diff --git a/.rulesync/commands/scaffold-plugin.md b/.rulesync/commands/scaffold-plugin.md index c220645d6..b1807a7ee 100644 --- a/.rulesync/commands/scaffold-plugin.md +++ b/.rulesync/commands/scaffold-plugin.md @@ -6,7 +6,7 @@ description: 'Generate a new overlay extension plugin under extensions//. Run `pnpm gen plugin $ARGUMENTS` in the repo root. Variants: `pnpm gen adapter ` for a vendor-adapter swap; `pnpm gen job-worker ` for a background-job worker. -The generator creates `extensions//plugin.ts` and registers it in `extensions.config.ts`. Then implement `register(ctx)`: +The generator creates `extensions//plugin.ts` and registers it in `extensions.config.ts`. Then implement `register(ctx)`. It only declares: a timer, connection, file or network read, process listener, global or `process.env` write goes in a `provide()` or router factory, closed with `c.onDispose()`, because a failed `SERVICE_MANIFEST` boot registers plugins into a container it then drops (ADR-0049). - `ctx.provide(TOKEN, factory)` - bind/override an adapter or service. Last registration wins, so an overlay loaded after the default-binding module replaces the default - keep the order in `extensions.config.ts` intentional. Sealed compliance tokens (`@openora/core/compliance`) are rejected at compile time + runtime. - `ctx.routers.add(namespace, (c) => router)` - mount oRPC routes. diff --git a/.rulesync/rules/messaging-and-microservices.md b/.rulesync/rules/messaging-and-microservices.md index 933c99080..f3d047f9b 100644 --- a/.rulesync/rules/messaging-and-microservices.md +++ b/.rulesync/rules/messaging-and-microservices.md @@ -37,5 +37,7 @@ Use the seams and contracts in code; this rule owns channel selection and the sa - An event reporting a committed change on a money, compliance, KYC, RG or account-status path uses `emitInTransaction(tx, ...)` inside the same transaction (the outbox is always on). `emit()` is best-effort and only for informational, catalog and cache-invalidation events. Both throw on an invalid payload. - A throwing handler is redelivered and dead-lettered to `oss:dlq:` after 5 deliveries (ADR-0045); plugin handlers are deduped per consumer via `processed_events`, a bare `bus.on` without a `consumer` is not. - A Redis Streams deployment uses `SERVICE_NAME` as its durable consumer-group identity. Every independently deployed service sharing Redis needs a distinct name. A subscription that must run once per deployment, however many services load its plugin (the audit writer), passes `{ consumerGroup }` as the third argument of `ctx.events.on`; services of one deployment using the same group name compete for each event, and `createApp` scopes the name as `/` so two deployments sharing one Redis do not. +- `SERVICE_MANIFEST` picks the modules a process boots; `createApp` applies it to any plugin list. A manifest lists every module its modules `dependsOn` and every module that binds a port they require; infra overlays always load and their dependencies count. Boot logs each `optionalPorts` entry left unbound with what its module skips. A split fails closed on controls: it refuses to boot while a selected module's `controlPorts` entry (an RG limit, a geo check, an audit record) or `AdminGuard`'s permission resolver or security policy - checked on every manifest boot, which builds the guard before it serves - is unbound and the plugin list has the module that binds it. Whatever is missing fails the boot once, with one manifest line adding every module needed and theirs; with the shipped modules every split carries iam, identity, audit, mail, compliance, wallet, gaming, player-management, exchange-rate and chat. Without a manifest nothing changes. +- `JOB_WORKERS` picks the job queues a process consumes (default all; names, `prefix*`, `!` to exclude). To keep a long-running queue off the API replicas, run an extra full instance with the same `SERVICE_NAME` and `JOB_WORKERS=`, and give the replicas `JOB_WORKERS=!`. Processes sharing a `SERVICE_NAME` load the same modules, every queue needs a consumer somewhere, and a handler never assumes it runs where its job was enqueued. - The shipped Redis Streams and BullMQ drivers do not honor `orderingKey`; use an overlay when strict ordering is required. - `REALTIME_TRANSPORT` is Redis Pub/Sub (`RedisPubSubRealtimeTransport`, auto-bound on `REDIS_URL`, no in-process fallback - ADR-0031) - fire-and-forget UI push, not a durable event: a message published while nobody is subscribed is simply lost, which is why a realtime channel carries a change _signal_ the client refetches, never the state itself. Channels, signals, presence and the revocation channel are prefixed with `REALTIME_NAMESPACE`, which defaults to `SERVICE_NAME`. The two are opposite requirements: `SERVICE_NAME` must differ per service (consumer group), `REALTIME_NAMESPACE` must match across the services that push to the same clients, and must differ between deployments sharing one Redis. A service that sets `SERVICE_MANIFEST` must set both, or the boot fails. A manifest entry takes one flag, `id:no-routes`: the module still binds its ports, handles its events and runs its jobs, but its router is discarded, so it serves no HTTP route. Its router factory still runs, because several modules construct the service their events and jobs read, or schedule their jobs, inside it (the audit writer assigns `svcRef` there); a factory that guards the HTTP surface takes `options.mounted` and skips that guard when it is `false`. Use it for a module a kept module `dependsOn` but whose surface the service must not expose (identity's sign-in, whose country gate lives in the compliance module). A manifest that names identity must name audit too, since identity `dependsOn` it, or the boot fails on the missing dependency. An unknown flag, or the flag on an infra overlay or an unknown id, fails the boot. `RealtimePresence`/`getOnlineUserIds` are shared across replicas via `RedisPresenceStore`, a per-channel sorted set scored by each member's last heartbeat and read with a TTL cutoff, so a crashed replica's members self-expire instead of lingering forever. diff --git a/.rulesync/subagents/plugin-author.md b/.rulesync/subagents/plugin-author.md index c4f482829..3cde47249 100644 --- a/.rulesync/subagents/plugin-author.md +++ b/.rulesync/subagents/plugin-author.md @@ -40,6 +40,8 @@ ctx.mcp.tool(defineMcpTool({...}), (c) => async (input, run) => output); // agen ctx.actions.register(defineActionType({...}), (c) => ({ precondition, execute })); // action type: a change a model may only propose ``` +`register()` only declares into `ctx`. A timer, connection, file or network read, process listener, global or `process.env` write goes in a `provide()` or router factory, closed with `c.onDispose()`: a failed `SERVICE_MANIFEST` boot registers the plugins it has not loaded into a container it then drops, to learn which binds which port (ADR-0049). + Agent tools and action types are served only through the `MCP_KERNEL` token, which checks the IAM grant of the admin behind the run, validates input, returns only the allow-listed output keys and declared error codes, and audits every tool call and action execution. A shape it cannot serve (eg a tool input that is not a top-level `z.object`, or an unbounded string) fails boot with the fix. `execute` must perform nothing on a replay while its effect still holds, and answer `already_applied`. The one-argument `ctx.mcp.tool(definition)` still registers but the kernel does not serve it. Contract types: `packages/core/src/contracts/adapters/mcp.ts`; why: ADR-0042. No decorators, no controllers - `{ id, dependsOn, register } satisfies Plugin` wired by the functional Container (ADR-0009). DB tables: a `pgTable` in the plugin's own `schema/index.ts`, then `pnpm regen`. diff --git a/AGENTS.md b/AGENTS.md index 3195bc0f9..7e6cf123a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -133,6 +133,8 @@ Use the seams and contracts in code; this rule owns channel selection and the sa - An event reporting a committed change on a money, compliance, KYC, RG or account-status path uses `emitInTransaction(tx, ...)` inside the same transaction (the outbox is always on). `emit()` is best-effort and only for informational, catalog and cache-invalidation events. Both throw on an invalid payload. - A throwing handler is redelivered and dead-lettered to `oss:dlq:` after 5 deliveries (ADR-0045); plugin handlers are deduped per consumer via `processed_events`, a bare `bus.on` without a `consumer` is not. - A Redis Streams deployment uses `SERVICE_NAME` as its durable consumer-group identity. Every independently deployed service sharing Redis needs a distinct name. A subscription that must run once per deployment, however many services load its plugin (the audit writer), passes `{ consumerGroup }` as the third argument of `ctx.events.on`; services of one deployment using the same group name compete for each event, and `createApp` scopes the name as `/` so two deployments sharing one Redis do not. +- `SERVICE_MANIFEST` picks the modules a process boots; `createApp` applies it to any plugin list. A manifest lists every module its modules `dependsOn` and every module that binds a port they require; infra overlays always load and their dependencies count. Boot logs each `optionalPorts` entry left unbound with what its module skips. A split fails closed on controls: it refuses to boot while a selected module's `controlPorts` entry (an RG limit, a geo check, an audit record) or `AdminGuard`'s permission resolver or security policy - checked on every manifest boot, which builds the guard before it serves - is unbound and the plugin list has the module that binds it. Whatever is missing fails the boot once, with one manifest line adding every module needed and theirs; with the shipped modules every split carries iam, identity, audit, mail, compliance, wallet, gaming, player-management, exchange-rate and chat. Without a manifest nothing changes. +- `JOB_WORKERS` picks the job queues a process consumes (default all; names, `prefix*`, `!` to exclude). To keep a long-running queue off the API replicas, run an extra full instance with the same `SERVICE_NAME` and `JOB_WORKERS=`, and give the replicas `JOB_WORKERS=!`. Processes sharing a `SERVICE_NAME` load the same modules, every queue needs a consumer somewhere, and a handler never assumes it runs where its job was enqueued. - The shipped Redis Streams and BullMQ drivers do not honor `orderingKey`; use an overlay when strict ordering is required. - `REALTIME_TRANSPORT` is Redis Pub/Sub (`RedisPubSubRealtimeTransport`, auto-bound on `REDIS_URL`, no in-process fallback - ADR-0031) - fire-and-forget UI push, not a durable event: a message published while nobody is subscribed is simply lost, which is why a realtime channel carries a change _signal_ the client refetches, never the state itself. Channels, signals, presence and the revocation channel are prefixed with `REALTIME_NAMESPACE`, which defaults to `SERVICE_NAME`. The two are opposite requirements: `SERVICE_NAME` must differ per service (consumer group), `REALTIME_NAMESPACE` must match across the services that push to the same clients, and must differ between deployments sharing one Redis. A service that sets `SERVICE_MANIFEST` must set both, or the boot fails. A manifest entry takes one flag, `id:no-routes`: the module still binds its ports, handles its events and runs its jobs, but its router is discarded, so it serves no HTTP route. Its router factory still runs, because several modules construct the service their events and jobs read, or schedule their jobs, inside it (the audit writer assigns `svcRef` there); a factory that guards the HTTP surface takes `options.mounted` and skips that guard when it is `false`. Use it for a module a kept module `dependsOn` but whose surface the service must not expose (identity's sign-in, whose country gate lives in the compliance module). A manifest that names identity must name audit too, since identity `dependsOn` it, or the boot fails on the missing dependency. An unknown flag, or the flag on an infra overlay or an unknown id, fails the boot. `RealtimePresence`/`getOnlineUserIds` are shared across replicas via `RedisPresenceStore`, a per-channel sorted set scored by each member's last heartbeat and read with a TTL cutoff, so a crashed replica's members self-expire instead of lingering forever. diff --git a/docs/adapters/error-tracking.md b/docs/adapters/error-tracking.md index 5be5dbb9c..eb8245708 100644 --- a/docs/adapters/error-tracking.md +++ b/docs/adapters/error-tracking.md @@ -63,19 +63,25 @@ export default { if (!dsn) { return; // self-disabling: no DSN -> platform stays logs-only } - Sentry.init({ dsn }); - ctx.provide(ERROR_TRACKING, () => ({ - captureException(error, context) { - Sentry.captureException(error, { - ...(context?.userId ? { user: { id: context.userId } } : {}), - tags: { ...context?.tags, ...(context?.traceId ? { trace_id: context.traceId } : {}) }, - ...(context?.extra ? { extra: context.extra } : {}), - }); - }, - })); + ctx.provide(ERROR_TRACKING, () => { + Sentry.init({ dsn }); + return { + captureException(error, context) { + Sentry.captureException(error, { + ...(context?.userId ? { user: { id: context.userId } } : {}), + tags: { ...context?.tags, ...(context?.traceId ? { trace_id: context.traceId } : {}) }, + ...(context?.extra ? { extra: context.extra } : {}), + }); + }, + }; + }); }, } as const satisfies Plugin; ``` +`Sentry.init` runs in the factory, which `createApp` resolves once for the container that serves, +not in `register()`: a failed `SERVICE_MANIFEST` boot registers plugins into a container it then +drops (ADR-0049). Initialising in a file preloaded with `node --import` works as well. + Register it last in `extensions.config.ts` (a `kind: 'infra'` overlay). A PostHog/Rollbar shop writes the equivalent overlay against the same port - core is unchanged. diff --git a/docs/adr/0014-job-queue-and-realtime-transport-seams.md b/docs/adr/0014-job-queue-and-realtime-transport-seams.md index d710e8f7b..a950538c2 100644 --- a/docs/adr/0014-job-queue-and-realtime-transport-seams.md +++ b/docs/adr/0014-job-queue-and-realtime-transport-seams.md @@ -7,6 +7,8 @@ > **Update (2026-07-25)**: [ADR-0039](./0039-tests-run-the-production-seams.md) deleted `InProcessJobQueue` - the test suite binds `BullMqJobQueue` too. `InProcessRealtimeTransport` stays, but as `createApp`'s production default rather than a test double, since core ships no realtime driver. The seam shapes are unchanged. > > **Update (2026-08-31)**: [ADR-0031](./0031-realtime-transport-sse-and-http-over-web-sockets.md) shipped `RedisPubSubRealtimeTransport` and deleted `InProcessRealtimeTransport` too - `REALTIME_TRANSPORT` now auto-binds on `REDIS_URL` exactly like `JOB_QUEUE`, with no in-process fallback either. Both seams from this ADR are now fully superseded on the production-default question; the seam shapes and the client-push-is-not-inter-module decision are still unchanged. +> +> **Update (2026-10-13)**: a worker still starts at boot against the resolved `JOB_QUEUE`, unless `JOB_WORKERS` leaves its queue to another process ([ADR-0049](./0049-per-process-job-worker-selection.md)). ## Context diff --git a/docs/adr/0017-extraction-readiness-manifest-outbox-command-ports.md b/docs/adr/0017-extraction-readiness-manifest-outbox-command-ports.md index 2ef137c99..3b5322f4a 100644 --- a/docs/adr/0017-extraction-readiness-manifest-outbox-command-ports.md +++ b/docs/adr/0017-extraction-readiness-manifest-outbox-command-ports.md @@ -4,6 +4,8 @@ **Status**: Accepted; implemented (manifest boot + outbox seam + `WALLET_COMMANDS` reference; outbox off by default). **Relates to**: ADR-0010 (event-driven broker direction), ADR-0014 (job-queue/realtime seams), ADR-0016 (event envelope). +> **Update (2026-10-13)**: `createApp` now applies `SERVICE_MANIFEST` to whatever plugin list it is given, so a host that passes a static list gets the same subset as one that calls `loadExtensions()`; until then such a host booted every module under the split service's own consumer group. `loadExtensions()` no longer filters, and `apps/api/src/extensions.ts` is long gone. A manifest must list every module its modules depend on and every module that binds a port they require - the dependencies of infra overlays included. It also fails closed on controls: a selected module's `controlPorts`, and `AdminGuard`'s permission resolver and security policy on every manifest boot, must be bound whenever the plugin list has the module that binds them. Whatever is missing, the boot fails once with one manifest line that adds it all ([ADR-0049](./0049-per-process-job-worker-selection.md)). `pnpm create:service` now sets `SERVICE_NAME` too. Which job queues a process consumes is chosen apart from its modules: [ADR-0049](./0049-per-process-job-worker-selection.md). + > **Update (2026-10-13)**: [ADR-0045](./0045-at-least-once-event-delivery.md) made the outbox always on - `OUTBOX_ENABLED` is gone, `createApp` always binds `OUTBOX` and starts the `OutboxRelay`, and money and compliance events are emitted with `emitInTransaction`. The "Opt-in" paragraph below no longer holds. The relay now claims rows atomically with a lease, and consumers dedup per handler through `processed_events` instead of on `eventId` alone. ## Context diff --git a/docs/adr/0033-analytics-as-a-read-only-domain.md b/docs/adr/0033-analytics-as-a-read-only-domain.md index efa8b8b10..2ae5da00d 100644 --- a/docs/adr/0033-analytics-as-a-read-only-domain.md +++ b/docs/adr/0033-analytics-as-a-read-only-domain.md @@ -3,6 +3,8 @@ **Date**: 2026-07-25 **Status**: Accepted +> **Update (2026-10-13)**: `SERVICE_MANIFEST=analytics` alone does not boot: the manifest must also list the modules analytics depends on (wallet, identity, profile, gaming) and theirs. The boot names every missing one ([ADR-0017](./0017-extraction-readiness-manifest-outbox-command-ports.md)). + ## Context A financial-analytics dashboard is requested (deposits/withdrawals by diff --git a/docs/adr/0036-rg-limit-enforcement-and-cooldown.md b/docs/adr/0036-rg-limit-enforcement-and-cooldown.md index 49526a888..2e686b8b2 100644 --- a/docs/adr/0036-rg-limit-enforcement-and-cooldown.md +++ b/docs/adr/0036-rg-limit-enforcement-and-cooldown.md @@ -8,6 +8,10 @@ refused server-side with a typed error. The paragraph below asserting the admin "still writes a limit outright in either direction" no longer holds; everything else in this ADR (the enforcement points, the deposit/wager gates, the player-path cool-down, and the concurrency reasoning) is unchanged. +**Update (2026-10-16)**: `RG_LIMITS` stays optional for its consumers in a full boot, but wallet +and gaming list it in `controlPorts`, so a `SERVICE_MANIFEST` boot that loads either refuses to +start without the compliance module while the plugin list has it: another process of the install +enforces the limits there ([ADR-0049](./0049-per-process-job-worker-selection.md)). > **Update (2026-10-15)**: RG spend now reads the wallet ledger for every product, and `loss` > stays unenforced for the reason in [ADR-0046](./0046-sportsbook-ledger-port.md). diff --git a/docs/adr/0049-per-process-job-worker-selection.md b/docs/adr/0049-per-process-job-worker-selection.md new file mode 100644 index 000000000..740493233 --- /dev/null +++ b/docs/adr/0049-per-process-job-worker-selection.md @@ -0,0 +1,128 @@ +# ADR-0049: Per-process job worker selection + +**Date**: 2026-10-13 +**Status**: Proposed +**Relates to**: ADR-0014 (job-queue seam), ADR-0017 (service manifest), ADR-0030 (`SERVICE_NAME`), ADR-0038 (routers before workers), ADR-0042 (MCP kernel). + +## Context + +`createApp` registers a worker for every queue a loaded module declares, so every process +consumes every queue. That is right for a homogeneous fleet of replicas, but not for a job that +runs for minutes and mostly waits on an outside service - an agent run waiting on a model, a +large export. Such jobs take a share of every API replica's capacity and stretch its deploys, +since a replica drains its workers before it stops. + +The service manifest cannot move them. A manifest service has its own `SERVICE_NAME`, so it +receives its own copy of every event, and a module pulled into it for its dependencies runs its +handlers a second time across the cluster. A job whose handler needs the full container cannot +live in a narrow service anyway: the MCP kernel (ADR-0042) is built in process from the tools and +action types of the modules that process loaded, so an agent run needs every module that +contributes them. Nor does a manifest service stay narrow: it fails closed on the controls the +rest of the install enforces (below). Every split loads iam and identity for AdminGuard, and with +them audit and mail; identity's controls pull in compliance, and compliance depends on wallet, +gaming, player-management and exchange-rate, player-management on chat. With the shipped modules +a split carries those ten before the module it was made for, which is close to the monolith. + +## Decision + +**`JOB_WORKERS` picks the queues a process consumes.** `createApp` reads it at boot and +registers a worker only for the queues it selects; the filter sits in `createApp`, so it holds +for any queue driver. Entries are separated by commas or spaces: + +| Value | Consumes | +| ------------------------- | ---------------------------------------- | +| unset, blank, `*`, `all` | every queue (the default, as before) | +| `none` | no queue | +| `reports.daily` | that queue | +| `reports.*` | every queue whose name starts `reports.` | +| `!pattern` | not the queues the pattern matches | +| only exclusions | every queue but the excluded ones | +| inclusions and exclusions | the included queues minus the excluded | + +A `*` anywhere but the end of a pattern, a bare `!`, `none` beside another entry, or `*` beside +a named queue fails the boot. So does an inclusion that matches no registered queue: a typo +would otherwise leave a worker process idle while its queue backs up. An exclusion that matches +nothing is allowed, because the module that registers the queue may not be installed. The boot +log lists the queues consumed and skipped. + +It is environment-only, like `SERVICE_MANIFEST` and `SERVICE_NAME`: which queues a process takes +is deployment topology, and one image runs as either kind of process. + +**Schedules are unchanged.** Every process still upserts its modules' schedules while building +routers (ADR-0038); the scheduler id dedupes them, and the selection only decides who consumes +the jobs they produce. + +**The deployment pattern is an extra full instance.** To keep `reports.*` off the API replicas, +run one more instance of the same image, with the same plugin list and the same `SERVICE_NAME`, +and `JOB_WORKERS=reports.*`; give the replicas `JOB_WORKERS=!reports.*`. Sharing the +`SERVICE_NAME` makes the worker compete with the replicas for events instead of receiving a copy, +and keeps it in the same realtime and presence namespace. It serves HTTP like any instance but +needs no ingress. + +**Add-ons follow the same rule.** The selection filters the workers registered through +`ctx.jobs.worker`, so an add-on moves its long-running jobs off the replicas only if it registers +them there, under a queue-name prefix of its own (`agents.*`, say) that one pattern selects. A +worker an add-on registers on `JOB_QUEUE` itself bypasses the selection and runs in every +process. + +**A manifest fails closed on controls.** Some ports back a money, compliance, authorization or +audit control and are read with `c.has()`. An install without the module that binds one has +nothing to enforce, so a full boot runs without it, as before. A split service is different: the +module runs in another process of the same install, and a split that skipped the control would +let through what that process refuses. A plugin declares such a port in `controlPorts`, with the +module that binds it, and a `SERVICE_MANIFEST` boot refuses to start while one is unbound and the +plugin list has that module. Declared today: `RG_LIMITS` (wallet, gaming), `GAME_GEO_CHECK` +(gaming), `GEO_CHECK_COMMANDS` (identity, bonus, notifications), `MCP_TOKEN_REVOCATION` (identity) +and `AUDIT_WRITER` (mail). A port whose absence already fails closed - no KYC policy means every +gated withdrawal needs KYC, no risk tags send auto-approval to a human - is not a control port. +`AdminGuard` holds its own ports to the same rule on every manifest boot, which builds the guard +before it serves, admin routes or not: without `ADMIN_PERMISSION_RESOLVER` (iam) it decides on the +static role table, so an admin whose role was revoked still acts, and without +`ADMIN_SECURITY_POLICY` (identity) it skips the 2FA enrolment and session checks. + +**A manifest is fixed in one edit.** A manifest boot fails once for everything the manifest leaves +out - a dependency, the module that binds a required port, the module behind a control - naming +each with what its module skips, and prints one manifest line that adds them all and what they +need in turn. Which module binds a port is learned only once the boot has already failed: what +the modules the boot registered bound is in its container, and every other plugin of the list +registers once, on its own, into a container that is then dropped. That is why `register()` +only declares into the registry - a timer, connection, file read, process listener, global or +`process.env` write there would run in that pass too - and a resource opens in a `provide()` or +router factory, which only the container that serves runs. A failed boot disposes its container, +closing what the boot had opened. + +### Alternatives considered + +- **A process role (`SERVICE_ROLE=api|worker`).** Fixes one topology and cannot say which + queues. +- **Role metadata on each worker registration.** A module cannot know how it is deployed. +- **A worker-only process without routers.** Router factories are boot hooks that workers rely + on (ADR-0038), and the kernel needs every module loaded. +- **A manifest service for the queue.** Its own `SERVICE_NAME` means its own copy of every event. +- **Pausing workers in the driver.** Driver-specific, and the decision belongs to the host. + +## Consequences + +**Positive:** + +- An operator can keep long-running jobs off the API replicas without code changes. +- Unset behaves exactly as before. + +**Negative / trade-offs:** + +- Every queue needs a consumer somewhere in the cluster; core cannot check that across + processes, so a queue excluded everywhere backs up silently. Queue-depth monitoring catches it. +- Processes that share a `SERVICE_NAME` must load the same modules. The streams driver acks an + event once per group, so a process without a handler for it loses the event for that group. +- When a manifest boot fails, every plugin of the list the boot had not registered registers + once, to learn what it binds. A `register()` that does more than declare - logs, reads + configuration and throws - does so in that pass. One that throws counts as binding what it + bound before the throw, and the error names it when a required port is left without a known + binder. + +**Neutral:** + +- Roll out the worker instance first, then the replicas with the exclusion; jobs enqueued in + between wait in Redis. +- `/health` does not report the selection: it is a public contract, and the boot log is the + record. diff --git a/docs/adr/0050-dependency-only-modules-provide-but-do-not-consume.md b/docs/adr/0050-dependency-only-modules-provide-but-do-not-consume.md new file mode 100644 index 000000000..591cf49fc --- /dev/null +++ b/docs/adr/0050-dependency-only-modules-provide-but-do-not-consume.md @@ -0,0 +1,58 @@ +# ADR-0050: Dependency-only modules provide but do not consume + +**Date**: 2026-10-13 +**Status**: Proposed +**Relates to**: ADR-0017 (service manifest), ADR-0030 (`SERVICE_NAME`), ADR-0049 (job worker selection). + +## Context + +A service manifest (ADR-0017) must list every module its modules depend on, and every module +that binds a port they require. Almost every module depends on `audit` or `identity`, so most +split services load them. A loaded module registers everything: its providers, but also its +event handlers, job workers, schedules, routers and MCP tools. Each split service has its own +`SERVICE_NAME`, so it receives its own copy of every event (ADR-0030). A module that two +services load only as a dependency therefore handles each event twice across the cluster - audit's +handlers, for one, would write each audit row once per service. Dedupe keyed on the consumer +does not help: the two copies run under different groups. + +Nothing ships split today, so nothing duplicates yet; extracting the first module would. + +## Decision + +Proposed, not implemented. + +**Separate the modules a process selects from the ones it only requires.** The selected modules +are the manifest's and the infra overlays; the required ones are the `dependsOn` closure and the +port providers. Every module still runs `register()`, so its providers bind and its services +work for the selected ones. `createApp` wires consumption only for registrations owned by a +selected module: event handlers, job workers, schedules and mounted routers. + +**The registry records the owner of each registration.** `ModuleRegistryImpl` already records it +for MCP tools and action types (`setOwner`); `events.on`, `jobs.worker` and `routers.add` would +carry it too. Router factories stay boot hooks that every module runs (ADR-0038); only the +mounting is filtered. Schedules move from router factories to a declarative, owned +`ctx.jobs.schedule`, so only a selected owner upserts them. + +**Required modules could then be pulled in automatically**, instead of failing the boot, since +loading one no longer changes what the service consumes. + +### Alternatives considered + +- **A consumer group per owning module**, so a module's handlers run once per cluster wherever it + is loaded. Infra overlays deliberately run per service, and it renames every durable group. + +## Consequences + +**Positive:** + +- No duplicate side effects across split services, and shorter manifests. + +**Negative / trade-offs:** + +- A breaking change for any deployment that relies on a dependency's handlers running in a + service that does not select it. +- Moving schedules out of router factories touches every module that schedules. + +**Neutral:** + +- A monolith is unaffected: every module is selected. diff --git a/docs/guides/downstream-consumer.md b/docs/guides/downstream-consumer.md index 2448b763a..ace5aa068 100644 --- a/docs/guides/downstream-consumer.md +++ b/docs/guides/downstream-consumer.md @@ -75,6 +75,35 @@ await listen(); Downstream consumers create their own thin entrypoint that calls `createApp` and bring their own `extensions.config.ts`. See `tools/templates/consumer/apps/api/src/main.ts.tpl` for the reference. +### Running more than one process + +Replicas of one API share a `SERVICE_NAME` (unset means `monolith`) and split the work: each event +and each job reaches one of them. + +- **Fewer modules in a process.** `SERVICE_MANIFEST` lists the module ids a process boots, and + `createApp` applies it to the `plugins` you pass. List every module your modules depend on and + every module that binds a port they require. A module still boots without an optional port, so + the boot log also lists each one the manifest leaves unbound and what its module skips without + it. A control - an RG limit, a country check, an audit record, the database grants and 2FA + checks behind admin routes - is different: while your plugin list has the module that enforces + it, a manifest that leaves it out refuses to boot. Whatever is missing, the boot fails once, + naming all of it, with one manifest line that adds every module needed and what those need in + turn. A process with a manifest needs a `SERVICE_NAME` of its own. + + With the shipped modules a split ends up close to the monolith. Every manifest boot checks + AdminGuard's ports, so every split loads iam and identity, and with them audit and mail. + Identity's controls pull in compliance, and compliance depends on wallet, gaming, + player-management and exchange-rate; player-management depends on chat. A split is those ten + modules plus the ones you meant to move, and until + [ADR-0050](../adr/0050-dependency-only-modules-provide-but-do-not-consume.md) lands each of the + ten handles every event once in the split and once in the rest of the install. + +- **Fewer queues in a process.** `JOB_WORKERS` picks the job queues a process consumes: names, + `prefix*` patterns, `!` to exclude. To keep a long-running queue off the API replicas, run one + more instance with the same plugins and `SERVICE_NAME` and `JOB_WORKERS=`, and give the + replicas `JOB_WORKERS=!`. Every queue needs a consumer somewhere. See + [ADR-0049](../adr/0049-per-process-job-worker-selection.md). + ## Seeding reference data (production) Migrations carry **structure only** (DDL: tables, indexes, constraints) - never data. diff --git a/docs/platform/architecture.md b/docs/platform/architecture.md index e3427ff1f..abc409bf9 100644 --- a/docs/platform/architecture.md +++ b/docs/platform/architecture.md @@ -102,7 +102,7 @@ Solid arrows are runtime/build dependencies; dashed arrows are **adapter seams** **Vendor adapters** - concrete implementations of a module's adapter interfaces (PSP, KYC vendor, igaming aggregator, chat), shipped as separate packages. The interface is the seam; the implementation is swappable. For the pooling/sweeping/reconciliation shape a custody-style payment vendor adds, see [docs/adapters/custody.md](../adapters/custody.md#custody-topology). -**Background jobs** - long-running work runs off the request path in a BullMQ worker shipped as an overlay plugin (`/scaffold-plugin -worker`): a module emits an event via the typed `EventBus`, the worker plugin subscribes in `register(ctx)` and processes the job. There is no standalone worker app. +**Background jobs** - long-running work runs off the request path on `JOB_QUEUE`: a module registers a worker with `ctx.jobs.worker` and enqueues to it. Every process consumes every queue by default; `JOB_WORKERS` narrows that, so a long-running queue can live on an extra instance that boots the same plugins while the API replicas skip it. See [ADR-0049](../adr/0049-per-process-job-worker-selection.md). **SDK & consumption surface** @@ -187,5 +187,6 @@ for the full direction; the shape: scaffolds a thin host); point its broker at the shared stream and its events keep flowing. The `no-cross-module-schema-read` lint warning flags the remaining shared-table couplings to retire first. Scale the stateless Hono API horizontally; - scale async work by moving consumers to their own services. See + scale async work by moving consumers to their own services, or narrow the queues a + process consumes with `JOB_WORKERS`. See [ADR-0017](../adr/0017-extraction-readiness-manifest-outbox-command-ports.md). diff --git a/docs/platform/system-design.md b/docs/platform/system-design.md index 7236b579c..b92dc9db9 100644 --- a/docs/platform/system-design.md +++ b/docs/platform/system-design.md @@ -276,15 +276,24 @@ sequenceDiagram ```mermaid flowchart TB CODE["Single codebase · extensions.config.ts"] - CODE --> MONO["SERVICE_MANIFEST unset to MONOLITH
all 15 modules, in-process seams"] - CODE --> SPLIT["SERVICE_MANIFEST=audit,identity,wallet,...
per-module thin host (pnpm create:service)"] - SPLIT --> S1["identity svc"] - SPLIT --> S2["wallet svc"] - SPLIT --> S3["casino svc"] - S1 & S2 & S3 ---|"durable broker (AMQP_URL)
events cross process · outbox at-least-once"| RMQ[("RabbitMQ/Kafka")] - S1 & S2 & S3 --- PG[("PostgreSQL")] + CODE --> MONO["SERVICE_MANIFEST unset: every module
API replicas: JOB_WORKERS=!reports.*"] + CODE --> WORKER["same plugins and SERVICE_NAME, no ingress
JOB_WORKERS=reports.*"] + CODE --> SPLIT["SERVICE_MANIFEST=wallet,identity,audit,...
own SERVICE_NAME (pnpm create:service)"] + MONO & WORKER & SPLIT ---|"Redis Streams events · BullMQ jobs (REDIS_URL)"| REDIS[("Redis")] + MONO & WORKER & SPLIT --- PG[("PostgreSQL")] ``` +A manifest lists every module its modules depend on and every module that binds a port they +require; `createApp` applies it to any plugin list. It logs each optional port a manifest leaves +unbound, with what the module skips without it, and refuses to boot over a control port - an RG +limit, a geo check, an audit record, AdminGuard's database grants - whose module the plugin list +has but the manifest leaves out. Whatever is missing fails the boot once, with one manifest line +that adds it all. A split service +needs a `SERVICE_NAME` of its own - its consumer group - while an instance that only takes a +share of the queues keeps the replicas' `SERVICE_NAME`, so it competes for events rather than +receiving a copy. `JOB_WORKERS` picks the queues a process consumes +([ADR-0049](../adr/0049-per-process-job-worker-selection.md)). + ## Reference - domain -> modules -> tables -> routes `docs/catalog.json` is the generated surface: every module, table, route, adapter port and @@ -294,14 +303,13 @@ hooks, so it is always current. Query it directly, or use the `oss-dev` MCP tool ## Cross-domain edges (lint-enforced — ADR-0015) -| From → To | Channel | -| --------------------------- | ------------------------------------------------------------------ | -| iam → identity | `dependsOn` (load order) | -| identity → audit | `dependsOn` (load order); writes its audit rows via `AUDIT_WRITER` | -| gaming → wallet | `WALLET_COMMANDS` synchronous command port (same `tx`, atomic) | -| player-management → profile | read-only `@openora/core/pam/schema/profile` | -| lobby → gaming | read-only `@openora/core/casino/schema/gaming` | -| any → any | domain **events** via `EventBus` — never money | +| From → To | Channel | +| --------------------------- | -------------------------------------------------------------- | +| iam → identity | `dependsOn` (load order) | +| gaming → wallet | `WALLET_COMMANDS` synchronous command port (same `tx`, atomic) | +| player-management → profile | read-only `@openora/core/pam/schema/profile` | +| lobby → gaming | read-only `@openora/core/casino/schema/gaming` | +| any → any | domain **events** via `EventBus` — never money | The adapter ports and 3 async seams (`MESSAGE_BROKER`, `JOB_QUEUE`, `REALTIME_TRANSPORT`) plus the transactional `OUTBOX` carry everything else. Money and needed-now reads stay synchronous diff --git a/docs/standards/module-structure.md b/docs/standards/module-structure.md index c6e729dc2..05980a376 100644 --- a/docs/standards/module-structure.md +++ b/docs/standards/module-structure.md @@ -62,6 +62,8 @@ await tags.evaluate(tx, { playerId }); Canonical seams to copy: `wallet -> tag` via `TAG_EVALUATION_COMMANDS` (optional port, resolved lazily with `c.has(...)` in the router factory), `compliance -> pam` via the `kyc.status.changed` domain event. +A port whose absence only degrades a feature, or already fails closed, goes in the plugin's `optionalPorts` with what the module skips without it; the boot logs each one left unbound. A port whose absence skips a money, compliance, authorization or audit control - an RG limit, a geo check, an audit record, a token revocation - goes in `controlPorts` instead, with the module that binds it (`wallet`, `gaming`, `identity`, `mail`). A full boot logs it the same way, since an install without that module has nothing to enforce; a `SERVICE_MANIFEST` boot refuses to start while it is unbound and the plugin list has that module, because another process of the install enforces the control. + Money + any needed-now mutation stay synchronous/transactional, never over events - use a command port: caller passes its own `tx` (`WALLET_COMMANDS.debit(tx, ...)`), atomic in-process yet splittable later; declare `dependsOn: ['']`. Cross-module schema reads are sanctioned but each one is an extraction blocker. ADR-0017. **Never write a deep (`../../`+) relative import that leaves your own top-level dir under `packages/core/src/` - reach every other zone through the package's own `@openora/core/*` subpath.** A relative import is only for staying inside your own module/zone (`./schema/index.js`, `../db/index.js`). The moment a `..` would cross into another domain, slice, or engine zone, it's a `@openora/core/*` import instead: diff --git a/packages/core/generators/src/config.ts b/packages/core/generators/src/config.ts index 6e258029e..62efc957d 100644 --- a/packages/core/generators/src/config.ts +++ b/packages/core/generators/src/config.ts @@ -487,7 +487,7 @@ export default function generator(plop: PlopTypes.NodePlopAPI): void { }); plop.setGenerator('service', { - description: 'Thin single-purpose service host (bakes SERVICE_MANIFEST)', + description: 'Thin single-purpose service host (bakes SERVICE_MANIFEST and SERVICE_NAME)', prompts: [ { type: 'input', diff --git a/packages/core/generators/src/templates/plugin.hbs b/packages/core/generators/src/templates/plugin.hbs index 4c807ed8c..41917073f 100644 --- a/packages/core/generators/src/templates/plugin.hbs +++ b/packages/core/generators/src/templates/plugin.hbs @@ -3,7 +3,8 @@ import type { CoreTokenCatalog, Plugin } from '@openora/core/server'; // Overlay plugin. Registered in extensions.config.ts (auto-wired in the OSS repo; // in a consumer repo add it to your apps/api/src/extensions.config.ts). // -// In register(ctx) - the typed ModuleRegistry - you can: +// register(ctx) only declares - open a timer or connection in a provide()/router factory, +// closed with c.onDispose(). In register(ctx) - the typed ModuleRegistry - you can: // ctx.provide(TOKEN, (c) => new Impl()) - bind/rebind a DI token (last wins) // ctx.routers.add(namespace, (c) => router) - mount an oRPC router // ctx.events.on(topic, (payload) => ...) - subscribe to a domain event diff --git a/packages/core/src/casino/gaming/plugin.ts b/packages/core/src/casino/gaming/plugin.ts index 3395cc457..1b8954486 100644 --- a/packages/core/src/casino/gaming/plugin.ts +++ b/packages/core/src/casino/gaming/plugin.ts @@ -55,6 +55,14 @@ export default { id: 'gaming', requiresPorts: [PLAY_ELIGIBILITY, IDENTITY_READER], dependsOn: ['wallet'], + controlPorts: [ + { port: RG_LIMITS, without: 'a round starts over a wager limit', boundBy: 'compliance' }, + { + port: GAME_GEO_CHECK, + without: 'a game starts in a country it is restricted in', + boundBy: 'compliance', + }, + ], register(ctx) { ctx.provide(GAME_ADAPTER, () => new MockGameAdapter()); ctx.provide(RNG_ADAPTER, () => new MockRngAdapter()); diff --git a/packages/core/src/contracts/adapters/broker.ts b/packages/core/src/contracts/adapters/broker.ts index dd141de50..0e9eb64d8 100644 --- a/packages/core/src/contracts/adapters/broker.ts +++ b/packages/core/src/contracts/adapters/broker.ts @@ -35,8 +35,9 @@ export type BrokerHandler = (envelope: EventEnvelope) => void | Promise; export type SubscribeOptions = { /** - * Forward hint for grouped consumers: Kafka consumer group / RabbitMQ queue group. - * When omitted, the adapter creates an exclusive per-process queue (fan-out). + * Overrides the consumer group (Kafka consumer group / RabbitMQ queue group). `EventBus` + * never sets it, so module handlers join the deployment's group - `SERVICE_NAME` in the + * Redis Streams driver - and each event reaches one replica per service. */ consumerGroup?: string; }; diff --git a/packages/core/src/contracts/adapters/job-queue.ts b/packages/core/src/contracts/adapters/job-queue.ts index b2f7ca3ff..d6e979abe 100644 --- a/packages/core/src/contracts/adapters/job-queue.ts +++ b/packages/core/src/contracts/adapters/job-queue.ts @@ -1,9 +1,8 @@ /** * Background-job seam. Modules enqueue durable, retryable, schedulable work - * through this adapter, so the queue driver is swappable: the default binding is - * an in-process queue (zero dependencies - good for `pnpm dev`, seed and tests); - * a downstream operator binds a durable driver (BullMQ/Redis is the reference - * overlay) to JOB_QUEUE without touching modules. Delivery is at-least-once, so + * through this adapter, so the queue driver is swappable: `createApp` binds the + * BullMQ driver on `REDIS_URL`, and an operator may bind another durable driver to + * JOB_QUEUE without touching modules. Delivery is at-least-once, so * handlers MUST be idempotent (use `idempotencyKey` + a DB guard for money jobs). * Distinct from MESSAGE_BROKER: the broker is "something happened" fan-out (the * EventBus facade); JOB_QUEUE is "execute this unit of work later, with delivery @@ -112,7 +111,7 @@ export type JobQueueAdapter = { repeat: RepeatOptions, ): Promise; unschedule(queue: QueueName, scheduleId: string): Promise; - /** Start consuming a queue. Called once per worker overlay during boot. */ + /** Start consuming a queue. `createApp` calls it at boot for each queue this process consumes (`JOB_WORKERS`). */ registerWorker(registration: WorkerRegistration): void; /** Graceful drain: stop accepting, finish in-flight, close connections. */ close(): Promise; diff --git a/packages/core/src/engagement/chat/plugin.ts b/packages/core/src/engagement/chat/plugin.ts index 4428fa511..d16852939 100644 --- a/packages/core/src/engagement/chat/plugin.ts +++ b/packages/core/src/engagement/chat/plugin.ts @@ -67,11 +67,12 @@ const RoomPurgeJobSchema = z.object({ roomId: UuidSchema }); const AccountClosedJobSchema = z.object({ userId: UuidSchema, closedAt: z.iso.datetime() }); const AccountReopenedJobSchema = z.object({ userId: UuidSchema }); +const logger = createLogger('chat'); + export default { id: 'chat', dependsOn: ['identity', 'audit'], register(ctx) { - const logger = createLogger('chat'); let expiryRef: ChatModerationExpiryService | undefined; let inviteRef: ChatRoomInviteService | undefined; ctx.provide(CHAT_REALTIME_TRANSPORT, (c) => c.get(REALTIME_TRANSPORT)); diff --git a/packages/core/src/engagement/notifications/plugin.ts b/packages/core/src/engagement/notifications/plugin.ts index ffcc0ef81..580be83af 100644 --- a/packages/core/src/engagement/notifications/plugin.ts +++ b/packages/core/src/engagement/notifications/plugin.ts @@ -561,11 +561,18 @@ export function buildRgLimitAdminUpdatedNotification( }); } +const logger = createLogger('notifications'); + export default { id: 'notifications', + controlPorts: [ + { + port: GEO_CHECK_COMMANDS, + without: 'welcome mails offer promotions without the country check', + boundBy: 'compliance', + }, + ], register(ctx) { - const logger = createLogger('notifications'); - // Subscriptions are wired before router factories run (boot order), so these refs are // null at registration but set from the router factory before any real event arrives. let svcRef: NotificationsService | null = null; diff --git a/packages/core/src/mail/plugin.ts b/packages/core/src/mail/plugin.ts index c2b25051f..69be46b83 100644 --- a/packages/core/src/mail/plugin.ts +++ b/packages/core/src/mail/plugin.ts @@ -19,6 +19,14 @@ const logger = createLogger('mail'); export default { id: 'mail', + optionalPorts: [{ port: PLAY_ELIGIBILITY, without: 'welcome mails never offer promotions' }], + controlPorts: [ + { + port: AUDIT_WRITER, + without: 'regulatory mails go out without an audit record', + boundBy: 'audit', + }, + ], register(ctx) { const encryptionSecret = process.env['AUTH_SECRET'] ?? ''; if (encryptionSecret.length < MIN_MAIL_ENCRYPTION_SECRET_LENGTH) { diff --git a/packages/core/src/pam/identity/plugin.ts b/packages/core/src/pam/identity/plugin.ts index 992790b52..9de213a3d 100644 --- a/packages/core/src/pam/identity/plugin.ts +++ b/packages/core/src/pam/identity/plugin.ts @@ -151,6 +151,19 @@ export default { id: 'identity', dependsOn: ['audit'], requiresPorts: [MAIL_DISPATCH], + optionalPorts: [{ port: PLAYER_PROVISIONING, without: 'registration is refused' }], + controlPorts: [ + { + port: GEO_CHECK_COMMANDS, + without: 'registration and sign-in skip the country check', + boundBy: 'compliance', + }, + { + port: MCP_TOKEN_REVOCATION, + without: "ending an admin's sessions leaves their MCP tokens valid", + boundBy: 'iam', + }, + ], register(ctx) { // Built once and shared by ADMIN_SECURITY_POLICY and the router below - two // separate `makeAdminSecurity(c)` calls would construct two independent diff --git a/packages/core/src/promo/bonus/plugin.ts b/packages/core/src/promo/bonus/plugin.ts index 416d49e1e..8da9c57d3 100644 --- a/packages/core/src/promo/bonus/plugin.ts +++ b/packages/core/src/promo/bonus/plugin.ts @@ -95,6 +95,13 @@ export default { id: 'bonus', dependsOn: ['audit'], requiresPorts: [PLAY_ELIGIBILITY], + controlPorts: [ + { + port: GEO_CHECK_COMMANDS, + without: 'public offers are listed to visitors from refused countries', + boundBy: 'compliance', + }, + ], register(ctx) { ctx.provide(BONUS_GRANTS, (c) => new GrantService(c.get(AUDIT_WRITER))); ctx.provide(BONUS_GRANT_LEDGER, () => bonusGrantLedger); diff --git a/packages/core/src/server/kernel/__tests__/container.test.ts b/packages/core/src/server/kernel/__tests__/container.test.ts index 4e1320d0a..875ace517 100644 --- a/packages/core/src/server/kernel/__tests__/container.test.ts +++ b/packages/core/src/server/kernel/__tests__/container.test.ts @@ -112,6 +112,17 @@ describe('Container', () => { expect(() => c.get(A)).toThrow(/Circular dependency/); }); + it('rethrows a refusing factory on every get, not as a circular dependency', () => { + const TOKEN = createToken('refuses'); + const c = new Container(); + c.register(TOKEN, () => { + throw new Error('refused'); + }); + + expect(() => c.get(TOKEN)).toThrow('refused'); + expect(() => c.get(TOKEN)).toThrow('refused'); + }); + it('runs disposers in reverse registration order', async () => { const c = new Container(); const order: string[] = []; diff --git a/packages/core/src/server/kernel/container.ts b/packages/core/src/server/kernel/container.ts index 22414be92..37da8a841 100644 --- a/packages/core/src/server/kernel/container.ts +++ b/packages/core/src/server/kernel/container.ts @@ -97,10 +97,14 @@ export class Container { } this.resolving.add(token); - const instance = factory(this) as TokenValue; - this.resolving.delete(token); - this.instances.set(token, instance); - return instance; + try { + const instance = factory(this) as TokenValue; + this.instances.set(token, instance); + return instance; + } finally { + // A factory that refuses keeps refusing with its own error, not a circular dependency. + this.resolving.delete(token); + } } onDispose(fn: () => void | Promise): void { diff --git a/packages/core/src/server/plugin-host/__tests__/register-side-effects.test.ts b/packages/core/src/server/plugin-host/__tests__/register-side-effects.test.ts new file mode 100644 index 000000000..80979634b --- /dev/null +++ b/packages/core/src/server/plugin-host/__tests__/register-side-effects.test.ts @@ -0,0 +1,97 @@ +import { readdirSync } from 'node:fs'; +import { basename, join } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import type { TokenCatalog } from '@openora/core/contracts'; +import { importPlugin } from '../load-plugins.js'; +import { selectServiceEntries } from '../service-manifest.js'; + +const CORE_SRC = fileURLToPath(new URL('../../../', import.meta.url)); +const CONFIGURED_SECRET = 'register-side-effects-secret-0000000000000000'; +const TIMERS = ['setTimeout', 'setInterval', 'setImmediate'] as const; + +async function loadCorePlugins() { + const paths = readdirSync(CORE_SRC, { recursive: true, encoding: 'utf8' }) + .filter((path) => basename(path) === 'plugin.ts') + .map((path) => join(CORE_SRC, path)); + return Promise.all( + paths.map(async (path) => { + const { default: plugin } = (await import(path)) as { default: { id: string } }; + const entry = { id: plugin.id, path }; + return { entry, plugin: await importPlugin(entry) }; + }), + ); +} + +function processState() { + return { + env: { ...process.env }, + listeners: Object.fromEntries( + process.eventNames().map((event) => [String(event), process.listenerCount(event)]), + ), + globals: Reflect.ownKeys(globalThis).map(String).sort(), + }; +} + +function activeResources() { + return process + .getActiveResourcesInfo() + .reduce>( + (counts, type) => ({ ...counts, [type]: (counts[type] ?? 0) + 1 }), + {}, + ); +} + +function grown(before: Record, after: Record) { + return Object.entries(after) + .filter(([type, count]) => count > (before[type] ?? 0)) + .map(([type]) => type); +} + +afterEach(() => { + vi.restoreAllMocks(); + vi.unstubAllEnvs(); +}); + +describe('the registration pass of a failed SERVICE_MANIFEST boot', () => { + it('registers every core plugin once, leaving no timer, handle, listener, global or env change behind', async () => { + vi.stubEnv('AUTH_SECRET', CONFIGURED_SECRET); + vi.stubEnv('WITHDRAWAL_PIN_HMAC_SECRET', CONFIGURED_SECRET); + const loaded = await loadCorePlugins(); + const registers = loaded.map(({ plugin }) => vi.spyOn(plugin, 'register')); + const state = processState(); + const resources = activeResources(); + const timers = TIMERS.map((name) => vi.spyOn(globalThis, name)); + + await expect( + selectServiceEntries({ + entries: loaded.map(({ entry }) => entry), + manifest: [{ id: 'wallet', noRoutes: false }], + readPlugin: (entry) => importPlugin(entry), + hostControls: [], + isBound: () => false, + }), + ).rejects.toThrow(/^SERVICE_MANIFEST=wallet leaves out/); + + expect(loaded.map(({ plugin }) => plugin.id)).toEqual( + expect.arrayContaining([ + 'audit', + 'identity', + 'mail', + 'notifications', + 'wallet', + 'compliance', + ]), + ); + expect( + Object.fromEntries( + loaded.map(({ plugin }, i) => [plugin.id, registers[i]?.mock.results.map((r) => r.type)]), + ), + ).toEqual(Object.fromEntries(loaded.map(({ plugin }) => [plugin.id, ['return']]))); + for (const timer of timers) { + expect(timer).not.toHaveBeenCalled(); + } + expect(grown(resources, activeResources())).toEqual([]); + expect(processState()).toEqual(state); + }, 60_000); +}); diff --git a/packages/core/src/server/plugin-host/__tests__/required-ports.test.ts b/packages/core/src/server/plugin-host/__tests__/required-ports.test.ts index 9bc0906c0..7a8944f61 100644 --- a/packages/core/src/server/plugin-host/__tests__/required-ports.test.ts +++ b/packages/core/src/server/plugin-host/__tests__/required-ports.test.ts @@ -1,11 +1,13 @@ import { describe, it, expect } from 'vitest'; import { Container } from '../../kernel/index.js'; import { createToken, type TokenCatalog } from '@openora/core/contracts'; -import { assertRequiredPorts } from '../load-plugins.js'; +import { assertRequiredPorts, unboundOptionalPorts } from '../load-plugins.js'; import type { Plugin } from '../define-plugin.js'; const WALLET_COMMANDS = createToken<{ debit: () => void }>('WALLET_COMMANDS'); -const catalog = { WALLET_COMMANDS } satisfies TokenCatalog; +const GEO_CHECK = createToken<{ check: () => void }>('GEO_CHECK'); +const RG_LIMITS = createToken<{ checkDeposit: () => void }>('RG_LIMITS'); +const catalog = { WALLET_COMMANDS, GEO_CHECK, RG_LIMITS } satisfies TokenCatalog; const consumer: Plugin = { id: 'gaming', @@ -31,3 +33,46 @@ describe('assertRequiredPorts (ADR-0024 boot fail-fast)', () => { expect(() => assertRequiredPorts([plain], container)).not.toThrow(); }); }); + +describe('unboundOptionalPorts', () => { + const identity: Plugin = { + id: 'identity', + optionalPorts: [{ port: GEO_CHECK, without: 'sign-in skips the country check' }], + register: () => {}, + }; + + it('names the plugin, the port and what the plugin skips without it', () => { + const container = new Container(); + + expect(unboundOptionalPorts([identity], container)).toEqual([ + { plugin: 'identity', port: 'GEO_CHECK', without: 'sign-in skips the country check' }, + ]); + }); + + it('is empty once the port is bound', () => { + const container = new Container(); + container.register(GEO_CHECK, () => ({ check: () => {} })); + + expect(unboundOptionalPorts([identity], container)).toEqual([]); + }); +}); + +describe('unboundOptionalPorts with control ports', () => { + const wallet: Plugin = { + id: 'wallet', + optionalPorts: [{ port: GEO_CHECK, without: 'nothing that matters here' }], + controlPorts: [ + { port: RG_LIMITS, without: 'deposit limits go unenforced', boundBy: 'compliance' }, + ], + register: () => {}, + }; + + it('logs an unbound control port with the optional ones', () => { + const container = new Container(); + + expect(unboundOptionalPorts([wallet], container).map(({ port }) => port)).toEqual([ + 'GEO_CHECK', + 'RG_LIMITS', + ]); + }); +}); diff --git a/packages/core/src/server/plugin-host/__tests__/service-manifest.test.ts b/packages/core/src/server/plugin-host/__tests__/service-manifest.test.ts index b705a2f80..f16eb1051 100644 --- a/packages/core/src/server/plugin-host/__tests__/service-manifest.test.ts +++ b/packages/core/src/server/plugin-host/__tests__/service-manifest.test.ts @@ -1,6 +1,15 @@ -import { describe, it, expect } from 'vitest'; -import { applyServiceManifest, parseServiceManifest } from '../service-manifest.js'; +import { describe, it, expect, vi } from 'vitest'; +import { createToken, type Token, type TokenCatalog } from '@openora/core/contracts'; +import { + applyServiceManifest, + assertServiceComplete, + parseServiceManifest, + selectServiceEntries, + type HostControl, + type ServiceManifestCheck, +} from '../service-manifest.js'; import type { PluginEntry } from '../load-plugins.js'; +import type { Plugin } from '../define-plugin.js'; const plain = (id: string) => ({ id, noRoutes: false }); @@ -78,3 +87,262 @@ describe('applyServiceManifest', () => { expect(() => applyServiceManifest(entries, [plain('nope')])).toThrow(/unknown module id/); }); }); + +const MAIL = createToken('MAIL'); +const RG_LIMITS = createToken('RG_LIMITS'); +const PERMISSIONS = createToken('PERMISSIONS'); +const catalog = { MAIL, RG_LIMITS, PERMISSIONS } satisfies TokenCatalog; +type Catalog = typeof catalog; + +const plugin = (id: string, shape: Partial> = {}): Plugin => ({ + id, + register: () => {}, + ...shape, +}); +const binds = + (...ports: Token[]): Plugin['register'] => + (ctx) => { + for (const port of ports) { + ctx.provide(port, () => ({})); + } + }; + +const ADMIN_GUARD_CONTROL: HostControl = { + owner: 'AdminGuard', + port: PERMISSIONS, + without: 'grants come from the static role table', + boundBy: 'iam', +}; + +/** A plugin list of `plugins` (those named in `infra` as overlays), checked against `manifest`. */ +function checkOf( + plugins: Plugin[], + manifest: string[], + { + infra = [], + bound = [], + hostControls = [], + }: { infra?: string[]; bound?: symbol[]; hostControls?: HostControl[] } = {}, +) { + const byId = new Map(plugins.map((p) => [p.id, p])); + const readPlugin = vi.fn(async (entry: PluginEntry) => { + const found = byId.get(entry.id); + if (!found) { + throw new Error(`no fixture plugin ${entry.id}`); + } + return found; + }); + const check: ServiceManifestCheck = { + entries: plugins.map((p) => ({ + id: p.id, + path: p.id, + ...(infra.includes(p.id) ? { kind: 'infra' as const } : {}), + })), + manifest: manifest.map((id) => ({ id, noRoutes: false })), + readPlugin, + hostControls, + isBound: (port) => bound.includes(port), + }; + return { check, readPlugin }; +} + +describe('selectServiceEntries', () => { + const graph = [ + plugin('audit'), + plugin('identity', { dependsOn: ['audit'] }), + plugin('wallet', { dependsOn: ['identity', 'audit'] }), + plugin('gaming', { dependsOn: ['wallet'] }), + plugin('bullmq'), + ]; + + it('boots a manifest that lists its dependencies, reading only what it selects', async () => { + const { check, readPlugin } = checkOf(graph, ['audit', 'identity', 'wallet'], { + infra: ['bullmq'], + }); + + const selected = await selectServiceEntries(check); + + expect(selected.map((e) => e.id)).toEqual(['audit', 'identity', 'wallet', 'bullmq']); + expect(readPlugin.mock.calls.map(([e]) => e.id).sort()).toEqual( + ['audit', 'bullmq', 'identity', 'wallet'].sort(), + ); + }); + + it('names every left-out dependency, transitive ones included, in one error', async () => { + const { check } = checkOf(graph, ['gaming']); + + await expect(selectServiceEntries(check)).rejects.toThrow( + /SERVICE_MANIFEST=gaming leaves out[\s\S]*- wallet \(required by gaming\)[\s\S]*- identity \(required by wallet\)[\s\S]*- audit \(required by wallet, identity\)[\s\S]*Add them: SERVICE_MANIFEST=gaming,wallet,identity,audit$/m, + ); + }); + + it("counts an infra overlay's dependencies", async () => { + const overlay = graph.map((p) => (p.id === 'bullmq' ? { ...p, dependsOn: ['identity'] } : p)); + const { check } = checkOf(overlay, ['audit'], { infra: ['bullmq'] }); + + await expect(selectServiceEntries(check)).rejects.toThrow(/- identity \(required by bullmq\)/); + }); + + it('flags a dependency the plugin list does not have at all', async () => { + const { check } = checkOf([plugin('audit', { dependsOn: ['ghost'] })], ['audit']); + const run = selectServiceEntries(check); + + await expect(run).rejects.toThrow(/- ghost \(required by audit\) - not in the plugin list/); + await expect(run).rejects.not.toThrow(/Add them/); + }); + + it('stops on a cycle among left-out modules', async () => { + const cyclic = [ + plugin('gaming', { dependsOn: ['wallet'] }), + plugin('wallet', { dependsOn: ['identity'] }), + plugin('identity', { dependsOn: ['wallet'] }), + ]; + + await expect(selectServiceEntries(checkOf(cyclic, ['gaming']).check)).rejects.toThrow( + /Add them: SERVICE_MANIFEST=gaming,wallet,identity$/m, + ); + }); + + it('still refuses an unknown id before reading anything', async () => { + const { check, readPlugin } = checkOf(graph, ['nope']); + + await expect(selectServiceEntries(check)).rejects.toThrow(/unknown module id/); + expect(readPlugin).not.toHaveBeenCalled(); + }); +}); + +describe('the one error a manifest boot fails with', () => { + const install = [ + plugin('audit'), + plugin('identity', { requiresPorts: [MAIL] }), + plugin('mail', { dependsOn: ['audit'], register: binds(MAIL) }), + plugin('wallet', { + dependsOn: ['identity'], + controlPorts: [{ port: RG_LIMITS, without: 'limits go unenforced', boundBy: 'compliance' }], + }), + plugin('compliance', { dependsOn: ['wallet', 'gaming'], register: binds(RG_LIMITS) }), + plugin('gaming', { dependsOn: ['wallet'] }), + plugin('iam', { dependsOn: ['identity'], register: binds(PERMISSIONS) }), + ]; + const withoutModules = (...ids: string[]) => install.filter((p) => !ids.includes(p.id)); + + it('adds every module a left-out dependency, a required port and a control need', async () => { + const { check } = checkOf(install, ['wallet'], { hostControls: [ADMIN_GUARD_CONTROL] }); + + await expect(selectServiceEntries(check)).rejects.toThrow( + 'SERVICE_MANIFEST=wallet leaves out what this service needs:\n' + + ' - identity (required by wallet, iam)\n' + + ' - wallet runs without RG_LIMITS (bound by compliance): limits go unenforced\n' + + ' - identity requires MAIL (bound by mail)\n' + + ' - AdminGuard runs without PERMISSIONS (bound by iam): grants come from the static role table\n' + + ' - gaming (required by compliance)\n' + + ' - audit (required by mail)\n' + + 'Add them: SERVICE_MANIFEST=wallet,identity,compliance,mail,iam,gaming,audit\n' + + 'A split service may not skip a control the rest of its install enforces', + ); + }); + + it('refuses an unbound required port, naming the module that binds it', async () => { + const { check } = checkOf(install, ['identity']); + + await expect(assertServiceComplete(check)).rejects.toThrow( + /- identity requires MAIL \(bound by mail\)\n {2}- audit \(required by mail\)\nAdd them: SERVICE_MANIFEST=identity,mail,audit\nA split service loads every module/, + ); + }); + + it('refuses an unbound control the plugin list has, adding its module and what it needs', async () => { + const { check } = checkOf(install, ['wallet', 'identity', 'mail', 'audit'], { bound: [MAIL] }); + + await expect(assertServiceComplete(check)).rejects.toThrow( + /- wallet runs without RG_LIMITS \(bound by compliance\): limits go unenforced\n {2}- gaming \(required by compliance\)\nAdd them: SERVICE_MANIFEST=wallet,identity,mail,audit,compliance,gaming\n/, + ); + }); + + it("refuses the host's controls on a split that needs nothing else", async () => { + const { check } = checkOf(install, ['audit'], { hostControls: [ADMIN_GUARD_CONTROL] }); + + await expect(assertServiceComplete(check)).rejects.toThrow( + /- AdminGuard runs without PERMISSIONS \(bound by iam\)[\s\S]*Add them: SERVICE_MANIFEST=audit,iam,identity,mail\n/, + ); + }); + + it('boots, reading only what it selects, once every port is bound', async () => { + const { check, readPlugin } = checkOf(install, ['wallet', 'identity', 'mail', 'audit'], { + bound: [MAIL, RG_LIMITS, PERMISSIONS], + hostControls: [ADMIN_GUARD_CONTROL], + }); + + await expect(assertServiceComplete(check)).resolves.toBeUndefined(); + expect(readPlugin.mock.calls.map(([e]) => e.id)).toEqual([ + 'audit', + 'identity', + 'mail', + 'wallet', + ]); + }); + + it('boots a control whose module the plugin list does not have', async () => { + const { check } = checkOf( + withoutModules('compliance', 'iam'), + ['wallet', 'identity', 'mail', 'audit'], + { bound: [MAIL], hostControls: [ADMIN_GUARD_CONTROL] }, + ); + + await expect(assertServiceComplete(check)).resolves.toBeUndefined(); + }); + + it('still refuses, offering nothing to add, when the selected module does not bind it', async () => { + const unbinding = install.map((p) => (p.id === 'compliance' ? { ...p, register: binds() } : p)); + const { check } = checkOf( + unbinding, + ['wallet', 'identity', 'mail', 'audit', 'compliance', 'gaming'], + { bound: [MAIL] }, + ); + const run = assertServiceComplete(check); + + await expect(run).rejects.toThrow(/- wallet runs without RG_LIMITS \(bound by compliance\)/); + await expect(run).rejects.not.toThrow(/Add them/); + }); + + it('names the module that failed to register when no binder of a required port is known', async () => { + const unconfigured = install.map((p) => + p.id === 'mail' + ? { + ...p, + register: () => { + throw new Error('mail: AUTH_SECRET is too short'); + }, + } + : p, + ); + const { check } = checkOf(unconfigured, ['identity']); + + await expect(assertServiceComplete(check)).rejects.toThrow( + /- identity requires MAIL, which no module in the plugin list binds[\s\S]*failed to register, so the ports they bind are unknown:\n {2}- mail: mail: AUTH_SECRET is too short$/, + ); + }); + + it('registers none of the plugins the boot registered a second time', async () => { + const registers = new Map(install.map(({ id, register }) => [id, vi.fn(register)])); + const { check } = checkOf( + install.map((p) => ({ ...p, register: registers.get(p.id) ?? p.register })), + ['identity'], + ); + + await expect( + assertServiceComplete({ ...check, registered: new Set(['identity']) }), + ).rejects.toThrow(/- identity requires MAIL \(bound by mail\)/); + expect(registers.get('identity')).not.toHaveBeenCalled(); + expect(registers.get('mail')).toHaveBeenCalledOnce(); + }); + + it('names a required port no module in the plugin list binds', async () => { + const { check } = checkOf(withoutModules('mail'), ['identity']); + const run = assertServiceComplete(check); + + await expect(run).rejects.toThrow( + /- identity requires MAIL, which no module in the plugin list binds/, + ); + await expect(run).rejects.not.toThrow(/Add them/); + }); +}); diff --git a/packages/core/src/server/plugin-host/define-plugin.ts b/packages/core/src/server/plugin-host/define-plugin.ts index d730358aa..0de58d812 100644 --- a/packages/core/src/server/plugin-host/define-plugin.ts +++ b/packages/core/src/server/plugin-host/define-plugin.ts @@ -138,7 +138,7 @@ export type ModuleRegistry = { listen(listener: TriggerListener): void; getAll(): readonly RegisteredTriggerListener[]; }; - // Started at boot against the resolved JOB_QUEUE (after all providers, so an overlay's durable driver is in effect). See ADR-0014. + // Started at boot against the resolved JOB_QUEUE (after all providers, so an overlay's durable driver is in effect), unless JOB_WORKERS leaves the queue to another process. See ADR-0014, ADR-0049. jobs: { worker(registration: WorkerRegistration): void; getAll(): WorkerRegistration[]; @@ -164,10 +164,39 @@ export type ModuleRegistry = { export type PluginContext = ModuleRegistry; +/** A port a plugin resolves with `c.has()`: it runs without one, minus what the port backs. */ +export type OptionalPort = { + port: C[keyof C]; + /** What the plugin skips while the port is unbound, as the boot log states it. */ + without: string; +}; + +/** + * An optional port behind a money, compliance, authorization or audit control. An install + * without the module that binds it has nothing to enforce, so a full boot runs without it. A + * split service runs beside a process that has that module, so it may not. + */ +export type ControlPort = OptionalPort & { + /** The module that binds the port, named when a SERVICE_MANIFEST leaves the port unbound. */ + boundBy: string; +}; + export type Plugin = { id: string; dependsOn?: readonly string[]; // Verified once after all plugins register - a missing port fails fast. See ADR-0024. requiresPorts?: readonly (C[keyof C] & Token)[]; + // Checked at the same point; an unbound one is logged, not fatal. + optionalPorts?: readonly OptionalPort[]; + // Logged like optionalPorts, but an unbound one fails a SERVICE_MANIFEST boot. + controlPorts?: readonly ControlPort[]; + /** + * Declares into `ctx` what the plugin contributes, and does nothing else: no timer, connection, + * file or network read, process listener, global, or `process.env` write. A SERVICE_MANIFEST + * boot that fails registers every plugin it has not registered yet, each into a container of + * its own that is then dropped, to learn which binds which port. Open a resource in a + * `provide()` or router factory, closing it with `c.onDispose()`: those run only for the + * container that serves. Reading configuration and throwing on a bad value is fine. + */ register(ctx: PluginContext): void | Promise; }; diff --git a/packages/core/src/server/plugin-host/index.ts b/packages/core/src/server/plugin-host/index.ts index 26c1484a6..0831ae579 100644 --- a/packages/core/src/server/plugin-host/index.ts +++ b/packages/core/src/server/plugin-host/index.ts @@ -1,5 +1,7 @@ export type { Plugin, + OptionalPort, + ControlPort, ModuleRegistry, PluginContext, RouterFactory, diff --git a/packages/core/src/server/plugin-host/load-extensions.ts b/packages/core/src/server/plugin-host/load-extensions.ts index d5cf12a12..7e08fe423 100644 --- a/packages/core/src/server/plugin-host/load-extensions.ts +++ b/packages/core/src/server/plugin-host/load-extensions.ts @@ -1,13 +1,12 @@ import { resolve, dirname, isAbsolute } from 'node:path'; import { accessSync } from 'node:fs'; -import { applyServiceManifest, parseServiceManifest } from './service-manifest.js'; import type { PluginEntry } from './load-plugins.js'; /** * Load the extensions.config.ts registry from disk and resolve plugin paths. * Resolves via EXTENSIONS_CONFIG env var or walks up from process.cwd(). - * Optional `filter` lets a composition root drop entries before the service - * manifest is applied. + * Optional `filter` lets a composition root drop entries. `SERVICE_MANIFEST` is + * applied by `createApp`, to this list or any other it is given. */ export async function loadExtensions( filter?: (entries: PluginEntry[]) => PluginEntry[], @@ -27,18 +26,7 @@ export async function loadExtensions( path: isAbsolute(entry.path) ? entry.path : resolve(configDir, entry.path), })); - const filtered = filter ? filter(resolved) : resolved; - - const manifest = parseServiceManifest(process.env['SERVICE_MANIFEST']); - const selected = applyServiceManifest(filtered, manifest); - if (manifest !== null) { - process.stdout.write( - `SERVICE_MANIFEST active: booting ${selected - .map((e) => (e.noRoutes ? `${e.id} (no routes)` : e.id)) - .join(', ')}\n`, - ); - } - return selected; + return filter ? filter(resolved) : resolved; } function findConfigUpwards(start: string): string { diff --git a/packages/core/src/server/plugin-host/load-plugins.ts b/packages/core/src/server/plugin-host/load-plugins.ts index 6200107c1..f9830eaab 100644 --- a/packages/core/src/server/plugin-host/load-plugins.ts +++ b/packages/core/src/server/plugin-host/load-plugins.ts @@ -1,8 +1,11 @@ import type { Container } from '../kernel/index.js'; +import { createLogger } from '../kernel/logger.js'; import type { TokenCatalog } from '@openora/core/contracts'; import type { Plugin } from './define-plugin.js'; import { ModuleRegistryImpl } from './module-registry.js'; +const logger = createLogger('plugin-host'); + export type PluginEntry = { id: string; /** @@ -20,7 +23,7 @@ export type PluginEntry = { noRoutes?: boolean; }; -function validateEntries(entries: unknown): asserts entries is PluginEntry[] { +export function validateEntries(entries: unknown): asserts entries is PluginEntry[] { if (!Array.isArray(entries)) { throw new Error( `extensions.config.ts must export \`extensions\` as an array of { id, path }. Got ${typeof entries}.`, @@ -93,21 +96,32 @@ export function topoSort(plugins: Plugin[]): Plugin(entry: PluginEntry): Promise> { + const mod = (await import(entry.path)) as { default?: Plugin }; + const plugin = mod.default; + if (!plugin || typeof plugin.register !== 'function') { + throw new Error(`Plugin at "${entry.path}" does not default-export a valid plugin`); + } + assertEntryMatchesPlugin(entry, plugin); + return plugin; +} + +/** + * Imports `entries`, registers them in dependency order and checks their ports. With + * `requiredPorts: 'defer'` an unbound required port is left to the caller, which a + * SERVICE_MANIFEST boot folds into one error with everything else the manifest leaves out. + */ export async function loadPlugins( entries: PluginEntry[], container: Container, + { requiredPorts = 'assert' }: { requiredPorts?: 'assert' | 'defer' } = {}, ): Promise> { validateEntries(entries); const plugins: Plugin[] = []; for (const entry of entries) { - const mod = (await import(entry.path)) as { default?: Plugin }; - const plugin = mod.default; - if (!plugin || typeof plugin.register !== 'function') { - throw new Error(`Plugin at "${entry.path}" does not default-export a valid plugin`); - } - assertEntryMatchesPlugin(entry, plugin); - plugins.push(plugin); + plugins.push(await importPlugin(entry)); } const ordered = topoSort(plugins); @@ -124,11 +138,36 @@ export async function loadPlugins( registry.setOwner(null); } - assertRequiredPorts(ordered, container); + if (requiredPorts === 'assert') { + assertRequiredPorts(ordered, container); + } + for (const { plugin, port, without } of unboundOptionalPorts(ordered, container)) { + logger.warn({ plugin, port }, `${plugin} runs without ${port}: ${without}`); + } return registry; } +export function portName(port: symbol) { + return String(port.description ?? port); +} + +/** + * The optional and control ports no plugin bound. A plugin boots without one but skips what it + * backs - a check, an audit record - so the boot logs each, typically when a SERVICE_MANIFEST + * leaves out the module that binds it. + */ +export function unboundOptionalPorts( + plugins: Plugin[], + container: Container, +): { plugin: string; port: string; without: string }[] { + return plugins.flatMap((plugin) => + [...(plugin.optionalPorts ?? []), ...(plugin.controlPorts ?? [])] + .filter(({ port }) => !container.has(port)) + .map(({ port, without }) => ({ plugin: plugin.id, port: portName(port), without })), + ); +} + /** * Boot-time fail-fast for cross-domain ports (ADR-0024). A domain depends ONLY on * the foundation and reaches another domain through a port token the other domain @@ -154,8 +193,9 @@ export function assertRequiredPorts( throw new Error( `[plugin-host] Required ports are unbound at boot:\n${unbound.join('\n')}\n` + `A domain reaches another domain only through a port the other domain binds. ` + - `Install the package that provides each port (eg @openora/wallet binds WALLET_COMMANDS) ` + - `or remove the plugin that requires it. See ADR-0024.`, + `Load the module that binds each port (eg wallet binds WALLET_COMMANDS) - with ` + + `SERVICE_MANIFEST set, add it to the manifest - or remove the plugin that requires it. ` + + `See ADR-0024.`, ); } } diff --git a/packages/core/src/server/plugin-host/service-manifest.ts b/packages/core/src/server/plugin-host/service-manifest.ts index 432cffd58..1095503dd 100644 --- a/packages/core/src/server/plugin-host/service-manifest.ts +++ b/packages/core/src/server/plugin-host/service-manifest.ts @@ -1,4 +1,8 @@ -import type { PluginEntry } from './load-plugins.js'; +import type { TokenCatalog } from '@openora/core/contracts'; +import { Container } from '../kernel/index.js'; +import type { ControlPort, Plugin } from './define-plugin.js'; +import { portName, validateEntries, type PluginEntry } from './load-plugins.js'; +import { ModuleRegistryImpl } from './module-registry.js'; // `identity:no-routes` loads the module for its ports, events and jobs but mounts none of // its routers. It is for a module a kept module `dependsOn` without wanting its HTTP surface. @@ -68,3 +72,261 @@ export function applyServiceManifest( .filter((e) => e.kind === 'infra' || flags.has(e.id)) .map((e) => (flags.get(e.id) ? { ...e, noRoutes: true } : e)); } + +/** A control the host process holds a manifest boot to, beside its plugins' `controlPorts`. */ +export type HostControl = ControlPort & { + /** What runs without the port, as the boot error names it. */ + owner: string; +}; + +/** What a SERVICE_MANIFEST boot is checked against. */ +export type ServiceManifestCheck = { + /** The whole plugin list: the manifest's modules and every one it leaves out. */ + entries: readonly PluginEntry[]; + manifest: readonly ServiceManifestEntry[]; + readPlugin: (entry: PluginEntry) => Promise>; + hostControls: readonly HostControl[]; + /** Whether the process has bound the port so far. */ + isBound: (port: C[keyof C]) => boolean; + /** + * The plugins this process has registered. What they bound is in its container, which + * `isBound` reads, so the boot error does not register them a second time. + */ + registered?: ReadonlySet; +}; + +type MissingModule = { kind: 'module'; id: string; requiredBy: string[]; listed: boolean }; + +type Finding = + | MissingModule + | { kind: 'required'; owner: string; port: string; boundBy: string | undefined } + | { kind: 'control'; owner: string; port: string; boundBy: string; without: string }; + +function findingLine(finding: Finding) { + switch (finding.kind) { + case 'module': + return finding.listed + ? ` - ${finding.id} (required by ${finding.requiredBy.join(', ')})` + : ` - ${finding.id} (required by ${finding.requiredBy.join(', ')}) - not in the plugin list either`; + case 'required': + return finding.boundBy === undefined + ? ` - ${finding.owner} requires ${finding.port}, which no module in the plugin list binds` + : ` - ${finding.owner} requires ${finding.port} (bound by ${finding.boundBy})`; + case 'control': + return ` - ${finding.owner} runs without ${finding.port} (bound by ${finding.boundBy}): ${finding.without}`; + } +} + +/** + * The modules that bind each of `ports`, in plugin-list order. Each plugin registers into a + * container of its own, so a later binding hides no earlier one. A plugin whose registration + * throws - a module this process does not load may lack its configuration - counts with what it + * bound before the throw, and is listed in `failed` with the error. + */ +async function portBinders( + plugins: readonly Plugin[], + ports: readonly C[keyof C][], +) { + const binders = new Map(); + const failed: string[] = []; + for (const plugin of plugins) { + const container = new Container(); + const registry = new ModuleRegistryImpl(container); + registry.setOwner(plugin.id); + try { + await plugin.register(registry); + } catch (err) { + failed.push(` - ${plugin.id}: ${err instanceof Error ? err.message : String(err)}`); + } + for (const port of ports.filter((p) => container.has(p))) { + binders.set(port, [...(binders.get(port) ?? []), plugin.id]); + } + } + return { binders, failed }; +} + +/** + * Everything the manifest leaves out. From the selected modules it follows each `dependsOn`, + * the module that binds each required port nothing bound, and the module behind each unbound + * control the plugin list has - the host's controls included - then does the same for every + * module that adds, until nothing new is needed. `added` is what to append to the manifest. + */ +function shortfall( + check: ServiceManifestCheck, + plugins: ReadonlyMap>, + binders: ReadonlyMap, +) { + const selected = new Set( + applyServiceManifest([...check.entries], check.manifest).map((e) => e.id), + ); + const loaded = new Set(selected); + const added: string[] = []; + const findings: Finding[] = []; + const missing = new Map(); + + const load = (id: string) => { + if (plugins.has(id) && !loaded.has(id)) { + loaded.add(id); + added.push(id); + } + }; + const bound = (port: C[keyof C]) => + check.isBound(port) || (binders.get(port) ?? []).some((id) => loaded.has(id)); + const requireModule = (id: string, by: string) => { + const known = missing.get(id); + if (known) { + known.requiredBy.push(by); + return; + } + const finding: MissingModule = { + kind: 'module', + id, + requiredBy: [by], + listed: plugins.has(id), + }; + missing.set(id, finding); + findings.push(finding); + load(id); + }; + const requirePort = (owner: string, port: C[keyof C]) => { + if (bound(port)) { + return; + } + const boundBy = binders.get(port)?.at(-1); + findings.push({ kind: 'required', owner, port: portName(port), boundBy }); + if (boundBy !== undefined) { + load(boundBy); + } + }; + const requireControl = (owner: string, { port, without, boundBy }: ControlPort) => { + if (bound(port) || !plugins.has(boundBy)) { + return; + } + findings.push({ kind: 'control', owner, port: portName(port), boundBy, without }); + load(boundBy); + }; + + const dependenciesRead = new Set(); + const portsRead = new Set(); + let hostRead = false; + for (;;) { + // A Set's iterator also visits what is added while it runs, so this closes over dependsOn. + for (const id of loaded) { + if (dependenciesRead.has(id)) { + continue; + } + dependenciesRead.add(id); + for (const dep of plugins.get(id)?.dependsOn ?? []) { + if (!selected.has(dep)) { + requireModule(dep, id); + } + } + } + const unread = [...loaded].filter((id) => !portsRead.has(id)); + if (unread.length === 0 && hostRead) { + return { findings, added }; + } + for (const id of unread) { + portsRead.add(id); + const plugin = plugins.get(id); + for (const port of plugin?.requiresPorts ?? []) { + requirePort(id, port); + } + for (const control of plugin?.controlPorts ?? []) { + requireControl(id, control); + } + } + if (!hostRead) { + hostRead = true; + for (const control of check.hostControls) { + requireControl(control.owner, control); + } + } + } +} + +/** + * The one error a manifest boot fails with: every module the manifest leaves out and why, and a + * manifest line that adds them all, so one edit fixes it. Reads every plugin in the list and + * registers each one the process has not, on its own, to learn which module binds which port. + */ +async function shortfallError(check: ServiceManifestCheck) { + const plugins = new Map>(); + for (const entry of check.entries) { + plugins.set(entry.id, await check.readPlugin(entry)); + } + const ports = [ + ...[...plugins.values()].flatMap((plugin) => [ + ...(plugin.requiresPorts ?? []), + ...(plugin.controlPorts ?? []).map(({ port }) => port), + ]), + ...check.hostControls.map(({ port }) => port), + ]; + const unregistered = [...plugins.values()].filter(({ id }) => !check.registered?.has(id)); + const { binders, failed } = await portBinders(unregistered, ports); + const { findings, added } = shortfall(check, plugins, binders); + // As the operator wrote it, flags included, so the suggested line keeps them. + const manifest = check.manifest + .map((m) => (m.noRoutes ? `${m.id}:${NO_ROUTES_FLAG}` : m.id)) + .join(','); + const unknownBinder = findings.some((f) => f.kind === 'required' && f.boundBy === undefined); + return new Error( + `SERVICE_MANIFEST=${manifest} leaves out what this service needs:\n` + + `${findings.map(findingLine).join('\n')}\n` + + (added.length > 0 ? `Add them: SERVICE_MANIFEST=${[manifest, ...added].join(',')}\n` : '') + + (findings.some(({ kind }) => kind === 'control') + ? 'A split service may not skip a control the rest of its install enforces: load the ' + + 'module that binds each port, or bind the port in an overlay.' + : 'A split service loads every module its modules depend on and every module that ' + + 'binds a port they require.') + + (unknownBinder && failed.length > 0 + ? `\nThese modules failed to register, so the ports they bind are unknown:\n${failed.join('\n')}` + : ''), + ); +} + +/** + * The entries a manifest boot loads: the manifest's modules plus every infra overlay. A + * manifest that leaves out a module they depend on fails before anything registers, with the + * error `assertServiceComplete` gives. Only the selected entries are read on the way to success. + */ +export async function selectServiceEntries(check: ServiceManifestCheck) { + validateEntries(check.entries); + const selected = applyServiceManifest([...check.entries], check.manifest); + const ids = new Set(selected.map((e) => e.id)); + for (const entry of selected) { + const { dependsOn = [] } = await check.readPlugin(entry); + if (dependsOn.some((dep) => !ids.has(dep))) { + throw await shortfallError(check); + } + } + return selected; +} + +/** + * Fails a manifest boot, once its modules registered and the composition root bound its own + * ports, while a selected module's required port is unbound, or a control - a selected module's + * `controlPorts` entry or one of the host's - is unbound and the plugin list has the module + * that binds it. That module then runs in another process of the same install, which enforces + * the control, so this one would let through what that one refuses. A list without the module + * has the control nowhere, and the boot goes on as a full one would. The error names everything + * the manifest leaves out, with a manifest line that boots. + */ +export async function assertServiceComplete( + check: ServiceManifestCheck, +) { + const listed = new Set(check.entries.map((e) => e.id)); + const selected = await Promise.all( + applyServiceManifest([...check.entries], check.manifest).map((e) => check.readPlugin(e)), + ); + const controls = [ + ...selected.flatMap((plugin) => plugin.controlPorts ?? []), + ...check.hostControls, + ]; + const refused = + selected.some((plugin) => (plugin.requiresPorts ?? []).some((port) => !check.isBound(port))) || + controls.some(({ port, boundBy }) => !check.isBound(port) && listed.has(boundBy)); + if (refused) { + throw await shortfallError(check); + } +} diff --git a/packages/core/src/server/runtime/__tests__/create-app-job-workers.int.test.ts b/packages/core/src/server/runtime/__tests__/create-app-job-workers.int.test.ts new file mode 100644 index 000000000..c7820e86b --- /dev/null +++ b/packages/core/src/server/runtime/__tests__/create-app-job-workers.int.test.ts @@ -0,0 +1,75 @@ +import { fileURLToPath } from 'node:url'; +import { describe, it, expect, vi, beforeAll, afterEach, afterAll } from 'vitest'; +import { Queue } from 'bullmq'; +import { JOB_QUEUE, queue } from '@openora/core/contracts'; +import { createTestRedis, redisUrlForWorker, type TestRedis } from '@openora/core/testing'; +import { createApp, type CreatedApp } from '../create-app.js'; + +// Nothing here runs a query; the pool connects lazily. +const DUMMY_DATABASE_URL = 'postgres://test:test@127.0.0.1:1/create_app_test'; +const POLL = { timeout: 5000, interval: 20 }; + +let redis: TestRedis; +let app: CreatedApp | null = null; +const rawQueues: Queue[] = []; +const saved = { redis: process.env['REDIS_URL'], workers: process.env['JOB_WORKERS'] }; + +function rawQueue(name: string): Queue { + const q = new Queue(name, { + connection: { url: redisUrlForWorker(), maxRetriesPerRequest: null }, + }); + rawQueues.push(q); + return q; +} + +beforeAll(async () => { + redis = await createTestRedis(); +}); + +afterEach(async () => { + await app?.close(); + app = null; + await Promise.allSettled(rawQueues.map((q) => q.close())); + rawQueues.length = 0; + await redis.flush(); + for (const [key, value] of [ + ['REDIS_URL', saved.redis], + ['JOB_WORKERS', saved.workers], + ] as const) { + if (value === undefined) { + delete process.env[key]; + } else { + process.env[key] = value; + } + } +}); + +afterAll(async () => { + await redis.quit(); +}); + +describe('createApp - JOB_WORKERS on the real queue driver', () => { + it('consumes the selected queue and leaves the other one waiting for another process', async () => { + process.env['REDIS_URL'] = redisUrlForWorker(); + process.env['JOB_WORKERS'] = 'fixture-base.daily'; + const path = fileURLToPath(new URL('./fixtures/base-plugin.ts', import.meta.url)); + app = await createApp({ + plugins: [{ id: 'fixture-base', path }], + databaseUrl: DUMMY_DATABASE_URL, + }); + const jobs = app.container.get(JOB_QUEUE); + + await jobs.enqueue(queue('fixture-base.weekly'), {}); + await jobs.enqueue(queue('fixture-base.daily'), {}); + + const daily = rawQueue('fixture-base.daily'); + const weekly = rawQueue('fixture-base.weekly'); + await vi.waitFor( + async () => expect(await daily.getJobCounts()).toMatchObject({ completed: 1 }), + POLL, + ); + // The daily job ran, so this process's workers are live; give the weekly queue the same chance. + await new Promise((resolve) => setTimeout(resolve, 300)); + expect(await weekly.getJobCounts()).toMatchObject({ waiting: 1, completed: 0, active: 0 }); + }); +}); diff --git a/packages/core/src/server/runtime/__tests__/create-app.test.ts b/packages/core/src/server/runtime/__tests__/create-app.test.ts index e61362a81..0736e06ee 100644 --- a/packages/core/src/server/runtime/__tests__/create-app.test.ts +++ b/packages/core/src/server/runtime/__tests__/create-app.test.ts @@ -23,13 +23,12 @@ import { } from '@openora/core/contracts'; import { SEALED_TOKENS } from '@openora/core/compliance'; import { redisUrlForWorker } from '@openora/core/testing'; -import { mock } from '../../../testing/mock.js'; -import { AUTH_SESSION, type SessionResolver } from '../../auth/index.js'; -import { createApp } from '../create-app.js'; -import { getCurrentClientMeta } from '../../kernel/request-context.js'; +import { makeJobQueue, mock } from '../../../testing/mock.js'; +import { ADMIN_GUARD, AUTH_SESSION, type SessionResolver } from '../../auth/index.js'; import { EVENT_BUS, createEventBus, + type Container, type EventBus, type EventHandler, type ProcessedEventLog, @@ -37,7 +36,12 @@ import { import { setErrorReporter } from '../../kernel/error-reporter.js'; import { wireErrorResponse } from '../../kernel/wire-error-response.js'; import type { PluginEntry } from '../../plugin-host/index.js'; +import { createApp } from '../create-app.js'; +import type { CoreTokenCatalog } from '../core-token-catalog.js'; +import { countedRegistrations } from './fixtures/counted-plugin.js'; +import { FIXTURE_LIMITS } from './fixtures/limits-plugin.js'; import { heardTriggers } from './fixtures/trigger-listener-plugin.js'; +import { getCurrentClientMeta } from '../../kernel/request-context.js'; // A syntactically valid but unreachable DB url - fine here because nothing in this // suite ever runs a query: DrizzleService's pg.Pool connects lazily, and neither @@ -45,6 +49,9 @@ import { heardTriggers } from './fixtures/trigger-listener-plugin.js'; // no session cookie (getSessionFromCtx short-circuits on a missing cookie). const DUMMY_DATABASE_URL = 'postgres://test:test@127.0.0.1:1/create_app_test'; +const fixture = (name: string) => + fileURLToPath(new URL(`./fixtures/${name}-plugin.ts`, import.meta.url)); + describe('createApp - distributed-only durable seams (ADR-0030)', () => { it('throws a clear, actionable error when no durable seam is bound', async () => { await expect(createApp({ plugins: [], databaseUrl: DUMMY_DATABASE_URL })).rejects.toThrow( @@ -158,14 +165,10 @@ describe('createApp - the sealed tokens', () => { describe('createApp - agent trigger listeners', () => { const saved = process.env['REDIS_URL']; - const fixture = (file: string) => fileURLToPath(new URL(`./fixtures/${file}`, import.meta.url)); - const recording: PluginEntry = { - id: 'trigger-listener', - path: fixture('trigger-listener-plugin.ts'), - }; + const recording: PluginEntry = { id: 'trigger-listener', path: fixture('trigger-listener') }; const throwing: PluginEntry = { id: 'throwing-trigger-listener', - path: fixture('throwing-trigger-listener-plugin.ts'), + path: fixture('throwing-trigger-listener'), }; afterEach(() => { @@ -452,6 +455,259 @@ describe('createApp - service name for the Redis Streams consumer group', () => }); }); +describe('createApp - SERVICE_MANIFEST and JOB_WORKERS on the plugin list it is given', () => { + const keys = [ + 'REDIS_URL', + 'SERVICE_MANIFEST', + 'SERVICE_NAME', + 'REALTIME_NAMESPACE', + 'JOB_WORKERS', + ] as const; + const saved = keys.map((key) => [key, process.env[key]] as const); + + afterEach(() => { + for (const [key, value] of saved) { + if (value === undefined) { + delete process.env[key]; + } else { + process.env[key] = value; + } + } + }); + + // A static list, as a consumer that never calls loadExtensions() passes it. + const plugins: PluginEntry[] = [ + { id: 'fixture-base', path: fixture('base') }, + { id: 'fixture-feature', path: fixture('feature') }, + { id: 'fixture-leaf', path: fixture('leaf') }, + ]; + + /** Boots the fixtures under `env` and returns the queues whose workers were registered. */ + async function bootQueues( + env: { SERVICE_MANIFEST?: string; JOB_WORKERS?: string }, + list: PluginEntry[] = plugins, + ) { + process.env['REDIS_URL'] = redisUrlForWorker(); + for (const key of [ + 'SERVICE_MANIFEST', + 'SERVICE_NAME', + 'REALTIME_NAMESPACE', + 'JOB_WORKERS', + ] as const) { + delete process.env[key]; + } + if (env.SERVICE_MANIFEST) { + process.env['SERVICE_MANIFEST'] = env.SERVICE_MANIFEST; + process.env['SERVICE_NAME'] = 'fixture'; + process.env['REALTIME_NAMESPACE'] = 'fixture'; + } + if (env.JOB_WORKERS) { + process.env['JOB_WORKERS'] = env.JOB_WORKERS; + } + const jobs = makeJobQueue(); + const created = await createApp({ plugins: list, databaseUrl: DUMMY_DATABASE_URL }, (c) => { + c.register(JOB_QUEUE, () => jobs); + }); + await created.close(); + return vi.mocked(jobs.registerWorker).mock.calls.map(([registration]) => registration.queue); + } + + it('boots only the modules the manifest selects', async () => { + expect(await bootQueues({ SERVICE_MANIFEST: 'fixture-base' })).toEqual([ + 'fixture-base.daily', + 'fixture-base.weekly', + ]); + }); + + it('names every module the manifest leaves out in one boot error', async () => { + await expect(bootQueues({ SERVICE_MANIFEST: 'fixture-leaf' })).rejects.toThrow( + /fixture-feature \(required by fixture-leaf\)[\s\S]*fixture-base \(required by fixture-feature\)[\s\S]*SERVICE_MANIFEST=fixture-leaf,fixture-feature,fixture-base/, + ); + }); + + it('registers only the workers JOB_WORKERS selects', async () => { + expect(await bootQueues({ JOB_WORKERS: 'fixture-base.*,!fixture-base.weekly' })).toEqual([ + 'fixture-base.daily', + ]); + expect(await bootQueues({ JOB_WORKERS: '!fixture-base.*' })).toEqual(['fixture-feature.sync']); + }); + + it('boots with an exclusion that matches no queue', async () => { + expect(await bootQueues({ JOB_WORKERS: '!absent.*' })).toHaveLength(3); + }); + + it('fails the boot on an inclusion that matches no queue', async () => { + await expect(bootQueues({ JOB_WORKERS: 'fixture-bsae.*' })).rejects.toThrow( + /does not register: fixture-bsae\.\*/, + ); + }); + + it('fails the boot on a malformed JOB_WORKERS', async () => { + await expect(bootQueues({ JOB_WORKERS: 'none,fixture-base.daily' })).rejects.toThrow( + /"none" stands alone/, + ); + }); + + it('starts a worker a router factory registers', async () => { + const late = [{ id: 'fixture-late-worker', path: fixture('late-worker') }]; + + expect(await bootQueues({}, late)).toEqual(['fixture-late-worker.run']); + expect(await bootQueues({ JOB_WORKERS: 'fixture-late-worker.*' }, late)).toEqual([ + 'fixture-late-worker.run', + ]); + }); +}); + +describe('createApp - a split service fails closed on a control it would skip', () => { + const keys = ['REDIS_URL', 'SERVICE_MANIFEST', 'SERVICE_NAME', 'REALTIME_NAMESPACE'] as const; + const saved = keys.map((key) => [key, process.env[key]] as const); + + afterEach(() => { + for (const [key, value] of saved) { + if (value === undefined) { + delete process.env[key]; + } else { + process.env[key] = value; + } + } + }); + + const entry = (id: string, name: string): PluginEntry => ({ id, path: fixture(name) }); + const limited = entry('fixture-limited', 'limited'); + const limits = entry('fixture-limits', 'limits'); + const admin = entry('fixture-admin', 'admin'); + const iam = entry('iam', 'iam'); + const identity = entry('identity', 'identity'); + const base = entry('fixture-base', 'base'); + + /** Boots `plugins` under `manifest` (none: a full boot) and closes the app again. */ + async function boot( + plugins: PluginEntry[], + manifest?: string, + configure?: (container: Container) => void, + ) { + process.env['REDIS_URL'] = redisUrlForWorker(); + delete process.env['SERVICE_MANIFEST']; + delete process.env['SERVICE_NAME']; + delete process.env['REALTIME_NAMESPACE']; + if (manifest) { + process.env['SERVICE_MANIFEST'] = manifest; + process.env['SERVICE_NAME'] = 'fixture'; + process.env['REALTIME_NAMESPACE'] = 'fixture'; + } + const created = await createApp({ plugins, databaseUrl: DUMMY_DATABASE_URL }, (c) => { + c.register(JOB_QUEUE, () => makeJobQueue()); + configure?.(c); + }); + await created.close(); + } + + it("refuses a module's unbound control port, naming the module that binds it", async () => { + await expect(boot([limited, limits], 'fixture-limited')).rejects.toThrow( + /- fixture-limited runs without FIXTURE_LIMITS \(bound by fixture-limits\): fixture limits go unenforced\nAdd them: SERVICE_MANIFEST=fixture-limited,fixture-limits/, + ); + }); + + it('registers each loaded plugin once and closes the connection it opened when it refuses', async () => { + const counted = entry('fixture-counted', 'counted'); + const openSockets = () => + process.getActiveResourcesInfo().filter((type) => type === 'TCPSocketWrap').length; + const socketsBefore = openSockets(); + countedRegistrations().count = 0; + + await expect(boot([counted, limits], 'fixture-counted')).rejects.toThrow( + /- fixture-counted runs without FIXTURE_LIMITS \(bound by fixture-limits\)/, + ); + + expect(countedRegistrations().count).toBe(1); + await vi.waitFor(() => expect(openSockets()).toBeLessThanOrEqual(socketsBefore), { + timeout: 2_000, + }); + }); + + it('boots once the manifest has that module, or the composition root binds the port', async () => { + await expect( + boot([limited, limits], 'fixture-limited,fixture-limits'), + ).resolves.toBeUndefined(); + await expect( + boot([limited, limits], 'fixture-limited', (c) => + c.registerUnsafe(FIXTURE_LIMITS, () => ({ check: () => true })), + ), + ).resolves.toBeUndefined(); + }); + + it('refuses admin routes without the ports AdminGuard checks them with', async () => { + await expect(boot([admin, iam, identity], 'fixture-admin')).rejects.toThrow( + /- AdminGuard runs without ADMIN_PERMISSION_RESOLVER \(bound by iam\)[\s\S]*- AdminGuard runs without ADMIN_SECURITY_POLICY \(bound by identity\)[\s\S]*Add them: SERVICE_MANIFEST=fixture-admin,iam,identity/, + ); + }); + + it('boots admin routes once the manifest has iam and identity', async () => { + await expect( + boot([admin, iam, identity], 'fixture-admin,iam,identity'), + ).resolves.toBeUndefined(); + }); + + it('refuses a split that serves no admin routes without them too', async () => { + await expect(boot([base, iam, identity], 'fixture-base')).rejects.toThrow( + /- AdminGuard runs without ADMIN_PERMISSION_RESOLVER \(bound by iam\)[\s\S]*Add them: SERVICE_MANIFEST=fixture-base,iam,identity/, + ); + }); + + it('builds AdminGuard at boot under a manifest, so a guard that cannot be built fails the boot', async () => { + const broken = (c: Container) => + c.register(ADMIN_GUARD, () => { + throw new Error('fixture guard cannot be built'); + }); + + await expect(boot([base], 'fixture-base', broken)).rejects.toThrow( + 'fixture guard cannot be built', + ); + await expect(boot([base], undefined, broken)).resolves.toBeUndefined(); + }); + + it('names a missing dependency, required port, control and guard port in one error whose manifest boots', async () => { + const plugins = [ + entry('fixture-split', 'split'), + entry('fixture-feature', 'feature'), + base, + entry('fixture-leaf', 'leaf'), + entry('fixture-dispatch', 'dispatch'), + limits, + iam, + identity, + ]; + + const refused = await boot(plugins, 'fixture-split').then( + () => new Error('booted'), + (err: unknown) => (err instanceof Error ? err : new Error(String(err))), + ); + + expect(refused.message).toContain( + 'SERVICE_MANIFEST=fixture-split leaves out what this service needs:\n' + + ' - fixture-feature (required by fixture-split, fixture-leaf)\n' + + ' - fixture-base (required by fixture-feature)\n' + + ' - fixture-split requires FIXTURE_DISPATCH (bound by fixture-dispatch)\n' + + ' - fixture-split runs without FIXTURE_LIMITS (bound by fixture-limits): fixture limits go unenforced\n', + ); + expect(refused.message).toMatch(/- AdminGuard runs without ADMIN_PERMISSION_RESOLVER/); + expect(refused.message).toMatch(/- fixture-leaf \(required by fixture-dispatch\)/); + const suggested = /^Add them: SERVICE_MANIFEST=(\S+)$/m.exec(refused.message)?.[1]; + expect(suggested).toBe( + 'fixture-split,fixture-feature,fixture-base,fixture-dispatch,fixture-limits,iam,identity,fixture-leaf', + ); + await expect(boot(plugins, suggested)).resolves.toBeUndefined(); + }); + + it('boots a full install unchanged, the controls left unbound', async () => { + await expect(boot([limited, admin])).resolves.toBeUndefined(); + }); + + it('boots a manifest whose plugin list has no module behind the controls', async () => { + await expect(boot([limited, admin], 'fixture-limited,fixture-admin')).resolves.toBeUndefined(); + }); +}); + describe('createApp - the player-activity stamp', () => { it('lands before the request that refreshed it returns', async () => { const saved = process.env['REDIS_URL']; diff --git a/packages/core/src/server/runtime/__tests__/fixtures/admin-plugin.ts b/packages/core/src/server/runtime/__tests__/fixtures/admin-plugin.ts new file mode 100644 index 000000000..9d749ebb9 --- /dev/null +++ b/packages/core/src/server/runtime/__tests__/fixtures/admin-plugin.ts @@ -0,0 +1,14 @@ +import type { TokenCatalog } from '@openora/core/contracts'; +import { ADMIN_GUARD } from '../../../auth/index.js'; +import type { Plugin } from '../../../plugin-host/define-plugin.js'; + +// Resolves the guard in its router factory, as every module with admin routes does. +export default { + id: 'fixture-admin', + register(ctx) { + ctx.routers.add('fixture-admin', (c) => { + c.get(ADMIN_GUARD); + return {}; + }); + }, +} satisfies Plugin; diff --git a/packages/core/src/server/runtime/__tests__/fixtures/base-plugin.ts b/packages/core/src/server/runtime/__tests__/fixtures/base-plugin.ts new file mode 100644 index 000000000..feafa016c --- /dev/null +++ b/packages/core/src/server/runtime/__tests__/fixtures/base-plugin.ts @@ -0,0 +1,12 @@ +import * as z from 'zod'; +import { queue, type TokenCatalog } from '@openora/core/contracts'; +import type { Plugin } from '../../../plugin-host/define-plugin.js'; + +export default { + id: 'fixture-base', + register(ctx) { + for (const name of ['fixture-base.daily', 'fixture-base.weekly']) { + ctx.jobs.worker({ queue: queue(name), schema: z.object({}), handler: async () => {} }); + } + }, +} satisfies Plugin; diff --git a/packages/core/src/server/runtime/__tests__/fixtures/counted-plugin.ts b/packages/core/src/server/runtime/__tests__/fixtures/counted-plugin.ts new file mode 100644 index 000000000..447b24e4a --- /dev/null +++ b/packages/core/src/server/runtime/__tests__/fixtures/counted-plugin.ts @@ -0,0 +1,20 @@ +import type { TokenCatalog } from '@openora/core/contracts'; +import type { Plugin } from '../../../plugin-host/define-plugin.js'; +import { FIXTURE_LIMITS } from './limits-plugin.js'; + +/** How often this plugin registered, on globalThis: the host imports this file by path, not the test. */ +export function countedRegistrations() { + const holder = globalThis as { __countedRegistrations?: { count: number } }; + holder.__countedRegistrations ??= { count: 0 }; + return holder.__countedRegistrations; +} + +export default { + id: 'fixture-counted', + controlPorts: [ + { port: FIXTURE_LIMITS, without: 'fixture limits go unenforced', boundBy: 'fixture-limits' }, + ], + register() { + countedRegistrations().count += 1; + }, +} satisfies Plugin; diff --git a/packages/core/src/server/runtime/__tests__/fixtures/dispatch-plugin.ts b/packages/core/src/server/runtime/__tests__/fixtures/dispatch-plugin.ts new file mode 100644 index 000000000..26b333de0 --- /dev/null +++ b/packages/core/src/server/runtime/__tests__/fixtures/dispatch-plugin.ts @@ -0,0 +1,12 @@ +import { createToken, type TokenCatalog } from '@openora/core/contracts'; +import type { Plugin } from '../../../plugin-host/define-plugin.js'; + +export const FIXTURE_DISPATCH = createToken<{ send: () => void }>('FIXTURE_DISPATCH'); + +export default { + id: 'fixture-dispatch', + dependsOn: ['fixture-leaf'], + register(ctx) { + ctx.provide(FIXTURE_DISPATCH, () => ({ send: () => {} })); + }, +} satisfies Plugin; diff --git a/packages/core/src/server/runtime/__tests__/fixtures/feature-plugin.ts b/packages/core/src/server/runtime/__tests__/fixtures/feature-plugin.ts new file mode 100644 index 000000000..d8ccc2417 --- /dev/null +++ b/packages/core/src/server/runtime/__tests__/fixtures/feature-plugin.ts @@ -0,0 +1,15 @@ +import * as z from 'zod'; +import { queue, type TokenCatalog } from '@openora/core/contracts'; +import type { Plugin } from '../../../plugin-host/define-plugin.js'; + +export default { + id: 'fixture-feature', + dependsOn: ['fixture-base'], + register(ctx) { + ctx.jobs.worker({ + queue: queue('fixture-feature.sync'), + schema: z.object({}), + handler: async () => {}, + }); + }, +} satisfies Plugin; diff --git a/packages/core/src/server/runtime/__tests__/fixtures/iam-plugin.ts b/packages/core/src/server/runtime/__tests__/fixtures/iam-plugin.ts new file mode 100644 index 000000000..2564dc2bc --- /dev/null +++ b/packages/core/src/server/runtime/__tests__/fixtures/iam-plugin.ts @@ -0,0 +1,12 @@ +import { ADMIN_PERMISSION_RESOLVER, type TokenCatalog } from '@openora/core/contracts'; +import type { Plugin } from '../../../plugin-host/define-plugin.js'; + +export default { + id: 'iam', + register(ctx) { + ctx.provide(ADMIN_PERMISSION_RESOLVER, () => ({ + getGrants: async () => [], + isSuperAdmin: async () => false, + })); + }, +} satisfies Plugin; diff --git a/packages/core/src/server/runtime/__tests__/fixtures/identity-plugin.ts b/packages/core/src/server/runtime/__tests__/fixtures/identity-plugin.ts new file mode 100644 index 000000000..215da4bb5 --- /dev/null +++ b/packages/core/src/server/runtime/__tests__/fixtures/identity-plugin.ts @@ -0,0 +1,12 @@ +import { ADMIN_SECURITY_POLICY, type TokenCatalog } from '@openora/core/contracts'; +import type { Plugin } from '../../../plugin-host/define-plugin.js'; + +export default { + id: 'identity', + register(ctx) { + ctx.provide(ADMIN_SECURITY_POLICY, () => ({ + assertEnrolled: async () => {}, + assertSessionIntact: async () => {}, + })); + }, +} satisfies Plugin; diff --git a/packages/core/src/server/runtime/__tests__/fixtures/late-worker-plugin.ts b/packages/core/src/server/runtime/__tests__/fixtures/late-worker-plugin.ts new file mode 100644 index 000000000..c320fe555 --- /dev/null +++ b/packages/core/src/server/runtime/__tests__/fixtures/late-worker-plugin.ts @@ -0,0 +1,17 @@ +import * as z from 'zod'; +import { queue, type TokenCatalog } from '@openora/core/contracts'; +import type { Plugin } from '../../../plugin-host/define-plugin.js'; + +export default { + id: 'fixture-late-worker', + register(ctx) { + ctx.routers.add('fixture-late-worker', () => { + ctx.jobs.worker({ + queue: queue('fixture-late-worker.run'), + schema: z.object({}), + handler: async () => {}, + }); + return {}; + }); + }, +} satisfies Plugin; diff --git a/packages/core/src/server/runtime/__tests__/fixtures/leaf-plugin.ts b/packages/core/src/server/runtime/__tests__/fixtures/leaf-plugin.ts new file mode 100644 index 000000000..331c88581 --- /dev/null +++ b/packages/core/src/server/runtime/__tests__/fixtures/leaf-plugin.ts @@ -0,0 +1,8 @@ +import type { TokenCatalog } from '@openora/core/contracts'; +import type { Plugin } from '../../../plugin-host/define-plugin.js'; + +export default { + id: 'fixture-leaf', + dependsOn: ['fixture-feature'], + register() {}, +} satisfies Plugin; diff --git a/packages/core/src/server/runtime/__tests__/fixtures/limited-plugin.ts b/packages/core/src/server/runtime/__tests__/fixtures/limited-plugin.ts new file mode 100644 index 000000000..af1a53c55 --- /dev/null +++ b/packages/core/src/server/runtime/__tests__/fixtures/limited-plugin.ts @@ -0,0 +1,11 @@ +import type { TokenCatalog } from '@openora/core/contracts'; +import type { Plugin } from '../../../plugin-host/define-plugin.js'; +import { FIXTURE_LIMITS } from './limits-plugin.js'; + +export default { + id: 'fixture-limited', + controlPorts: [ + { port: FIXTURE_LIMITS, without: 'fixture limits go unenforced', boundBy: 'fixture-limits' }, + ], + register() {}, +} satisfies Plugin; diff --git a/packages/core/src/server/runtime/__tests__/fixtures/limits-plugin.ts b/packages/core/src/server/runtime/__tests__/fixtures/limits-plugin.ts new file mode 100644 index 000000000..08eabe964 --- /dev/null +++ b/packages/core/src/server/runtime/__tests__/fixtures/limits-plugin.ts @@ -0,0 +1,11 @@ +import { createToken, type TokenCatalog } from '@openora/core/contracts'; +import type { Plugin } from '../../../plugin-host/define-plugin.js'; + +export const FIXTURE_LIMITS = createToken<{ check: () => boolean }>('FIXTURE_LIMITS'); + +export default { + id: 'fixture-limits', + register(ctx) { + ctx.provide(FIXTURE_LIMITS, () => ({ check: () => true })); + }, +} satisfies Plugin; diff --git a/packages/core/src/server/runtime/__tests__/fixtures/split-plugin.ts b/packages/core/src/server/runtime/__tests__/fixtures/split-plugin.ts new file mode 100644 index 000000000..8a6ec5d7e --- /dev/null +++ b/packages/core/src/server/runtime/__tests__/fixtures/split-plugin.ts @@ -0,0 +1,14 @@ +import type { TokenCatalog } from '@openora/core/contracts'; +import type { Plugin } from '../../../plugin-host/define-plugin.js'; +import { FIXTURE_DISPATCH } from './dispatch-plugin.js'; +import { FIXTURE_LIMITS } from './limits-plugin.js'; + +export default { + id: 'fixture-split', + dependsOn: ['fixture-feature'], + requiresPorts: [FIXTURE_DISPATCH], + controlPorts: [ + { port: FIXTURE_LIMITS, without: 'fixture limits go unenforced', boundBy: 'fixture-limits' }, + ], + register() {}, +} satisfies Plugin; diff --git a/packages/core/src/server/runtime/__tests__/job-workers.test.ts b/packages/core/src/server/runtime/__tests__/job-workers.test.ts new file mode 100644 index 000000000..2dfa5e1ef --- /dev/null +++ b/packages/core/src/server/runtime/__tests__/job-workers.test.ts @@ -0,0 +1,98 @@ +import { describe, it, expect } from 'vitest'; +import { parseJobWorkers, selectJobWorkers } from '../job-workers.js'; + +const registered = [ + { queue: 'reports.daily' }, + { queue: 'reports.weekly' }, + { queue: 'reportsx' }, + { queue: 'promo-sweep' }, + { queue: 'mail' }, +]; + +const consumedBy = (raw: string) => + selectJobWorkers(registered, parseJobWorkers(raw)).consumed.map((r) => r.queue); + +describe('parseJobWorkers', () => { + it('reads unset or blank as every queue, with no selection at all', () => { + expect(parseJobWorkers(undefined)).toBeNull(); + expect(parseJobWorkers('')).toBeNull(); + expect(parseJobWorkers(' ')).toBeNull(); + }); + + it('reads the keywords and patterns, split on commas or spaces', () => { + expect(parseJobWorkers('*')).toEqual({ include: null, exclude: [] }); + expect(parseJobWorkers('all')).toEqual({ include: null, exclude: [] }); + expect(parseJobWorkers('none')).toEqual({ include: [], exclude: [] }); + expect(parseJobWorkers('mail reports.*,,promo-*')).toEqual({ + include: ['mail', 'reports.*', 'promo-*'], + exclude: [], + }); + expect(parseJobWorkers('!reports.*')).toEqual({ include: null, exclude: ['reports.*'] }); + expect(parseJobWorkers('*,!mail')).toEqual({ include: null, exclude: ['mail'] }); + expect(parseJobWorkers('reports.*,!reports.weekly')).toEqual({ + include: ['reports.*'], + exclude: ['reports.weekly'], + }); + }); + + it.each([ + ['none,mail', /"none" stands alone/], + ['*,mail', /exclude with "!"/], + ['re*ports', /only at its end/], + ['*mail', /only at its end/], + ['!', /names no queue/], + ])('refuses %s', (raw, error) => { + expect(() => parseJobWorkers(raw)).toThrow(error); + }); +}); + +describe('selectJobWorkers', () => { + it('consumes every queue without a selection', () => { + expect(selectJobWorkers(registered, null)).toEqual({ + consumed: registered, + skipped: [], + idleExclusions: [], + }); + }); + + it('matches a name exactly and a prefix pattern by its prefix', () => { + expect(consumedBy('mail')).toEqual(['mail']); + expect(consumedBy('reports.*')).toEqual(['reports.daily', 'reports.weekly']); + expect(consumedBy('promo-*')).toEqual(['promo-sweep']); + }); + + it('consumes every queue but the excluded ones when only exclusions are given', () => { + expect(consumedBy('!reports.*')).toEqual(['reportsx', 'promo-sweep', 'mail']); + expect(consumedBy('!*')).toEqual([]); + }); + + it('takes the included queues minus the excluded ones', () => { + expect(consumedBy('reports.*,mail,!reports.weekly')).toEqual(['reports.daily', 'mail']); + }); + + it('consumes nothing with none, and reports what it skips', () => { + const { consumed, skipped } = selectJobWorkers(registered, parseJobWorkers('none')); + + expect(consumed).toEqual([]); + expect(skipped).toEqual(registered); + }); + + it('fails on an inclusion no registered queue matches, naming it and the queues', () => { + expect(() => selectJobWorkers(registered, parseJobWorkers('reprots.*,mail'))).toThrow( + /does not register: reprots\.\*\. Registered queues: reports\.daily, .*mail/, + ); + }); + + it('allows an exclusion nothing matches and reports it as idle', () => { + const selected = selectJobWorkers(registered, parseJobWorkers('!absent.*,!mail')); + + expect(selected.idleExclusions).toEqual(['absent.*']); + expect(selected.consumed.map((r) => r.queue)).not.toContain('mail'); + }); + + it('keeps two workers on one queue together', () => { + const twice = [{ queue: 'mail' }, { queue: 'mail' }, { queue: 'reportsx' }]; + + expect(selectJobWorkers(twice, parseJobWorkers('mail')).consumed).toHaveLength(2); + }); +}); diff --git a/packages/core/src/server/runtime/create-app.ts b/packages/core/src/server/runtime/create-app.ts index c7c6decab..5dc818bec 100644 --- a/packages/core/src/server/runtime/create-app.ts +++ b/packages/core/src/server/runtime/create-app.ts @@ -57,7 +57,14 @@ import { OutboxRelay, } from '../db/index.js'; import { AdminGuard, ADMIN_GUARD, SessionResolver, AUTH_SESSION } from '../auth/index.js'; -import { loadPlugins, type PluginEntry } from '../plugin-host/index.js'; +import { loadPlugins, parseServiceManifest, type PluginEntry } from '../plugin-host/index.js'; +import { importPlugin } from '../plugin-host/load-plugins.js'; +import { + assertServiceComplete, + selectServiceEntries, + type HostControl, + type ServiceManifestCheck, +} from '../plugin-host/service-manifest.js'; import { allowedTriggers, createMcpKernel } from '../mcp/index.js'; import { authorizeWithAdminGuard } from '../mcp/authorize.js'; import type { McpHttpTransport } from '../mcp/http-transport.js'; @@ -73,6 +80,7 @@ import { import { loadPlatformConfig, resolvePlatformConfigPath } from '../kernel/platform-config-loader.js'; import { wireErrorBody } from '../kernel/wire-error-response.js'; import type { CoreTokenCatalog } from './core-token-catalog.js'; +import { parseJobWorkers, selectJobWorkers } from './job-workers.js'; // Path prefixes safe to cache at the HTTP layer: public, non-personalized reads // only (public CMS content and the game catalogue). NOTHING @@ -131,7 +139,7 @@ function resolveServiceName(): string { if (serviceName) { return serviceName; } - if (process.env['SERVICE_MANIFEST']) { + if (parseServiceManifest(process.env['SERVICE_MANIFEST']) !== null) { throw new Error( '[create-app] SERVICE_MANIFEST is set but SERVICE_NAME is not. A split service needs a stable ' + 'name of its own: it becomes the MESSAGE_BROKER consumer group, and every service sharing a ' + @@ -151,7 +159,7 @@ function resolveRealtimeNamespace(serviceName: string): string { if (namespace) { return namespace; } - if (process.env['SERVICE_MANIFEST']) { + if (parseServiceManifest(process.env['SERVICE_MANIFEST']) !== null) { throw new Error( '[create-app] SERVICE_MANIFEST is set but REALTIME_NAMESPACE is not. A split service would ' + 'fall back to its own name, so its channels, signals, presence and session revocation ' + @@ -230,6 +238,25 @@ function requestMeta(c: Context, trustedProxies: TrustedProxies) { }; } +// The ports AdminGuard reads with has(). An install without iam or identity runs without them +// everywhere; a split service whose install has them must not, or an admin whose role was +// revoked in the database, or who never enrolled 2FA, still acts through it. +const ADMIN_GUARD_CONTROLS = [ + { + owner: 'AdminGuard', + port: ADMIN_PERMISSION_RESOLVER, + boundBy: 'iam', + without: + 'grants come from the static role table, so an admin whose role was revoked still acts', + }, + { + owner: 'AdminGuard', + port: ADMIN_SECURITY_POLICY, + boundBy: 'identity', + without: 'admins skip the 2FA enrolment and session-integrity checks', + }, +] as const satisfies readonly HostControl[]; + const MCP_MAX_BODY_BYTES = 1_048_576; const MCP_SDK_PACKAGE = '@modelcontextprotocol/sdk'; @@ -374,8 +401,14 @@ async function captureRawBody(req: Request): Promise { } /** - * The single composition root: wires the DI container, boots every plugin in - * `config.plugins`, and mounts the resulting oRPC router on a Hono app. + * The single composition root: wires the DI container, boots the plugins in + * `config.plugins` that `SERVICE_MANIFEST` selects (every one when it is unset), and + * mounts the resulting oRPC router on a Hono app. A manifest that leaves out a module its + * modules depend on, the module that binds a port they require, or the module behind a + * control - a selected module's `controlPorts`, or AdminGuard's - that the plugin list has, + * fails the boot with one error naming all of them; a manifest boot resolves AdminGuard + * before it serves. + * `JOB_WORKERS` picks the job queues this process consumes (every one when it is unset). * Production is distributed-only (ADR-0030, superseding ADR-0010/0014/0016/0028): * `REDIS_URL` auto-binds `JOB_QUEUE`/`RATE_LIMITER`/`CACHE`/`MESSAGE_BROKER`/ * `REALTIME_TRANSPORT` to their Redis-backed drivers - the broker being the @@ -389,23 +422,44 @@ async function captureRawBody(req: Request): Promise { * `container.dispose()` (via `close()`) runs disposers in REVERSE registration * order, so DRIZZLE is registered before JOB_QUEUE to guarantee workers drain * before the DB closes. A router namespace registered by more than one plugin - * throws at boot, not at request time. + * throws at boot, not at request time. A boot that fails disposes the container before it + * rethrows: the caller gets no app to close, so what the boot opened - the Redis client, a + * connection `configure` bound - is closed here. */ export async function createApp( config: CreateAppConfig, configure?: (container: Container) => void | Promise, ): Promise { + const container = new Container(); + try { + return await bootApp(container, config, configure); + } catch (err) { + await container + .dispose() + .catch((disposeErr: unknown) => + createLogger('create-app').error({ err: disposeErr }, 'closing a failed boot failed'), + ); + throw err; + } +} + +async function bootApp( + container: Container, + config: CreateAppConfig, + configure?: (container: Container) => void | Promise, +) { // Parsed first so a malformed entry fails the boot before anything is bound. const trustedProxies = resolveTrustedProxies( config.trustedProxies, process.env['TRUSTED_PROXIES'], ); + const jobWorkerSelection = parseJobWorkers(process.env['JOB_WORKERS']); + const manifest = parseServiceManifest(process.env['SERVICE_MANIFEST']); if (config.databaseUrl) { process.env['DATABASE_URL'] = config.databaseUrl; } - const container = new Container(); container.register(DRIZZLE, () => { const svc = new DrizzleService(); container.onDispose(() => svc.dispose()); @@ -479,10 +533,39 @@ export async function createApp( // first running every plugin's side effects. const corsOrigins = resolveCorsOrigins(config.cors); - const registry = await loadPlugins(config.plugins, container); + const bootLogger = createLogger('create-app'); + const manifestCheck: ServiceManifestCheck | null = + manifest === null + ? null + : { + entries: config.plugins, + manifest, + readPlugin: (entry) => importPlugin(entry), + hostControls: ADMIN_GUARD_CONTROLS, + isBound: (port) => container.has(port), + }; + const plugins = manifestCheck ? await selectServiceEntries(manifestCheck) : config.plugins; + if (manifestCheck) { + bootLogger.info( + { modules: plugins.map((e) => (e.noRoutes ? `${e.id} (no routes)` : e.id)) }, + 'SERVICE_MANIFEST active', + ); + } + + const registry = await loadPlugins(plugins, container, { + requiredPorts: manifestCheck ? 'defer' : 'assert', + }); await configure?.(container); assertDurableSeamsBound(container); + // After configure(), so a port the composition root binds counts. + if (manifestCheck) { + await assertServiceComplete({ + ...manifestCheck, + registered: new Set(plugins.map(({ id }) => id)), + }); + container.get(ADMIN_GUARD); + } // Every sealed token is settled once the plugins and the composition root have registered: a // holder of the returned container can neither rebind a regulator-mandated service a module @@ -562,7 +645,20 @@ export async function createApp( // Built now so a broken tool or action-type factory fails boot, not the first call. const mcpTransport = await loadMcpTransport(container, container.get(MCP_KERNEL)); - for (const registration of registry.jobs.getAll()) { + // After the router factories: a factory may register a worker too. + const workers = selectJobWorkers(registry.jobs.getAll(), jobWorkerSelection); + if (jobWorkerSelection !== null) { + const queuesOf = (list: typeof workers.consumed) => [...new Set(list.map((w) => w.queue))]; + bootLogger.info( + { + consumed: queuesOf(workers.consumed), + skipped: queuesOf(workers.skipped), + idleExclusions: workers.idleExclusions, + }, + 'JOB_WORKERS active', + ); + } + for (const registration of workers.consumed) { jobQueue.registerWorker(registration); } diff --git a/packages/core/src/server/runtime/job-workers.ts b/packages/core/src/server/runtime/job-workers.ts new file mode 100644 index 000000000..c6417bb95 --- /dev/null +++ b/packages/core/src/server/runtime/job-workers.ts @@ -0,0 +1,104 @@ +/** + * Which job queues this process consumes, from `JOB_WORKERS`. `include: null` is every queue, + * `[]` none; a pattern is a queue name or a prefix ending in `*`. + */ +export type JobWorkerSelection = { + include: readonly string[] | null; + exclude: readonly string[]; +}; + +export type SelectedJobWorkers = { + consumed: T[]; + skipped: T[]; + /** Exclusions no registered queue matched - allowed, an optional module may be absent. */ + idleExclusions: string[]; +}; + +function invalid(entry: string, rule: string): Error { + return new Error(`[create-app] JOB_WORKERS: "${entry}" ${rule}.`); +} + +function assertPattern(pattern: string, entry: string): void { + if (pattern.length === 0) { + throw invalid(entry, 'names no queue'); + } + if (pattern.slice(0, -1).includes('*')) { + throw invalid(entry, 'may use "*" only at its end, as a prefix'); + } +} + +/** + * Parses `JOB_WORKERS`: comma- or space-separated queue names and `prefix*` patterns, `!` before + * one to exclude it, `*`/`all` for every queue and `none` for none. Unset or blank is null - + * every queue, as before the variable existed. Only exclusions mean every queue but those. + */ +export function parseJobWorkers(raw: string | undefined): JobWorkerSelection | null { + const entries = (raw ?? '').split(/[,\s]+/).filter(Boolean); + if (entries.length === 0) { + return null; + } + + const include: string[] = []; + const exclude: string[] = []; + let all = false; + let none = false; + for (const entry of entries) { + if (entry.startsWith('!')) { + const pattern = entry.slice(1); + assertPattern(pattern, entry); + exclude.push(pattern); + } else if (entry === '*' || entry === 'all') { + all = true; + } else if (entry === 'none') { + none = true; + } else { + assertPattern(entry, entry); + include.push(entry); + } + } + if (none && entries.length > 1) { + throw invalid(entries.join(','), 'combines "none" with other entries; "none" stands alone'); + } + if (all && include.length > 0) { + throw invalid(entries.join(','), 'combines every queue with named ones; exclude with "!"'); + } + + return { include: none ? [] : include.length > 0 ? include : null, exclude }; +} + +function matches(pattern: string, queue: string): boolean { + return pattern.endsWith('*') ? queue.startsWith(pattern.slice(0, -1)) : queue === pattern; +} + +/** + * Splits the registered workers into the ones this process consumes and the ones it leaves to + * another. A queue named for inclusion that nothing registers fails the boot: a typo would + * otherwise leave a worker process idle while its queue backs up. + */ +export function selectJobWorkers( + registrations: readonly T[], + selection: JobWorkerSelection | null, +): SelectedJobWorkers { + if (selection === null) { + return { consumed: [...registrations], skipped: [], idleExclusions: [] }; + } + + const queues = [...new Set(registrations.map((r) => r.queue))]; + const unmatched = (selection.include ?? []).filter((p) => !queues.some((q) => matches(p, q))); + if (unmatched.length > 0) { + throw new Error( + `[create-app] JOB_WORKERS names queues this process does not register: ` + + `${unmatched.join(', ')}. Registered queues: ${queues.join(', ') || '(none)'}. ` + + 'Fix the pattern, or load the module that registers the queue (check SERVICE_MANIFEST).', + ); + } + + const consumes = (queue: string): boolean => + (selection.include === null || selection.include.some((p) => matches(p, queue))) && + !selection.exclude.some((p) => matches(p, queue)); + return { + consumed: registrations.filter((r) => consumes(r.queue)), + skipped: registrations.filter((r) => !consumes(r.queue)), + idleExclusions: selection.exclude.filter((p) => !queues.some((q) => matches(p, q))), + }; +} diff --git a/packages/core/src/wallet/plugin.ts b/packages/core/src/wallet/plugin.ts index 4ad2e9b9a..70e0ba21b 100644 --- a/packages/core/src/wallet/plugin.ts +++ b/packages/core/src/wallet/plugin.ts @@ -98,6 +98,9 @@ export default { // registered - so both ports are bound by then regardless of load order. id: 'wallet', dependsOn: ['identity', 'audit'], + controlPorts: [ + { port: RG_LIMITS, without: 'deposit and wager limits go unenforced', boundBy: 'compliance' }, + ], register(ctx) { // Set inside the router factory (container access); the job worker's handler // closes over this ref, same shape as pam/tag's evalSvc - subscriptions/workers diff --git a/packages/testing/src/__tests__/service-manifest-boot.e2e.test.ts b/packages/testing/src/__tests__/service-manifest-boot.e2e.test.ts new file mode 100644 index 000000000..d3c87e6e6 --- /dev/null +++ b/packages/testing/src/__tests__/service-manifest-boot.e2e.test.ts @@ -0,0 +1,74 @@ +import { readFileSync } from 'node:fs'; +import * as z from 'zod'; +import { describe, it, expect, beforeAll, afterAll } from 'vitest'; +import { loadExtensions } from '@openora/core/server'; +import { setupTestDb, bootTestApp, type TestDb, type TestApp } from '../index.js'; + +const ServiceExampleSchema = z.object({ name: z.string(), manifest: z.array(z.string()).min(1) }); +const createServiceExample = ServiceExampleSchema.parse( + JSON.parse( + readFileSync(new URL('../../../../tools/create/service-example.json', import.meta.url), 'utf8'), + ), +); +const SUGGESTED_MANIFEST = /^Add them: SERVICE_MANIFEST=(\S+)$/m; +const SPLIT_ENV = ['SERVICE_MANIFEST', 'SERVICE_NAME', 'REALTIME_NAMESPACE'] as const; + +let db: TestDb; +let split: TestApp; + +/** Boots every installed module under `manifest`, as a split service named `split-e2e`. */ +async function bootSplit(manifest: string) { + const saved = SPLIT_ENV.map((key) => [key, process.env[key]] as const); + process.env['SERVICE_MANIFEST'] = manifest; + process.env['SERVICE_NAME'] = 'split-e2e'; + process.env['REALTIME_NAMESPACE'] = 'split-e2e'; + try { + return await bootTestApp({ plugins: await loadExtensions(), databaseUrl: db.url }); + } finally { + for (const [key, value] of saved) { + if (value === undefined) { + delete process.env[key]; + } else { + process.env[key] = value; + } + } + } +} + +beforeAll(async () => { + process.env['BETTER_AUTH_SECRET'] ??= 'e2e-test-better-auth-secret-please-change-000000'; + process.env['AUTH_SECRET'] ??= process.env['BETTER_AUTH_SECRET']; + process.env['WITHDRAWAL_PIN_HMAC_SECRET'] ??= 'e2e-test-withdrawal-pin-hmac-secret-000000'; + process.env['NODE_ENV'] ??= 'test'; + + db = await setupTestDb(); + split = await bootSplit(createServiceExample.manifest.join(',')); +}, 60_000); + +afterAll(async () => { + await split?.close(); + await db?.dispose(); +}); + +describe('a split service booted from the pnpm create:service example', () => { + it('serves its modules, guards their admin routes, and serves none it leaves out', async () => { + expect((await split.app.request('/health')).status).toBe(200); + expect((await split.app.request('/compliance/country-rules')).status).toBe(401); + expect((await split.app.request('/cms/pages')).status).toBe(404); + }); + + it('is the manifest the first boot of SERVICE_MANIFEST=wallet tells to use', async () => { + const refusal = await bootSplit('wallet').then( + async (booted) => { + await booted.close(); + return 'SERVICE_MANIFEST=wallet booted'; + }, + (err: unknown) => (err instanceof Error ? err.message : String(err)), + ); + + expect(refusal).toMatch(/- identity \(required by wallet/); + expect(refusal).toMatch(/- wallet runs without RG_LIMITS \(bound by compliance\)/); + expect(refusal).toMatch(/- identity requires MAIL_DISPATCH \(bound by mail\)/); + expect(SUGGESTED_MANIFEST.exec(refusal)?.[1]).toBe(createServiceExample.manifest.join(',')); + }); +}); diff --git a/tools/create/create-service.ts b/tools/create/create-service.ts index 8a660c214..9549e36ff 100644 --- a/tools/create/create-service.ts +++ b/tools/create/create-service.ts @@ -8,22 +8,31 @@ * * Usage: * pnpm create:service - * pnpm create:service wallet identity,wallet + * + * Run without arguments, it prints an example that boots (service-example.json, which an E2E + * boots too). The manifest lists every module its modules depend on, every module that binds a + * port they require, and every module behind a control they would otherwise skip + * (`controlPorts`, AdminGuard's); a boot that misses any prints one manifest line adding all of + * them. The host bakes the service name as its SERVICE_NAME - its own Redis Streams consumer + * group. * * After scaffolding: `pnpm install`, build deps, then * pnpm -F @openora/-service dev # uses the baked-in manifest - * SERVICE_MANIFEST=identity,wallet,chat pnpm -F @openora/-service dev # override + * SERVICE_MANIFEST= pnpm -F @openora/-service dev # override */ import { existsSync, mkdirSync, writeFileSync, readFileSync } from 'node:fs'; import { join, dirname } from 'node:path'; import { fileURLToPath } from 'node:url'; const repoRoot = join(dirname(fileURLToPath(import.meta.url)), '../..'); +const example = JSON.parse( + readFileSync(join(repoRoot, 'tools/create/service-example.json'), 'utf8'), +) as { name: string; manifest: string[] }; const [, , rawName, rawManifest] = process.argv; if (!rawName || !rawManifest) { console.error('Usage: pnpm create:service '); - console.error('Example: pnpm create:service wallet identity,wallet'); + console.error(`Example: pnpm create:service ${example.name} ${example.manifest.join(',')}`); process.exit(1); } @@ -68,7 +77,7 @@ const files: Record = { build: 'tsc', dev: 'node --import tsx --watch src/main.ts', start: 'node dist/main.js', - typecheck: 'tsc --noEmit', + 'check:types': 'tsc --noEmit', }, dependencies: { '@openora/core': 'workspace:*', @@ -94,16 +103,22 @@ const files: Record = { ) + '\n', 'src/main.ts': `import { createApp, loadExtensions } from '@openora/core/server'; +import { user, session, account, verification, twoFactor } from '@openora/core/pam/schema/identity'; // ${name} service: boots only the modules it owns. The module code is shared with -// the monolith (root extensions.config.ts) - this host just narrows it via the -// SERVICE_MANIFEST seam. Override the default at runtime with SERVICE_MANIFEST. +// the monolith (root extensions.config.ts) - createApp narrows it via the +// SERVICE_MANIFEST seam. Override the defaults at runtime with SERVICE_MANIFEST and +// SERVICE_NAME; the name is this service's own consumer group. process.env['SERVICE_MANIFEST'] ??= '${manifest.join(',')}'; +process.env['SERVICE_NAME'] ??= '${name}'; async function bootstrap() { const plugins = await loadExtensions(); // Routes come from the loaded plugins; createApp exposes runtime API docs. - const { listen } = await createApp({ plugins }); + const { listen } = await createApp({ + plugins, + authSchema: { user, session, account, verification, twoFactor }, + }); await listen(); } @@ -121,9 +136,34 @@ for (const [rel, content] of Object.entries(files)) { console.log(`Created apps/${name} (${pkgName}) booting modules: ${manifest.join(', ')}`); console.log('Next:'); console.log(' pnpm install'); -console.log(` pnpm -F ${pkgName} dev # boots the baked-in manifest`); +console.log( + ` DATABASE_URL=... REDIS_URL=... CORS_ORIGINS=... pnpm -F ${pkgName} dev # boots the baked-in manifest`, +); console.log(` SERVICE_MANIFEST=... pnpm -F ${pkgName} dev # override at runtime`); console.log('To exclude these modules from the monolith, drop them from extensions.config.ts'); console.log( 'and run them only via this host (set REDIS_URL or bind MESSAGE_BROKER in an overlay; AMQP_URL only enables the outbox).', ); +console.warn('Before you deploy it:'); +console.warn( + ' - A module this service and another process both load handles each event in both, under each', +); +console.warn( + ' SERVICE_NAME - audit writes every row twice. Until ADR-0050 lands, split modules no other process loads.', +); +console.warn( + ' - A manifest that leaves out a module behind a control (RG limits, country checks, audit', +); +console.warn( + ' records, admin grants) refuses to boot and prints one manifest line with every module to', +); +console.warn( + ' add. Every split loads iam and identity for AdminGuard, and identity brings compliance,', +); +console.warn( + ' which depends on wallet, gaming, player-management and exchange-rate: about ten modules.', +); +console.warn( + ' - The boot logs each other optional port the manifest leaves unbound and what its module', +); +console.warn(' skips without it. Read those lines.'); diff --git a/tools/create/service-example.json b/tools/create/service-example.json new file mode 100644 index 000000000..cc292bbfa --- /dev/null +++ b/tools/create/service-example.json @@ -0,0 +1,15 @@ +{ + "name": "payments", + "manifest": [ + "wallet", + "identity", + "audit", + "compliance", + "mail", + "iam", + "player-management", + "gaming", + "exchange-rate", + "chat" + ] +} diff --git a/tools/templates/consumer/__dot__rulesync/commands/scaffold-plugin.md b/tools/templates/consumer/__dot__rulesync/commands/scaffold-plugin.md index 89c37d4e3..eb91959a1 100644 --- a/tools/templates/consumer/__dot__rulesync/commands/scaffold-plugin.md +++ b/tools/templates/consumer/__dot__rulesync/commands/scaffold-plugin.md @@ -7,7 +7,10 @@ Run `pnpm gen plugin $ARGUMENTS` in the repo root. For a vendor swap use `pnpm gen adapter `; for a react-sdk page use `pnpm gen page `. After the generator finishes, open `apps/api/src/extensions//plugin.ts` and implement -`register(ctx)`: +`register(ctx)`. It only declares: a timer, connection, file or network read, process listener, +global or `process.env` write goes in a `provide()` or router factory, closed with +`c.onDispose()`, because a failed `SERVICE_MANIFEST` boot registers plugins into a container it +then drops. - `ctx.provide(TOKEN, factory)` - bind an adapter/service (sealed compliance tokens are rejected at compile + runtime).