packages feed

pgmq-inspect (empty) → 0.7.0.0

raw patch · 64 files changed

+6880/−0 lines, 64 filesdep +aesondep +asyncdep +base

Dependencies added: aeson, async, base, bytestring, case-insensitive, containers, directory, effectful-core, ephemeral-pg, generic-lens, hasql, hasql-pool, hs-opentelemetry-api, http-client, http-types, lens, network, network-uri, openapi3, optparse-applicative, pg-migrate, pgmq-core, pgmq-effectful, pgmq-hasql, pgmq-inspect, pgmq-migration, postgresql-connection-string, postgresql-libpq, process, random, stm, streaming-commons, tasty, tasty-golden, tasty-hunit, text, time, unix, vector, wai, wai-cors, wai-websockets, warp, websockets

Files

+ CHANGELOG.md view
@@ -0,0 +1,33 @@+# Revision history for pgmq-inspect++## 0.7.0.0 -- 2026-10-26++Initial release with the six-package 0.7.0.0 family.++### New Features++- Mountable read-only HTTP inspection for non-destructive live/archive paging,+  lookup, metrics and bindings, with explicit-origin CORS, shared nonblocking+  admission and safe errors.+- Standalone `pgmq-inspect` executable with checked options, database URL environment+  fallback, CORS flags, actual bound-port reporting and isolated connections with+  finite statement/lock/connect/acquisition budgets. The CLI supports+  optparse-applicative 0.19; the library does not depend on the CLI parser.+- OpenAPI 3 at `/openapi.json`, advertised by the service descriptor. Relative links+  and server URLs support prefix mounts and the canonical trailing-slash root.+- WebSocket metrics subscriptions with shared per-queue refresh scheduling, bounded+  outboxes and admission, generation-safe snapshots, poll convergence, frame and+  connection caps, overflow signaling and scoped shutdown through+  `withInspectRuntime`.+- A runtime-wide LISTEN backend refcounts channels, bounds buffered batches,+  reconnects with backoff and restores LISTENs. Polling remains authoritative.+  `InspectRuntime` exposes separate listener health and per-channel acknowledgement.+  Graceful frame delivery shares one one-second window across runner/runtime cleanup.++### Other Changes++- Tests validate actual response schemas, exact int64 client operations and restricted+  role access. The inspection guide documents deployment, grants, routes, frames and+  measured workload limits; design note 021 retains the runtime contract and evidence.+- Ephemeral PostgreSQL roots are created/validated as owner-only (`0700`), existing+  owned roots are tightened, unsafe paths are rejected and registry contents survive.
+ LICENSE view
@@ -0,0 +1,20 @@+Copyright (c) 2025 Nadeem Bitar++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.
+ app/Main.hs view
@@ -0,0 +1,31 @@+module Main (main) where++import Control.Concurrent.Async (wait)+import Control.Exception (bracket)+import Data.Text qualified as T+import Data.Version (showVersion)+import Hasql.Pool qualified as Pool+import Options.Applicative (execParser)+import Paths_pgmq_inspect (version)+import Pgmq.Inspect+import PgmqInspectCli+import System.Environment (lookupEnv)+import System.Exit (ExitCode (..), exitWith)+import System.IO (BufferMode (..), hPutStrLn, hSetBuffering, stderr, stdout)++main :: IO ()+main = do+  hSetBuffering stdout LineBuffering+  opts <- execParser parserInfo+  fallback <- lookupEnv "PGMQ_INSPECT_DATABASE_URL"+  raw <- either failWith pure (resolveDatabaseUrl (databaseUrl opts) fallback)+  (cfg, settings) <- either failWith pure (standaloneConfig opts raw)+  bracket (Pool.acquire (standalonePoolConfig opts settings)) Pool.release $ \pool ->+    withInspectServer cfg (poolEnv pool (if noListener opts then Nothing else Just settings)) $ \running -> do+      let bind = host opts+          urlHost = if ':' `elem` bind && not ('[' `elem` bind) then "[" <> bind <> "]" else bind+      putStrLn ("pgmq-inspect " <> showVersion version <> " listening on http://" <> urlHost <> ":" <> show (serverPort running))+      wait (serverThread running)++failWith :: T.Text -> IO a+failWith message = hPutStrLn stderr ("pgmq-inspect: " <> T.unpack message) >> exitWith (ExitFailure 2)
+ app/PgmqInspectCli.hs view
@@ -0,0 +1,107 @@+-- | Private standalone configuration, shared with the executable's tests.+module PgmqInspectCli+  ( Options (..),+    parserInfo,+    resolveDatabaseUrl,+    standaloneConfig,+    standalonePoolConfig,+  )+where++import Control.Lens ((&), (.~))+import Data.Generics.Labels ()+import Data.List.NonEmpty (nonEmpty)+import Data.Text (Text)+import Data.Text qualified as T+import Data.Text.Encoding qualified as TE+import Data.Text.Read qualified as TR+import Data.Version (showVersion)+import Hasql.Connection.Settings qualified as Settings+import Hasql.Pool.Config qualified as PoolConfig+import Hasql.Session qualified as Session+import Options.Applicative+import Paths_pgmq_inspect (version)+import Pgmq.Inspect.Config+import PostgresqlConnectionString qualified as ConnInfo++-- Keep option arithmetic in Integer until checked conversion, including ms -> us.+data Options = Options+  { databaseUrl :: !(Maybe String),+    host :: !String,+    port :: !Integer,+    corsOrigins :: ![String],+    pollIntervalMs :: !Integer,+    poolSize :: !Integer,+    noListener :: !Bool+  }++parserInfo :: ParserInfo Options+parserInfo =+  info+    (optionsParser <**> helper <**> versionOption)+    ( fullDesc+        <> header "pgmq-inspect - read-only HTTP and WebSocket inspection of pgmq queues"+        <> progDesc "The pgmq schema must already be installed. Uses dedicated inspection connections with finite database deadlines."+        <> failureCode 2+    )+  where+    versionOption = infoOption (showVersion version) (long "version" <> help "Print the version and exit")++optionsParser :: Parser Options+optionsParser =+  Options+    <$> optional (strOption (long "database-url" <> metavar "CONNINFO" <> help "libpq connection string or URL; defaults to PGMQ_INSPECT_DATABASE_URL"))+    <*> strOption (long "host" <> metavar "HOST" <> value "127.0.0.1" <> showDefault <> help "Bind address")+    <*> option boundedInteger (long "port" <> metavar "PORT" <> value 9092 <> showDefault <> help "Bind port (0 chooses a free port)")+    <*> many (strOption (long "cors-origin" <> metavar "ORIGIN" <> help "Allowed HTTP(S) browser origin; repeatable; CORS disabled when absent"))+    <*> option boundedInteger (long "poll-interval-ms" <> metavar "MS" <> value 5000 <> showDefault <> help "Authoritative WebSocket polling interval")+    <*> option boundedInteger (long "pool-size" <> metavar "N" <> value 4 <> showDefault <> help "Dedicated inspection pool size (shared read admission defaults to 2)")+    <*> switch (long "no-listener" <> help "Disable LISTEN acceleration; keep the authoritative poll feed")++boundedInteger :: ReadM Integer+boundedInteger = eitherReader $ \raw ->+  if null raw || length raw > 128 || not (all (\c -> c >= '0' && c <= '9') raw)+    then Left "expected at most 128 decimal digits"+    else case TR.decimal (T.pack raw) of+      Right (number, rest) | T.null rest -> Right number+      _ -> Left "expected a decimal integer"++resolveDatabaseUrl :: Maybe String -> Maybe String -> Either Text Text+resolveDatabaseUrl explicit fromEnv = case explicit of+  Just raw -> Right (T.pack raw)+  Nothing -> case fromEnv of+    Just raw | not (null raw) -> Right (T.pack raw)+    _ -> Left "no database URL: pass --database-url or set PGMQ_INSPECT_DATABASE_URL"++standaloneConfig :: Options -> Text -> Either Text (InspectServerConfig, Settings.Settings)+standaloneConfig opts raw = do+  parsed <- either (const (Left badUrl)) Right (ConnInfo.parse raw)+  if parsed == mempty then Left badUrl else Right ()+  if port opts < 0 || port opts > 65535 then Left "--port must be in 0..65535" else Right ()+  if pollIntervalMs opts <= 0 || pollIntervalMs opts > toInteger (maxBound :: Int) `div` 1000 then Left "--poll-interval-ms is out of range" else Right ()+  if poolSize opts <= 0 || poolSize opts > toInteger (maxBound :: Int) then Left "--pool-size is out of range" else Right ()+  let cors = maybe CorsDisabled CorsAllowOrigins (nonEmpty (map (TE.encodeUtf8 . T.pack) (corsOrigins opts)))+      cfg = defaultInspectConfig & #corsPolicy .~ cors & #pollIntervalUs .~ fromInteger (pollIntervalMs opts * 1000)+  checked <- validateInspectConfig cfg+  let (existingOptions, withoutOptions) = maybe ("", parsed) id (ConnInfo.interceptParam "options" parsed)+      withoutConnectTimeout = maybe withoutOptions snd (ConnInfo.interceptParam "connect_timeout" withoutOptions)+      bounded =+        withoutConnectTimeout+          <> ConnInfo.param "options" (existingOptions <> " -c statement_timeout=2000 -c lock_timeout=250")+          <> ConnInfo.param "connect_timeout" "2"+      settings = Settings.connectionString (ConnInfo.toUrl bounded) <> Settings.applicationName "pgmq-inspect"+      server = defaultInspectServerConfig & #bindHost .~ host opts & #bindPort .~ fromInteger (port opts) & #inspect .~ checked+  pure (server, settings)+  where+    badUrl = "the database URL is empty or not a valid libpq connection string"++-- | Init SQL also defends the deadlines on every pooled connection. The startup+-- options above apply to the separate listener and preserve other supplied GUCs.+standalonePoolConfig :: Options -> Settings.Settings -> PoolConfig.Config+standalonePoolConfig opts settings =+  PoolConfig.settings+    [ PoolConfig.size (fromInteger (poolSize opts)),+      PoolConfig.acquisitionTimeout 1,+      PoolConfig.staticConnectionSettings settings,+      PoolConfig.initSession (Session.script "SET statement_timeout = '2000ms'; SET lock_timeout = '250ms'")+    ]
+ pgmq-inspect.cabal view
@@ -0,0 +1,259 @@+cabal-version: 3.4+name: pgmq-inspect+version: 0.7.0.0+synopsis: Embeddable HTTP and WebSocket inspection surface for PGMQ queues+description:+  A sister package to pgmq-hs: an embeddable WAI application and a convenience+  runner that expose read-only queue inspection over HTTP and WebSocket. Queue+  listings, total and visible depth metrics, non-destructive message and archive+  browsing, message lookup by id, topic bindings, routing tests, notification+  throttles, and health probes, with configurable CORS. The core pgmq-hs+  libraries gain no web dependency; this package depends on them, never the+  other way around.++homepage: https://github.com/shinzui/pgmq-hs+license: MIT+license-file: LICENSE+author: Nadeem Bitar+maintainer: Nadeem Bitar+category: Database, Web+build-type: Simple+extra-doc-files: CHANGELOG.md+extra-source-files:+  test/client/*.mjs+  test/fixtures/inspection-role.sql+  test/fixtures/pgmq-1.12.0.sql+  test/golden/*.json++common warnings+  ghc-options:+    -Wall+    -Wcompat+    -Widentities+    -Wincomplete-uni-patterns+    -Wincomplete-record-updates+    -Wredundant-constraints+    -fhide-source-paths+    -Wmissing-export-lists+    -Wpartial-fields+    -Wmissing-deriving-strategies++library+  import: warnings+  exposed-modules:+    Pgmq.Inspect+    Pgmq.Inspect.Config+    Pgmq.Inspect.Cors+    Pgmq.Inspect.Env+    Pgmq.Inspect.Error+    Pgmq.Inspect.Http+    Pgmq.Inspect.OpenApi+    Pgmq.Inspect.Server+    Pgmq.Inspect.WebSocket+    Pgmq.Inspect.Wire++  other-modules:+    Paths_pgmq_inspect+    Pgmq.Inspect.Budget+    Pgmq.Inspect.Frames+    Pgmq.Inspect.Lifecycle+    Pgmq.Inspect.Listener+    Pgmq.Inspect.Outbox+    Pgmq.Inspect.Refresh++  autogen-modules: Paths_pgmq_inspect+  default-extensions:+    DataKinds+    DeriveGeneric+    DuplicateRecordFields+    GeneralisedNewtypeDeriving+    ImportQualifiedPost+    LambdaCase+    NamedFieldPuns+    OverloadedLabels+    OverloadedStrings+    TypeOperators++  build-depends:+    aeson ^>=2.2,+    async ^>=2.2,+    base >=4.18 && <5,+    bytestring >=0.11 && <0.13,+    containers >=0.6 && <0.8,+    effectful-core ^>=2.6 || ^>=2.7,+    generic-lens ^>=2.2 || ^>=2.3,+    hasql ^>=1.10,+    hasql-pool ^>=1.4,+    hs-opentelemetry-api >=1.0 && <2,+    http-types ^>=0.12,+    lens ^>=5.3,+    network ^>=3.2,+    network-uri ^>=2.6,+    openapi3 ^>=3.2.5,+    pgmq-core >=0.7 && <0.8,+    pgmq-effectful >=0.7 && <0.8,+    pgmq-hasql >=0.7 && <0.8,+    postgresql-libpq >=0.10.1 && <0.12,+    stm >=2.5 && <2.6,+    streaming-commons ^>=0.2,+    text ^>=2.1,+    time ^>=1.14,+    vector ^>=0.13,+    wai ^>=3.2,+    wai-cors ^>=0.2.7,+    wai-websockets ^>=3.0,+    warp ^>=3.4,+    websockets ^>=0.13,++  hs-source-dirs: src+  default-language: GHC2024++executable pgmq-inspect+  import: warnings+  default-language: GHC2024+  hs-source-dirs: app+  main-is: Main.hs+  ghc-options:+    -threaded+    -rtsopts+    -with-rtsopts=-N++  autogen-modules: Paths_pgmq_inspect+  other-modules:+    Paths_pgmq_inspect+    PgmqInspectCli++  default-extensions:+    DeriveGeneric+    DuplicateRecordFields+    ImportQualifiedPost+    OverloadedLabels+    OverloadedStrings++  build-depends:+    async ^>=2.2,+    base >=4.18 && <5,+    generic-lens ^>=2.2 || ^>=2.3,+    hasql ^>=1.10,+    hasql-pool ^>=1.4,+    lens ^>=5.3,+    optparse-applicative ^>=0.19,+    pgmq-inspect >=0.7 && <0.8,+    postgresql-connection-string ^>=0.1,+    text ^>=2.1,++test-suite pgmq-inspect-test+  import: warnings+  default-language: GHC2024+  build-tool-depends: pgmq-inspect:pgmq-inspect >=0.7 && <0.8+  type: exitcode-stdio-1.0+  hs-source-dirs:+    test+    src+    app++  main-is: Main.hs+  ghc-options:+    -threaded+    -rtsopts+    -with-rtsopts=-N++  autogen-modules: Paths_pgmq_inspect+  other-modules:+    CliSpec+    ConfigSpec+    CorsSpec+    EmbeddingExample+    EphemeralDb+    ErrorSpec+    HttpSpec+    IsolationSpec+    ListenerSpec+    LiveFeedSpec+    OpenApiSpec+    OutboxSpec+    Paths_pgmq_inspect+    Pgmq.Inspect+    Pgmq.Inspect.Budget+    Pgmq.Inspect.Config+    Pgmq.Inspect.Cors+    Pgmq.Inspect.Env+    Pgmq.Inspect.Error+    Pgmq.Inspect.Frames+    Pgmq.Inspect.Http+    Pgmq.Inspect.Lifecycle+    Pgmq.Inspect.Listener+    Pgmq.Inspect.OpenApi+    Pgmq.Inspect.Outbox+    Pgmq.Inspect.Refresh+    Pgmq.Inspect.Server+    Pgmq.Inspect.WebSocket+    Pgmq.Inspect.Wire+    PgmqInspectCli+    PrivilegesSpec+    RawWebSocketSpec+    RefreshSpec+    RootPermissionsSpec+    RouterSpec+    ServerSpec+    SmokeSpec+    TestServer+    WebSocketSpec+    WireGoldenSpec++  default-extensions:+    DataKinds+    DeriveGeneric+    DuplicateRecordFields+    GeneralisedNewtypeDeriving+    ImportQualifiedPost+    LambdaCase+    NamedFieldPuns+    OverloadedLabels+    OverloadedStrings+    TypeOperators++  build-depends:+    aeson,+    async,+    base >=4.18 && <5,+    bytestring,+    case-insensitive,+    containers,+    directory,+    effectful-core ^>=2.6 || ^>=2.7,+    ephemeral-pg >=0.3.1.0,+    generic-lens,+    hasql,+    hasql-pool ^>=1.4,+    hs-opentelemetry-api >=1.0 && <2,+    http-client ^>=0.7,+    http-types,+    lens,+    network,+    network-uri ^>=2.6,+    openapi3 ^>=3.2.5,+    optparse-applicative ^>=0.19,+    pg-migrate,+    pgmq-core >=0.7 && <0.8,+    pgmq-effectful >=0.7 && <0.8,+    pgmq-hasql >=0.7 && <0.8,+    pgmq-migration >=0.7 && <0.8,+    postgresql-connection-string ^>=0.1,+    postgresql-libpq >=0.10.1 && <0.12,+    process,+    random ^>=1.2,+    stm,+    streaming-commons ^>=0.2,+    tasty ^>=1.5,+    tasty-golden ^>=2.3.6,+    tasty-hunit ^>=0.10,+    text,+    time,+    unix,+    vector,+    wai,+    wai-cors ^>=0.2.7,+    wai-websockets ^>=3.0,+    warp,+    websockets,
+ src/Pgmq/Inspect.hs view
@@ -0,0 +1,23 @@+-- | Umbrella re-export: everything a host needs to configure, build, and run+-- the inspection surface. See @docs/user/queue-inspection.md@ for the guide.+module Pgmq.Inspect+  ( module Pgmq.Inspect.Config,+    module Pgmq.Inspect.Env,+    module Pgmq.Inspect.Server,+    module Pgmq.Inspect.Error,+    module Pgmq.Inspect.Wire,+    module Pgmq.Inspect.WebSocket,+    openApiDocument,+    routeTable,+    PathSegment (..),+  )+where++import Pgmq.Inspect.Config+import Pgmq.Inspect.Env+import Pgmq.Inspect.Error+import Pgmq.Inspect.Http (PathSegment (..), routeTable)+import Pgmq.Inspect.OpenApi (openApiDocument)+import Pgmq.Inspect.Server+import Pgmq.Inspect.WebSocket+import Pgmq.Inspect.Wire
+ src/Pgmq/Inspect/Budget.hs view
@@ -0,0 +1,20 @@+-- | One runtime-wide nonblocking admission gate for HTTP and feed reads.+module Pgmq.Inspect.Budget (InspectionBudget, withInspectionBudget, withInspectionPermit) where++import Control.Concurrent.STM+import Control.Exception (bracket, finally, mask)+import Pgmq.Inspect.Config++newtype InspectionBudget = InspectionBudget (TVar Int)++withInspectionBudget :: InspectConfig -> (InspectionBudget -> IO a) -> IO a+withInspectionBudget cfg k = bracket (InspectionBudget <$> newTVarIO (maxConcurrentReads cfg)) (const (pure ())) k++-- | Reject promptly when full; release on success, failure and cancellation.+-- No retries or detached work are created here.+withInspectionPermit :: InspectionBudget -> IO a -> IO (Maybe a)+withInspectionPermit (InspectionBudget slots) action = mask $ \restore -> do+  acquired <- atomically $ do+    n <- readTVar slots+    if n <= 0 then pure False else writeTVar slots (n - 1) >> pure True+  if acquired then Just <$> (restore action `finally` atomically (modifyTVar' slots (+ 1))) else pure Nothing
+ src/Pgmq/Inspect/Config.hs view
@@ -0,0 +1,124 @@+-- | Configuration for the inspection surface and for the convenience runner.+--+-- The surface itself is configured by 'InspectConfig'; where it is served+-- from is 'InspectServerConfig'. A host that mounts the 'Network.Wai.Application'+-- into its own server needs only the former.+module Pgmq.Inspect.Config+  ( CorsPolicy (..),+    InspectConfig (..),+    defaultInspectConfig,+    validateInspectConfig,+    InspectServerConfig (..),+    defaultInspectServerConfig,+  )+where++import Data.ByteString (ByteString)+import Data.ByteString.Char8 qualified as BS+import Data.Int (Int32)+import Data.List.NonEmpty (NonEmpty)+import Data.Text (Text)+import GHC.Generics (Generic)+import Network.URI (URI (..), URIAuth (..), parseURI)+import Numeric.Natural (Natural)++-- | Which browser origins may call the surface cross-origin.+--+-- There is deliberately no \"any origin\" constructor and credentials are+-- never allowed: the browser forbids a wildcard origin together with+-- credentials, and an inspection surface that assumes a trusted network has+-- no business opting into either. List the origins you serve a browser UI+-- from, for example @CorsAllowOrigins ("http://localhost:5173" :| [])@.+data CorsPolicy+  = -- | Send no CORS headers at all. The default.+    CorsDisabled+  | -- | Answer requests whose @Origin@ header is exactly one of these with the+    -- CORS headers that let the page read the response; serve any other origin+    -- normally but without those headers.+    CorsAllowOrigins !(NonEmpty ByteString)+  deriving stock (Eq, Show, Generic)++-- | Behaviour of the surface. See 'defaultInspectConfig' for the defaults.+data InspectConfig = InspectConfig+  { corsPolicy :: !CorsPolicy,+    -- | Page size when a request omits @limit@.+    defaultPageLimit :: !Int32,+    -- | Largest page a request may ask for; larger values are capped to this.+    maxPageLimit :: !Int32,+    -- | Budget for the readiness probe's round trip, in microseconds.+    readinessTimeoutUs :: !Int,+    -- | WebSocket feed (see "Pgmq.Inspect.WebSocket"): how often each subscribed+    -- queue is polled for metrics, in microseconds.+    pollIntervalUs :: !Int,+    -- | WebSocket feed: how long to coalesce insert notifications for one+    -- queue before reading its metrics, in microseconds.+    notifyDebounceUs :: !Int,+    -- | WebSocket feed: concurrent connections accepted before rejecting.+    wsMaxConnections :: !Int,+    -- | WebSocket feed: queues one connection may watch at once.+    wsMaxSubscriptions :: !Int,+    -- | WebSocket feed: frames buffered per connection before the oldest are+    -- dropped and an in-band overflow error is sent.+    wsQueueCapacity :: !Natural,+    maxConcurrentReads :: !Int,+    metricsMinRefreshUs :: !Int,+    wsMaxQueues :: !Int,+    wsMaxFrameBytes :: !Int+  }+  deriving stock (Eq, Show, Generic)++-- | CORS disabled, pages of 50 capped at 500, a one-second readiness budget,+-- a five-second poll, a 100 ms notify debounce, 100 connections, 100+-- subscriptions per connection, 256 buffered frames.+defaultInspectConfig :: InspectConfig+defaultInspectConfig =+  InspectConfig+    { corsPolicy = CorsDisabled,+      defaultPageLimit = 50,+      maxPageLimit = 500,+      readinessTimeoutUs = 1_000_000,+      pollIntervalUs = 5_000_000,+      notifyDebounceUs = 100_000,+      wsMaxConnections = 100,+      wsMaxSubscriptions = 100,+      wsQueueCapacity = 256,+      maxConcurrentReads = 2,+      metricsMinRefreshUs = 1_000_000,+      wsMaxQueues = 100,+      wsMaxFrameBytes = 16_384+    }++-- | Where the convenience runner listens.+data InspectServerConfig = InspectServerConfig+  { -- | Interface to bind. Loopback by default; bind @*@ only behind an+    -- authenticating reverse proxy or on a trusted network.+    bindHost :: !String,+    -- | Port to bind. @0@ asks for a free port on the configured bindHost.+    bindPort :: !Int,+    inspect :: !InspectConfig+  }+  deriving stock (Eq, Show, Generic)++-- | @127.0.0.1:9092@ with 'defaultInspectConfig'. shibuya-metrics defaults to+-- 9090 and kiroku-metrics to 9091, so the three surfaces coexist on one host.+defaultInspectServerConfig :: InspectServerConfig+defaultInspectServerConfig =+  InspectServerConfig {bindHost = "127.0.0.1", bindPort = 9092, inspect = defaultInspectConfig}++-- | Check limits before allocating sockets, queues or threads.+validateInspectConfig :: InspectConfig -> Either Text InspectConfig+validateInspectConfig cfg+  | defaultPageLimit cfg < 1 || defaultPageLimit cfg > maxPageLimit cfg = Left "defaultPageLimit must be positive and no greater than maxPageLimit"+  | maxPageLimit cfg == maxBound = Left "maxPageLimit must leave room for one lookahead row"+  | any (<= 0) [readinessTimeoutUs cfg, pollIntervalUs cfg, notifyDebounceUs cfg, wsMaxConnections cfg, wsMaxSubscriptions cfg, maxConcurrentReads cfg, metricsMinRefreshUs cfg, wsMaxQueues cfg, wsMaxFrameBytes cfg] = Left "intervals and capacity limits must be positive"+  | wsQueueCapacity cfg == 0 = Left "wsQueueCapacity must be positive"+  | CorsAllowOrigins origins <- corsPolicy cfg, not (all validOrigin origins) = Left "CORS origins must be explicit HTTP(S) origins without credentials, path, query or fragment"+  | otherwise = Right cfg+  where+    validOrigin raw = case parseURI (BS.unpack raw) of+      Just URI {uriScheme, uriAuthority = Just URIAuth {uriUserInfo, uriRegName, uriPort}, uriPath, uriQuery, uriFragment} ->+        uriScheme `elem` ["http:", "https:"] && null uriUserInfo && not (null uriRegName) && uriRegName /= "*" && null uriPath && null uriQuery && null uriFragment && validPort uriPort+      _ -> False+    validPort "" = True+    validPort (':' : digits) = not (null digits) && length digits <= 5 && all (\c -> c >= '0' && c <= '9') digits && (read digits :: Int) <= 65535+    validPort _ = False
+ src/Pgmq/Inspect/Cors.hs view
@@ -0,0 +1,31 @@+-- | CORS (Cross-Origin Resource Sharing) support: the browser mechanism that+-- lets a page served from one origin read responses from another. Off by+-- default; on only for an explicit list of origins; never with credentials.+module Pgmq.Inspect.Cors (corsMiddleware) where++import Data.List.NonEmpty qualified as NE+import Network.Wai (Middleware)+import Network.Wai.Middleware.Cors+  ( CorsResourcePolicy (..),+    cors,+    simpleCorsResourcePolicy,+  )+import Pgmq.Inspect.Config (CorsPolicy (..))++-- | 'CorsDisabled' adds no headers and intercepts nothing. 'CorsAllowOrigins'+-- answers preflight @OPTIONS@ requests from a listed origin and adds+-- @Access-Control-Allow-Origin@ (echoing the origin, with @Vary: Origin@) to+-- actual responses; a request whose @Origin@ is not listed is served normally+-- with no CORS headers, which the browser then refuses to expose to the page.+corsMiddleware :: CorsPolicy -> Middleware+corsMiddleware CorsDisabled = id+corsMiddleware (CorsAllowOrigins origins) = cors (const (Just policy))+  where+    policy =+      simpleCorsResourcePolicy+        { corsOrigins = Just (NE.toList origins, False),+          corsMethods = ["GET", "OPTIONS"],+          corsRequestHeaders = ["Content-Type"],+          corsVaryOrigin = True,+          corsIgnoreFailures = True+        }
+ src/Pgmq/Inspect/Env.hs view
@@ -0,0 +1,60 @@+-- | How the surface reaches the database: a runner for the @Pgmq@ effect,+-- supplied by the host, plus the connection settings the WebSocket feed's+-- listener uses.+module Pgmq.Inspect.Env+  ( Inspection,+    InspectEnv (..),+    runnerFromInterpreter,+    poolEnv,+    tracedPoolEnv,+  )+where++import Effectful (Eff, IOE, runEff)+import Effectful.Error.Static (Error, runError)+import Hasql.Connection.Settings (Settings)+import Hasql.Pool (Pool)+import OpenTelemetry.Trace.Core qualified as OTel+import Pgmq.Effectful (Pgmq, PgmqRuntimeError, runPgmq, runPgmqTraced)++-- | A database program the surface wants run: every handler is one of these.+type Inspection a = Eff '[Pgmq, Error PgmqRuntimeError, IOE] a++-- | The host's side of the contract.+data InspectEnv = InspectEnv+  { -- | Run one program. Which interpreter does so is the host's choice+    -- ('poolEnv' for the plain one, 'tracedPoolEnv' for the OpenTelemetry one,+    -- or anything else, including a mock for tests).+    runInspection :: forall a. Inspection a -> IO (Either PgmqRuntimeError a),+    -- | Settings for the dedicated connection the WebSocket feed LISTENs on+    -- (see "Pgmq.Inspect.WebSocket"). 'Nothing' makes the feed poll-only.+    listenerSettings :: !(Maybe Settings)+  }++-- | Turn an interpreter of the @Pgmq@ effect into a runner, discarding the+-- call stack that 'runError' pairs with the error.+runnerFromInterpreter ::+  (forall a. Eff '[Pgmq, Error PgmqRuntimeError, IOE] a -> Eff '[Error PgmqRuntimeError, IOE] a) ->+  (forall a. Inspection a -> IO (Either PgmqRuntimeError a))+runnerFromInterpreter interpretPgmq action =+  either (Left . snd) Right <$> runEff (runError @PgmqRuntimeError (interpretPgmq action))++-- | The plain interpreter over a dedicated inspection pool. The host must set+-- finite PostgreSQL statement_timeout and lock_timeout, plus acquisition/connect+-- budgets. Never borrow worker capacity. The admission cap limits live work,+-- but hasql's blackholed-reply limitation (IR-6) still applies.+poolEnv :: Pool -> Maybe Settings -> InspectEnv+poolEnv pool settings =+  InspectEnv+    { runInspection = runnerFromInterpreter (runPgmq pool),+      listenerSettings = settings+    }++-- | The OpenTelemetry-traced interpreter over a connection pool: every+-- handler's database work becomes a span under the host's tracer.+tracedPoolEnv :: Pool -> OTel.Tracer -> Maybe Settings -> InspectEnv+tracedPoolEnv pool tracer settings =+  InspectEnv+    { runInspection = runnerFromInterpreter (runPgmqTraced pool tracer),+      listenerSettings = settings+    }
+ src/Pgmq/Inspect/Error.hs view
@@ -0,0 +1,125 @@+-- | The error envelope and the mapping from the library's typed runtime error+-- onto HTTP statuses and codes.+--+-- Every error body is @{"error":{"code":…,"message":…,"details":…}}@. @code@+-- is a stable snake_case identifier a client may switch on; @message@ is a+-- human sentence safe to show verbatim and free to change; @details@ is+-- optional structured context. The vocabulary is listed in+-- @docs/design/021-inspection-surface-wire-contract.md@.+module Pgmq.Inspect.Error+  ( ErrorBody (..),+    inspectionBusy,+    errorResponse,+    jsonResponse,+    notFound,+    methodNotAllowed,+    queueNotFound,+    messageNotFound,+    invalidCursor,+    invalidLimit,+    invalidMessageId,+    invalidRoutingKey,+    websocketUpgradeRequired,+    websocketUnavailable,+    sqlState,+    describeRuntimeError,+    classifyRuntimeError,+  )+where++import Data.Aeson (ToJSON (..), Value, encode, object, (.=))+import Data.ByteString.Lazy (ByteString)+import Data.Text (Text)+import GHC.Generics (Generic)+import Hasql.Errors qualified as HasqlErrors+import Network.HTTP.Types (Status, hContentType, status500, status503)+import Network.Wai (Response, responseLBS)+import Pgmq.Effectful (PgmqRuntimeError (..), isTransient)++data ErrorBody = ErrorBody+  { errorCode :: !Text,+    errorMessage :: !Text,+    errorDetails :: !(Maybe Value)+  }+  deriving stock (Eq, Show, Generic)++instance ToJSON ErrorBody where+  toJSON e =+    object+      [ "error"+          .= object+            ( ["code" .= errorCode e, "message" .= errorMessage e]+                <> maybe [] (\d -> ["details" .= d]) (errorDetails e)+            )+      ]++jsonResponse :: Status -> ByteString -> Response+jsonResponse status = responseLBS status [(hContentType, "application/json")]++errorResponse :: Status -> ErrorBody -> Response+errorResponse status = jsonResponse status . encode++notFound, methodNotAllowed, websocketUpgradeRequired, websocketUnavailable :: ErrorBody+notFound = ErrorBody "not_found" "No route matches this path." Nothing+methodNotAllowed = ErrorBody "method_not_allowed" "Only GET is served." Nothing+websocketUpgradeRequired =+  ErrorBody "websocket_upgrade_required" "This path serves WebSocket upgrades only." Nothing+websocketUnavailable =+  ErrorBody "websocket_unavailable" "The WebSocket feed is not available on this server." Nothing++queueNotFound :: Text -> ErrorBody+queueNotFound name =+  ErrorBody "queue_not_found" "No queue with that name exists." (Just (object ["queue_name" .= name]))++messageNotFound :: Text -> Text -> ErrorBody+messageNotFound name raw =+  ErrorBody+    "message_not_found"+    "No message with that id exists in that table."+    (Just (object ["queue_name" .= name, "msg_id" .= raw]))++invalidCursor, invalidLimit, invalidMessageId, invalidRoutingKey :: Text -> ErrorBody+invalidCursor raw =+  ErrorBody "invalid_cursor" "from must be a non-negative integer msg_id." (Just (object ["from" .= raw]))+invalidLimit raw =+  ErrorBody "invalid_limit" "limit must be a positive integer." (Just (object ["limit" .= raw]))+invalidMessageId raw =+  ErrorBody "invalid_message_id" "msg_id must be a non-negative integer." (Just (object ["msg_id" .= raw]))+invalidRoutingKey raw =+  ErrorBody "invalid_routing_key" "routing_key is not a valid pgmq routing key." (Just (object ["routing_key" .= raw]))++-- | The SQLSTATE the server reported, when the error is a server-reported+-- statement error. Mirrors the pattern 'isTransient' reads.+sqlState :: PgmqRuntimeError -> Maybe Text+sqlState = \case+  PgmqSessionError+    ( HasqlErrors.StatementSessionError+        _+        _+        _+        _+        _+        (HasqlErrors.ServerStatementError (HasqlErrors.ServerError code _ _ _ _))+      ) -> Just code+  PgmqSessionError (HasqlErrors.ScriptSessionError _ (HasqlErrors.ServerError code _ _ _ _)) -> Just code+  _ -> Nothing++-- | Display-safe database failure. Raw driver diagnostics belong in host-side logs.+describeRuntimeError :: PgmqRuntimeError -> (Text, Maybe Value)+describeRuntimeError err =+  ( "The database could not complete the inspection request.",+    fmap (\code -> object ["sqlstate" .= code]) (sqlState err)+  )++-- | Queue identity is established in the handler, not inferred from 42P01 here.+classifyRuntimeError :: PgmqRuntimeError -> (Status, ErrorBody)+classifyRuntimeError err+  | sqlState err == Just "57014" = (status503, ErrorBody "inspection_timeout" "The inspection query exceeded its configured database deadline." details)+  | isTransient err = (status503, ErrorBody "database_unavailable" "The database is temporarily unavailable; retry the request." details)+  | otherwise = (status500, ErrorBody "database_error" message details)+  where+    (message, details) = describeRuntimeError err++-- | The runtime's shared read capacity is occupied; retry later.+inspectionBusy :: ErrorBody+inspectionBusy = ErrorBody "inspection_busy" "Inspection capacity is busy; retry the request." Nothing
+ src/Pgmq/Inspect/Frames.hs view
@@ -0,0 +1,58 @@+-- | Internal protocol representation. Public constructors are re-exported by WebSocket.+module Pgmq.Inspect.Frames (ClientFrame (..), ServerFrame (..), UpdateSource (..)) where++import Data.Aeson+import Data.Aeson.Types (Parser)+import Data.Text (Text)+import Pgmq.Hasql.Statements.Types (QueueMetrics)++data ClientFrame = Subscribe !Text | Unsubscribe !Text | Ping+  deriving stock (Eq, Show)++data UpdateSource = NotifySource | PollSource+  deriving stock (Eq, Show)++data ServerFrame+  = Snapshot !Text !QueueMetrics !Bool+  | Update !Text !QueueMetrics !UpdateSource+  | Pong+  | ErrorFrame !Text !Text !(Maybe Text)+  | Goodbye+  deriving stock (Show)++instance FromJSON ClientFrame where+  parseJSON = withObject "ClientFrame" $ \o -> do+    tag <- o .: "type" :: Parser Text+    case tag of+      "subscribe" -> Subscribe <$> o .: "queue"+      "unsubscribe" -> Unsubscribe <$> o .: "queue"+      "ping" -> pure Ping+      _ -> fail "unknown frame type"++instance ToJSON ClientFrame where+  toJSON = object . clientFields+  toEncoding = pairs . mconcat . clientFields++clientFields :: (KeyValue e kv) => ClientFrame -> [kv]+clientFields (Subscribe q) = ["type" .= ("subscribe" :: Text), "queue" .= q]+clientFields (Unsubscribe q) = ["type" .= ("unsubscribe" :: Text), "queue" .= q]+clientFields Ping = ["type" .= ("ping" :: Text)]++instance ToJSON UpdateSource where+  toJSON = toJSON . sourceText+  toEncoding = toEncoding . sourceText++sourceText :: UpdateSource -> Text+sourceText NotifySource = "notify"+sourceText PollSource = "poll"++instance ToJSON ServerFrame where+  toJSON = object . serverFields+  toEncoding = pairs . mconcat . serverFields++serverFields :: (KeyValue e kv) => ServerFrame -> [kv]+serverFields (Snapshot q metrics push) = ["type" .= ("snapshot" :: Text), "queue" .= q, "metrics" .= metrics, "push" .= push]+serverFields (Update q metrics source) = ["type" .= ("update" :: Text), "queue" .= q, "metrics" .= metrics, "source" .= source]+serverFields Pong = ["type" .= ("pong" :: Text)]+serverFields (ErrorFrame code message queue) = ["type" .= ("error" :: Text), "code" .= code, "message" .= message, "queue" .= queue]+serverFields Goodbye = ["type" .= ("goodbye" :: Text)]
+ src/Pgmq/Inspect/Http.hs view
@@ -0,0 +1,291 @@+-- | The HTTP router: one case per route, matching on 'pathInfo' so the+-- application behaves identically at the root and under a host's prefix.+module Pgmq.Inspect.Http+  ( PathSegment (..),+    routeTable,+    httpApp,+  )+where++import Control.Lens ((^.))+import Data.Aeson (ToJSON, encode)+import Data.Generics.Labels ()+import Data.Int (Int32, Int64)+import Data.List (find)+import Data.Text (Text)+import Data.Text qualified as T+import Data.Text.Encoding qualified as TE+import Data.Text.Read qualified as TR+import Data.Vector (Vector)+import Data.Vector qualified as V+import Data.Version (showVersion)+import GHC.Clock (getMonotonicTimeNSec)+import Network.HTTP.Types+  ( Method,+    methodGet,+    mkStatus,+    status200,+    status400,+    status404,+    status405,+    status503,+  )+import Network.Wai (Application, Request, Response, pathInfo, queryString, requestMethod)+import Network.Wai.Handler.WebSockets qualified as WaiWS+import Network.WebSockets qualified as WS+import Paths_pgmq_inspect (version)+import Pgmq.Effectful (allQueueMetrics, listNotifyInsertThrottles, listTopicBindings, testRouting)+import Pgmq.Effectful.Effect (listQueuesUnvalidated, lookupArchivedMessage, lookupMessage, peekArchivedMessages, peekMessages, queueMetricsUnvalidated)+import Pgmq.Hasql.Statements.Types (LookupMessage (..), PeekMessages (..))+import Pgmq.Inspect.Budget+import Pgmq.Inspect.Config (InspectConfig (..))+import Pgmq.Inspect.Env (InspectEnv (..), Inspection)+import Pgmq.Inspect.Error+import Pgmq.Inspect.OpenApi (openApiJson)+import Pgmq.Inspect.Wire+import Pgmq.Types (ArchivedMessage (..), MessageId (..), TopicBinding (..), UnvalidatedQueue (..), parseRoutingKey)+import System.Timeout (timeout)++data PathSegment = Literal !Text | Placeholder !Text+  deriving stock (Eq, Show)++-- | Every route the router serves. Plan 31's OpenAPI coverage test reads it.+routeTable :: [(Method, [PathSegment])]+routeTable =+  [ (methodGet, []),+    (methodGet, [Literal "openapi.json"]),+    (methodGet, [Literal "queues"]),+    (methodGet, [Literal "queues", Placeholder "queue"]),+    (methodGet, [Literal "queues", Placeholder "queue", Literal "metrics"]),+    (methodGet, [Literal "queues", Placeholder "queue", Literal "messages"]),+    (methodGet, [Literal "queues", Placeholder "queue", Literal "messages", Placeholder "msg_id"]),+    (methodGet, [Literal "queues", Placeholder "queue", Literal "archive"]),+    (methodGet, [Literal "queues", Placeholder "queue", Literal "archive", Placeholder "msg_id"]),+    (methodGet, [Literal "queues", Placeholder "queue", Literal "bindings"]),+    (methodGet, [Literal "metrics"]),+    (methodGet, [Literal "bindings"]),+    (methodGet, [Literal "routing", Literal "test"]),+    (methodGet, [Literal "notify", Literal "throttles"]),+    (methodGet, [Literal "health", Literal "live"]),+    (methodGet, [Literal "health", Literal "ready"]),+    (methodGet, [Literal "ws"])+  ]++httpApp :: InspectionBudget -> InspectConfig -> InspectEnv -> WS.ServerApp -> Application+httpApp budget cfg env wsApp original respond+  | requestMethod req /= methodGet = respond (errorResponse status405 methodNotAllowed)+  | otherwise = respond =<< admittedRoute+  where+    -- WAI retains the empty final segment at a canonical prefix mount (/pgmq/).+    req = if pathInfo original == [""] then original {pathInfo = []} else original+    admittedRoute+      | pathInfo req `elem` [[], ["openapi.json"], ["health", "live"], ["ws"]] = route+      | not (any (matches (pathInfo req) . snd) routeTable) = route+      | Left bad <- validateInput = pure (errorResponse status400 bad)+      | otherwise = maybe busyResponse id <$> withInspectionPermit budget route+    busyResponse+      | pathInfo req == ["health", "ready"] = jsonResponse status503 (encode (ReadinessBody False [DependencyCheck "postgres" False Nothing (Just "Inspection capacity is busy; retry the probe.")]))+      | otherwise = errorResponse status503 inspectionBusy+    validateInput = case pathInfo req of+      ["queues", _, kind] | kind `elem` ["messages", "archive"] -> () <$ parsePage cfg req+      ["queues", _, kind, mid] | kind `elem` ["messages", "archive"] -> () <$ parseMessageId mid+      ["routing", "test"] -> case queryParam "routing_key" req of+        Just raw | Right _ <- parseRoutingKey raw -> Right ()+        raw -> Left (invalidRoutingKey (maybe "" id raw))+      _ -> Right ()+    route = case pathInfo req of+      ["queues", _] -> routeSuccess+      "queues" : q : _ | any (matches (pathInfo req) . snd) routeTable -> queueScoped env q routeSuccess+      _ -> routeSuccess+    routeSuccess = case pathInfo req of+      [] -> pure descriptor+      ["openapi.json"] -> pure (jsonResponse status200 openApiJson)+      ["queues"] -> listQueuesRoute env+      ["queues", q] -> queueRoute env q+      ["queues", q, "metrics"] -> queueMetricsRoute env q+      ["queues", q, "messages"] -> messagesPageRoute cfg env q req+      ["queues", q, "messages", mid] -> messageRoute env q mid+      ["queues", q, "archive"] -> archivePageRoute cfg env q req+      ["queues", q, "archive", mid] -> archivedMessageRoute env q mid+      ["queues", q, "bindings"] -> queueBindingsRoute env q+      ["metrics"] -> allMetricsRoute env+      ["bindings"] -> allBindingsRoute env+      ["routing", "test"] -> routingTestRoute env req+      ["notify", "throttles"] -> throttlesRoute env+      ["health", "live"] -> pure (jsonResponse status200 (encode (LivenessBody True)))+      ["health", "ready"] -> readinessRoute cfg env+      ["ws"] -> pure (maybe (errorResponse (mkStatus 426 "Upgrade Required") websocketUpgradeRequired) id (WaiWS.websocketsApp WS.defaultConnectionOptions {WS.connectionFramePayloadSizeLimit = WS.SizeLimit (fromIntegral (wsMaxFrameBytes cfg)), WS.connectionMessageDataSizeLimit = WS.SizeLimit (fromIntegral (wsMaxFrameBytes cfg))} wsApp req))+      _ -> pure (errorResponse status404 notFound)++descriptor :: Response+descriptor =+  jsonResponse status200 . encode $+    ServiceDescriptor+      { serviceName = "pgmq-inspect",+        serviceVersion = T.pack (showVersion version),+        websocketPath = "ws",+        openapiPath = "openapi.json"+      }++-- | Run a program and encode its result, or map its failure.+runJson :: (ToJSON a) => InspectEnv -> Inspection a -> IO Response+runJson env program = runWith env program (jsonResponse status200 . encode)++runWith :: InspectEnv -> Inspection a -> (a -> Response) -> IO Response+runWith env program onOk =+  either (uncurry errorResponse . classifyRuntimeError) onOk <$> runInspection env program++listQueuesRoute :: InspectEnv -> IO Response+listQueuesRoute env = runJson env listQueuesUnvalidated++queueRoute :: InspectEnv -> Text -> IO Response+queueRoute env name =+  runWith env listQueuesUnvalidated $ \queues ->+    case find ((== name) . unvalidatedName) queues of+      Nothing -> errorResponse status404 (queueNotFound name)+      Just q -> jsonResponse status200 (encode q)++queueMetricsRoute :: InspectEnv -> Text -> IO Response+queueMetricsRoute env name = runScopedWith env name (queueMetricsUnvalidated name) (jsonResponse status200 . encode)++allMetricsRoute :: InspectEnv -> IO Response+allMetricsRoute env = runJson env allQueueMetrics++readinessRoute :: InspectConfig -> InspectEnv -> IO Response+readinessRoute cfg env = do+  started <- getMonotonicTimeNSec+  outcome <- timeout (readinessTimeoutUs cfg) (runInspection env listQueuesUnvalidated)+  finished <- getMonotonicTimeNSec+  let latency = Just (fromIntegral ((finished - started) `div` 1_000_000))+      check = case outcome of+        Nothing -> DependencyCheck "postgres" False latency (Just "readiness probe timed out")+        Just (Left err) -> DependencyCheck "postgres" False latency (Just (fst (describeRuntimeError err)))+        Just (Right _) -> DependencyCheck "postgres" True latency Nothing+      body = ReadinessBody {ready = checkHealthy check, checks = [check]}+  pure (jsonResponse (if ready body then status200 else status503) (encode body))++-- | @from@ (exclusive msg_id cursor, optional) and @limit@ (optional, defaulted,+-- capped). Malformed values are rejected before any database work.+parsePage :: InspectConfig -> Request -> Either ErrorBody (Maybe MessageId, Int32)+parsePage cfg req = do+  from <- case queryParam "from" req of+    Nothing -> Right Nothing+    Just raw -> maybe (Left (invalidCursor raw)) (Right . Just . MessageId) (parseNonNegative raw)+  limit <- case queryParam "limit" req of+    Nothing -> Right (defaultPageLimit cfg)+    Just raw -> case parseDecimalInteger raw of+      Just n | n > 0 -> Right (fromIntegral (min (fromIntegral (maxPageLimit cfg)) n))+      _ -> Left (invalidLimit raw)+  pure (from, limit)++queryParam :: Text -> Request -> Maybe Text+queryParam key req =+  case lookup (TE.encodeUtf8 key) (queryString req) of+    Just (Just raw) -> Just (TE.decodeUtf8Lenient raw)+    Just Nothing -> Just ""+    Nothing -> Nothing++-- Parse arbitrary precision before narrowing; fixed-width decimal silently wraps.+-- Bound token length so adversarial inputs do not build enormous Integers.+parseDecimalInteger :: Text -> Maybe Integer+parseDecimalInteger raw+  | T.null raw || T.length raw > 128 || not (T.all (\c -> c >= '0' && c <= '9') raw) = Nothing+  | otherwise = case TR.decimal raw of+      Right (n, rest) | T.null rest -> Just n+      _ -> Nothing++parseNonNegative :: Text -> Maybe Int64+parseNonNegative raw = do+  n <- parseDecimalInteger raw+  if n <= fromIntegral (maxBound :: Int64) then Just (fromInteger n) else Nothing++parseMessageId :: Text -> Either ErrorBody MessageId+parseMessageId raw = maybe (Left (invalidMessageId raw)) (Right . MessageId) (parseNonNegative raw)++-- | Ask for one row more than the page, return the page, and point at the+-- next one only when that extra row existed.+toPage :: Int32 -> (a -> MessageId) -> Vector a -> Page a+toPage limit cursorOf rows =+  let n = fromIntegral limit+      page = V.toList (V.take n rows)+   in Page {items = page, nextCursor = if V.length rows > n then cursorOf <$> lastOf page else Nothing}+  where+    lastOf [] = Nothing+    lastOf xs = Just (last xs)++messagesPageRoute :: InspectConfig -> InspectEnv -> Text -> Request -> IO Response+messagesPageRoute cfg env name req =+  case parsePage cfg req of+    Left bad -> pure (errorResponse status400 bad)+    Right (from, limit) ->+      runScopedWith env name (peekMessages PeekMessages {unvalidatedQueueName = name, afterMessageId = from, limit = limit + 1}) $+        jsonResponse status200 . encode . toPage limit (^. #messageId)++archivePageRoute :: InspectConfig -> InspectEnv -> Text -> Request -> IO Response+archivePageRoute cfg env name req =+  case parsePage cfg req of+    Left bad -> pure (errorResponse status400 bad)+    Right (from, limit) ->+      runScopedWith env name (peekArchivedMessages PeekMessages {unvalidatedQueueName = name, afterMessageId = from, limit = limit + 1}) $+        jsonResponse status200 . encode . toPage limit (\a -> archivedMessage a ^. #messageId)++messageRoute :: InspectEnv -> Text -> Text -> IO Response+messageRoute env name raw =+  case parseMessageId raw of+    Left bad -> pure (errorResponse status400 bad)+    Right mid ->+      runScopedWith env name (lookupMessage LookupMessage {unvalidatedQueueName = name, messageId = mid}) $+        maybe (errorResponse status404 (messageNotFound name raw)) (jsonResponse status200 . encode)++archivedMessageRoute :: InspectEnv -> Text -> Text -> IO Response+archivedMessageRoute env name raw =+  case parseMessageId raw of+    Left bad -> pure (errorResponse status400 bad)+    Right mid ->+      runScopedWith env name (lookupArchivedMessage LookupMessage {unvalidatedQueueName = name, messageId = mid}) $+        maybe (errorResponse status404 (messageNotFound name raw)) (jsonResponse status200 . encode)++queueBindingsRoute :: InspectEnv -> Text -> IO Response+queueBindingsRoute env name =+  runScopedWith env name listTopicBindings $ \bindings ->+    jsonResponse status200 (encode (filter ((== name) . bindingQueueName) bindings))++allBindingsRoute :: InspectEnv -> IO Response+allBindingsRoute env = runJson env listTopicBindings++routingTestRoute :: InspectEnv -> Request -> IO Response+routingTestRoute env req =+  case queryParam "routing_key" req of+    Nothing -> pure (errorResponse status400 (invalidRoutingKey ""))+    Just raw -> case parseRoutingKey raw of+      Left _ -> pure (errorResponse status400 (invalidRoutingKey raw))+      Right key -> runJson env (testRouting key)++throttlesRoute :: InspectEnv -> IO Response+throttlesRoute env = runJson env listNotifyInsertThrottles++-- Only metadata-confirmed absence is a queue 404. Global schema failures stay 500.+queueScoped :: InspectEnv -> Text -> IO Response -> IO Response+queueScoped env name action = do+  listed <- runInspection env listQueuesUnvalidated+  case listed of+    Left err -> pure (uncurry errorResponse (classifyRuntimeError err))+    Right queues | not (any ((== name) . unvalidatedName) queues) -> pure (errorResponse status404 (queueNotFound name))+    Right _ -> action++matches :: [Text] -> [PathSegment] -> Bool+matches [] [] = True+matches (x : xs) (Literal y : ys) = x == y && matches xs ys+matches (_ : xs) (Placeholder _ : ys) = matches xs ys+matches _ _ = False++runScopedWith :: InspectEnv -> Text -> Inspection a -> (a -> Response) -> IO Response+runScopedWith env name program onOk = do+  result <- runInspection env program+  case result of+    Left err | sqlState err == Just "42P01" -> do+      queues <- runInspection env listQueuesUnvalidated+      pure $ case queues of+        Right qs | not (any ((== name) . unvalidatedName) qs) -> errorResponse status404 (queueNotFound name)+        _ -> uncurry errorResponse (classifyRuntimeError err)+    _ -> pure (either (uncurry errorResponse . classifyRuntimeError) onOk result)
+ src/Pgmq/Inspect/Lifecycle.hs view
@@ -0,0 +1,37 @@+-- | Internal, observable socket/thread ownership used by the public runner and+-- deterministic lifecycle tests. A released ephemeral port may be reused by the+-- OS, so tests inspect the owned Socket rather than racing to bind that port again.+module Pgmq.Inspect.Lifecycle (withBoundSocket, withRunningSocket) where++import Control.Concurrent.Async (Async, waitSTM, withAsync)+import Control.Concurrent.STM+import Control.Exception (bracket, finally)+import Data.Streaming.Network (bindPortTCP)+import Data.String (fromString)+import Network.Socket qualified as Socket+import Network.Wai (Application)+import Network.Wai.Handler.Warp qualified as Warp+import Pgmq.Inspect.Config++withBoundSocket :: InspectServerConfig -> (Socket.Socket -> IO a) -> IO a+withBoundSocket cfg k = do+  if bindPort cfg < 0 || bindPort cfg > 65535 then ioError (userError "bindPort must be between 0 and 65535") else pure ()+  bracket (bindPortTCP (bindPort cfg) (fromString (bindHost cfg))) Socket.close k++-- | The caller owns the socket. Startup is acknowledged or fails synchronously;+-- graceful application shutdown runs before Warp is cancelled and joined.+withRunningSocket :: InspectServerConfig -> Socket.Socket -> Application -> IO () -> (Int -> Async () -> IO a) -> IO a+withRunningSocket cfg socket application shutdown k = do+  port <- fromIntegral <$> Socket.socketPort socket+  started <- newEmptyTMVarIO+  let settings =+        Warp.setBeforeMainLoop+          (atomically (putTMVar started ()))+          (Warp.setHost (fromString (bindHost cfg)) (Warp.setPort port Warp.defaultSettings))+  withAsync (Warp.runSettingsSocket settings socket application) $ \worker -> do+    atomically ((waitSTM worker >> throwSTM (userError "Inspection server exited during startup")) `orElse` readTMVar started)+    ( withAsync (k port worker) $ \continuation -> do+        outcome <- atomically ((Left <$> waitSTM worker) `orElse` (Right <$> waitSTM continuation))+        either (const (ioError (userError "Inspection server exited"))) pure outcome+      )+      `finally` shutdown
+ src/Pgmq/Inspect/Listener.hs view
@@ -0,0 +1,155 @@+-- | One runtime-wide LISTEN connection with refcounted channels. Registry changes+-- and socket readiness share an interruptible STM wait whose fd registration is+-- always released. Reconnect reissues every desired LISTEN; polling is independent.+-- The backend application_name is @pgmq-inspect-listener@. Health and channel+-- acknowledgement are separate observations; neither promises installed triggers.+module Pgmq.Inspect.Listener+  ( Listener,+    SubscriberKey,+    withListener,+    subscribeChannel,+    unsubscribeChannel,+    listenerHealthy,+    listenerAcknowledged,+    listenerChannelCount,+  )+where++import Control.Concurrent (threadDelay, threadWaitReadSTM, yield)+import Control.Concurrent.Async (link, withAsync)+import Control.Concurrent.STM+import Control.Exception+import Control.Monad (forM_, unless, when)+import Data.IORef+import Data.Map.Strict (Map)+import Data.Map.Strict qualified as Map+import Data.Set (Set)+import Data.Set qualified as Set+import Data.Text (Text)+import Data.Text qualified as T+import Data.Text.Encoding qualified as TE+import Data.Unique (Unique, newUnique)+import Database.PostgreSQL.LibPQ qualified as PQ+import Hasql.Connection qualified as Connection+import Hasql.Connection.Settings (Settings)+import Hasql.Connection.Settings qualified as Settings+import Hasql.Session qualified as Session++newtype SubscriberKey = SubscriberKey Unique+  deriving newtype (Eq, Ord)++data Listener = Listener+  { listenerRegistry :: !(TVar (Map Text (Map SubscriberKey (STM ())))),+    registryChanged :: !(TVar Bool),+    healthy :: !(TVar Bool),+    acknowledged :: !(TVar (Set Text))+  }++withListener :: Settings -> (Listener -> IO a) -> IO a+withListener settings k = do+  listener <- Listener <$> newTVarIO Map.empty <*> newTVarIO False <*> newTVarIO False <*> newTVarIO Set.empty+  let cleanup = atomically $ do+        writeTVar (healthy listener) False+        writeTVar (acknowledged listener) Set.empty+        writeTVar (listenerRegistry listener) Map.empty+  (withAsync (reconnect settings listener 100000) $ \task -> link task >> k listener) `finally` cleanup++subscribeChannel :: Listener -> Text -> STM () -> IO SubscriberKey+subscribeChannel listener channel wake = mask_ $ do+  key <- SubscriberKey <$> newUnique+  atomically $ do+    registry <- readTVar (listenerRegistry listener)+    let callbacks = Map.findWithDefault Map.empty channel registry+    writeTVar (listenerRegistry listener) (Map.insert channel (Map.insert key wake callbacks) registry)+    when (Map.null callbacks) (writeTVar (registryChanged listener) True)+  pure key++unsubscribeChannel :: Listener -> Text -> SubscriberKey -> IO ()+unsubscribeChannel listener channel key = atomically $ do+  registry <- readTVar (listenerRegistry listener)+  case Map.lookup channel registry of+    Nothing -> pure ()+    Just callbacks -> do+      let remaining = Map.delete key callbacks+      if Map.null remaining+        then do+          writeTVar (listenerRegistry listener) (Map.delete channel registry)+          writeTVar (registryChanged listener) True+        else writeTVar (listenerRegistry listener) (Map.insert channel remaining registry)++listenerHealthy :: Listener -> STM Bool+listenerHealthy = readTVar . healthy++listenerAcknowledged :: Listener -> Text -> STM Bool+listenerAcknowledged listener channel = Set.member channel <$> readTVar (acknowledged listener)++listenerChannelCount :: Listener -> STM Int+listenerChannelCount listener = Map.size <$> readTVar (listenerRegistry listener)++reconnect :: Settings -> Listener -> Int -> IO ()+reconnect settings listener delay = do+  wasHealthy <- newIORef False+  outcome <- try @SomeException $ bracket acquire Connection.release (serve listener wasHealthy)+  atomically $ writeTVar (healthy listener) False >> writeTVar (acknowledged listener) Set.empty+  case outcome of+    Left err | Just (_ :: SomeAsyncException) <- fromException err -> throwIO err+    _ -> pure ()+  succeeded <- readIORef wasHealthy+  let pause = if succeeded then 100000 else delay+  threadDelay pause+  reconnect settings listener (if succeeded then 100000 else min 5000000 (delay * 2))+  where+    acquire = Connection.acquire (settings <> Settings.applicationName "pgmq-inspect-listener") >>= either (const (ioError (userError "Listener connection failed"))) pure++serve :: Listener -> IORef Bool -> Connection.Connection -> IO ()+serve listener wasHealthy connection = loop Set.empty+  where+    loop listening = do+      desired <- atomically $ do+        registry <- readTVar (listenerRegistry listener)+        writeTVar (registryChanged listener) False+        pure (Map.keysSet registry)+      forM_ (Set.toList (listening Set.\\ desired)) $ \channel -> command "UNLISTEN" channel+      forM_ (Set.toList (desired Set.\\ listening)) $ \channel -> command "LISTEN" channel+      atomically $ writeTVar (acknowledged listener) desired >> writeTVar (healthy listener) True+      writeIORef wasHealthy True+      full <-+        Connection.use+          connection+          ( Session.onLibpqConnection+              ( \raw -> do+                  count <- drain raw (0 :: Int)+                  if count == 64+                    then pure (Right True, raw)+                    else do+                      descriptor <- PQ.socket raw >>= maybe (ioError (userError "Listener has no socket")) pure+                      -- Always unregister, including registry changes and async cancellation.+                      bracket (threadWaitReadSTM descriptor) snd $ \(readable, _) -> do+                        inputReady <-+                          atomically $+                            (readTVar (registryChanged listener) >>= check >> pure False)+                              `orElse` (readable >> pure True)+                        when inputReady $ do+                          consumed <- PQ.consumeInput raw+                          status <- PQ.status raw+                          unless (consumed && status == PQ.ConnectionOk) (ioError (userError "Listener connection dropped"))+                      pure (Right False, raw)+              )+          )+          >>= either (const (ioError (userError "Listener receive failed"))) pure+      when full yield+      loop desired+    command operation channel = do+      let quoted = "\"" <> T.replace "\"" "\"\"" channel <> "\""+      Connection.use connection (Session.script (operation <> " " <> quoted)) >>= either (const (ioError (userError "Listener channel synchronization failed"))) pure+    drain raw count+      | count >= 64 = pure count+      | otherwise =+          PQ.notifies raw >>= \case+            Nothing -> pure count+            Just notification -> do+              let channel = TE.decodeUtf8Lenient (PQ.notifyRelname notification)+              atomically $ do+                registry <- readTVar (listenerRegistry listener)+                sequence_ (Map.findWithDefault Map.empty channel registry)+              drain raw (count + 1)
+ src/Pgmq/Inspect/OpenApi.hs view
@@ -0,0 +1,185 @@+-- | OpenAPI 3 HTTP contract. Route coverage and real-response validation live in OpenApiSpec.+module Pgmq.Inspect.OpenApi+  ( openApiDocument,+    openApiJson,+    documentedPaths,+  )+where++import Data.Aeson (encode)+import Data.ByteString.Lazy qualified as LBS+import Data.HashMap.Strict.InsOrd.Compat qualified as InsOrd+import Data.OpenApi (AdditionalProperties (..), Components (..), Info (..), MediaTypeObject (..), OpenApi (..), OpenApiItems (..), OpenApiType (..), Operation (..), Param (..), ParamLocation (..), PathItem (..), Reference (..), Referenced (..), Response (..), Responses (..), Schema (..), Server (..))+import Data.Text (Text)+import Data.Text qualified as T+import Data.Version (showVersion)+import Paths_pgmq_inspect (version)++openApiDocument :: OpenApi+openApiDocument =+  mempty+    { _openApiInfo =+        mempty+          { _infoTitle = "pgmq-inspect",+            _infoVersion = T.pack (showVersion version),+            _infoDescription =+              Just+                "Read-only inspection of pgmq queues. Every path is relative to wherever the \+                \application is mounted. Field names are pgmq's own SQL column names in snake_case; \+                \once published they are never removed or re-typed. The WebSocket feed at ws is \+                \described in docs/user/queue-inspection.md, not here."+          },+      _openApiServers = [Server "." Nothing mempty],+      _openApiPaths = InsOrd.fromList paths,+      _openApiComponents = mempty {_componentsSchemas = InsOrd.fromList schemas}+    }++openApiJson :: LBS.ByteString+openApiJson = encode openApiDocument++documentedPaths :: [FilePath]+documentedPaths = map fst paths++paths :: [(FilePath, PathItem)]+paths =+  [ ("/", getOp "serviceDescriptor" "Service descriptor" [] (ok "ServiceDescriptor") []),+    ("/queues", getOp "listQueues" "List every queue, including names this library would reject" [] (okArray "Queue") []),+    ("/queues/{queue}", getOp "getQueue" "One queue" [queueParam] (ok "Queue") [(404, "queue_not_found")]),+    ("/queues/{queue}/metrics", getOp "getQueueMetrics" "Metrics for one queue" [queueParam] (ok "QueueMetrics") [(404, "queue_not_found")]),+    ("/metrics", getOp "getAllMetrics" "Metrics for every queue; scans rows across queues, subject to the inspection budget" [] (okArray "QueueMetrics") []),+    ("/queues/{queue}/messages", getOp "peekMessages" "Non-destructive keyset page of a queue" [queueParam, fromParam, limitParam] (ok "MessagePage") [(400, "invalid_cursor, invalid_limit"), (404, "queue_not_found")]),+    ("/queues/{queue}/messages/{msg_id}", getOp "lookupMessage" "One message by id" [queueParam, msgIdParam] (ok "Message") [(400, "invalid_message_id"), (404, "queue_not_found, message_not_found")]),+    ("/queues/{queue}/archive", getOp "peekArchive" "Keyset page of a queue's archive" [queueParam, fromParam, limitParam] (ok "ArchivedMessagePage") [(400, "invalid_cursor, invalid_limit"), (404, "queue_not_found")]),+    ("/queues/{queue}/archive/{msg_id}", getOp "lookupArchivedMessage" "One archived message by id" [queueParam, msgIdParam] (ok "ArchivedMessage") [(400, "invalid_message_id"), (404, "queue_not_found, message_not_found")]),+    ("/queues/{queue}/bindings", getOp "listQueueBindings" "Topic bindings of one queue" [queueParam] (okArray "TopicBinding") [(404, "queue_not_found")]),+    ("/bindings", getOp "listBindings" "Every topic binding" [] (okArray "TopicBinding") []),+    ("/routing/test", getOp "testRouting" "Queues a routing key would reach" [routingKeyParam] (okArray "RoutingMatch") [(400, "invalid_routing_key")]),+    ("/notify/throttles", getOp "listNotifyThrottles" "Insert-notification throttle rows" [] (okArray "NotifyInsertThrottle") []),+    ("/health/live", getOp "liveness" "Process liveness" [] (ok "Liveness") []),+    ("/health/ready", readinessOp),+    ("/openapi.json", getOp "openApi" "This document" [] (okAny "The OpenAPI 3 document") [])+  ]++getOp :: Text -> Text -> [Referenced Param] -> Referenced Response -> [(Int, Text)] -> PathItem+getOp opId summary params success errors =+  mempty+    { _pathItemGet =+        Just+          mempty+            { _operationOperationId = Just opId,+              _operationSummary = Just summary,+              _operationParameters = params,+              _operationResponses =+                mempty+                  { _responsesResponses =+                      InsOrd.fromList+                        ( (200, success)+                            : [(status, errorResponse codes) | (status, codes) <- errors]+                              <> [ (405, errorResponse "method_not_allowed"),+                                   (500, errorResponse "database_error"),+                                   (503, errorResponse "database_unavailable, inspection_busy, inspection_timeout")+                                 ]+                        )+                  }+            }+    }++-- Override after getOp so its common 503 cannot overwrite the readiness body.+readinessOp :: PathItem+readinessOp =+  let base = getOp "readiness" "Database readiness" [] (ok "Readiness") []+      patch op =+        op+          { _operationResponses =+              (_operationResponses op)+                { _responsesResponses =+                    InsOrd.insert 503 (ok "Readiness") (_responsesResponses (_operationResponses op))+                }+          }+   in base {_pathItemGet = fmap patch (_pathItemGet base)}++errorResponse :: Text -> Referenced Response+errorResponse codes =+  Inline+    mempty+      { _responseDescription = "Error envelope; code is one of: " <> codes,+        _responseContent = InsOrd.singleton "application/json" (jsonOf (Ref (Reference "ErrorEnvelope")))+      }++ok :: Text -> Referenced Response+ok name = Inline mempty {_responseDescription = "OK", _responseContent = InsOrd.singleton "application/json" (jsonOf (Ref (Reference name)))}++okArray :: Text -> Referenced Response+okArray name = Inline mempty {_responseDescription = "OK", _responseContent = InsOrd.singleton "application/json" (jsonOf (Inline (arrayOf (Ref (Reference name)))))}++okAny :: Text -> Referenced Response+okAny description = Inline mempty {_responseDescription = description, _responseContent = InsOrd.singleton "application/json" (jsonOf (Inline (typed OpenApiObject)))}++jsonOf :: Referenced Schema -> MediaTypeObject+jsonOf s = mempty {_mediaTypeObjectSchema = Just s}++queueParam, msgIdParam, fromParam, limitParam, routingKeyParam :: Referenced Param+queueParam = Inline mempty {_paramName = "queue", _paramIn = ParamPath, _paramRequired = Just True, _paramDescription = Just "Queue name as stored in pgmq.meta; any server-accepted name, not only names this library validates", _paramSchema = Just (Inline (typed OpenApiString))}+msgIdParam = Inline mempty {_paramName = "msg_id", _paramDescription = Just "Non-negative int64 ID, at most 128 decimal digits; browser clients must preserve exact integer tokens", _paramIn = ParamPath, _paramRequired = Just True, _paramSchema = Just (Inline int64)}+fromParam = Inline mempty {_paramName = "from", _paramIn = ParamQuery, _paramRequired = Just False, _paramDescription = Just "Exclusive msg_id cursor, 0..9223372036854775807, at most 128 decimal digits. Echo next_cursor here; moving-view pages are not a transaction snapshot.", _paramSchema = Just (Inline int64)}+limitParam = Inline mempty {_paramName = "limit", _paramIn = ParamQuery, _paramRequired = Just False, _paramDescription = Just "Positive decimal page size (at most 128 digits), capped before narrowing; default 50, cap 500, both host-configurable. Larger values are accepted and capped", _paramSchema = Just (Inline (typed OpenApiInteger) {_schemaMinimum = Just 1})}+routingKeyParam = Inline mempty {_paramName = "routing_key", _paramIn = ParamQuery, _paramRequired = Just True, _paramSchema = Just (Inline (typed OpenApiString))}++typed :: OpenApiType -> Schema+typed t = mempty {_schemaType = Just t, _schemaAdditionalProperties = if t == OpenApiObject then Just (AdditionalPropertiesAllowed True) else Nothing}++nullable :: Schema -> Schema+nullable s = s {_schemaNullable = Just True}++dateTime :: Schema+dateTime = (typed OpenApiString) {_schemaFormat = Just "date-time"}++int64, int32 :: Schema+int64 = (typed OpenApiInteger) {_schemaFormat = Just "int64", _schemaMinimum = Just 0, _schemaMaximum = Just 9223372036854775807}+int32 = (typed OpenApiInteger) {_schemaFormat = Just "int32"}++arrayOf :: Referenced Schema -> Schema+arrayOf item = (typed OpenApiArray) {_schemaItems = Just (OpenApiItemsObject item)}++objectOf :: [(Text, Schema)] -> Schema+objectOf fields =+  (typed OpenApiObject)+    { _schemaProperties = InsOrd.fromList [(name, Inline s) | (name, s) <- fields],+      _schemaRequired = map fst fields -- every key is always present; nullable keys carry null+    }++messageFields :: [(Text, Schema)]+messageFields =+  [ ("msg_id", int64),+    ("read_ct", int64),+    ("enqueued_at", dateTime),+    ("last_read_at", nullable dateTime),+    ("vt", dateTime),+    ("message", mempty {_schemaAdditionalProperties = Just (AdditionalPropertiesAllowed True), _schemaDescription = Just "Any JSON value; a SQL NULL body is JSON null"}),+    ("headers", mempty {_schemaAdditionalProperties = Just (AdditionalPropertiesAllowed True), _schemaDescription = Just "Any JSON value, including null; absent and JSON-null optional headers encode as null"})+  ]++schemas :: [(Text, Schema)]+schemas =+  [ ("Queue", objectOf [("queue_name", typed OpenApiString), ("is_partitioned", typed OpenApiBoolean), ("is_unlogged", typed OpenApiBoolean), ("created_at", dateTime)]),+    ("Message", objectOf messageFields),+    ("ArchivedMessage", objectOf (messageFields <> [("archived_at", dateTime)])),+    ("QueueMetrics", objectOf [("queue_name", typed OpenApiString), ("queue_length", int64 {_schemaDescription = Just "Every message in the queue table, visible or not"}), ("queue_visible_length", int64 {_schemaDescription = Just "Messages whose visibility timeout has expired, i.e. available to consumers now"}), ("newest_msg_age_sec", nullable int32), ("oldest_msg_age_sec", nullable int32), ("total_messages", int64), ("scrape_time", dateTime), ("default_partition_length", nullable int64 {_schemaDescription = Just "PGMQ 1.13 planner estimate of rows in the queue and archive default partitions; null on 1.12 and for ordinary queues"})]),+    ("TopicBinding", objectOf [("pattern", typed OpenApiString), ("queue_name", typed OpenApiString), ("bound_at", dateTime), ("compiled_regex", typed OpenApiString)]),+    ("RoutingMatch", objectOf [("pattern", typed OpenApiString), ("queue_name", typed OpenApiString), ("compiled_regex", typed OpenApiString)]),+    ("NotifyInsertThrottle", objectOf [("queue_name", typed OpenApiString), ("throttle_interval_ms", int32), ("last_notified_at", dateTime)]),+    ("MessagePage", page "Message"),+    ("ArchivedMessagePage", page "ArchivedMessage"),+    ("ErrorBody", (objectOf [("code", typed OpenApiString), ("message", typed OpenApiString)]) {_schemaProperties = InsOrd.fromList [("code", Inline (typed OpenApiString)), ("message", Inline (typed OpenApiString)), ("details", Inline (typed OpenApiObject))], _schemaRequired = ["code", "message"]}),+    ("ErrorEnvelope", (typed OpenApiObject) {_schemaProperties = InsOrd.singleton "error" (Ref (Reference "ErrorBody")), _schemaRequired = ["error"]}),+    ("ServiceDescriptor", objectOf [("service", typed OpenApiString), ("version", typed OpenApiString), ("websocket_path", typed OpenApiString), ("openapi_path", typed OpenApiString)]),+    ("Liveness", objectOf [("alive", typed OpenApiBoolean)]),+    ("Readiness", objectOf [("ready", typed OpenApiBoolean), ("checks", arrayOf (Ref (Reference "DependencyCheck")))]),+    ("DependencyCheck", objectOf [("name", typed OpenApiString), ("healthy", typed OpenApiBoolean), ("latency_ms", nullable int64), ("error", nullable (typed OpenApiString))])+  ]+  where+    page item =+      (typed OpenApiObject)+        { _schemaProperties = InsOrd.fromList [("items", Inline (arrayOf (Ref (Reference item)))), ("next_cursor", Inline int64 {_schemaDescription = Just "msg_id of the last item; absent on the last page"})],+          _schemaRequired = ["items"]+        }
+ src/Pgmq/Inspect/Outbox.hs view
@@ -0,0 +1,73 @@+-- | Bounded transport state, with generation tags checked immediately before writing.+module Pgmq.Inspect.Outbox+  ( Outbox,+    Generation,+    SubscriptionTag,+    newOutbox,+    enqueue,+    enqueueTagged,+    takeOutgoing,+    clearQueue,+    queuedFrames,+    tagCurrent,+    overflowFrame,+  )+where++import Control.Concurrent.STM+import Control.Monad (forM_, void, when)+import Data.Map.Strict (Map)+import Data.Map.Strict qualified as Map+import Data.Text (Text)+import Data.Unique (Unique)+import Numeric.Natural (Natural)+import Pgmq.Inspect.Frames++type Generation = Unique++type SubscriptionTag = (TVar (Map Text Generation), Text, Generation)++data Outbox = Outbox !(TBQueue (Maybe SubscriptionTag, ServerFrame)) !(TVar Bool)++newOutbox :: Natural -> IO Outbox+newOutbox capacity = Outbox <$> newTBQueueIO capacity <*> newTVarIO False++enqueue :: Outbox -> ServerFrame -> STM ()+enqueue out = enqueueTagged out Nothing++enqueueTagged :: Outbox -> Maybe SubscriptionTag -> ServerFrame -> STM ()+enqueueTagged (Outbox frames overflow) tag frame = do+  full <- isFullTBQueue frames+  when full $ void (readTBQueue frames) >> writeTVar overflow True+  writeTBQueue frames (tag, frame)++tagCurrent :: SubscriptionTag -> STM Bool+tagCurrent (generations, queue, generation) = (== Just generation) . Map.lookup queue <$> readTVar generations++takeOutgoing :: Outbox -> STM (Bool, ServerFrame)+takeOutgoing box@(Outbox frames overflow) = do+  (tag, frame) <- readTBQueue frames+  valid <- maybe (pure True) tagCurrent tag+  if not valid+    then takeOutgoing box+    else do+      dropped <- readTVar overflow+      writeTVar overflow False+      pure (dropped, frame)++queuedFrames :: Outbox -> STM Natural+queuedFrames (Outbox frames _) = lengthTBQueue frames++overflowFrame :: ServerFrame+overflowFrame = ErrorFrame "overflow" "The server dropped frames because this connection was not reading them; re-subscribe or re-read the queue over HTTP." Nothing++-- | Explicit subscribe/unsubscribe discards any frames still queued for that name.+clearQueue :: Outbox -> Text -> STM ()+clearQueue (Outbox frames _) name = do+  queued <- flushTBQueue frames+  forM_ queued $ \item@(_, frame) -> when (frameQueue frame /= Just name) (writeTBQueue frames item)+  where+    frameQueue (Snapshot q _ _) = Just q+    frameQueue (Update q _ _) = Just q+    frameQueue (ErrorFrame _ _ q) = q+    frameQueue _ = Nothing
+ src/Pgmq/Inspect/Refresh.hs view
@@ -0,0 +1,344 @@+-- | Runtime-wide fair scheduling. Only fixed workers perform database work.+-- Notifications and missed polls coalesce; subscription count never multiplies scans.+module Pgmq.Inspect.Refresh+  ( MetricsCoordinator,+    PushHooks (..),+    Subscriber,+    newSubscriber,+    subscriberOutbox,+    withCoordinator,+    withCoordinatorUsing,+    subscribeQueue,+    unsubscribeQueue,+    releaseSubscriber,+    stopCoordinator,+    coordinatorQueueCount,+    coordinatorSubscriberCount,+    coordinatorInFlight,+  )+where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async (link, withAsync)+import Control.Concurrent.STM+import Control.Exception+import Control.Monad (forM_, forever, when)+import Data.Aeson (Value (..), toJSON)+import Data.Aeson.KeyMap qualified as KM+import Data.List (find, sortOn)+import Data.Map.Strict (Map)+import Data.Map.Strict qualified as Map+import Data.Maybe (isJust)+import Data.Text (Text)+import Data.Text qualified as T+import Data.Unique (Unique, newUnique)+import Data.Word (Word64)+import GHC.Clock (getMonotonicTimeNSec)+import Numeric.Natural (Natural)+import Pgmq.Effectful.Effect qualified as E+import Pgmq.Hasql.Statements.Types (QueueMetrics)+import Pgmq.Inspect.Budget+import Pgmq.Inspect.Config+import Pgmq.Inspect.Env+import Pgmq.Inspect.Error+import Pgmq.Inspect.Frames+import Pgmq.Inspect.Outbox+import Pgmq.Types++-- | Register one distinct eligible queue, returning its idempotent cleanup.+newtype PushHooks = PushHooks {registerPush :: Text -> STM () -> IO (IO ())}++data Subscriber = Subscriber+  { subscriberId :: !Unique,+    subscriberGenerations :: !(TVar (Map Text Generation)),+    subscriberFaultAt :: !(TVar (Maybe Word64)),+    subscriberOutbox :: !Outbox+  }++data Subscription = Subscription !Subscriber !Generation !Bool++-- Bool: initial snapshot has been atomically published.+data Entry = Entry+  { entryGeneration :: !Generation,+    entrySubscribers :: !(Map Unique Subscription),+    entryMetrics :: !(Maybe QueueMetrics),+    entryPushCleanup :: !(Maybe (IO ())),+    entryLastStart :: !(Maybe Word64),+    entryPollAt :: !Word64,+    entryDirtyAt :: !(Maybe Word64),+    entryFlight :: !(Maybe Unique)+  }++data MetricsCoordinator = MetricsCoordinator+  { coordinatorConfig :: !InspectConfig,+    coordinatorEnv :: !InspectEnv,+    coordinatorBudget :: !InspectionBudget,+    coordinatorPush :: !(Maybe PushHooks),+    coordinatorEntries :: !(TVar (Map Text Entry)),+    coordinatorClock :: !(TVar Word64),+    coordinatorStopped :: !(TVar Bool)+  }++data Job = Job !Text !Generation !Unique !(Map Unique Generation) !UpdateSource++newSubscriber :: Natural -> IO Subscriber+newSubscriber capacity = Subscriber <$> newUnique <*> newTVarIO Map.empty <*> newTVarIO Nothing <*> newOutbox capacity++withCoordinator :: InspectConfig -> InspectEnv -> InspectionBudget -> Maybe PushHooks -> (MetricsCoordinator -> IO a) -> IO a+withCoordinator = withCoordinatorUsing ((`div` 1000) <$> getMonotonicTimeNSec)++-- | Clock seam for deterministic scheduling and generation tests (microseconds).+withCoordinatorUsing :: IO Word64 -> InspectConfig -> InspectEnv -> InspectionBudget -> Maybe PushHooks -> (MetricsCoordinator -> IO a) -> IO a+withCoordinatorUsing clock cfg env budget hooks k = do+  now <- clock+  coordinator <- MetricsCoordinator cfg env budget hooks <$> newTVarIO Map.empty <*> newTVarIO now <*> newTVarIO False+  let ticker = forever $ do+        current <- clock+        atomically $ do+          writeTVar (coordinatorClock coordinator) current+          modifyTVar' (coordinatorEntries coordinator) (collectIdle cfg current)+        threadDelay 10000+      workers 0 action = action+      workers n action = withAsync (worker clock coordinator) $ \task -> link task >> workers (n - 1) action+      cleanup = do+        cleanups <- atomically $ do+          stopCoordinator coordinator+          entries <- readTVar (coordinatorEntries coordinator)+          writeTVar (coordinatorEntries coordinator) Map.empty+          pure [cleanup' | entry <- Map.elems entries, Just cleanup' <- [entryPushCleanup entry]]+        sequence_ cleanups+  (withAsync ticker $ \task -> link task >> workers (maxConcurrentReads cfg) (k coordinator `finally` atomically (stopCoordinator coordinator))) `finally` cleanup++stopCoordinator :: MetricsCoordinator -> STM ()+stopCoordinator = flip writeTVar True . coordinatorStopped++coordinatorQueueCount :: MetricsCoordinator -> STM Int+coordinatorQueueCount coordinator = length . filter (not . Map.null . entrySubscribers) . Map.elems <$> readTVar (coordinatorEntries coordinator)++coordinatorSubscriberCount :: MetricsCoordinator -> Text -> STM Int+coordinatorSubscriberCount coordinator name = maybe 0 (Map.size . entrySubscribers) . Map.lookup name <$> readTVar (coordinatorEntries coordinator)++coordinatorInFlight :: MetricsCoordinator -> STM Int+coordinatorInFlight coordinator = length . filter (isJust . entryFlight) . Map.elems <$> readTVar (coordinatorEntries coordinator)++-- Keep only bounded cooldown leases after unsubscribe, so unsubscribe/re-subscribe+-- cannot reset the minimum refresh interval. No metadata, metrics or listener remains.+collectIdle :: InspectConfig -> Word64 -> Map Text Entry -> Map Text Entry+collectIdle cfg now = Map.filter $ \entry -> not (Map.null (entrySubscribers entry)) || isJust (entryFlight entry) || not (cooled cfg now entry)++cooled :: InspectConfig -> Word64 -> Entry -> Bool+cooled cfg now entry = maybe True (\started -> now >= started && now - started >= fromIntegral (metricsMinRefreshUs cfg)) (entryLastStart entry)++subscribeQueue :: MetricsCoordinator -> Subscriber -> Text -> IO ()+subscribeQueue coordinator subscriber name = do+  generation <- newUnique+  atomically $ do+    stopped <- readTVar (coordinatorStopped coordinator)+    unlessStopped stopped $ do+      current <- readTVar (subscriberGenerations subscriber)+      now <- readTVar (coordinatorClock coordinator)+      entries <- collectIdle cfg now <$> readTVar (coordinatorEntries coordinator)+      let existing = Map.lookup name entries+          activeCount = length (filter (not . Map.null . entrySubscribers) (Map.elems entries))+          fresh = maybe True (Map.null . entrySubscribers) existing+          reject code message = enqueue (subscriberOutbox subscriber) (ErrorFrame code message (Just name))+      if T.null name || T.length name > 47 || T.any (== '\0') name+        then enqueue (subscriberOutbox subscriber) (ErrorFrame "invalid_frame" "Queue names must contain 1 to 47 characters without NUL." Nothing)+        else+          if not (Map.member name current) && Map.size current >= wsMaxSubscriptions cfg+            then reject "too_many_subscriptions" "This connection has reached its subscription limit."+            else+              if fresh && activeCount >= wsMaxQueues cfg+                then reject "too_many_queues" "The runtime has reached its distinct queue limit."+                else+                  if not (isJust existing) && toInteger (Map.size entries) >= 2 * toInteger (wsMaxQueues cfg)+                    then reject "inspection_busy" "Queue refresh cooldown capacity is busy; retry the subscription."+                    else do+                      let entry = maybe (Entry generation Map.empty Nothing Nothing Nothing now Nothing Nothing) id existing+                          -- A generation tag invalidates queued frames from the previous subscription.+                          tag = (subscriberGenerations subscriber, name, generation)+                          active = isJust (entryMetrics entry)+                      clearQueue (subscriberOutbox subscriber) name+                      writeTVar (subscriberGenerations subscriber) (Map.insert name generation current)+                      forM_ (entryMetrics entry) $ \metrics -> enqueueTagged (subscriberOutbox subscriber) (Just tag) (Snapshot name metrics (isJust (entryPushCleanup entry)))+                      let updated = entry {entrySubscribers = Map.insert (subscriberId subscriber) (Subscription subscriber generation active) (entrySubscribers entry)}+                      writeTVar (coordinatorEntries coordinator) (Map.insert name updated entries)+  where+    cfg = coordinatorConfig coordinator+    unlessStopped True _ = pure ()+    unlessStopped False action = action++unsubscribeQueue :: MetricsCoordinator -> Subscriber -> Text -> IO ()+unsubscribeQueue coordinator subscriber name = do+  generation <- newUnique+  cleanup <- atomically $ do+    clearQueue (subscriberOutbox subscriber) name+    modifyTVar' (subscriberGenerations subscriber) (Map.delete name)+    entries <- readTVar (coordinatorEntries coordinator)+    case Map.lookup name entries of+      Nothing -> pure Nothing+      Just entry -> do+        let remaining = Map.delete (subscriberId subscriber) (entrySubscribers entry)+            empty = Map.null remaining+            updated =+              entry+                { entrySubscribers = remaining,+                  entryGeneration = if empty then generation else entryGeneration entry,+                  entryMetrics = if empty then Nothing else entryMetrics entry,+                  entryPushCleanup = if empty then Nothing else entryPushCleanup entry,+                  entryDirtyAt = if empty then Nothing else entryDirtyAt entry+                }+        writeTVar (coordinatorEntries coordinator) (Map.insert name updated entries)+        pure (if empty then entryPushCleanup entry else Nothing)+  sequence_ cleanup++releaseSubscriber :: MetricsCoordinator -> Subscriber -> IO ()+releaseSubscriber coordinator subscriber = do+  names <- Map.keys <$> readTVarIO (subscriberGenerations subscriber)+  mapM_ (unsubscribeQueue coordinator subscriber) names++worker :: IO Word64 -> MetricsCoordinator -> IO ()+worker clock coordinator = forever $ do+  token <- newUnique+  job@(Job name _ _ _ _) <- atomically (chooseJob coordinator token)+  now <- clock+  atomically $ modifyTVar' (coordinatorEntries coordinator) (Map.adjust (\entry -> entry {entryLastStart = Just now}) name)+  result <- try @SomeException $ withInspectionPermit (coordinatorBudget coordinator) (readMetrics coordinator name)+  outcome <- case result of+    Right Nothing -> pure (Left inspectionBusy)+    Right (Just value) -> pure value+    Left err+      | Just (_ :: SomeAsyncException) <- fromException err -> throwIO err+      | otherwise -> pure (Left (ErrorBody "database_error" "The database could not complete the inspection request." Nothing))+  case outcome of+    Right (queue, metrics) -> do+      ensurePush coordinator job queue+      atomically (completeJob coordinator job (Right metrics)) >>= sequence_+    Left err -> atomically (completeJob coordinator job (Left err)) >>= sequence_++chooseJob :: MetricsCoordinator -> Unique -> STM Job+chooseJob coordinator token = do+  stopped <- readTVar (coordinatorStopped coordinator)+  check (not stopped)+  now <- readTVar (coordinatorClock coordinator)+  entries <- readTVar (coordinatorEntries coordinator)+  let eligible (_, entry) = not (Map.null (entrySubscribers entry)) && not (isJust (entryFlight entry)) && cooled cfg now entry && due entry+      due entry = any (\(Subscription _ _ active) -> not active) (Map.elems (entrySubscribers entry)) || entryPollAt entry <= now || maybe False (\dirty -> now >= dirty && now - dirty >= fromIntegral (notifyDebounceUs cfg)) (entryDirtyAt entry)+      candidates = sortOn (\(name, entry) -> (entryLastStart entry, name)) (filter eligible (Map.toList entries))+  case candidates of+    [] -> retry+    (name, entry) : _ -> do+      let initial = not (isJust (entryMetrics entry))+          pollDue = entryPollAt entry <= now+          hintDue = maybe False (\dirty -> now >= dirty && now - dirty >= fromIntegral (notifyDebounceUs cfg)) (entryDirtyAt entry)+          -- A single authoritative read may satisfy both drivers. Preserve the+          -- notify tag while also advancing the independently due poll deadline.+          source = if hintDue && not initial then NotifySource else PollSource+          subscribers = Map.map (\(Subscription _ generation _) -> generation) (entrySubscribers entry)+          updated =+            entry+              { entryFlight = Just token,+                entryLastStart = Just now,+                entryDirtyAt = Nothing,+                entryPollAt = if initial || pollDue then now + fromIntegral (pollIntervalUs cfg) else entryPollAt entry+              }+      writeTVar (coordinatorEntries coordinator) (Map.insert name updated entries)+      pure (Job name (entryGeneration entry) token subscribers source)+  where+    cfg = coordinatorConfig coordinator++readMetrics :: MetricsCoordinator -> Text -> IO (Either ErrorBody (UnvalidatedQueue, QueueMetrics))+readMetrics coordinator name = do+  outcome <- runInspection env $ do+    queues <- E.listQueuesUnvalidated+    case find ((== name) . unvalidatedName) queues of+      Nothing -> pure (Left (queueNotFound name))+      Just queue -> Right . (queue,) <$> E.queueMetricsUnvalidated name+  case outcome of+    Right value -> pure value+    Left err | sqlState err == Just "42P01" -> do+      checked <- runInspection env E.listQueuesUnvalidated+      pure $ case checked of+        Right queues | not (any ((== name) . unvalidatedName) queues) -> Left (queueNotFound name)+        _ -> Left (snd (classifyRuntimeError err))+    Left err -> pure (Left (snd (classifyRuntimeError err)))+  where+    env = coordinatorEnv coordinator++ensurePush :: MetricsCoordinator -> Job -> UnvalidatedQueue -> IO ()+ensurePush coordinator (Job name generation token _ _) queue =+  case (coordinatorPush coordinator, parseQueueName name) of+    (Just hooks, Right validated) | not (unvalidatedIsPartitioned queue) -> mask $ \restore -> do+      needed <- atomically $ do+        entries <- readTVar (coordinatorEntries coordinator)+        pure $ case Map.lookup name entries of+          Just entry -> entryGeneration entry == generation && entryFlight entry == Just token && not (isJust (entryPushCleanup entry)) && not (Map.null (entrySubscribers entry))+          _ -> False+      when needed $ do+        cleanup <- registerPush hooks (notifyChannelName validated) (markDirty coordinator name generation)+        retained <- atomically $ do+          stopped <- readTVar (coordinatorStopped coordinator)+          entries <- readTVar (coordinatorEntries coordinator)+          case Map.lookup name entries of+            Just entry | not stopped && entryGeneration entry == generation && entryFlight entry == Just token && not (Map.null (entrySubscribers entry)) -> do+              writeTVar (coordinatorEntries coordinator) (Map.insert name entry {entryPushCleanup = Just cleanup} entries)+              pure True+            _ -> pure False+        if retained then pure () else restore cleanup+    _ -> pure ()++markDirty :: MetricsCoordinator -> Text -> Generation -> STM ()+markDirty coordinator name generation = do+  now <- readTVar (coordinatorClock coordinator)+  modifyTVar' (coordinatorEntries coordinator) (Map.adjust (\entry -> if entryGeneration entry == generation then entry {entryDirtyAt = Just (maybe now id (entryDirtyAt entry))} else entry) name)++completeJob :: MetricsCoordinator -> Job -> Either ErrorBody QueueMetrics -> STM (Maybe (IO ()))+completeJob coordinator (Job name generation token participants source) outcome = do+  stopped <- readTVar (coordinatorStopped coordinator)+  entries <- readTVar (coordinatorEntries coordinator)+  now <- readTVar (coordinatorClock coordinator)+  case Map.lookup name entries of+    Just entry | entryFlight entry == Just token -> do+      let eligible subscriber current = maybe True (== current) (Map.lookup (subscriberId subscriber) participants)+          publish (Subscription subscriber current active) = do+            let tag = (subscriberGenerations subscriber, name, current)+            case outcome of+              Left err+                | errorCode err == "queue_not_found" -> pure ()+                | otherwise -> do+                    lastFault <- readTVar (subscriberFaultAt subscriber)+                    let transient = errorCode err `elem` ["inspection_busy", "inspection_timeout", "database_unavailable"]+                        elapsed = maybe True (\prior -> now >= prior && now - prior >= fromIntegral (pollIntervalUs (coordinatorConfig coordinator))) lastFault+                    when (not transient || elapsed) $ do+                      writeTVar (subscriberFaultAt subscriber) (Just now)+                      enqueueTagged (subscriberOutbox subscriber) (Just tag) (ErrorFrame (errorCode err) (errorMessage err) (Just name))+              Right metrics+                | not active -> enqueueTagged (subscriberOutbox subscriber) (Just tag) (Snapshot name metrics (isJust (entryPushCleanup entry)))+                | source == NotifySource || maybe True ((/= fingerprint metrics) . fingerprint) (entryMetrics entry) -> enqueueTagged (subscriberOutbox subscriber) (Just tag) (Update name metrics source)+                | otherwise -> pure ()+      when (not stopped && entryGeneration entry == generation) $+        forM_ (Map.elems (entrySubscribers entry)) $+          \sub@(Subscription subscriber current _) -> when (eligible subscriber current) (publish sub)+      let valid = not stopped && entryGeneration entry == generation+          successful = case outcome of Right _ -> True; _ -> False+          subscribers = Map.map (\(Subscription subscriber current active) -> if valid && successful && eligible subscriber current then Subscription subscriber current True else Subscription subscriber current active) (entrySubscribers entry)+          updated =+            entry+              { entryFlight = Nothing,+                entrySubscribers = subscribers,+                entryMetrics = if valid then either (const (entryMetrics entry)) Just outcome else entryMetrics entry,+                entryDirtyAt = case outcome of Left _ | valid -> Just now; _ -> entryDirtyAt entry+              }+      -- Confirmed disappearance removes the subscription, but never a global 42P01.+      if valid && either ((== "queue_not_found") . errorCode) (const False) outcome+        then do+          forM_ (Map.elems subscribers) $ \(Subscription subscriber current _) -> modifyTVar' (subscriberGenerations subscriber) (\gs -> if Map.lookup name gs == Just current then Map.delete name gs else gs)+          -- Errors above remain visible even though the generation is removed.+          forM_ (Map.elems subscribers) $ \(Subscription subscriber _ _) -> enqueue (subscriberOutbox subscriber) (ErrorFrame "queue_not_found" "No queue with that name exists." (Just name))+          writeTVar (coordinatorEntries coordinator) (Map.insert name updated {entrySubscribers = Map.empty, entryMetrics = Nothing, entryDirtyAt = Nothing, entryPushCleanup = Nothing} entries)+          pure (entryPushCleanup entry)+        else writeTVar (coordinatorEntries coordinator) (Map.insert name updated entries) >> pure Nothing+    _ -> pure Nothing++fingerprint :: QueueMetrics -> Value+fingerprint metrics = case toJSON metrics of Object fields -> Object (KM.delete "scrape_time" fields); value -> value
+ src/Pgmq/Inspect/Server.hs view
@@ -0,0 +1,147 @@+-- | The mountable application and the convenience runner.+--+-- 'withInspectApplication' hands the host a WAI 'Application' to mount+-- anywhere; the host that mounts it under a prefix must strip the prefix+-- from 'Network.Wai.pathInfo' before delegating, as every WAI mounting helper+-- does. 'withInspectServer' and 'runInspectServer' serve it with Warp.+module Pgmq.Inspect.Server+  ( withInspectApplication,+    withInspectApplicationWith,+    InspectRuntime (..),+    withInspectRuntime,+    rejectingWebSocketApp,+    RunningInspectServer (..),+    withInspectServer,+    runInspectServer,+  )+where++import Control.Concurrent.Async (Async, wait)+import Control.Concurrent.STM+import Control.Exception (finally)+import Control.Monad (void)+import Data.Aeson (encode)+import Data.ByteString.Lazy qualified as LBS+import Data.Text (Text)+import GHC.Clock (getMonotonicTimeNSec)+import Hasql.Connection.Settings qualified+import Network.Wai (Application)+import Network.WebSockets qualified as WS+import Pgmq.Inspect.Budget (withInspectionBudget)+import Pgmq.Inspect.Config (InspectConfig (..), InspectServerConfig (..), validateInspectConfig)+import Pgmq.Inspect.Cors (corsMiddleware)+import Pgmq.Inspect.Env (InspectEnv (..))+import Pgmq.Inspect.Error (websocketUnavailable)+import Pgmq.Inspect.Http (httpApp)+import Pgmq.Inspect.Lifecycle qualified as Lifecycle+import Pgmq.Inspect.Listener qualified as Listener+import Pgmq.Inspect.Refresh (PushHooks (..), stopCoordinator, withCoordinator)+import Pgmq.Inspect.WebSocket qualified as Feed++-- | The mountable HTTP application and shared metrics feed. Its optional LISTEN+-- connection, scheduler and transports are released when the continuation exits.+withInspectApplication :: InspectConfig -> InspectEnv -> (Application -> IO a) -> IO a+withInspectApplication cfg env k = withInspectRuntime cfg env (k . runtimeApplication)++-- | Shared lifetime of HTTP admission and the metrics feed. Listener health is+-- independent of readiness; a reconnecting listener never disables polling.+data InspectRuntime = InspectRuntime+  { runtimeApplication :: !Application,+    runtimeRequestShutdown :: !(IO ()),+    runtimeAwaitDrained :: !(Int -> IO ()),+    -- | Current-session LISTEN acknowledgement for a channel. Resolve channel+    -- names with notifyChannelName; health alone does not acknowledge a channel.+    runtimeListenerAcknowledged :: !(Text -> STM Bool),+    -- | Connection health, independent of HTTP readiness and trigger existence.+    runtimeListenerHealthy :: !(STM Bool)+  }++withInspectRuntime :: InspectConfig -> InspectEnv -> (InspectRuntime -> IO a) -> IO a+withInspectRuntime cfg env k = do+  checked cfg+  optionalListener (listenerSettings env) $ \listener ->+    withInspectionBudget cfg $ \budget ->+      withCoordinator cfg env budget (pushHooks <$> listener) $ \coordinator -> do+        state <- Feed.newWebSocketState (wsMaxConnections cfg)+        shutdownStarted <- newTVarIO Nothing+        let shutdown = do+              now <- getMonotonicTimeNSec+              atomically $ do+                stopCoordinator coordinator+                Feed.requestShutdown state+                started <- readTVar shutdownStarted+                case started of+                  Nothing -> writeTVar shutdownStarted (Just now)+                  Just _ -> pure ()+            runtime =+              InspectRuntime+                { runtimeApplication = corsMiddleware (corsPolicy cfg) (httpApp budget cfg env (Feed.websocketApp cfg coordinator state)),+                  runtimeRequestShutdown = shutdown,+                  runtimeAwaitDrained = Feed.awaitDrained state,+                  runtimeListenerAcknowledged = maybe (const (pure False)) Listener.listenerAcknowledged listener,+                  runtimeListenerHealthy = maybe (pure False) Listener.listenerHealthy listener+                }+            cleanup = do+              shutdown+              now <- getMonotonicTimeNSec+              started <- readTVarIO shutdownStarted+              let elapsedUs = maybe 0 (\start -> min 1000000 ((now - start) `div` 1000)) started+              Feed.awaitDrained state (1000000 - fromIntegral elapsedUs)+              Feed.closeConnections state+        k runtime `finally` cleanup++-- | The application with an explicit WebSocket seam.+withInspectApplicationWith :: InspectConfig -> InspectEnv -> WS.ServerApp -> (Application -> IO a) -> IO a+withInspectApplicationWith cfg env wsApp k =+  checked cfg+    >> withInspectionBudget+      cfg+      ( \budget ->+          k (corsMiddleware (corsPolicy cfg) (httpApp budget cfg env wsApp))+      )++-- | Answer every upgrade with an ordinary 503 carrying the JSON error envelope.+rejectingWebSocketApp :: WS.ServerApp+rejectingWebSocketApp pending =+  WS.rejectRequestWith+    pending+    WS.defaultRejectRequest+      { WS.rejectCode = 503,+        WS.rejectMessage = "Service Unavailable",+        WS.rejectHeaders = [("Content-Type", "application/json")],+        WS.rejectBody = LBS.toStrict (encode websocketUnavailable)+      }++data RunningInspectServer = RunningInspectServer+  { -- | The bound port; informative when 'bindPort' was @0@.+    serverPort :: !Int,+    serverThread :: !(Async ())+  }++-- | Bind, acknowledge startup, and serve within the continuation’s scope.+withInspectServer :: InspectServerConfig -> InspectEnv -> (RunningInspectServer -> IO a) -> IO a+withInspectServer scfg env k = do+  checked (inspect scfg)+  Lifecycle.withBoundSocket scfg $ \socket ->+    withInspectRuntime (inspect scfg) env $ \runtime ->+      Lifecycle.withRunningSocket+        scfg+        socket+        (runtimeApplication runtime)+        (runtimeRequestShutdown runtime >> runtimeAwaitDrained runtime 1000000)+        (\port worker -> k (RunningInspectServer port worker))++runInspectServer :: InspectServerConfig -> InspectEnv -> IO ()+runInspectServer scfg env = withInspectServer scfg env (void . wait . serverThread)++checked :: InspectConfig -> IO ()+checked = either (ioError . userError . show) (const (pure ())) . validateInspectConfig++optionalListener :: Maybe Hasql.Connection.Settings.Settings -> (Maybe Listener.Listener -> IO a) -> IO a+optionalListener Nothing k = k Nothing+optionalListener (Just settings) k = Listener.withListener settings (k . Just)++pushHooks :: Listener.Listener -> PushHooks+pushHooks listener = PushHooks $ \channel wake -> do+  key <- Listener.subscribeChannel listener channel wake+  pure (Listener.unsubscribeChannel listener channel key)
+ src/Pgmq/Inspect/WebSocket.hs view
@@ -0,0 +1,162 @@+-- | Queue metrics feed at @/ws@. Client frames are+-- @{"type":"subscribe","queue":"orders"}@, @unsubscribe@ with the same queue,+-- and @{"type":"ping"}@. A successful subscribe publishes a @snapshot@ before+-- any @update@; snapshots carry @queue@, @metrics@ and @push@, updates carry+-- @queue@, @metrics@ and @source@ (@poll@ or @notify@). No messages are streamed.+--+-- @push@ means notification interest was registered for a validated,+-- non-partitioned queue, not that a listener or trigger is healthy. Polling is+-- authoritative even when every notification is lost. A runtime shares one+-- rate-limited metrics read per distinct queue across connections.+--+-- @pong@ answers a JSON ping; WebSocket control pings run every 30 seconds.+-- @error@ carries @code@, safe @message@ and an always-present nullable @queue@.+-- Bounded outboxes drop the oldest frame and report @overflow@ before the next+-- retained frame; re-read HTTP or re-subscribe after loss. Unsubscribe invalidates+-- queued generations. Bytes already handed to the socket writer cannot be recalled.+-- @goodbye@ precedes a server close, with a bounded best-effort shutdown drain.+--+-- Server JSON examples (QueueMetrics uses its owning domain encoding):+--+-- > {"type":"snapshot","queue":"orders","metrics":{"queue_name":"orders","queue_length":3,"newest_msg_age_sec":1,"oldest_msg_age_sec":9,"total_messages":12,"scrape_time":"2026-09-30T12:00:00Z","queue_visible_length":2,"default_partition_length":null},"push":true}+--+-- > {"type":"update","queue":"orders","metrics":{"queue_name":"orders","queue_length":3,"newest_msg_age_sec":1,"oldest_msg_age_sec":9,"total_messages":12,"scrape_time":"2026-09-30T12:00:00Z","queue_visible_length":2,"default_partition_length":null},"source":"notify"}+--+-- > {"type":"pong"}+--+-- > {"type":"error","code":"overflow","message":"The server dropped frames because this connection was not reading them; re-subscribe or re-read the queue over HTTP.","queue":null}+--+-- > {"type":"goodbye"}+module Pgmq.Inspect.WebSocket+  ( module Pgmq.Inspect.Frames,+    WebSocketState,+    newWebSocketState,+    requestShutdown,+    activeConnections,+    awaitDrained,+    closeConnections,+    websocketApp,+  )+where++import Control.Concurrent.Async+import Control.Concurrent.STM+import Control.Exception+import Control.Monad (forever, unless, when)+import Data.Aeson (eitherDecode, encode)+import Data.ByteString.Lazy qualified as LBS+import Data.List (isInfixOf)+import Data.Map.Strict qualified as Map+import Data.Unique (Unique, newUnique)+import Network.WebSockets qualified as WS+import Pgmq.Inspect.Config+import Pgmq.Inspect.Error+import Pgmq.Inspect.Frames+import Pgmq.Inspect.Outbox+import Pgmq.Inspect.Refresh+import System.Timeout (timeout)++-- | Connection count includes handshakes. Tasks are registered before serving.+data WebSocketState = WebSocketState+  { stateLimit :: !Int,+    stateCount :: !(TVar Int),+    stateShutdown :: !(TVar Bool),+    stateTasks :: !(TVar (Map.Map Unique (Async ())))+  }++newWebSocketState :: Int -> IO WebSocketState+newWebSocketState limit = WebSocketState limit <$> newTVarIO 0 <*> newTVarIO False <*> newTVarIO Map.empty++requestShutdown :: WebSocketState -> STM ()+requestShutdown = flip writeTVar True . stateShutdown++activeConnections :: WebSocketState -> STM Int+activeConnections = readTVar . stateCount++awaitDrained :: WebSocketState -> Int -> IO ()+awaitDrained state budget = do+  _ <- timeout budget (atomically (activeConnections state >>= check . (== 0)))+  pure ()++-- | Stop and join remaining transport/handshake tasks, also for host-owned Warp.+closeConnections :: WebSocketState -> IO ()+closeConnections state = readTVarIO (stateTasks state) >>= mapM_ cancel . Map.elems++websocketApp :: InspectConfig -> MetricsCoordinator -> WebSocketState -> WS.ServerApp+websocketApp cfg coordinator state pending = mask $ \restore -> do+  slot <- atomically $ do+    count <- readTVar (stateCount state)+    stopped <- readTVar (stateShutdown state)+    if stopped || count >= stateLimit state+      then pure False+      else writeTVar (stateCount state) (count + 1) >> pure True+  if not slot+    then+      restore $+        WS.rejectRequestWith+          pending+          WS.defaultRejectRequest+            { WS.rejectCode = 429,+              WS.rejectMessage = "Too Many Requests",+              WS.rejectHeaders = [("Content-Type", "application/json")],+              WS.rejectBody = LBS.toStrict (encode inspectionBusy)+            }+    else flip finally (atomically (modifyTVar' (stateCount state) (subtract 1))) $ do+      key <- newUnique+      start <- newEmptyTMVarIO+      withAsync (atomically (takeTMVar start) >> restore serve) $ \task -> do+        registered <- atomically $ do+          stopped <- readTVar (stateShutdown state)+          if stopped+            then pure False+            else do+              modifyTVar' (stateTasks state) (Map.insert key task)+              putTMVar start ()+              pure True+        when registered $ restore (wait task) `finally` atomically (modifyTVar' (stateTasks state) (Map.delete key))+  where+    serve =+      ( do+          connection <- WS.acceptRequest pending+          subscriber <- newSubscriber (wsQueueCapacity cfg)+          transport connection subscriber `finally` releaseSubscriber coordinator subscriber+      )+        `catch` \(_ :: WS.ConnectionException) -> pure ()+    transport connection subscriber = WS.withPingThread connection 30 (pure ()) $ do+      let outbox = subscriberOutbox subscriber+          writer = do+            (overflow, frame) <- atomically (takeOutgoing outbox)+            when overflow (WS.sendTextData connection (encode overflowFrame))+            WS.sendTextData connection (encode frame)+            case frame of+              Goodbye -> WS.sendClose connection ("shutdown" :: LBS.ByteString)+              _ -> writer+          finishWithError code message = do+            atomically (enqueue outbox (ErrorFrame code message Nothing))+            -- Capacity one must still deliver the error before goodbye.+            atomically (queuedFrames outbox >>= check . (== 0))+            atomically (enqueue outbox Goodbye)+            atomically retry+          reader =+            ( forever $ do+                bytes <- WS.receiveData connection :: IO LBS.ByteString+                if LBS.length bytes > fromIntegral (wsMaxFrameBytes cfg)+                  then finishWithError "frame_too_large" "The client frame exceeded the configured size limit."+                  else case eitherDecode bytes of+                    Left _ -> atomically (enqueue outbox (ErrorFrame "invalid_frame" "Expected a subscribe, unsubscribe or ping JSON frame." Nothing))+                    Right frame -> do+                      stopped <- readTVarIO (stateShutdown state)+                      unless stopped $ case frame of+                        Subscribe name -> subscribeQueue coordinator subscriber name+                        Unsubscribe name -> unsubscribeQueue coordinator subscriber name+                        Ping -> atomically (enqueue outbox Pong)+            )+              `catch` \err -> case err of+                WS.ParseException message | "limit" `isInfixOf` message -> finishWithError "frame_too_large" "The client frame exceeded the configured size limit."+                WS.ParseException _ -> finishWithError "invalid_frame" "The WebSocket frame was invalid."+                _ -> throwIO (err :: WS.ConnectionException)+          shutdown = do+            atomically (readTVar (stateShutdown state) >>= check)+            atomically (enqueue outbox Goodbye)+            atomically retry+      withAsync writer $ \writerTask -> race_ (wait writerTask) (race_ reader shutdown)
+ src/Pgmq/Inspect/Wire.hs view
@@ -0,0 +1,84 @@+-- | Wire types that are the surface's own (not records of pgmq-core or+-- pgmq-hasql, which carry their own instances). Every encoding here follows+-- the policy in @docs/design/020-json-wire-encodings.md@: snake_case keys,+-- hand-written instances, published shapes frozen. 'Page' is the one place a+-- key may be absent: @next_cursor@ is omitted on the last page by convention.+module Pgmq.Inspect.Wire+  ( Page (..),+    ServiceDescriptor (..),+    LivenessBody (..),+    DependencyCheck (..),+    ReadinessBody (..),+  )+where++import Data.Aeson (ToJSON (..), object, (.=))+import Data.Int (Int64)+import Data.Text (Text)+import GHC.Generics (Generic)+import Pgmq.Types (MessageId)++-- | One page of a keyset-paged read.+data Page a = Page+  { items :: ![a],+    -- | The @msg_id@ to pass back as @from@ for the next page; 'Nothing' on the+    -- last page, in which case the key is omitted from the JSON.+    nextCursor :: !(Maybe MessageId)+  }+  deriving stock (Eq, Show, Generic)++instance (ToJSON a) => ToJSON (Page a) where+  toJSON page =+    object (("items" .= items page) : maybe [] (\c -> ["next_cursor" .= c]) (nextCursor page))++-- | What @GET /@ returns: enough for a client to find the rest.+data ServiceDescriptor = ServiceDescriptor+  { serviceName :: !Text,+    serviceVersion :: !Text,+    -- | Relative to the mount root, so it is right under any prefix.+    websocketPath :: !Text,+    openapiPath :: !Text+  }+  deriving stock (Eq, Show, Generic)++instance ToJSON ServiceDescriptor where+  toJSON d =+    object+      [ "service" .= serviceName d,+        "version" .= serviceVersion d,+        "websocket_path" .= websocketPath d,+        "openapi_path" .= openapiPath d+      ]++newtype LivenessBody = LivenessBody {alive :: Bool}+  deriving stock (Eq, Show, Generic)++instance ToJSON LivenessBody where+  toJSON b = object ["alive" .= alive b]++-- | One dependency the readiness probe checked.+data DependencyCheck = DependencyCheck+  { checkName :: !Text,+    checkHealthy :: !Bool,+    checkLatencyMs :: !(Maybe Int64),+    checkError :: !(Maybe Text)+  }+  deriving stock (Eq, Show, Generic)++instance ToJSON DependencyCheck where+  toJSON c =+    object+      [ "name" .= checkName c,+        "healthy" .= checkHealthy c,+        "latency_ms" .= checkLatencyMs c,+        "error" .= checkError c+      ]++data ReadinessBody = ReadinessBody+  { ready :: !Bool,+    checks :: ![DependencyCheck]+  }+  deriving stock (Eq, Show, Generic)++instance ToJSON ReadinessBody where+  toJSON b = object ["ready" .= ready b, "checks" .= checks b]
+ test/CliSpec.hs view
@@ -0,0 +1,106 @@+module CliSpec (tests, databaseTests) where++import Control.Concurrent.Async (mapConcurrently)+import Control.Concurrent.STM+import Control.Exception (bracket)+import Control.Monad (forM_)+import Data.List (isInfixOf)+import Data.Text qualified as T+import EphemeralPg qualified as Pg+import Hasql.Connection qualified as Connection+import Hasql.Decoders qualified as D+import Hasql.Pool qualified as Pool+import Hasql.Session qualified as Session+import Hasql.Statement (unpreparable)+import HttpSpec (session, withQueue)+import Options.Applicative (ParserResult (..), defaultPrefs, execParserPure)+import PgmqInspectCli+import System.Environment (getEnvironment)+import System.Exit (ExitCode (..))+import System.IO (hGetLine)+import System.Process+import System.Timeout (timeout)+import Test.Tasty+import Test.Tasty.HUnit+import TestServer++parse :: [String] -> IO Options+parse args = case execParserPure defaultPrefs parserInfo args of+  Success opts -> pure opts+  _ -> assertFailure "valid options did not parse"++tests :: TestTree+tests =+  testGroup+    "Standalone CLI"+    [ testCase "version/help do not need a database" $ forM_ ["--version", "--help"] $ \arg -> do+        (code, out, _) <- readProcessWithExitCode "pgmq-inspect" [arg] ""+        code @?= ExitSuccess+        assertBool "expected output" ((if arg == "--version" then "0.7.0.0" else "--database-url") `isInfixOf` out),+      testCase "URL errors are one line, exit 2, and never echo secrets" $ do+        inherited <- getEnvironment+        forM_ [[], ["--database-url", "password='sensitive url' not-valid"]] $ \args -> do+          let p = (proc "pgmq-inspect" args) {env = Just (filter ((/= "PGMQ_INSPECT_DATABASE_URL") . fst) inherited)}+          (code, _, err) <- readCreateProcessWithExitCode p ""+          code @?= ExitFailure 2+          length (lines err) @?= 1+          assertBool "secret sanitized" (not ("sensitive" `isInfixOf` err)),+      testCase "explicit URL takes precedence; absent environment is rejected" $ do+        resolveDatabaseUrl (Just "explicit") (Just "env") @?= Right "explicit"+        resolveDatabaseUrl Nothing (Just "env") @?= Right "env"+        assertBool "missing rejected" (either (const True) (const False) (resolveDatabaseUrl Nothing Nothing)),+      testCase "numeric and CORS bounds checked before connection allocation" $ do+        forM_ [["--port", "65536"], ["--pool-size", "0"], ["--poll-interval-ms", "0"], ["--poll-interval-ms", replicate 128 '9'], ["--cors-origin", "*"]] $ \args -> do+          opts <- parse args+          assertBool (show args) (either (const True) (const False) (standaloneConfig opts "dbname=unused"))+        forM_ [replicate 129 '9', "-1", "1x"] $ \raw -> case execParserPure defaultPrefs parserInfo ["--port", raw] of+          Failure _ -> pure ()+          _ -> assertFailure "unbounded/malformed option accepted"+    ]++databaseTests :: IO (Pool.Pool, Pg.Database) -> TestTree+databaseTests database =+  testGroup+    "Standalone database"+    [ testCase "actual binary uses environment URL, ephemeral port and CORS" $ do+        (pool, db) <- database+        inherited <- getEnvironment+        withQueue pool $ \_ -> do+          let p =+                (proc "pgmq-inspect" ["--port", "0", "--no-listener", "--cors-origin", "https://ui.example", "--poll-interval-ms", "1000"])+                  { env = Just (("PGMQ_INSPECT_DATABASE_URL", T.unpack (Pg.connectionString db)) : filter ((/= "PGMQ_INSPECT_DATABASE_URL") . fst) inherited),+                    std_out = CreatePipe+                  }+          withCreateProcess p $ \_ out _ _ -> case out of+            Nothing -> assertFailure "missing stdout pipe"+            Just output -> do+              line <- timeout 10000000 (hGetLine output) >>= maybe (assertFailure "startup timed out") pure+              let portNumber = read (reverse (takeWhile (/= ':') (reverse line)))+              assertBool "actual port" (portNumber > 0)+              r <- getWith portNumber "GET" [("Origin", "https://ui.example")] "/queues"+              replyStatus r @?= 200+              lookup "Access-Control-Allow-Origin" (replyHeaders r) @?= Just "https://ui.example"+              spec <- get portNumber "/openapi.json"+              replyStatus spec @?= 200,+      testCase "every pooled connection and listener settings enforce deadlines, preserve supplied GUC" $ do+        (_, db) <- database+        opts <- parse ["--pool-size", "4"]+        (_, settings) <- either (assertFailure . T.unpack) pure (standaloneConfig opts (Pg.connectionString db <> " options='-c work_mem=8MB -c statement_timeout=0'"))+        let probe = Session.statement () (unpreparable "SELECT current_setting('statement_timeout'), current_setting('lock_timeout'), current_setting('work_mem')" mempty (D.singleRow ((,,) <$> D.column (D.nonNullable D.text) <*> D.column (D.nonNullable D.text) <*> D.column (D.nonNullable D.text))))+        bracket (Pool.acquire (standalonePoolConfig opts settings)) Pool.release $ \pool -> do+          arrived <- newTVarIO (0 :: Int)+          let heldProbe = do+                row <- probe+                Session.onLibpqConnection+                  ( \raw -> do+                      atomically (modifyTVar' arrived (+ 1))+                      atomically (readTVar arrived >>= check . (== 4))+                      pure (Right (), raw)+                  )+                pure row+          values <- mapConcurrently (const (session pool heldProbe)) [1 .. 4 :: Int]+          values @?= replicate 4 ("2s", "250ms", "8MB")+        bracket (Connection.acquire settings >>= either (assertFailure . show) pure) Connection.release $ \conn -> do+          value <- Connection.use conn probe >>= either (assertFailure . show) pure+          value @?= ("2s", "250ms", "8MB")+    ]
+ test/ConfigSpec.hs view
@@ -0,0 +1,33 @@+module ConfigSpec (tests) where++import Data.Either (isLeft)+import Data.List.NonEmpty (NonEmpty (..))+import Pgmq.Inspect.Config+import Test.Tasty+import Test.Tasty.HUnit++tests :: TestTree+tests =+  testGroup+    "Checked configuration"+    [ testCase "defaults are valid" $ validateInspectConfig defaultInspectConfig @?= Right defaultInspectConfig,+      testGroup+        "invalid values rejected before allocation"+        [ testCase label $ assertBool label (isLeft (validateInspectConfig cfg))+        | (label, cfg) <-+            [ ("zero default", defaultInspectConfig {defaultPageLimit = 0}),+              ("negative default", defaultInspectConfig {defaultPageLimit = -1}),+              ("default above max", defaultInspectConfig {defaultPageLimit = 501}),+              ("zero max", defaultInspectConfig {maxPageLimit = 0}),+              ("lookahead overflow", defaultInspectConfig {maxPageLimit = maxBound}),+              ("zero outbox", defaultInspectConfig {wsQueueCapacity = 0}),+              ("zero reads", defaultInspectConfig {maxConcurrentReads = 0}),+              ("negative interval", defaultInspectConfig {pollIntervalUs = -1}),+              ("wildcard", defaultInspectConfig {corsPolicy = CorsAllowOrigins ("*" :| [])}),+              ("wildcard host", defaultInspectConfig {corsPolicy = CorsAllowOrigins ("http://*" :| [])}),+              ("credentials", defaultInspectConfig {corsPolicy = CorsAllowOrigins ("http://user@host" :| [])}),+              ("origin path", defaultInspectConfig {corsPolicy = CorsAllowOrigins ("http://host/path" :| [])}),+              ("bad port", defaultInspectConfig {corsPolicy = CorsAllowOrigins ("http://host:65536" :| [])})+            ]+        ]+    ]
+ test/CorsSpec.hs view
@@ -0,0 +1,32 @@+module CorsSpec (tests) where++import Data.List.NonEmpty (NonEmpty (..))+import Pgmq.Inspect.Config+import RouterSpec (mockEnv)+import Test.Tasty+import Test.Tasty.HUnit+import TestServer++tests :: TestTree+tests =+  testGroup+    "CORS"+    [ testCase "no CORS headers when disabled" $ withTestServer defaultInspectConfig mockEnv $ \port -> do+        r <- getWith port "GET" [("Origin", "http://ui.example")] "/queues"+        replyStatus r @?= 200+        lookup "Access-Control-Allow-Origin" (replyHeaders r) @?= Nothing,+      testCase "a listed origin gets Access-Control-Allow-Origin and preflight" $ withTestServer cfg mockEnv $ \port -> do+        r <- getWith port "GET" [("Origin", "http://ui.example")] "/queues"+        lookup "Access-Control-Allow-Origin" (replyHeaders r) @?= Just "http://ui.example"+        lookup "Vary" (replyHeaders r) @?= Just "Origin"+        lookup "Access-Control-Allow-Credentials" (replyHeaders r) @?= Nothing+        preflight <- getWith port "OPTIONS" [("Origin", "http://ui.example"), ("Access-Control-Request-Method", "GET")] "/queues"+        replyStatus preflight @?= 200+        assertBool "allows GET" (lookup "Access-Control-Allow-Methods" (replyHeaders preflight) /= Nothing),+      testCase "an unlisted origin gets none" $ withTestServer cfg mockEnv $ \port -> do+        r <- getWith port "GET" [("Origin", "http://other.example")] "/queues"+        replyStatus r @?= 200+        lookup "Access-Control-Allow-Origin" (replyHeaders r) @?= Nothing+    ]+  where+    cfg = defaultInspectConfig {corsPolicy = CorsAllowOrigins ("http://ui.example" :| [])}
+ test/EmbeddingExample.hs view
@@ -0,0 +1,18 @@+-- | Compiled guide example: the host owns inspection capacity and its server.+module EmbeddingExample (mountUnder, withEmbeddedInspection) where++import Data.List (stripPrefix)+import Data.Text (Text)+import Network.HTTP.Types (status404)+import Network.Wai (Application, pathInfo, responseLBS)+import Pgmq.Inspect++mountUnder :: [Text] -> Application -> Application+mountUnder prefix app req respond =+  case stripPrefix prefix (pathInfo req) of+    Just rest -> app req {pathInfo = rest} respond+    Nothing -> respond (responseLBS status404 [] "Not found")++withEmbeddedInspection :: InspectConfig -> InspectEnv -> (Application -> IO a) -> IO a+withEmbeddedInspection config env k =+  withInspectApplication config env (k . mountUnder ["pgmq"])
+ test/EphemeralDb.hs view
@@ -0,0 +1,138 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Test database infrastructure using ephemeral-pg.+--+-- This is a copy of pgmq-hasql\'s EphemeralDb helper, adjusted for+-- pgmq-inspect\'s test suite. Keeping a local copy avoids cross-package+-- test-helper sharing (which would require a shared test-helper library+-- or fragile cross-package hs-source-dirs).+module EphemeralDb+  ( withPgmqDb,+    withPgmqPool,+    ephemeralConfig,+    ensureOwnerOnlyRoot,+    installPgmqNative,+    StartError,+    Database,+  )+where++import Control.Exception (bracket)+import Control.Monad (filterM, unless, when)+import Data.Bits ((.&.))+import Data.List.NonEmpty (NonEmpty (..))+import Data.Monoid (Last (..))+import Data.Text.IO qualified as TextIO+import Database.PostgreSQL.Migrate+  ( defaultRunOptions,+    migrationPlan,+    runMigrationPlan,+  )+import EphemeralPg+  ( Config (temporaryRoot),+    Database,+    StartError,+    connectionSettings,+    defaultCacheConfig,+    defaultConfig,+    withCachedConfig,+  )+import Hasql.Connection.Settings qualified as ConnectionSettings+import Hasql.Pool qualified as Pool+import Hasql.Pool.Config qualified as PoolConfig+import Hasql.Session qualified as Session+import Pgmq.Migration qualified as Migration+import System.Directory (doesFileExist)+import System.Environment (lookupEnv)+import System.IO.Error (catchIOError, isAlreadyExistsError)+import System.Posix.Directory qualified as Posix+import System.Posix.Files (deviceID, fileID, fileMode, fileOwner, getSymbolicLinkStatus, isDirectory, setFileMode)+import System.Posix.User (getEffectiveUserID)++-- | Root directory for ephemeral PostgreSQL clusters.+--+-- ephemeral-pg reaps abandoned clusters at startup, but only within its own+-- temporary root. With 'temporaryRoot' unset that root is @$TMPDIR@, which+-- @nix develop@ makes unique per shell, so a run never reclaims what an earlier+-- session abandoned. Pinning one root across sessions keeps them reachable.+--+-- The path is keyed by effective uid. It is a fixed location created @0700@, so+-- a run under a different user -- the Nix build sandbox -- must not collide with+-- a directory it cannot write to.+ephemeralRoot :: IO FilePath+ephemeralRoot = do+  uid <- getEffectiveUserID+  pure ("/tmp/ephpg-pgmq-hs-" <> show uid)++-- | Establish a real effective-uid-owned root without deleting its contents.+ensureOwnerOnlyRoot :: FilePath -> IO ()+ensureOwnerOnlyRoot root = do+  catchIOError (Posix.createDirectory root 0o700) $ \err ->+    unless (isAlreadyExistsError err) (ioError err)+  uid <- getEffectiveUserID+  before <- getSymbolicLinkStatus root+  unless (isDirectory before && fileOwner before == uid) $+    ioError (userError ("Unsafe ephemeral PostgreSQL root: " <> root))+  setFileMode root 0o700+  after <- getSymbolicLinkStatus root+  unless+    ( isDirectory after+        && fileOwner after == uid+        && deviceID before == deviceID after+        && fileID before == fileID after+        && fileMode after .&. 0o7777 == 0o700+    )+    (ioError (userError ("Ephemeral PostgreSQL root changed or is not owner-only: " <> root)))++-- | Cached-startup configuration pinned to 'ephemeralRoot'.+ephemeralConfig :: IO Config+ephemeralConfig = do+  root <- ephemeralRoot+  ensureOwnerOnlyRoot root+  pure defaultConfig {temporaryRoot = Last (Just root)}++-- | Run an action with a temporary PostgreSQL database that has the+-- pgmq schema installed.+withPgmqDb :: (Pool.Pool -> Database -> IO a) -> IO (Either StartError a)+withPgmqDb action =+  ephemeralConfig >>= \config ->+    withCachedConfig config defaultCacheConfig $ \db -> do+      let connSettings = connectionSettings db+          poolConfig =+            PoolConfig.settings+              [ PoolConfig.size 3,+                PoolConfig.staticConnectionSettings connSettings+              ]+      bracket (Pool.acquire poolConfig) Pool.release $ \pool -> do+        version <- lookupEnv "PGMQ_TEST_SCHEMA_VERSION"+        case version of+          Just "1.12.0" -> do+            paths <- filterM doesFileExist ["test/fixtures/pgmq-1.12.0.sql", "pgmq-inspect/test/fixtures/pgmq-1.12.0.sql"]+            path <- case paths of+              candidate : _ -> pure candidate+              [] -> error "Missing packaged PGMQ 1.12.0 test fixture"+            sql <- TextIO.readFile path+            Pool.use pool (Session.script sql) >>= either (error . show) pure+          Nothing -> installPgmqNative connSettings+          Just "1.13.0" -> installPgmqNative connSettings+          Just invalid -> error ("Invalid PGMQ_TEST_SCHEMA_VERSION: " <> invalid)+        required <- (== Just "1") <$> lookupEnv "PGMQ_REQUIRE_PARTMAN"+        -- Required runs fail for absent or unusable pg_partman, not just missing metadata.+        partman <- Pool.use pool (Session.script "CREATE SCHEMA IF NOT EXISTS partman; CREATE EXTENSION IF NOT EXISTS pg_partman SCHEMA partman")+        case partman of+          Left err -> when required (error ("PGMQ_REQUIRE_PARTMAN=1: " <> show err))+          Right () -> pure ()+        action pool db++-- | Apply the native pgmq migration ledger to the database at the given+-- connection settings, failing loudly on any error.+installPgmqNative :: ConnectionSettings.Settings -> IO ()+installPgmqNative connSettings = do+  component <- either (error . ("Invalid PGMQ migration component: " <>) . show) pure Migration.pgmqMigrations+  plan <- either (error . ("Invalid PGMQ migration plan: " <>) . show) pure (migrationPlan (component :| []))+  installResult <- runMigrationPlan defaultRunOptions connSettings plan+  either (error . ("Migration failed: " <>) . show) (const (pure ())) installResult++-- | Alias for 'withPgmqDb' kept for parallelism with pgmq-hasql.+withPgmqPool :: (Pool.Pool -> IO a) -> IO (Either StartError a)+withPgmqPool action = withPgmqDb (\pool _ -> action pool)
+ test/ErrorSpec.hs view
@@ -0,0 +1,124 @@+module ErrorSpec (tests, serverError, failingEnv) where++import Control.Concurrent.Async (wait, withAsync)+import Control.Concurrent.MVar+import Control.Exception (bracket)+import Control.Monad (void)+import Data.Aeson (Value (..))+import Data.ByteString.Lazy.Char8 qualified as LBS+import Data.IORef+import Data.List (isInfixOf)+import Data.Text (Text)+import Data.Time (UTCTime (..), fromGregorian)+import Effectful (liftIO)+import Effectful.Dispatch.Dynamic (interpret)+import Effectful.Error.Static (throwError)+import Hasql.Errors qualified as H+import Hasql.Pool qualified as Pool+import Hasql.Pool.Config qualified as PC+import Pgmq.Effectful (PgmqRuntimeError (..))+import Pgmq.Effectful.Effect (Pgmq (..))+import Pgmq.Inspect.Config+import Pgmq.Inspect.Env+import Pgmq.Types (UnvalidatedQueue (..))+import RouterSpec (errorCodeOf, lookupKey, mockEnv)+import System.Timeout (timeout)+import Test.Tasty+import Test.Tasty.HUnit+import TestServer++serverError :: Text -> PgmqRuntimeError+serverError code = PgmqSessionError (H.StatementSessionError 1 0 "SECRET_SQL" ["SECRET_PARAMS"] True (H.ServerStatementError (H.ServerError code "SECRET_MESSAGE" (Just "SECRET_DETAIL") (Just "SECRET_HINT") Nothing)))++failingEnv :: PgmqRuntimeError -> InspectEnv+failingEnv err = InspectEnv (runnerFromInterpreter (interpret (\_ _ -> throwError err))) Nothing++tests :: TestTree+tests =+  testGroup+    "Safe errors and shared admission"+    [ testCase "an unreachable database answers 503 database_unavailable" $+        bracket (Pool.acquire (PC.settings [PC.size 1, PC.staticConnectionSettings "host=127.0.0.1 port=1 user=nobody dbname=nonexistent connect_timeout=1"])) Pool.release $ \pool ->+          withTestServer defaultInspectConfig (poolEnv pool Nothing) $ \port -> do+            r <- get port "/queues"+            replyStatus r @?= 503+            errorCodeOf r >>= (@?= String "database_unavailable"),+      testCase "global missing schema/table remains 500; diagnostics never reach public responses" $+        mapM_+          ( \code -> withTestServer defaultInspectConfig (failingEnv (serverError code)) $ \port -> do+              r <- get port "/queues"+              replyStatus r @?= 500+              errorCodeOf r >>= (@?= String "database_error")+              assertBool "no raw diagnostics" (not ("SECRET" `isInfixOf` LBS.unpack (replyBody r)))+              ready <- get port "/health/ready"+              replyStatus ready @?= 503+              assertBool "readiness safe" (not ("SECRET" `isInfixOf` LBS.unpack (replyBody ready)))+          )+          ["42P01", "23505"],+      testCase "statement deadline maps to inspection_timeout" $+        withTestServer defaultInspectConfig (failingEnv (serverError "57014")) $ \port -> do+          r <- get port "/queues"+          replyStatus r @?= 503+          errorCodeOf r >>= (@?= String "inspection_timeout"),+      testCase "a missing queue answers 404 queue_not_found for every scoped route" $+        withTestServer defaultInspectConfig mockEnv $ \port ->+          mapM_+            ( \suffix -> do+                r <- get port ("/queues/missing" <> suffix)+                replyStatus r @?= 404+                errorCodeOf r >>= (@?= String "queue_not_found")+            )+            ["", "/metrics", "/bindings", "/messages", "/archive", "/messages/1", "/archive/1"],+      testCase "42P01 on an existing queue is database_error; a confirmed concurrent drop is queue_not_found" $+        mapM_+          ( \dropped -> do+              n <- newIORef (0 :: Int)+              let env =+                    InspectEnv+                      ( runnerFromInterpreter+                          ( interpret+                              ( \_ -> \case+                                  ListQueuesUnvalidated -> liftIO $ do+                                    calls <- atomicModifyIORef' n (\x -> (x + 1, x))+                                    pure [UnvalidatedQueue "orders" (UTCTime (fromGregorian 2026 1 1) 0) False False | calls == 0 || not dropped]+                                  QueueMetricsUnvalidated _ -> throwError (serverError "42P01")+                                  _ -> error "unexpected operation"+                              )+                          )+                      )+                      Nothing+              withTestServer defaultInspectConfig env $ \port -> do+                r <- get port "/queues/orders/metrics"+                replyStatus r @?= if dropped then 404 else 500+                errorCodeOf r >>= (@?= String (if dropped then "queue_not_found" else "database_error"))+          )+          [False, True],+      testCase "saturated admission rejects promptly and retains only bounded live work" $ do+        started <- newEmptyMVar+        release <- newEmptyMVar+        calls <- newIORef (0 :: Int)+        let env =+              InspectEnv+                ( \program -> do+                    atomicModifyIORef' calls (\n -> (n + 1, ()))+                    putMVar started ()+                    takeMVar release+                    runInspection mockEnv program+                )+                Nothing+        withTestServer defaultInspectConfig {maxConcurrentReads = 1} env $ \port ->+          withAsync (get port "/queues") $ \first -> do+            takeMVar started+            outcome <- timeout 500000 (get port "/queues")+            case outcome of+              Nothing -> assertFailure "admission queued a request"+              Just r -> (replyStatus r @?= 503) >> (errorCodeOf r >>= (@?= String "inspection_busy"))+            ready <- get port "/health/ready"+            replyStatus ready @?= 503+            body <- decodeReply ready+            lookupKey "ready" body @?= Just (Bool False)+            readIORef calls >>= (@?= 1)+            get port "/health/live" >>= (\r -> replyStatus r @?= 200)+            putMVar release ()+            void (wait first)+    ]
+ test/HttpSpec.hs view
@@ -0,0 +1,226 @@+module HttpSpec (tests, withQueue, sendN, session) where++import Control.Exception (bracket)+import Control.Lens ((^.))+import Control.Monad (void)+import Data.Aeson+import Data.Generics.Labels ()+import Data.Int (Int32, Int64)+import Data.Text qualified as T+import Data.Time (UTCTime)+import Data.Vector qualified as V+import EphemeralDb+import EphemeralPg qualified as Pg+import Hasql.Connection qualified as Connection+import Hasql.Decoders qualified as D+import Hasql.Pool qualified as Pool+import Hasql.Pool.Config qualified as PC+import Hasql.Session qualified as Session+import Hasql.Statement (unpreparable)+import Pgmq.Effectful qualified+import Pgmq.Hasql.Sessions qualified as S+import Pgmq.Hasql.Statements.Types qualified as H+import Pgmq.Inspect.Config+import Pgmq.Inspect.Env+import Pgmq.Types+import RouterSpec (errorCodeOf, lookupKey)+import System.Random (randomRIO)+import System.Timeout (timeout)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit+import TestServer++session :: Pool.Pool -> Session.Session a -> IO a+session pool action = Pool.use pool action >>= either (assertFailure . show) pure++withQueue :: Pool.Pool -> (QueueName -> IO a) -> IO a+withQueue pool k = do+  n <- randomRIO (1, 1000000000 :: Int)+  q <- either (error . show) pure (parseQueueName ("inspect_" <> T.pack (show n)))+  bracket (q <$ session pool (S.createQueue q)) (\name -> void (session pool (S.dropQueue name))) k++sendN :: Pool.Pool -> QueueName -> Int -> IO [MessageId]+sendN pool q n = session pool (S.batchSendMessage (H.BatchSendMessage q [MessageBody (object ["n" .= i]) | i <- [1 .. n]] Nothing))++baseFor :: QueueName -> String+baseFor q = "/queues/" <> T.unpack (queueNameToText q)++data PageOf a = PageOf [a] (Maybe MessageId)++instance (FromJSON a) => FromJSON (PageOf a) where+  parseJSON = withObject "PageOf" $ \o -> PageOf <$> o .: "items" <*> o .:? "next_cursor"++images :: QueueName -> Session.Session [(Int64, Int32, Maybe UTCTime, UTCTime)]+images q =+  Session.statement+    ()+    ( unpreparable+        ("select msg_id, read_ct, last_read_at, vt from pgmq.q_" <> queueNameToText q <> " order by msg_id")+        mempty+        (D.rowList ((,,,) <$> D.column (D.nonNullable D.int8) <*> D.column (D.nonNullable D.int4) <*> D.column (D.nullable D.timestamptz) <*> D.column (D.nonNullable D.timestamptz)))+    )++tests :: IO (Pool.Pool, Pg.Database) -> TestTree+tests database =+  testGroup+    "HTTP database integration"+    [ testCase "lock deadline bounds inspection while a consumer uses worker capacity" $ do+        result <- withPgmqDb $ \workers database' -> withQueue workers $ \q -> do+          void (sendN workers q 1)+          bracket (Connection.acquire (Pg.connectionSettings database') >>= either (assertFailure . show) pure) Connection.release $ \locker -> do+            Connection.use locker (Session.script "BEGIN; LOCK TABLE pgmq.meta IN ACCESS EXCLUSIVE MODE") >>= either (assertFailure . show) pure+            bracket (Pool.acquire (PC.settings [PC.size 1, PC.acquisitionTimeout 1, PC.staticConnectionSettings (Pg.connectionSettings database'), PC.initSession (Session.script "SET statement_timeout = '1s'; SET lock_timeout = '50ms'")])) Pool.release $ \inspectionPool ->+              withTestServer defaultInspectConfig (poolEnv inspectionPool Nothing) $ \port -> do+                outcome <- timeout 2000000 (get port "/queues")+                case outcome of+                  Nothing -> assertFailure "lock deadline failed"+                  Just r -> do+                    replyStatus r @?= 503+                    errorCodeOf r >>= (@?= String "database_unavailable")+                consumed <- timeout 500000 (session workers (S.readMessage (H.ReadMessage q 0 (Just 1) Nothing)))+                fmap V.length consumed @?= Just 1+        either (assertFailure . show) pure result,+      testCase "browsing over HTTP leaves read_ct at zero and every message available" $ do+        (pool, _) <- database+        withQueue pool $ \q -> do+          ids <- sendN pool q 5+          before <- session pool (images q)+          withTestServer defaultInspectConfig (poolEnv pool Nothing) $ \port -> do+            let base = baseFor q+            PageOf first cursor <- decodeReply =<< get port (base <> "/messages?limit=3") :: IO (PageOf Message)+            length first @?= 3+            cursor @?= Just (ids !! 2)+            PageOf second lastCursor <- decodeReply =<< get port (base <> "/messages?limit=3&from=" <> show (unMessageId (ids !! 2))) :: IO (PageOf Message)+            length second @?= 2+            lastCursor @?= Nothing+            r <- get port (base <> "/messages/" <> show (unMessageId (ids !! 0)))+            replyStatus r @?= 200+          after <- session pool (images q)+          before @?= after+          consumed <- session pool (S.readMessage (H.ReadMessage q 0 (Just 10) Nothing))+          V.length consumed @?= 5+          V.toList (V.map (^. #readCount) consumed) @?= [1, 1, 1, 1, 1],+      testCase "GET /queues lists the created queue and lookup returns it" $ do+        (pool, _) <- database+        withQueue pool $ \q -> withTestServer defaultInspectConfig (poolEnv pool Nothing) $ \port -> do+          rows <- decodeReply =<< get port "/queues" :: IO [Value]+          assertBool "listed" (Just (String (queueNameToText q)) `elem` map (lookupKey "queue_name") rows)+          row <- decodeReply =<< get port (baseFor q) :: IO Value+          lookupKey "queue_name" row @?= Just (String (queueNameToText q)),+      testCase "metrics distinguish total and visible depth after a lease" $ do+        (_, _) <- database+        withPgmqPool+          ( \p -> withQueue p $ \q -> do+              void (sendN p q 3)+              withTestServer defaultInspectConfig (poolEnv p Nothing) $ \port -> do+                initial <- decodeReply =<< get port (baseFor q <> "/metrics") :: IO Value+                lookupKey "queue_length" initial @?= Just (Number 3)+                lookupKey "queue_visible_length" initial @?= Just (Number 3)+                void (session p (S.readMessage (H.ReadMessage q 30 (Just 1) Nothing)))+                leased <- decodeReply =<< get port (baseFor q <> "/metrics") :: IO Value+                lookupKey "queue_length" leased @?= Just (Number 3)+                lookupKey "queue_visible_length" leased @?= Just (Number 2)+          )+          >>= either (assertFailure . show) pure,+      testCase "GET /metrics lists both queues on an isolated database" $ do+        (_, _) <- database+        withPgmqPool+          ( \p -> withQueue p $ \q -> withQueue p $ \q2 -> withTestServer defaultInspectConfig (poolEnv p Nothing) $ \port -> do+              rows <- decodeReply =<< get port "/metrics" :: IO [Value]+              let names = map (lookupKey "queue_name") rows+              assertBool "both queues" (all (\name -> Just (String (queueNameToText name)) `elem` names) [q, q2])+          )+          >>= either (assertFailure . show) pure,+      testCase "archive paging and by-id lookup retain archived_at" $ do+        (pool, _) <- database+        withQueue pool $ \q -> do+          ids <- sendN pool q 2+          mapM_ (\mid -> session pool (S.archiveMessage (H.MessageQuery q mid))) ids+          withTestServer defaultInspectConfig (poolEnv pool Nothing) $ \port -> do+            PageOf first cursor <- decodeReply =<< get port (baseFor q <> "/archive?limit=1") :: IO (PageOf ArchivedMessage)+            length first @?= 1+            cursor @?= Just (ids !! 0)+            PageOf lastPage next <- decodeReply =<< get port (baseFor q <> "/archive?limit=1&from=" <> show (unMessageId (ids !! 0))) :: IO (PageOf ArchivedMessage)+            length lastPage @?= 1+            next @?= Nothing+            row <- decodeReply =<< get port (baseFor q <> "/archive/" <> show (unMessageId (ids !! 0))) :: IO Value+            assertBool "has archived_at" (lookupKey "archived_at" row /= Nothing),+      testCase "valid missing IDs are message_not_found in either table" $ do+        (pool, _) <- database+        withQueue pool $ \q -> withTestServer defaultInspectConfig (poolEnv pool Nothing) $ \port ->+          mapM_+            ( \kind -> do+                r <- get port (baseFor q <> "/" <> kind <> "/9223372036854775807")+                replyStatus r @?= 404+                errorCodeOf r >>= (@?= String "message_not_found")+            )+            ["messages", "archive"],+      testCase "oversized page limit is capped with a lookahead cursor" $ do+        (pool, _) <- database+        withQueue pool $ \q -> do+          void (sendN pool q 4)+          withTestServer defaultInspectConfig {defaultPageLimit = 2, maxPageLimit = 3} (poolEnv pool Nothing) $ \port -> do+            PageOf rows cursor <- decodeReply =<< get port (baseFor q <> "/messages?limit=10000") :: IO (PageOf Message)+            length rows @?= 3+            assertBool "more" (cursor /= Nothing),+      testCase "bindings, routing and throttle routes use domain encodings" $ do+        (pool, _) <- database+        withQueue pool $ \q -> do+          pattern' <- either (error . show) pure (parseTopicPattern "orders.*")+          void (session pool (S.bindTopic (H.BindTopic pattern' q)))+          session pool (S.enableNotifyInsert (H.EnableNotifyInsert q (Just 250)))+          withTestServer defaultInspectConfig (poolEnv pool Nothing) $ \port -> do+            rows <- decodeReply =<< get port (baseFor q <> "/bindings") :: IO [Value]+            length rows @?= 1+            allRows <- decodeReply =<< get port "/bindings" :: IO [Value]+            assertBool "global binding" (all (`elem` allRows) rows)+            matches <- decodeReply =<< get port "/routing/test?routing_key=orders.created" :: IO [Value]+            assertBool "route match" (Just (String (queueNameToText q)) `elem` map (lookupKey "queue_name") matches)+            bad <- get port "/routing/test?routing_key=bad%20key"+            replyStatus bad @?= 400+            throttles <- decodeReply =<< get port "/notify/throttles" :: IO [Value]+            assertBool "250ms row" (any (\r -> lookupKey "queue_name" r == Just (String (queueNameToText q)) && lookupKey "throttle_interval_ms" r == Just (Number 250)) throttles),+      testCase "readiness includes a healthy postgres check and numeric latency" $ do+        (pool, _) <- database+        withTestServer defaultInspectConfig (poolEnv pool Nothing) $ \port -> do+          r <- get port "/health/ready"+          replyStatus r @?= 200+          row <- decodeReply r :: IO Value+          lookupKey "ready" row @?= Just (Bool True),+      testCase "mixed-case metadata names remain inspectable and encoded paths resolve" $ do+        (_, _) <- database+        withPgmqPool+          ( \p -> do+              session p (Session.script "select pgmq.create('MixedCase_q')")+              withTestServer defaultInspectConfig (poolEnv p Nothing) $ \port -> do+                mapM_ (\path -> get port path >>= (\r -> replyStatus r @?= 200)) ["/queues/MixedCase_q", "/queues/MixedCase_q/metrics", "/queues/MixedCase_q/messages", "/queues/%4dixedCase_q/archive"]+          )+          >>= either (assertFailure . show) pure,+      testCase "row cap does not cap payload bytes" $ do+        (pool, _) <- database+        withQueue pool $ \q -> do+          void (session pool (S.sendMessage (H.SendMessage q (MessageBody (String (T.replicate (1024 * 1024) "x"))) Nothing)))+          withTestServer defaultInspectConfig (poolEnv pool Nothing) $ \port -> do+            r <- get port (baseFor q <> "/messages?limit=1")+            replyStatus r @?= 200+            PageOf rows _ <- decodeReply r :: IO (PageOf Message)+            length rows @?= 1,+      testCase "dedicated inspection statement deadline preserves worker pool progress" $ do+        (pool, db) <- database+        bracket (Pool.acquire (PC.settings [PC.size 1, PC.acquisitionTimeout 1, PC.staticConnectionSettings (Pg.connectionSettings db), PC.initSession (Session.script "SET statement_timeout = '100ms'; SET lock_timeout = '50ms'")])) Pool.release $ \inspectionPool -> do+          -- The slow host runner uses only inspection capacity, never worker connections.+          let env =+                InspectEnv+                  ( \_ -> do+                      result <- Pool.use inspectionPool (Session.script "select pg_sleep(10)")+                      pure (either (Left . Pgmq.Effectful.fromUsageError) (const (error "slow query unexpectedly succeeded")) result)+                  )+                  Nothing+          withTestServer defaultInspectConfig env $ \port -> do+            r <- timeout 2000000 (get port "/queues")+            case r of+              Nothing -> assertFailure "statement timeout failed"+              Just reply -> errorCodeOf reply >>= (@?= String "inspection_timeout")+            worker <- timeout 500000 (session pool S.listQueuesUnvalidated)+            assertBool "worker pool progresses" (maybe False (const True) worker)+    ]
+ test/IsolationSpec.hs view
@@ -0,0 +1,80 @@+module IsolationSpec (tests) where++import Control.Concurrent.Async+import Control.Concurrent.MVar+import Control.Concurrent.STM+import Control.Exception (bracket, finally, uninterruptibleMask_)+import Control.Monad (replicateM_, void)+import EphemeralPg qualified as Pg+import Hasql.Pool qualified as Pool+import Hasql.Pool.Config qualified as PC+import Hasql.Session qualified as Session+import HttpSpec (session, withQueue)+import Network.Wai.Handler.Warp qualified as Warp+import Pgmq.Effectful (fromUsageError)+import Pgmq.Inspect+import Pgmq.Types (queueNameToText)+import System.Timeout (timeout)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit+import TestServer+import WebSocketSpec (sendClient, withWs)++tests :: IO (Pool.Pool, Pg.Database) -> TestTree+tests database =+  testGroup+    "WebSocket inspection isolation"+    [ testCase "uncancellable driver work stays bounded while worker capacity progresses" $ do+        (workers, db) <- database+        withQueue workers $ \queue ->+          bracket (Pool.acquire (PC.settings [PC.size 1, PC.acquisitionTimeout 1, PC.staticConnectionSettings (Pg.connectionSettings db), PC.initSession (Session.script "SET statement_timeout = '2000ms'; SET lock_timeout = '250ms'")])) Pool.release $ \inspection -> do+            entered <- newEmptyMVar+            release <- newEmptyMVar+            ready <- newEmptyMVar+            parked <- newEmptyMVar+            starts <- newTVarIO (0 :: Int)+            let base = poolEnv inspection Nothing+                env =+                  base+                    { runInspection = \program -> do+                        atomically (modifyTVar' starts (+ 1))+                        -- Model IR-6's uncancellable reply wait while holding a real,+                        -- dedicated pool connection. Always release this gate in finally.+                        outcome <-+                          Pool.use+                            inspection+                            ( Session.onLibpqConnection+                                ( \raw -> do+                                    uninterruptibleMask_ (putMVar entered () >> takeMVar release)+                                    pure (Right (), raw)+                                )+                            )+                        case outcome of+                          Left err -> pure (Left (fromUsageError err))+                          Right () -> runInspection base program+                    }+                config = defaultInspectConfig {maxConcurrentReads = 1}+            withAsync (withInspectRuntime config env (\runtime -> putMVar ready runtime >> takeMVar parked)) $ \owner -> do+              runtime <- takeMVar ready+              Warp.testWithApplication (pure (runtimeApplication runtime)) $ \port -> withWs port $ \connection -> do+                sendClient connection (Subscribe (queueNameToText queue))+                takeMVar entered+                withAsync (cancel owner) $ \canceller ->+                  ( do+                      pending <- timeout 100000 (wait canceller)+                      pending @?= Nothing+                      -- HTTP and background reads share the held permit. Rejections+                      -- must not create retries, threads or worker-pool acquisitions.+                      replicateM_ 10 $ do+                        reply <- get port "/queues"+                        replyStatus reply @?= 503+                      readTVarIO starts >>= (@?= 1)+                      progressed <- timeout 1000000 (session workers (Session.script "SELECT 1"))+                      progressed @?= Just ()+                  )+                    `finally` void (tryPutMVar release ())+              -- The released driver slice lets cancellation join every owned task.+              waitCatch owner >>= (\_ -> pure ())+            reused <- timeout 1000000 (session inspection (Session.script "SELECT 1"))+            reused @?= Just ()+    ]
+ test/ListenerSpec.hs view
@@ -0,0 +1,74 @@+module ListenerSpec (tests) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async (cancel, withAsync)+import Control.Concurrent.MVar+import Control.Concurrent.STM+import Control.Monad (forM_, replicateM, void)+import EphemeralDb (withPgmqDb)+import EphemeralPg qualified as Pg+import Hasql.Decoders qualified as D+import Hasql.Pool qualified as Pool+import Hasql.Session qualified as Session+import Hasql.Statement (unpreparable)+import HttpSpec (session)+import Pgmq.Inspect.Listener+import System.Timeout (timeout)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit++waitSTM :: STM a -> IO a+waitSTM action = timeout 2000000 (atomically action) >>= maybe (assertFailure "listener acknowledgement timed out") pure++backendCount :: Pool.Pool -> IO Int+backendCount pool = fromIntegral <$> session pool (Session.statement () (unpreparable "select count(*)::int from pg_stat_activity where application_name = 'pgmq-inspect-listener'" mempty (D.singleRow (D.column (D.nonNullable D.int4)))))++waitNoBackend :: Pool.Pool -> IO ()+waitNoBackend pool = timeout 2000000 loop >>= maybe (assertFailure "listener backend leaked") pure+  where+    loop = backendCount pool >>= \n -> if n == 0 then pure () else threadDelay 10000 >> loop++tests :: TestTree+tests =+  testGroup+    "WebSocket listener ownership"+    [ testCase "refcounted channels use one backend and UNLISTEN after last subscriber" $ do+        result <- withPgmqDb $ \pool database -> withListener (Pg.connectionSettings database) $ \listener -> do+          delivered <- newTVarIO (0 :: Int)+          keys <- replicateM 10 (subscribeChannel listener "inspect_test" (modifyTVar' delivered (+ 1)))+          waitSTM (listenerAcknowledged listener "inspect_test" >>= check)+          backendCount pool >>= (@?= 1)+          atomically (listenerChannelCount listener) >>= (@?= 1)+          unsubscribeChannel listener "inspect_test" (keys !! 0)+          session pool (Session.script "NOTIFY inspect_test")+          waitSTM (readTVar delivered >>= check . (== 9))+          forM_ (drop 1 keys) (unsubscribeChannel listener "inspect_test")+          waitSTM (listenerAcknowledged listener "inspect_test" >>= check . not)+          atomically (listenerChannelCount listener) >>= (@?= 0)+          session pool (Session.script "NOTIFY inspect_test")+          threadDelay 20000+          readTVarIO delivered >>= (@?= 9)+        either (assertFailure . show) pure result,+      testCase "cancellation during fd wait clears health, registrations and backend" $ do+        result <- withPgmqDb $ \pool database -> forM_ [1 .. 10 :: Int] $ \_ -> do+          ready <- newEmptyMVar+          withAsync (withListener (Pg.connectionSettings database) (\listener -> putMVar ready listener >> (newEmptyMVar >>= takeMVar))) $ \owner -> do+            listener <- takeMVar ready+            void (subscribeChannel listener "inspect_cancel" (pure ()))+            waitSTM (listenerAcknowledged listener "inspect_cancel" >>= check)+            cancel owner+            atomically (listenerHealthy listener) >>= (@?= False)+            atomically (listenerChannelCount listener) >>= (@?= 0)+            waitNoBackend pool+        either (assertFailure . show) pure result,+      testCase "buffered notification batches do not starve new channel acknowledgement" $ do+        result <- withPgmqDb $ \pool database -> withListener (Pg.connectionSettings database) $ \listener -> do+          delivered <- newTVarIO (0 :: Int)+          void (subscribeChannel listener "inspect_flood" (modifyTVar' delivered (+ 1)))+          waitSTM (listenerAcknowledged listener "inspect_flood" >>= check)+          session pool (Session.script "DO $$ BEGIN FOR i IN 1..300 LOOP PERFORM pg_notify('inspect_flood', i::text); END LOOP; END $$")+          void (subscribeChannel listener "inspect_new" (pure ()))+          waitSTM (listenerAcknowledged listener "inspect_new" >>= check)+          waitSTM (readTVar delivered >>= check . (== 300))+        either (assertFailure . show) pure result+    ]
+ test/LiveFeedSpec.hs view
@@ -0,0 +1,127 @@+module LiveFeedSpec (tests) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.STM+import Control.Monad (replicateM_, void)+import Data.Aeson (Value (..))+import Data.Text (Text)+import Data.Text qualified as T+import EphemeralDb (withPgmqDb)+import EphemeralPg qualified as Pg+import Hasql.Decoders qualified as D+import Hasql.Pool qualified as Pool+import Hasql.Session qualified as Session+import Hasql.Statement (unpreparable)+import HttpSpec (sendN, session, withQueue)+import Network.WebSockets qualified as WS+import Pgmq.Hasql.Sessions qualified as S+import Pgmq.Hasql.Statements.Types qualified as H+import Pgmq.Inspect+import Pgmq.Types (notifyChannelName, queueNameToText)+import RouterSpec (mockEnv)+import System.Timeout (timeout)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit+import TestServer+import WebSocketSpec hiding (tests)++cfg :: InspectConfig+cfg = defaultInspectConfig {pollIntervalUs = 200000, metricsMinRefreshUs = 100000, notifyDebounceUs = 20000}++waitAcknowledged :: InspectRuntime -> Text -> IO ()+waitAcknowledged runtime channel = timeout 10000000 (atomically (runtimeListenerAcknowledged runtime channel >>= check)) >>= maybe (assertFailure "LISTEN not acknowledged") pure++hasLength :: Int -> Value -> Bool+hasLength n frame = (frameField "metrics" frame >>= frameField "queue_length") == Just (Number (fromIntegral n))++requireUpdate :: WS.Connection -> (Value -> Bool) -> IO Value+requireUpdate connection predicate = awaitFrame 2000000 connection (\frame -> frameType frame == Just "update" && predicate frame) >>= maybe (assertFailure "feed update timed out") pure++tests :: IO (Pool.Pool, Pg.Database) -> TestTree+tests database =+  testGroup+    "WebSocket listener-accelerated feed"+    [ testCase "registered push interest and notify updates work with polling far away" $ do+        (pool, db) <- database+        withQueue pool $ \queue -> do+          session pool (S.enableNotifyInsert (H.EnableNotifyInsert queue (Just 0)))+          withTestRuntime cfg {pollIntervalUs = 10000000} (poolEnv pool (Just (Pg.connectionSettings db))) $ \port runtime -> withWs port $ \connection -> do+            sendClient connection (Subscribe (queueNameToText queue))+            snapshot <- receive connection+            frameField "push" snapshot @?= Just (Bool True)+            waitAcknowledged runtime (notifyChannelName queue)+            void (sendN pool queue 1)+            frame <- requireUpdate connection (hasLength 1)+            frameField "source" frame @?= Just (String "notify"),+      testCase "notify burst coalesces and fans out one refresh to both clients" $ do+        (pool, db) <- database+        withQueue pool $ \queue -> do+          session pool (S.enableNotifyInsert (H.EnableNotifyInsert queue (Just 0)))+          withTestRuntime cfg {pollIntervalUs = 10000000} (poolEnv pool (Just (Pg.connectionSettings db))) $ \port runtime -> withWs port $ \a -> withWs port $ \b -> do+            sendClient a (Subscribe (queueNameToText queue))+            sendClient b (Subscribe (queueNameToText queue))+            void (receive a)+            void (receive b)+            waitAcknowledged runtime (notifyChannelName queue)+            replicateM_ 5 (void (sendN pool queue 1))+            first <- requireUpdate a (hasLength 5)+            second <- requireUpdate b (hasLength 5)+            frameField "source" first @?= Just (String "notify")+            first @?= second+            let countExtra n+                  | n >= 4 = pure n+                  | otherwise = awaitFrame 200000 a (\frame -> frameType frame == Just "update") >>= maybe (pure n) (const (countExtra (n + 1)))+            extra <- countExtra (0 :: Int)+            assertBool "fewer than five notify updates for the burst" (extra < 4),+      testCase "push means registration even without a trigger; poll remains authoritative" $ do+        (pool, db) <- database+        withQueue pool $ \queue ->+          withTestRuntime cfg (poolEnv pool (Just (Pg.connectionSettings db))) $ \port runtime -> withWs port $ \connection -> do+            sendClient connection (Subscribe (queueNameToText queue))+            snapshot <- receive connection+            frameField "push" snapshot @?= Just (Bool True)+            waitAcknowledged runtime (notifyChannelName queue)+            void (sendN pool queue 1)+            update <- requireUpdate connection (hasLength 1)+            frameField "source" update @?= Just (String "poll"),+      testCase "listener connection failure does not disable snapshots or readiness" $+        withTestRuntime cfg mockEnv {listenerSettings = Just "host=127.0.0.1 port=1 user=nobody dbname=nonexistent connect_timeout=1"} $ \port runtime -> withWs port $ \connection -> do+          sendClient connection (Subscribe "orders")+          snapshot <- receive connection+          frameField "push" snapshot @?= Just (Bool True)+          atomically (runtimeListenerHealthy runtime) >>= (@?= False)+          get port "/health/ready" >>= (\reply -> replyStatus reply @?= 200),+      testCase "foreign mixed-case names are poll-only with a configured listener" $ do+        result <- withPgmqDb $ \pool db -> do+          session pool (Session.script "select pgmq.create('Mixed_Case')")+          withTestRuntime cfg (poolEnv pool (Just (Pg.connectionSettings db))) $ \port _ -> withWs port $ \connection -> do+            sendClient connection (Subscribe "Mixed_Case")+            snapshot <- receive connection+            frameField "push" snapshot @?= Just (Bool False)+            session pool (Session.script "select pgmq.send('Mixed_Case', '{\"x\":1}'::jsonb)")+            update <- requireUpdate connection (hasLength 1)+            frameField "source" update @?= Just (String "poll")+        either (assertFailure . show) pure result,+      testCase "killing the LISTEN backend preserves convergence and resumes notify pushes" $ do+        -- Isolated cluster: terminating this runtime's listener cannot disrupt other cases.+        result <- withPgmqDb $ \pool db -> withQueue pool $ \queue -> do+          session pool (S.enableNotifyInsert (H.EnableNotifyInsert queue (Just 0)))+          withTestRuntime cfg (poolEnv pool (Just (Pg.connectionSettings db))) $ \port runtime -> withWs port $ \connection -> do+            sendClient connection (Subscribe (queueNameToText queue))+            void (receive connection)+            waitAcknowledged runtime (notifyChannelName queue)+            oldPid <- session pool (Session.statement () (unpreparable "select pid::int from pg_stat_activity where application_name = 'pgmq-inspect-listener'" mempty (D.singleRow (D.column (D.nonNullable D.int4)))))+            killed <- session pool (Session.statement () (unpreparable "select count(pg_terminate_backend(pid))::int from pg_stat_activity where application_name = 'pgmq-inspect-listener'" mempty (D.singleRow (D.column (D.nonNullable D.int4)))))+            assertBool "terminated a listener" (killed >= 1)+            void (sendN pool queue 1)+            void (requireUpdate connection (hasLength 1))+            let waitReplacement = do+                  replaced <- session pool (Session.statement () (unpreparable ("select count(*)::int from pg_stat_activity where application_name = 'pgmq-inspect-listener' and pid <> " <> T.pack (show oldPid)) mempty (D.singleRow (D.column (D.nonNullable D.int4)))))+                  if replaced > 0 then pure () else threadDelay 10000 >> waitReplacement+            timeout 10000000 waitReplacement >>= maybe (assertFailure "listener did not reconnect") pure+            waitAcknowledged runtime (notifyChannelName queue)+            void (sendN pool queue 1)+            update <- requireUpdate connection (\frame -> hasLength 2 frame && frameField "source" frame == Just (String "notify"))+            frameField "source" update @?= Just (String "notify")+        either (assertFailure . show) pure result+    ]
+ test/Main.hs view
@@ -0,0 +1,66 @@+module Main (main) where++import CliSpec qualified+import ConfigSpec qualified+import Control.Concurrent.Async+import Control.Concurrent.STM+import Control.Exception+import CorsSpec qualified+import EphemeralDb+import ErrorSpec qualified+import HttpSpec qualified+import IsolationSpec qualified+import ListenerSpec qualified+import LiveFeedSpec qualified+import OpenApiSpec qualified+import OutboxSpec qualified+import PrivilegesSpec qualified+import RawWebSocketSpec qualified+import RefreshSpec qualified+import RootPermissionsSpec qualified+import RouterSpec qualified+import ServerSpec qualified+import SmokeSpec qualified+import Test.Tasty+import WebSocketSpec qualified+import WireGoldenSpec qualified++main :: IO ()+main =+  defaultMain $+    testGroup+      "pgmq-inspect"+      [ RootPermissionsSpec.tests ensureOwnerOnlyRoot,+        CliSpec.tests,+        ConfigSpec.tests,+        CorsSpec.tests,+        ErrorSpec.tests,+        RouterSpec.tests,+        WireGoldenSpec.tests,+        WebSocketSpec.tests,+        ListenerSpec.tests,+        OpenApiSpec.tests,+        OutboxSpec.tests,+        PrivilegesSpec.tests,+        RawWebSocketSpec.tests,+        RefreshSpec.tests,+        ServerSpec.tests,+        SmokeSpec.tests,+        withResource acquire release (\resource -> testGroup "Database" [OpenApiSpec.databaseTests (snd <$> resource), CliSpec.databaseTests (snd <$> resource), HttpSpec.tests (snd <$> resource), WebSocketSpec.databaseTests (snd <$> resource), LiveFeedSpec.tests (snd <$> resource), IsolationSpec.tests (snd <$> resource)])+      ]+  where+    acquire = mask $ \restore -> do+      started <- newEmptyTMVarIO+      stop <- newEmptyTMVarIO+      worker <- async $ restore $ do+        result <- try @SomeException $ withPgmqDb $ \pool db -> do+          atomically (putTMVar started (Right (pool, db)))+          atomically (takeTMVar stop)+        case result of+          Left e -> atomically (voidTryPut started (Left e))+          Right (Left e) -> atomically (voidTryPut started (Left (toException (userError (show e)))))+          Right (Right ()) -> pure ()+      pair <- restore (atomically (takeTMVar started) >>= either throwIO pure) `onException` cancel worker+      pure ((worker, stop), pair)+    release ((worker, stop), _) = atomically (putTMVar stop ()) >> wait worker+    voidTryPut var value = () <$ tryPutTMVar var value
+ test/OpenApiSpec.hs view
@@ -0,0 +1,175 @@+module OpenApiSpec (tests, databaseTests) where++import Control.Concurrent.Async (wait, withAsync)+import Control.Concurrent.MVar+import Control.Monad (forM_)+import Data.Aeson (Value (..), eitherDecode, object, (.=))+import Data.ByteString.Char8 qualified as BS+import Data.HashMap.Strict.InsOrd.Compat qualified as O+import Data.List (intercalate, sort)+import Data.OpenApi (Components (..), MediaTypeObject (..), OpenApi (..), OpenApiItems (..), OpenApiType (..), Operation (..), PathItem (..), Reference (..), Referenced (..), Response (..), Responses (..), Schema (..))+import Data.OpenApi.Schema.Validation (validateJSON)+import Data.Text qualified as T+import EmbeddingExample (mountUnder)+import EphemeralDb (withPgmqPool)+import EphemeralPg qualified as Pg+import ErrorSpec (failingEnv, serverError)+import Hasql.Pool qualified as Pool+import Hasql.Session qualified as Session+import HttpSpec (session, withQueue)+import Network.Wai.Handler.Warp qualified as Warp+import Network.WebSockets qualified as WS+import Pgmq.Inspect.Config+import Pgmq.Inspect.Env+import Pgmq.Inspect.Http+import Pgmq.Inspect.OpenApi+import Pgmq.Inspect.Server (withInspectApplication)+import Pgmq.Types (queueNameToText)+import RouterSpec (mockEnv)+import System.Exit (ExitCode (..))+import System.Process (readProcessWithExitCode)+import Test.Tasty+import Test.Tasty.Golden (NoCreateFile (..), goldenVsString)+import Test.Tasty.HUnit+import TestServer++coverage :: [(String, String)] -> Bool+coverage served = sort served == sort documentedRoutes++servedRoutes :: [(String, String)]+servedRoutes = [(BS.unpack method, "/" <> intercalate "/" (map segment segments)) | (method, segments) <- routeTable, segments /= [Literal "ws"]]+  where+    segment (Literal t) = T.unpack t+    segment (Placeholder t) = "{" <> T.unpack t <> "}"++documentedRoutes :: [(String, String)]+documentedRoutes =+  [ (method, path)+  | (path, item) <- O.toList (_openApiPaths openApiDocument),+    (method, op) <- [("GET", _pathItemGet item), ("POST", _pathItemPost item), ("PUT", _pathItemPut item), ("DELETE", _pathItemDelete item), ("HEAD", _pathItemHead item), ("OPTIONS", _pathItemOptions item), ("PATCH", _pathItemPatch item), ("TRACE", _pathItemTrace item)],+    Just _ <- [op]+  ]++-- Resolve the actual operation/status/media schema, rather than independently+-- choosing a domain schema from a fixture's filename.+-- openapi3 3.2.5's validator ignores the OAS 3 nullable keyword (verified in+-- Internal.Schema.Validation). Translate it ONLY in the validation view to its+-- supported union/null representation. The served schema remains valid OAS 3.0.+validationSchema :: Schema -> Schema+validationSchema schema =+  let ref (Inline s) = Inline (validationSchema s)+      ref r = r+      nested =+        schema+          { _schemaProperties = fmap ref (_schemaProperties schema),+            _schemaItems = fmap (\case OpenApiItemsObject r -> OpenApiItemsObject (ref r); OpenApiItemsArray rs -> OpenApiItemsArray (map ref rs)) (_schemaItems schema),+            _schemaNullable = Nothing+          }+   in if _schemaNullable schema == Just True+        then mempty {_schemaOneOf = Just [Inline nested, Inline mempty {_schemaType = Just OpenApiNull}]}+        else nested++validateReply :: String -> Reply -> IO ()+validateReply path reply = do+  value <- decodeReply reply+  let schemas = fmap validationSchema (_componentsSchemas (_openApiComponents openApiDocument))+      resolve (Inline s) = Just s+      resolve (Ref (Reference name)) = O.lookup name schemas+      schema = do+        item <- O.lookup path (_openApiPaths openApiDocument)+        op <- _pathItemGet item+        responseRef <- O.lookup (replyStatus reply) (_responsesResponses (_operationResponses op))+        response <- case responseRef of Inline r -> Just r; _ -> Nothing+        media <- O.lookup "application/json" (_responseContent response)+        _mediaTypeObjectSchema media >>= resolve+  s <- maybe (assertFailure ("no schema for " <> path <> " " <> show (replyStatus reply))) pure schema+  assertEqual (path <> " " <> show (replyStatus reply)) [] (validateJSON schemas (validationSchema s) value)++tests :: TestTree+tests =+  testGroup+    "OpenAPI contract"+    [ localOption (NoCreateFile True) $ goldenVsString "document golden" "test/golden/openapi.json" (pure openApiJson),+      testCase "HTTP paths and methods covered; fake route proves drift detection" $ do+        assertEqual "route coverage" (sort documentedRoutes) (sort servedRoutes)+        assertBool "extra route rejected" (not (coverage (("GET", "/forgotten") : servedRoutes)))+        assertBool "extra method rejected" (not (coverage (("POST", "/queues") : servedRoutes)))+        forM_ (O.elems (_openApiPaths openApiDocument)) $ \item -> do+          assertBool "GET exists" (maybe False (const True) (_pathItemGet item))+          assertBool "POST absent" (maybe True (const False) (_pathItemPost item)),+      testCase "served document round-trips and descriptor advertises it" $+        withTestServer defaultInspectConfig mockEnv $ \port -> do+          r <- get port "/openapi.json"+          replyBody r @?= openApiJson+          case eitherDecode (replyBody r) :: Either String OpenApi of+            Left e -> assertFailure e+            Right parsed -> parsed @?= openApiDocument+          get port "/" >>= validateReply "/",+      testCase "actual failures validate by status, including readiness 503" $ do+        withTestServer defaultInspectConfig mockEnv $ \port -> do+          forM_ [("/queues/{queue}", "/queues/absent"), ("/queues/{queue}/messages/{msg_id}", "/queues/orders/messages/0"), ("/queues/{queue}/messages", "/queues/orders/messages?from=bad"), ("/routing/test", "/routing/test?routing_key=bad%20key")] $ \(template, path) -> get port path >>= validateReply template+          getWith port "POST" [] "/queues" >>= validateReply "/queues"+        forM_ ["57014", "42P01", "55P03"] $ \code ->+          withTestServer defaultInspectConfig (failingEnv (serverError code)) $ \port -> do+            get port "/queues" >>= validateReply "/queues"+            get port "/health/ready" >>= validateReply "/health/ready",+      testCase "saturated readiness has nullable latency; document bypasses database admission" $ do+        started <- newEmptyMVar+        release <- newEmptyMVar+        let env = InspectEnv (\program -> putMVar started () >> takeMVar release >> runInspection mockEnv program) Nothing+        withTestServer defaultInspectConfig {maxConcurrentReads = 1} env $ \port ->+          withAsync (get port "/queues") $ \pending -> do+            takeMVar started+            ready <- get port "/health/ready"+            replyStatus ready @?= 503+            validateReply "/health/ready" ready+            document <- get port "/openapi.json"+            replyStatus document @?= 200+            putMVar release ()+            _ <- wait pending+            pure (),+      testCase "validator detects incompatible bodies and required keys" $ do+        let schemas = fmap validationSchema (_componentsSchemas (_openApiComponents openApiDocument))+        schema <- maybe (assertFailure "Message schema missing") pure (O.lookup "Message" schemas)+        let integer = mempty {_schemaType = Just OpenApiInteger}+        assertBool "non-null rejects null" (not (null (validateJSON mempty (validationSchema integer) Null)))+        validateJSON mempty (validationSchema integer {_schemaNullable = Just True}) Null @?= []+        assertBool "nullable still rejects text" (not (null (validateJSON mempty (validationSchema integer {_schemaNullable = Just True}) (String "wrong"))))+        assertBool "required fields enforced" (not (null (validateJSON schemas schema (String "wrong"))))+    ]++databaseTests :: IO (Pool.Pool, Pg.Database) -> TestTree+databaseTests database =+  testGroup+    "OpenAPI real responses and client"+    [ testCase "nonempty domain responses, arbitrary headers and both page forms validate" $ do+        result <- withPgmqPool $ \pool -> withQueue pool $ \q -> do+          let name = queueNameToText q+              base = "/queues/" <> T.unpack name+          session pool (Session.script ("INSERT INTO pgmq.q_" <> name <> " (msg_id,vt,message,headers) OVERRIDING SYSTEM VALUE VALUES (0, now(), 'true', NULL), (1, now(), '[1]', '{}'), (2, now(), '\"body\"', 'true'), (9007199254740993, now(), 'null', '[1,true]'), (9223372036854775807, now(), '42', '\"scalar\"'); SELECT pgmq.bind_topic('inspect.*', '" <> name <> "'); SELECT pgmq.enable_notify_insert('" <> name <> "',250)"))+          withTestServer defaultInspectConfig (poolEnv pool Nothing) $ \port -> do+            forM_ [("/", "/"), ("/queues", "/queues"), ("/queues/{queue}", base), ("/queues/{queue}/metrics", base <> "/metrics"), ("/metrics", "/metrics"), ("/queues/{queue}/messages", base <> "/messages?limit=1"), ("/queues/{queue}/messages", base <> "/messages?limit=500"), ("/queues/{queue}/messages", base <> "/messages?from=9007199254740993"), ("/queues/{queue}/messages/{msg_id}", base <> "/messages/9223372036854775807"), ("/queues/{queue}/bindings", base <> "/bindings"), ("/bindings", "/bindings"), ("/routing/test", "/routing/test?routing_key=inspect.created"), ("/notify/throttles", "/notify/throttles"), ("/health/live", "/health/live"), ("/health/ready", "/health/ready"), ("/openapi.json", "/openapi.json")] $ \(template, path) -> do+              r <- get port path+              assertEqual (path <> " " <> show (replyBody r)) 200 (replyStatus r)+              validateReply template r+            session pool (Session.script ("SELECT pgmq.archive('" <> name <> "',9007199254740993); SELECT pgmq.archive('" <> name <> "',9223372036854775807)"))+            forM_ [("/queues/{queue}/archive", base <> "/archive?limit=1"), ("/queues/{queue}/archive", base <> "/archive?from=9007199254740993"), ("/queues/{queue}/archive/{msg_id}", base <> "/archive/9223372036854775807")] $ \(template, path) -> get port path >>= validateReply template+        either (assertFailure . show) pure result,+      testCase "client generated from served operations preserves prefixes, encoded names and int64 tokens" $ do+        (pool, _) <- database+        withQueue pool $ \q -> do+          -- A space is accepted by upstream and exercises percent encoding; use+          -- an isolated table/name derived from this test's random queue.+          let name = "Foreign " <> queueNameToText q+          session pool (Session.script ("SELECT pgmq.create('" <> name <> "'); INSERT INTO pgmq.\"q_" <> T.toLower name <> "\" (msg_id,vt,message) OVERRIDING SYSTEM VALUE VALUES (9007199254740993, now(), '{}'),(9223372036854775807, now(), '{}')"))+          withInspectApplication defaultInspectConfig (poolEnv pool Nothing) $ \app ->+            forM_ ["", "/pgmq"] $ \prefix ->+              Warp.testWithApplication (pure (mountUnder (if null prefix then [] else ["pgmq"]) app)) $ \port -> do+                (code, out, err) <- readProcessWithExitCode "node" ["test/client/inspection-client.mjs", "http://127.0.0.1:" <> show port <> prefix <> "/", T.unpack name] ""+                assertEqual (out <> err) ExitSuccess code+                WS.runClient "127.0.0.1" port (prefix <> "/ws") $ \connection -> do+                  WS.sendTextData connection ("{\"type\":\"ping\"}" :: T.Text)+                  pong <- WS.receiveData connection+                  (eitherDecode pong :: Either String Value) @?= Right (object ["type" .= ("pong" :: T.Text)])+          session pool (Session.script ("SELECT pgmq.drop_queue('" <> name <> "')"))+    ]
+ test/OutboxSpec.hs view
@@ -0,0 +1,53 @@+module OutboxSpec (tests) where++import Control.Concurrent.Async+import Control.Concurrent.STM+import Control.Monad (forM_)+import Data.Aeson (encode)+import Data.Map.Strict qualified as Map+import Data.Unique (newUnique)+import Pgmq.Inspect.Frames+import Pgmq.Inspect.Outbox+import Test.Tasty+import Test.Tasty.HUnit++tests :: TestTree+tests =+  testGroup+    "WebSocket outbox"+    [ testGroup+        "bounded outbox"+        [ testCase ("capacity " <> show capacity <> " drops oldest and signals before retained frames") $ do+            box <- newOutbox capacity+            writerBarrier <- newEmptyTMVarIO+            withAsync (atomically (takeTMVar writerBarrier) >> atomically (takeOutgoing box)) $ \writer -> do+              atomically $ forM_ ["first", "second", "third"] $ \code -> enqueue box (ErrorFrame code "test" Nothing)+              atomically (queuedFrames box) >>= (@?= capacity)+              atomically (putTMVar writerBarrier ())+              (lost, first) <- wait writer+              lost @?= True+              let expected = if capacity == 1 then "third" else "second"+              encode first @?= encode (ErrorFrame expected "test" Nothing)+              if capacity == 2+                then do+                  (again, retained) <- atomically (takeOutgoing box)+                  again @?= False+                  encode retained @?= encode (ErrorFrame "third" "test" Nothing)+                else pure ()+              atomically (queuedFrames box) >>= (@?= 0)+        | capacity <- [1, 2]+        ],+      testGroup+        "generation tags"+        [ testCase "unsubscribed generations are discarded by the writer" $ do+            box <- newOutbox 2+            old <- newUnique+            generations <- newTVarIO (Map.singleton "orders" old)+            atomically $ do+              enqueueTagged box (Just (generations, "orders", old)) Pong+              writeTVar generations Map.empty+              enqueue box Goodbye+            (_, frame) <- atomically (takeOutgoing box)+            encode frame @?= encode Goodbye+        ]+    ]
+ test/PrivilegesSpec.hs view
@@ -0,0 +1,62 @@+module PrivilegesSpec (tests) where++import Control.Exception (bracket)+import Control.Monad (forM_)+import Data.Aeson (Value (..), eitherDecode)+import Data.Text (Text)+import Data.Text.IO qualified as T+import EphemeralDb+import EphemeralPg qualified as Pg+import Hasql.Connection qualified as Connection+import Hasql.Connection.Settings qualified as Settings+import Hasql.Decoders qualified as D+import Hasql.Pool qualified as Pool+import Hasql.Pool.Config qualified as PC+import Hasql.Session qualified as Session+import Hasql.Statement (unpreparable)+import HttpSpec (session)+import Network.WebSockets qualified as WS+import Pgmq.Effectful (PgmqRuntimeError (..))+import Pgmq.Inspect.Config+import Pgmq.Inspect.Env+import Pgmq.Inspect.Error (sqlState)+import RouterSpec (lookupKey)+import Test.Tasty+import Test.Tasty.HUnit+import TestServer++tests :: TestTree+tests =+  testGroup+    "Inspection role privileges"+    [ testCase "all reads and live snapshot work; consumption/mutation denied; future queues inherit SELECT" $ do+        result <- withPgmqDb $ \owner db -> do+          session owner (Session.script "SELECT pgmq.create('role_current'); SELECT pgmq.send('role_current','{}'); SELECT pgmq.send('role_current','{}'); SELECT pgmq.archive('role_current',1); SELECT pgmq.bind_topic('role.*','role_current'); SELECT pgmq.enable_notify_insert('role_current',250)")+          sql <- T.readFile "test/fixtures/inspection-role.sql"+          session owner (Session.script sql)+          session owner (Session.script "SELECT pgmq.create('role_future'); SELECT pgmq.send('role_future','{}')")+          let settings = Pg.connectionSettings db <> Settings.user "pgmq_inspector"+              poolConfig = PC.settings [PC.size 2, PC.acquisitionTimeout 1, PC.staticConnectionSettings settings]+          bracket (Pool.acquire poolConfig) Pool.release $ \pool ->+            withTestServer defaultInspectConfig (poolEnv pool (Just settings)) $ \port -> do+              forM_ ["/", "/openapi.json", "/queues", "/queues/role_current", "/queues/role_current/metrics", "/queues/role_current/messages", "/queues/role_current/messages/2", "/queues/role_current/archive", "/queues/role_current/archive/1", "/queues/role_current/bindings", "/metrics", "/bindings", "/routing/test?routing_key=role.created", "/notify/throttles", "/health/live", "/health/ready", "/queues/role_future/messages", "/queues/role_future/metrics"] $ \path -> do+                r <- get port path+                assertEqual (path <> " failed: " <> show (replyBody r)) 200 (replyStatus r)+              WS.runClient "127.0.0.1" port "/ws" $ \conn -> do+                WS.sendTextData conn ("{\"type\":\"subscribe\",\"queue\":\"role_current\"}" :: Text)+                bytes <- WS.receiveData conn+                row <- either assertFailure pure (eitherDecode bytes :: Either String Value)+                lookupKey "type" row @?= Just (String "snapshot")+          bracket (Connection.acquire settings >>= either (assertFailure . show) pure) Connection.release $ \conn -> do+            effective <- Connection.use conn (Session.statement () (unpreparable "SELECT current_setting('default_transaction_read_only'), current_setting('statement_timeout'), current_setting('lock_timeout')" mempty (D.singleRow ((,,) <$> D.column (D.nonNullable D.text) <*> D.column (D.nonNullable D.text) <*> D.column (D.nonNullable D.text))))) >>= either (assertFailure . show) pure+            effective @?= ("on", "2s", "250ms")+            -- Turn off the default transaction safeguard to prove the ACL itself+            -- denies mutation, including sequence advancement.+            Connection.use conn (Session.script "SET default_transaction_read_only=off") >>= either (assertFailure . show) pure+            forM_ ["SELECT pgmq.read('role_current',0,1)", "SELECT pgmq.pop('role_current')", "SELECT pgmq.send('role_current','{}')", "SELECT pgmq.archive('role_current',2)", "SELECT pgmq.delete('role_current',2)", "SELECT pgmq.purge_queue('role_current')", "SELECT pgmq.drop_queue('role_current')", "SELECT pgmq.create('role_forbidden')", "UPDATE pgmq.q_role_current SET read_ct=read_ct+1", "SELECT nextval('pgmq.q_role_current_msg_id_seq')"] $ \sql' -> do+              answer <- Connection.use conn (Session.script sql')+              case answer of+                Left err -> assertEqual ("wrong failure for " <> show sql') (Just "42501") (sqlState (PgmqSessionError err))+                Right () -> assertFailure ("write unexpectedly permitted: " <> show sql')+        either (assertFailure . show) pure result+    ]
+ test/RawWebSocketSpec.hs view
@@ -0,0 +1,53 @@+-- | Send raw client frames so size limits are tested before a body is provided,+-- and across fragments that individually fit but collectively exceed the cap.+module RawWebSocketSpec (tests) where++import Control.Exception (bracket)+import Data.ByteString qualified as BS+import Data.ByteString.Char8 qualified as B8+import Network.Socket qualified as Socket+import Network.Socket.ByteString qualified as SocketBS+import Pgmq.Inspect.Config+import RouterSpec (mockEnv)+import System.Timeout (timeout)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit+import TestServer++withRaw :: Int -> (Socket.Socket -> IO a) -> IO a+withRaw port k = bracket (Socket.socket Socket.AF_INET Socket.Stream Socket.defaultProtocol) Socket.close $ \socket -> do+  Socket.connect socket (Socket.SockAddrInet (fromIntegral port) (Socket.tupleToHostAddress (127, 0, 0, 1)))+  SocketBS.sendAll socket "GET /ws HTTP/1.1\r\nHost: localhost\r\nConnection: Upgrade\r\nUpgrade: websocket\r\nSec-WebSocket-Version: 13\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\r\n"+  header <- timeout 2000000 (untilBytes socket "\r\n\r\n") >>= maybe (assertFailure "handshake timed out") pure+  assertBool "upgraded" ("101" `BS.isInfixOf` header)+  k socket++untilBytes :: Socket.Socket -> BS.ByteString -> IO BS.ByteString+untilBytes socket needle = go BS.empty+  where+    go accumulated+      | needle `BS.isInfixOf` accumulated = pure accumulated+      | otherwise = do+          chunk <- SocketBS.recv socket 4096+          if BS.null chunk then assertFailure ("socket closed before " <> B8.unpack needle) else go (accumulated <> chunk)++tests :: TestTree+tests =+  testGroup+    "WebSocket raw size limits"+    [ testCase "huge declared frame is rejected without sending its payload" $+        withTestRuntime config mockEnv $ \port _ -> withRaw port $ \socket -> do+          -- Masked text, uint64 length = 1 GiB, followed by the four-byte mask.+          SocketBS.sendAll socket (BS.pack [0x81, 0xff, 0, 0, 0, 0, 0x40, 0, 0, 0, 0, 0, 0, 0])+          result <- timeout 2000000 (untilBytes socket "frame_too_large")+          assertBool "header-only size rejection" (maybe False (const True) result),+      testCase "fragmented message enforces assembled cap before unbounded accumulation" $+        withTestRuntime config mockEnv $ \port _ -> withRaw port $ \socket -> do+          let payload = BS.replicate 40 120+          SocketBS.sendAll socket (BS.pack [0x01, 0xa8, 0, 0, 0, 0] <> payload)+          SocketBS.sendAll socket (BS.pack [0x80, 0xa8, 0, 0, 0, 0] <> payload)+          result <- timeout 2000000 (untilBytes socket "frame_too_large")+          assertBool "assembled size rejection" (maybe False (const True) result)+    ]+  where+    config = defaultInspectConfig {wsMaxFrameBytes = 64}
+ test/RefreshSpec.hs view
@@ -0,0 +1,258 @@+module RefreshSpec (tests) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async+import Control.Concurrent.MVar+import Control.Concurrent.STM+import Control.Exception (finally)+import Control.Monad (forM_, replicateM, replicateM_, void)+import Data.Aeson (encode)+import Data.Map.Strict qualified as Map+import Data.Text (Text)+import Data.Time (UTCTime (..), fromGregorian)+import Data.Word (Word64)+import Effectful (liftIO)+import Effectful.Dispatch.Dynamic (interpret)+import ErrorSpec (serverError)+import Pgmq.Effectful.Effect qualified as E+import Pgmq.Hasql.Statements.Types qualified as H+import Pgmq.Inspect.Budget+import Pgmq.Inspect.Config+import Pgmq.Inspect.Env+import Pgmq.Inspect.Frames+import Pgmq.Inspect.Outbox+import Pgmq.Inspect.Refresh+import Pgmq.Types (UnvalidatedQueue (..))+import System.Timeout (timeout)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit++cfg :: InspectConfig+cfg = defaultInspectConfig {pollIntervalUs = 200000, metricsMinRefreshUs = 100000, notifyDebounceUs = 20000}++epoch :: UTCTime+epoch = UTCTime (fromGregorian 2026 1 1) 0++metrics :: Text -> Int -> H.QueueMetrics+metrics name count = H.QueueMetrics name (fromIntegral count) Nothing Nothing (fromIntegral count) epoch (fromIntegral count) Nothing++data Probe = Probe+  {calls :: TVar (Map.Map Text Int), live :: TVar (Map.Map Text Int), peak :: TVar Int}++newProbe :: IO Probe+newProbe = Probe <$> newTVarIO Map.empty <*> newTVarIO Map.empty <*> newTVarIO 0++probeEnv :: Probe -> (Text -> Int -> IO H.QueueMetrics) -> InspectEnv+probeEnv probe readOne =+  InspectEnv+    ( runnerFromInterpreter+        ( interpret+            ( \_ -> \case+                E.ListQueuesUnvalidated -> pure [UnvalidatedQueue name epoch False False | name <- ["orders", "other"]]+                E.QueueMetricsUnvalidated name -> liftIO $ do+                  index <- atomically $ do+                    counts <- readTVar (calls probe)+                    let index = Map.findWithDefault 0 name counts + 1+                    writeTVar (calls probe) (Map.insert name index counts)+                    modifyTVar' (live probe) (Map.insertWith (+) name 1)+                    active <- sum . Map.elems <$> readTVar (live probe)+                    modifyTVar' (peak probe) (max active)+                    pure index+                  readOne name index `finally` atomically (modifyTVar' (live probe) (Map.adjust (subtract 1) name))+                _ -> error "RefreshSpec unexpected effect"+            )+        )+    )+    Nothing++await :: STM a -> IO a+await action = timeout 2000000 (atomically action) >>= maybe (assertFailure "coordinator barrier timed out") pure++next :: Subscriber -> IO ServerFrame+next subscriber = snd <$> await (takeOutgoing (subscriberOutbox subscriber))++callCount :: Probe -> Text -> STM Int+callCount probe name = Map.findWithDefault 0 name <$> readTVar (calls probe)++settled :: Probe -> MetricsCoordinator -> Text -> Int -> IO ()+settled probe coordinator name expected = await $ do+  n <- callCount probe name+  check (n == expected)+  coordinatorInFlight coordinator >>= check . (== 0)++withClock :: InspectConfig -> InspectEnv -> Maybe PushHooks -> (TVar Word64 -> MetricsCoordinator -> IO a) -> IO a+withClock config env hooks k = do+  clock <- newTVarIO 1000000+  withInspectionBudget config $ \budget -> withCoordinatorUsing (readTVarIO clock) config env budget hooks (k clock)++tests :: TestTree+tests =+  testGroup+    "WebSocket shared coordinator"+    [ testCase "100 subscriptions share one in-flight snapshot and one metrics read per poll" $ do+        probe <- newProbe+        started <- newEmptyMVar+        release <- newEmptyMVar+        let env =+              probeEnv+                probe+                ( \name n -> do+                    if n == 1 then putMVar started () >> takeMVar release else pure ()+                    pure (metrics name 0)+                )+        withClock cfg env Nothing $ \clock coordinator -> do+          subscribers <- replicateM 100 (newSubscriber 4)+          subscribeQueue coordinator (subscribers !! 0) "orders"+          takeMVar started+          forM_ (drop 1 subscribers) $ \subscriber -> subscribeQueue coordinator subscriber "orders"+          atomically (coordinatorSubscriberCount coordinator "orders") >>= (@?= 100)+          putMVar release ()+          forM_ subscribers $ \subscriber -> next subscriber >>= (\case Snapshot "orders" _ False -> pure (); frame -> assertFailure (show frame))+          settled probe coordinator "orders" 1+          -- Re-subscription reuses shared current metrics and cannot bypass the floor.+          replicateM_ 100 (subscribeQueue coordinator (subscribers !! 0) "orders")+          atomically (callCount probe "orders") >>= (@?= 1)+          atomically (writeTVar clock 1200000)+          settled probe coordinator "orders" 2+          forM_ (drop 1 subscribers) $ \subscriber -> atomically (queuedFrames (subscriberOutbox subscriber)) >>= (@?= 0),+      testCase "hints coalesce globally and obey the minimum interval even when a poll is due" $ do+        probe <- newProbe+        hint <- newEmptyMVar+        let hooks = PushHooks (\_ callback -> putMVar hint callback >> pure (pure ()))+        withClock cfg (probeEnv probe (\name _ -> pure (metrics name 0))) (Just hooks) $ \clock coordinator -> do+          a <- newSubscriber 4+          b <- newSubscriber 4+          subscribeQueue coordinator a "orders"+          subscribeQueue coordinator b "orders"+          next a >>= (\case Snapshot _ _ True -> pure (); frame -> assertFailure (show frame))+          void (next b)+          notify <- takeMVar hint+          atomically (replicateM_ 100 notify >> writeTVar clock 1050000)+          threadDelay 30000+          atomically (callCount probe "orders") >>= (@?= 1)+          atomically (writeTVar clock 1200000)+          forM_ [a, b] $ \subscriber -> next subscriber >>= (\case Update _ _ NotifySource -> pure (); frame -> assertFailure (show frame))+          settled probe coordinator "orders" 2,+      testCase "a slow queue cannot overlap itself or starve another queue" $ do+        probe <- newProbe+        started <- newEmptyMVar+        release <- newEmptyMVar+        let env =+              probeEnv+                probe+                ( \name n -> do+                    if name == "orders" && n == 1 then putMVar started () >> takeMVar release else pure ()+                    pure (metrics name n)+                )+        withClock cfg env Nothing $ \clock coordinator -> do+          a <- newSubscriber 8+          b <- newSubscriber 8+          subscribeQueue coordinator a "orders"+          takeMVar started+          subscribeQueue coordinator b "other"+          void (next b)+          atomically (writeTVar clock 1300000)+          void (next b)+          atomically (callCount probe "orders") >>= (@?= 1)+          atomically (readTVar (peak probe)) >>= (\n -> assertBool "global bound" (n <= maxConcurrentReads cfg))+          active <- atomically (Map.findWithDefault 0 "orders" <$> readTVar (live probe))+          active @?= 1+          putMVar release (),+      testCase "unsubscribe/re-subscribe discards stale completion and preserves refresh floor" $ do+        probe <- newProbe+        started <- newEmptyMVar+        release <- newEmptyMVar+        let env =+              probeEnv+                probe+                ( \name n -> do+                    if n == 1 then putMVar started () >> takeMVar release else pure ()+                    pure (metrics name n)+                )+        withClock cfg env Nothing $ \clock coordinator -> do+          subscriber <- newSubscriber 4+          subscribeQueue coordinator subscriber "orders"+          takeMVar started+          unsubscribeQueue coordinator subscriber "orders"+          subscribeQueue coordinator subscriber "orders"+          putMVar release ()+          settled probe coordinator "orders" 1+          atomically (queuedFrames (subscriberOutbox subscriber)) >>= (@?= 0)+          atomically (writeTVar clock 1200000)+          frame <- next subscriber+          case frame of+            Snapshot "orders" actual False -> encode actual @?= encode (metrics "orders" 2)+            _ -> assertFailure (show frame),+      testCase "HTTP-held admission reports busy and keeps a single bounded retry" $ do+        probe <- newProbe+        clock <- newTVarIO 1000000+        started <- newEmptyMVar+        release <- newEmptyMVar+        let config = cfg {maxConcurrentReads = 1}+        withInspectionBudget config $ \budget ->+          withAsync (withInspectionPermit budget (putMVar started () >> takeMVar release)) $ \held -> do+            takeMVar started+            withCoordinatorUsing (readTVarIO clock) config (probeEnv probe (\name n -> pure (metrics name n))) budget Nothing $ \coordinator -> do+              subscriber <- newSubscriber 4+              subscribeQueue coordinator subscriber "orders"+              next subscriber >>= (\case ErrorFrame "inspection_busy" _ _ -> pure (); frame -> assertFailure (show frame))+              atomically (callCount probe "orders") >>= (@?= 0)+              putMVar release ()+              void (wait held)+              atomically (writeTVar clock 1200000)+              next subscriber >>= (\case Snapshot _ _ _ -> pure (); frame -> assertFailure (show frame))+              settled probe coordinator "orders" 1,+      testCase "statement deadline is in-band and a later poll recovers" $ do+        probe <- newProbe+        attempts <- newTVarIO (0 :: Int)+        let base = probeEnv probe (\name n -> pure (metrics name n))+            env =+              base+                { runInspection = \action -> do+                    n <- atomically $ do+                      current <- readTVar attempts+                      writeTVar attempts (current + 1)+                      pure current+                    if n == 0 then pure (Left (serverError "57014")) else runInspection base action+                }+        withClock cfg env Nothing $ \clock coordinator -> do+          subscriber <- newSubscriber 4+          subscribeQueue coordinator subscriber "orders"+          next subscriber >>= (\case ErrorFrame "inspection_timeout" _ _ -> pure (); frame -> assertFailure (show frame))+          atomically (writeTVar clock 1200000)+          next subscriber >>= (\case Snapshot _ _ False -> pure (); frame -> assertFailure (show frame)),+      testCase "snapshot publication precedes an update even when a hint races initial metrics" $ do+        probe <- newProbe+        beforePublish <- newEmptyMVar+        release <- newEmptyMVar+        let hooks = PushHooks (\_ hint -> putMVar beforePublish hint >> takeMVar release >> pure (pure ()))+        withClock cfg (probeEnv probe (\name n -> pure (metrics name n))) (Just hooks) $ \clock coordinator -> do+          subscriber <- newSubscriber 4+          subscribeQueue coordinator subscriber "orders"+          hint <- takeMVar beforePublish+          atomically (hint >> writeTVar clock 1300000)+          atomically (queuedFrames (subscriberOutbox subscriber)) >>= (@?= 0)+          putMVar release ()+          next subscriber >>= (\case Snapshot _ actual True -> encode actual @?= encode (metrics "orders" 1); frame -> assertFailure (show frame))+          next subscriber >>= (\case Update _ actual _ -> encode actual @?= encode (metrics "orders" 2); frame -> assertFailure (show frame)),+      testCase "partitioned metadata is poll-only even when a push hook exists" $ do+        registered <- newTVarIO False+        let env =+              InspectEnv+                ( runnerFromInterpreter+                    ( interpret+                        ( \_ -> \case+                            E.ListQueuesUnvalidated -> pure [UnvalidatedQueue "orders" epoch True False]+                            E.QueueMetricsUnvalidated name -> pure (metrics name 0)+                            _ -> error "unexpected partition test effect"+                        )+                    )+                )+                Nothing+            hooks = PushHooks (\_ _ -> atomically (writeTVar registered True) >> pure (pure ()))+        withClock cfg env (Just hooks) $ \_ coordinator -> do+          subscriber <- newSubscriber 4+          subscribeQueue coordinator subscriber "orders"+          next subscriber >>= (\case Snapshot _ _ False -> pure (); frame -> assertFailure (show frame))+          readTVarIO registered >>= (@?= False)+    ]
+ test/RootPermissionsSpec.hs view
@@ -0,0 +1,63 @@+-- | Check each suite's local root helper using private disposable fixtures.+module RootPermissionsSpec (tests) where++import Control.Exception (IOException, bracket, try)+import Data.Bits ((.&.))+import System.Directory (getTemporaryDirectory, removePathForcibly)+import System.Posix.Directory qualified as Posix+import System.Posix.Files (createSymbolicLink, fileMode, getSymbolicLinkStatus, isDirectory, setFileMode)+import System.Posix.Temp (mkdtemp)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertBool, assertFailure, testCase, (@?=))++tests :: (FilePath -> IO ()) -> TestTree+tests ensure =+  testGroup+    "RootPermissionsSpec"+    [ testCase "missing root is a real 0700 directory" $ private $ \parent -> do+        let root = parent <> "/root"+        ensure root+        status <- getSymbolicLinkStatus root+        assertBool "real directory" (isDirectory status)+        fileMode status .&. 0o7777 @?= 0o700,+      testCase "owned 0755 directory is tightened without losing contents" $ private $ \parent -> do+        let root = parent <> "/root"+        Posix.createDirectory root 0o755+        setFileMode root 0o755+        writeFile (root <> "/sentinel") "preserve"+        ensure root+        mode root >>= (@?= 0o700)+        readFile (root <> "/sentinel") >>= (@?= "preserve"),+      testCase "repeated call preserves contents and permissions" $ private $ \parent -> do+        let root = parent <> "/root"+        ensure root+        writeFile (root <> "/sentinel") "preserve"+        ensure root+        mode root >>= (@?= 0o700)+        readFile (root <> "/sentinel") >>= (@?= "preserve"),+      testCase "regular file is rejected unchanged" $ private $ \parent -> do+        let root = parent <> "/root"+        writeFile root "preserve"+        setFileMode root 0o644+        rejected (ensure root)+        mode root >>= (@?= 0o644)+        readFile root >>= (@?= "preserve"),+      testCase "symlink is rejected without chmod of its target" $ private $ \parent -> do+        let root = parent <> "/root"+            target = parent <> "/target"+        Posix.createDirectory target 0o755+        setFileMode target 0o755+        writeFile (target <> "/sentinel") "preserve"+        createSymbolicLink target root+        rejected (ensure root)+        mode target >>= (@?= 0o755)+        readFile (target <> "/sentinel") >>= (@?= "preserve")+    ]+  where+    mode path = (.&. 0o7777) . fileMode <$> getSymbolicLinkStatus path+    rejected action = do+      result <- try action :: IO (Either IOException ())+      case result of+        Left _ -> pure ()+        Right () -> assertFailure "unsafe root accepted"+    private action = bracket (getTemporaryDirectory >>= \tmp -> mkdtemp (tmp <> "/pgmq-root-test-")) removePathForcibly action
+ test/RouterSpec.hs view
@@ -0,0 +1,112 @@+module RouterSpec (tests, mockEnv, lookupKey, errorCodeOf) where++import Control.Monad (forM_)+import Data.Aeson (Value)+import Data.Aeson qualified as Aeson+import Data.Aeson.KeyMap qualified as KM+import Data.Time (UTCTime (..), fromGregorian)+import Data.Vector qualified as V+import Effectful (runEff)+import Effectful.Dispatch.Dynamic (interpret)+import Effectful.Error.Static (runError)+import Pgmq.Effectful (PgmqRuntimeError)+import Pgmq.Effectful.Effect (Pgmq (..))+import Pgmq.Hasql.Statements.Types qualified as H+import Pgmq.Inspect.Config (defaultInspectConfig)+import Pgmq.Inspect.Env (InspectEnv (..))+import Pgmq.Types (UnvalidatedQueue (..))+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (testCase, (@?=))+import TestServer++-- | An environment whose runner never opens a connection: the handler answers+-- the four operations the Milestone 1 routes need and errors on any other.+mockEnv :: InspectEnv+mockEnv = InspectEnv {runInspection = runMock, listenerSettings = Nothing}+  where+    runMock action =+      either (Left . snd) Right+        <$> runEff+          ( runError @PgmqRuntimeError+              ( interpret+                  ( \_ -> \case+                      ListQueuesUnvalidated -> pure [cannedQueue]+                      QueueMetricsUnvalidated name -> pure (cannedMetrics name)+                      AllQueueMetrics -> pure [cannedMetrics "orders"]+                      ListTopicBindings -> pure []+                      PeekMessages _ -> pure V.empty+                      PeekArchivedMessages _ -> pure V.empty+                      LookupMessage _ -> pure Nothing+                      LookupArchivedMessage _ -> pure Nothing+                      ListNotifyInsertThrottles -> pure []+                      TestRouting _ -> pure []+                      _ -> error "RouterSpec: operation not canned"+                  )+                  action+              )+          )+    epoch = UTCTime (fromGregorian 2026 1 1) 0+    cannedQueue = UnvalidatedQueue "orders" epoch False False+    cannedMetrics name = H.QueueMetrics name 3 (Just 1) (Just 9) 12 epoch 2 Nothing++tests :: TestTree+tests =+  testGroup+    "Router (mock interpreter, no database)"+    [ testCase "integer overflow and malformed input are rejected before database work" $+        withTestServer defaultInspectConfig mockEnv $ \port -> forM_ ["-1", "9223372036854775808", "18446744073709551617", replicate 128 '9', replicate 129 '1', "1x", "", "%2B1"] $ \raw -> do+          reply <- get port ("/queues/orders/messages?from=" <> raw)+          replyStatus reply @?= 400+          errorCodeOf reply >>= (@?= Aeson.String "invalid_cursor")+          one <- get port ("/queues/orders/messages/" <> if null raw then "abc" else raw)+          replyStatus one @?= 400,+      testCase "maximum int64 and large capped limit remain exact" $+        withTestServer defaultInspectConfig mockEnv $ \port -> do+          reply <- get port "/queues/orders/messages?from=9223372036854775807&limit=999999999999999999999999999999999999999"+          replyStatus reply @?= 200,+      testCase "GET / describes the service" $+        withTestServer defaultInspectConfig mockEnv $ \port -> do+          reply <- get port "/"+          replyStatus reply @?= 200+          body <- decodeReply reply+          lookupKey "service" body @?= Just (Aeson.String "pgmq-inspect")+          lookupKey "websocket_path" body @?= Just (Aeson.String "ws"),+      testCase "GET /queues lists the canned queue with pgmq column names" $+        withTestServer defaultInspectConfig mockEnv $ \port -> do+          reply <- get port "/queues"+          replyStatus reply @?= 200+          rows <- decodeReply reply+          map (lookupKey "queue_name") rows @?= [Just (Aeson.String "orders")],+      testCase "GET /queues/orders/metrics distinguishes total and visible depth" $+        withTestServer defaultInspectConfig mockEnv $ \port -> do+          body <- decodeReply =<< get port "/queues/orders/metrics"+          lookupKey "queue_length" body @?= Just (Aeson.Number 3)+          lookupKey "queue_visible_length" body @?= Just (Aeson.Number 2)+          lookupKey "default_partition_length" body @?= Just Aeson.Null,+      testCase "an unknown path is 404 not_found" $+        withTestServer defaultInspectConfig mockEnv $ \port -> do+          reply <- get port "/nope"+          replyStatus reply @?= 404+          code <- errorCodeOf reply+          code @?= "not_found",+      testCase "POST is 405 method_not_allowed" $+        withTestServer defaultInspectConfig mockEnv $ \port -> do+          reply <- getWith port "POST" [] "/queues"+          replyStatus reply @?= 405,+      testCase "limit=abc is 400 invalid_limit before any database work" $+        withTestServer defaultInspectConfig mockEnv $ \port -> do+          reply <- get port "/queues/orders/messages?limit=abc"+          replyStatus reply @?= 400+          code <- errorCodeOf reply+          code @?= "invalid_limit"+    ]++lookupKey :: Aeson.Key -> Value -> Maybe Value+lookupKey k = \case+  Aeson.Object o -> KM.lookup k o+  _ -> Nothing++errorCodeOf :: Reply -> IO Value+errorCodeOf reply = do+  body <- decodeReply reply+  maybe (fail "no error.code") pure (lookupKey "error" body >>= lookupKey "code")
+ test/ServerSpec.hs view
@@ -0,0 +1,105 @@+module ServerSpec (tests) where++import Control.Concurrent.Async+import Control.Concurrent.MVar+import Control.Exception+import Data.Aeson (Value (..))+import Data.IORef+import Data.List (isInfixOf)+import Network.HTTP.Client qualified as Http+import Network.HTTP.Types (statusCode)+import Network.Socket qualified as Socket+import Network.Wai (pathInfo)+import Network.Wai.Handler.Warp qualified as Warp+import Pgmq.Inspect+import Pgmq.Inspect.Lifecycle qualified as Lifecycle+import RouterSpec (errorCodeOf, mockEnv)+import System.IO.Error (isUserError)+import Test.Tasty+import Test.Tasty.HUnit+import TestServer++tests :: TestTree+tests =+  testGroup+    "Server lifecycle and mounting"+    [ testCase "a plain GET /ws is 426" $ withTestServer defaultInspectConfig mockEnv $ \port -> do+        r <- get port "/ws"+        replyStatus r @?= 426+        errorCodeOf r >>= (@?= String "websocket_upgrade_required"),+      testCase "upgrade seam returns JSON 503" $+        withInspectApplicationWith defaultInspectConfig mockEnv rejectingWebSocketApp $ \app ->+          Warp.testWithApplication (pure app) $ \port -> do+            r <- getWith port "GET" upgradeHeaders "/ws"+            replyStatus r @?= 503+            errorCodeOf r >>= (@?= String "websocket_unavailable"),+      testCase "HTTP, encoded queue names and upgrades use stripped pathInfo under a prefix" $+        withInspectApplicationWith defaultInspectConfig mockEnv rejectingWebSocketApp $ \app ->+          Warp.testWithApplication (pure (\req respond -> app req {pathInfo = drop 1 (pathInfo req)} respond)) $ \port -> do+            r <- get port "/pgmq/queues/%6frders"+            replyStatus r @?= 200+            upgrade <- getWith port "GET" upgradeHeaders "/pgmq/ws"+            replyStatus upgrade @?= 503,+      testCase "occupied port fails before continuation" $+        withInspectServer config mockEnv $ \server -> do+          ran <- newIORef False+          result <- try @IOException (withInspectServer config {bindPort = serverPort server} mockEnv (\_ -> writeIORef ran True))+          assertBool "bind failure propagated" (either (const True) (const False) result)+          readIORef ran >>= (@?= False),+      testCase "invalid host fails before continuation" $ do+        result <- try @IOException (withInspectServer config {bindHost = "invalid host !"} mockEnv (const (assertFailure "continued after bad host")))+        assertBool "host failure" (either (const True) (const False) result),+      testCase "invalid port and config are rejected" $+        mapM_+          ( \cfg -> do+              result <- try @IOException (withInspectServer cfg mockEnv (const (assertFailure "continued after invalid configuration")))+              assertBool "invalid configuration failed" (either (const True) (const False) result)+          )+          [config {bindPort = -1}, config {bindPort = 65536}, config {inspect = defaultInspectConfig {maxPageLimit = maxBound}}],+      testCase "normal and exceptional exits join thread and close the actual owned socket" $ do+        saved <- newEmptyMVar+        observedServer (\socket server -> putMVar saved (socket, server))+        (socket, server) <- takeMVar saved+        closedSocket socket+        poll (serverThread server) >>= (\state -> assertBool "server joined" (maybe False (const True) state))+        result <- try @IOException $ observedServer (\owned _ -> putMVar saved (owned, server) >> throwIO (userError "intentional callback failure"))+        case result of+          Left err -> assertBool "callback error propagated" (isUserError err && "intentional callback failure" `isInfixOf` show err)+          Right () -> assertFailure "callback error was lost"+        (failedSocket, _) <- takeMVar saved+        closedSocket failedSocket,+      testCase "cancellation immediately after bind closes the actual startup socket" $ do+        bound <- newEmptyMVar+        withAsync (Lifecycle.withBoundSocket config (\socket -> putMVar bound socket >> threadBlock)) $ \owner -> do+          socket <- takeMVar bound+          cancel owner+          closedSocket socket,+      testCase "cancellation in a running callback joins thread and closes owned socket" $ do+        started <- newEmptyMVar+        withAsync (observedServer (\socket server -> putMVar started (socket, server) >> threadBlock)) $ \owner -> do+          (socket, server) <- takeMVar started+          cancel owner+          closedSocket socket+          poll (serverThread server) >>= (\state -> assertBool "server joined" (maybe False (const True) state)),+      testCase "port 0 honors configured loopback host" $+        withInspectServer config {bindHost = "::1"} mockEnv $ \server -> do+          assertBool "OS assigned port" (serverPort server > 0)+          manager <- Http.newManager Http.defaultManagerSettings+          req <- Http.parseRequest ("http://[::1]:" <> show (serverPort server) <> "/health/live")+          response <- Http.httpLbs req manager+          statusCode (Http.responseStatus response) @?= 200+    ]+  where+    -- These are the same private resource combinators used by withInspectServer.+    observedServer action = Lifecycle.withBoundSocket config $ \socket ->+      withInspectRuntime defaultInspectConfig mockEnv $ \runtime ->+        Lifecycle.withRunningSocket+          config+          socket+          (runtimeApplication runtime)+          (runtimeRequestShutdown runtime >> runtimeAwaitDrained runtime 1000000)+          (\port worker -> action socket (RunningInspectServer port worker))+    closedSocket socket = Socket.withFdSocket socket (\descriptor -> descriptor @?= -1)+    config = defaultInspectServerConfig {bindPort = 0}+    threadBlock = newEmptyMVar >>= takeMVar :: IO ()+    upgradeHeaders = [("Connection", "Upgrade"), ("Upgrade", "websocket"), ("Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ=="), ("Sec-WebSocket-Version", "13")]
+ test/SmokeSpec.hs view
@@ -0,0 +1,62 @@+-- | Reproducible guide transcript, using only a disposable database. Set+-- PGMQ_INSPECT_RECORD_TRANSCRIPT to retain the successful curl/WS conversation.+module SmokeSpec (tests) where++import Control.Monad (forM)+import Data.Aeson (Value (..), eitherDecode)+import Data.ByteString.Lazy.Char8 qualified as LBS+import Data.Text (Text)+import Data.Text qualified as T+import EphemeralDb+import EphemeralPg qualified as Pg+import Hasql.Session qualified as Session+import HttpSpec (session)+import Network.WebSockets qualified as WS+import RouterSpec (lookupKey)+import System.Environment (lookupEnv)+import System.Exit (ExitCode (..))+import System.IO (hGetLine)+import System.Process+import System.Timeout (timeout)+import Test.Tasty+import Test.Tasty.HUnit++tests :: TestTree+tests = testCase "standalone guide curl/WS transcript and Ctrl-C cleanup" $ do+  result <- withPgmqDb $ \owner db -> do+    session owner (Session.script "SELECT pgmq.create('demo'); SELECT pgmq.send('demo','{\"hello\":\"world\"}'); SELECT pgmq.send('demo','{\"hello\":\"world\"}'); SELECT pgmq.send('demo','{\"hello\":\"world\"}'); SELECT pgmq.archive('demo',1); SELECT pgmq.bind_topic('demo.*','demo'); SELECT pgmq.enable_notify_insert('demo',250)")+    let p =+          (proc "pgmq-inspect" ["--database-url", T.unpack (Pg.connectionString db), "--port", "0", "--no-listener", "--cors-origin", "http://localhost:5173"])+            { std_out = CreatePipe,+              create_group = True+            }+    withCreateProcess p $ \_ output _ child -> do+      startup <- maybe (assertFailure "no stdout pipe") (\h -> timeout 10000000 (hGetLine h) >>= maybe (assertFailure "startup timeout") pure) output+      let port = read (reverse (takeWhile (/= ':') (reverse startup)))+          base = "http://127.0.0.1:" <> show (port :: Int)+      curls <- forM ["/", "/queues", "/queues/demo", "/queues/demo/metrics", "/metrics", "/queues/demo/messages?limit=1", "/queues/demo/messages?from=2&limit=1", "/queues/demo/messages/2", "/queues/demo/archive", "/queues/demo/archive/1", "/queues/demo/bindings", "/bindings", "/routing/test?routing_key=demo.created", "/notify/throttles", "/health/live", "/health/ready", "/openapi.json", "/queues/demo/messages?from=bad"] $ \path -> do+        (code, body, err) <- readProcessWithExitCode "curl" ["-sS", base <> path] ""+        assertEqual err ExitSuccess code+        case eitherDecode (LBS.pack body) :: Either String Value of+          Left e -> assertFailure e+          Right _ -> pure ()+        pure ("$ curl -sS \"$base" <> path <> "\"\n" <> body <> "\n")+      (corsCode, cors, corsErr) <- readProcessWithExitCode "curl" ["-sS", "-i", "-H", "Origin: http://localhost:5173", base <> "/health/live"] ""+      assertEqual corsErr ExitSuccess corsCode+      frames <- WS.runClient "127.0.0.1" port "/ws" $ \connection -> do+        WS.sendTextData connection ("{\"type\":\"subscribe\",\"queue\":\"demo\"}" :: Text)+        snapshot <- WS.receiveData connection+        row <- either assertFailure pure (eitherDecode snapshot :: Either String Value)+        lookupKey "type" row @?= Just (String "snapshot")+        lookupKey "push" row @?= Just (Bool False)+        WS.sendTextData connection ("{\"type\":\"ping\"}" :: Text)+        pong <- WS.receiveData connection+        interruptProcessGroupOf child+        goodbye <- timeout 3000000 (WS.receiveData connection) >>= maybe (assertFailure "no Ctrl-C goodbye") pure+        final <- either assertFailure pure (eitherDecode goodbye :: Either String Value)+        lookupKey "type" final @?= Just (String "goodbye")+        pure (unlines ["> {\"type\":\"subscribe\",\"queue\":\"demo\"}", LBS.unpack snapshot, "> {\"type\":\"ping\"}", LBS.unpack pong, "> Ctrl-C to server", LBS.unpack goodbye])+      exited <- timeout 5000000 (waitForProcess child)+      assertBool "process exited" (maybe False (const True) exited)+      lookupEnv "PGMQ_INSPECT_RECORD_TRANSCRIPT" >>= mapM_ (\path -> writeFile path (startup <> "\n" <> concat curls <> "$ curl -sS -i -H 'Origin: http://localhost:5173' \"$base/health/live\"\n" <> filter (/= '\r') cors <> "\nWebSocket /ws:\n" <> frames))+  either (assertFailure . show) pure result
+ test/TestServer.hs view
@@ -0,0 +1,60 @@+-- | Start the surface on an OS-assigned loopback port and talk to it with+-- http-client.+module TestServer+  ( withTestServer,+    withTestRuntime,+    Reply (..),+    get,+    getWith,+    decodeReply,+  )+where++import Control.Exception (finally)+import Data.Aeson (FromJSON, eitherDecode)+import Data.ByteString (ByteString)+import Data.ByteString.Lazy qualified as LBS+import Data.CaseInsensitive (CI)+import Network.HTTP.Client qualified as Http+import Network.HTTP.Types (Method, methodGet)+import Network.HTTP.Types.Header (HeaderName)+import Network.Wai.Handler.Warp qualified as Warp+import Pgmq.Inspect.Config (InspectConfig, InspectServerConfig (..))+import Pgmq.Inspect.Env (InspectEnv)+import Pgmq.Inspect.Server (InspectRuntime (..), RunningInspectServer (..), withInspectRuntime, withInspectServer)+import Test.Tasty.HUnit (assertFailure)++withTestServer :: InspectConfig -> InspectEnv -> (Int -> IO a) -> IO a+withTestServer cfg env k =+  withInspectServer InspectServerConfig {bindHost = "127.0.0.1", bindPort = 0, inspect = cfg} env $ \server ->+    k (serverPort server)++data Reply = Reply+  { replyStatus :: !Int,+    replyHeaders :: ![(CI ByteString, ByteString)],+    replyBody :: !LBS.ByteString+  }++get :: Int -> String -> IO Reply+get port = getWith port methodGet []++getWith :: Int -> Method -> [(HeaderName, ByteString)] -> String -> IO Reply+getWith port method headers path = do+  manager <- Http.newManager Http.defaultManagerSettings+  request <- Http.parseRequest ("http://127.0.0.1:" <> show port <> path)+  response <- Http.httpLbs request {Http.method = method, Http.requestHeaders = headers} manager+  pure+    Reply+      { replyStatus = fromEnum (Http.responseStatus response),+        replyHeaders = Http.responseHeaders response,+        replyBody = Http.responseBody response+      }++decodeReply :: (FromJSON a) => Reply -> IO a+decodeReply reply = either (assertFailure . ("undecodable body: " <>)) pure (eitherDecode (replyBody reply))++-- | Feed shutdown happens while the host-owned Warp server is still alive.+withTestRuntime :: InspectConfig -> InspectEnv -> (Int -> InspectRuntime -> IO a) -> IO a+withTestRuntime cfg env k = withInspectRuntime cfg env $ \runtime ->+  Warp.testWithApplication (pure (runtimeApplication runtime)) $ \port ->+    k port runtime `finally` (runtimeRequestShutdown runtime >> runtimeAwaitDrained runtime 1000000)
+ test/WebSocketSpec.hs view
@@ -0,0 +1,200 @@+module WebSocketSpec (tests, databaseTests, withWs, sendClient, receive, awaitFrame, frameType, frameField) where++import Control.Concurrent.Async+import Control.Concurrent.MVar+import Control.Exception+import Control.Monad (forM_, void)+import Data.Aeson+import Data.Aeson.Key qualified as Key+import Data.ByteString.Lazy.Char8 qualified as LBS+import Data.List (isInfixOf)+import Data.Text (Text)+import Data.Text qualified as T+import Data.Time (UTCTime (..), fromGregorian)+import EphemeralPg qualified as Pg+import ErrorSpec (failingEnv, serverError)+import Hasql.Pool qualified as Pool+import HttpSpec (sendN, withQueue)+import Network.Wai.Handler.Warp qualified as Warp+import Network.WebSockets qualified as WS+import Pgmq.Hasql.Statements.Types qualified as H+import Pgmq.Inspect+import Pgmq.Inspect.Outbox (overflowFrame)+import Pgmq.Types (queueNameToText)+import RouterSpec (lookupKey, mockEnv)+import System.Timeout (timeout)+import Test.Tasty+import Test.Tasty.Golden (NoCreateFile (..), goldenVsString)+import Test.Tasty.HUnit+import TestServer++withWs :: Int -> (WS.Connection -> IO a) -> IO a+withWs port = WS.runClient "127.0.0.1" port "/ws"++sendClient :: WS.Connection -> ClientFrame -> IO ()+sendClient connection = WS.sendTextData connection . encode++receive :: WS.Connection -> IO Value+receive connection = do+  bytes <- timeout 2000000 (WS.receiveData connection :: IO LBS.ByteString) >>= maybe (assertFailure "WebSocket receive timed out") pure+  either (assertFailure . ("invalid server JSON: " <>)) pure (eitherDecode bytes)++frameField :: Text -> Value -> Maybe Value+frameField = lookupKey . Key.fromText++frameType :: Value -> Maybe Text+frameType value = case frameField "type" value of Just (String tag) -> Just tag; _ -> Nothing++awaitFrame :: Int -> WS.Connection -> (Value -> Bool) -> IO (Maybe Value)+awaitFrame budget connection predicate = timeout budget loop+  where+    loop = do+      bytes <- WS.receiveData connection :: IO LBS.ByteString+      value <- either (assertFailure . show) pure (eitherDecode bytes)+      if predicate value then pure value else loop++config :: InspectConfig+config = defaultInspectConfig {pollIntervalUs = 200000, metricsMinRefreshUs = 100000, notifyDebounceUs = 20000}++tests :: TestTree+tests =+  testGroup+    "WebSocket"+    [ goldenTests,+      testCase "ping and malformed JSON are answered through the writer" $+        withTestRuntime config mockEnv $ \port _ -> withWs port $ \connection -> do+          sendClient connection Ping+          receive connection >>= (\frame -> frameType frame @?= Just "pong")+          WS.sendTextData connection ("not json" :: Text)+          receive connection >>= (\frame -> frameField "code" frame @?= Just (String "invalid_frame")),+      testCase "invalid queue frames do not echo the client payload" $+        withTestRuntime config mockEnv $ \port _ -> withWs port $ \connection -> do+          sendClient connection (Subscribe (T.replicate 100 "SECRET"))+          frame <- receive connection+          frameField "code" frame @?= Just (String "invalid_frame")+          frameField "queue" frame @?= Just Null+          assertBool "no raw client payload" (not ("SECRET" `isInfixOf` LBS.unpack (encode frame))),+      testCase "unknown queue returns a queue-scoped error and the connection survives" $+        withTestRuntime config mockEnv $ \port _ -> withWs port $ \connection -> do+          sendClient connection (Subscribe "missing")+          frame <- receive connection+          frameField "code" frame @?= Just (String "queue_not_found")+          frameField "queue" frame @?= Just (String "missing")+          sendClient connection Ping+          receive connection >>= (\answer -> frameType answer @?= Just "pong"),+      testCase "per-connection subscription cap is enforced before another read" $+        withTestRuntime config {wsMaxSubscriptions = 1} mockEnv $ \port _ -> withWs port $ \connection -> do+          sendClient connection (Subscribe "orders")+          receive connection >>= (\frame -> frameType frame @?= Just "snapshot")+          sendClient connection (Subscribe "other")+          receive connection >>= (\frame -> frameField "code" frame @?= Just (String "too_many_subscriptions")),+      testCase "distinct queue cap is shared across connections" $+        withTestRuntime config {wsMaxQueues = 1} mockEnv $ \port _ -> withWs port $ \first -> withWs port $ \second -> do+          sendClient first (Subscribe "orders")+          void (receive first)+          sendClient second (Subscribe "other")+          receive second >>= (\frame -> frameField "code" frame @?= Just (String "too_many_queues")),+      testCase "connection cap rejects the extra upgrade" $+        withTestRuntime config {wsMaxConnections = 1} mockEnv $ \port _ -> withWs port $ \first -> do+          sendClient first Ping+          void (receive first)+          outcome <- try @WS.HandshakeException (withWs port (const (pure ())))+          case outcome of+            Left _ -> pure ()+            Right () -> assertFailure "extra connection was accepted",+      testCase "oversized frame is bounded before receiving its body, even with capacity one" $+        withTestRuntime config {wsMaxFrameBytes = 64, wsQueueCapacity = 1} mockEnv $ \port _ -> withWs port $ \connection -> do+          WS.sendTextData connection (T.replicate 1024 "x")+          frame <- receive connection+          frameField "code" frame @?= Just (String "frame_too_large")+          receive connection >>= (\bye -> frameType bye @?= Just "goodbye"),+      testCase "database diagnostics do not reach feed errors" $+        withTestRuntime config (failingEnv (serverError "23505")) $ \port _ -> withWs port $ \connection -> do+          sendClient connection (Subscribe "orders")+          frame <- receive connection+          frameField "code" frame @?= Just (String "database_error")+          assertBool "no SQL/parameters/server diagnostics" (not ("SECRET" `isInfixOf` LBS.unpack (encode frame))),+      testCase "goodbye precedes the server close" $+        withTestRuntime config mockEnv $ \port runtime -> withWs port $ \connection -> do+          sendClient connection (Subscribe "orders")+          void (receive connection)+          runtimeRequestShutdown runtime+          receive connection >>= (\frame -> frameType frame @?= Just "goodbye")+          closed <- try @WS.ConnectionException (WS.receiveData connection :: IO LBS.ByteString)+          case closed of+            Left (WS.CloseRequest _ _) -> pure ()+            other -> assertFailure (show other),+      testCase "cancelling the convenience runner drains goodbye before Warp exits" $ do+        started <- newEmptyMVar+        stop <- newEmptyMVar+        withAsync (withInspectServer defaultInspectServerConfig {bindPort = 0, inspect = config} mockEnv (\server -> putMVar started (serverPort server) >> takeMVar stop)) $ \owner -> do+          port <- takeMVar started+          withWs port $ \connection -> do+            sendClient connection Ping+            void (receive connection)+            cancel owner+            receive connection >>= (\frame -> frameType frame @?= Just "goodbye"),+      testCase "host-owned Warp may outlive the runtime without retaining feed tasks" $ do+        readyRuntime <- newEmptyMVar+        stop <- newEmptyMVar+        withAsync (withInspectRuntime config mockEnv (\runtime -> putMVar readyRuntime runtime >> takeMVar stop)) $ \owner -> do+          runtime <- takeMVar readyRuntime+          Warp.testWithApplication (pure (runtimeApplication runtime)) $ \port -> withWs port $ \connection -> do+            sendClient connection Ping+            void (receive connection)+            putMVar stop ()+            receive connection >>= (\frame -> frameType frame @?= Just "goodbye")+            finished <- timeout 2000000 (wait owner)+            finished @?= Just ()+    ]++goldenTests :: TestTree+goldenTests =+  localOption (NoCreateFile True) $+    testGroup+      "frame encodings"+      ( [goldenVsString name ("test/golden/" <> name <> ".json") (pure bytes) | (name, bytes) <- fixtures]+          <> [ testCase "client round trips and server encoder parity" $ do+                 forM_ [Subscribe "orders", Unsubscribe "orders", Ping] $ \frame -> eitherDecode (encode frame) @?= Right frame+                 forM_ serverFrames $ \frame -> eitherDecode (encode frame) @?= Right (toJSON frame)+             ]+      )+  where+    epoch = UTCTime (fromGregorian 2026 9 30) (12 * 3600)+    metrics = H.QueueMetrics "orders" 3 (Just 1) (Just 9) 12 epoch 2 Nothing+    serverFrames = [Snapshot "orders" metrics True, Update "orders" metrics NotifySource, Update "orders" metrics PollSource, Pong, overflowFrame, Goodbye]+    fixtures =+      zip ["ws-subscribe", "ws-unsubscribe", "ws-ping"] (map encode [Subscribe "orders", Unsubscribe "orders", Ping])+        <> zip ["ws-snapshot", "ws-update-notify", "ws-update-poll", "ws-pong", "ws-error", "ws-goodbye"] (map encode serverFrames)++databaseTests :: IO (Pool.Pool, Pg.Database) -> TestTree+databaseTests database =+  testGroup+    "WebSocket database feed"+    [ testCase "subscribe sends snapshot then authoritative poll-sourced updates" $ do+        (pool, _) <- database+        withQueue pool $ \queue -> withTestRuntime config (poolEnv pool Nothing) $ \port _ -> withWs port $ \connection -> do+          let name = queueNameToText queue+          sendClient connection (Subscribe name)+          snapshot <- receive connection+          frameType snapshot @?= Just "snapshot"+          frameField "push" snapshot @?= Just (Bool False)+          (frameField "metrics" snapshot >>= frameField "queue_length") @?= Just (Number 0)+          void (sendN pool queue 3)+          update <- awaitFrame 2000000 connection (\frame -> frameType frame == Just "update" && (frameField "metrics" frame >>= frameField "queue_length") == Just (Number 3))+          case update of+            Nothing -> assertFailure "poll failed to converge"+            Just frame -> frameField "source" frame @?= Just (String "poll"),+      testCase "nonempty idle queues emit age changes even without writes" $ do+        (pool, _) <- database+        withQueue pool $ \queue -> do+          void (sendN pool queue 1)+          withTestRuntime config (poolEnv pool Nothing) $ \port _ -> withWs port $ \connection -> do+            sendClient connection (Subscribe (queueNameToText queue))+            snapshot <- receive connection+            let initialAge = frameField "metrics" snapshot >>= frameField "oldest_msg_age_sec"+            update <- awaitFrame 2000000 connection (\frame -> frameType frame == Just "update" && (frameField "metrics" frame >>= frameField "oldest_msg_age_sec") /= initialAge)+            case update of+              Nothing -> assertFailure "idle age change was suppressed"+              Just frame -> frameField "source" frame @?= Just (String "poll")+    ]
+ test/WireGoldenSpec.hs view
@@ -0,0 +1,29 @@+module WireGoldenSpec (tests) where++import Data.Aeson (encode)+import Pgmq.Inspect.Error (ErrorBody (..), invalidCursor)+import Pgmq.Inspect.Wire+import Pgmq.Types (MessageId (..))+import Test.Tasty (TestTree, localOption, testGroup)+import Test.Tasty.Golden (NoCreateFile (..), goldenVsString)++tests :: TestTree+tests =+  localOption (NoCreateFile True) $+    testGroup+      "Wire shapes (golden)"+      [ goldenVsString "page with a next cursor" "test/golden/page-more.json" $+          pure (encode (Page [1 :: Int, 2, 3] (Just (MessageId 3)))),+        goldenVsString "last page omits next_cursor" "test/golden/page-last.json" $+          pure (encode (Page [4 :: Int] Nothing)),+        goldenVsString "error envelope with details" "test/golden/error-invalid-cursor.json" $+          pure (encode (invalidCursor "x")),+        goldenVsString "error envelope without details" "test/golden/error-plain.json" $+          pure (encode (ErrorBody "not_found" "No route matches this path." Nothing)),+        goldenVsString "service descriptor" "test/golden/descriptor.json" $+          pure (encode (ServiceDescriptor "pgmq-inspect" "0.1.0.0" "ws" "openapi.json")),+        goldenVsString "liveness" "test/golden/health-live.json" $+          pure (encode (LivenessBody True)),+        goldenVsString "readiness" "test/golden/health-ready.json" $+          pure (encode (ReadinessBody False [DependencyCheck "postgres" False (Just 12) (Just "timed out")]))+      ]
+ test/client/inspection-client.mjs view
@@ -0,0 +1,50 @@+// Dependency-free generated-operation fixture. Requires JSON.parse reviver source+// context (Node >=22); a production browser must support it or use a lossless parser.+import assert from 'node:assert/strict';+const [mount, queue] = process.argv.slice(2);+const lossless = text => JSON.parse(text, (_key, value, context) => {+  if (typeof value === 'number' && Number.isInteger(value) && !Number.isSafeInteger(value)) {+    assert.ok(context?.source, 'exact source token support required');+    return BigInt(context.source);+  }+  return value;+});+const requested = [];+async function read(url) {+  requested.push(new URL(url).pathname);+  const response = await fetch(url);+  assert.equal(response.status, 200, await response.clone().text());+  return lossless(await response.text());+}+const descriptor = await read(mount);+assert.equal(descriptor.openapi_path, 'openapi.json');+assert.equal(descriptor.websocket_path, 'ws');+const specUrl = new URL(descriptor.openapi_path, mount);+const spec = await read(specUrl);+assert.equal(spec.servers[0].url, '.');+const base = new URL(spec.servers[0].url, specUrl);+const client = {};+for (const [path, methods] of Object.entries(spec.paths)) {+  for (const [method, operation] of Object.entries(methods)) {+    if (method !== 'get') continue;+    client[operation.operationId] = (params = {}, query = {}) => {+      const rendered = path.replace(/\{([^}]+)\}/g, (_, name) => encodeURIComponent(String(params[name])));+      const url = new URL(rendered.replace(/^\//, ''), base);+      url.search = new URLSearchParams(query).toString();+      return read(url);+    };+  }+}+const queues = await client.listQueues();+assert.ok(queues.some(q => q.queue_name === queue));+const page = await client.peekMessages({queue}, {limit: 1});+assert.equal(page.items[0].msg_id, 9007199254740993n);+assert.equal(page.next_cursor, 9007199254740993n);+const last = await client.peekMessages({queue}, {from: page.next_cursor.toString()});+assert.equal(last.items[0].msg_id, 9223372036854775807n);+assert.ok(!Object.hasOwn(last, 'next_cursor'));+const single = await client.lookupMessage({queue, msg_id: last.items[0].msg_id.toString()});+assert.equal(single.msg_id, 9223372036854775807n);+assert.ok(requested.every(path => path.startsWith(new URL(mount).pathname)));+assert.ok(requested.includes(new URL('queues', mount).pathname));+console.log(JSON.stringify({mount: new URL(mount).pathname, exact_ids: [page.next_cursor.toString(), single.msg_id.toString()], requested}));
+ test/fixtures/inspection-role.sql view
@@ -0,0 +1,22 @@+-- Run as the actual queue-creating owner, in the database containing pgmq.+-- Review existing worker grants BEFORE removing PUBLIC function execution.+CREATE ROLE pgmq_inspector LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOINHERIT NOREPLICATION NOBYPASSRLS;+ALTER ROLE pgmq_inspector SET default_transaction_read_only = on;+ALTER ROLE pgmq_inspector SET statement_timeout = '2s';+ALTER ROLE pgmq_inspector SET lock_timeout = '250ms';+GRANT USAGE ON SCHEMA pgmq TO pgmq_inspector;+GRANT SELECT ON ALL TABLES IN SCHEMA pgmq TO pgmq_inspector;+GRANT SELECT ON ALL SEQUENCES IN SCHEMA pgmq TO pgmq_inspector;+ALTER DEFAULT PRIVILEGES IN SCHEMA pgmq GRANT SELECT ON TABLES TO pgmq_inspector;+ALTER DEFAULT PRIVILEGES IN SCHEMA pgmq GRANT SELECT ON SEQUENCES TO pgmq_inspector;+REVOKE EXECUTE ON ALL FUNCTIONS IN SCHEMA pgmq FROM PUBLIC;+-- This is owner-wide, across schemas: schema-local REVOKE cannot undo the+-- global default PUBLIC EXECUTE grant. Coordinate this with the owner's apps.+ALTER DEFAULT PRIVILEGES REVOKE EXECUTE ON FUNCTIONS FROM PUBLIC;+GRANT EXECUTE ON FUNCTION+  pgmq.list_queues(), pgmq.format_table_name(text,text),+  pgmq.metrics(text), pgmq.metrics_all(),+  pgmq.list_topic_bindings(), pgmq.list_topic_bindings(text),+  pgmq.test_routing(text), pgmq.validate_routing_key(text),+  pgmq.list_notify_insert_throttles()+TO pgmq_inspector;
+ test/fixtures/pgmq-1.12.0.sql view
@@ -0,0 +1,2166 @@+------------------------------------------------------------+-- Schema, tables, records, privileges, indexes, etc+------------------------------------------------------------+-- When installed as an extension, we don't need to create the `pgmq` schema+-- because it is automatically created by postgres due to being declared in+-- the extension control file+DO+$$+BEGIN+    IF (SELECT NOT EXISTS( SELECT 1 FROM pg_extension WHERE extname = 'pgmq')) THEN+      CREATE SCHEMA IF NOT EXISTS pgmq;+    END IF;+END+$$;++-- Table where queues and metadata about them is stored+CREATE TABLE IF NOT EXISTS pgmq.meta (+    queue_name VARCHAR UNIQUE NOT NULL,+    is_partitioned BOOLEAN NOT NULL,+    is_unlogged BOOLEAN NOT NULL,+    created_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL+);++-- Grant permission to pg_monitor to all tables and sequences+-- These grants are intentionally placed here (after creating `pgmq.meta` but before creating other tables). This+-- allows the `pg_dump` output for a fresh installation to match the output for an installation that followed the+-- upgrade path.+GRANT USAGE ON SCHEMA pgmq TO pg_monitor;+GRANT SELECT ON ALL TABLES IN SCHEMA pgmq TO pg_monitor;+GRANT SELECT ON ALL SEQUENCES IN SCHEMA pgmq TO pg_monitor;+ALTER DEFAULT PRIVILEGES IN SCHEMA pgmq GRANT SELECT ON TABLES TO pg_monitor;+ALTER DEFAULT PRIVILEGES IN SCHEMA pgmq GRANT SELECT ON SEQUENCES TO pg_monitor;++-- Table to track notification throttling for queues+CREATE UNLOGGED TABLE IF NOT EXISTS pgmq.notify_insert_throttle (+    queue_name           VARCHAR UNIQUE NOT NULL -- Queue name (without 'q_' prefix)+       CONSTRAINT notify_insert_throttle_meta_queue_name_fk+            REFERENCES pgmq.meta (queue_name)+            ON DELETE CASCADE,+    throttle_interval_ms INTEGER NOT NULL DEFAULT 0, -- Min milliseconds between notifications (0 = no throttling)+    last_notified_at     TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT to_timestamp(0) -- Timestamp of last sent notification+);++CREATE INDEX IF NOT EXISTS idx_notify_throttle_active+    ON pgmq.notify_insert_throttle (queue_name, last_notified_at)+    WHERE throttle_interval_ms > 0;++CREATE TABLE IF NOT EXISTS pgmq.topic_bindings+(+    pattern        text NOT NULL, -- Wildcard pattern for routing key matching (* = one segment, # = zero or more segments)+    queue_name     text NOT NULL  -- Name of the queue that receives messages when pattern matches+        CONSTRAINT topic_bindings_meta_queue_name_fk+            REFERENCES pgmq.meta (queue_name)+            ON DELETE CASCADE,+    bound_at       TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, -- Timestamp when the binding was created+    compiled_regex text GENERATED ALWAYS AS (+        -- Pre-compile the pattern to regex for faster matching+        -- This avoids runtime compilation on every send_topic call+        '^' ||+        replace(+                replace(+                        regexp_replace(pattern, '([.+?{}()|\[\]\\^$])', '\\\1', 'g'),+                        '*', '[^.]+'+                ),+                '#', '.*'+        ) || '$'+        ) STORED,                 -- Computed column: stores the compiled regex pattern+    CONSTRAINT topic_bindings_unique_pattern_queue UNIQUE (pattern, queue_name)+);++-- Create covering index for better performance when scanning patterns+-- Includes queue_name and compiled_regex to allow index-only scans (no table access needed)+CREATE INDEX IF NOT EXISTS idx_topic_bindings_covering ON pgmq.topic_bindings (pattern) INCLUDE (queue_name, compiled_regex);++-- Allow the following `pgmq` tables to be dumped by `pg_dump` when pgmq is installed as an extension+DO+$$+BEGIN+    IF EXISTS(SELECT 1 FROM pg_extension WHERE extname = 'pgmq') THEN+        PERFORM pg_catalog.pg_extension_config_dump('pgmq.meta', '');+        PERFORM pg_catalog.pg_extension_config_dump('pgmq.notify_insert_throttle', '');+        PERFORM pg_catalog.pg_extension_config_dump('pgmq.topic_bindings', '');+    END IF;+END+$$;++-- This type has the shape of a message in a queue, and is often returned by+-- pgmq functions that return messages.+-- Note: Changing the order of fields in this type is a breaking change -- our Rust Diesel client implementation+-- expects a specific order of fields.+CREATE TYPE pgmq.message_record AS (+    msg_id BIGINT,+    read_ct INTEGER,+    enqueued_at TIMESTAMP WITH TIME ZONE,+    last_read_at TIMESTAMP WITH TIME ZONE,+    vt TIMESTAMP WITH TIME ZONE,+    message JSONB,+    headers JSONB+);++-- Note: Changing the order of fields in this type is a breaking change -- our Rust Diesel client implementation+-- expects a specific order of fields.+CREATE TYPE pgmq.queue_record AS (+    queue_name VARCHAR,+    is_partitioned BOOLEAN,+    is_unlogged BOOLEAN,+    created_at TIMESTAMP WITH TIME ZONE+);++------------------------------------------------------------+-- Functions+------------------------------------------------------------++-- prevents race conditions during queue creation by acquiring a transaction-level advisory lock+-- uses a transaction advisory lock maintain the lock until transaction commit+-- a race condition would still exist if lock was released before commit+CREATE FUNCTION pgmq.acquire_queue_lock(queue_name TEXT)+RETURNS void AS $$+BEGIN+  PERFORM pg_advisory_xact_lock(hashtext('pgmq.queue_' || queue_name));+END;+$$ LANGUAGE plpgsql;++-- read_grouped_round_robin+-- reads messages while preserving FIFO within groups and interleaving across groups (layered round-robin)+CREATE FUNCTION pgmq.read_grouped_rr(+    queue_name TEXT,+    vt INTEGER,+    qty INTEGER+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    sql TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+        $QUERY$+        WITH fifo_groups AS (+            -- Determine the absolute head (oldest) message id per FIFO group, regardless of visibility+            SELECT+                COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') AS fifo_key,+                MIN(msg_id) AS head_msg_id+            FROM pgmq.%1$I+            GROUP BY COALESCE(headers->>'x-pgmq-group', '_default_fifo_group')+        ),+        eligible_groups AS (+            -- Only groups whose head message is currently visible+            -- Acquire a transaction-level advisory lock per group to prevent concurrent selection+            SELECT+                g.fifo_key,+                g.head_msg_id,+                ROW_NUMBER() OVER (ORDER BY g.head_msg_id) AS group_priority+            FROM fifo_groups g+            JOIN pgmq.%2$I h ON h.msg_id = g.head_msg_id+            WHERE h.vt <= clock_timestamp()+              AND pg_try_advisory_xact_lock(pg_catalog.hashtextextended(g.fifo_key, 0))+        ),+        available_messages AS (+            -- All currently visible messages starting at the head for each eligible group+            SELECT+                m.msg_id,+                eg.group_priority,+                ROW_NUMBER() OVER (+                    PARTITION BY eg.fifo_key+                    ORDER BY m.msg_id+                ) AS msg_rank_in_group+            FROM pgmq.%3$I m+            JOIN eligible_groups eg+              ON COALESCE(m.headers->>'x-pgmq-group', '_default_fifo_group') = eg.fifo_key+            WHERE m.vt <= clock_timestamp()+              AND m.msg_id >= eg.head_msg_id+        ),+        ordered_messages AS (+            -- Layered round-robin: take rank 1 of all groups by group_priority, then rank 2, etc.+            -- Assign selection order before locking+            SELECT msg_id, ROW_NUMBER() OVER (ORDER BY msg_rank_in_group, group_priority) as selection_order+            FROM available_messages+        ),+        selected_messages AS (+            -- Lock the messages in the correct order, preserving selection_order+            SELECT om.msg_id, om.selection_order+            FROM ordered_messages om+            JOIN pgmq.%4$I m ON m.msg_id = om.msg_id+            WHERE om.selection_order <= $1+            ORDER BY om.selection_order+            FOR UPDATE OF m SKIP LOCKED+        ),+        updated_messages AS (+            UPDATE pgmq.%5$I m+            SET+                vt = clock_timestamp() + %6$L,+                read_ct = read_ct + 1,+                last_read_at = clock_timestamp()+            FROM selected_messages sm+            WHERE m.msg_id = sm.msg_id+              AND m.vt <= clock_timestamp() -- final guard to avoid duplicate reads under races+            RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers, sm.selection_order+        )+        SELECT msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers+        FROM updated_messages+        ORDER BY selection_order;+        $QUERY$,+        qtable, qtable, qtable, qtable, qtable, make_interval(secs => vt)+    );+    RETURN QUERY EXECUTE sql USING qty;+END;+$$ LANGUAGE plpgsql;++-- read_grouped_rr_with_poll+-- reads messages using round-robin layering across groups, with polling support+CREATE FUNCTION pgmq.read_grouped_rr_with_poll(+    queue_name TEXT,+    vt INTEGER,+    qty INTEGER,+    max_poll_seconds INTEGER DEFAULT 5,+    poll_interval_ms INTEGER DEFAULT 100+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    r pgmq.message_record;+    stop_at TIMESTAMP;+BEGIN+    stop_at := clock_timestamp() + make_interval(secs => max_poll_seconds);+    LOOP+      IF (SELECT clock_timestamp() >= stop_at) THEN+        RETURN;+      END IF;++      FOR r IN+        SELECT * FROM pgmq.read_grouped_rr(queue_name, vt, qty)+      LOOP+        RETURN NEXT r;+      END LOOP;+      IF FOUND THEN+        RETURN;+      ELSE+        PERFORM pg_sleep(poll_interval_ms::numeric / 1000);+      END IF;+    END LOOP;+END;+$$ LANGUAGE plpgsql;++-- read_grouped_head:  read the head of N different FIFO groups in a single operation.+-- This supports horizontal scaling by processing groups in parallel while ensuring message ordering is preserved per group.+CREATE FUNCTION pgmq.read_grouped_head(+    queue_name TEXT,+    vt INTEGER,+    qty INTEGER+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    sql TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+        $QUERY$+        WITH fifo_groups AS (+            -- Determine the absolute head (oldest) message id per FIFO group, regardless of visibility+            SELECT+                COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') AS fifo_key,+                MIN(msg_id) AS head_msg_id+            FROM pgmq.%1$I+            GROUP BY COALESCE(headers->>'x-pgmq-group', '_default_fifo_group')+        ),+        selected_messages AS (+            -- Take at most 1 message per group+            SELECT g.head_msg_id msg_id+            FROM fifo_groups g+            JOIN pgmq.%1$I q ON q.msg_id = g.head_msg_id+	        WHERE q.vt <= clock_timestamp()+            ORDER BY q.msg_id+            LIMIT $1+            FOR UPDATE SKIP LOCKED+        )+        UPDATE pgmq.%1$I m+        SET+            vt = clock_timestamp() + %2$L,+            read_ct = read_ct + 1,+            last_read_at = clock_timestamp()+        FROM selected_messages sm+        WHERE m.msg_id = sm.msg_id+        RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers;+        $QUERY$,+        qtable, make_interval(secs => vt)+    );+    RETURN QUERY EXECUTE sql USING qty;+END;+$$ LANGUAGE plpgsql;++-- read_grouped_head_with_poll+-- reads the head of N different FIFO groups in a single operation, with polling support+CREATE FUNCTION pgmq.read_grouped_head_with_poll(+    queue_name TEXT,+    vt INTEGER,+    qty INTEGER,+    max_poll_seconds INTEGER DEFAULT 5,+    poll_interval_ms INTEGER DEFAULT 100+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    r pgmq.message_record;+    stop_at TIMESTAMPTZ;+BEGIN+    stop_at := clock_timestamp() + make_interval(secs => max_poll_seconds);+    LOOP+      IF clock_timestamp() >= stop_at THEN+        RETURN;+      END IF;++      FOR r IN+        SELECT * FROM pgmq.read_grouped_head(queue_name, vt, qty)+      LOOP+        RETURN NEXT r;+      END LOOP;+      IF FOUND THEN+        RETURN;+      ELSE+        PERFORM pg_sleep(poll_interval_ms::numeric / 1000);+      END IF;+    END LOOP;+END;+$$ LANGUAGE plpgsql;++-- a helper to format table names and check for invalid characters+CREATE FUNCTION pgmq.format_table_name(queue_name text, prefix text)+RETURNS TEXT AS $$+BEGIN+    IF queue_name ~ '\$|;|--|'''+    THEN+        RAISE EXCEPTION 'queue name contains invalid characters: $, ;, --, or \''';+    END IF;+    RETURN lower(prefix || '_' || queue_name);+END;+$$ LANGUAGE plpgsql;++-- read+-- reads a number of messages from a queue, setting a visibility timeout on them+CREATE FUNCTION pgmq.read(+    queue_name TEXT,+    vt INTEGER,+    qty INTEGER,+    conditional JSONB DEFAULT '{}'+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    sql TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+        $QUERY$+        WITH cte AS+        (+            SELECT msg_id+            FROM pgmq.%I+            WHERE vt <= clock_timestamp() AND CASE+                WHEN %L != '{}'::jsonb THEN (message @> %2$L)::integer+                ELSE 1+            END = 1+            ORDER BY msg_id ASC+            LIMIT $1+            FOR UPDATE SKIP LOCKED+        )+        UPDATE pgmq.%I m+        SET+            last_read_at = clock_timestamp(),+            vt = clock_timestamp() + %L,+            read_ct = read_ct + 1+        FROM cte+        WHERE m.msg_id = cte.msg_id+        RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers;+        $QUERY$,+        qtable, conditional, qtable, make_interval(secs => vt)+    );+    RETURN QUERY EXECUTE sql USING qty;+END;+$$ LANGUAGE plpgsql;++-- read_grouped+-- reads messages with AWS SQS FIFO-style batch retrieval behavior+-- attempts to return as many messages as possible from the same message group+CREATE FUNCTION pgmq.read_grouped(+    queue_name TEXT,+    vt INTEGER,+    qty INTEGER+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    sql TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+        $QUERY$+        WITH fifo_groups AS (+            -- Find the minimum msg_id for each FIFO group that's ready to be processed+            SELECT+                COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') as fifo_key,+                MIN(msg_id) as min_msg_id+            FROM pgmq.%I+            WHERE vt <= clock_timestamp()+            GROUP BY COALESCE(headers->>'x-pgmq-group', '_default_fifo_group')+        ),+        locked_groups AS (+            -- Lock the first available message in each FIFO group+            SELECT+                m.msg_id,+                fg.fifo_key+            FROM pgmq.%I m+            INNER JOIN fifo_groups fg ON+                COALESCE(m.headers->>'x-pgmq-group', '_default_fifo_group') = fg.fifo_key+                AND m.msg_id = fg.min_msg_id+            WHERE m.vt <= clock_timestamp()+            ORDER BY m.msg_id ASC+            FOR UPDATE SKIP LOCKED+        ),+        group_priorities AS (+            -- Assign priority to groups based on their oldest message+            SELECT+                fifo_key,+                msg_id as min_msg_id,+                ROW_NUMBER() OVER (ORDER BY msg_id) as group_priority+            FROM locked_groups+        ),+        filtered_groups as (+            SELECT * FROM group_priorities gp+            WHERE NOT EXISTS (+                -- Ensure no earlier message in this group is currently being processed+                SELECT 1+                FROM pgmq.%I m2+                WHERE COALESCE(m2.headers->>'x-pgmq-group', '_default_fifo_group') = gp.fifo_key+                AND m2.vt > clock_timestamp()+                AND m2.msg_id < gp.min_msg_id+            )+        ),+        available_messages as (+            SELECT gp.fifo_key, t.msg_id,gp.group_priority,+                ROW_NUMBER() OVER (PARTITION BY gp.fifo_key ORDER BY t.msg_id) as msg_rank_in_group+            FROM filtered_groups gp+            CROSS JOIN LATERAL (+                SELECT *+                FROM pgmq.%I t+                WHERE COALESCE(t.headers->>'x-pgmq-group', '_default_fifo_group') = gp.fifo_key+                AND t.vt <= clock_timestamp()+                ORDER BY msg_id+                LIMIT $1  -- tip to limit query impact, we know we need at most qty in each group+            ) t+            ORDER BY gp.group_priority+        ),+        batch_selection AS (+            -- Select messages to fill batch, prioritizing earliest group+            SELECT+                msg_id,+                ROW_NUMBER() OVER (ORDER BY group_priority, msg_rank_in_group) as overall_rank+            FROM available_messages+        ),+        selected_messages AS (+            -- Limit to requested quantity+            SELECT msg_id+            FROM batch_selection+            WHERE overall_rank <= $1+            ORDER BY msg_id+            FOR UPDATE SKIP LOCKED+        )+        UPDATE pgmq.%I m+        SET+            vt = clock_timestamp() + %L,+            read_ct = read_ct + 1,+            last_read_at = clock_timestamp()+        FROM selected_messages sm+        WHERE m.msg_id = sm.msg_id+        RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers;+        $QUERY$,+        qtable, qtable, qtable, qtable, qtable, make_interval(secs => vt)+    );+    RETURN QUERY EXECUTE sql USING qty;+END;+$$ LANGUAGE plpgsql;++-- read_grouped_with_poll+-- reads messages with AWS SQS FIFO-style batch retrieval behavior, with polling support+CREATE FUNCTION pgmq.read_grouped_with_poll(+    queue_name TEXT,+    vt INTEGER,+    qty INTEGER,+    max_poll_seconds INTEGER DEFAULT 5,+    poll_interval_ms INTEGER DEFAULT 100+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    r pgmq.message_record;+    stop_at TIMESTAMP;+BEGIN+    stop_at := clock_timestamp() + make_interval(secs => max_poll_seconds);+    LOOP+      IF (SELECT clock_timestamp() >= stop_at) THEN+        RETURN;+      END IF;++      FOR r IN+        SELECT * FROM pgmq.read_grouped(queue_name, vt, qty)+      LOOP+        RETURN NEXT r;+      END LOOP;+      IF FOUND THEN+        RETURN;+      ELSE+        PERFORM pg_sleep(poll_interval_ms::numeric / 1000);+      END IF;+    END LOOP;+END;+$$ LANGUAGE plpgsql;++---- read_with_poll+---- reads a number of messages from a queue, setting a visibility timeout on them+CREATE FUNCTION pgmq.read_with_poll(+    queue_name TEXT,+    vt INTEGER,+    qty INTEGER,+    max_poll_seconds INTEGER DEFAULT 5,+    poll_interval_ms INTEGER DEFAULT 100,+    conditional JSONB DEFAULT '{}'+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    r pgmq.message_record;+    stop_at TIMESTAMP;+    sql TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    stop_at := clock_timestamp() + make_interval(secs => max_poll_seconds);+    LOOP+      IF (SELECT clock_timestamp() >= stop_at) THEN+        RETURN;+      END IF;++      sql := FORMAT(+          $QUERY$+          WITH cte AS+          (+              SELECT msg_id+              FROM pgmq.%I+              WHERE vt <= clock_timestamp() AND CASE+                  WHEN %L != '{}'::jsonb THEN (message @> %2$L)::integer+                  ELSE 1+              END = 1+              ORDER BY msg_id ASC+              LIMIT $1+              FOR UPDATE SKIP LOCKED+          )+          UPDATE pgmq.%I m+          SET+              last_read_at = clock_timestamp(),+              vt = clock_timestamp() + %L,+              read_ct = read_ct + 1+          FROM cte+          WHERE m.msg_id = cte.msg_id+          RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers;+          $QUERY$,+          qtable, conditional, qtable, make_interval(secs => vt)+      );++      FOR r IN+        EXECUTE sql USING qty+      LOOP+        RETURN NEXT r;+      END LOOP;+      IF FOUND THEN+        RETURN;+      ELSE+        PERFORM pg_sleep(poll_interval_ms::numeric / 1000);+      END IF;+    END LOOP;+END;+$$ LANGUAGE plpgsql;++---- archive+---- removes a message from the queue, and sends it to the archive, where its+---- saved permanently.+CREATE FUNCTION pgmq.archive(+    queue_name TEXT,+    msg_id BIGINT+)+RETURNS BOOLEAN AS $$+DECLARE+    sql TEXT;+    result BIGINT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+    atable TEXT := pgmq.format_table_name(queue_name, 'a');+BEGIN+    sql := FORMAT(+        $QUERY$+        WITH archived AS (+            DELETE FROM pgmq.%I+            WHERE msg_id = $1+            RETURNING msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers+        )+        INSERT INTO pgmq.%I (msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers)+        SELECT msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers+        FROM archived+        RETURNING msg_id;+        $QUERY$,+        qtable, atable+    );+    EXECUTE sql USING msg_id INTO result;+    RETURN NOT (result IS NULL);+END;+$$ LANGUAGE plpgsql;++---- archive+---- removes an array of message ids from the queue, and sends it to the archive,+---- where these messages will be saved permanently.+CREATE FUNCTION pgmq.archive(+    queue_name TEXT,+    msg_ids BIGINT[]+)+RETURNS SETOF BIGINT AS $$+DECLARE+    sql TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+    atable TEXT := pgmq.format_table_name(queue_name, 'a');+BEGIN+    sql := FORMAT(+        $QUERY$+        WITH archived AS (+            DELETE FROM pgmq.%I+            WHERE msg_id = ANY($1)+            RETURNING msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers+        )+        INSERT INTO pgmq.%I (msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers)+        SELECT msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers+        FROM archived+        RETURNING msg_id;+        $QUERY$,+        qtable, atable+    );+    RETURN QUERY EXECUTE sql USING msg_ids;+END;+$$ LANGUAGE plpgsql;++---- delete+---- deletes a message id from the queue permanently+CREATE FUNCTION pgmq.delete(+    queue_name TEXT,+    msg_id BIGINT+)+RETURNS BOOLEAN AS $$+DECLARE+    sql TEXT;+    result BIGINT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+        $QUERY$+        DELETE FROM pgmq.%I+        WHERE msg_id = $1+        RETURNING msg_id+        $QUERY$,+        qtable+    );+    EXECUTE sql USING msg_id INTO result;+    RETURN NOT (result IS NULL);+END;+$$ LANGUAGE plpgsql;++---- delete+---- deletes an array of message ids from the queue permanently+CREATE FUNCTION pgmq.delete(+    queue_name TEXT,+    msg_ids BIGINT[]+)+RETURNS SETOF BIGINT AS $$+DECLARE+    sql TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+        $QUERY$+        DELETE FROM pgmq.%I+        WHERE msg_id = ANY($1)+        RETURNING msg_id+        $QUERY$,+        qtable+    );+    RETURN QUERY EXECUTE sql USING msg_ids;+END;+$$ LANGUAGE plpgsql;++-- send: actual implementation+CREATE FUNCTION pgmq.send(+    queue_name TEXT,+    msg JSONB,+    headers JSONB,+    delay TIMESTAMP WITH TIME ZONE+) RETURNS SETOF BIGINT AS $$+DECLARE+    sql TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+            $QUERY$+        INSERT INTO pgmq.%I (vt, message, headers)+        VALUES ($2, $1, $3)+        RETURNING msg_id;+        $QUERY$,+            qtable+           );+    RETURN QUERY EXECUTE sql USING msg, delay, headers;+END;+$$ LANGUAGE plpgsql;++-- send: 2 args, no delay or headers+CREATE FUNCTION pgmq.send(+    queue_name TEXT,+    msg JSONB+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send(queue_name, msg, NULL, clock_timestamp());+$$ LANGUAGE sql;++-- send: 3 args with headers+CREATE FUNCTION pgmq.send(+    queue_name TEXT,+    msg JSONB,+    headers JSONB+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send(queue_name, msg, headers, clock_timestamp());+$$ LANGUAGE sql;++-- send: 3 args with integer delay+CREATE FUNCTION pgmq.send(+    queue_name TEXT,+    msg JSONB,+    delay INTEGER+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send(queue_name, msg, NULL, clock_timestamp() + make_interval(secs => delay));+$$ LANGUAGE sql;++-- send: 3 args with timestamp+CREATE FUNCTION pgmq.send(+    queue_name TEXT,+    msg JSONB,+    delay TIMESTAMP WITH TIME ZONE+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send(queue_name, msg, NULL, delay);+$$ LANGUAGE sql;++-- send: 4 args with integer delay+CREATE FUNCTION pgmq.send(+    queue_name TEXT,+    msg JSONB,+    headers JSONB,+    delay INTEGER+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send(queue_name, msg, headers, clock_timestamp() + make_interval(secs => delay));+$$ LANGUAGE sql;++-- _validate_batch_params: Private function to validate batch parameters+CREATE FUNCTION pgmq._validate_batch_params(+    msgs JSONB[],+    headers JSONB[]+) RETURNS void AS $$+BEGIN+    -- Validate that msgs is not NULL or empty+    IF msgs IS NULL OR array_length(msgs, 1) IS NULL THEN+        RAISE EXCEPTION 'msgs cannot be NULL or empty';+    END IF;++    -- Validate that headers array length matches msgs array length if headers is provided+    -- Note: array_length returns NULL for empty arrays, so we use COALESCE to treat empty arrays as length 0+    IF headers IS NOT NULL AND COALESCE(array_length(headers, 1), 0) != COALESCE(array_length(msgs, 1), 0) THEN+        RAISE EXCEPTION 'headers array length (%) must match msgs array length (%)',+            COALESCE(array_length(headers, 1), 0), COALESCE(array_length(msgs, 1), 0);+    END IF;+END;+$$ LANGUAGE plpgsql;++-- _send_batch: Private function that performs the actual batch insert without validation+CREATE FUNCTION pgmq._send_batch(+    queue_name TEXT,+    msgs JSONB[],+    headers JSONB[],+    delay TIMESTAMP WITH TIME ZONE+) RETURNS SETOF BIGINT AS $$+DECLARE+    sql TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+            $QUERY$+        INSERT INTO pgmq.%I (vt, message, headers)+        SELECT $2, unnest($1), unnest(coalesce($3, ARRAY[]::jsonb[]))+        RETURNING msg_id;+        $QUERY$,+            qtable+           );+    RETURN QUERY EXECUTE sql USING msgs, delay, headers;+END;+$$ LANGUAGE plpgsql;++-- send_batch: Public function with validation+CREATE FUNCTION pgmq.send_batch(+    queue_name TEXT,+    msgs JSONB[],+    headers JSONB[],+    delay TIMESTAMP WITH TIME ZONE+) RETURNS SETOF BIGINT AS $$+BEGIN+    PERFORM pgmq._validate_batch_params(msgs, headers);+    RETURN QUERY SELECT * FROM pgmq._send_batch(queue_name, msgs, headers, delay);+END;+$$ LANGUAGE plpgsql;++-- send batch: 2 args+CREATE FUNCTION pgmq.send_batch(+    queue_name TEXT,+    msgs JSONB[]+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send_batch(queue_name, msgs, NULL, clock_timestamp());+$$ LANGUAGE sql;++-- send batch: 3 args with headers+CREATE FUNCTION pgmq.send_batch(+    queue_name TEXT,+    msgs JSONB[],+    headers JSONB[]+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send_batch(queue_name, msgs, headers, clock_timestamp());+$$ LANGUAGE sql;++-- send batch: 3 args with integer delay+CREATE FUNCTION pgmq.send_batch(+    queue_name TEXT,+    msgs JSONB[],+    delay INTEGER+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send_batch(queue_name, msgs, NULL, clock_timestamp() + make_interval(secs => delay));+$$ LANGUAGE sql;++-- send batch: 3 args with timestamp+CREATE FUNCTION pgmq.send_batch(+    queue_name TEXT,+    msgs JSONB[],+    delay TIMESTAMP WITH TIME ZONE+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send_batch(queue_name, msgs, NULL, delay);+$$ LANGUAGE sql;++-- send_batch: 4 args with integer delay+CREATE FUNCTION pgmq.send_batch(+    queue_name TEXT,+    msgs JSONB[],+    headers JSONB[],+    delay INTEGER+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send_batch(queue_name, msgs, headers, clock_timestamp() + make_interval(secs => delay));+$$ LANGUAGE sql;++-- returned by pgmq.metrics() and pgmq.metrics_all+-- Note: Changing the order of fields in this type is a breaking change -- our Rust Diesel client implementation+-- expects a specific order of fields.+CREATE TYPE pgmq.metrics_result AS (+    queue_name text,+    queue_length bigint,+    newest_msg_age_sec int,+    oldest_msg_age_sec int,+    total_messages bigint,+    scrape_time timestamp with time zone,+    queue_visible_length bigint+);++-- get metrics for a single queue+CREATE FUNCTION pgmq.metrics(queue_name TEXT)+RETURNS pgmq.metrics_result AS $$+DECLARE+    result_row pgmq.metrics_result;+    query TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    query := FORMAT(+        $QUERY$+        WITH q_summary AS (+            SELECT+                count(*) as queue_length,+                count(CASE WHEN vt <= NOW() THEN 1 END) as queue_visible_length,+                EXTRACT(epoch FROM (NOW() - max(enqueued_at)))::int as newest_msg_age_sec,+                EXTRACT(epoch FROM (NOW() - min(enqueued_at)))::int as oldest_msg_age_sec,+                NOW() as scrape_time+            FROM pgmq.%I+        ),+        all_metrics AS (+            SELECT CASE+                WHEN is_called THEN last_value ELSE 0+                END as total_messages+            FROM pgmq.%I+        )+        SELECT+            %L as queue_name,+            q_summary.queue_length,+            q_summary.newest_msg_age_sec,+            q_summary.oldest_msg_age_sec,+            all_metrics.total_messages,+            q_summary.scrape_time,+            q_summary.queue_visible_length+        FROM q_summary, all_metrics+        $QUERY$,+        qtable, qtable || '_msg_id_seq', queue_name+    );+    EXECUTE query INTO result_row;+    RETURN result_row;+END;+$$ LANGUAGE plpgsql;++-- get metrics for all queues+CREATE FUNCTION pgmq."metrics_all"()+RETURNS SETOF pgmq.metrics_result AS $$+DECLARE+    row_name RECORD;+    result_row pgmq.metrics_result;+BEGIN+    FOR row_name IN SELECT queue_name FROM pgmq.meta LOOP+        result_row := pgmq.metrics(row_name.queue_name);+        RETURN NEXT result_row;+    END LOOP;+END;+$$ LANGUAGE plpgsql;++-- list queues+CREATE FUNCTION pgmq."list_queues"()+RETURNS SETOF pgmq.queue_record AS $$+BEGIN+  RETURN QUERY SELECT * FROM pgmq.meta;+END+$$ LANGUAGE plpgsql;++-- purge queue, deleting all entries in it.+CREATE OR REPLACE FUNCTION pgmq."purge_queue"(queue_name TEXT)+RETURNS BIGINT AS $$+DECLARE+  deleted_count INTEGER;+  qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+  -- Get the row count before truncating+  EXECUTE format('SELECT count(*) FROM pgmq.%I', qtable) INTO deleted_count;++  -- Use TRUNCATE for better performance on large tables+  EXECUTE format('TRUNCATE TABLE pgmq.%I', qtable);++  -- Return the number of purged rows+  RETURN deleted_count;+END+$$ LANGUAGE plpgsql;++-- unassign archive, so it can be kept when a queue is deleted+CREATE FUNCTION pgmq."detach_archive"(queue_name TEXT)+RETURNS VOID AS $$+DECLARE+  atable TEXT := pgmq.format_table_name(queue_name, 'a');+BEGIN+  RAISE WARNING 'detach_archive(queue_name) is deprecated and is a no-op. It will be removed in PGMQ v2.0. Archive tables are no longer member objects.';+END+$$ LANGUAGE plpgsql;++-- pop: implementation+CREATE FUNCTION pgmq.pop(queue_name TEXT, qty INTEGER DEFAULT 1)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    sql TEXT;+    result pgmq.message_record;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+        $QUERY$+        WITH cte AS+            (+                SELECT msg_id+                FROM pgmq.%I+                WHERE vt <= clock_timestamp()+                ORDER BY msg_id ASC+                LIMIT $1+                FOR UPDATE SKIP LOCKED+            )+        DELETE from pgmq.%I+        WHERE msg_id IN (select msg_id from cte)+        RETURNING msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers;+        $QUERY$,+        qtable, qtable+    );+    RETURN QUERY EXECUTE sql USING qty;+END;+$$ LANGUAGE plpgsql;++-- Sets timestamp vt of a message, returns it+CREATE FUNCTION pgmq.set_vt(queue_name TEXT, msg_id BIGINT, vt TIMESTAMP WITH TIME ZONE)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    sql TEXT;+    result pgmq.message_record;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+        $QUERY$+        UPDATE pgmq.%I+        SET vt = $1+        WHERE msg_id = $2+        RETURNING msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers;+        $QUERY$, +        qtable+    );+    RETURN QUERY EXECUTE sql USING vt, msg_id;+END;+$$ LANGUAGE plpgsql;++-- Sets integer vt of a message, returns it+CREATE FUNCTION pgmq.set_vt(queue_name TEXT, msg_id BIGINT, vt INTEGER)+RETURNS SETOF pgmq.message_record AS $$+    SELECT * FROM pgmq.set_vt(queue_name, msg_id, clock_timestamp() + make_interval(secs => vt));+$$ LANGUAGE sql;++-- Sets timestamp vt of multiple messages, returns them+CREATE FUNCTION pgmq.set_vt(+    queue_name TEXT,+    msg_ids BIGINT[],+    vt TIMESTAMP WITH TIME ZONE+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    sql TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+        $QUERY$+        UPDATE pgmq.%I+        SET vt = $1+        WHERE msg_id = ANY($2)+        RETURNING msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers;+        $QUERY$,+        qtable+    );+    RETURN QUERY EXECUTE sql USING vt, msg_ids;+END;+$$ LANGUAGE plpgsql;++-- Sets integer vt of multiple messages, returns them+CREATE FUNCTION pgmq.set_vt(+    queue_name TEXT,+    msg_ids BIGINT[],+    vt INTEGER+)+RETURNS SETOF pgmq.message_record AS $$+    SELECT * FROM pgmq.set_vt(queue_name, msg_ids, clock_timestamp() + make_interval(secs => vt));+$$ LANGUAGE sql;++CREATE FUNCTION pgmq._get_pg_partman_schema()+RETURNS TEXT AS $$+  SELECT+    extnamespace::regnamespace::text+  FROM+    pg_extension+  WHERE+    extname = 'pg_partman';+$$ LANGUAGE SQL;++CREATE FUNCTION pgmq.drop_queue(queue_name TEXT, partitioned BOOLEAN)+RETURNS BOOLEAN AS $$+DECLARE+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+    fq_qtable TEXT := 'pgmq.' || qtable;+    atable TEXT := pgmq.format_table_name(queue_name, 'a');+    fq_atable TEXT := 'pgmq.' || atable;+BEGIN+    RAISE WARNING 'drop_queue(queue_name, partitioned) is deprecated and will be removed in PGMQ v2.0. Use drop_queue(queue_name) instead';++    PERFORM pgmq.drop_queue(queue_name);++    RETURN TRUE;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.drop_queue(queue_name TEXT)+RETURNS BOOLEAN AS $$+DECLARE+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+    qtable_seq TEXT := qtable || '_msg_id_seq';+    fq_qtable TEXT := 'pgmq.' || qtable;+    atable TEXT := pgmq.format_table_name(queue_name, 'a');+    fq_atable TEXT := 'pgmq.' || atable;+    partitioned BOOLEAN;+BEGIN+    PERFORM pgmq.acquire_queue_lock(queue_name);+    EXECUTE FORMAT(+        $QUERY$+        SELECT is_partitioned FROM pgmq.meta WHERE queue_name = %L+        $QUERY$,+        queue_name+    ) INTO partitioned;++    -- check if the queue exists+    IF NOT EXISTS (+        SELECT 1+        FROM information_schema.tables+        WHERE table_name = qtable and table_schema = 'pgmq'+    ) THEN+        RAISE NOTICE 'pgmq queue `%` does not exist', queue_name;+        RETURN FALSE;+    END IF;++    EXECUTE FORMAT(+        $QUERY$+        DROP TABLE IF EXISTS pgmq.%I+        $QUERY$,+        qtable+    );++    EXECUTE FORMAT(+        $QUERY$+        DROP TABLE IF EXISTS pgmq.%I+        $QUERY$,+        atable+    );++     IF EXISTS (+          SELECT 1+          FROM information_schema.tables+          WHERE table_name = 'meta' and table_schema = 'pgmq'+     ) THEN+        EXECUTE FORMAT(+            $QUERY$+            DELETE FROM pgmq.meta WHERE queue_name = %L+            $QUERY$,+            queue_name+        );+     END IF;++     IF partitioned THEN+        EXECUTE FORMAT(+          $QUERY$+          DELETE FROM %I.part_config where parent_table in (%L, %L)+          $QUERY$,+          pgmq._get_pg_partman_schema(), fq_qtable, fq_atable+        );+     END IF;++    RETURN TRUE;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.validate_queue_name(queue_name TEXT)+RETURNS void AS $$+BEGIN+  IF length(queue_name) > 47 THEN+    -- complete table identifier must be <= 63+    -- https://www.postgresql.org/docs/17/sql-syntax-lexical.html#SQL-SYNTAX-IDENTIFIERS+    -- e.g. template_pgmq_q_my_queue is an identifier for my_queue when partitioned+    -- template_pgmq_q_ (16) + <a max length queue name> (47) = 63 +    RAISE EXCEPTION 'queue name is too long, maximum length is 47 characters';+  END IF;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq._belongs_to_pgmq(table_name TEXT)+RETURNS BOOLEAN AS $$+DECLARE+    sql TEXT;+    result BOOLEAN;+BEGIN+  SELECT EXISTS (+    SELECT 1+    FROM pg_depend+    WHERE refobjid = (SELECT oid FROM pg_extension WHERE extname = 'pgmq')+    AND objid = (+        SELECT oid+        FROM pg_class+        WHERE relname = table_name+    )+  ) INTO result;+  RETURN result;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.create_non_partitioned(queue_name TEXT)+RETURNS void AS $$+DECLARE+  qtable TEXT := pgmq.format_table_name(queue_name, 'q');+  qtable_seq TEXT := qtable || '_msg_id_seq';+  atable TEXT := pgmq.format_table_name(queue_name, 'a');+BEGIN+  PERFORM pgmq.validate_queue_name(queue_name);+  PERFORM pgmq.acquire_queue_lock(queue_name);++  EXECUTE FORMAT(+    $QUERY$+    CREATE TABLE IF NOT EXISTS pgmq.%I (+        msg_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,+        read_ct INT DEFAULT 0 NOT NULL,+        enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+        last_read_at TIMESTAMP WITH TIME ZONE,+        vt TIMESTAMP WITH TIME ZONE NOT NULL,+        message JSONB,+        headers JSONB+    )+    $QUERY$,+    qtable+  );++  EXECUTE FORMAT(+    $QUERY$+    CREATE TABLE IF NOT EXISTS pgmq.%I (+      msg_id BIGINT PRIMARY KEY,+      read_ct INT DEFAULT 0 NOT NULL,+      enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+      last_read_at TIMESTAMP WITH TIME ZONE,+      archived_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+      vt TIMESTAMP WITH TIME ZONE NOT NULL,+      message JSONB,+      headers JSONB+    );+    $QUERY$,+    atable+  );++  EXECUTE FORMAT(+    $QUERY$+    CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (vt ASC);+    $QUERY$,+    qtable || '_vt_idx', qtable+  );++  EXECUTE FORMAT(+    $QUERY$+    CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (archived_at);+    $QUERY$,+    'archived_at_idx_' || queue_name, atable+  );++  EXECUTE FORMAT(+    $QUERY$+    INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged)+    VALUES (%L, false, false)+    ON CONFLICT+    DO NOTHING;+    $QUERY$,+    queue_name+  );++END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.create_unlogged(queue_name TEXT)+RETURNS void AS $$+DECLARE+  qtable TEXT := pgmq.format_table_name(queue_name, 'q');+  qtable_seq TEXT := qtable || '_msg_id_seq';+  atable TEXT := pgmq.format_table_name(queue_name, 'a');+BEGIN+  PERFORM pgmq.validate_queue_name(queue_name);+  PERFORM pgmq.acquire_queue_lock(queue_name);++  EXECUTE FORMAT(+    $QUERY$+    CREATE UNLOGGED TABLE IF NOT EXISTS pgmq.%I (+        msg_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,+        read_ct INT DEFAULT 0 NOT NULL,+        enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+        last_read_at TIMESTAMP WITH TIME ZONE,+        vt TIMESTAMP WITH TIME ZONE NOT NULL,+        message JSONB,+        headers JSONB+    )+    $QUERY$,+    qtable+  );++  EXECUTE FORMAT(+    $QUERY$+    CREATE TABLE IF NOT EXISTS pgmq.%I (+      msg_id BIGINT PRIMARY KEY,+      read_ct INT DEFAULT 0 NOT NULL,+      enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+      last_read_at TIMESTAMP WITH TIME ZONE,+      archived_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+      vt TIMESTAMP WITH TIME ZONE NOT NULL,+      message JSONB,+      headers JSONB+    );+    $QUERY$,+    atable+  );++  EXECUTE FORMAT(+    $QUERY$+    CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (vt ASC);+    $QUERY$,+    qtable || '_vt_idx', qtable+  );++  EXECUTE FORMAT(+    $QUERY$+    CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (archived_at);+    $QUERY$,+    'archived_at_idx_' || queue_name, atable+  );++  EXECUTE FORMAT(+    $QUERY$+    INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged)+    VALUES (%L, false, true)+    ON CONFLICT+    DO NOTHING;+    $QUERY$,+    queue_name+  );+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq._get_partition_col(partition_interval TEXT)+RETURNS TEXT AS $$+DECLARE+  num INTEGER;+BEGIN+    BEGIN+        num := partition_interval::INTEGER;+        RETURN 'msg_id';+    EXCEPTION+        WHEN others THEN+            RETURN 'enqueued_at';+    END;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq._extension_exists(extension_name TEXT)+    RETURNS BOOLEAN+    LANGUAGE SQL+AS $$+SELECT EXISTS (+    SELECT 1+    FROM pg_extension+    WHERE extname = extension_name+)+$$;++CREATE FUNCTION pgmq._ensure_pg_partman_installed()+RETURNS void AS $$+BEGIN+  IF NOT pgmq._extension_exists('pg_partman') THEN+    RAISE EXCEPTION 'pg_partman is required for partitioned queues';+  END IF;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq._get_pg_partman_major_version()+RETURNS INT+LANGUAGE SQL+AS $$+  SELECT split_part(extversion, '.', 1)::INT+  FROM pg_extension+  WHERE extname = 'pg_partman'+$$;++CREATE FUNCTION pgmq.create_partitioned(+  queue_name TEXT,+  partition_interval TEXT DEFAULT '10000',+  retention_interval TEXT DEFAULT '100000'+)+RETURNS void AS $$+DECLARE+  partition_col TEXT;+  a_partition_col TEXT;+  qtable TEXT := pgmq.format_table_name(queue_name, 'q');+  qtable_seq TEXT := qtable || '_msg_id_seq';+  atable TEXT := pgmq.format_table_name(queue_name, 'a');+  fq_qtable TEXT := 'pgmq.' || qtable;+  fq_atable TEXT := 'pgmq.' || atable;+BEGIN+  PERFORM pgmq.validate_queue_name(queue_name);+  PERFORM pgmq.acquire_queue_lock(queue_name);+  PERFORM pgmq._ensure_pg_partman_installed();+  SELECT pgmq._get_partition_col(partition_interval) INTO partition_col;++  EXECUTE FORMAT(+    $QUERY$+    CREATE TABLE IF NOT EXISTS pgmq.%I (+        msg_id BIGINT GENERATED ALWAYS AS IDENTITY,+        read_ct INT DEFAULT 0 NOT NULL,+        enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+        last_read_at TIMESTAMP WITH TIME ZONE,+        vt TIMESTAMP WITH TIME ZONE NOT NULL,+        message JSONB,+        headers JSONB+    ) PARTITION BY RANGE (%I)+    $QUERY$,+    qtable, partition_col+  );++  -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md+  -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema.+  EXECUTE FORMAT(+    $QUERY$+    SELECT %I.create_parent(+      p_parent_table := %L,+      p_control := %L,+      p_interval := %L,+      p_type := case+        when pgmq._get_pg_partman_major_version() = 5 then 'range'+        else 'native'+      end+    )+    $QUERY$,+    pgmq._get_pg_partman_schema(),+    fq_qtable,+    partition_col,+    partition_interval+  );++  EXECUTE FORMAT(+    $QUERY$+    CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (%I);+    $QUERY$,+    qtable || '_part_idx', qtable, partition_col+  );++  EXECUTE FORMAT(+    $QUERY$+    UPDATE %I.part_config+    SET+        retention = %L,+        retention_keep_table = false,+        retention_keep_index = true,+        automatic_maintenance = 'on'+    WHERE parent_table = %L;+    $QUERY$,+    pgmq._get_pg_partman_schema(),+    retention_interval,+    'pgmq.' || qtable+  );++  EXECUTE FORMAT(+    $QUERY$+    INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged)+    VALUES (%L, true, false)+    ON CONFLICT+    DO NOTHING;+    $QUERY$,+    queue_name+  );++  IF partition_col = 'enqueued_at' THEN+    a_partition_col := 'archived_at';+  ELSE+    a_partition_col := partition_col;+  END IF;++  EXECUTE FORMAT(+    $QUERY$+    CREATE TABLE IF NOT EXISTS pgmq.%I (+      msg_id BIGINT NOT NULL,+      read_ct INT DEFAULT 0 NOT NULL,+      enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+      last_read_at TIMESTAMP WITH TIME ZONE,+      archived_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+      vt TIMESTAMP WITH TIME ZONE NOT NULL,+      message JSONB,+      headers JSONB+    ) PARTITION BY RANGE (%I);+    $QUERY$,+    atable, a_partition_col+  );++  -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md+  -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema.+  EXECUTE FORMAT(+    $QUERY$+    SELECT %I.create_parent(+      p_parent_table := %L,+      p_control := %L,+      p_interval := %L,+      p_type := case+        when pgmq._get_pg_partman_major_version() = 5 then 'range'+        else 'native'+      end+    )+    $QUERY$,+    pgmq._get_pg_partman_schema(),+    fq_atable,+    a_partition_col,+    partition_interval+  );++  EXECUTE FORMAT(+    $QUERY$+    UPDATE %I.part_config+    SET+        retention = %L,+        retention_keep_table = false,+        retention_keep_index = true,+        automatic_maintenance = 'on'+    WHERE parent_table = %L;+    $QUERY$,+    pgmq._get_pg_partman_schema(),+    retention_interval,+    'pgmq.' || atable+  );++  EXECUTE FORMAT(+    $QUERY$+    CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (archived_at);+    $QUERY$,+    'archived_at_idx_' || queue_name, atable+  );++END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.create(queue_name TEXT)+RETURNS void AS $$+BEGIN+    PERFORM pgmq.create_non_partitioned(queue_name);+END;+$$ LANGUAGE plpgsql;++-- _create_fifo_index_if_not_exists+-- internal function to create GIN index on headers for better FIFO performance+CREATE OR REPLACE FUNCTION pgmq._create_fifo_index_if_not_exists(queue_name TEXT)+RETURNS void AS $$+DECLARE+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+    index_name TEXT := qtable || '_fifo_idx';+BEGIN+    -- Create GIN index on headers for efficient FIFO key lookups+    EXECUTE FORMAT(+        $QUERY$+        CREATE INDEX IF NOT EXISTS %I ON pgmq.%I USING GIN (headers);+        $QUERY$,+        index_name, qtable+    );+END;+$$ LANGUAGE plpgsql;++-- create_fifo_index+-- creates a GIN index on the headers column to improve FIFO read performance+CREATE FUNCTION pgmq.create_fifo_index(queue_name TEXT)+RETURNS void AS $$+BEGIN+    PERFORM pgmq._create_fifo_index_if_not_exists(queue_name);+END;+$$ LANGUAGE plpgsql;++-- create_fifo_indexes_all+-- creates FIFO indexes on all existing queues+CREATE FUNCTION pgmq.create_fifo_indexes_all()+RETURNS void AS $$+DECLARE+    queue_record RECORD;+BEGIN+    FOR queue_record IN SELECT queue_name FROM pgmq.meta LOOP+        PERFORM pgmq.create_fifo_index(queue_record.queue_name);+    END LOOP;+END;+$$ LANGUAGE plpgsql;++CREATE OR REPLACE FUNCTION pgmq.convert_archive_partitioned(+  table_name TEXT,+  partition_interval TEXT DEFAULT '10000',+  retention_interval TEXT DEFAULT '100000',+  leading_partition INT DEFAULT 10+)+RETURNS void AS $$+DECLARE+  a_table_name TEXT := pgmq.format_table_name(table_name, 'a');+  a_table_name_old TEXT := pgmq.format_table_name(table_name, 'a') || '_old';+  qualified_a_table_name TEXT := format('pgmq.%I', a_table_name);+  partition_col TEXT;+  a_partition_col TEXT;+BEGIN++  PERFORM c.relkind+    FROM pg_class c+    JOIN pg_namespace n ON n.oid = c.relnamespace+    WHERE c.relname = a_table_name+    AND c.relkind = 'p';++  IF FOUND THEN+    RAISE NOTICE 'Table %s is already partitioned', a_table_name;+    RETURN;+  END IF;++  PERFORM c.relkind+    FROM pg_class c+    JOIN pg_namespace n ON n.oid = c.relnamespace+    WHERE c.relname = a_table_name+    AND c.relkind = 'r';++  IF NOT FOUND THEN+    RAISE NOTICE 'Table %s does not exists', a_table_name;+    RETURN;+  END IF;++  SELECT pgmq._get_partition_col(partition_interval) INTO partition_col;++  -- For archive tables, use archived_at for time-based partitioning+  IF partition_col = 'enqueued_at' THEN+    a_partition_col := 'archived_at';+  ELSE+    a_partition_col := partition_col;+  END IF;++  EXECUTE 'ALTER TABLE ' || qualified_a_table_name || ' RENAME TO ' || a_table_name_old;++  -- When partitioning by time (archived_at), we need to exclude constraints and indexes+  -- because the existing PRIMARY KEY on msg_id alone is incompatible with partitioning by archived_at.+  -- When partitioning by msg_id, we can keep all constraints including PRIMARY KEY.+  IF a_partition_col = 'archived_at' THEN+    EXECUTE format( 'CREATE TABLE pgmq.%I (LIKE pgmq.%I including defaults including generated including storage including comments) PARTITION BY RANGE (%I)', a_table_name, a_table_name_old, a_partition_col );+  ELSE+    EXECUTE format( 'CREATE TABLE pgmq.%I (LIKE pgmq.%I including all) PARTITION BY RANGE (%I)', a_table_name, a_table_name_old, a_partition_col );+  END IF;++  EXECUTE 'ALTER INDEX pgmq.archived_at_idx_' || table_name || ' RENAME TO archived_at_idx_' || table_name || '_old';+  EXECUTE 'CREATE INDEX archived_at_idx_'|| table_name || ' ON ' || qualified_a_table_name ||'(archived_at)';++  -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md+  -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema.+  EXECUTE FORMAT(+    $QUERY$+    SELECT %I.create_parent(+      p_parent_table := %L,+      p_control := %L,+      p_interval := %L,+      p_type := case+        when pgmq._get_pg_partman_major_version() = 5 then 'range'+        else 'native'+      end+    )+    $QUERY$,+    pgmq._get_pg_partman_schema(),+    qualified_a_table_name,+    a_partition_col,+    partition_interval+  );++  EXECUTE FORMAT(+    $QUERY$+    UPDATE %I.part_config+    SET+      retention = %L,+      retention_keep_table = false,+      retention_keep_index = false,+      infinite_time_partitions = true+    WHERE+      parent_table = %L;+    $QUERY$,+    pgmq._get_pg_partman_schema(),+    retention_interval,+    qualified_a_table_name+  );+END;+$$ LANGUAGE plpgsql;++CREATE OR REPLACE FUNCTION pgmq.notify_queue_listeners()+RETURNS TRIGGER AS $$+DECLARE+  queue_name_extracted TEXT; -- Queue name extracted from trigger table name+  updated_count        INTEGER; -- Number of rows updated (0 or 1)+BEGIN+  queue_name_extracted := substring(TG_TABLE_NAME from 3);++  UPDATE pgmq.notify_insert_throttle+  SET last_notified_at = clock_timestamp()+  WHERE queue_name = queue_name_extracted+    AND (+      throttle_interval_ms = 0 -- No throttling configured+          OR clock_timestamp() - last_notified_at >=+             (throttle_interval_ms * INTERVAL '1 millisecond') -- Throttle interval has elapsed+    );++  -- Check how many rows were updated (will be 0 or 1)+  GET DIAGNOSTICS updated_count = ROW_COUNT;++  IF updated_count > 0 THEN+    PERFORM PG_NOTIFY('pgmq.' || TG_TABLE_NAME || '.' || TG_OP, NULL);+  END IF;++RETURN NEW;+END;+$$ LANGUAGE plpgsql;++CREATE OR REPLACE FUNCTION pgmq.enable_notify_insert(queue_name TEXT, throttle_interval_ms INTEGER DEFAULT 250)+RETURNS void AS $$+DECLARE+  qtable TEXT := pgmq.format_table_name(queue_name, 'q');+  v_queue_name TEXT := queue_name;+  v_throttle_interval_ms INTEGER := throttle_interval_ms;+BEGIN+  -- Validate that throttle_interval_ms is non-negative+  IF v_throttle_interval_ms < 0 THEN+    RAISE EXCEPTION 'throttle_interval_ms must be non-negative';+  END IF;++  -- Validate that the queue table exists+  IF NOT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'pgmq' AND table_name = qtable) THEN+    RAISE EXCEPTION 'Queue "%" does not exist. Create it first using pgmq.create()', v_queue_name;+  END IF;++  PERFORM pgmq.disable_notify_insert(v_queue_name);++  INSERT INTO pgmq.notify_insert_throttle (queue_name, throttle_interval_ms)+  VALUES (v_queue_name, v_throttle_interval_ms)+  ON CONFLICT ON CONSTRAINT notify_insert_throttle_queue_name_key DO UPDATE+      SET throttle_interval_ms = EXCLUDED.throttle_interval_ms,+          last_notified_at = to_timestamp(0);++  EXECUTE FORMAT(+    $QUERY$+    CREATE CONSTRAINT TRIGGER trigger_notify_queue_insert_listeners+    AFTER INSERT ON pgmq.%I+    DEFERRABLE FOR EACH ROW+    EXECUTE PROCEDURE pgmq.notify_queue_listeners()+    $QUERY$,+    qtable+  );+END;+$$ LANGUAGE plpgsql;++CREATE OR REPLACE FUNCTION pgmq.disable_notify_insert(queue_name TEXT)+RETURNS void AS $$+DECLARE+  qtable TEXT := pgmq.format_table_name(queue_name, 'q');+  v_queue_name TEXT := queue_name;+BEGIN+  EXECUTE FORMAT(+    $QUERY$+    DROP TRIGGER IF EXISTS trigger_notify_queue_insert_listeners ON pgmq.%I;+    $QUERY$,+    qtable+  );++  DELETE FROM pgmq.notify_insert_throttle nit WHERE nit.queue_name = v_queue_name;+END;+$$ LANGUAGE plpgsql;++CREATE OR REPLACE FUNCTION pgmq.list_notify_insert_throttles()+    RETURNS TABLE+            (+                queue_name           text,+                throttle_interval_ms integer,+                last_notified_at     TIMESTAMP WITH TIME ZONE+            )+    LANGUAGE sql+    STABLE+AS+$$+    SELECT queue_name, throttle_interval_ms, last_notified_at+    FROM pgmq.notify_insert_throttle+    ORDER BY queue_name;+$$;++CREATE OR REPLACE FUNCTION pgmq.update_notify_insert(queue_name text, throttle_interval_ms integer)+    RETURNS void+    LANGUAGE plpgsql+AS+$$+BEGIN+    IF throttle_interval_ms < 0 THEN+        RAISE EXCEPTION 'throttle_interval_ms must be non-negative, got: %', throttle_interval_ms;+    END IF;++    IF NOT EXISTS (SELECT 1 FROM pgmq.meta WHERE meta.queue_name = update_notify_insert.queue_name) THEN+        RAISE EXCEPTION 'Queue "%" does not exist. Create the queue first using pgmq.create()', queue_name;+    END IF;++    IF NOT EXISTS (SELECT 1 FROM pgmq.notify_insert_throttle WHERE notify_insert_throttle.queue_name = update_notify_insert.queue_name) THEN+        RAISE EXCEPTION 'Queue "%" does not have notify_insert enabled. Enable it first using pgmq.enable_notify_insert()', queue_name;+    END IF;++    UPDATE pgmq.notify_insert_throttle+    SET throttle_interval_ms = update_notify_insert.throttle_interval_ms,+        last_notified_at = to_timestamp(0)+    WHERE notify_insert_throttle.queue_name = update_notify_insert.queue_name;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.validate_routing_key(routing_key text)+    RETURNS boolean+    LANGUAGE plpgsql+    IMMUTABLE+AS+$$+BEGIN+    -- Valid routing key examples:+    --   "logs.error"+    --   "app.user-service.auth"+    --   "system_events.db.connection_failed"+    --+    -- Invalid routing key examples:+    --   ""                     - empty+    --   ".logs.error"          - starts with dot+    --   "logs.error."          - ends with dot+    --   "logs..error"          - consecutive dots+    --   "logs.error!"          - invalid character+    --   "logs error"           - space not allowed+    --   "logs.*"               - wildcards not allowed in routing keys++    IF routing_key IS NULL OR routing_key = '' THEN+        RAISE EXCEPTION 'routing_key cannot be NULL or empty';+    END IF;++    IF length(routing_key) > 255 THEN+        RAISE EXCEPTION 'routing_key length cannot exceed 255 characters, got % characters', length(routing_key);+    END IF;++    IF routing_key !~ '^[a-zA-Z0-9._-]+$' THEN+        RAISE EXCEPTION 'routing_key contains invalid characters. Only alphanumeric, dots, hyphens, and underscores are allowed. Got: %', routing_key;+    END IF;++    IF routing_key ~ '^\.' THEN+        RAISE EXCEPTION 'routing_key cannot start with a dot. Got: %', routing_key;+    END IF;++    IF routing_key ~ '\.$' THEN+        RAISE EXCEPTION 'routing_key cannot end with a dot. Got: %', routing_key;+    END IF;++    IF routing_key ~ '\.\.' THEN+        RAISE EXCEPTION 'routing_key cannot contain consecutive dots. Got: %', routing_key;+    END IF;++    RETURN true;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.validate_topic_pattern(pattern text)+    RETURNS boolean+    LANGUAGE plpgsql+    IMMUTABLE+AS+$$+BEGIN+    -- Valid pattern examples:+    --   "logs.*"           - matches one segment after logs. (e.g., logs.error, logs.info)+    --   "logs.#"           - matches one or more segments after logs. (e.g., logs.error, logs.api.error)+    --   "*.error"          - matches one segment before .error (e.g., app.error, db.error)+    --   "#.error"          - matches one or more segments before .error (e.g., app.error, x.y.error)+    --   "app.*.#"          - mixed wildcards (one segment then one or more)+    --   "#"                - catch-all pattern, matches any routing key+    --+    -- Invalid pattern examples:+    --   ".logs.*"          - starts with dot+    --   "logs.*."          - ends with dot+    --   "logs..error"      - consecutive dots+    --   "logs.**"          - consecutive stars+    --   "logs.##"          - consecutive hashes+    --   "logs.*#"          - adjacent wildcards+    --   "logs.error!"      - invalid character++    IF pattern IS NULL OR pattern = '' THEN+        RAISE EXCEPTION 'pattern cannot be NULL or empty';+    END IF;++    IF length(pattern) > 255 THEN+        RAISE EXCEPTION 'pattern length cannot exceed 255 characters, got % characters', length(pattern);+    END IF;++    IF pattern !~ '^[a-zA-Z0-9._\-*#]+$' THEN+        RAISE EXCEPTION 'pattern contains invalid characters. Only alphanumeric, dots, hyphens, underscores, *, and # are allowed. Got: %', pattern;+    END IF;++    IF pattern ~ '^\.' THEN+        RAISE EXCEPTION 'pattern cannot start with a dot. Got: %', pattern;+    END IF;++    IF pattern ~ '\.$' THEN+        RAISE EXCEPTION 'pattern cannot end with a dot. Got: %', pattern;+    END IF;++    IF pattern ~ '\.\.' THEN+        RAISE EXCEPTION 'pattern cannot contain consecutive dots. Got: %', pattern;+    END IF;++    IF pattern ~ '\*\*' THEN+        RAISE EXCEPTION 'pattern cannot contain consecutive stars (**). Use # for multi-segment matching. Got: %', pattern;+    END IF;++    IF pattern ~ '##' THEN+        RAISE EXCEPTION 'pattern cannot contain consecutive hashes (##). A single # already matches zero or more segments. Got: %', pattern;+    END IF;++    IF pattern ~ '\*#' OR pattern ~ '#\*' THEN+        RAISE EXCEPTION 'pattern cannot contain adjacent wildcards (*# or #*). Separate wildcards with dots. Got: %', pattern;+    END IF;++    RETURN true;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.bind_topic(pattern text, queue_name text)+    RETURNS void+    LANGUAGE plpgsql+AS+$$+BEGIN+    PERFORM pgmq.validate_topic_pattern(pattern);+    IF queue_name IS NULL OR queue_name = '' THEN+        RAISE EXCEPTION 'queue_name cannot be NULL or empty';+    END IF;++    IF NOT EXISTS (SELECT 1 FROM pgmq.meta WHERE meta.queue_name = bind_topic.queue_name) THEN+        RAISE EXCEPTION 'Queue "%" does not exist. Create the queue first using pgmq.create()', queue_name;+    END IF;++    INSERT INTO pgmq.topic_bindings (pattern, queue_name)+    VALUES (pattern, queue_name)+    ON CONFLICT ON CONSTRAINT topic_bindings_unique_pattern_queue DO NOTHING;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.unbind_topic(pattern text, queue_name text)+    RETURNS boolean+    LANGUAGE plpgsql+AS+$$+DECLARE+    rows_deleted integer;+BEGIN+    IF pattern IS NULL OR pattern = '' THEN+        RAISE EXCEPTION 'pattern cannot be NULL or empty';+    END IF;++    IF queue_name IS NULL OR queue_name = '' THEN+        RAISE EXCEPTION 'queue_name cannot be NULL or empty';+    END IF;++    DELETE+    FROM pgmq.topic_bindings+    WHERE topic_bindings.pattern = unbind_topic.pattern+      AND topic_bindings.queue_name = unbind_topic.queue_name;++    GET DIAGNOSTICS rows_deleted = ROW_COUNT;++    IF rows_deleted > 0 THEN+        RETURN true;+    ELSE+        RETURN false;+    END IF;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.test_routing(routing_key text)+    RETURNS TABLE+            (+                pattern        text,+                queue_name     text,+                compiled_regex text+            )+    LANGUAGE plpgsql+    STABLE+AS+$$+BEGIN+    PERFORM pgmq.validate_routing_key(routing_key);+    RETURN QUERY+        SELECT b.pattern,+               b.queue_name,+               b.compiled_regex+        FROM pgmq.topic_bindings b+        WHERE routing_key ~ b.compiled_regex+        ORDER BY b.pattern;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.send_topic(routing_key text, msg jsonb, headers jsonb, delay integer)+    RETURNS integer+    LANGUAGE plpgsql+    VOLATILE+AS+$$+DECLARE+    b             RECORD;+    matched_count integer := 0;+BEGIN+    PERFORM pgmq.validate_routing_key(routing_key);++    IF msg IS NULL THEN+        RAISE EXCEPTION 'msg cannot be NULL';+    END IF;++    IF delay < 0 THEN+        RAISE EXCEPTION 'delay cannot be negative, got: %', delay;+    END IF;++    -- Filter matching patterns in SQL for better performance (uses index)+    -- Any failure will rollback the entire transaction+    FOR b IN+        SELECT DISTINCT tb.queue_name+        FROM pgmq.topic_bindings tb+        WHERE routing_key ~ tb.compiled_regex+        ORDER BY tb.queue_name -- Deterministic ordering, deduplicated by queue_name+        LOOP+            PERFORM pgmq.send(b.queue_name, msg, headers, delay);+            matched_count := matched_count + 1;+        END LOOP;++    RETURN matched_count;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.send_topic(routing_key text, msg jsonb)+    RETURNS integer+    LANGUAGE plpgsql+    VOLATILE+AS+$$+BEGIN+    RETURN pgmq.send_topic(routing_key, msg, NULL, 0);+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.send_topic(routing_key text, msg jsonb, delay integer)+    RETURNS integer+    LANGUAGE plpgsql+    VOLATILE+AS+$$+BEGIN+    RETURN pgmq.send_topic(routing_key, msg, NULL, delay);+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.list_topic_bindings()+    RETURNS TABLE+            (+                pattern        text,+                queue_name     text,+                bound_at       TIMESTAMP WITH TIME ZONE,+                compiled_regex text+            )+    LANGUAGE sql+    STABLE+AS+$$+    SELECT pattern, queue_name, bound_at, compiled_regex+    FROM pgmq.topic_bindings+    ORDER BY bound_at DESC, pattern, queue_name;+$$;++CREATE OR REPLACE FUNCTION pgmq.list_topic_bindings(queue_name text)+    RETURNS TABLE+            (+                pattern        text,+                queue_name     text,+                bound_at       TIMESTAMP WITH TIME ZONE,+                compiled_regex text+            )+    LANGUAGE sql+    STABLE+AS+$$+    SELECT pattern, tb.queue_name, bound_at, compiled_regex+    FROM pgmq.topic_bindings tb+    WHERE tb.queue_name = list_topic_bindings.queue_name+    ORDER BY bound_at DESC, pattern;+$$;++-- send_batch_topic: Base implementation with TIMESTAMP WITH TIME ZONE delay+CREATE OR REPLACE FUNCTION pgmq.send_batch_topic(+    routing_key text,+    msgs jsonb[],+    headers jsonb[],+    delay TIMESTAMP WITH TIME ZONE+)+    RETURNS TABLE(queue_name text, msg_id bigint)+    LANGUAGE plpgsql+    VOLATILE+AS+$$+DECLARE+    b RECORD;+BEGIN+    PERFORM pgmq.validate_routing_key(routing_key);++    -- Validate batch parameters once (not per queue)+    PERFORM pgmq._validate_batch_params(msgs, headers);++    -- Filter matching patterns in SQL for better performance (uses index)+    -- Any failure will rollback the entire transaction+    FOR b IN+        SELECT DISTINCT tb.queue_name+        FROM pgmq.topic_bindings tb+        WHERE routing_key ~ tb.compiled_regex+        ORDER BY tb.queue_name -- Deterministic ordering, deduplicated by queue_name+        LOOP+            -- Use private _send_batch to avoid redundant validation+            RETURN QUERY+            SELECT b.queue_name, batch_result.msg_id+            FROM pgmq._send_batch(b.queue_name, msgs, headers, delay) AS batch_result(msg_id);+        END LOOP;++    RETURN;+END;+$$;++-- send_batch_topic: 2 args (routing_key, msgs)+CREATE OR REPLACE FUNCTION pgmq.send_batch_topic(+    routing_key text,+    msgs jsonb[]+)+    RETURNS TABLE(queue_name text, msg_id bigint)+    LANGUAGE sql+    VOLATILE+AS+$$+    SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, NULL, clock_timestamp());+$$;++-- send_batch_topic: 3 args with headers+CREATE OR REPLACE FUNCTION pgmq.send_batch_topic(+    routing_key text,+    msgs jsonb[],+    headers jsonb[]+)+    RETURNS TABLE(queue_name text, msg_id bigint)+    LANGUAGE sql+    VOLATILE+AS+$$+    SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, headers, clock_timestamp());+$$;++-- send_batch_topic: 3 args with integer delay+CREATE OR REPLACE FUNCTION pgmq.send_batch_topic(+    routing_key text,+    msgs jsonb[],+    delay integer+)+    RETURNS TABLE(queue_name text, msg_id bigint)+    LANGUAGE sql+    VOLATILE+AS+$$+    SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, NULL, clock_timestamp() + make_interval(secs => delay));+$$;++-- send_batch_topic: 3 args with timestamp delay+CREATE OR REPLACE FUNCTION pgmq.send_batch_topic(+    routing_key text,+    msgs jsonb[],+    delay TIMESTAMP WITH TIME ZONE+)+    RETURNS TABLE(queue_name text, msg_id bigint)+    LANGUAGE sql+    VOLATILE+AS+$$+    SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, NULL, delay);+$$;++-- send_batch_topic: 4 args with integer delay+CREATE OR REPLACE FUNCTION pgmq.send_batch_topic(+    routing_key text,+    msgs jsonb[],+    headers jsonb[],+    delay integer+)+    RETURNS TABLE(queue_name text, msg_id bigint)+    LANGUAGE sql+    VOLATILE+AS+$$+    SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, headers, clock_timestamp() + make_interval(secs => delay));+$$;
+ test/golden/descriptor.json view
@@ -0,0 +1,1 @@+{"openapi_path":"openapi.json","service":"pgmq-inspect","version":"0.1.0.0","websocket_path":"ws"}
+ test/golden/error-invalid-cursor.json view
@@ -0,0 +1,1 @@+{"error":{"code":"invalid_cursor","details":{"from":"x"},"message":"from must be a non-negative integer msg_id."}}
+ test/golden/error-plain.json view
@@ -0,0 +1,1 @@+{"error":{"code":"not_found","message":"No route matches this path."}}
+ test/golden/health-live.json view
@@ -0,0 +1,1 @@+{"alive":true}
+ test/golden/health-ready.json view
@@ -0,0 +1,1 @@+{"checks":[{"error":"timed out","healthy":false,"latency_ms":12,"name":"postgres"}],"ready":false}
+ test/golden/openapi.json view
@@ -0,0 +1,1 @@+{"info":{"description":"Read-only inspection of pgmq queues. Every path is relative to wherever the application is mounted. Field names are pgmq's own SQL column names in snake_case; once published they are never removed or re-typed. The WebSocket feed at ws is described in docs/user/queue-inspection.md, not here.","title":"pgmq-inspect","version":"0.7.0.0"},"servers":[{"url":"."}],"paths":{"/":{"get":{"summary":"Service descriptor","operationId":"serviceDescriptor","responses":{"200":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ServiceDescriptor"}}},"description":"OK"},"405":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: method_not_allowed"},"500":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_error"},"503":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_unavailable, inspection_busy, inspection_timeout"}}}},"/queues":{"get":{"summary":"List every queue, including names this library would reject","operationId":"listQueues","responses":{"200":{"content":{"application/json":{"schema":{"items":{"$ref":"#/components/schemas/Queue"},"type":"array"}}},"description":"OK"},"405":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: method_not_allowed"},"500":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_error"},"503":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_unavailable, inspection_busy, inspection_timeout"}}}},"/queues/{queue}":{"get":{"summary":"One queue","operationId":"getQueue","parameters":[{"description":"Queue name as stored in pgmq.meta; any server-accepted name, not only names this library validates","in":"path","name":"queue","required":true,"schema":{"type":"string"}}],"responses":{"200":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/Queue"}}},"description":"OK"},"404":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: queue_not_found"},"405":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: method_not_allowed"},"500":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_error"},"503":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_unavailable, inspection_busy, inspection_timeout"}}}},"/queues/{queue}/metrics":{"get":{"summary":"Metrics for one queue","operationId":"getQueueMetrics","parameters":[{"description":"Queue name as stored in pgmq.meta; any server-accepted name, not only names this library validates","in":"path","name":"queue","required":true,"schema":{"type":"string"}}],"responses":{"200":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/QueueMetrics"}}},"description":"OK"},"404":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: queue_not_found"},"405":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: method_not_allowed"},"500":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_error"},"503":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_unavailable, inspection_busy, inspection_timeout"}}}},"/metrics":{"get":{"summary":"Metrics for every queue; scans rows across queues, subject to the inspection budget","operationId":"getAllMetrics","responses":{"200":{"content":{"application/json":{"schema":{"items":{"$ref":"#/components/schemas/QueueMetrics"},"type":"array"}}},"description":"OK"},"405":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: method_not_allowed"},"500":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_error"},"503":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_unavailable, inspection_busy, inspection_timeout"}}}},"/queues/{queue}/messages":{"get":{"summary":"Non-destructive keyset page of a queue","operationId":"peekMessages","parameters":[{"description":"Queue name as stored in pgmq.meta; any server-accepted name, not only names this library validates","in":"path","name":"queue","required":true,"schema":{"type":"string"}},{"description":"Exclusive msg_id cursor, 0..9223372036854775807, at most 128 decimal digits. Echo next_cursor here; moving-view pages are not a transaction snapshot.","in":"query","name":"from","required":false,"schema":{"format":"int64","maximum":9223372036854775807,"minimum":0,"type":"integer"}},{"description":"Positive decimal page size (at most 128 digits), capped before narrowing; default 50, cap 500, both host-configurable. Larger values are accepted and capped","in":"query","name":"limit","required":false,"schema":{"minimum":1,"type":"integer"}}],"responses":{"200":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/MessagePage"}}},"description":"OK"},"400":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: invalid_cursor, invalid_limit"},"404":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: queue_not_found"},"405":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: method_not_allowed"},"500":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_error"},"503":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_unavailable, inspection_busy, inspection_timeout"}}}},"/queues/{queue}/messages/{msg_id}":{"get":{"summary":"One message by id","operationId":"lookupMessage","parameters":[{"description":"Queue name as stored in pgmq.meta; any server-accepted name, not only names this library validates","in":"path","name":"queue","required":true,"schema":{"type":"string"}},{"description":"Non-negative int64 ID, at most 128 decimal digits; browser clients must preserve exact integer tokens","in":"path","name":"msg_id","required":true,"schema":{"format":"int64","maximum":9223372036854775807,"minimum":0,"type":"integer"}}],"responses":{"200":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/Message"}}},"description":"OK"},"400":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: invalid_message_id"},"404":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: queue_not_found, message_not_found"},"405":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: method_not_allowed"},"500":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_error"},"503":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_unavailable, inspection_busy, inspection_timeout"}}}},"/queues/{queue}/archive":{"get":{"summary":"Keyset page of a queue's archive","operationId":"peekArchive","parameters":[{"description":"Queue name as stored in pgmq.meta; any server-accepted name, not only names this library validates","in":"path","name":"queue","required":true,"schema":{"type":"string"}},{"description":"Exclusive msg_id cursor, 0..9223372036854775807, at most 128 decimal digits. Echo next_cursor here; moving-view pages are not a transaction snapshot.","in":"query","name":"from","required":false,"schema":{"format":"int64","maximum":9223372036854775807,"minimum":0,"type":"integer"}},{"description":"Positive decimal page size (at most 128 digits), capped before narrowing; default 50, cap 500, both host-configurable. Larger values are accepted and capped","in":"query","name":"limit","required":false,"schema":{"minimum":1,"type":"integer"}}],"responses":{"200":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ArchivedMessagePage"}}},"description":"OK"},"400":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: invalid_cursor, invalid_limit"},"404":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: queue_not_found"},"405":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: method_not_allowed"},"500":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_error"},"503":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_unavailable, inspection_busy, inspection_timeout"}}}},"/queues/{queue}/archive/{msg_id}":{"get":{"summary":"One archived message by id","operationId":"lookupArchivedMessage","parameters":[{"description":"Queue name as stored in pgmq.meta; any server-accepted name, not only names this library validates","in":"path","name":"queue","required":true,"schema":{"type":"string"}},{"description":"Non-negative int64 ID, at most 128 decimal digits; browser clients must preserve exact integer tokens","in":"path","name":"msg_id","required":true,"schema":{"format":"int64","maximum":9223372036854775807,"minimum":0,"type":"integer"}}],"responses":{"200":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ArchivedMessage"}}},"description":"OK"},"400":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: invalid_message_id"},"404":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: queue_not_found, message_not_found"},"405":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: method_not_allowed"},"500":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_error"},"503":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_unavailable, inspection_busy, inspection_timeout"}}}},"/queues/{queue}/bindings":{"get":{"summary":"Topic bindings of one queue","operationId":"listQueueBindings","parameters":[{"description":"Queue name as stored in pgmq.meta; any server-accepted name, not only names this library validates","in":"path","name":"queue","required":true,"schema":{"type":"string"}}],"responses":{"200":{"content":{"application/json":{"schema":{"items":{"$ref":"#/components/schemas/TopicBinding"},"type":"array"}}},"description":"OK"},"404":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: queue_not_found"},"405":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: method_not_allowed"},"500":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_error"},"503":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_unavailable, inspection_busy, inspection_timeout"}}}},"/bindings":{"get":{"summary":"Every topic binding","operationId":"listBindings","responses":{"200":{"content":{"application/json":{"schema":{"items":{"$ref":"#/components/schemas/TopicBinding"},"type":"array"}}},"description":"OK"},"405":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: method_not_allowed"},"500":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_error"},"503":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_unavailable, inspection_busy, inspection_timeout"}}}},"/routing/test":{"get":{"summary":"Queues a routing key would reach","operationId":"testRouting","parameters":[{"in":"query","name":"routing_key","required":true,"schema":{"type":"string"}}],"responses":{"200":{"content":{"application/json":{"schema":{"items":{"$ref":"#/components/schemas/RoutingMatch"},"type":"array"}}},"description":"OK"},"400":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: invalid_routing_key"},"405":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: method_not_allowed"},"500":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_error"},"503":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_unavailable, inspection_busy, inspection_timeout"}}}},"/notify/throttles":{"get":{"summary":"Insert-notification throttle rows","operationId":"listNotifyThrottles","responses":{"200":{"content":{"application/json":{"schema":{"items":{"$ref":"#/components/schemas/NotifyInsertThrottle"},"type":"array"}}},"description":"OK"},"405":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: method_not_allowed"},"500":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_error"},"503":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_unavailable, inspection_busy, inspection_timeout"}}}},"/health/live":{"get":{"summary":"Process liveness","operationId":"liveness","responses":{"200":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/Liveness"}}},"description":"OK"},"405":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: method_not_allowed"},"500":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_error"},"503":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_unavailable, inspection_busy, inspection_timeout"}}}},"/health/ready":{"get":{"summary":"Database readiness","operationId":"readiness","responses":{"200":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/Readiness"}}},"description":"OK"},"405":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: method_not_allowed"},"500":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_error"},"503":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/Readiness"}}},"description":"OK"}}}},"/openapi.json":{"get":{"summary":"This document","operationId":"openApi","responses":{"200":{"content":{"application/json":{"schema":{"additionalProperties":true,"type":"object"}}},"description":"The OpenAPI 3 document"},"405":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: method_not_allowed"},"500":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_error"},"503":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/ErrorEnvelope"}}},"description":"Error envelope; code is one of: database_unavailable, inspection_busy, inspection_timeout"}}}}},"components":{"schemas":{"Queue":{"additionalProperties":true,"properties":{"created_at":{"format":"date-time","type":"string"},"is_partitioned":{"type":"boolean"},"is_unlogged":{"type":"boolean"},"queue_name":{"type":"string"}},"required":["queue_name","is_partitioned","is_unlogged","created_at"],"type":"object"},"Message":{"additionalProperties":true,"properties":{"enqueued_at":{"format":"date-time","type":"string"},"headers":{"additionalProperties":true,"description":"Any JSON value, including null; absent and JSON-null optional headers encode as null"},"last_read_at":{"format":"date-time","nullable":true,"type":"string"},"message":{"additionalProperties":true,"description":"Any JSON value; a SQL NULL body is JSON null"},"msg_id":{"format":"int64","maximum":9223372036854775807,"minimum":0,"type":"integer"},"read_ct":{"format":"int64","maximum":9223372036854775807,"minimum":0,"type":"integer"},"vt":{"format":"date-time","type":"string"}},"required":["msg_id","read_ct","enqueued_at","last_read_at","vt","message","headers"],"type":"object"},"ArchivedMessage":{"additionalProperties":true,"properties":{"archived_at":{"format":"date-time","type":"string"},"enqueued_at":{"format":"date-time","type":"string"},"headers":{"additionalProperties":true,"description":"Any JSON value, including null; absent and JSON-null optional headers encode as null"},"last_read_at":{"format":"date-time","nullable":true,"type":"string"},"message":{"additionalProperties":true,"description":"Any JSON value; a SQL NULL body is JSON null"},"msg_id":{"format":"int64","maximum":9223372036854775807,"minimum":0,"type":"integer"},"read_ct":{"format":"int64","maximum":9223372036854775807,"minimum":0,"type":"integer"},"vt":{"format":"date-time","type":"string"}},"required":["msg_id","read_ct","enqueued_at","last_read_at","vt","message","headers","archived_at"],"type":"object"},"QueueMetrics":{"additionalProperties":true,"properties":{"default_partition_length":{"description":"PGMQ 1.13 planner estimate of rows in the queue and archive default partitions; null on 1.12 and for ordinary queues","format":"int64","maximum":9223372036854775807,"minimum":0,"nullable":true,"type":"integer"},"newest_msg_age_sec":{"format":"int32","nullable":true,"type":"integer"},"oldest_msg_age_sec":{"format":"int32","nullable":true,"type":"integer"},"queue_length":{"description":"Every message in the queue table, visible or not","format":"int64","maximum":9223372036854775807,"minimum":0,"type":"integer"},"queue_name":{"type":"string"},"queue_visible_length":{"description":"Messages whose visibility timeout has expired, i.e. available to consumers now","format":"int64","maximum":9223372036854775807,"minimum":0,"type":"integer"},"scrape_time":{"format":"date-time","type":"string"},"total_messages":{"format":"int64","maximum":9223372036854775807,"minimum":0,"type":"integer"}},"required":["queue_name","queue_length","queue_visible_length","newest_msg_age_sec","oldest_msg_age_sec","total_messages","scrape_time","default_partition_length"],"type":"object"},"TopicBinding":{"additionalProperties":true,"properties":{"bound_at":{"format":"date-time","type":"string"},"compiled_regex":{"type":"string"},"pattern":{"type":"string"},"queue_name":{"type":"string"}},"required":["pattern","queue_name","bound_at","compiled_regex"],"type":"object"},"RoutingMatch":{"additionalProperties":true,"properties":{"compiled_regex":{"type":"string"},"pattern":{"type":"string"},"queue_name":{"type":"string"}},"required":["pattern","queue_name","compiled_regex"],"type":"object"},"NotifyInsertThrottle":{"additionalProperties":true,"properties":{"last_notified_at":{"format":"date-time","type":"string"},"queue_name":{"type":"string"},"throttle_interval_ms":{"format":"int32","type":"integer"}},"required":["queue_name","throttle_interval_ms","last_notified_at"],"type":"object"},"MessagePage":{"additionalProperties":true,"properties":{"items":{"items":{"$ref":"#/components/schemas/Message"},"type":"array"},"next_cursor":{"description":"msg_id of the last item; absent on the last page","format":"int64","maximum":9223372036854775807,"minimum":0,"type":"integer"}},"required":["items"],"type":"object"},"ArchivedMessagePage":{"additionalProperties":true,"properties":{"items":{"items":{"$ref":"#/components/schemas/ArchivedMessage"},"type":"array"},"next_cursor":{"description":"msg_id of the last item; absent on the last page","format":"int64","maximum":9223372036854775807,"minimum":0,"type":"integer"}},"required":["items"],"type":"object"},"ErrorBody":{"additionalProperties":true,"properties":{"code":{"type":"string"},"details":{"additionalProperties":true,"type":"object"},"message":{"type":"string"}},"required":["code","message"],"type":"object"},"ErrorEnvelope":{"additionalProperties":true,"properties":{"error":{"$ref":"#/components/schemas/ErrorBody"}},"required":["error"],"type":"object"},"ServiceDescriptor":{"additionalProperties":true,"properties":{"openapi_path":{"type":"string"},"service":{"type":"string"},"version":{"type":"string"},"websocket_path":{"type":"string"}},"required":["service","version","websocket_path","openapi_path"],"type":"object"},"Liveness":{"additionalProperties":true,"properties":{"alive":{"type":"boolean"}},"required":["alive"],"type":"object"},"Readiness":{"additionalProperties":true,"properties":{"checks":{"items":{"$ref":"#/components/schemas/DependencyCheck"},"type":"array"},"ready":{"type":"boolean"}},"required":["ready","checks"],"type":"object"},"DependencyCheck":{"additionalProperties":true,"properties":{"error":{"nullable":true,"type":"string"},"healthy":{"type":"boolean"},"latency_ms":{"format":"int64","maximum":9223372036854775807,"minimum":0,"nullable":true,"type":"integer"},"name":{"type":"string"}},"required":["name","healthy","latency_ms","error"],"type":"object"}}},"openapi":"3.0.0"}
+ test/golden/page-last.json view
@@ -0,0 +1,1 @@+{"items":[4]}
+ test/golden/page-more.json view
@@ -0,0 +1,1 @@+{"items":[1,2,3],"next_cursor":3}
+ test/golden/ws-error.json view
@@ -0,0 +1,1 @@+{"type":"error","code":"overflow","message":"The server dropped frames because this connection was not reading them; re-subscribe or re-read the queue over HTTP.","queue":null}
+ test/golden/ws-goodbye.json view
@@ -0,0 +1,1 @@+{"type":"goodbye"}
+ test/golden/ws-ping.json view
@@ -0,0 +1,1 @@+{"type":"ping"}
+ test/golden/ws-pong.json view
@@ -0,0 +1,1 @@+{"type":"pong"}
+ test/golden/ws-snapshot.json view
@@ -0,0 +1,1 @@+{"type":"snapshot","queue":"orders","metrics":{"queue_name":"orders","queue_length":3,"newest_msg_age_sec":1,"oldest_msg_age_sec":9,"total_messages":12,"scrape_time":"2026-09-30T12:00:00Z","queue_visible_length":2,"default_partition_length":null},"push":true}
+ test/golden/ws-subscribe.json view
@@ -0,0 +1,1 @@+{"type":"subscribe","queue":"orders"}
+ test/golden/ws-unsubscribe.json view
@@ -0,0 +1,1 @@+{"type":"unsubscribe","queue":"orders"}
+ test/golden/ws-update-notify.json view
@@ -0,0 +1,1 @@+{"type":"update","queue":"orders","metrics":{"queue_name":"orders","queue_length":3,"newest_msg_age_sec":1,"oldest_msg_age_sec":9,"total_messages":12,"scrape_time":"2026-09-30T12:00:00Z","queue_visible_length":2,"default_partition_length":null},"source":"notify"}
+ test/golden/ws-update-poll.json view
@@ -0,0 +1,1 @@+{"type":"update","queue":"orders","metrics":{"queue_name":"orders","queue_length":3,"newest_msg_age_sec":1,"oldest_msg_age_sec":9,"total_messages":12,"scrape_time":"2026-09-30T12:00:00Z","queue_visible_length":2,"default_partition_length":null},"source":"poll"}