{"_id":"@anyq/pgmq","name":"@anyq/pgmq","dist-tags":{"latest":"0.5.0"},"versions":{"0.5.0":{"name":"@anyq/pgmq","version":"0.5.0","description":"Postgres (pgmq) adapter for anyq","type":"module","main":"./dist/index.js","module":"./dist/index.js","types":"./dist/index.d.ts","exports":{".":{"import":"./dist/index.js","types":"./dist/index.d.ts"}},"scripts":{"build":"bun build ./src/index.ts --outdir ./dist --target bun && bun run build:types","build:types":"tsc --emitDeclarationOnly --outDir dist","typecheck":"tsc --noEmit","test":"bun test","clean":"rm -rf dist"},"dependencies":{"@anyq/core":"0.5.0","pg":"^8.13.0"},"devDependencies":{"@types/bun":"latest","@types/pg":"^8.11.10","typescript":"^5.4.5"},"peerDependencies":{"@anyq/core":"0.5.0"},"keywords":["anyq","queue","pgmq","postgres","postgresql","message-queue"],"license":"Apache-2.0","author":"Shantanu Sharma","repository":{"type":"git","url":"https://github.com/sns45/anyq","directory":"packages/pgmq"},"homepage":"https://github.com/sns45/anyq#readme","bugs":{"url":"https://github.com/sns45/anyq/issues"},"publishConfig":{"access":"public"},"_id":"@anyq/pgmq@0.5.0","_integrity":"sha512-CoRI0EZCG9sfLXXr1ffYidCNg1RiiQrECsVpmkRV38nsucHPLH2iIUIiTifRt80eeWRJhyTlMSX1ISvJDGlvNw==","_nodeVersion":"26.3.0","_npmVersion":"10.8.3","shasum":"afda54107fd6c748c4e1dd85270df3ce56a3a47c","dist":{"integrity":"sha512-CoRI0EZCG9sfLXXr1ffYidCNg1RiiQrECsVpmkRV38nsucHPLH2iIUIiTifRt80eeWRJhyTlMSX1ISvJDGlvNw==","shasum":"afda54107fd6c748c4e1dd85270df3ce56a3a47c","tarball":"https://registry.npmjs.org/@anyq/pgmq/-/pgmq-0.5.0.tgz","fileCount":15,"unpackedSize":255656,"signatures":[{"keyid":"SHA256:DhQ8wR5APBvFHLF/+Tc+AYvPOdTpcIDqOhxsBHRwC7U","sig":"MEQCIA3gWIF8GIfVjRVsCyW3/rG4GOoFYCFD07qaJmdlbhjIAiBvW2DmvHKoqH9AIwEvnv+GuH9x5g9Ub5mWHAdXRN0Mvg=="}]},"_npmUser":{"name":"sns45","email":"notifyshantanu@gmail.com"},"directories":{},"maintainers":[{"name":"sns45","email":"notifyshantanu@gmail.com"}],"_npmOperationalInternal":{"host":"s3://npm-registry-packages-npm-production","tmp":"tmp/pgmq_0.5.0_1788809006076_0.9730812339746748"},"_hasShrinkwrap":false}},"time":{"created":"2026-09-07T19:23:25.946Z","0.5.0":"2026-09-07T19:23:26.212Z","modified":"2026-09-07T19:23:26.404Z"},"maintainers":[{"name":"sns45","email":"notifyshantanu@gmail.com"}],"description":"Postgres (pgmq) adapter for anyq","homepage":"https://github.com/sns45/anyq#readme","keywords":["anyq","queue","pgmq","postgres","postgresql","message-queue"],"repository":{"type":"git","url":"https://github.com/sns45/anyq","directory":"packages/pgmq"},"author":"Shantanu Sharma","bugs":{"url":"https://github.com/sns45/anyq/issues"},"license":"Apache-2.0","readme":"# @anyq/pgmq\n\nPostgres adapter for [anyq](https://github.com/sns45/anyq), backed by [pgmq](https://github.com/pgmq/pgmq). Run a real queue on the database you already operate, with the same producer and consumer API as every other anyq driver.\n\n## Install\n\n```bash\nbun add @anyq/core @anyq/pgmq\n```\n\npgmq must be present in the target database. The adapter runs `CREATE EXTENSION IF NOT EXISTS pgmq` on connect when the `pgmq` schema is missing (set `autoInstall: false` to disable). Where you cannot create extensions, use pgmq's SQL only install:\n\n```bash\npsql -f pgmq-extension/sql/pgmq.sql postgres://user:pass@host:5432/db\n```\n\n### Managed Postgres: check the extension allowlist first\n\nAs of Sept 7, 2026, `pgmq` is not on the extension allowlist of the managed services below, so `CREATE EXTENSION pgmq` is refused there and the adapter's `autoInstall` fails with a `ConfigurationError`. Skype's `pgq` is absent from the same lists.\n\n| Service | Checked against | pgmq listed? |\n|---|---|---|\n| Amazon RDS for PostgreSQL | [Extension versions](https://docs.aws.amazon.com/AmazonRDS/latest/PostgreSQLReleaseNotes/postgresql-extensions.html), PostgreSQL 9.6 to 19 | No |\n| Amazon Aurora PostgreSQL | [Extensions supported](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraPostgreSQLReleaseNotes/AuroraPostgreSQL.Extensions.html), PostgreSQL 10 to 18 | No |\n| Azure Database for PostgreSQL flexible server | [Extensions by name](https://learn.microsoft.com/en-us/azure/postgresql/extensions/concepts-extensions-versions), article dated 2026-07-10 | No |\n| Google Cloud SQL for PostgreSQL | [Configure extensions](https://docs.cloud.google.com/sql/docs/postgres/extensions), updated 2026-08-28 | No |\n| Google AlloyDB | [Supported extensions](https://docs.cloud.google.com/alloydb/docs/reference/extensions), updated 2026-08-26 | No |\n| Neon | [Postgres extensions](https://neon.com/docs/extensions/pg-extensions) | No |\n| Supabase | [Supabase Queues](https://supabase.com/docs/guides/queues) | Yes, Supabase Queues is built on pgmq |\n\nOn the services that do not list it, pgmq documents a SQL only install for exactly this case: pgmq 1.x is plain SQL and PL/pgSQL, so `psql -f pgmq-extension/sql/pgmq.sql <url>` creates the `pgmq` schema without extension privileges. Run that as the database owner, then connect with `autoInstall: false`. This adapter detects a SQL only install (it looks for `pgmq.read`, not for an extension row). The SQL only path has not been verified on those services by this project, and allowlists change, so check the vendor page before relying on this note.\n\nFor local work the pgmq project publishes a Postgres image:\n\n```bash\ndocker run -d --name pgmq -e POSTGRES_PASSWORD=postgres -p 5432:5432 quay.io/tembo/pg17-pgmq:latest\n```\n\n## Quick start\n\n```typescript\nimport { createPgmqProducer, createPgmqConsumer } from '@anyq/pgmq';\n\nconst pg = { connectionString: 'postgres://postgres:postgres@localhost:5432/postgres' };\n\nconst producer = createPgmqProducer<{ orderId: string }>({ queueName: 'orders', pg });\nconst consumer = createPgmqConsumer<{ orderId: string }>({\n  queueName: 'orders',\n  pg,\n  consumer: { visibilityTimeout: 30 },\n  deadLetterQueue: { enabled: true, destination: 'orders_dlq', maxDeliveryAttempts: 3, includeError: true },\n});\n\nawait producer.connect();\nawait consumer.connect();\n\nawait consumer.subscribe(async (message) => {\n  console.log(message.body, 'attempt', message.deliveryAttempt);\n  await message.ack();\n});\n\nawait producer.publish({ orderId: '123' });\nawait producer.publish({ orderId: '124' }, { delaySeconds: 60 });\n```\n\n## How it maps onto pgmq\n\n| anyq | pgmq |\n|---|---|\n| `publish` with `delaySeconds` | `pgmq.send(queue, msg, headers, delay)` |\n| `publishBatch` | `pgmq.send_batch` (one call per distinct delay) |\n| poll | `pgmq.read_with_poll(queue, vt, qty, ...)`, falling back to `pgmq.read` |\n| `ack()` | `pgmq.delete` |\n| `nack(true)` | `pgmq.set_vt(queue, id, 0)`: visible again at once |\n| `nack(false)` | `pgmq.archive`: kept in `pgmq.a_<queue>`, never silently dropped |\n| `extendDeadline(seconds)` | `pgmq.set_vt(queue, id, seconds)` (absolute from now, like SQS) |\n| `deliveryAttempt` | `read_ct` |\n| `park` (retry strategies) | `pgmq.set_vt(queue, id, delay)`, native, no in process downgrade |\n| dead letter | `pgmq.send` to the DLQ queue, then `pgmq.delete` the original |\n\nHeaders travel in pgmq's `headers` jsonb column; the routing `key` is stored there under the reserved header `x-anyq-key`. Dead lettered messages carry `x-original-queue`, `x-death-time`, `x-delivery-attempts` and, unless `includeError` is false, `x-death-reason`.\n\n**Headers are text.** String values are stored as they are. `Buffer` values must be valid UTF-8; they are stored as text and arrive on the consumer as strings. A `Buffer` that is not valid UTF-8 is rejected with a `SerializationError` at publish time rather than being altered (no U+FFFD replacement), which matches the Go adapter. Binary header values must be encoded by the caller, for example as base64.\n\nThe visibility timeout is the processing lease. A message that is neither acked nor nacked becomes visible again when `vt` lapses, with `read_ct` incremented, so `deliveryAttempt` is always truthful.\n\nThree rules follow from that lease:\n\n* **Rows read together are handled together.** Every row returned by one poll carries its own lease, so the consumer runs their handlers concurrently (up to `concurrency`); running them one after another would let the later leases expire while the earlier handlers run.\n* **An explicit settlement is final, once it has succeeded.** `ack()` or `nack()` called inside a handler wins over `autoAck`, and later `ack`, `nack` or `extendDeadline` calls on a settled message are no ops. A handler that nacks with requeue and then returns normally gets the redelivery it asked for. The settled state is recorded only after the SQL succeeds, so an `ack` or `nack` that throws (connection blip, timeout) can simply be called again. Operations on one delivery are serialised in call order (the Go adapter uses a per delivery mutex for the same reason), so `nack(true)` followed at once by `ack()` or `extendDeadline()` keeps the requeue: the later call is a no op.\n* **A poll never overlaps itself.** The loop waits for every handler from one read to finish, including when one of them ends in a `fail` decision, before it reads again or lets `disconnect()` return. `concurrency` is therefore a hard ceiling, not a target.\n* **Settlement is by message id, not by a receipt token.** pgmq has no per delivery receipt (unlike SQS), so `ack`, `nack` and `extendDeadline` apply to whatever delivery of that `msg_id` exists when they run. A handler that outlives its lease can therefore settle a newer delivery that another consumer currently holds. Keep handlers idempotent, call `extendDeadline` before the lease lapses on slow work, and treat a redelivery as possible after any lease expiry.\n* **Dead lettering is one transaction.** The DLQ insert and the source delete commit together. If the source row is already gone (settled elsewhere, or the lease expired and another consumer finished it) the transaction rolls back, so no orphan DLQ copy is written. If the transfer fails for any other reason nothing is archived or deleted; the lease is left to expire and the message is retried after `visibilityTimeout`.\n\n## Configuration\n\n| Option | Default | Notes |\n|---|---|---|\n| `queueName` | required | Must match `^[a-zA-Z0-9_]{1,47}$` (pgmq limit) |\n| `pg.connectionString` | | Or `host`, `port`, `user`, `password`, `database`, `ssl`, `max` |\n| `pool` | | Reuse an existing `pg.Pool`; the adapter will not end it |\n| `autoCreate` | `true` | `pgmq.create` for the queue and DLQ on connect |\n| `autoInstall` | `true` | `CREATE EXTENSION IF NOT EXISTS pgmq` when the schema is missing |\n| `producer.delaySeconds` | `0` | Default delay when `PublishOptions.delaySeconds` is absent |\n| `consumer.visibilityTimeout` | `30` | Seconds a read message stays invisible |\n| `consumer.longPollSeconds` | `5` | Server side wait via `read_with_poll`; `0` uses `read` plus `pollingInterval` |\n| `consumer.longPollIntervalMs` | `100` | Check interval inside a long poll |\n| `consumer.pollingInterval` | `1000` | Sleep in ms when the queue is empty and long polling is off |\n| `consumer.maxMessages` | `100` | Upper bound per read |\n| `deadLetterQueue` | disabled | `destination` defaults to `<queueName>_dlq` when enabled |\n\n`healthCheck()` reports round trip latency plus `queueLength` and `totalMessages` from `pgmq.metrics`.\n\n## Retry strategies\n\nAll `@anyq/core` strategies work end to end. `park` decisions (for example from `backpressurePause`) are scheduled natively through `set_vt`, so the consumer never blocks in process and no downgrade warning is logged at startup. Without a strategy, a failing handler leaves the message to reappear after the visibility timeout, and the adapter dead letters it once `deadLetterQueue.maxDeliveryAttempts` is reached.\n\n## Tests\n\nThe integration suite needs a pgmq Postgres and skips itself otherwise:\n\n```bash\ndocker run -d --name pgmq -e POSTGRES_PASSWORD=postgres -p 5433:5432 quay.io/tembo/pg17-pgmq:latest\nPGMQ_URL=postgres://postgres:postgres@localhost:5433/postgres bun test\n```\n\n## License\n\nApache License 2.0. See [LICENSE](https://github.com/sns45/anyq/blob/main/LICENSE).\n","readmeFilename":"README.md","_rev":"1-63bc05e287a7c9bb88ff3f421643c258"}