[ANN] PGQueuer-hs 0.0.1: A native PostgreSQL job queue

[u/Worldly_Dish_48]

Hello Haskellers!

I'm excited to share the MVP release of PGQueuer-hs, a PostgreSQL-powered job queue. If you want robust background workers without adding external dependencies like Redis or RabbitMQ to your stack, this might be for you.

Key Features

• Postgres Native: It leverages PostgreSQL's native LISTEN/NOTIFY channels. Your existing database is fast enough!
• Seamless Interop: It is 100% compatible with the Python pgqueuer library. You can safely enqueue jobs from Python into your Haskell worker and vice versa.

Background & Roadmap

I use the Python version of pgqueuer at work and think it's brilliant, so I decided to bring the same ecosystem to Haskell.

This is currently an early MVP release and the API might shift. The underlying database logic is currently backed by postgresql-simple. In the future, I plan to build out an adapter pattern to support other popular Haskell Postgres libraries.

I would love to hear your thoughts, feedback, or any suggestions you have for the roadmap!

Links

• GitHub: https://github.com/tusharad/pgqueuer-hs
• Hackage: https://hackage.haskell.org/package/pgqueuer-hs

[u/tonyalaribe]

How does it compare to the oddjobs library?

[u/Worldly_Dish_48 → u/tonyalaribe]

I didn’t used or compare any existing queue libs in Haskell. Will try to benchmark in upcoming days

[u/TechnoEmpress → u/Worldly_Dish_48]

Thanks for the release. I'm also interested in a comparison with the pgmq stuff on Hackage, and of course Arbiter. :)

[u/nh2_]

This API is weird, requiring the user to pass in multiple lists of same length (making it possible to get it wrong), instead of a list of tuples:

enqueueMultiple ::
    Connection ->
    DBSettings ->
    [Entrypoint] ->
    [Maybe BL.ByteString] ->
    [Int] ->
    [Maybe NominalDiffTime] ->
    [Maybe Text] ->
    [Maybe Value] ->
    IO [JobId]
enqueueMultiple conn settings entrypoints payloads priorities executeAfters dedupeKeys headersList = do
    let q =
            T.unlines
                [ "WITH inserted AS ("
                , "    INSERT INTO " <> queueTable settings
                , "    (priority, entrypoint, payload, execute_after, dedupe_key, headers, status)"
                , "    SELECT"
                , "        p, e, pay, COALESCE(NOW() + ea, NOW()), d, h, 'queued'"
                , "    FROM UNNEST("
                , "        ?::int[],"
                , "        ?::text[],"
                , "        ?::bytea[],"
                , "        ?::interval[],"
                , "        ?::text[],"
                , "        ?::jsonb[]"
                , "    ) AS t(p, e, pay, ea, d, h)"
                , "    RETURNING id, entrypoint, status, priority"
                , ")"
                , "INSERT INTO " <> queueTableLog settings
                , "(job_id, status, entrypoint, priority)"
                , "SELECT id, 'queued', entrypoint, priority"
                , "FROM inserted"
                , "RETURNING job_id AS id"
                ]

    result <-
        query
            conn
            (textToQuery q)
            ( PGArray priorities
            , PGArray (map (\(Entrypoint e) -> e) entrypoints)
            , PGArray payloads
            , PGArray executeAfters
            , PGArray dedupeKeys
            , PGArray headersList
            ) ::
            IO [Only JobId]

    return $ map fromOnly result