packages feed

pgqueuer-hs (empty) → 0.0.1.0

raw patch · 13 files changed

+1688/−0 lines, 13 filesdep +aesondep +basedep +bytestringsetup-changed

Dependencies added: aeson, base, bytestring, containers, pgqueuer-hs, postgresql-simple, tasty, tasty-hunit, text, time, uuid

Files

+ CHANGELOG.md view
@@ -0,0 +1,11 @@+# Changelog for `pgqueuer-hs`++All notable changes to this project will be documented in this file.++The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),+and this project adheres to the+[Haskell Package Versioning Policy](https://pvp.haskell.org/).++## Unreleased++## 0.1.0.0 - YYYY-MM-DD
+ LICENSE view
@@ -0,0 +1,7 @@+Copyright 2026 Tushar Adhatrao++Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ README.md view
@@ -0,0 +1,114 @@+# pgqueuer-hs: PostgreSQL-powered job queues for Haskell++A PostgreSQL-powered job queue library. fully compatible Haskell implementation of [pgqueuer](https://github.com/janbjorge/pgqueuer).++## Overview++Your PostgreSQL database is already a job queue.++`pgqueuer-hs` turns PostgreSQL into a fast, reliable background job processor. Jobs live in the same database as your application data. One stack, full ACID guarantees, and no separate message broker to run.++### Key Features++- **PostgreSQL-Native**: Jobs stored directly in PostgreSQL with ACID guarantees+- **Transactional Enqueue**: Enqueue jobs in the same transaction as your application data+- **Safe Concurrency**: `FOR UPDATE SKIP LOCKED` prevents duplicate processing+- **Per-Entrypoint Limits**: Configure concurrency limits per job type+- **Global Limits**: Optional global concurrency control across all entrypoints+- **Instant Dispatch**: `LISTEN/NOTIFY` wakes workers immediately when jobs arrive+- **Deferred Jobs**: Schedule job execution with `execute_after`+- **Deduplication**: Prevent duplicate jobs with `dedupe_key`+- **Cross-Language Compatible**: Jobs created in Python can be processed by Haskell (and vice versa)+- **Job Tracking**: Complete logging of job status transitions+- **Error Handling**: Failed jobs can be retried or held for manual inspection++## Database Schema++The Haskell port uses the **exact same schema** as Python pgqueuer:++## Examples++Fully working examples are available in `./example` directory++## Installation++Add `pgqueuer-hs` to your `package.yaml` or `project.cabal` dependencies:++```yaml+dependencies:+  - pgqueuer-hs+  - postgresql-simple+  - uuid+  - time+  - text+  - aeson++## Quick start++Install schema if not installed.++```haskell+{-# LANGUAGE OverloadedStrings #-}++import Data.UUID.V4 (nextRandom)+import PGQueuer+import Data.Either (isLeft)+import Data.UUID.V4 (nextRandom)++main :: IO ()+main = do+    let conStr = "postgresql://queue_user:queue_pass@localhost:5432/queue_db"+    queueMgrId <- nextRandom+    withQueueManager conStr defaultDBSettings queueMgrId $ \qm -> do+        -- setup schema+        eInstalled <- verifyStructure qm+        when (isLeft eInstalled) (installSchema qm)++        -- Enqueue a job+        let ep = Entrypoint "hello"+        let params = [EntrypointExecutionParameter ep 0]+        qm1 <- registerEntrypoint qm ep (\_ -> pure ())+        _ <- enqueue qm1 ep Nothing 0 Nothing Nothing Nothing++        -- Dequeue jobs+        pickedJobs <- dequeue qm1 20 params Nothing 3+        mapM_ (\job -> jobStatus job) pickedJobs+```++## Core Concepts+### Architecture++PGQueuer creates a self-contained ecosystem within your PostgreSQL database:+  1. pgqueuer table: The primary ledger for active jobs (status: queued, picked).+  2. pgqueuer_log table: An unlogged table used for fast, high-volume event logging of job state transitions.+  3. pgqueuer_statistics table: Aggregates queue throughput and metrics.+  4. pgqueuer_schedules table: Manages cron-like recurring jobs and future executions.+  +  Database Triggers: Automatically emit pub/sub notifications via fn_pgqueuer_changed when queue states mutate.++### Status Lifecycle++- Jobs transition through various states defined by JobStatus:+- Queued - Waiting for a worker.+- Picked - Claimed by a worker (protected by a heartbeat timeout).+- Successful - Completed without errors.+- Failed / Exception - Encountered an error (eligible for retry).+- Canceled / Deleted - Terminated or scrubbed.++> [!IMPORTANT]  +> Codebase is currently highly unstable.++## Notes++- Postgresql-simple is currently being used as the primary database driver. In the future, adapter drivers will be implemented.+- Scheduling is not supported right now.++## License++This project is licensed under the MIT License - see the LICENSE file for details.++## See Also++- [Python pgqueuer](https://github.com/janbjorge/pgqueuer)+- [PostgreSQL Documentation](https://www.postgresql.org/docs/)+- [postgresql-simple](http://hackage.haskell.org/package/postgresql-simple)
+ Setup.hs view
@@ -0,0 +1,3 @@+import Distribution.Simple++main = defaultMain
+ pgqueuer-hs.cabal view
@@ -0,0 +1,73 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.38.1.+--+-- see: https://github.com/sol/hpack++name:           pgqueuer-hs+version:        0.0.1.0+synopsis:       PostgreSQL powered Job queue+description:    Please see the README on GitHub at <https://github.com/tusharad/pgqueuer-hs#readme>+category:       Database+homepage:       https://github.com/tusharad/pgqueuer-hs#readme+bug-reports:    https://github.com/tusharad/pgqueuer-hs/issues+author:         Tushar Adhatrao+maintainer:     tusharadhatrao@gmail.com+copyright:      2026 Tushar Adhatrao+license:        MIT+license-file:   LICENSE+build-type:     Simple+extra-source-files:+    README.md+    CHANGELOG.md++source-repository head+  type: git+  location: https://github.com/tusharad/pgqueuer-hs++library+  exposed-modules:+      PGQueuer+      PGQueuer.Listener+      PGQueuer.Query+      PGQueuer.Schema+      PGQueuer.Settings+      PGQueuer.Types+  other-modules:+      Paths_pgqueuer_hs+  hs-source-dirs:+      src+  ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wmissing-export-lists -Wmissing-home-modules -Wredundant-constraints -Wno-partial-fields+  build-depends:+      aeson >=2.0 && <3+    , base >=4.7 && <5+    , bytestring >=0.10.4.0 && <0.13+    , containers >=0.6 && <0.8+    , postgresql-simple >=0.7 && <0.9+    , text >=1.2 && <3+    , time >=1.9 && <2+    , uuid >=1.3 && <2+  default-language: Haskell2010++test-suite pgqueuer-hs-test+  type: exitcode-stdio-1.0+  main-is: Spec.hs+  other-modules:+      Tests.PGQueuer+      Paths_pgqueuer_hs+  hs-source-dirs:+      test+  ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wmissing-export-lists -Wmissing-home-modules -Wredundant-constraints -Wno-partial-fields -threaded -rtsopts -with-rtsopts=-N+  build-depends:+      aeson >=2.0 && <3+    , base >=4.7 && <5+    , bytestring >=0.10.4.0 && <0.13+    , containers >=0.6 && <0.8+    , pgqueuer-hs+    , postgresql-simple >=0.7 && <0.9+    , tasty >=1.4 && <2+    , tasty-hunit >=0.10 && <1+    , text >=1.2 && <3+    , time >=1.9 && <2+    , uuid >=1.3 && <2+  default-language: Haskell2010
+ src/PGQueuer.hs view
@@ -0,0 +1,278 @@+{- |+Module      : PGQueuer+Description : Main interface for the PGQueuer job queue ecosystem.+Copyright   : (c) Tushar Adhatrao, 2026+License     : MIT+Maintainer  : tusharadhatrao@gmail.com+Stability   : experimental++This module provides the high-level API for interacting with PGQueuer.+It re-exports essential types and settings required to configure,+manage, and interact with the database-backed queue.+-}+module PGQueuer (+    -- * Queue Manager+    QueueManager (..),+    createQueueManager,+    withQueueManager,++    -- * Entrypoint Management+    registerEntrypoint,++    -- * Job Operations+    workerLoop,+    enqueue,+    enqueueMultiple,+    dequeue,+    markJobAsCancelled,+    requeueJobs,+    retryJob,+    updateHeartbeat,+    logJobs,++    -- * Queue Statistics and Management+    getQueueSize,+    clearQueue,+    listFailedJobs,+    listJobStatusById,++    -- * Schema management+    verifyStructure,+    installSchema,+    uninstallSchema,++    -- * Re-exported modules+    module PGQueuer.Types,+    module PGQueuer.Settings,+) where++import Control.Concurrent (threadDelay)+import Control.Exception (bracket)+import Control.Monad (forever)+import Data.Aeson (Value)+import Data.ByteString (ByteString)+import qualified Data.ByteString.Lazy as BL+import Data.Map.Strict (Map)+import qualified Data.Map.Strict as Map+import Data.Text (Text)+import Data.Time (NominalDiffTime)+import Data.UUID (UUID)+import Database.PostgreSQL.Simple (Connection, close, connectPostgreSQL)++import qualified PGQueuer.Query as Q+import PGQueuer.Schema (install, uninstall, verifyStructure_)+import PGQueuer.Settings+import PGQueuer.Types++-- | Core state context for managing Queue.+data QueueManager = QueueManager+    { qmConnection :: Connection+    -- ^ Database connection+    , qmSettings :: DBSettings+    -- ^ Database settings+    , qmEntrypoints :: Map Text EntrypointHandler+    -- ^ Registered entrypoint handlers+    , qmQueueManagerId :: UUID+    -- ^ Unique identifier for this QueueManager instance+    }++-- | Entrypoint handler type+type EntrypointHandler = Job -> IO ()++-- | Create a new QueueManager from a connection+createQueueManager ::+    -- | Database connection+    Connection ->+    -- | Database settings+    DBSettings ->+    -- | Unique identifier for this QueueManager instance+    UUID ->+    IO QueueManager+createQueueManager conn settings queueMgrId = do+    return $+        QueueManager+            { qmConnection = conn+            , qmSettings = settings+            , qmEntrypoints = Map.empty+            , qmQueueManagerId = queueMgrId+            }++-- | Register an entrypoint handler+registerEntrypoint ::+    -- | QueueManager instance+    QueueManager ->+    -- | Entrypoint to register+    Entrypoint ->+    -- | Handler function for the entrypoint+    EntrypointHandler ->+    IO QueueManager+registerEntrypoint qm (Entrypoint ep) handler = do+    return+        qm+            { qmEntrypoints = Map.insert ep handler (qmEntrypoints qm)+            }++-- | Continuously dequeue jobs and dispatch them to registered handlers.+workerLoop :: QueueManager -> [EntrypointExecutionParameter] -> IO ()+workerLoop qm params = forever $ do+    jobs <- dequeue qm defaultBatchSize params Nothing defaultHeartbeatTimeout+    if null jobs+        then threadDelay 1000000+        else mapM_ (dispatchJob qm) jobs++dispatchJob :: QueueManager -> Job -> IO ()+dispatchJob qm job =+    case Map.lookup entrypointName (qmEntrypoints qm) of+        Nothing -> fail $ "No handler registered for entrypoint: " ++ show entrypointName+        Just handler -> handler job+  where+    Entrypoint entrypointName = jobEntrypoint job++-- ============================================================================+-- Queue operations (wrappers around Query module)+-- ============================================================================++-- | Enqueue a single job+enqueue ::+    -- | QueueManager instance+    QueueManager ->+    -- | Entrypoint for the job+    Entrypoint ->+    -- | Optional payload for the job+    Maybe BL.ByteString ->+    -- | Priority of the job+    Int ->+    -- | Optional delay before the job can be executed+    Maybe NominalDiffTime ->+    -- | Optional deduplication key for the job+    Maybe Text ->+    -- | Optional headers for the job+    Maybe Value ->+    IO [JobId]+enqueue qm =+    Q.enqueueSingle+        (qmConnection qm)+        (qmSettings qm)++-- | Enqueue multiple jobs+enqueueMultiple ::+    -- | QueueManager instance+    QueueManager ->+    -- | List of entrypoints for the jobs+    [Entrypoint] ->+    -- | List of optional payloads for the jobs+    [Maybe BL.ByteString] ->+    -- | List of priorities for the jobs+    [Int] ->+    -- | List of optional delays before the jobs can be executed+    [Maybe NominalDiffTime] ->+    -- | List of optional deduplication keys for the jobs+    [Maybe Text] ->+    -- | List of optional headers for the jobs+    [Maybe Value] ->+    IO [JobId]+enqueueMultiple qm =+    Q.enqueueMultiple+        (qmConnection qm)+        (qmSettings qm)++-- | Dequeue jobs+dequeue ::+    -- | QueueManager instance+    QueueManager ->+    -- | Batch size for dequeuing jobs+    Int ->+    -- | List of entrypoint execution parameters+    [EntrypointExecutionParameter] ->+    -- | Optional maximum number of jobs to dequeue+    Maybe Int ->+    -- | Heartbeat timeout in seconds+    Int ->+    IO [Job]+dequeue qm batchSize params =+    Q.dequeue+        (qmConnection qm)+        (qmSettings qm)+        batchSize+        params+        (qmQueueManagerId qm)++-- | Log job status changes+logJobs ::+    -- | QueueManager instance+    QueueManager ->+    -- | List of job status changes with optional traceback+    [(JobId, JobStatus, Maybe Value)] ->+    IO ()+logJobs qm =+    Q.logJobs (qmConnection qm) (qmSettings qm)++-- | Get queue size statistics+getQueueSize :: QueueManager -> IO [QueueStatistics]+getQueueSize qm = Q.queueSize (qmConnection qm) (qmSettings qm)++-- | Clear the queue+clearQueue ::+    -- | QueueManager instance+    QueueManager ->+    -- | Optional list of entrypoints to clear; if Nothing, clears all+    Maybe [Entrypoint] ->+    IO ()+clearQueue qm = Q.clearQueue (qmConnection qm) (qmSettings qm)++-- | List failed jobs+listFailedJobs :: QueueManager -> Int -> IO [Job]+listFailedJobs qm = Q.listFailedJobs (qmConnection qm) (qmSettings qm)++-- | List Job status by Id+listJobStatusById :: QueueManager -> [JobId] -> IO [(JobId, JobStatus)]+listJobStatusById qm = Q.jobStatusById (qmConnection qm) (qmSettings qm)++-- | Mark jobs as cancelled+markJobAsCancelled :: QueueManager -> [JobId] -> IO ()+markJobAsCancelled qm = Q.markJobAsCancelled (qmConnection qm) (qmSettings qm)++-- | Update heartbeat+updateHeartbeat :: QueueManager -> [JobId] -> IO ()+updateHeartbeat qm = Q.updateHeartbeat (qmConnection qm) (qmSettings qm)++-- | Requeue jobs+requeueJobs :: QueueManager -> [JobId] -> IO ()+requeueJobs qm = Q.requeueJobs (qmConnection qm) (qmSettings qm)++-- | Retry a job+retryJob :: QueueManager -> Job -> NominalDiffTime -> Maybe Value -> IO ()+retryJob qm = Q.retryJob (qmConnection qm) (qmSettings qm)++-- ============================================================================+-- Schema management+-- ============================================================================++-- | Verify schema structure+verifyStructure :: QueueManager -> IO (Either String ())+verifyStructure qm = PGQueuer.Schema.verifyStructure_ (qmConnection qm) (qmSettings qm)++-- | Install schema+installSchema :: QueueManager -> IO ()+installSchema qm = install (qmConnection qm) (qmSettings qm)++-- | Uninstall schema+uninstallSchema :: QueueManager -> IO ()+uninstallSchema qm = uninstall (qmConnection qm) (qmSettings qm)++-- | Run a QueueManager within a bracket (for resource safety)+withQueueManager ::+    -- | Connection string for PostgreSQL+    ByteString ->+    -- | Database settings+    DBSettings ->+    -- | Unique identifier for this QueueManager instance+    UUID ->+    -- | Action to run with the QueueManager+    (QueueManager -> IO a) ->+    IO a+withQueueManager connStr settings queueMgrId action = do+    bracket+        (connectPostgreSQL connStr >>= \conn -> createQueueManager conn settings queueMgrId)+        (close . qmConnection)+        action
+ src/PGQueuer/Listener.hs view
@@ -0,0 +1,91 @@+{-# LANGUAGE OverloadedStrings #-}++module PGQueuer.Listener (+    listen,+    unlisten,+    unlistenAll,+    getNextNotification,+    notifyJobCancellation,+    notifyHealthCheck,+    parseEventPayload,+) where++import Control.Monad (void)+import Data.Aeson ((.=))+import qualified Data.Aeson as Aeson+import qualified Data.ByteString.Char8 as BS+import qualified Data.ByteString.Lazy as BL+import qualified Data.Text as T+import qualified Data.Text.Encoding as TE+import Data.Time (getCurrentTime)+import Data.UUID.V4 (nextRandom)+import Database.PostgreSQL.Simple (Connection, Only (..), execute, execute_)+import Database.PostgreSQL.Simple.Notification (Notification, getNotification)+import Database.PostgreSQL.Simple.Types (Query (..))+import PGQueuer.Types (Channel (..), defaultChannel)++-- | Listen for notifications on a specific channel+listen :: Connection -> T.Text -> IO ()+listen conn channel =+    void $ execute_ conn (Query $ TE.encodeUtf8 ("LISTEN " <> channel))++-- | Stop listening on a specific channel+unlisten :: Connection -> T.Text -> IO ()+unlisten conn channel =+    void $ execute_ conn (Query $ TE.encodeUtf8 ("UNLISTEN " <> channel))++-- | Stop listening on all channels+unlistenAll :: Connection -> IO ()+unlistenAll conn =+    void $ execute_ conn (Query "UNLISTEN *;")++-- | Get the next notification, blocking until available+getNextNotification :: Connection -> IO (Maybe Notification)+getNextNotification conn = Just <$> getNotification conn++sendNotification :: Connection -> Aeson.Value -> IO ()+sendNotification conn payload =+    void $ execute conn "NOTIFY pgqueuer, ?" (Only payloadText)+  where+    payloadText = TE.decodeUtf8 . BL.toStrict $ Aeson.encode payload++notificationChannel :: T.Text+notificationChannel = channelText+  where+    Channel channelText = defaultChannel++-- | Send a job cancellation notification+notifyJobCancellation :: Connection -> Int -> IO ()+notifyJobCancellation conn jobId = do+    sentAt <- getCurrentTime+    sendNotification+        conn+        ( Aeson.object+            [ "channel" .= notificationChannel+            , "sent_at" .= sentAt+            , "type" .= ("cancellation_event" :: T.Text)+            , "ids" .= ([jobId] :: [Int])+            ]+        )++-- | Send a health check notification+notifyHealthCheck :: Connection -> IO ()+notifyHealthCheck conn = do+    sentAt <- getCurrentTime+    healthCheckId <- nextRandom+    sendNotification+        conn+        ( Aeson.object+            [ "channel" .= notificationChannel+            , "sent_at" .= sentAt+            , "type" .= ("health_check_event" :: T.Text)+            , "id" .= healthCheckId+            ]+        )++-- | Parse event payload from notification+parseEventPayload :: BS.ByteString -> Either String String+parseEventPayload payload =+    case Aeson.eitherDecodeStrict' payload :: Either String Aeson.Value of+        Left err -> Left err+        Right _ -> Right (BS.unpack payload)
+ src/PGQueuer/Query.hs view
@@ -0,0 +1,417 @@+{-# LANGUAGE OverloadedStrings #-}++module PGQueuer.Query (+    enqueueSingle,+    enqueueMultiple,+    dequeue,+    logJobs,+    queueSize,+    queuedWork,+    retryJob,+    requeueJobs,+    markJobAsCancelled,+    updateHeartbeat,+    jobStatusById,+    clearQueue,+    listFailedJobs,+) where++import Data.Aeson (Value)+import qualified Data.ByteString.Lazy as BL+import Data.Text (Text)+import qualified Data.Text.Encoding as TE+import Data.Time (NominalDiffTime, addUTCTime)+import Data.UUID (UUID)+import Database.PostgreSQL.Simple+import Database.PostgreSQL.Simple.Types (PGArray (..), Query (..))+import PGQueuer.Settings+import PGQueuer.Types++import qualified Data.Text as T++textToQuery :: Text -> Query+textToQuery = Query . TE.encodeUtf8++-- | Enqueue a single job+enqueueSingle ::+    Connection ->+    DBSettings ->+    Entrypoint ->+    Maybe BL.ByteString ->+    Int ->+    Maybe NominalDiffTime ->+    Maybe Text ->+    Maybe Value ->+    IO [JobId]+enqueueSingle conn settings entrypoint payload priority executeAfter dedupeKey headers = do+    enqueueMultiple+        conn+        settings+        [entrypoint]+        (maybeToList payload)+        [priority]+        (maybeToList executeAfter)+        (maybeToList dedupeKey)+        (maybeToList headers)+  where+    maybeToList Nothing = []+    maybeToList (Just x) = [Just x]++-- | Enqueue multiple jobs+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++-- | Dequeue jobs respecting concurrency limits+dequeue ::+    Connection ->+    DBSettings ->+    Int -> -- batch size+    [EntrypointExecutionParameter] -> -- per-entrypoint parameters+    UUID -> -- queue manager id+    Maybe Int -> -- global concurrency limit+    Int -> -- heartbeat timeout in seconds+    IO [Job]+dequeue conn settings batchSize params queueMgrId globalLimit heartbeatTimeoutSecs = do+    let entrypoints = map paramEntrypoint params+        concurrencyLimits = map paramConcurrencyLimit params++        q =+            T.unlines+                [ "WITH"+                , "runtime AS ("+                , "    SELECT"+                , "        ?::int    AS batch_size,"+                , "        ?::uuid   AS queue_manager_id,"+                , "        ?::bigint AS global_limit"+                , "),"+                , "params AS ("+                , "    SELECT"+                , "        UNNEST(?::text[])   AS entrypoint,"+                , "        UNNEST(?::bigint[]) AS concurrency_limit"+                , "),"+                , "picked AS ("+                , "    SELECT entrypoint, COUNT(*) AS total"+                , "    FROM " <> queueTable settings <> " q"+                , "    WHERE q.queue_manager_id IS NOT NULL"+                , "      AND EXISTS (SELECT 1 FROM params p WHERE p.entrypoint = q.entrypoint)"+                , "    GROUP BY q.entrypoint"+                , "),"+                , "worker_load AS ("+                , "    SELECT COUNT(*) AS total"+                , "    FROM " <> queueTable settings <> " q"+                , "    WHERE q.queue_manager_id = (SELECT queue_manager_id FROM runtime)"+                , "      AND EXISTS (SELECT 1 FROM params p WHERE p.entrypoint = q.entrypoint)"+                , "),"+                , "available AS ("+                , "    SELECT p.entrypoint"+                , "    FROM params p"+                , "    LEFT JOIN picked pk ON pk.entrypoint = p.entrypoint"+                , "    WHERE p.concurrency_limit <= 0"+                , "       OR COALESCE(pk.total, 0) < p.concurrency_limit"+                , "),"+                , "next_queued_src AS ("+                , "    SELECT q.id, q.priority"+                , "    FROM available a"+                , "    CROSS JOIN LATERAL ("+                , "        SELECT q2.id, q2.priority"+                , "        FROM " <> queueTable settings <> " q2"+                , "        WHERE q2.entrypoint = a.entrypoint"+                , "          AND q2.status = 'queued'"+                , "          AND q2.execute_after < NOW()"+                , "        ORDER BY q2.priority DESC, q2.id ASC"+                , "        LIMIT (SELECT batch_size FROM runtime)"+                , "        FOR UPDATE SKIP LOCKED"+                , "    ) q"+                , "    WHERE ((SELECT global_limit FROM runtime) IS NULL"+                , "           OR (SELECT total FROM worker_load) < (SELECT global_limit FROM runtime))"+                , "    ORDER BY q.priority DESC, q.id ASC"+                , "    LIMIT (SELECT batch_size FROM runtime)"+                , "),"+                , "next_queued AS ("+                , "    SELECT id FROM next_queued_src"+                , "),"+                , "next_stale AS ("+                , "    SELECT q.id"+                , "    FROM " <> queueTable settings <> " q"+                , "    JOIN params p ON p.entrypoint = q.entrypoint"+                , "    WHERE q.status = 'picked'"+                , "      AND q.heartbeat < NOW() - INTERVAL '" <> T.pack (show heartbeatTimeoutSecs) <> " seconds'"+                , "      AND q.execute_after < NOW()"+                , "      AND ((SELECT global_limit FROM runtime) IS NULL"+                , "           OR (SELECT total FROM worker_load) < (SELECT global_limit FROM runtime))"+                , "    ORDER BY q.priority DESC, q.id ASC"+                , "    FOR UPDATE SKIP LOCKED"+                , "    LIMIT (SELECT batch_size FROM runtime)"+                , "),"+                , "eligible AS ("+                , "    SELECT id FROM ("+                , "        SELECT id, 0 AS src FROM next_queued"+                , "        UNION ALL"+                , "        SELECT id, 1 AS src FROM next_stale"+                , "    ) combined"+                , "    ORDER BY src, id"+                , "    LIMIT (SELECT batch_size FROM runtime)"+                , "),"+                , "claimed AS ("+                , "    UPDATE " <> queueTable settings+                , "    SET status = 'picked',"+                , "        updated   = NOW(),"+                , "        heartbeat = NOW(),"+                , "        queue_manager_id = (SELECT queue_manager_id FROM runtime)"+                , "    WHERE id IN (SELECT id FROM eligible)"+                , "    RETURNING *"+                , "),"+                , "log_pick AS ("+                , "    INSERT INTO " <> queueTableLog settings <> " (job_id, status, entrypoint, priority)"+                , "    SELECT id, status, entrypoint, priority FROM claimed"+                , ")"+                , "SELECT id, priority, created, updated, heartbeat, execute_after, status::text AS status, entrypoint, payload, attempts, queue_manager_id, headers"+                , "FROM claimed"+                , "ORDER BY priority DESC, id ASC"+                ]+    query+        conn+        (textToQuery q)+        ( batchSize+        , queueMgrId+        , globalLimit+        , PGArray $ map (\(Entrypoint e) -> e) entrypoints+        , PGArray concurrencyLimits+        )++-- | Log job completions and status changes+logJobs ::+    Connection ->+    DBSettings ->+    [(JobId, JobStatus, Maybe Value)] -> -- (job_id, status, traceback)+    IO ()+logJobs conn settings jobStatuses = do+    let jobIds = map (\(JobId i, _, _) -> i) jobStatuses+        statuses = map (\(_, s, _) -> jobStatusToText s) jobStatuses+        tracebacks = map (\(_, _, tb) -> tb) jobStatuses++        q =+            T.unlines+                [ "WITH job_status AS ("+                , "    SELECT"+                , "        UNNEST(?::integer[])   AS id,"+                , "        UNNEST(?::" <> queueStatusType settings <> "[]) AS status,"+                , "        UNNEST(?::JSONB[])     AS traceback"+                , "), deleted AS ("+                , "    DELETE FROM " <> queueTable settings+                , "    WHERE id = ANY(SELECT js.id FROM job_status js WHERE js.status != 'failed')"+                , "    RETURNING id, entrypoint, priority"+                , "), held AS ("+                , "    UPDATE " <> queueTable settings+                , "    SET status = 'failed', updated = NOW(), queue_manager_id = NULL"+                , "    WHERE id = ANY(SELECT js.id FROM job_status js WHERE js.status = 'failed')"+                , "    RETURNING id, entrypoint, priority"+                , "), all_resolved AS ("+                , "    SELECT id, entrypoint, priority FROM deleted"+                , "    UNION ALL"+                , "    SELECT id, entrypoint, priority FROM held"+                , "), merged AS ("+                , "    SELECT"+                , "        job_status.id           AS id,"+                , "        job_status.status       AS status,"+                , "        job_status.traceback    AS traceback,"+                , "        all_resolved.entrypoint AS entrypoint,"+                , "        all_resolved.priority   AS priority"+                , "    FROM job_status"+                , "    INNER JOIN all_resolved"+                , "        ON all_resolved.id = job_status.id"+                , ")"+                , "INSERT INTO " <> queueTableLog settings <> " ("+                , "    job_id,"+                , "    status,"+                , "    entrypoint,"+                , "    priority,"+                , "    traceback"+                , ")"+                , "SELECT id, status, entrypoint, priority, traceback FROM merged"+                ]+    _ <- execute conn (textToQuery q) (PGArray jobIds, PGArray statuses, PGArray tracebacks)+    return ()++-- | Get queue size statistics+queueSize :: Connection -> DBSettings -> IO [QueueStatistics]+queueSize conn settings = do+    let q =+            T.unlines+                [ "SELECT"+                , "    count(*) AS count,"+                , "    entrypoint,"+                , "    priority,"+                , "    status::text AS status"+                , "FROM " <> queueTable settings+                , "GROUP BY entrypoint, priority, status"+                , "ORDER BY count, entrypoint, priority, status"+                ]+    query_ conn (textToQuery q)++-- | Get queued work count for specific entrypoints+queuedWork :: Connection -> DBSettings -> [Entrypoint] -> IO Int+queuedWork conn settings entrypoints = do+    let eps = map (\(Entrypoint e) -> e) entrypoints+        q =+            T.unlines+                [ "SELECT COUNT(*)"+                , "FROM " <> queueTable settings+                , "WHERE entrypoint = ANY(?::text[])"+                , "  AND status = 'queued'"+                ]+    result <- query conn (textToQuery q) (Only $ PGArray eps) :: IO [Only Int]+    case result of+        [Only count] -> return count+        _ -> return 0++-- | Retry a failed job+retryJob ::+    Connection ->+    DBSettings ->+    Job ->+    NominalDiffTime -> -- delay+    Maybe Value -> -- traceback+    IO ()+retryJob conn settings job delay _traceback = do+    let newExecuteAfter = addUTCTime delay (jobExecuteAfter job)+        newAttempts = jobAttempts job + 1+        q =+            T.unlines+                [ "UPDATE " <> queueTable settings+                , "SET status = 'queued',"+                , "    execute_after = ?,"+                , "    attempts = ?,"+                , "    queue_manager_id = NULL,"+                , "    updated = NOW()"+                , "WHERE id = ?"+                ]+    _ <- execute conn (textToQuery q) (newExecuteAfter, newAttempts, jobId job)+    return ()++-- | Requeue multiple jobs+requeueJobs :: Connection -> DBSettings -> [JobId] -> IO ()+requeueJobs conn settings jobIds = do+    let q =+            T.unlines+                [ "UPDATE " <> queueTable settings+                , "SET status = 'queued',"+                , "    queue_manager_id = NULL,"+                , "    updated = NOW()"+                , "WHERE id = ANY(?::integer[])"+                ]+    _ <- execute conn (textToQuery q) (Only $ PGArray jobIds)+    return ()++-- | Mark jobs as cancelled+markJobAsCancelled :: Connection -> DBSettings -> [JobId] -> IO ()+markJobAsCancelled conn settings jobIds = do+    let q =+            T.unlines+                [ "UPDATE " <> queueTable settings+                , "SET status = 'canceled', updated = NOW()"+                , "WHERE id = ANY(?::integer[])"+                ]+    _ <- execute conn (textToQuery q) (Only $ PGArray jobIds)+    return ()++-- | Update heartbeat for active jobs+updateHeartbeat :: Connection -> DBSettings -> [JobId] -> IO ()+updateHeartbeat conn settings jobIds = do+    let q =+            T.unlines+                [ "UPDATE " <> queueTable settings+                , "SET heartbeat = NOW()"+                , "WHERE id = ANY(?::integer[]) AND status = 'picked'"+                ]+    _ <- execute conn (textToQuery q) (Only $ PGArray jobIds)+    return ()++-- | Get job status by IDs+jobStatusById :: Connection -> DBSettings -> [JobId] -> IO [(JobId, JobStatus)]+jobStatusById conn settings jobIds = do+    let q =+            T.unlines+                [ "SELECT id, status::text AS status"+                , "FROM " <> queueTable settings+                , "WHERE id = ANY(?::integer[])"+                ]+    query conn (textToQuery q) (Only $ PGArray jobIds) :: IO [(JobId, JobStatus)]++-- | Clear entire queue or by entrypoint+clearQueue :: Connection -> DBSettings -> Maybe [Entrypoint] -> IO ()+clearQueue conn settings mbEntrypoints = do+    case mbEntrypoints of+        Nothing -> do+            let q = "DELETE FROM " <> queueTable settings+            _ <- execute_ conn (textToQuery q)+            return ()+        Just entrypoints -> do+            let eps = map (\(Entrypoint e) -> e) entrypoints+                q =+                    T.unlines+                        [ "DELETE FROM " <> queueTable settings+                        , "WHERE entrypoint = ANY(?::text[])"+                        ]+            _ <- execute conn (textToQuery q) (Only $ PGArray eps)+            return ()++-- | List failed jobs+listFailedJobs :: Connection -> DBSettings -> Int -> IO [Job]+listFailedJobs conn settings limit = do+    let q =+            T.unlines+                [ "SELECT id, priority, created, updated, heartbeat, execute_after, status::text AS status, entrypoint, payload, attempts, queue_manager_id, headers"+                , "FROM " <> queueTable settings+                , "WHERE status = 'failed'"+                , "ORDER BY updated DESC"+                , "LIMIT " <> T.pack (show limit)+                ]+    query_ conn (textToQuery q)
+ src/PGQueuer/Schema.hs view
@@ -0,0 +1,188 @@+{-# LANGUAGE OverloadedStrings #-}++module PGQueuer.Schema (+    install,+    uninstall,+    verifyStructure_,+) where++import Data.String (fromString)+import qualified Data.Text as T+import Database.PostgreSQL.Simple (Connection, execute_)+import qualified Database.PostgreSQL.Simple as PG+import Database.PostgreSQL.Simple.Types (Only (..))+import PGQueuer.Settings++-- | Install the schema in the database+install :: Connection -> DBSettings -> IO ()+install conn _settings = do+    _ <- execute_ conn "CREATE TYPE pgqueuer_status AS ENUM ('queued', 'picked', 'successful', 'exception', 'canceled', 'deleted', 'failed');"+    _ <- execute_ conn $ textToQuery queueTableSQL+    _ <- execute_ conn $ textToQuery queueLogTableSQL+    _ <- execute_ conn $ textToQuery statisticsTableSQL+    _ <- execute_ conn $ textToQuery schedulesTableSQL+    _ <- execute_ conn $ textToQuery triggerFunctionSQL+    _ <- execute_ conn $ textToQuery triggerSQL+    return ()++-- | Uninstall the schema from the database+uninstall :: Connection -> DBSettings -> IO ()+uninstall conn _settings = do+    _ <- execute_ conn "DROP TRIGGER IF EXISTS tg_pgqueuer_changed ON pgqueuer;"+    _ <- execute_ conn "DROP TABLE IF EXISTS pgqueuer;"+    _ <- execute_ conn "DROP TABLE IF EXISTS pgqueuer_log;"+    _ <- execute_ conn "DROP TABLE IF EXISTS pgqueuer_statistics;"+    _ <- execute_ conn "DROP TABLE IF EXISTS pgqueuer_schedules;"+    _ <- execute_ conn "DROP TYPE IF EXISTS pgqueuer_status;"+    _ <- execute_ conn "DROP FUNCTION IF EXISTS fn_pgqueuer_changed();"+    return ()++-- | Verify the schema is properly installed+verifyStructure_ :: Connection -> DBSettings -> IO (Either String ())+verifyStructure_ conn _settings = do+    result <- PG.query_ conn "SELECT EXISTS (SELECT FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = 'pgqueuer');" :: IO [Only Bool]+    case result of+        [Only True] -> return $ Right ()+        _ -> return $ Left "Queue table is missing. Please run 'pgqueuer install' to set up the schema."++textToQuery :: T.Text -> PG.Query+textToQuery = fromString . T.unpack++queueTableSQL :: T.Text+queueTableSQL =+    T.unlines+        [ "CREATE TABLE IF NOT EXISTS pgqueuer ("+        , "    id SERIAL PRIMARY KEY,"+        , "    priority INT NOT NULL,"+        , "    queue_manager_id UUID,"+        , "    created TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL,"+        , "    updated TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL,"+        , "    heartbeat TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL,"+        , "    execute_after TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL,"+        , "    status pgqueuer_status NOT NULL,"+        , "    entrypoint TEXT NOT NULL,"+        , "    dedupe_key TEXT,"+        , "    payload BYTEA,"+        , "    headers JSONB,"+        , "    attempts INT NOT NULL DEFAULT 0"+        , ");"+        , "CREATE INDEX IF NOT EXISTS pgqueuer_priority_id_idx ON pgqueuer (priority ASC, id DESC)"+        , "    INCLUDE (id) WHERE status = 'queued';"+        , "CREATE INDEX IF NOT EXISTS pgqueuer_updated_id_idx ON pgqueuer (updated ASC, id DESC)"+        , "    INCLUDE (id) WHERE status = 'picked';"+        , "CREATE INDEX IF NOT EXISTS pgqueuer_manager_id_idx ON pgqueuer (queue_manager_id)"+        , "    WHERE queue_manager_id IS NOT NULL;"+        , "CREATE INDEX IF NOT EXISTS pgqueuer_ep_prio_id_idx ON pgqueuer (entrypoint, priority DESC, id ASC)"+        , "    WHERE status = 'queued';"+        , "CREATE INDEX IF NOT EXISTS pgqueuer_ep_ea_idx ON pgqueuer (entrypoint, execute_after)"+        , "    WHERE status = 'queued';"+        , "CREATE UNIQUE INDEX IF NOT EXISTS pgqueuer_unique_dedupe_key ON pgqueuer (dedupe_key)"+        , "    WHERE ((status IN ('queued', 'picked') AND dedupe_key IS NOT NULL));"+        ]++queueLogTableSQL :: T.Text+queueLogTableSQL =+    T.unlines+        [ "CREATE UNLOGGED TABLE IF NOT EXISTS pgqueuer_log ("+        , "    id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,"+        , "    created TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL,"+        , "    job_id BIGINT NOT NULL,"+        , "    status pgqueuer_status NOT NULL,"+        , "    priority INT NOT NULL,"+        , "    entrypoint TEXT NOT NULL,"+        , "    traceback JSONB DEFAULT NULL,"+        , "    aggregated BOOLEAN DEFAULT FALSE"+        , ");"+        , "CREATE INDEX IF NOT EXISTS pgqueuer_log_not_aggregated ON pgqueuer_log (entrypoint, priority, status, created) WHERE not aggregated;"+        , "CREATE INDEX IF NOT EXISTS pgqueuer_log_created ON pgqueuer_log (created);"+        , "CREATE INDEX IF NOT EXISTS pgqueuer_log_status ON pgqueuer_log (status);"+        , "CREATE INDEX IF NOT EXISTS pgqueuer_log_job_id_status ON pgqueuer_log (job_id, created DESC);"+        ]++statisticsTableSQL :: T.Text+statisticsTableSQL =+    T.unlines+        [ "CREATE TABLE IF NOT EXISTS pgqueuer_statistics ("+        , "    id SERIAL PRIMARY KEY,"+        , "    created TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT DATE_TRUNC('sec', NOW() at time zone 'UTC'),"+        , "    count BIGINT NOT NULL,"+        , "    priority INT NOT NULL,"+        , "    status pgqueuer_status NOT NULL,"+        , "    entrypoint TEXT NOT NULL"+        , ");"+        , "CREATE UNIQUE INDEX IF NOT EXISTS pgqueuer_statistics_unique_count ON pgqueuer_statistics ("+        , "    priority,"+        , "    DATE_TRUNC('sec', created at time zone 'UTC'),"+        , "    status,"+        , "    entrypoint"+        , ");"+        ]++schedulesTableSQL :: T.Text+schedulesTableSQL =+    T.unlines+        [ "CREATE TABLE IF NOT EXISTS pgqueuer_schedules ("+        , "    id SERIAL PRIMARY KEY,"+        , "    expression TEXT NOT NULL,"+        , "    entrypoint TEXT NOT NULL,"+        , "    heartbeat TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL,"+        , "    created TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL,"+        , "    updated TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL,"+        , "    next_run TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL,"+        , "    last_run TIMESTAMP WITH TIME ZONE,"+        , "    status pgqueuer_status DEFAULT 'queued',"+        , "    UNIQUE (expression, entrypoint)"+        , ");"+        ]++triggerFunctionSQL :: T.Text+triggerFunctionSQL =+    T.unlines+        [ "CREATE OR REPLACE FUNCTION fn_pgqueuer_changed() RETURNS TRIGGER AS $$"+        , "DECLARE"+        , "    to_emit BOOLEAN := false;"+        , "BEGIN"+        , "    IF TG_OP = 'UPDATE' AND OLD IS DISTINCT FROM NEW THEN"+        , "        to_emit := true;"+        , "    ELSIF TG_OP = 'DELETE' THEN"+        , "        to_emit := true;"+        , "    ELSIF TG_OP = 'INSERT' THEN"+        , "        to_emit := true;"+        , "    ELSIF TG_OP = 'TRUNCATE' THEN"+        , "        to_emit := true;"+        , "    END IF;"+        , ""+        , "    IF to_emit THEN"+        , "        PERFORM pg_notify("+        , "            'pgqueuer',"+        , "            json_build_object("+        , "                'channel', 'pgqueuer',"+        , "                'operation', lower(TG_OP),"+        , "                'sent_at', NOW(),"+        , "                'table', TG_TABLE_NAME,"+        , "                'type', 'table_changed_event'"+        , "            )::text"+        , "        );"+        , "    END IF;"+        , ""+        , "    IF TG_OP IN ('INSERT', 'UPDATE') THEN"+        , "        RETURN NEW;"+        , "    ELSIF TG_OP = 'DELETE' THEN"+        , "        RETURN OLD;"+        , "    ELSE"+        , "        RETURN NULL;"+        , "    END IF;"+        , "END;"+        , "$$ LANGUAGE plpgsql;"+        ]++triggerSQL :: T.Text+triggerSQL =+    T.unlines+        [ "CREATE TRIGGER tg_pgqueuer_changed"+        , "AFTER INSERT OR UPDATE OR DELETE ON pgqueuer"+        , "FOR EACH ROW EXECUTE FUNCTION fn_pgqueuer_changed();"+        , "CREATE TRIGGER tg_pgqueuer_changed_truncate"+        , "AFTER TRUNCATE ON pgqueuer"+        , "FOR EACH STATEMENT EXECUTE FUNCTION fn_pgqueuer_changed();"+        ]
+ src/PGQueuer/Settings.hs view
@@ -0,0 +1,49 @@+{-# LANGUAGE OverloadedStrings #-}++module PGQueuer.Settings (+    DBSettings (..),+    defaultDBSettings,+    queueTable,+    queueTableLog,+    statisticsTable,+    schedulesTable,+    queueStatusType,+    function,+    trigger,+) where++import Data.Text (Text)+import PGQueuer.Types (Channel, defaultChannel)++data DBSettings = DBSettings+    { dbPrefix :: Text+    , dbChannel :: Channel+    }++defaultDBSettings :: DBSettings+defaultDBSettings =+    DBSettings+        { dbPrefix = ""+        , dbChannel = defaultChannel+        }++queueTable :: DBSettings -> Text+queueTable settings = dbPrefix settings <> "pgqueuer"++queueTableLog :: DBSettings -> Text+queueTableLog settings = dbPrefix settings <> "pgqueuer_log"++statisticsTable :: DBSettings -> Text+statisticsTable settings = dbPrefix settings <> "pgqueuer_statistics"++schedulesTable :: DBSettings -> Text+schedulesTable settings = dbPrefix settings <> "pgqueuer_schedules"++queueStatusType :: DBSettings -> Text+queueStatusType settings = dbPrefix settings <> "pgqueuer_status"++function :: DBSettings -> Text+function settings = dbPrefix settings <> "fn_pgqueuer_changed"++trigger :: DBSettings -> Text+trigger settings = dbPrefix settings <> "tg_pgqueuer_changed"
+ src/PGQueuer/Types.hs view
@@ -0,0 +1,266 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings #-}++module PGQueuer.Types (+    JobId (..),+    ScheduleId (..),+    Entrypoint (..),+    Channel (..),+    EntrypointExecutionParameter (..),+    Job (..),+    JobStatus (..),+    QueueStatistics (..),+    OnFailure (..),+    onFailureToText,+    textToOnFailure,+    defaultHeartbeatTimeout,+    defaultBatchSize,+    jobStatusToText,+    defaultChannel,+) where++import Data.Aeson (Value)+import qualified Data.ByteString.Lazy as BL+import Data.Maybe (fromMaybe)+import Data.Text (Text)+import Data.Time (UTCTime)+import Data.UUID (UUID)+import Database.PostgreSQL.Simple.FromField (FromField (..))+import Database.PostgreSQL.Simple.FromRow (FromRow (..), field)+import Database.PostgreSQL.Simple.ToField (ToField (..))+import GHC.Generics (Generic)++newtype JobId = JobId Int+    deriving stock (Show, Eq, Ord, Generic)+    deriving newtype (FromField, ToField)++newtype ScheduleId = ScheduleId Int+    deriving stock (Show, Eq, Ord, Generic)+    deriving newtype (FromField, ToField)++newtype Entrypoint = Entrypoint Text+    deriving stock (Show, Eq, Ord, Generic)+    deriving newtype (FromField, ToField)++newtype CronExpression = CronExpression Text+    deriving stock (Show, Eq, Ord, Generic)+    deriving newtype (FromField, ToField)++newtype Channel = Channel Text+    deriving stock (Show, Eq, Ord, Generic)+    deriving newtype (FromField, ToField)++data JobStatus+    = Queued+    | Picked+    | Successful+    | Failed+    | Exception+    | Canceled+    | Deleted+    deriving stock (Show, Eq, Ord, Generic, Bounded, Enum)++jobStatusToText :: JobStatus -> Text+jobStatusToText Queued = "queued"+jobStatusToText Picked = "picked"+jobStatusToText Successful = "successful"+jobStatusToText Failed = "failed"+jobStatusToText Exception = "exception"+jobStatusToText Canceled = "canceled"+jobStatusToText Deleted = "deleted"++textToJobStatus :: Text -> Maybe JobStatus+textToJobStatus "queued" = Just Queued+textToJobStatus "picked" = Just Picked+textToJobStatus "successful" = Just Successful+textToJobStatus "failed" = Just Failed+textToJobStatus "exception" = Just Exception+textToJobStatus "canceled" = Just Canceled+textToJobStatus "deleted" = Just Deleted+textToJobStatus _ = Nothing++instance FromField JobStatus where+    fromField f v = fromField f v >>= \t -> return (fromMaybe Queued (textToJobStatus t))++instance ToField JobStatus where+    toField = toField . jobStatusToText++data Operation+    = Insert+    | Update+    | Delete+    | Truncate+    deriving stock (Show, Eq, Ord, Generic, Bounded, Enum)++data Event+    = TableChangedEvent+        { eventChannel :: Channel+        , eventSentAt :: UTCTime+        , eventOperation :: Operation+        , eventTable :: Text+        }+    | CancellationEvent+        { eventChannel :: Channel+        , eventSentAt :: UTCTime+        , eventIds :: [JobId]+        }+    | HealthCheckEvent+        { eventChannel :: Channel+        , eventSentAt :: UTCTime+        , eventId :: UUID+        }+    deriving stock (Show, Generic)++data Job = Job+    { jobId :: JobId+    , jobPriority :: Int+    , jobCreated :: UTCTime+    , jobUpdated :: UTCTime+    , jobHeartbeat :: UTCTime+    , jobExecuteAfter :: UTCTime+    , jobStatus :: JobStatus+    , jobEntrypoint :: Entrypoint+    , jobPayload :: Maybe BL.ByteString+    , jobAttempts :: Int+    , jobQueueManagerId :: Maybe UUID+    , jobHeaders :: Maybe Value+    }+    deriving stock (Show, Eq, Generic)++data LogEntry = LogEntry+    { logCreated :: UTCTime+    , logJobId :: JobId+    , logStatus :: JobStatus+    , logPriority :: Int+    , logEntrypoint :: Entrypoint+    , logTraceback :: Maybe Value+    , logAggregated :: Bool+    }+    deriving stock (Show, Eq, Generic)++data QueueStatistics = QueueStatistics+    { statsCount :: Int+    , statsEntrypoint :: Entrypoint+    , statsPriority :: Int+    , statsStatus :: JobStatus+    }+    deriving stock (Show, Eq, Generic)++data LogStatistics = LogStatistics+    { logStatsCount :: Int+    , logStatsCreated :: UTCTime+    , logStatsEntrypoint :: Entrypoint+    , logStatsPriority :: Int+    , logStatsStatus :: JobStatus+    }+    deriving stock (Show, Eq, Generic)++data Schedule = Schedule+    { scheduleId :: ScheduleId+    , scheduleExpression :: CronExpression+    , scheduleEntrypoint :: Entrypoint+    , scheduleHeartbeat :: UTCTime+    , scheduleCreated :: UTCTime+    , scheduleUpdated :: UTCTime+    , scheduleNextRun :: UTCTime+    , scheduleLastRun :: Maybe UTCTime+    , scheduleStatus :: JobStatus+    }+    deriving stock (Show, Eq, Generic)++data TracebackRecord = TracebackRecord+    { traceJobId :: JobId+    , traceTimestamp :: UTCTime+    , traceExceptionType :: Text+    , traceExceptionMessage :: Text+    , traceTraceback :: Text+    }+    deriving stock (Show, Eq, Generic)++data OnFailure+    = OnDelete+    | OnHold+    deriving stock (Show, Eq, Ord, Generic, Bounded, Enum)++onFailureToText :: OnFailure -> Text+onFailureToText OnDelete = "delete"+onFailureToText OnHold = "hold"++textToOnFailure :: Text -> Maybe OnFailure+textToOnFailure "delete" = Just OnDelete+textToOnFailure "hold" = Just OnHold+textToOnFailure _ = Nothing++data EntrypointExecutionParameter = EntrypointExecutionParameter+    { paramEntrypoint :: Entrypoint+    , paramConcurrencyLimit :: Int+    }+    deriving stock (Show, Eq, Generic)++defaultChannel :: Channel+defaultChannel = Channel "ch_pgqueuer"++defaultBatchSize :: Int+defaultBatchSize = 100++defaultHeartbeatTimeout :: Int+defaultHeartbeatTimeout = 300 -- 5 minutes in seconds++instance FromRow Job where+    fromRow =+        Job+            <$> field -- jobId+            <*> field -- jobPriority+            <*> field -- jobCreated+            <*> field -- jobUpdated+            <*> field -- jobHeartbeat+            <*> field -- jobExecuteAfter+            <*> field -- jobStatus+            <*> field -- jobEntrypoint+            <*> field -- jobPayload+            <*> field -- jobAttempts+            <*> field -- jobQueueManagerId+            <*> field -- jobHeaders++instance FromRow LogEntry where+    fromRow =+        LogEntry+            <$> field -- logCreated+            <*> field -- logJobId+            <*> field -- logStatus+            <*> field -- logPriority+            <*> field -- logEntrypoint+            <*> field -- logTraceback+            <*> field -- logAggregated++instance FromRow QueueStatistics where+    fromRow =+        QueueStatistics+            <$> field -- statsCount+            <*> field -- statsEntrypoint+            <*> field -- statsPriority+            <*> field -- statsStatus++instance FromRow LogStatistics where+    fromRow =+        LogStatistics+            <$> field -- logStatsCount+            <*> field -- logStatsCreated+            <*> field -- logStatsEntrypoint+            <*> field -- logStatsPriority+            <*> field -- logStatsStatus++instance FromRow Schedule where+    fromRow =+        Schedule+            <$> field -- scheduleId+            <*> field -- scheduleExpression+            <*> field -- scheduleEntrypoint+            <*> field -- scheduleHeartbeat+            <*> field -- scheduleCreated+            <*> field -- scheduleUpdated+            <*> field -- scheduleNextRun+            <*> field -- scheduleLastRun+            <*> field -- scheduleStatus
+ test/Spec.hs view
@@ -0,0 +1,4 @@+import Tests.PGQueuer++main :: IO ()+main = runTests
+ test/Tests/PGQueuer.hs view
@@ -0,0 +1,187 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE OverloadedStrings #-}++module Tests.PGQueuer (runTests) where++import Test.Tasty+import Test.Tasty.HUnit++import Data.Aeson+import Data.Either (isLeft, isRight)+import qualified Data.Map.Strict as Map+import Data.Maybe (fromMaybe, isJust, listToMaybe)+import Data.UUID.V4 (nextRandom)+import GHC.Generics+import PGQueuer++runTests :: IO ()+runTests =+    defaultMain $+        dependentTestGroup+            "All tests"+            AllFinish+            [ schemaTests+            , entrypointTests+            , enqueueDequeueJobs+            , enqueueDequeueMultipleJobs+            , logAndListJobs+            , roundTripJSONPayload+            ]++withFreshQueue :: (QueueManager -> IO a) -> IO a+withFreshQueue action = do+    let conStr = "postgresql://queue_user:queue_pass@localhost:5432/queue_db"+    let dbSetting = defaultDBSettings+    queueMgrId <- nextRandom+    withQueueManager conStr dbSetting queueMgrId $ \qm -> do+        uninstallSchema qm+        installSchema qm+        action qm++expectOne :: String -> [a] -> (a -> Assertion) -> Assertion+expectOne message values assertion =+    case listToMaybe values of+        Nothing -> assertFailure message+        Just value -> assertion value++schemaTests :: TestTree+schemaTests = do+    testGroup "Schema related tests" [checkInstallUninstallSchema]++checkInstallUninstallSchema :: TestTree+checkInstallUninstallSchema = testCase "installed schema should should be deleted safely" $ do+    withFreshQueue $ \qm -> do+        schemaInstalled <- verifyStructure qm+        assertBool "Schema is installed and verified" (isRight schemaInstalled)+        uninstallSchema qm+        schemaInstalled2 <- verifyStructure qm+        assertBool "Schema is removed and verified" (isLeft schemaInstalled2)++entrypointTests :: TestTree+entrypointTests = testCase "Adding an entrypoint in qm adds entrypoint" $ do+    withFreshQueue $ \qm -> do+        let ep = Entrypoint "hello"+        qm1 <- registerEntrypoint qm ep (\_ -> pure ())+        let entrypoints = qmEntrypoints qm1+        assertBool "qm should contain entrypoint hello" $ isJust (Map.lookup "hello" entrypoints)++enqueueDequeueJobs :: TestTree+enqueueDequeueJobs =+    testCase "Enqueue jobs should have entry in PGQ table" $ do+        withFreshQueue $ \qm -> do+            let ep = Entrypoint "hello"+            let params = [EntrypointExecutionParameter ep 0]+            qm1 <- registerEntrypoint qm ep (\_ -> pure ())+            _ <- enqueue qm1 ep Nothing 0 Nothing Nothing Nothing+            stats <- getQueueSize qm1+            expectOne "List is empty" stats $ \stat ->+                assertBool "Should contain exactly one entry" $ statsCount stat == 1++            pickedJobs <- dequeue qm1 20 params Nothing 3+            expectOne "dequeue returned no jobs" pickedJobs $ \job -> do+                assertEqual "dequeued job should be picked" Picked (jobStatus job)+                status <- listJobStatusById qm1 [jobId job]+                expectOne "job status lookup failed" status $ \(_, jobStatus') ->+                    assertEqual "job should be picked" Picked jobStatus'++enqueueDequeueMultipleJobs :: TestTree+enqueueDequeueMultipleJobs =+    testCase "Enqueue multiple jobs dequeue them" $ do+        withFreshQueue $ \qm -> do+            let ep = Entrypoint "hello"+            let params = [EntrypointExecutionParameter ep 0]+            qm1 <- registerEntrypoint qm ep (\_ -> pure ())+            _ <-+                enqueueMultiple+                    qm1+                    (replicate 10 ep)+                    (replicate 10 Nothing)+                    (replicate 10 0)+                    (replicate 10 Nothing)+                    (replicate 10 Nothing)+                    (replicate 10 Nothing)+            stats <- getQueueSize qm1+            expectOne "statistics are empty" stats $ \stat ->+                assertEqual "Should contain exactly 10 entries" 10 (statsCount stat)++            pickedJobs <- dequeue qm1 20 params Nothing 3+            assertEqual "Should dequeue 10 jobs" 10 (length pickedJobs)+            assertBool "All jobs should be picked" (all ((== Picked) . jobStatus) pickedJobs)+            stats1 <- getQueueSize qm1+            expectOne "statistics are empty" stats1 $ \stat ->+                assertEqual "picked jobs should be tracked" Picked (statsStatus stat)++logAndListJobs :: TestTree+logAndListJobs =+    testCase "Update job status, retry, requeue and clear queue" $ do+        withFreshQueue $ \qm -> do+            let ep = Entrypoint "hello"+            let params = [EntrypointExecutionParameter ep 0]+            qm1 <- registerEntrypoint qm ep (\_ -> pure ())+            jobIds <-+                enqueueMultiple+                    qm1+                    (replicate 2 ep)+                    (replicate 2 Nothing)+                    (replicate 2 0)+                    (replicate 2 Nothing)+                    (replicate 2 Nothing)+                    (replicate 2 Nothing)+            assertEqual "enqueueMultiple should return two job ids" 2 (length jobIds)+            pickedJobs <- dequeue qm1 2 params Nothing 3+            case pickedJobs of+                [job1, job2] -> do+                    updateHeartbeat qm1 [jobId job1]+                    heartbeatStatus <- listJobStatusById qm1 [jobId job1]+                    expectOne "heartbeat status lookup failed" heartbeatStatus $ \(_, status) ->+                        assertEqual "picked job should stay picked" Picked status++                    logJobs qm1 [(jobId job1, Failed, Nothing)]+                    failedJobs <- listFailedJobs qm1 10+                    expectOne "Need exactly one failed job" failedJobs $ \failedJob -> do+                        assertEqual "failed job id mismatch" (jobId job1) (jobId failedJob)+                        retryJob qm1 failedJob 0 Nothing++                    retryStatus <- listJobStatusById qm1 [jobId job1]+                    expectOne "retry status lookup failed" retryStatus $ \(_, status) ->+                        assertEqual "retried job should be queued" Queued status++                    requeueJobs qm1 [jobId job2]+                    requeueStatus <- listJobStatusById qm1 [jobId job2]+                    expectOne "requeue status lookup failed" requeueStatus $ \(_, status) ->+                        assertEqual "requeued job should be queued" Queued status++                    markJobAsCancelled qm1 [jobId job2]+                    cancelledStatus <- listJobStatusById qm1 [jobId job2]+                    expectOne "cancelled status lookup failed" cancelledStatus $ \(_, status) ->+                        assertEqual "cancelled job should be cancelled" Canceled status++                    clearQueue qm1 (Just [ep])+                    remaining <- getQueueSize qm1+                    assertBool "queue should be empty after clearQueue" (null remaining)+                _ -> assertFailure "There should be exactly two picked jobs"++data BasicType = BasicType+    { name :: String+    , age :: Int+    }+    deriving (Show, Eq, Generic, FromJSON, ToJSON)++roundTripJSONPayload :: TestTree+roundTripJSONPayload = do+    testCase "Enqueue jobs should have entry in PGQ table" $ do+        withFreshQueue $ \qm -> do+            let ep = Entrypoint "hello"+            let params = [EntrypointExecutionParameter ep 0]+            qm1 <- registerEntrypoint qm ep (\_ -> pure ())+            let payload = encode $ BasicType "Hello" 25+            _ <- enqueue qm1 ep (Just payload) 0 Nothing Nothing Nothing++            pickedJobs <- dequeue qm1 20 params Nothing 3+            expectOne "pickup job lookup failed" pickedJobs $ \job ->+                case decode (fromMaybe "{}" $ jobPayload job) of+                    Nothing -> assertFailure "decoding of payload failed"+                    Just x -> do+                        assertEqual "round trip name" "Hello" (name x)+                        assertEqual "round trip name" 25 (age x)