diff --git a/.claude/agents/backend-engineer.md b/.claude/agents/backend-engineer.md index dbcc18dd..a4bcb030 100644 --- a/.claude/agents/backend-engineer.md +++ b/.claude/agents/backend-engineer.md @@ -4,7 +4,7 @@ model: sonnet description: Implements ONLY the backend slice of a feature — HTTP API/services, business logic, DB schema/migrations, events, and the backend's own unit tests — against a frozen API contract. Does NOT build UI, the independent test suite, deploy wiring, or docs. Use for backend implementation in a contract-first fan-out. # Figma is reserved for frontend-engineer; pure-code agent gets core tools only (no MCP). tools: Task, Bash, Glob, Grep, LS, Read, Edit, MultiEdit, Write, NotebookEdit, WebFetch, WebSearch, TodoWrite -skills: [api-contract-first, feature-flags, verification-protocol, model-cascade] +skills: [api-contract-first, feature-flags, logging, verification-protocol, model-cascade] --- You are a **backend engineer** for FuzeFront. You implement the **backend slice only**. diff --git a/.claude/agents/frontend-engineer.md b/.claude/agents/frontend-engineer.md index 2ccbc44a..3e8d6a02 100644 --- a/.claude/agents/frontend-engineer.md +++ b/.claude/agents/frontend-engineer.md @@ -5,7 +5,7 @@ description: Implements ONLY the UI slice of a feature — a design-system-first # SOLE owner of the Figma MCP plugin (design-to-code). All other domain agents have # Figma removed from their tool grant — it is reserved here for the UI/design-system slice. tools: "*" -skills: [fuzefront-ui-package, design-system-inheritance, design-system-conformance, ui-frame-contract, frontend-design, feature-flags, ui-runtime-validation, verification-protocol, model-cascade] +skills: [fuzefront-ui-package, design-system-inheritance, design-system-conformance, ui-frame-contract, frontend-design, feature-flags, logging, ui-runtime-validation, verification-protocol, model-cascade] --- You are a **frontend engineer**. You implement the **UI slice only**. diff --git a/.claude/agents/fuzeinfra-expert.md b/.claude/agents/fuzeinfra-expert.md index 362bad69..3e088662 100644 --- a/.claude/agents/fuzeinfra-expert.md +++ b/.claude/agents/fuzeinfra-expert.md @@ -48,6 +48,13 @@ Apps attach via `networks: { FuzeInfra: { external: true } }` and reach services ## CI/CD (`.github/workflows/`) `deploy-prod.yml` (validate → ArgoCD sync), `helm-validate.yml` (helm lint + **kubeconform `-ignore-missing-schemas`** for Traefik CRDs), `infrastructure-tests.yml` (pytest against a live stack), `deploy-ec2.yml`, `claude-ci-autofix.yml` + `grafana-crit-fix.yml` (Claude-driven autofix bots), `auto-merge.yml`, `telegram-pr-merged.yml`, `update-ignore-list.yml`. +## Read access to prod for anyone (including consuming repos) — `cluster-query` +`.github/workflows/cluster-query.yml` (`workflow_dispatch`, `runs-on: staging`) is **self-service read-only `kubectl` against the prod cluster**, and it is **not** FuzeInfra-only — any repo whose token can write Actions on FuzeInfra can dispatch it. Tell consumers this rather than relaying cluster state by hand: +```bash +gh workflow run cluster-query.yml --repo izzywdev/FuzeInfra -f kubectl_args='-n get pods -o wide' +``` +Guard (executable in `tests/test_cluster_query_guard.py`): a read verb must be present (`get describe logs top events version api-resources api-versions explain cluster-info config`); every mutating/exec token is refused; **`Secret` reads and `--raw` are blocked** because FuzeInfra's job logs are **public** and a read whose *output* is a credential leaks it (this happened on 2026-07-29 with `LITELLM_MASTER_KEY`; `--raw` would have printed the runner's cluster-admin kubeconfig). SealedSecrets are readable on purpose. The dispatch credential needs **Actions: write** — `FUZEINFRA_DISPATCH_TOKEN` (Contents-only) does not cover it. Consumer-facing doc: `docs/consuming-repos/CLUSTER_QUERY.md`. Recovering a live Secret value goes through the operator SSH path in `docs/SECRETS_MANAGEMENT.md` §4, never here. + ## Gotchas (learned the hard way — verify they're still in the code) - **Prod is GitOps. Never hand-deploy or `kubectl patch`/`edit` prod resources** — ArgoCD `selfHeal` reverts out-of-band changes within seconds. Change `helm/fuzeinfra` (or values), commit to `main`, let ArgoCD sync. (This bit the Grafana dashboard fix: the kubectl patch didn't persist; it had to go through Git.) - **Grafana v13 table panels**: pre-v39 schemas fail with "Error loading: table". Fix = migrate `custom.displayMode` → `custom.cellOptions` and bump `schemaVersion` to `39` in the dashboard JSON, then ship via Git→ArgoCD. Note (as of last check): only `cluster-overview`, `fuzeinfra-services`, and `kubernetes-pods` are at v39 — `kubernetes-nodes`/`logs-explorer` are still 38 and `infrastructure-overview` is 27, so re-check before assuming a given dashboard is migrated. (Leave legitimate `legend.displayMode`/bargauge `displayMode` alone — only table-cell `custom.displayMode` needs the swap.) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2eb8d2be..aefbb8e6 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -154,9 +154,12 @@ jobs: # core + shared/kafka must be built first — the jest moduleNameMapper # resolves them from their dist/ output. - name: build core + shared (jest deps) + # shared MUST build before core — @fuzefront/core now imports + # @fuzefront/shared/kafka (the schema registry), and tsc resolves it + # through shared/dist, so shared has to be current first. run: | - npm run -w @fuzefront/core build npm run -w @fuzefront/shared build + npm run -w @fuzefront/core build - name: security — API-token jest suite (DB mocked, no Postgres) run: | diff --git a/agent-templates/schema/role-manifest.schema.json b/agent-templates/schema/role-manifest.schema.json index 87809834..f10670fd 100644 --- a/agent-templates/schema/role-manifest.schema.json +++ b/agent-templates/schema/role-manifest.schema.json @@ -80,6 +80,40 @@ "metadata": { "type": "object", "description": "Passed through as agent `metadata` (free-form tracking)." + }, + "a2a": { + "type": "object", + "additionalProperties": false, + "description": "OPTIONAL A2A discoverability/publication block. Mirrors the frozen contract FuzeAgent/agent-templates/contracts/a2a/v1/schema/role-a2a-extension.schema.json. Every field has a derived default, so no existing role.json needs it. The card projection reads role/name/description/services/metadata/coordinator regardless; this block only lets a role improve discoverability (examples/tags) or opt out of publication.", + "properties": { + "publish": { + "type": "boolean", + "default": true, + "description": "false hides this role from the public card. Still reachable on the EXTENDED card if the caller is allowlisted (authz.md §5)." + }, + "extendedOnly": { + "type": "boolean", + "default": false, + "description": "true publishes this skill ONLY on the authenticated extended card, never on the anonymous /.well-known/agent-card.json. Use for skills whose mere existence is sensitive." + }, + "tags": { + "type": "array", + "items": { "type": "string" }, + "description": "Extra tags merged with the derived tags. Derived tags are never removed." + }, + "examples": { + "type": "array", + "items": { "type": "string" }, + "description": "Example prompts a caller can send to this skill — the primary signal a calling agent uses to decide fit. Absent examples make a skill effectively undiscoverable." + }, + "inputModes": { "type": "array", "items": { "type": "string" } }, + "outputModes": { "type": "array", "items": { "type": "string" } }, + "scopes": { + "type": "array", + "items": { "type": "string" }, + "description": "OAuth scopes required for THIS skill, projected into the skill's securityRequirements." + } + } } } } diff --git a/agent-templates/sync/role_loader.py b/agent-templates/sync/role_loader.py index 9cf3685f..d71f87fe 100644 --- a/agent-templates/sync/role_loader.py +++ b/agent-templates/sync/role_loader.py @@ -4,9 +4,12 @@ followed by the base guardrail block and any role-specific `system_append`. """ import json +import logging import os import re +log = logging.getLogger(__name__) + HERE = os.path.dirname(os.path.abspath(__file__)) TEMPLATES_ROOT = os.path.dirname(HERE) # agent-templates/ REPO_ROOT = os.path.dirname(TEMPLATES_ROOT) # repo root (personas live under .claude/agents) @@ -76,14 +79,29 @@ def agent_payload(manifest): # or set-but-empty -> "") — the API rejects an empty/invalid url. Also drop the # matching mcp_toolset so the agent creates cleanly with only its configured servers; # re-provision after setting the URL to add the server + tool back. + # + # A drop is NEVER silent (FA-13): a required server logs at WARNING (its tools are + # missing until the URL is set), an `optional: true` server logs at INFO. A silent + # strip was the worst failure mode — an agent came up tool-less with no signal at all. servers = expand_env(manifest.get("mcp_servers", [])) valid, dropped = [], set() for s in servers: + # `optional` is our own hint, never part of the API payload — strip it either way. + optional = bool(s.pop("optional", False)) url = s.get("url", "") if url and "${" not in url: valid.append(s) + continue + name = s.get("name") + dropped.add(name) + reason = "url unset/empty" if not url else f"url unresolved ({url!r})" + if optional: + log.info("MCP server %r on agent %r dropped (optional): %s — continuing without it.", + name, manifest.get("name"), reason) else: - dropped.add(s.get("name")) + log.warning("MCP server %r on agent %r dropped: %s — its tools will be MISSING until " + "the URL is configured; re-provision after setting it.", + name, manifest.get("name"), reason) tools = expand_env(manifest.get("tools", [])) if dropped: tools = [t for t in tools diff --git a/backend/core/package.json b/backend/core/package.json index 64440591..be1ddfe1 100644 --- a/backend/core/package.json +++ b/backend/core/package.json @@ -10,6 +10,7 @@ "test": "jest" }, "dependencies": { + "@fuzefront/shared": "1.0.0", "cors": "^2.8.5", "express": "^4.19.2", "helmet": "^7.1.0", diff --git a/backend/core/src/events/kafkaPublisher.ts b/backend/core/src/events/kafkaPublisher.ts new file mode 100644 index 00000000..19066c33 --- /dev/null +++ b/backend/core/src/events/kafkaPublisher.ts @@ -0,0 +1,172 @@ +import { Knex } from 'knex' +import { ZodSchema } from 'zod' +import { + createKafkaClient, + TypedProducer, + FuzeEvent, + dlqTopic, + schemaForTopic, + partitionKeyForPayload, +} from '@fuzefront/shared/kafka' +import { + OutboxRecord, + OutboxRelayHandle, + startOutboxRelay, +} from './outboxRelay' + +export interface KafkaPublisherConfig { + brokers: string[] + clientId?: string +} + +export interface KafkaOutboxPublisher { + /** Publish an outbox record to Kafka (validates via the shared schema registry). */ + publish: (record: OutboxRecord) => Promise + /** Route an exhausted record to its `.dlq`. */ + deadLetter: (record: OutboxRecord) => Promise + /** Disconnect the underlying producer (graceful shutdown). */ + disconnect: () => Promise +} + +/** Minimal producer surface the publisher needs — satisfied by `TypedProducer`. */ +export interface ProducerLike { + send( + topic: string, + event: FuzeEvent, + schema: ZodSchema, + options?: { key?: string } + ): Promise + raw: { send(payload: { topic: string; messages: Array<{ key?: string; value: string }> }): Promise } + disconnect(): Promise +} + +/** + * The generic transport wiring, decoupled from how the producer is obtained so + * it is unit-testable with a fake. Builds the `FuzeEvent` envelope, derives the + * partition key, and validates against the shared schema registry (unmapped + * topics publish raw). `getProducer` is called lazily/memoised by the caller. + */ +export function makeOutboxPublisher(getProducer: () => Promise): KafkaOutboxPublisher { + const publish = async (record: OutboxRecord): Promise => { + // A connect/send failure throws → the relay leaves the row 'pending'. + const p = await getProducer() + const event: FuzeEvent = { + version: '1.0', + topic: record.topic as FuzeEvent['topic'], + correlationId: record.correlationId, + occurredAt: new Date().toISOString(), + payload: record.payload, + } + const key = partitionKeyForPayload(record.payload) + const schema = schemaForTopic(record.topic) + if (schema) { + await p.send(record.topic, event, schema as ZodSchema, { key }) + } else { + await p.raw.send({ + topic: record.topic, + messages: [{ key, value: JSON.stringify(event) }], + }) + } + } + + const deadLetter = async (record: OutboxRecord): Promise => { + const p = await getProducer() + await p.raw.send({ + topic: dlqTopic(record.topic), + messages: [ + { value: JSON.stringify({ raw: record, reason: 'outbox max attempts exhausted' }) }, + ], + }) + } + + const disconnect = async (): Promise => { + // Only disconnect a producer that was actually created. + const p = await getProducer().catch(() => null) + if (p) await p.disconnect() + } + + return { publish, deadLetter, disconnect } +} + +/** + * Builds an outbox publisher backed by a lazily-connected Kafka `TypedProducer`. + */ +export function createKafkaOutboxPublisher(config: KafkaPublisherConfig): KafkaOutboxPublisher { + let producer: TypedProducer | null = null + let connecting: Promise | null = null + + const getProducer = async (): Promise => { + if (producer) return producer + if (!connecting) { + connecting = (async () => { + const kafka = createKafkaClient({ + clientId: config.clientId || 'fuzefront-outbox-relay', + brokers: config.brokers, + }) + const p = new TypedProducer(kafka) + await p.connect() + producer = p + return p + })().catch(err => { + connecting = null // don't cache a failed connection + throw err + }) + } + return connecting + } + + const base = makeOutboxPublisher(getProducer) + return { + ...base, + disconnect: async () => { + if (producer) { + await producer.disconnect() + producer = null + connecting = null + } + }, + } +} + +export interface OutboxRelayFromEnvHandle extends OutboxRelayHandle { + disconnect: () => Promise +} + +/** + * Start the transactional-outbox relay with the Kafka transport wired from the + * environment — the one-call, install-and-go entry point for any backend + * service. Returns null (a no-op) when no broker is configured, so events stay + * durably in `event_outbox` until one is. + */ +export function startOutboxRelayFromEnv(opts: { + db: Knex + brokers?: string + clientId?: string + intervalMs?: number + logger?: { info: (m: string) => void; error: (m: string) => void } +}): OutboxRelayFromEnvHandle | null { + const brokersRaw = opts.brokers ?? process.env.KAFKA_BROKERS + if (!brokersRaw) { + opts.logger?.info('KAFKA_BROKERS unset — outbox relay disabled (events held in event_outbox)') + return null + } + const brokers = brokersRaw + .split(',') + .map(b => b.trim()) + .filter(Boolean) + + const { publish, deadLetter, disconnect } = createKafkaOutboxPublisher({ + brokers, + clientId: opts.clientId, + }) + + const handle = startOutboxRelay({ + db: opts.db, + publish, + onDeadLetter: deadLetter, + intervalMs: opts.intervalMs ?? Number(process.env.OUTBOX_RELAY_INTERVAL_MS || 1000), + logger: opts.logger, + }) + + return { stop: handle.stop, disconnect } +} diff --git a/backend/core/src/index.ts b/backend/core/src/index.ts index 27e86b6d..f76e0446 100644 --- a/backend/core/src/index.ts +++ b/backend/core/src/index.ts @@ -8,3 +8,4 @@ export * from './types/shared' export * from './bootstrap' export * from './events/outbox' export * from './events/outboxRelay' +export * from './events/kafkaPublisher' diff --git a/backend/core/tests/events/kafkaPublisher.test.ts b/backend/core/tests/events/kafkaPublisher.test.ts new file mode 100644 index 00000000..a03c1b9a --- /dev/null +++ b/backend/core/tests/events/kafkaPublisher.test.ts @@ -0,0 +1,101 @@ +import { makeOutboxPublisher, startOutboxRelayFromEnv, ProducerLike } from '../../src/events/kafkaPublisher' +import { OutboxRecord } from '../../src/events/outboxRelay' + +type SendCall = { topic: string; event: any; hasSchema: boolean; key?: string } +type RawCall = { topic: string; key?: string; value: string } + +function fakeProducer() { + const sends: SendCall[] = [] + const raws: RawCall[] = [] + let disconnected = false + const producer: ProducerLike = { + async send(topic, event, schema, options) { + sends.push({ topic, event, hasSchema: !!schema, key: options?.key }) + }, + raw: { + async send(payload) { + for (const m of payload.messages) raws.push({ topic: payload.topic, key: m.key, value: m.value }) + return undefined + }, + }, + async disconnect() { + disconnected = true + }, + } + return { producer, sends, raws, get disconnected() { return disconnected } } +} + +const rec = (topic: string, payload: unknown, correlationId = 'c1'): OutboxRecord => ({ + id: 'id-1', + topic, + payload, + correlationId, + attempts: 0, +}) + +describe('makeOutboxPublisher', () => { + it('validates against the registry schema and keys by organizationId for a mapped topic', async () => { + const f = fakeProducer() + const pub = makeOutboxPublisher(async () => f.producer) + + await pub.publish(rec('identity.org.created', { organizationId: 'org-9', slug: 'acme' })) + + expect(f.sends).toHaveLength(1) + expect(f.raws).toHaveLength(0) + const call = f.sends[0] + expect(call.topic).toBe('identity.org.created') + expect(call.hasSchema).toBe(true) + expect(call.key).toBe('org-9') + expect(call.event).toMatchObject({ + version: '1.0', + topic: 'identity.org.created', + correlationId: 'c1', + payload: { organizationId: 'org-9', slug: 'acme' }, + }) + expect(typeof call.event.occurredAt).toBe('string') + }) + + it('publishes raw (no schema) for an unmapped topic, still keyed by the entity id', async () => { + const f = fakeProducer() + const pub = makeOutboxPublisher(async () => f.producer) + + await pub.publish(rec('billing.trial.ending', { userId: 'user-3' })) + + expect(f.sends).toHaveLength(0) + expect(f.raws).toHaveLength(1) + expect(f.raws[0].topic).toBe('billing.trial.ending') + expect(f.raws[0].key).toBe('user-3') + expect(JSON.parse(f.raws[0].value)).toMatchObject({ topic: 'billing.trial.ending', payload: { userId: 'user-3' } }) + }) + + it('dead-letters to .dlq', async () => { + const f = fakeProducer() + const pub = makeOutboxPublisher(async () => f.producer) + + await pub.deadLetter(rec('identity.org.created', { organizationId: 'org-9' })) + + expect(f.raws).toHaveLength(1) + expect(f.raws[0].topic).toBe('identity.org.created.dlq') + expect(JSON.parse(f.raws[0].value)).toMatchObject({ reason: 'outbox max attempts exhausted' }) + }) + + it('propagates a producer failure so the relay keeps the row pending', async () => { + const pub = makeOutboxPublisher(async () => { + throw new Error('kafka down') + }) + await expect(pub.publish(rec('identity.org.created', { organizationId: 'o' }))).rejects.toThrow('kafka down') + }) +}) + +describe('startOutboxRelayFromEnv', () => { + it('is a no-op (returns null) when no broker is configured', () => { + const logs: string[] = [] + const handle = startOutboxRelayFromEnv({ + db: {} as any, + brokers: '', + logger: { info: m => logs.push(m), error: () => undefined }, + }) + expect(handle).toBeNull() + expect(logs.join(' ')).toMatch(/disabled/) + }) +}) diff --git a/backend/security/src/services/outboxRelay.ts b/backend/security/src/services/outboxRelay.ts index dd219534..9dd104d6 100644 --- a/backend/security/src/services/outboxRelay.ts +++ b/backend/security/src/services/outboxRelay.ts @@ -1,79 +1,11 @@ -// Outbox relay wiring for security-service. +// Outbox relay for security-service. // -// The generic drain loop lives in @fuzefront/core (startOutboxRelay); this -// module injects the Kafka transport: it maps each topic to its frozen Zod -// schema, derives the partition key (entity id) for per-entity ordering, and -// dead-letters rows that exhaust their retries. One relay per shared DB drains -// the `event_outbox` table written transactionally by the route handlers. -import { - createKafkaClient, - TypedProducer, - FuzeEvent, - dlqTopic, - TOPICS, - identityUserCreatedSchemaV1, - identityUserUpdatedSchemaV1, - identityUserDeletedSchemaV1, - identityOrgCreatedSchemaV1, - identityOrgUpdatedSchemaV1, - identityOrgDeletedSchemaV1, - identityMembershipAddedSchemaV1, - identityMembershipRemovedSchemaV1, - notifyEmailRequestedSchemaV1, - portalCreatedSchemaV1, -} from '@fuzefront/shared/kafka' -import { - db, - startOutboxRelay, - OutboxRecord, - OutboxRelayHandle, -} from '@fuzefront/core' -import type { ZodSchema } from 'zod' - -// Topic -> payload schema. Publishing validates against the frozen schema; an -// unmapped topic is published WITHOUT validation (raw) so a newly-emitted topic -// never gets its rows stuck 'pending' before its schema is added here. -const SCHEMA_BY_TOPIC: Record> = { - [TOPICS.IDENTITY_USER_CREATED]: identityUserCreatedSchemaV1, - [TOPICS.IDENTITY_USER_UPDATED]: identityUserUpdatedSchemaV1, - [TOPICS.IDENTITY_USER_DELETED]: identityUserDeletedSchemaV1, - [TOPICS.IDENTITY_ORG_CREATED]: identityOrgCreatedSchemaV1, - [TOPICS.IDENTITY_ORG_UPDATED]: identityOrgUpdatedSchemaV1, - [TOPICS.IDENTITY_ORG_DELETED]: identityOrgDeletedSchemaV1, - [TOPICS.IDENTITY_MEMBERSHIP_ADDED]: identityMembershipAddedSchemaV1, - [TOPICS.IDENTITY_MEMBERSHIP_REMOVED]: identityMembershipRemovedSchemaV1, - [TOPICS.NOTIFY_EMAIL_REQUESTED]: notifyEmailRequestedSchemaV1, - [TOPICS.PORTAL_CREATED]: portalCreatedSchemaV1, -} - -// Kafka message key = entity id, so all events for one org/user land on a -// single partition and stay ordered. -function partitionKey(payload: any): string | undefined { - return ( - payload?.organizationId ?? - payload?.userId ?? - payload?.portalId ?? - payload?.entityId ?? - undefined - ) -} - -let producer: TypedProducer | null = null -async function getProducer(): Promise { - if (producer) return producer - const brokers = (process.env.KAFKA_BROKERS as string) - .split(',') - .map(b => b.trim()) - .filter(Boolean) - const kafka = createKafkaClient({ - clientId: process.env.KAFKA_CLIENT_ID || 'fuzefront-outbox-relay', - brokers, - }) - const p = new TypedProducer(kafka) - await p.connect() - producer = p - return p -} +// The generic wiring (topic→schema validation, partition-key derivation, Kafka +// publish/DLQ adapter) now lives in @fuzefront/core so every backend service — +// and the Python `fuzefront-events` mirror — shares one install-and-go surface. +// This module is a thin binding: it hands core the service's `db` singleton and +// a labelled logger. +import { db, startOutboxRelayFromEnv, OutboxRelayHandle } from '@fuzefront/core' /** * Start the transactional-outbox relay if a Kafka broker is configured. Returns @@ -81,56 +13,9 @@ async function getProducer(): Promise { * `event_outbox` (and reconcile-on-login still provisions), exactly as before. */ export function startOutboxRelayIfConfigured(): OutboxRelayHandle | null { - if (!process.env.KAFKA_BROKERS) { - console.log( - 'ℹ️ KAFKA_BROKERS unset — outbox relay disabled (events held in event_outbox)' - ) - return null - } - - const publish = async (record: OutboxRecord): Promise => { - // A connect/send failure throws → the row stays 'pending' and is retried. - const p = await getProducer() - const event: FuzeEvent = { - version: '1.0', - topic: record.topic as FuzeEvent['topic'], - correlationId: record.correlationId, - occurredAt: new Date().toISOString(), - payload: record.payload, - } - const key = partitionKey(record.payload) - const schema = SCHEMA_BY_TOPIC[record.topic] - if (schema) { - await p.send(record.topic, event, schema, { key }) - } else { - await p.raw.send({ - topic: record.topic, - messages: [{ key, value: JSON.stringify(event) }], - }) - } - } - - const onDeadLetter = async (record: OutboxRecord): Promise => { - const p = await getProducer() - await p.raw.send({ - topic: dlqTopic(record.topic), - messages: [ - { - value: JSON.stringify({ - raw: record, - reason: 'outbox max attempts exhausted', - }), - }, - ], - }) - } - - console.log('🚀 Starting outbox relay (event_outbox → Kafka)') - return startOutboxRelay({ + return startOutboxRelayFromEnv({ db, - publish, - onDeadLetter, - intervalMs: Number(process.env.OUTBOX_RELAY_INTERVAL_MS || 1000), + clientId: process.env.KAFKA_CLIENT_ID || 'fuzefront-outbox-relay', logger: { info: m => console.log(`[outbox-relay] ${m}`), error: m => console.error(`[outbox-relay] ${m}`), diff --git a/shared/src/kafka/index.ts b/shared/src/kafka/index.ts index be3ec9b7..eca778e6 100644 --- a/shared/src/kafka/index.ts +++ b/shared/src/kafka/index.ts @@ -1,6 +1,7 @@ // Kafka client + schemas — filled in by later tasks export * from './types'; export * from './schemas'; +export * from './registry'; export * from './client'; export * from './producer'; export * from './consumer'; diff --git a/shared/src/kafka/registry.ts b/shared/src/kafka/registry.ts new file mode 100644 index 00000000..f2463bc7 --- /dev/null +++ b/shared/src/kafka/registry.ts @@ -0,0 +1,75 @@ +import { ZodTypeAny } from 'zod'; +import { TOPICS } from './types'; +import { + appRegisteredSchemaV1, + appActivatedSchemaV1, + appSuspendedSchemaV1, + appHeartbeatSchemaV1, + billingLlmUsageSchemaV1, + billingUsageRecordedSchemaV1, + billingSubscriptionChangedSchemaV1, + billingPaymentCompletedSchemaV1, + identityUserCreatedSchemaV1, + identityUserUpdatedSchemaV1, + identityUserDeletedSchemaV1, + identityOrgCreatedSchemaV1, + identityOrgUpdatedSchemaV1, + identityOrgDeletedSchemaV1, + identityMembershipAddedSchemaV1, + identityMembershipRemovedSchemaV1, + notifyEmailRequestedSchemaV1, + notifyEmailStatusSchemaV1, + portalCreatedSchemaV1, +} from './schemas'; + +/** + * Single source of truth mapping each topic to the frozen Zod schema for its + * payload. Producers and the outbox relay look schemas up here to validate + * before publishing; a topic with no entry is published without validation. + * + * Any language binding (e.g. the Python `fuzefront-events` package) mirrors this + * registry so validation and topic coverage stay identical across the family. + */ +export const SCHEMA_BY_TOPIC: Readonly> = { + [TOPICS.APP_REGISTERED]: appRegisteredSchemaV1, + [TOPICS.APP_ACTIVATED]: appActivatedSchemaV1, + [TOPICS.APP_SUSPENDED]: appSuspendedSchemaV1, + [TOPICS.APP_HEARTBEAT]: appHeartbeatSchemaV1, + [TOPICS.BILLING_LLM_USAGE]: billingLlmUsageSchemaV1, + [TOPICS.BILLING_USAGE_RECORDED]: billingUsageRecordedSchemaV1, + [TOPICS.BILLING_SUBSCRIPTION_CHANGED]: billingSubscriptionChangedSchemaV1, + [TOPICS.BILLING_PAYMENT_COMPLETED]: billingPaymentCompletedSchemaV1, + [TOPICS.IDENTITY_USER_CREATED]: identityUserCreatedSchemaV1, + [TOPICS.IDENTITY_USER_UPDATED]: identityUserUpdatedSchemaV1, + [TOPICS.IDENTITY_USER_DELETED]: identityUserDeletedSchemaV1, + [TOPICS.IDENTITY_ORG_CREATED]: identityOrgCreatedSchemaV1, + [TOPICS.IDENTITY_ORG_UPDATED]: identityOrgUpdatedSchemaV1, + [TOPICS.IDENTITY_ORG_DELETED]: identityOrgDeletedSchemaV1, + [TOPICS.IDENTITY_MEMBERSHIP_ADDED]: identityMembershipAddedSchemaV1, + [TOPICS.IDENTITY_MEMBERSHIP_REMOVED]: identityMembershipRemovedSchemaV1, + [TOPICS.NOTIFY_EMAIL_REQUESTED]: notifyEmailRequestedSchemaV1, + [TOPICS.NOTIFY_EMAIL_STATUS]: notifyEmailStatusSchemaV1, + [TOPICS.PORTAL_CREATED]: portalCreatedSchemaV1, +}; + +/** Returns the payload schema for a topic, or undefined if none is registered. */ +export function schemaForTopic(topic: string): ZodTypeAny | undefined { + return SCHEMA_BY_TOPIC[topic]; +} + +/** + * Derives the Kafka partition key from an event payload so all events for one + * entity stay ordered on a single partition. Falls through the common id fields; + * returns undefined when none is present (round-robin). Defined here — in the + * contract — so every language binding derives the key identically. + */ +export function partitionKeyForPayload(payload: any): string | undefined { + return ( + payload?.organizationId ?? + payload?.userId ?? + payload?.portalId ?? + payload?.entityId ?? + payload?.appId ?? + undefined + ); +} diff --git a/shared/tests/kafka/registry.test.ts b/shared/tests/kafka/registry.test.ts new file mode 100644 index 00000000..9d04924a --- /dev/null +++ b/shared/tests/kafka/registry.test.ts @@ -0,0 +1,57 @@ +import { + SCHEMA_BY_TOPIC, + schemaForTopic, + partitionKeyForPayload, + TOPICS, +} from '../../src/kafka'; + +describe('SCHEMA_BY_TOPIC / schemaForTopic', () => { + it('resolves the lifecycle topics to a validating schema', () => { + const lifecycle = [ + TOPICS.IDENTITY_ORG_CREATED, + TOPICS.IDENTITY_ORG_UPDATED, + TOPICS.IDENTITY_ORG_DELETED, + TOPICS.IDENTITY_USER_UPDATED, + TOPICS.IDENTITY_USER_DELETED, + TOPICS.IDENTITY_MEMBERSHIP_ADDED, + TOPICS.IDENTITY_MEMBERSHIP_REMOVED, + ]; + for (const topic of lifecycle) { + const schema = schemaForTopic(topic); + expect(schema).toBeDefined(); + // the resolved schema actually validates its payload family + expect(typeof schema!.safeParse).toBe('function'); + } + }); + + it('returns undefined for an unmapped topic (raw publish path)', () => { + expect(schemaForTopic('billing.trial.ending')).toBeUndefined(); + expect(schemaForTopic('not.a.topic')).toBeUndefined(); + }); + + it('every registered key is a known topic value', () => { + const topicValues = new Set(Object.values(TOPICS)); + for (const key of Object.keys(SCHEMA_BY_TOPIC)) { + expect(topicValues.has(key)).toBe(true); + } + }); + + it('the org.created schema resolved via the registry rejects a bad payload', () => { + const schema = schemaForTopic(TOPICS.IDENTITY_ORG_CREATED)!; + expect(schema.safeParse({ organizationId: 'not-a-uuid' }).success).toBe(false); + }); +}); + +describe('partitionKeyForPayload', () => { + it('prefers organizationId, then userId, then portalId', () => { + expect(partitionKeyForPayload({ organizationId: 'o1', userId: 'u1' })).toBe('o1'); + expect(partitionKeyForPayload({ userId: 'u1', portalId: 'p1' })).toBe('u1'); + expect(partitionKeyForPayload({ portalId: 'p1' })).toBe('p1'); + expect(partitionKeyForPayload({ entityId: 'e1' })).toBe('e1'); + }); + + it('returns undefined when no id field is present', () => { + expect(partitionKeyForPayload({ foo: 'bar' })).toBeUndefined(); + expect(partitionKeyForPayload(null)).toBeUndefined(); + }); +});