From f25cb3c06b6c14055636a90b935409fdcd888f02 Mon Sep 17 00:00:00 2001 From: Copilot <223556219+Copilot@users.noreply.github.com> Date: Wed, 28 Oct 2026 17:31:12 +0000 Subject: [PATCH 1/3] feat(hub): add PostgreSQL background jobs Add separate on-demand and scheduled BullMQ consumers, explicit release setup, validated job contracts, and isolated PostgreSQL test coverage without porting product jobs. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Co-authored-by: Nishiki <1121087+nshki@users.noreply.github.com> --- hub/.env.example | 4 + hub/README.md | 2 +- hub/docs/README.md | 1 + hub/docs/architecture.md | 4 +- hub/docs/background-jobs.md | 93 ++++++ hub/docs/database.md | 2 + hub/docs/development.md | 6 + hub/docs/testing-and-operations.md | 8 +- hub/package-lock.json | 205 ++++++++++++- hub/package.json | 11 +- .../bullmq/create-job-processor/index.test.ts | 23 ++ .../bullmq/create-job-processor/index.ts | 33 ++ .../bullmq/create-job-queue/index.ts | 25 ++ .../bullmq/create-job-worker/index.db.test.ts | 290 ++++++++++++++++++ .../bullmq/create-job-worker/index.test.ts | 49 +++ .../bullmq/create-job-worker/index.ts | 61 ++++ .../integrations/bullmq/enqueue-job/index.ts | 14 + .../bullmq/get-job-connection/index.ts | 18 ++ .../bullmq/migrate-job-database/index.ts | 14 + .../bullmq/parse-job-payload/index.ts | 9 + hub/server/integrations/bullmq/types.ts | 35 +++ .../upsert-job-schedule/index.db.test.ts | 63 ++++ .../bullmq/upsert-job-schedule/index.ts | 44 +++ hub/server/setup-jobs/index.ts | 28 ++ hub/server/workers/configuration.ts | 19 ++ hub/server/workers/index.ts | 25 ++ .../workers/run-job-worker/index.test.ts | 186 +++++++++++ hub/server/workers/run-job-worker/index.ts | 57 ++++ hub/test/database/global-setup/index.ts | 2 + .../database/reset-database/index.db.test.ts | 121 ++++++++ hub/test/database/reset-database/index.ts | 2 +- 31 files changed, 1444 insertions(+), 10 deletions(-) create mode 100644 hub/docs/background-jobs.md create mode 100644 hub/server/integrations/bullmq/create-job-processor/index.test.ts create mode 100644 hub/server/integrations/bullmq/create-job-processor/index.ts create mode 100644 hub/server/integrations/bullmq/create-job-queue/index.ts create mode 100644 hub/server/integrations/bullmq/create-job-worker/index.db.test.ts create mode 100644 hub/server/integrations/bullmq/create-job-worker/index.test.ts create mode 100644 hub/server/integrations/bullmq/create-job-worker/index.ts create mode 100644 hub/server/integrations/bullmq/enqueue-job/index.ts create mode 100644 hub/server/integrations/bullmq/get-job-connection/index.ts create mode 100644 hub/server/integrations/bullmq/migrate-job-database/index.ts create mode 100644 hub/server/integrations/bullmq/parse-job-payload/index.ts create mode 100644 hub/server/integrations/bullmq/types.ts create mode 100644 hub/server/integrations/bullmq/upsert-job-schedule/index.db.test.ts create mode 100644 hub/server/integrations/bullmq/upsert-job-schedule/index.ts create mode 100644 hub/server/setup-jobs/index.ts create mode 100644 hub/server/workers/configuration.ts create mode 100644 hub/server/workers/index.ts create mode 100644 hub/server/workers/run-job-worker/index.test.ts create mode 100644 hub/server/workers/run-job-worker/index.ts create mode 100644 hub/test/database/reset-database/index.db.test.ts diff --git a/hub/.env.example b/hub/.env.example index ed8cba81..ed2821cd 100644 --- a/hub/.env.example +++ b/hub/.env.example @@ -4,6 +4,10 @@ PORT=3001 # API logging threshold: trace, debug, info, warn, error, fatal, or silent. LOG_LEVEL=info +# Independent worker concurrency. Both default to one (supported range: 1-100). +ON_DEMAND_JOB_CONCURRENCY=1 +SCHEDULED_JOB_CONCURRENCY=1 + # Local development only. The devcontainer supplies its own database hostname. DATABASE_URL=postgresql://replica:replica@127.0.0.1:5432/replica diff --git a/hub/README.md b/hub/README.md index fba50194..47ef69db 100644 --- a/hub/README.md +++ b/hub/README.md @@ -4,7 +4,7 @@ Hub is being rebuilt on `build/hub/new` using TypeScript, Express, React, Vite, The API has one route: `GET /api/health`. The frontend displays the existing Replica logo and supports light and dark themes. Existing components, features, and pages are kept in Storybook, but they are not connected to the new API. -Contributors can use PostgreSQL and Prisma locally or in the devcontainer. The database has no product models yet. Sign-in, saved data, live updates, background jobs, and deployment are not implemented. +Contributors can use PostgreSQL and Prisma locally or in the devcontainer. The [background-job foundation](docs/background-jobs.md) provides PostgreSQL-backed BullMQ queues and independent on-demand and scheduled workers, without product jobs yet. The database has no product models. Sign-in, saved product data, live updates, and deployment are not implemented. Use [Development](docs/development.md) to start the API and frontend. Review [Architecture](docs/architecture.md), [code architecture conventions](docs/backend-architecture.md), and [coding conventions](docs/coding-conventions.md) before adding behavior. diff --git a/hub/docs/README.md b/hub/docs/README.md index ec3bd8fe..5d471ed3 100644 --- a/hub/docs/README.md +++ b/hub/docs/README.md @@ -8,6 +8,7 @@ Hub is being rebuilt. The [Hub overview](../README.md) describes what the baseli - [Development](development.md): local API and frontend setup. - [Database](database.md): Prisma schema conventions, migrations, and the shared client. - [Logging](logging.md): application events, automatic request records, and payload safety. +- [Background jobs](background-jobs.md): PostgreSQL queues, workers, schedules, setup, and test isolation. - [Testing and operations](testing-and-operations.md): baseline checks and their limits. - [Storybook](storybook.md): retained presentation, fixtures, and screenshot evidence. - [Prototype canvas](prototype-canvas.md): the separately retained extension. diff --git a/hub/docs/architecture.md b/hub/docs/architecture.md index eb7d24c5..bf72e320 100644 --- a/hub/docs/architecture.md +++ b/hub/docs/architecture.md @@ -8,7 +8,7 @@ The team has chosen TypeScript, Express for the API, React and Vite for the fron The API responds to a health check, and the frontend displays the logo with light and dark themes. Storybook keeps the existing components and pages available for reuse. -Contributors can use PostgreSQL and Prisma locally or in the devcontainer. The database has no product models yet. Sign-in, deployment, background jobs, and product APIs have not been built. This app cannot yet find repositories, import Builds, host previews, save product data, or support reviews. +Contributors can use PostgreSQL and Prisma locally or in the devcontainer. The database has no product models yet. Sign-in, deployment, and product APIs have not been built. This app cannot yet find repositories, import Builds, host previews, save product data, or support reviews. ## Backend @@ -20,6 +20,8 @@ The [database guide](database.md) defines the multi-file Prisma schema and share The [Pino integration](logging.md) writes structured application and HTTP logs to stdout. `createApp()` installs request logging and JSON parsing before routes, so new endpoints inherit completion records and server-generated request IDs without explicit logging calls. +The [background-job foundation](background-jobs.md) uses BullMQ with PostgreSQL and separate on-demand and scheduled worker processes. Setup migrates BullMQ's own schema independently of Prisma. Product processors and schedules are not registered yet. + ## Frontend ```mermaid diff --git a/hub/docs/background-jobs.md b/hub/docs/background-jobs.md new file mode 100644 index 00000000..77f7650f --- /dev/null +++ b/hub/docs/background-jobs.md @@ -0,0 +1,93 @@ +# Background jobs + +Hub uses BullMQ's PostgreSQL backend for internal background work. The foundation provides queues, validated job payloads, recurring schedules, and independent worker commands. No product jobs or production schedules are registered yet. The Rails jobs have not been ported. + +## Queues and workers + +Two stable queue names separate execution capacity: + +| Queue | Work | Command from `hub/` | +| --- | --- | --- | +| `on-demand` | Work requested by an application action | `npm run worker:on-demand` | +| `scheduled` | Recurring work and work spawned by recurring sweeps | `npm run worker:scheduled` | + +Each command starts one Node.js process consuming only its assigned queue. The API starts neither worker. Multiple processes may consume the same queue when more capacity is needed. Separate queues do not isolate PostgreSQL, CPU, or external service quotas. + +`ON_DEMAND_JOB_CONCURRENCY` and `SCHEDULED_JOB_CONCURRENCY` independently set the number of active jobs per worker process. Both default to one and accept integers from 1 through 100. CPU-heavy JavaScript still blocks that worker's event loop; move that work to a subprocess or an appropriate execution boundary when adding the domain. + +`server/workers/configuration.ts` is the composition point for processors and schedules. Its empty registrations are intentional until product domains exist. Worker startup logs readiness after connecting; unknown job names fail permanently rather than disappearing. + +Workers stop accepting work on `SIGINT` or `SIGTERM` and wait for active jobs. After 30 seconds, the command exits unsuccessfully rather than waiting forever. Interrupted work can be retried by BullMQ's stalled-job recovery. Deployment stop timeouts must allow this shutdown window. Long-running domain work needs its own bounded execution and interruption policy; a queue lock is not a business lock. + +## Database and setup + +Jobs use `DATABASE_URL`, the same database as application data, with BullMQ-owned tables and functions in the `bullmq` schema. Prisma owns only the application schema. Do not copy BullMQ tables into Prisma models or change its migration files. + +From `hub/`, after installing dependencies and starting PostgreSQL: + +```sh +npm run setup +``` + +This runs `prisma migrate deploy`, then BullMQ migrations, then schedule registration. A failed step stops setup. For local work that only needs the job schema, use `npm run jobs:setup`. + +Run setup once as a coordinated release step before starting the new application version. Do not run it in every API or worker startup. Migrations are not one transaction across Prisma and BullMQ, and changes must remain compatible with old processes during a rolling upgrade. BullMQ schema downgrades are not supported. + +After building, the compiled equivalents are: + +```sh +npm run start:setup +npm start +npm run start:worker:on-demand +npm run start:worker:scheduled +``` + +Run the last three commands as separate services, not sequential commands in one shell. These roles can use one application image in a future Compose deployment, with PostgreSQL shared and only the web service exposed. Production frontend hosting, an application image, service health checks, and release orchestration are not implemented by this foundation. + +Each queue or worker owns a small query pool and BullMQ's additional notification connections. The query pool is limited to two connections; concurrency does not increase that limit. Budget database connections across all API, producer, and worker instances. Creation functions require an explicit URL so tests and other callers cannot accidentally inherit a previously imported database client. + +## Defining work + +Keep stable names, a Zod payload schema, and retry options in a job definition. Producers need the definition, not the domain's execution code. Register a processor at the worker composition point using `createJobProcessor(definition, process, shouldRetry)`. The optional `shouldRetry(error)` predicate makes an execution failure permanent when it returns false; errors otherwise follow the job's retry options. + +Use the functions under `server/integrations/bullmq/`: + +| Function | Contract | +| --- | --- | +| `migrateJobDatabase(databaseUrl)` | Explicitly migrate BullMQ's schema and close the migration connection. | +| `createJobQueue(options)` | Open a client for one named PostgreSQL queue. The caller must close it. | +| `enqueueJob(queue, definition, payload, options)` | Validate JSON payload data and enqueue work with job-specific retries and optional delay, priority, or job ID. Failure rejects the call. | +| `createJobProcessor(definition, process, shouldRetry)` | Bind typed execution to runtime payload validation. Invalid stored payloads fail permanently. | +| `createJobWorker(options)` | Construct a stopped consumer for registered processors. The caller starts and closes it. Duplicate processor names are rejected. | +| `upsertJobSchedule(queue, schedule)` | Validate and register an interval or timezone-explicit cron schedule on `scheduled`. | + +`runJobWorker()` owns process signals and consumer lifetime outside the integration. Integration code does not import domains, repositories, or the shared logger; worker composition injects processors and logging. + +Payloads must serialize as JSON. Prefer record IDs and necessary revision identifiers over snapshots, credentials, or private text. Load current records and recheck access when execution begins. Do not rely on TypeScript alone: enqueue and execution both validate input. + +Retries default to one attempt. A definition can specify `attempts`, `backoff`, and `priority`; an enqueue can override priority. Choose retries according to the operation, not a universal policy. Retention keeps at most 1,000 successful jobs for up to one day and 5,000 failed jobs for up to seven days. BullMQ applies removal lazily as jobs finish, not as a strict expiry timer. Job records can retain payloads and failure messages, so restrict database access and keep secrets out of thrown errors as well as payloads. + +Worker logs contain queue, job name, job ID, attempt counts, and error types, never payloads, results, or raw exception messages. Domain functions should report safe failure codes and necessary context through the existing logging policy. Connection failures and command failures produce explicit error records. + +## Schedules and delivery guarantees + +A schedule has a stable ID within a stable queue. Setup calls `upsertJobSchedule()` with that identity, so repeated releases update one schedule instead of adding another. Removing an entry from code does not remove its persisted schedule: explicitly call BullMQ's `removeJobScheduler()` when retiring or renaming one. Coordinate updates through one release step so old and new processes do not overwrite each other's definitions. + +Keep work spawned by a recurring sweep on `scheduled`; otherwise a sweep can consume on-demand capacity indirectly. Scheduled times are not execution deadlines, and a busy queue can run less often than its configured interval. Define missed-run and overlap rules when each domain is introduced. + +Assume at-least-once execution. A stable job or scheduler ID does not replace idempotent handlers or keyed exclusion across workers and queues. Sharing PostgreSQL also does not make a Prisma write and a BullMQ enqueue atomic. Future domains that need guaranteed submission must explicitly implement an outbox or another durable recovery contract. + +## Tests + +Ordinary `npm test` requires no database. Use `.db.test.ts` cases and `npm run test:database` for real BullMQ behavior. The [database harness](testing-and-operations.md#parallel-database-tests) applies Prisma and BullMQ setup to a unique template before cloning a database for each Vitest worker. + +Pass the current worker's resolved `DATABASE_URL` to every queue and consumer. Do not use the administrative `TEST_DATABASE_URL` or a production connection. Each test registers its own synthetic processors and schedules; setup does not install production schedules. + +Close consumers, queue clients, and any owned connections before the next case resets data. Clearing Vitest's module cache does not stop consumers. Cleanup clears job and schedule tables but preserves `public._prisma_migrations` and `bullmq.migration`; preserving a migration ledger is not preserving queued work. + +## References + +- [BullMQ PostgreSQL backend](https://docs.bullmq.io/guide/postgresql) +- [Job Schedulers](https://docs.bullmq.io/guide/job-schedulers) +- [Managing schedules](https://docs.bullmq.io/guide/job-schedulers/manage-job-schedulers) +- [Worker shutdown](https://docs.bullmq.io/guide/workers/graceful-shutdown) diff --git a/hub/docs/database.md b/hub/docs/database.md index 6cb52918..1663e7f1 100644 --- a/hub/docs/database.md +++ b/hub/docs/database.md @@ -2,6 +2,8 @@ Hub uses PostgreSQL and Prisma. Product models and migrations have not been added. +[Background jobs](background-jobs.md) use the same database with BullMQ-owned tables and functions in the `bullmq` schema. Prisma does not own those objects. `npm run setup` applies Prisma migrations before BullMQ setup and schedule registration. + [Compose](../.devcontainer/compose.yaml) pins the development database image. [package.json](../package.json) pins the Prisma packages to the same version. Dependency updates must follow the [seven-day dependency policy](../../docs/testing-and-review.md#update-dependencies). Choose the newest stable PostgreSQL major available from both Azure Database for PostgreSQL and Amazon RDS or Aurora. Managed services can offer different patch releases; the local image tracks a stable patch of that shared major. Check provider availability again when changing the major, rather than assuming that an upstream release is available everywhere. diff --git a/hub/docs/development.md b/hub/docs/development.md index b91ecf9f..452682d3 100644 --- a/hub/docs/development.md +++ b/hub/docs/development.md @@ -40,6 +40,12 @@ Open . Vite serves the frontend and forwards `/api` reque The API reads an optional `PORT` environment variable, defaulting to 3001. If you change it, update the local Vite proxy target to match. The editor forwards ports 3001, 5173, and 6006 from the devcontainer. API startup reads `.env` when present; Compose's environment takes precedence. +## Background workers + +The API and frontend do not start background consumers. After starting PostgreSQL, run `npm run setup` once, then start `npm run worker:on-demand` and `npm run worker:scheduled` in separate terminals. Both use `DATABASE_URL`; their concurrency settings are in `.env.example`. + +No product jobs or recurring schedules are installed yet. See [Background jobs](background-jobs.md) for definitions, shutdown, migrations, and the compiled commands intended for separate deployment services. + ## Database lifecycle PostgreSQL stores data in a named Compose volume. Stopping or rebuilding the devcontainer preserves it. For a local Node.js setup, stop the database with `docker compose -f .devcontainer/compose.yaml down`; this also preserves data. Do not add `--volumes` unless you intend to permanently delete the local database. diff --git a/hub/docs/testing-and-operations.md b/hub/docs/testing-and-operations.md index 4730dc20..8babdf32 100644 --- a/hub/docs/testing-and-operations.md +++ b/hub/docs/testing-and-operations.md @@ -51,15 +51,15 @@ Use `.test.ts` or `.test.tsx` for ordinary cases. Use `.db.test.ts` or `.db.test Start the PostgreSQL service as described in [Development](development.md) and set `TEST_DATABASE_URL` using `.env.example`. The devcontainer supplies this variable. Use a local database user with permission to create and drop databases, never a production connection. -The database command uses `--passWithNoTests` so an empty selection exits successfully. It still fails when tests fail. There are no database test cases yet; an empty run does not prepare or exercise the databases. +The database command uses `--passWithNoTests` so an empty selection exits successfully. It still fails when tests fail. The job integration cases exercise real PostgreSQL queues, schedules, and database resets. Database-enabled jsdom tests use the browser setup and database hooks together. They can arrange records through server code, but frontend code must not import server modules. The setup does not start an API or connect frontend requests to one. The database setup follows this lifecycle: -1. Before workers start, setup creates a template database with a unique name for each project and run. It applies the current Prisma schema with `prisma db push`, then copies the template for every worker slot. +1. Before workers start, setup creates a template database with a unique name for each project and run. It applies the current Prisma schema with `prisma db push` and BullMQ's migrations, closes those setup connections, then copies the template for every worker slot. 2. Each worker receives its own `DATABASE_URL` before application modules load. Setup uses Vitest's [`VITEST_POOL_ID`](https://vitest.dev/guide/migration/jest#envs), which stays within the configured worker limit. -3. Before every case, setup empties application tables across user schemas and resets generated identity counters. It preserves Prisma's migration records. Each case creates its own records; there is no shared seed data. +3. Before every case, setup empties application and job tables across user schemas and resets generated identity counters. It preserves `public._prisma_migrations` and `bullmq.migration`. Each case creates its own records; there is no shared seed data. 4. After the run, teardown drops only the databases it created, even if tests left connections open. Unique database names also isolate simultaneous runs. Before clearing tables, cleanup checks that the database name belongs to the test namespace and matches the connected database. @@ -68,6 +68,8 @@ The helpers under `test/` prepare databases before Prisma Client and application Hub CI runs database tests against PostgreSQL on every supported Node major. Only that command starts the database service. +Job tests use each worker's resolved `DATABASE_URL`, never `TEST_DATABASE_URL`. Close their consumers, queues, and owned connections before the next case's reset; clearing the module cache does not stop running jobs. See [Background jobs](background-jobs.md#tests) for job-specific lifecycle rules. + ## Shared code The extension and build action still use Hub's Controls and canvas code. Keep their checks working: diff --git a/hub/package-lock.json b/hub/package-lock.json index 6594c6db..25ceab39 100644 --- a/hub/package-lock.json +++ b/hub/package-lock.json @@ -13,8 +13,10 @@ "@primer/react-brand": "0.78.0", "@prisma/adapter-pg": "7.10.1", "@prisma/client": "7.10.1", + "bullmq": "6.4.0", "dotenv": "18.0.8", "express": "5.3.0", + "pg": "8.23.1", "pino": "10.4.0", "pino-http": "11.0.0", "react": "19.3.0", @@ -45,7 +47,6 @@ "globals": "17.13.0", "jiti": "2.7.0", "jsdom": "30.1.3", - "pg": "8.23.1", "prettier": "3.9.10", "prisma": "7.10.1", "storybook": "10.6.2", @@ -1840,6 +1841,84 @@ "react": ">=16" } }, + "node_modules/@msgpackr-extract/msgpackr-extract-darwin-arm64": { + "version": "3.0.4", + "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-darwin-arm64/-/msgpackr-extract-darwin-arm64-3.0.4.tgz", + "integrity": "sha512-LCkGo6JDfaBhgST7UpPWgNgLINpcpabaHfyz5OBx75nUYxBsaEPxjnyNjWpeb/xBup/682QnBfRBy2/LvPutZQ==", + "cpu": [ + "arm64" + ], + "license": "MIT", + "optional": true, + "os": [ + "darwin" + ] + }, + "node_modules/@msgpackr-extract/msgpackr-extract-darwin-x64": { + "version": "3.0.4", + "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-darwin-x64/-/msgpackr-extract-darwin-x64-3.0.4.tgz", + "integrity": "sha512-zExlW9zUJKZH/tOtVMttwjKa4Xm/3KcNjnE3dPN92uCktwavMxpgCA3MoJK/DOnTWsQgo224OaST27/mPNAf+w==", + "cpu": [ + "x64" + ], + "license": "MIT", + "optional": true, + "os": [ + "darwin" + ] + }, + "node_modules/@msgpackr-extract/msgpackr-extract-linux-arm": { + "version": "3.0.4", + "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-linux-arm/-/msgpackr-extract-linux-arm-3.0.4.tgz", + "integrity": "sha512-Tg3yX65f5GbtXLkrYEHE5oibZG9epyYWas7FogTTEJeDEF9JlXJzKgXaNhT3UXlTOeA+AfZpYZYZ0uPj7Cfquw==", + "cpu": [ + "arm" + ], + "license": "MIT", + "optional": true, + "os": [ + "linux" + ] + }, + "node_modules/@msgpackr-extract/msgpackr-extract-linux-arm64": { + "version": "3.0.4", + "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-linux-arm64/-/msgpackr-extract-linux-arm64-3.0.4.tgz", + "integrity": "sha512-dgX0P/9wGPJeHFBG+ZmhgE6bmtMt7NP5CRBGyyktpopdk/mW4POnrpQsSLtKI1dwpc+pPLuXHDh6vvskyQE/sw==", + "cpu": [ + "arm64" + ], + "license": "MIT", + "optional": true, + "os": [ + "linux" + ] + }, + "node_modules/@msgpackr-extract/msgpackr-extract-linux-x64": { + "version": "3.0.4", + "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-linux-x64/-/msgpackr-extract-linux-x64-3.0.4.tgz", + "integrity": "sha512-8TNXMEjJc3QEy7R/x1INhgiU+XakDAFUzBhaz7+Rbrs8NH5UQeHQxxmzsSBJGyV6I1jW79undiQm8tOI+D+8FQ==", + "cpu": [ + "x64" + ], + "license": "MIT", + "optional": true, + "os": [ + "linux" + ] + }, + "node_modules/@msgpackr-extract/msgpackr-extract-win32-x64": { + "version": "3.0.4", + "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-win32-x64/-/msgpackr-extract-win32-x64-3.0.4.tgz", + "integrity": "sha512-CmCXPQrkbwExx3j946/PtHWHbYJiCRBRDl4BlkRQcJB/YOwQxJRTpoo7aTsortjgoJ1x7opzTSxn7C+ASSLVjQ==", + "cpu": [ + "x64" + ], + "license": "MIT", + "optional": true, + "os": [ + "win32" + ] + }, "node_modules/@napi-rs/wasm-runtime": { "version": "1.2.4", "resolved": "https://registry.npmjs.org/@napi-rs/wasm-runtime/-/wasm-runtime-1.2.4.tgz", @@ -5146,6 +5225,54 @@ "node": "^6 || ^7 || ^8 || ^9 || ^10 || ^11 || ^12 || >=13.7" } }, + "node_modules/bullmq": { + "version": "6.4.0", + "resolved": "https://registry.npmjs.org/bullmq/-/bullmq-6.4.0.tgz", + "integrity": "sha512-wEx8b4tDmW1upAOXS0SCy0No+DBTAkV82GFP/rb0EAM3qtAOAXAXKJ2OKiWq4g7xZ1Bv+WpkwvvJx1mxWY+taw==", + "license": "MIT", + "dependencies": { + "cron-parser": "5.10.2", + "msgpackr": "2.1.0", + "node-abort-controller": "3.1.1", + "semver": "7.8.5", + "tslib": "2.8.1" + }, + "engines": { + "node": ">=14.17.0" + }, + "peerDependencies": { + "bullmq-otel": ">=2.0.0", + "ioredis": ">=5.0.0", + "pg": ">=8.0.0", + "redis": ">=5.0.0" + }, + "peerDependenciesMeta": { + "bullmq-otel": { + "optional": true + }, + "ioredis": { + "optional": true + }, + "pg": { + "optional": true + }, + "redis": { + "optional": true + } + } + }, + "node_modules/bullmq/node_modules/semver": { + "version": "7.8.5", + "resolved": "https://registry.npmjs.org/semver/-/semver-7.8.5.tgz", + "integrity": "sha512-Y7/KDsb8LjooZpwaqGyulO6DQlksgCncchHGk+sZIY4SBvUocMBEFH5Ur1fI4dV+Jvl0w6cjvucaIi40puRioA==", + "license": "ISC", + "bin": { + "semver": "bin/semver.js" + }, + "engines": { + "node": ">=10" + } + }, "node_modules/bundle-name": { "version": "4.1.0", "resolved": "https://registry.npmjs.org/bundle-name/-/bundle-name-4.1.0.tgz", @@ -5385,6 +5512,18 @@ "node": ">=6.6.0" } }, + "node_modules/cron-parser": { + "version": "5.10.2", + "resolved": "https://registry.npmjs.org/cron-parser/-/cron-parser-5.10.2.tgz", + "integrity": "sha512-JfAYMY0LZaWQTuict8PhtF/ksSnq8xefSDmQUnnoqvkPYkU/y3UP0DTWzJKFBz0P1IXsDG8q0vY3zHkoFZOhvw==", + "license": "MIT", + "dependencies": { + "luxon": "^3.7.2" + }, + "engines": { + "node": ">=18" + } + }, "node_modules/cross-spawn": { "version": "7.0.6", "resolved": "https://registry.npmjs.org/cross-spawn/-/cross-spawn-7.0.6.tgz", @@ -5731,7 +5870,7 @@ "version": "2.1.2", "resolved": "https://registry.npmjs.org/detect-libc/-/detect-libc-2.1.2.tgz", "integrity": "sha512-Btj2BOOO83o3WyH59e8MgXsxEQVcarkUOpEYrubB0urwnN10yQ364rsiByU11nZlqWYZm05i/of7io4mzihBtQ==", - "dev": true, + "devOptional": true, "license": "Apache-2.0", "engines": { "node": ">=8" @@ -7469,6 +7608,15 @@ "url": "https://github.com/sponsors/wellwelwel" } }, + "node_modules/luxon": { + "version": "3.7.2", + "resolved": "https://registry.npmjs.org/luxon/-/luxon-3.7.2.tgz", + "integrity": "sha512-vtEhXh/gNjI9Yg1u4jX/0YVPMvxzHuGgCm6tC5kZyb08yjGWGnqAjGJvcXbqQR2P3MyMEFnRbpcdFS6PBcLqew==", + "license": "MIT", + "engines": { + "node": ">=12" + } + }, "node_modules/lz-string": { "version": "1.5.0", "resolved": "https://registry.npmjs.org/lz-string/-/lz-string-1.5.0.tgz", @@ -7607,6 +7755,37 @@ "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==", "license": "MIT" }, + "node_modules/msgpackr": { + "version": "2.1.0", + "resolved": "https://registry.npmjs.org/msgpackr/-/msgpackr-2.1.0.tgz", + "integrity": "sha512-p/pBCVO63CsvvpkomUnNNag6+n38rULuDA6HHe70o2gtC8ODI52foF/4ko2qQcp6OiErJXTmrZeXmsGGHsIQNQ==", + "license": "MIT", + "optionalDependencies": { + "msgpackr-extract": "^3.0.4" + } + }, + "node_modules/msgpackr-extract": { + "version": "3.0.4", + "resolved": "https://registry.npmjs.org/msgpackr-extract/-/msgpackr-extract-3.0.4.tgz", + "integrity": "sha512-4kmO/MdyUIkLIvTPr8VHLil4AtoKIoniWPIEk5+CDy0xnWC84azhSFmuJ7PxZdsYtiP5kEeQsORAVIeMgxT+Hw==", + "hasInstallScript": true, + "license": "MIT", + "optional": true, + "dependencies": { + "node-gyp-build-optional-packages": "5.2.2" + }, + "bin": { + "download-msgpackr-prebuilds": "bin/download-prebuilds.js" + }, + "optionalDependencies": { + "@msgpackr-extract/msgpackr-extract-darwin-arm64": "3.0.4", + "@msgpackr-extract/msgpackr-extract-darwin-x64": "3.0.4", + "@msgpackr-extract/msgpackr-extract-linux-arm": "3.0.4", + "@msgpackr-extract/msgpackr-extract-linux-arm64": "3.0.4", + "@msgpackr-extract/msgpackr-extract-linux-x64": "3.0.4", + "@msgpackr-extract/msgpackr-extract-win32-x64": "3.0.4" + } + }, "node_modules/mysql2": { "version": "3.24.4", "resolved": "https://registry.npmjs.org/mysql2/-/mysql2-3.24.4.tgz", @@ -7684,6 +7863,27 @@ "url": "https://opencollective.com/express" } }, + "node_modules/node-abort-controller": { + "version": "3.1.1", + "resolved": "https://registry.npmjs.org/node-abort-controller/-/node-abort-controller-3.1.1.tgz", + "integrity": "sha512-AGK2yQKIjRuqnc6VkX2Xj5d+QW8xZ87pa1UK6yA6ouUyuxfHuMP6umE5QK7UmTeOAymo+Zx1Fxiuw9rVx8taHQ==", + "license": "MIT" + }, + "node_modules/node-gyp-build-optional-packages": { + "version": "5.2.2", + "resolved": "https://registry.npmjs.org/node-gyp-build-optional-packages/-/node-gyp-build-optional-packages-5.2.2.tgz", + "integrity": "sha512-s+w+rBWnpTMwSFbaE0UXsRlg7hU4FjekKU4eyAih5T8nJuNZT1nNsskXpxmeqSK9UzkBl6UgRlnKc8hz8IEqOw==", + "license": "MIT", + "optional": true, + "dependencies": { + "detect-libc": "^2.0.1" + }, + "bin": { + "node-gyp-build-optional-packages": "bin.js", + "node-gyp-build-optional-packages-optional": "optional.js", + "node-gyp-build-optional-packages-test": "build-test.js" + } + }, "node_modules/node-releases": { "version": "2.0.55", "resolved": "https://registry.npmjs.org/node-releases/-/node-releases-2.0.55.tgz", @@ -9368,7 +9568,6 @@ "version": "2.8.1", "resolved": "https://registry.npmjs.org/tslib/-/tslib-2.8.1.tgz", "integrity": "sha512-oJFu94HQb+KVduSUQL7wnpmqnfmLsOA/nAh6b6EH0wCEoK0/mPeXU6c3wKDV83MkOuHPRHtSXKKU99IBazS/2w==", - "dev": true, "license": "0BSD" }, "node_modules/type-check": { diff --git a/hub/package.json b/hub/package.json index f24effed..259f8e00 100644 --- a/hub/package.json +++ b/hub/package.json @@ -13,6 +13,14 @@ "db:migrate": "prisma migrate dev", "db:deploy": "prisma migrate deploy", "db:studio": "prisma studio", + "jobs:setup": "node server/setup-jobs/index.ts", + "setup": "npm run db:deploy && npm run jobs:setup", + "worker:on-demand": "node server/workers/index.ts on-demand", + "worker:scheduled": "node server/workers/index.ts scheduled", + "start:worker:on-demand": "node dist/server/workers/index.js on-demand", + "start:worker:scheduled": "node dist/server/workers/index.js scheduled", + "start:jobs:setup": "node dist/server/setup-jobs/index.js", + "start:setup": "npm run db:deploy && npm run start:jobs:setup", "controls:build": "node script/build-controls-bridge.mjs", "controls:assets:check": "node script/build-controls-bridge.mjs --check", "controls:test": "node --test script/tests/build-controls.test.mjs", @@ -42,10 +50,12 @@ "@prisma/client": "7.10.1", "@primer/octicons-react": "19.40.0", "@primer/react-brand": "0.78.0", + "bullmq": "6.4.0", "dotenv": "18.0.8", "express": "5.3.0", "pino": "10.4.0", "pino-http": "11.0.0", + "pg": "8.23.1", "react": "19.3.0", "react-dom": "19.3.0", "zod": "4.6.5" @@ -74,7 +84,6 @@ "globals": "17.13.0", "jiti": "2.7.0", "jsdom": "30.1.3", - "pg": "8.23.1", "prettier": "3.9.10", "prisma": "7.10.1", "storybook": "10.6.2", diff --git a/hub/server/integrations/bullmq/create-job-processor/index.test.ts b/hub/server/integrations/bullmq/create-job-processor/index.test.ts new file mode 100644 index 00000000..73096540 --- /dev/null +++ b/hub/server/integrations/bullmq/create-job-processor/index.test.ts @@ -0,0 +1,23 @@ +import { describe, expect, it, vi } from "vitest" +import { z } from "zod" + +import { createJobProcessor } from "." + +describe("createJobProcessor()", () => { + it("preserves unexpected schema errors as retryable failures", async () => { + const failure = new Error("Schema dependency failure") + const process = vi.fn(() => Promise.resolve("must not run")) + const processor = createJobProcessor( + { + name: "schema-failure", + schema: z.string().transform(() => { + throw failure + }), + }, + process, + ) + + await expect(processor.process("valid input")).rejects.toBe(failure) + expect(process).not.toHaveBeenCalled() + }) +}) diff --git a/hub/server/integrations/bullmq/create-job-processor/index.ts b/hub/server/integrations/bullmq/create-job-processor/index.ts new file mode 100644 index 00000000..ac65e8c4 --- /dev/null +++ b/hub/server/integrations/bullmq/create-job-processor/index.ts @@ -0,0 +1,33 @@ +import { UnrecoverableError } from "bullmq" +import { z } from "zod" + +import { parseJobPayload } from "../parse-job-payload/index.ts" +import type { JobDefinition, JobProcessor } from "../types.ts" + +/** Bind typed domain input without making queue producers import execution code. */ +export function createJobProcessor( + definition: JobDefinition, + process: (data: T) => Promise, + shouldRetry?: (error: unknown) => boolean, +): JobProcessor { + return { + name: definition.name, + async process(payload) { + let data: T + try { + data = parseJobPayload(definition, payload) + } catch (error) { + if (error instanceof z.ZodError) throw new UnrecoverableError("Job payload is invalid.") + throw error + } + try { + return await process(data) + } catch (error) { + if (shouldRetry && !shouldRetry(error)) { + throw new UnrecoverableError("Job failed with a nonretryable error.") + } + throw error + } + }, + } +} diff --git a/hub/server/integrations/bullmq/create-job-queue/index.ts b/hub/server/integrations/bullmq/create-job-queue/index.ts new file mode 100644 index 00000000..8c74645c --- /dev/null +++ b/hub/server/integrations/bullmq/create-job-queue/index.ts @@ -0,0 +1,25 @@ +import { createPostgresBackend, Queue } from "bullmq" +import { z } from "zod" + +import { getJobConnection } from "../get-job-connection/index.ts" +import type { JobConnection, JobQueue } from "../types.ts" + +export function createJobQueue({ name, databaseUrl, logger }: JobConnection): JobQueue { + z.enum(["on-demand", "scheduled"]).parse(name) + const queue: JobQueue = new Queue( + name, + { + connection: getJobConnection(databaseUrl), + defaultJobOptions: { + attempts: 1, + removeOnComplete: { age: 24 * 60 * 60, count: 1_000 }, + removeOnFail: { age: 7 * 24 * 60 * 60, count: 5_000 }, + }, + }, + createPostgresBackend, + ) + queue.on("error", (error) => + logger.error({ queue: name, errorType: error.name }, "Job queue connection failed"), + ) + return queue +} diff --git a/hub/server/integrations/bullmq/create-job-worker/index.db.test.ts b/hub/server/integrations/bullmq/create-job-worker/index.db.test.ts new file mode 100644 index 00000000..5416b627 --- /dev/null +++ b/hub/server/integrations/bullmq/create-job-worker/index.db.test.ts @@ -0,0 +1,290 @@ +import pino from "pino" +import { describe, expect, it, onTestFinished, vi } from "vitest" +import { z } from "zod" + +import { createJobProcessor } from "../create-job-processor" +import { createJobQueue } from "../create-job-queue" +import { enqueueJob } from "../enqueue-job" +import type { JobDefinition, JobProcessor, JobQueueName } from "../types" +import { upsertJobSchedule } from "../upsert-job-schedule" +import { createJobWorker } from "." + +function getDatabaseUrl() { + const url = process.env.DATABASE_URL + if (!url) throw new Error("The database test harness must provide DATABASE_URL.") + return url +} + +function openQueue(name: JobQueueName, databaseUrl = getDatabaseUrl()) { + const queue = createJobQueue({ + name, + databaseUrl, + logger: pino({ level: "silent" }), + }) + onTestFinished(() => queue.close()) + return queue +} + +async function openWorker( + name: JobQueueName, + processors: JobProcessor[], + databaseUrl = getDatabaseUrl(), + logger = pino({ level: "silent" }), +) { + const worker = createJobWorker({ name, databaseUrl, logger, processors }) + let running: Promise | undefined + onTestFinished(async () => { + await worker.close() + await running + }) + await worker.waitUntilReady() + return { + worker, + start: () => { + running = worker.run() + }, + } +} + +describe("job processing with PostgreSQL", () => { + it("uses the supplied worker database and processes heterogeneous typed handlers only after run", async () => { + const databaseUrl = getDatabaseUrl() + vi.stubEnv("DATABASE_URL", "postgresql://127.0.0.1:1/not-the-worker-database") + const queue = openQueue("on-demand", databaseUrl) + const numbers = { name: "double", schema: z.object({ value: z.number() }) } + const words = { name: "uppercase", schema: z.object({ value: z.string() }) } + const { worker, start } = await openWorker( + "on-demand", + [ + createJobProcessor(numbers, ({ value }) => Promise.resolve(value * 2)), + createJobProcessor(words, ({ value }) => Promise.resolve(value.toUpperCase())), + ], + databaseUrl, + ) + const numberJob = await enqueueJob(queue, numbers, { value: 21 }) + const wordJob = await enqueueJob(queue, words, { value: "ready" }) + + expect(worker.isRunning()).toBe(false) + expect(await numberJob.getState()).toBe("waiting") + start() + + await vi.waitFor( + async () => { + expect(await numberJob.getState()).toBe("completed") + expect(await wordJob.getState()).toBe("completed") + }, + { timeout: 5_000 }, + ) + expect((await queue.getJob(numberJob.id!))?.returnvalue).toBe(42) + expect((await queue.getJob(wordJob.id!))?.returnvalue).toBe("READY") + }, 15_000) + + it("rejects invalid producer data and permanently rejects invalid stored payloads and unknown names", async () => { + const queue = openQueue("on-demand") + const definition: JobDefinition = { + name: "validated", + schema: z.object({ count: z.number().int().positive() }), + } + const process = vi.fn(() => Promise.resolve("must not run")) + const { start } = await openWorker("on-demand", [createJobProcessor(definition, process)]) + + await expect(enqueueJob(queue, definition, { count: "invalid" })).rejects.toThrow() + expect(await queue.getJobCountByTypes("waiting", "delayed", "failed")).toBe(0) + const malformed = await queue.add(definition.name, { count: "invalid" }, { attempts: 3 }) + const unknown = await queue.add("unregistered", { count: 1 }, { attempts: 3 }) + start() + + await vi.waitFor( + async () => { + expect(await malformed.getState()).toBe("failed") + expect(await unknown.getState()).toBe("failed") + }, + { timeout: 5_000 }, + ) + expect((await queue.getJob(malformed.id!))?.attemptsMade).toBe(1) + expect((await queue.getJob(unknown.id!))?.attemptsMade).toBe(1) + expect(process).not.toHaveBeenCalled() + }, 15_000) + + it("retries ordinary errors but stops immediately when the handler marks an error permanent", async () => { + const queue = openQueue("on-demand") + const definition = { + name: "retryable", + schema: z.object({ label: z.string() }), + options: { attempts: 3, backoff: { type: "fixed", delay: 10 } }, + } + const retry = vi.fn(({ label }: { label: string }) => { + if (retry.mock.calls.length < 2) { + return Promise.reject(new Error("Temporary dependency failure")) + } + return Promise.resolve(label) + }) + const permanent = vi.fn(() => Promise.reject(new Error("private-error-marker"))) + const lines: string[] = [] + const logger = pino({ level: "info" }, { write: (line) => lines.push(line) }) + const { worker, start } = await openWorker( + "on-demand", + [ + createJobProcessor(definition, retry), + createJobProcessor({ ...definition, name: "permanent" }, permanent, () => false), + ], + getDatabaseUrl(), + logger, + ) + const retryJob = await enqueueJob(queue, definition, { label: "recovered" }) + const permanentJob = await enqueueJob( + queue, + { ...definition, name: "permanent" }, + { label: "private-payload-marker" }, + ) + start() + + await vi.waitFor( + async () => { + expect(await retryJob.getState()).toBe("completed") + expect(await permanentJob.getState()).toBe("failed") + }, + { timeout: 5_000 }, + ) + expect(retry).toHaveBeenCalledTimes(2) + expect(permanent).toHaveBeenCalledTimes(1) + expect((await queue.getJob(retryJob.id!))?.returnvalue).toBe("recovered") + expect((await queue.getJob(permanentJob.id!))?.attemptsMade).toBe(1) + expect(() => worker.emit("error", new Error("private-connection-marker"))).not.toThrow() + expect(lines.length).toBeGreaterThan(0) + expect(lines.join("")).not.toMatch( + /private-error-marker|private-payload-marker|private-connection-marker/, + ) + }, 15_000) + + it("keeps a stable schedule id while updating its payload, interval, and job defaults", async () => { + const queue = openQueue("scheduled") + const definition = { + name: "periodic", + schema: z.object({ version: z.number() }), + options: { attempts: 2, priority: 4, backoff: { type: "fixed", delay: 20 } }, + } + + await upsertJobSchedule(queue, { + id: "stable-schedule", + definition, + payload: { version: 1 }, + repeat: { every: 60_000 }, + }) + const updatedDefinition = { + ...definition, + name: "updated-periodic", + options: { attempts: 4, priority: 2, backoff: { type: "fixed", delay: 30 } }, + } + await upsertJobSchedule(queue, { + id: "stable-schedule", + definition: updatedDefinition, + payload: { version: 2 }, + repeat: { every: 120_000 }, + }) + + expect(await queue.getJobSchedulersCount()).toBe(1) + expect(await queue.getJobScheduler("stable-schedule")).toMatchObject({ + name: updatedDefinition.name, + every: 120_000, + template: { data: { version: 2 }, opts: updatedDefinition.options }, + }) + }, 15_000) + + it("actually executes recurring jobs rather than only storing schedule metadata", async () => { + const queue = openQueue("scheduled") + const definition = { + name: "recurring", + schema: z.object({ marker: z.string() }), + } + const process = vi.fn(({ marker }: { marker: string }) => Promise.resolve(marker)) + const { start } = await openWorker("scheduled", [createJobProcessor(definition, process)]) + await upsertJobSchedule(queue, { + id: "recurring-schedule", + definition, + payload: { marker: "tick" }, + repeat: { every: 100 }, + }) + start() + + await vi.waitFor(() => expect(process.mock.calls.length).toBeGreaterThanOrEqual(2), { + timeout: 5_000, + }) + expect(process.mock.calls.every(([payload]) => payload.marker === "tick")).toBe(true) + }, 15_000) + + it("processes on-demand work while the scheduled worker has an active job and a backlog", async () => { + const scheduled = openQueue("scheduled") + const onDemand = openQueue("on-demand") + const definition = { name: "isolated", schema: z.object({ value: z.number() }) } + let release: (() => void) | undefined + const gate = new Promise((resolve) => { + release = resolve + }) + const scheduledProcess = vi.fn(async () => { + await gate + return "released" + }) + const scheduledWorker = await openWorker("scheduled", [ + createJobProcessor(definition, scheduledProcess), + ]) + const onDemandWorker = await openWorker("on-demand", [ + createJobProcessor(definition, ({ value }) => Promise.resolve(value)), + ]) + const active = await enqueueJob(scheduled, definition, { value: 1 }) + const waiting = await enqueueJob(scheduled, definition, { value: 2 }) + scheduledWorker.start() + onDemandWorker.start() + + try { + await vi.waitFor(() => expect(scheduledProcess).toHaveBeenCalledTimes(1), { + timeout: 5_000, + }) + const urgent = await enqueueJob(onDemand, definition, { value: 3 }) + await vi.waitFor(async () => expect(await urgent.getState()).toBe("completed"), { + timeout: 5_000, + }) + + expect(await active.getState()).toBe("active") + expect(await waiting.getState()).toBe("waiting") + expect((await onDemand.getJob(urgent.id!))?.returnvalue).toBe(3) + } finally { + release?.() + } + }, 15_000) + + it("rejects queue write failures instead of returning a successful enqueue", async () => { + const queue = openQueue("on-demand") + const definition = { name: "unavailable", schema: z.object({ value: z.number() }) } + const failure = new Error("Queue unavailable") + vi.spyOn(queue, "add").mockRejectedValueOnce(failure) + + await expect(enqueueJob(queue, definition, { value: 1 })).rejects.toBe(failure) + }, 15_000) + + it("validates producer data and honors per-job overrides alongside definition defaults", async () => { + const queue = openQueue("on-demand") + const definition = { + name: "configured", + schema: z.object({ label: z.string().trim() }), + options: { attempts: 3, priority: 8, backoff: { type: "fixed", delay: 20 } }, + } + + const job = await enqueueJob( + queue, + definition, + { label: " normalized " }, + { jobId: "explicit-job-id", priority: 2, delay: 60_000 }, + ) + + expect(job.id).toBe("explicit-job-id") + expect(job.data).toEqual({ label: "normalized" }) + expect(job.opts).toMatchObject({ + attempts: 3, + backoff: definition.options.backoff, + priority: 2, + delay: 60_000, + }) + expect(await job.getState()).toBe("delayed") + }, 15_000) +}) diff --git a/hub/server/integrations/bullmq/create-job-worker/index.test.ts b/hub/server/integrations/bullmq/create-job-worker/index.test.ts new file mode 100644 index 00000000..15b82ca1 --- /dev/null +++ b/hub/server/integrations/bullmq/create-job-worker/index.test.ts @@ -0,0 +1,49 @@ +import { Socket } from "node:net" +import pino from "pino" +import { describe, expect, it, vi } from "vitest" +import { z } from "zod" + +import { createJobProcessor } from "../create-job-processor" +import { createJobWorker } from "." + +describe("createJobWorker()", () => { + it("rejects duplicate processor names before opening a database connection", () => { + const connect = vi.spyOn(Socket.prototype, "connect").mockImplementation(() => { + throw new Error("Invalid workers must not open a database connection.") + }) + const first = createJobProcessor( + { name: "duplicate", schema: z.object({ count: z.number() }) }, + ({ count }) => Promise.resolve(count), + ) + const second = createJobProcessor( + { name: "duplicate", schema: z.object({ label: z.string() }) }, + ({ label }) => Promise.resolve(label), + ) + + expect(() => + createJobWorker({ + name: "on-demand", + databaseUrl: "postgresql://127.0.0.1:1/unreachable", + logger: pino({ level: "silent" }), + processors: [first, second], + }), + ).toThrow() + expect(connect).not.toHaveBeenCalled() + }) + + it("imports the job entry points without configuration or network access", async () => { + vi.stubEnv("DATABASE_URL", undefined) + const connect = vi.spyOn(Socket.prototype, "connect").mockImplementation(() => { + throw new Error("Importing job modules must not open a connection.") + }) + + await Promise.all([ + import("../create-job-queue"), + import("../create-job-worker"), + import("../migrate-job-database"), + import("../upsert-job-schedule"), + ]) + + expect(connect).not.toHaveBeenCalled() + }) +}) diff --git a/hub/server/integrations/bullmq/create-job-worker/index.ts b/hub/server/integrations/bullmq/create-job-worker/index.ts new file mode 100644 index 00000000..c9ddd1e7 --- /dev/null +++ b/hub/server/integrations/bullmq/create-job-worker/index.ts @@ -0,0 +1,61 @@ +import { createPostgresBackend, UnrecoverableError, Worker } from "bullmq" +import type { JobProgress, PostgresConnectionOptions, PostgresQueueBackend } from "bullmq" +import { z } from "zod" + +import { getJobConnection } from "../get-job-connection/index.ts" +import type { JobConnection, JobProcessor } from "../types.ts" + +export function createJobWorker({ + name, + databaseUrl, + logger, + processors, + concurrency = 1, +}: JobConnection & { processors: JobProcessor[]; concurrency?: number }) { + z.enum(["on-demand", "scheduled"]).parse(name) + z.number().int().min(1).max(100).parse(concurrency) + const handlers = new Map(processors.map((processor) => [processor.name, processor])) + if (handlers.size !== processors.length) { + throw new Error("Job processor names must be unique within a worker.") + } + + const worker = new Worker< + unknown, + unknown, + string, + PostgresQueueBackend, + JobProgress, + PostgresConnectionOptions + >( + name, + async (job) => { + const processor = handlers.get(job.name) + if (!processor) throw new UnrecoverableError("Job name is not registered for this worker.") + return await processor.process(job.data) + }, + { connection: getJobConnection(databaseUrl), concurrency, autorun: false }, + createPostgresBackend, + ) + worker.on("error", (error) => + logger.error({ queue: name, errorType: error.name }, "Job worker connection failed"), + ) + worker.on("failed", (job, error) => { + logger.error( + { + queue: name, + jobId: job?.id, + jobName: job?.name, + attemptsMade: job?.attemptsMade, + errorType: error.name, + }, + "Job attempt failed", + ) + }) + worker.on("completed", (job) => { + logger.info( + { queue: name, jobId: job.id, jobName: job.name, attemptsMade: job.attemptsMade }, + "Job completed", + ) + }) + return worker +} diff --git a/hub/server/integrations/bullmq/enqueue-job/index.ts b/hub/server/integrations/bullmq/enqueue-job/index.ts new file mode 100644 index 00000000..7a01a146 --- /dev/null +++ b/hub/server/integrations/bullmq/enqueue-job/index.ts @@ -0,0 +1,14 @@ +import type { JobsOptions } from "bullmq" + +import { parseJobPayload } from "../parse-job-payload/index.ts" +import type { JobDefinition, JobQueue } from "../types.ts" + +export async function enqueueJob( + queue: JobQueue, + definition: JobDefinition, + payload: T, + options: Pick = {}, +) { + const data = parseJobPayload(definition, payload) + return await queue.add(definition.name, data, { ...definition.options, ...options }) +} diff --git a/hub/server/integrations/bullmq/get-job-connection/index.ts b/hub/server/integrations/bullmq/get-job-connection/index.ts new file mode 100644 index 00000000..4ca58a87 --- /dev/null +++ b/hub/server/integrations/bullmq/get-job-connection/index.ts @@ -0,0 +1,18 @@ +import { z } from "zod" + +const databaseUrlSchema = z.url().refine((value) => { + const url = new URL(value) + return ["postgres:", "postgresql:"].includes(url.protocol) && url.pathname.length > 1 +}, "Job connections require a PostgreSQL database URL.") + +/** Leave pool ownership with BullMQ, including its dedicated notification connection. */ +export function getJobConnection(databaseUrl: string) { + return { + connectionString: databaseUrlSchema.parse(databaseUrl), + max: 2, + connectionTimeoutMillis: 10_000, + statement_timeout: 30_000, + query_timeout: 30_000, + application_name: "replica-hub-jobs", + } +} diff --git a/hub/server/integrations/bullmq/migrate-job-database/index.ts b/hub/server/integrations/bullmq/migrate-job-database/index.ts new file mode 100644 index 00000000..4e780013 --- /dev/null +++ b/hub/server/integrations/bullmq/migrate-job-database/index.ts @@ -0,0 +1,14 @@ +import { runMigrations } from "bullmq" +import pg from "pg" + +import { getJobConnection } from "../get-job-connection/index.ts" + +export async function migrateJobDatabase(databaseUrl: string) { + const client = new pg.Client(getJobConnection(databaseUrl)) + try { + await client.connect() + return await runMigrations(client) + } finally { + await client.end() + } +} diff --git a/hub/server/integrations/bullmq/parse-job-payload/index.ts b/hub/server/integrations/bullmq/parse-job-payload/index.ts new file mode 100644 index 00000000..99703d37 --- /dev/null +++ b/hub/server/integrations/bullmq/parse-job-payload/index.ts @@ -0,0 +1,9 @@ +import { z } from "zod" + +import type { JobDefinition } from "../types.ts" + +export function parseJobPayload(definition: JobDefinition, payload: unknown) { + const data = definition.schema.parse(payload) + z.json().parse(data) + return data +} diff --git a/hub/server/integrations/bullmq/types.ts b/hub/server/integrations/bullmq/types.ts new file mode 100644 index 00000000..2c69209c --- /dev/null +++ b/hub/server/integrations/bullmq/types.ts @@ -0,0 +1,35 @@ +import type { JobsOptions, PostgresConnectionOptions, PostgresQueueBackend, Queue } from "bullmq" +import type { Logger } from "pino" +import type { z } from "zod" + +export type JobQueueName = "on-demand" | "scheduled" + +export interface JobConnection { + name: JobQueueName + databaseUrl: string + logger: Logger +} + +export interface JobDefinition { + name: string + schema: z.ZodType + options?: Pick +} + +export interface JobProcessor { + name: string + process: (payload: unknown) => Promise +} + +export type JobQueue = Queue< + unknown, + unknown, + string, + unknown, + unknown, + string, + PostgresQueueBackend, + PostgresConnectionOptions +> + +export type JobScheduleRepeat = { every: number } | { pattern: string; tz: string } diff --git a/hub/server/integrations/bullmq/upsert-job-schedule/index.db.test.ts b/hub/server/integrations/bullmq/upsert-job-schedule/index.db.test.ts new file mode 100644 index 00000000..02626e88 --- /dev/null +++ b/hub/server/integrations/bullmq/upsert-job-schedule/index.db.test.ts @@ -0,0 +1,63 @@ +import pino from "pino" +import { describe, expect, it, onTestFinished } from "vitest" +import { z } from "zod" + +import { createJobQueue } from "../create-job-queue" +import type { JobDefinition, JobScheduleRepeat } from "../types" +import { upsertJobSchedule } from "." + +describe("upsertJobSchedule()", () => { + it("rejects the on-demand queue and malformed schedules without storing jobs", async () => { + const databaseUrl = process.env.DATABASE_URL + if (!databaseUrl) throw new Error("The database test harness must provide DATABASE_URL.") + const logger = pino({ level: "silent" }) + const scheduled = createJobQueue({ name: "scheduled", databaseUrl, logger }) + const onDemand = createJobQueue({ name: "on-demand", databaseUrl, logger }) + onTestFinished(async () => { + await Promise.all([scheduled.close(), onDemand.close()]) + }) + const definition: JobDefinition = { + name: "validated-schedule", + schema: z.object({ count: z.number().int().positive() }), + } + + await expect( + upsertJobSchedule(onDemand, { + id: "wrong-queue", + definition, + payload: { count: 1 }, + repeat: { every: 100 }, + }), + ).rejects.toThrow() + const invalidRepeats: JobScheduleRepeat[] = [ + { every: 0 }, + { every: -1 }, + { every: 1.5 }, + { every: Number.NaN }, + { pattern: "not a cron expression", tz: "UTC" }, + { pattern: "0 * * * *", tz: "Not/A-Timezone" }, + ] + for (const repeat of invalidRepeats) { + await expect( + upsertJobSchedule(scheduled, { + id: "invalid-repeat", + definition, + payload: { count: 1 }, + repeat, + }), + ).rejects.toThrow() + } + await expect( + upsertJobSchedule(scheduled, { + id: "invalid-payload", + definition, + payload: { count: "invalid" }, + repeat: { every: 100 }, + }), + ).rejects.toThrow() + + expect(await scheduled.getJobSchedulersCount()).toBe(0) + expect(await onDemand.getJobSchedulersCount()).toBe(0) + expect(await scheduled.getJobCountByTypes("waiting", "delayed")).toBe(0) + }, 15_000) +}) diff --git a/hub/server/integrations/bullmq/upsert-job-schedule/index.ts b/hub/server/integrations/bullmq/upsert-job-schedule/index.ts new file mode 100644 index 00000000..37b42c51 --- /dev/null +++ b/hub/server/integrations/bullmq/upsert-job-schedule/index.ts @@ -0,0 +1,44 @@ +import { z } from "zod" + +import { parseJobPayload } from "../parse-job-payload/index.ts" +import type { JobDefinition, JobQueue, JobScheduleRepeat } from "../types.ts" + +const repeatSchema = z.union([ + z.object({ every: z.number().int().positive() }).strict(), + z + .object({ + pattern: z.string().min(1), + tz: z.string().refine((value) => { + try { + new Intl.DateTimeFormat("en", { timeZone: value }) + return true + } catch (error) { + if (error instanceof RangeError) return false + throw error + } + }, "Schedules require a valid timezone."), + }) + .strict(), +]) + +export async function upsertJobSchedule( + queue: JobQueue, + { + id, + definition, + payload, + repeat, + }: { id: string; definition: JobDefinition; payload: T; repeat: JobScheduleRepeat }, +) { + if (queue.name !== "scheduled") { + throw new Error("Recurring jobs must use the scheduled queue.") + } + z.string().min(1).parse(id) + const settings = repeatSchema.parse(repeat) + const data = parseJobPayload(definition, payload) + return await queue.upsertJobScheduler(id, settings, { + name: definition.name, + data, + opts: definition.options, + }) +} diff --git a/hub/server/setup-jobs/index.ts b/hub/server/setup-jobs/index.ts new file mode 100644 index 00000000..d910f46a --- /dev/null +++ b/hub/server/setup-jobs/index.ts @@ -0,0 +1,28 @@ +import "dotenv/config" +import { z } from "zod" + +import { createJobQueue } from "../integrations/bullmq/create-job-queue/index.ts" +import { migrateJobDatabase } from "../integrations/bullmq/migrate-job-database/index.ts" +import { upsertJobSchedule } from "../integrations/bullmq/upsert-job-schedule/index.ts" +import { logger } from "../integrations/pino/index.ts" +import { jobSchedules } from "../workers/configuration.ts" + +try { + const databaseUrl = z.string().min(1).parse(process.env.DATABASE_URL) + const schemaVersion = await migrateJobDatabase(databaseUrl) + const queue = createJobQueue({ name: "scheduled", databaseUrl, logger }) + try { + for (const schedule of jobSchedules) { + await upsertJobSchedule(queue, schedule) + } + } finally { + await queue.close() + } + logger.info({ schemaVersion, scheduleCount: jobSchedules.length }, "Job setup completed") +} catch (error) { + logger.fatal( + { errorType: error instanceof Error ? error.name : "UnknownError" }, + "Job setup failed; check database access, schema permissions, and schedule definitions", + ) + process.exit(1) +} diff --git a/hub/server/workers/configuration.ts b/hub/server/workers/configuration.ts new file mode 100644 index 00000000..d815c0b0 --- /dev/null +++ b/hub/server/workers/configuration.ts @@ -0,0 +1,19 @@ +import type { + JobDefinition, + JobProcessor, + JobQueueName, + JobScheduleRepeat, +} from "../integrations/bullmq/types.ts" + +// Product processors and schedules are added here when their domains are built. +export const jobProcessors: Record = { + "on-demand": [], + scheduled: [], +} + +export const jobSchedules: { + id: string + definition: JobDefinition + payload: unknown + repeat: JobScheduleRepeat +}[] = [] diff --git a/hub/server/workers/index.ts b/hub/server/workers/index.ts new file mode 100644 index 00000000..743cc450 --- /dev/null +++ b/hub/server/workers/index.ts @@ -0,0 +1,25 @@ +import "dotenv/config" +import { z } from "zod" + +import { logger } from "../integrations/pino/index.ts" +import { jobProcessors } from "./configuration.ts" +import { runJobWorker } from "./run-job-worker/index.ts" + +try { + const name = z.enum(["on-demand", "scheduled"]).parse(process.argv[2]) + const databaseUrl = z.string().min(1).parse(process.env.DATABASE_URL) + const variable = name === "on-demand" ? "ON_DEMAND_JOB_CONCURRENCY" : "SCHEDULED_JOB_CONCURRENCY" + const concurrency = z.coerce + .number() + .int() + .min(1) + .max(100) + .parse(process.env[variable] ?? 1) + await runJobWorker({ name, databaseUrl, concurrency, logger, processors: jobProcessors[name] }) +} catch (error) { + logger.fatal( + { errorType: error instanceof Error ? error.name : "UnknownError" }, + "Job worker failed; check database access, worker configuration, and shutdown time", + ) + process.exit(1) +} diff --git a/hub/server/workers/run-job-worker/index.test.ts b/hub/server/workers/run-job-worker/index.test.ts new file mode 100644 index 00000000..49d4ec69 --- /dev/null +++ b/hub/server/workers/run-job-worker/index.test.ts @@ -0,0 +1,186 @@ +import pino from "pino" +import { describe, expect, it, onTestFinished, vi } from "vitest" + +import { runJobWorker } from "." + +const { createJobWorker } = vi.hoisted(() => ({ createJobWorker: vi.fn() })) +vi.mock("../../integrations/bullmq/create-job-worker/index.ts", () => ({ + createJobWorker, +})) + +function deferred() { + let resolve!: () => void + const promise = new Promise((complete) => { + resolve = complete + }) + return { promise, resolve: () => resolve() } +} + +describe("runJobWorker()", () => { + it.each(["SIGINT", "SIGTERM"] as const)( + "drains the worker and removes both signal listeners after %s", + async (signal) => { + const run = deferred() + const close = deferred() + const worker = { + waitUntilReady: vi.fn().mockResolvedValue(undefined), + run: vi.fn().mockReturnValue(run.promise), + close: vi.fn().mockReturnValue(close.promise), + } + createJobWorker.mockReturnValue(worker) + const once = vi.spyOn(process, "once").mockReturnValue(process) + const removeListener = vi.spyOn(process, "removeListener") + const completed = vi.fn() + const task = runJobWorker({ + name: "on-demand", + databaseUrl: "postgresql://127.0.0.1:1/not-used", + logger: pino({ level: "silent" }), + processors: [], + concurrency: 2, + }).then(completed) + const stop = once.mock.calls.find(([event]) => event === signal)?.[1] + onTestFinished(async () => { + stop?.() + close.resolve() + run.resolve() + await task + }) + if (!stop) throw new Error("The worker must register its shutdown handler.") + await vi.waitFor(() => expect(worker.run).toHaveBeenCalledOnce()) + + stop() + await vi.waitFor(() => expect(worker.close).toHaveBeenCalledOnce()) + expect(completed).not.toHaveBeenCalled() + close.resolve() + await Promise.resolve() + expect(completed).not.toHaveBeenCalled() + run.resolve() + await task + + expect(removeListener).toHaveBeenCalledWith("SIGINT", stop) + expect(removeListener).toHaveBeenCalledWith("SIGTERM", stop) + }, + ) + + it("closes the worker and propagates startup failures without starting the run loop", async () => { + const failure = new Error("Startup failed") + const worker = { + waitUntilReady: vi.fn().mockRejectedValue(failure), + run: vi.fn(), + close: vi.fn().mockResolvedValue(undefined), + } + createJobWorker.mockReturnValue(worker) + const once = vi.spyOn(process, "once").mockReturnValue(process) + const removeListener = vi.spyOn(process, "removeListener") + + await expect( + runJobWorker({ + name: "scheduled", + databaseUrl: "postgresql://127.0.0.1:1/not-used", + logger: pino({ level: "silent" }), + processors: [], + }), + ).rejects.toBe(failure) + + expect(worker.run).not.toHaveBeenCalled() + expect(worker.close).toHaveBeenCalledOnce() + const stop = once.mock.calls.find(([event]) => event === "SIGINT")?.[1] + expect(removeListener).toHaveBeenCalledWith("SIGINT", stop) + expect(removeListener).toHaveBeenCalledWith("SIGTERM", stop) + }) + + it("stops during pending readiness without starting the worker or reporting it ready", async () => { + const ready = deferred() + const worker = { + waitUntilReady: vi.fn().mockReturnValue(ready.promise), + run: vi.fn().mockResolvedValue(undefined), + close: vi.fn().mockResolvedValue(undefined), + } + createJobWorker.mockReturnValue(worker) + const once = vi.spyOn(process, "once").mockReturnValue(process) + const lines: string[] = [] + const logger = pino({ level: "info" }, { write: (line) => lines.push(line) }) + const task = runJobWorker({ + name: "scheduled", + databaseUrl: "postgresql://127.0.0.1:1/not-used", + logger, + processors: [], + }) + const stop = once.mock.calls.find(([event]) => event === "SIGTERM")?.[1] + onTestFinished(async () => { + stop?.() + ready.resolve() + await task + }) + if (!stop) throw new Error("The worker must register its shutdown handler.") + + stop() + await vi.waitFor(() => expect(worker.close).toHaveBeenCalledOnce()) + await task + ready.resolve() + await ready.promise + + expect(worker.run).not.toHaveBeenCalled() + expect(lines.join("")).not.toContain("Job worker ready") + }) + + it("closes the worker and propagates failures from the run loop", async () => { + const failure = new Error("Worker run failed") + const worker = { + waitUntilReady: vi.fn().mockResolvedValue(undefined), + run: vi.fn().mockRejectedValue(failure), + close: vi.fn().mockResolvedValue(undefined), + } + createJobWorker.mockReturnValue(worker) + vi.spyOn(process, "once").mockReturnValue(process) + + await expect( + runJobWorker({ + name: "on-demand", + databaseUrl: "postgresql://127.0.0.1:1/not-used", + logger: pino({ level: "silent" }), + processors: [], + }), + ).rejects.toBe(failure) + expect(worker.close).toHaveBeenCalledOnce() + }) + + it.each(["close", "run"] as const)( + "rejects shutdowns when %s exceeds the configured deadline", + async (pendingPhase) => { + vi.useFakeTimers() + const run = deferred() + const close = deferred() + const worker = { + waitUntilReady: vi.fn().mockResolvedValue(undefined), + run: vi.fn().mockReturnValue(run.promise), + close: vi + .fn() + .mockReturnValue(pendingPhase === "close" ? close.promise : Promise.resolve()), + } + createJobWorker.mockReturnValue(worker) + const once = vi.spyOn(process, "once").mockReturnValue(process) + const task = runJobWorker({ + name: "scheduled", + databaseUrl: "postgresql://127.0.0.1:1/not-used", + logger: pino({ level: "silent" }), + processors: [], + shutdownTimeoutMs: 25, + }) + onTestFinished(async () => { + close.resolve() + run.resolve() + await task.catch(() => {}) + }) + await vi.waitFor(() => expect(worker.run).toHaveBeenCalledOnce()) + const stop = once.mock.calls.find(([event]) => event === "SIGTERM")?.[1] + if (!stop) throw new Error("The worker must register its shutdown handler.") + + const rejected = expect(task).rejects.toThrow(/deadline/i) + stop() + await vi.advanceTimersByTimeAsync(25) + await rejected + expect(worker.close).toHaveBeenCalledOnce() + }, + ) +}) diff --git a/hub/server/workers/run-job-worker/index.ts b/hub/server/workers/run-job-worker/index.ts new file mode 100644 index 00000000..c5a91dae --- /dev/null +++ b/hub/server/workers/run-job-worker/index.ts @@ -0,0 +1,57 @@ +import { z } from "zod" + +import { createJobWorker } from "../../integrations/bullmq/create-job-worker/index.ts" +import type { JobConnection, JobProcessor } from "../../integrations/bullmq/types.ts" + +export async function runJobWorker({ + shutdownTimeoutMs = 30_000, + ...options +}: JobConnection & { + processors: JobProcessor[] + concurrency?: number + shutdownTimeoutMs?: number +}) { + z.number().int().positive().parse(shutdownTimeoutMs) + const worker = createJobWorker(options) + let stopped = false + let resolveStop: () => void + function stop() { + stopped = true + resolveStop() + } + const stopping = new Promise((resolve) => { + resolveStop = resolve + }) + process.once("SIGINT", stop) + process.once("SIGTERM", stop) + + let running: Promise | undefined + try { + await Promise.race([worker.waitUntilReady(), stopping]) + if (stopped) return + running = worker.run() + options.logger.info({ queue: options.name }, "Job worker ready") + await Promise.race([running, stopping]) + } finally { + process.removeListener("SIGINT", stop) + process.removeListener("SIGTERM", stop) + options.logger.info({ queue: options.name }, "Job worker stopping") + let timer: ReturnType | undefined + async function drain() { + await worker.close() + await running + } + try { + await Promise.race([ + drain(), + new Promise((_, reject) => { + timer = setTimeout(() => { + reject(new Error("Job worker exceeded its shutdown deadline.")) + }, shutdownTimeoutMs) + }), + ]) + } finally { + clearTimeout(timer) + } + } +} diff --git a/hub/test/database/global-setup/index.ts b/hub/test/database/global-setup/index.ts index cf60e93d..f6a7ddf6 100644 --- a/hub/test/database/global-setup/index.ts +++ b/hub/test/database/global-setup/index.ts @@ -7,6 +7,7 @@ import { promisify } from "node:util" import pg from "pg" import type { TestProject } from "vitest/node" +import { migrateJobDatabase } from "../../../server/integrations/bullmq/migrate-job-database/index.ts" import { getWorkerDatabase } from "../get-worker-database/index.ts" export default async function setup(project: TestProject) { @@ -65,6 +66,7 @@ export default async function setup(project: TestProject) { timeout: 60_000, }, ) + await migrateJobDatabase(templateUrl.toString()) for (let worker = 1; worker <= database.workers; worker++) { const name = `${database.prefix}_${worker}` diff --git a/hub/test/database/reset-database/index.db.test.ts b/hub/test/database/reset-database/index.db.test.ts new file mode 100644 index 00000000..6d59cea5 --- /dev/null +++ b/hub/test/database/reset-database/index.db.test.ts @@ -0,0 +1,121 @@ +import pg from "pg" +import pino from "pino" +import { describe, expect, it, onTestFinished, vi } from "vitest" +import { z } from "zod" + +import { createJobProcessor } from "../../../server/integrations/bullmq/create-job-processor" +import { createJobQueue } from "../../../server/integrations/bullmq/create-job-queue" +import { createJobWorker } from "../../../server/integrations/bullmq/create-job-worker" +import { enqueueJob } from "../../../server/integrations/bullmq/enqueue-job" +import { migrateJobDatabase } from "../../../server/integrations/bullmq/migrate-job-database" +import { upsertJobSchedule } from "../../../server/integrations/bullmq/upsert-job-schedule" +import { resetDatabase } from "." + +describe("resetDatabase() with job storage", () => { + it("preserves migration ledgers across repeated resets while clearing jobs, schedules, and application rows", async () => { + const databaseUrl = process.env.DATABASE_URL + if (!databaseUrl) throw new Error("The database test harness must provide DATABASE_URL.") + const client = new pg.Client({ connectionString: databaseUrl }) + await client.connect() + let createdLedger = false + let createdProbe = false + let insertedMigration = false + onTestFinished(async () => { + try { + if (createdProbe) await client.query("DROP TABLE public.jobs_reset_probe") + if (insertedMigration) { + await client.query( + `DELETE FROM public."_prisma_migrations" WHERE id = 'jobs-reset-regression'`, + ) + } + if (createdLedger) await client.query(`DROP TABLE public."_prisma_migrations"`) + } finally { + await client.end() + } + }) + const existing = await client.query<{ ledger: string | null }>( + `SELECT to_regclass('public._prisma_migrations')::text AS ledger`, + ) + if (!existing.rows[0]?.ledger) { + await client.query(` + CREATE TABLE public."_prisma_migrations" ( + id text PRIMARY KEY, + checksum text NOT NULL, + migration_name text NOT NULL, + started_at timestamptz NOT NULL DEFAULT now(), + finished_at timestamptz, + rolled_back_at timestamptz, + logs text, + applied_steps_count integer NOT NULL DEFAULT 0 + ) + `) + createdLedger = true + } + await client.query( + `INSERT INTO public."_prisma_migrations" (id, checksum, migration_name) + VALUES ('jobs-reset-regression', 'fixture-checksum', 'jobs-reset-regression')`, + ) + insertedMigration = true + await client.query("CREATE TABLE public.jobs_reset_probe (value text NOT NULL)") + createdProbe = true + const initialMigration = await migrateJobDatabase(databaseUrl) + expect(initialMigration).toBeGreaterThan(0) + const bullmqLedger = await client.query("SELECT * FROM bullmq.migration") + const prismaLedger = await client.query('SELECT * FROM public."_prisma_migrations"') + expect(bullmqLedger.rows.length).toBeGreaterThan(0) + + const logger = pino({ level: "silent" }) + const definition = { name: "after-reset", schema: z.object({ cycle: z.number() }) } + for (const cycle of [1, 2]) { + await client.query("INSERT INTO public.jobs_reset_probe (value) VALUES ('clear me')") + const scheduled = createJobQueue({ name: "scheduled", databaseUrl, logger }) + const onDemand = createJobQueue({ name: "on-demand", databaseUrl, logger }) + try { + await enqueueJob(onDemand, definition, { cycle }) + await upsertJobSchedule(scheduled, { + id: "reset-schedule", + definition, + payload: { cycle }, + repeat: { every: 60_000 }, + }) + } finally { + await Promise.all([scheduled.close(), onDemand.close()]) + } + + await resetDatabase(databaseUrl) + + expect((await client.query("SELECT * FROM bullmq.migration")).rows).toEqual(bullmqLedger.rows) + expect((await client.query('SELECT * FROM public."_prisma_migrations"')).rows).toEqual( + prismaLedger.rows, + ) + expect((await client.query("SELECT * FROM public.jobs_reset_probe")).rows).toEqual([]) + expect(await migrateJobDatabase(databaseUrl)).toBe(initialMigration) + + const freshScheduled = createJobQueue({ name: "scheduled", databaseUrl, logger }) + const freshOnDemand = createJobQueue({ name: "on-demand", databaseUrl, logger }) + const worker = createJobWorker({ + name: "on-demand", + databaseUrl, + logger, + processors: [createJobProcessor(definition, ({ cycle }) => Promise.resolve(cycle))], + }) + let running: Promise | undefined + try { + expect(await freshScheduled.getJobSchedulersCount()).toBe(0) + expect(await freshScheduled.getJobCountByTypes("waiting", "delayed")).toBe(0) + expect(await freshOnDemand.getJobCountByTypes("waiting", "delayed")).toBe(0) + await worker.waitUntilReady() + running = worker.run() + const job = await enqueueJob(freshOnDemand, definition, { cycle }) + await vi.waitFor(async () => expect(await job.getState()).toBe("completed"), { + timeout: 5_000, + }) + expect((await freshOnDemand.getJob(job.id!))?.returnvalue).toBe(cycle) + } finally { + await worker.close() + await running + await Promise.all([freshScheduled.close(), freshOnDemand.close()]) + } + } + }, 30_000) +}) diff --git a/hub/test/database/reset-database/index.ts b/hub/test/database/reset-database/index.ts index dfc51cf6..9860f176 100644 --- a/hub/test/database/reset-database/index.ts +++ b/hub/test/database/reset-database/index.ts @@ -25,7 +25,7 @@ export async function resetDatabase(databaseUrl: string) { throw new Error("Connected database does not match the expected test worker database.") } const tables = await client.query<{ schemaname: string; tablename: string }>( - "SELECT schemaname, tablename FROM pg_tables WHERE schemaname <> 'information_schema' AND schemaname !~ '^pg_' AND tablename <> '_prisma_migrations'", + "SELECT schemaname, tablename FROM pg_tables WHERE schemaname <> 'information_schema' AND schemaname !~ '^pg_' AND NOT (schemaname = 'public' AND tablename = '_prisma_migrations') AND NOT (schemaname = 'bullmq' AND tablename = 'migration')", ) if (tables.rows.length) { const names = tables.rows.map( From 2f11a568ef6303f4e2ec65e3b2ee3732881e40f0 Mon Sep 17 00:00:00 2001 From: Copilot <223556219+Copilot@users.noreply.github.com> Date: Wed, 28 Oct 2026 18:41:30 +0000 Subject: [PATCH 2/3] refactor(hub): share job test lifecycle helpers Extract worker database URL access and end-of-test queue disposal while retaining explicit reset-test resource ownership. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Co-authored-by: Nishiki <1121087+nshki@users.noreply.github.com> --- hub/docs/testing-and-operations.md | 2 ++ .../bullmq/create-job-worker/index.db.test.ts | 19 ++------------- .../upsert-job-schedule/index.db.test.ts | 15 ++++-------- .../database/get-database-url/index.test.ts | 23 +++++++++++++++++++ hub/test/database/get-database-url/index.ts | 5 ++++ .../database/reset-database/index.db.test.ts | 4 ++-- hub/test/jobs/open-queue/index.ts | 16 +++++++++++++ 7 files changed, 54 insertions(+), 30 deletions(-) create mode 100644 hub/test/database/get-database-url/index.test.ts create mode 100644 hub/test/database/get-database-url/index.ts create mode 100644 hub/test/jobs/open-queue/index.ts diff --git a/hub/docs/testing-and-operations.md b/hub/docs/testing-and-operations.md index 8babdf32..6d23554f 100644 --- a/hub/docs/testing-and-operations.md +++ b/hub/docs/testing-and-operations.md @@ -70,6 +70,8 @@ Hub CI runs database tests against PostgreSQL on every supported Node major. Onl Job tests use each worker's resolved `DATABASE_URL`, never `TEST_DATABASE_URL`. Close their consumers, queues, and owned connections before the next case's reset; clearing the module cache does not stop running jobs. See [Background jobs](background-jobs.md#tests) for job-specific lifecycle rules. +Use `test/database/get-database-url` to read the worker URL and `test/jobs/open-queue` for a queue that closes at test completion. Tests that reset data mid-case must create and close their queues explicitly before resetting. + ## Shared code The extension and build action still use Hub's Controls and canvas code. Keep their checks working: diff --git a/hub/server/integrations/bullmq/create-job-worker/index.db.test.ts b/hub/server/integrations/bullmq/create-job-worker/index.db.test.ts index 5416b627..96540a44 100644 --- a/hub/server/integrations/bullmq/create-job-worker/index.db.test.ts +++ b/hub/server/integrations/bullmq/create-job-worker/index.db.test.ts @@ -2,29 +2,14 @@ import pino from "pino" import { describe, expect, it, onTestFinished, vi } from "vitest" import { z } from "zod" +import { getDatabaseUrl } from "../../../../test/database/get-database-url" +import { openQueue } from "../../../../test/jobs/open-queue" import { createJobProcessor } from "../create-job-processor" -import { createJobQueue } from "../create-job-queue" import { enqueueJob } from "../enqueue-job" import type { JobDefinition, JobProcessor, JobQueueName } from "../types" import { upsertJobSchedule } from "../upsert-job-schedule" import { createJobWorker } from "." -function getDatabaseUrl() { - const url = process.env.DATABASE_URL - if (!url) throw new Error("The database test harness must provide DATABASE_URL.") - return url -} - -function openQueue(name: JobQueueName, databaseUrl = getDatabaseUrl()) { - const queue = createJobQueue({ - name, - databaseUrl, - logger: pino({ level: "silent" }), - }) - onTestFinished(() => queue.close()) - return queue -} - async function openWorker( name: JobQueueName, processors: JobProcessor[], diff --git a/hub/server/integrations/bullmq/upsert-job-schedule/index.db.test.ts b/hub/server/integrations/bullmq/upsert-job-schedule/index.db.test.ts index 02626e88..c2d0cb6d 100644 --- a/hub/server/integrations/bullmq/upsert-job-schedule/index.db.test.ts +++ b/hub/server/integrations/bullmq/upsert-job-schedule/index.db.test.ts @@ -1,21 +1,14 @@ -import pino from "pino" -import { describe, expect, it, onTestFinished } from "vitest" +import { describe, expect, it } from "vitest" import { z } from "zod" -import { createJobQueue } from "../create-job-queue" +import { openQueue } from "../../../../test/jobs/open-queue" import type { JobDefinition, JobScheduleRepeat } from "../types" import { upsertJobSchedule } from "." describe("upsertJobSchedule()", () => { it("rejects the on-demand queue and malformed schedules without storing jobs", async () => { - const databaseUrl = process.env.DATABASE_URL - if (!databaseUrl) throw new Error("The database test harness must provide DATABASE_URL.") - const logger = pino({ level: "silent" }) - const scheduled = createJobQueue({ name: "scheduled", databaseUrl, logger }) - const onDemand = createJobQueue({ name: "on-demand", databaseUrl, logger }) - onTestFinished(async () => { - await Promise.all([scheduled.close(), onDemand.close()]) - }) + const scheduled = openQueue("scheduled") + const onDemand = openQueue("on-demand") const definition: JobDefinition = { name: "validated-schedule", schema: z.object({ count: z.number().int().positive() }), diff --git a/hub/test/database/get-database-url/index.test.ts b/hub/test/database/get-database-url/index.test.ts new file mode 100644 index 00000000..7031687a --- /dev/null +++ b/hub/test/database/get-database-url/index.test.ts @@ -0,0 +1,23 @@ +import { describe, expect, it, vi } from "vitest" + +import { getDatabaseUrl } from "." + +describe("getDatabaseUrl()", () => { + it("returns the current worker URL rather than the administrative URL", () => { + const url = "postgresql://127.0.0.1/replica_test_0123456789abcdef_1" + vi.stubEnv("DATABASE_URL", url) + vi.stubEnv("TEST_DATABASE_URL", "postgresql://127.0.0.1/postgres") + + expect(getDatabaseUrl()).toBe(url) + }) + + it.each([undefined, ""])( + "rejects a missing worker URL without an administrative fallback: %s", + (url) => { + vi.stubEnv("DATABASE_URL", url) + vi.stubEnv("TEST_DATABASE_URL", "postgresql://127.0.0.1/postgres") + + expect(() => getDatabaseUrl()).toThrow("The database test harness must provide DATABASE_URL.") + }, + ) +}) diff --git a/hub/test/database/get-database-url/index.ts b/hub/test/database/get-database-url/index.ts new file mode 100644 index 00000000..7f1b75fd --- /dev/null +++ b/hub/test/database/get-database-url/index.ts @@ -0,0 +1,5 @@ +export function getDatabaseUrl() { + const url = process.env.DATABASE_URL + if (!url) throw new Error("The database test harness must provide DATABASE_URL.") + return url +} diff --git a/hub/test/database/reset-database/index.db.test.ts b/hub/test/database/reset-database/index.db.test.ts index 6d59cea5..c08b2eab 100644 --- a/hub/test/database/reset-database/index.db.test.ts +++ b/hub/test/database/reset-database/index.db.test.ts @@ -9,12 +9,12 @@ import { createJobWorker } from "../../../server/integrations/bullmq/create-job- import { enqueueJob } from "../../../server/integrations/bullmq/enqueue-job" import { migrateJobDatabase } from "../../../server/integrations/bullmq/migrate-job-database" import { upsertJobSchedule } from "../../../server/integrations/bullmq/upsert-job-schedule" +import { getDatabaseUrl } from "../get-database-url" import { resetDatabase } from "." describe("resetDatabase() with job storage", () => { it("preserves migration ledgers across repeated resets while clearing jobs, schedules, and application rows", async () => { - const databaseUrl = process.env.DATABASE_URL - if (!databaseUrl) throw new Error("The database test harness must provide DATABASE_URL.") + const databaseUrl = getDatabaseUrl() const client = new pg.Client({ connectionString: databaseUrl }) await client.connect() let createdLedger = false diff --git a/hub/test/jobs/open-queue/index.ts b/hub/test/jobs/open-queue/index.ts new file mode 100644 index 00000000..3113045e --- /dev/null +++ b/hub/test/jobs/open-queue/index.ts @@ -0,0 +1,16 @@ +import pino from "pino" +import { onTestFinished } from "vitest" + +import { createJobQueue } from "../../../server/integrations/bullmq/create-job-queue/index.ts" +import type { JobQueueName } from "../../../server/integrations/bullmq/types.ts" +import { getDatabaseUrl } from "../../database/get-database-url/index.ts" + +export function openQueue(name: JobQueueName, databaseUrl = getDatabaseUrl()) { + const queue = createJobQueue({ + name, + databaseUrl, + logger: pino({ level: "silent" }), + }) + onTestFinished(() => queue.close()) + return queue +} From 8f37334314b6a1f31e2ef245843325fcd2524d0a Mon Sep 17 00:00:00 2001 From: Copilot <223556219+Copilot@users.noreply.github.com> Date: Wed, 28 Oct 2026 18:49:24 +0000 Subject: [PATCH 3/3] test(hub): focus jobs coverage on critical behavior Remove low-value schema and helper edge cases, import-only checks, and mocked enqueue failures. Retain real persistence, schedule identity, isolation, retries, and shutdown safeguards. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Co-authored-by: Nishiki <1121087+nshki@users.noreply.github.com> --- .../bullmq/create-job-processor/index.test.ts | 23 ------------ .../bullmq/create-job-worker/index.db.test.ts | 9 ----- .../bullmq/create-job-worker/index.test.ts | 16 --------- .../upsert-job-schedule/index.db.test.ts | 36 ++----------------- .../database/get-database-url/index.test.ts | 23 ------------ 5 files changed, 3 insertions(+), 104 deletions(-) delete mode 100644 hub/server/integrations/bullmq/create-job-processor/index.test.ts delete mode 100644 hub/test/database/get-database-url/index.test.ts diff --git a/hub/server/integrations/bullmq/create-job-processor/index.test.ts b/hub/server/integrations/bullmq/create-job-processor/index.test.ts deleted file mode 100644 index 73096540..00000000 --- a/hub/server/integrations/bullmq/create-job-processor/index.test.ts +++ /dev/null @@ -1,23 +0,0 @@ -import { describe, expect, it, vi } from "vitest" -import { z } from "zod" - -import { createJobProcessor } from "." - -describe("createJobProcessor()", () => { - it("preserves unexpected schema errors as retryable failures", async () => { - const failure = new Error("Schema dependency failure") - const process = vi.fn(() => Promise.resolve("must not run")) - const processor = createJobProcessor( - { - name: "schema-failure", - schema: z.string().transform(() => { - throw failure - }), - }, - process, - ) - - await expect(processor.process("valid input")).rejects.toBe(failure) - expect(process).not.toHaveBeenCalled() - }) -}) diff --git a/hub/server/integrations/bullmq/create-job-worker/index.db.test.ts b/hub/server/integrations/bullmq/create-job-worker/index.db.test.ts index 96540a44..c119ed51 100644 --- a/hub/server/integrations/bullmq/create-job-worker/index.db.test.ts +++ b/hub/server/integrations/bullmq/create-job-worker/index.db.test.ts @@ -238,15 +238,6 @@ describe("job processing with PostgreSQL", () => { } }, 15_000) - it("rejects queue write failures instead of returning a successful enqueue", async () => { - const queue = openQueue("on-demand") - const definition = { name: "unavailable", schema: z.object({ value: z.number() }) } - const failure = new Error("Queue unavailable") - vi.spyOn(queue, "add").mockRejectedValueOnce(failure) - - await expect(enqueueJob(queue, definition, { value: 1 })).rejects.toBe(failure) - }, 15_000) - it("validates producer data and honors per-job overrides alongside definition defaults", async () => { const queue = openQueue("on-demand") const definition = { diff --git a/hub/server/integrations/bullmq/create-job-worker/index.test.ts b/hub/server/integrations/bullmq/create-job-worker/index.test.ts index 15b82ca1..8d4280f2 100644 --- a/hub/server/integrations/bullmq/create-job-worker/index.test.ts +++ b/hub/server/integrations/bullmq/create-job-worker/index.test.ts @@ -30,20 +30,4 @@ describe("createJobWorker()", () => { ).toThrow() expect(connect).not.toHaveBeenCalled() }) - - it("imports the job entry points without configuration or network access", async () => { - vi.stubEnv("DATABASE_URL", undefined) - const connect = vi.spyOn(Socket.prototype, "connect").mockImplementation(() => { - throw new Error("Importing job modules must not open a connection.") - }) - - await Promise.all([ - import("../create-job-queue"), - import("../create-job-worker"), - import("../migrate-job-database"), - import("../upsert-job-schedule"), - ]) - - expect(connect).not.toHaveBeenCalled() - }) }) diff --git a/hub/server/integrations/bullmq/upsert-job-schedule/index.db.test.ts b/hub/server/integrations/bullmq/upsert-job-schedule/index.db.test.ts index c2d0cb6d..593ee030 100644 --- a/hub/server/integrations/bullmq/upsert-job-schedule/index.db.test.ts +++ b/hub/server/integrations/bullmq/upsert-job-schedule/index.db.test.ts @@ -2,14 +2,12 @@ import { describe, expect, it } from "vitest" import { z } from "zod" import { openQueue } from "../../../../test/jobs/open-queue" -import type { JobDefinition, JobScheduleRepeat } from "../types" import { upsertJobSchedule } from "." describe("upsertJobSchedule()", () => { - it("rejects the on-demand queue and malformed schedules without storing jobs", async () => { - const scheduled = openQueue("scheduled") + it("rejects recurring work on the on-demand queue without storing jobs", async () => { const onDemand = openQueue("on-demand") - const definition: JobDefinition = { + const definition = { name: "validated-schedule", schema: z.object({ count: z.number().int().positive() }), } @@ -22,35 +20,7 @@ describe("upsertJobSchedule()", () => { repeat: { every: 100 }, }), ).rejects.toThrow() - const invalidRepeats: JobScheduleRepeat[] = [ - { every: 0 }, - { every: -1 }, - { every: 1.5 }, - { every: Number.NaN }, - { pattern: "not a cron expression", tz: "UTC" }, - { pattern: "0 * * * *", tz: "Not/A-Timezone" }, - ] - for (const repeat of invalidRepeats) { - await expect( - upsertJobSchedule(scheduled, { - id: "invalid-repeat", - definition, - payload: { count: 1 }, - repeat, - }), - ).rejects.toThrow() - } - await expect( - upsertJobSchedule(scheduled, { - id: "invalid-payload", - definition, - payload: { count: "invalid" }, - repeat: { every: 100 }, - }), - ).rejects.toThrow() - - expect(await scheduled.getJobSchedulersCount()).toBe(0) expect(await onDemand.getJobSchedulersCount()).toBe(0) - expect(await scheduled.getJobCountByTypes("waiting", "delayed")).toBe(0) + expect(await onDemand.getJobCountByTypes("waiting", "delayed")).toBe(0) }, 15_000) }) diff --git a/hub/test/database/get-database-url/index.test.ts b/hub/test/database/get-database-url/index.test.ts deleted file mode 100644 index 7031687a..00000000 --- a/hub/test/database/get-database-url/index.test.ts +++ /dev/null @@ -1,23 +0,0 @@ -import { describe, expect, it, vi } from "vitest" - -import { getDatabaseUrl } from "." - -describe("getDatabaseUrl()", () => { - it("returns the current worker URL rather than the administrative URL", () => { - const url = "postgresql://127.0.0.1/replica_test_0123456789abcdef_1" - vi.stubEnv("DATABASE_URL", url) - vi.stubEnv("TEST_DATABASE_URL", "postgresql://127.0.0.1/postgres") - - expect(getDatabaseUrl()).toBe(url) - }) - - it.each([undefined, ""])( - "rejects a missing worker URL without an administrative fallback: %s", - (url) => { - vi.stubEnv("DATABASE_URL", url) - vi.stubEnv("TEST_DATABASE_URL", "postgresql://127.0.0.1/postgres") - - expect(() => getDatabaseUrl()).toThrow("The database test harness must provide DATABASE_URL.") - }, - ) -})