diff --git a/docs/docs/Services/Awareness-as-a-Service.md b/docs/docs/Services/Awareness-as-a-Service.md index a00b10911..bfbb7db0e 100644 --- a/docs/docs/Services/Awareness-as-a-Service.md +++ b/docs/docs/Services/Awareness-as-a-Service.md @@ -141,11 +141,11 @@ AaaS is designed to be dropped in with **zero receiver-side changes**: 1. **Backfill.** AaaS runs on the same node as evault-core's Neo4j. The `backfill` script reads existing MetaEnvelopes straight from the graph and seeds the `packets` table (history only — it does not queue deliveries). -2. **Catch-all seeding.** On every launch, AaaS ensures each platform currently - in the registry has an approved consumer and a catch-all subscription - pointing at `/api/webhook`. Existing platforms therefore keep - receiving every packet exactly as before, and can later narrow their - subscriptions to specific ontologies or eVaults. +2. **Catch-all reconciliation.** On every launch and once per configured sync + interval, AaaS ensures each platform currently in the registry has an + approved consumer and an active catch-all subscription pointing at + `/api/webhook`. Existing and newly registered platforms therefore + keep receiving every packet exactly as before. 3. **evault-core switch.** evault-core's `deliverWebhooks`/`getActivePlatforms` are removed; a single `notifyAwareness` POST forwards each packet to AaaS. @@ -162,6 +162,7 @@ AaaS is designed to be dropped in with **zero receiver-side changes**: | `AAAS_JWT_SECRET` | Signs portal session JWTs | | `AWARENESS_MAX_ATTEMPTS` | Delivery attempts before dead-lettering (default 3) | | `AWARENESS_DELIVERY_POLL_MS` | Delivery engine poll interval (default 2000) | +| `AWARENESS_REGISTRY_SYNC_MS` | Registry catch-all reconciliation interval (default 60000; 0 disables periodic sync) | | `NEO4J_URI` / `NEO4J_USER` / `NEO4J_PASSWORD` | Standard eVault Neo4j vars — reused by the one-time backfill | | `PUBLIC_AWARENESS_API_URL` | (portal) AaaS API base URL | @@ -172,6 +173,6 @@ AaaS is designed to be dropped in with **zero receiver-side changes**: pnpm --filter awareness-service-api build pnpm --filter awareness-service-api migration:run pnpm --filter awareness-service-api backfill # one-time, from Neo4j -pnpm --filter awareness-service-api dev # API (seeds catch-all on launch) +pnpm --filter awareness-service-api dev # API (keeps registry catch-alls synced) pnpm --filter awareness-portal dev # portal ``` diff --git a/infrastructure/evault-core/src/core/protocol/graphql-server.ts b/infrastructure/evault-core/src/core/protocol/graphql-server.ts index f54e059ab..d68f0b5e8 100644 --- a/infrastructure/evault-core/src/core/protocol/graphql-server.ts +++ b/infrastructure/evault-core/src/core/protocol/graphql-server.ts @@ -113,24 +113,44 @@ export class GraphQLServer { return; } - try { - await axios.post( - new URL( - "/ingest", - process.env.AWARENESS_SERVICE_URL, - ).toString(), - { ...webhookPayload, requestingPlatform }, - { - headers: { - "Content-Type": "application/json", - "x-ingest-secret": - process.env.AWARENESS_INGEST_SECRET ?? "", + const ingestUrl = new URL( + "/ingest", + process.env.AWARENESS_SERVICE_URL, + ).toString(); + const maxAttempts = 3; + + for (let attempt = 1; attempt <= maxAttempts; attempt += 1) { + try { + const response = await axios.post( + ingestUrl, + { ...webhookPayload, requestingPlatform }, + { + headers: { + "Content-Type": "application/json", + "x-ingest-secret": + process.env.AWARENESS_INGEST_SECRET ?? "", + }, + timeout: 5000, }, - timeout: 5000, - }, - ); - } catch (error) { - console.log("Awareness ingest delivery failed"); + ); + console.log( + `[webhook] AaaS accepted id=${webhookPayload?.id} status=${response.status} attempt=${attempt}`, + ); + return; + } catch (error: any) { + const status = error?.response?.status ?? "no-response"; + const code = error?.code ?? "unknown"; + const message = error?.message ?? "unknown error"; + console.error( + `[webhook] AaaS ingest failed id=${webhookPayload?.id} status=${status} code=${code} attempt=${attempt}/${maxAttempts}: ${message}`, + ); + + if (attempt < maxAttempts) { + await new Promise((resolve) => + setTimeout(resolve, 250 * 2 ** (attempt - 1)), + ); + } + } } } @@ -388,6 +408,7 @@ export class GraphQLServer { evaultPublicKey: this.evaultPublicKey, data: input.payload, schemaId: input.ontology, + operation: "create" as const, }; // Fire-and-forget ingest to AaaS @@ -532,6 +553,7 @@ export class GraphQLServer { evaultPublicKey: this.evaultPublicKey, data: result.mergedPayload ?? input.payload, schemaId: input.ontology, + operation: "update" as const, }; // Fire-and-forget ingest to AaaS @@ -756,6 +778,7 @@ export class GraphQLServer { evaultPublicKey: this.evaultPublicKey, data: input.payload, schemaId: input.ontology, + operation: "create" as const, }; // Fire-and-forget ingest to AaaS @@ -927,6 +950,7 @@ export class GraphQLServer { data: result.bindingDocument, schemaId: BINDING_DOCUMENT_ONTOLOGY, + operation: "create" as const, }; this.notifyAwareness( webhookPayload, @@ -1033,6 +1057,7 @@ export class GraphQLServer { data: result, schemaId: BINDING_DOCUMENT_ONTOLOGY, + operation: "update" as const, }; this.notifyAwareness( webhookPayload, @@ -1217,6 +1242,7 @@ export class GraphQLServer { evaultPublicKey: this.evaultPublicKey, data: input.payload, schemaId: input.ontology, + operation: "create" as const, }; this.notifyAwareness( @@ -1470,6 +1496,7 @@ export class GraphQLServer { evaultPublicKey: this.evaultPublicKey, data: result.mergedPayload ?? input.payload, schemaId: input.ontology, + operation: "update" as const, }; // Fire-and-forget ingest to AaaS diff --git a/services/awareness-service/README.md b/services/awareness-service/README.md index 7f804ff43..c9f9f8d70 100644 --- a/services/awareness-service/README.md +++ b/services/awareness-service/README.md @@ -59,7 +59,9 @@ Service**. ## Backward compatibility -On launch AaaS seeds a catch-all subscription for every platform currently in -the registry, so existing webhook receivers keep getting every packet at -`/api/webhook` with no change. Consumers can later narrow to specific -ontologies / eVaults. +On launch and periodically thereafter, AaaS reconciles a catch-all subscription +for every platform currently in the registry, so existing and newly registered +webhook receivers keep getting every packet at `/api/webhook` with no +change. `AWARENESS_REGISTRY_SYNC_MS` controls the interval (default 60000; set +to 0 to disable periodic reconciliation). Non-registry consumers can narrow +their own subscriptions to specific ontologies / eVaults. diff --git a/services/awareness-service/api/src/config.ts b/services/awareness-service/api/src/config.ts index 71df8fc4b..e67c45047 100644 --- a/services/awareness-service/api/src/config.ts +++ b/services/awareness-service/api/src/config.ts @@ -11,6 +11,15 @@ function required(name: string): string { return value; } +function timerInterval(name: string, fallback: number): number { + const raw = process.env[name]; + if (raw === undefined || !/^\d+$/.test(raw)) return fallback; + const value = Number(raw); + return Number.isSafeInteger(value) && value <= 2_147_483_647 + ? value + : fallback; +} + export const config = { /** Postgres connection string for the AaaS database. */ databaseUrl: process.env.AWARENESS_DATABASE_URL ?? "", @@ -32,6 +41,8 @@ export const config = { process.env.AWARENESS_DELIVERY_POLL_MS ?? "2000", 10, ), + /** How often registry platforms are reconciled with catch-all subscriptions. */ + registrySyncMs: timerInterval("AWARENESS_REGISTRY_SYNC_MS", 60000), /** Public base URL of the AaaS API, used to build W3DS auth callbacks. */ publicUrl: process.env.AWARENESS_PUBLIC_URL ?? "http://localhost:4100", dbCaCert: process.env.DB_CA_CERT, diff --git a/services/awareness-service/api/src/controllers/AdminController.ts b/services/awareness-service/api/src/controllers/AdminController.ts index 936215e88..564feb4d7 100644 --- a/services/awareness-service/api/src/controllers/AdminController.ts +++ b/services/awareness-service/api/src/controllers/AdminController.ts @@ -4,6 +4,8 @@ import { AccessApplication } from "../database/entities/AccessApplication"; import { Consumer } from "../database/entities/Consumer"; import { DeadLetter } from "../database/entities/DeadLetter"; import { Delivery } from "../database/entities/Delivery"; +import { Packet } from "../database/entities/Packet"; +import { Subscription } from "../database/entities/Subscription"; import { adminAuth } from "../middleware/portalAuth"; /** @@ -33,6 +35,40 @@ export function adminRouter(): Router { res.json({ applications: apps }); }); + // Inspect effective targets without exposing webhook signing secrets. + router.get("/api/admin/subscriptions", async (req, res, next) => { + try { + const target = + typeof req.query.target === "string" ? req.query.target : null; + const qb = AppDataSource.getRepository(Subscription) + .createQueryBuilder("s") + .innerJoin(Consumer, "c", "c.id = s.consumerId") + .select([ + 's.id AS "subscriptionId"', + 's.targetUrl AS "targetUrl"', + 's.isCatchAll AS "isCatchAll"', + 's.active AS "active"', + 's.ontologyFilter AS "ontologyFilter"', + 's.evaultFilter AS "evaultFilter"', + 's.createdAt AS "createdAt"', + 'c.id AS "consumerId"', + 'c.ename AS "consumerEname"', + 'c.status AS "consumerStatus"', + 'c.webhookBaseUrl AS "webhookBaseUrl"', + ]) + .orderBy("s.createdAt", "ASC"); + if (target) { + qb.where("s.targetUrl ILIKE :target", { + target: `%${target}%`, + }); + } + const subscriptions = await qb.getRawMany(); + res.json({ count: subscriptions.length, subscriptions }); + } catch (error) { + next(error); + } + }); + router.post("/api/admin/applications/:id/approve", async (req, res) => { const appRepo = AppDataSource.getRepository(AccessApplication); const application = await appRepo.findOne({ @@ -100,6 +136,54 @@ export function adminRouter(): Router { res.json({ deadLetters }); }); + // Definitive end-to-end trace for one awareness packet: every matched + // subscription, resolved target, owning consumer and delivery outcome. + router.get("/api/admin/packets/:id/deliveries", async (req, res, next) => { + try { + const rows = await AppDataSource.getRepository(Delivery) + .createQueryBuilder("d") + .innerJoin(Subscription, "s", "s.id = d.subscriptionId") + .innerJoin(Consumer, "c", "c.id = s.consumerId") + .select([ + 'd.id AS "deliveryId"', + 'd.packetId AS "packetId"', + 'd.status AS "status"', + 'd.attempts AS "attempts"', + 'd.nextAttemptAt AS "nextAttemptAt"', + 'd.deliveredAt AS "deliveredAt"', + 'd.lastResponseStatus AS "lastResponseStatus"', + 'd.lastError AS "lastError"', + 's.id AS "subscriptionId"', + 's.targetUrl AS "targetUrl"', + 's.isCatchAll AS "isCatchAll"', + 's.active AS "subscriptionActive"', + 's.ontologyFilter AS "ontologyFilter"', + 's.evaultFilter AS "evaultFilter"', + 'c.id AS "consumerId"', + 'c.ename AS "consumerEname"', + 'c.status AS "consumerStatus"', + ]) + .where("d.packetId = :packetId", { packetId: req.params.id }) + .orderBy("d.createdAt", "ASC") + .getRawMany(); + + const packetExists = await AppDataSource.getRepository( + Packet, + ).exists({ + where: { id: req.params.id }, + }); + + res.json({ + packetId: req.params.id, + packetExists, + deliveryCount: rows.length, + deliveries: rows, + }); + } catch (error) { + next(error); + } + }); + // Replay re-queues the original delivery and resolves the dead letter. router.post("/api/admin/dead-letters/:id/replay", async (req, res) => { const dlRepo = AppDataSource.getRepository(DeadLetter); diff --git a/services/awareness-service/api/src/database/entities/Delivery.ts b/services/awareness-service/api/src/database/entities/Delivery.ts index 2e997c305..1c0170da2 100644 --- a/services/awareness-service/api/src/database/entities/Delivery.ts +++ b/services/awareness-service/api/src/database/entities/Delivery.ts @@ -41,6 +41,11 @@ export class Delivery { @Column({ type: "varchar" }) contentHash!: string; + /** Immutable event snapshot; prevents later packet upserts changing this delivery. */ + @Column({ type: "jsonb", nullable: true }) + // `any` avoids TypeORM DeepPartial rejecting arbitrary JSON object values. + payload!: any; + @Column({ type: "varchar", default: "pending" }) status!: DeliveryStatus; diff --git a/services/awareness-service/api/src/database/migrations/1780404367749-AddDeliveryPayload.ts b/services/awareness-service/api/src/database/migrations/1780404367749-AddDeliveryPayload.ts new file mode 100644 index 000000000..171eb0f26 --- /dev/null +++ b/services/awareness-service/api/src/database/migrations/1780404367749-AddDeliveryPayload.ts @@ -0,0 +1,18 @@ +import { MigrationInterface, QueryRunner } from "typeorm"; + +/** Preserve the exact event body on each queued delivery. */ +export class AddDeliveryPayload1780404367749 implements MigrationInterface { + name = "AddDeliveryPayload1780404367749"; + + public async up(queryRunner: QueryRunner): Promise { + await queryRunner.query( + `ALTER TABLE "deliveries" ADD "payload" jsonb`, + ); + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.query( + `ALTER TABLE "deliveries" DROP COLUMN "payload"`, + ); + } +} diff --git a/services/awareness-service/api/src/index.ts b/services/awareness-service/api/src/index.ts index 7a71799df..53cdde070 100644 --- a/services/awareness-service/api/src/index.ts +++ b/services/awareness-service/api/src/index.ts @@ -50,7 +50,9 @@ async function start(): Promise { // Backward compat: keep every currently-registered platform receiving // everything, the same way evault-core's old fanout did. - await new SeedService().seedCatchAll(); + const seedService = new SeedService(); + await seedService.syncCatchAll(); + seedService.start(); const deliveryEngine = new DeliveryEngine(); deliveryEngine.start(); diff --git a/services/awareness-service/api/src/services/DeliveryEngine.ts b/services/awareness-service/api/src/services/DeliveryEngine.ts index 378d43eb5..0d79cd7ed 100644 --- a/services/awareness-service/api/src/services/DeliveryEngine.ts +++ b/services/awareness-service/api/src/services/DeliveryEngine.ts @@ -59,8 +59,33 @@ export class DeliveryEngine { this.running = true; try { const claimed = await this.claimBatch(); - for (const delivery of claimed) { - await this.attemptDelivery(delivery); + // A slow or malicious subscription must not block unrelated + // platforms in the same batch. Each request has its own timeout; + // drain the claimed batch concurrently so failures are isolated. + const results = await Promise.allSettled( + claimed.map((delivery) => this.attemptDelivery(delivery)), + ); + const rejected = results + .map((result, index) => ({ result, delivery: claimed[index] })) + .filter( + (item): item is { + result: PromiseRejectedResult; + delivery: Delivery; + } => item.result.status === "rejected", + ); + for (const { result, delivery } of rejected) { + const message = + result.reason instanceof Error + ? result.reason.message + : String(result.reason); + await AppDataSource.getRepository(Delivery).update(delivery.id, { + status: "failed", + lastError: message, + nextAttemptAt: new Date(), + }); + console.error( + `[aaas] delivery ${delivery.id} task failed before completion: ${message}`, + ); } this.dbDown = false; } catch (err) { @@ -97,7 +122,13 @@ export class DeliveryEngine { // already hit the cap) are terminal and never re-claimed. `id IN ( SELECT id FROM deliveries - WHERE status IN ('pending', 'failed') + WHERE ( + status IN ('pending', 'failed') + OR ( + status = 'delivering' + AND "nextAttemptAt" <= now() - interval '1 minute' + ) + ) AND attempts < :maxAttempts AND "nextAttemptAt" <= now() ORDER BY "nextAttemptAt" @@ -124,7 +155,7 @@ export class DeliveryEngine { where: { id: delivery.packetId }, }); - if (!subscription || !packet) { + if (!subscription || (!packet && !delivery.payload)) { await this.fail( delivery, subscription, @@ -134,12 +165,16 @@ export class DeliveryEngine { return; } - const payload: AwarenessPayload = { - id: packet.id, - w3id: packet.w3id, - evaultPublicKey: packet.evaultPublicKey, - data: packet.data, - schemaId: packet.ontology, + // New rows carry an immutable snapshot so rapid updates to the same + // envelope cannot overwrite an older event before it is delivered. + // Fall back to Packet for deliveries created before this column existed. + const payload: AwarenessPayload = delivery.payload ?? { + id: packet!.id, + w3id: packet!.w3id, + evaultPublicKey: packet!.evaultPublicKey, + data: packet!.data, + schemaId: packet!.ontology, + operation: packet!.operation, }; const headers: Record = { diff --git a/services/awareness-service/api/src/services/IngestService.ts b/services/awareness-service/api/src/services/IngestService.ts index 4351a94a1..aab5a942c 100644 --- a/services/awareness-service/api/src/services/IngestService.ts +++ b/services/awareness-service/api/src/services/IngestService.ts @@ -60,7 +60,17 @@ export class IngestService { // Dedupe deliveries by payload content rather than packet id alone: // re-ingesting an updated envelope (new hash) must queue a new delivery, // while a retried POST of the same payload (same hash) must not. - const hash = contentHash(payload.data ?? null); + // The operation is part of the event identity. A delete can carry the + // same (often null) data as another event and must still be delivered. + const deliveryPayload: AwarenessPayload = { + id: payload.id, + w3id: payload.w3id ?? null, + evaultPublicKey: payload.evaultPublicKey ?? null, + data: payload.data ?? null, + schemaId: payload.schemaId, + operation: payload.operation ?? "create", + }; + const hash = contentHash(deliveryPayload); const deliveryRepo = AppDataSource.getRepository(Delivery); const rows = subscriptions.map((sub) => @@ -68,6 +78,7 @@ export class IngestService { subscriptionId: sub.id, packetId: packet.id, contentHash: hash, + payload: deliveryPayload, status: "pending", attempts: 0, nextAttemptAt: new Date(), diff --git a/services/awareness-service/api/src/services/SeedService.ts b/services/awareness-service/api/src/services/SeedService.ts index ca3032162..5f12ecba2 100644 --- a/services/awareness-service/api/src/services/SeedService.ts +++ b/services/awareness-service/api/src/services/SeedService.ts @@ -6,13 +6,44 @@ import { config } from "../config"; /** * Backward-compat seeding. Before AaaS, evault-core fanned out every webhook to - * every registered platform. To preserve that behaviour, on launch we ensure - * each platform currently in the registry has an approved consumer and a - * catch-all subscription (empty filters) pointing at `/api/webhook`. + * every registered platform. To preserve that behaviour, on launch and at a + * configured interval we ensure each platform currently in the registry has + * an approved consumer and an active catch-all subscription (empty filters) + * pointing at `/api/webhook`. * - * Idempotent: existing catch-all subscriptions are left untouched. + * Idempotent: valid existing catch-all subscriptions are reused. */ export class SeedService { + private timer?: NodeJS.Timeout; + private syncing = false; + + start(): void { + if (!config.registryUrl || config.registrySyncMs <= 0) return; + this.timer = setInterval(() => { + void this.syncCatchAll().catch((err) => { + console.error("[seed] registry reconciliation failed:", err); + }); + }, config.registrySyncMs); + console.log( + `[seed] registry reconciliation started (poll ${config.registrySyncMs}ms)`, + ); + } + + stop(): void { + if (this.timer) clearInterval(this.timer); + } + + /** Prevent overlapping registry requests when one reconciliation is slow. */ + async syncCatchAll(): Promise<{ seeded: number; total: number }> { + if (this.syncing) return { seeded: 0, total: 0 }; + this.syncing = true; + try { + return await this.seedCatchAll(); + } finally { + this.syncing = false; + } + } + async seedCatchAll(): Promise<{ seeded: number; total: number }> { if (!config.registryUrl) { console.warn("[seed] PUBLIC_REGISTRY_URL not set, skipping"); @@ -34,6 +65,7 @@ export class SeedService { const consumerRepo = AppDataSource.getRepository(Consumer); const subRepo = AppDataSource.getRepository(Subscription); let seeded = 0; + const currentTargets = new Map(); for (const platformUrl of platforms) { let host: string; @@ -47,6 +79,7 @@ export class SeedService { } const ename = `catchall:${host}`; + currentTargets.set(ename, targetUrl); let consumer = await consumerRepo.findOne({ where: { ename } }); if (!consumer) { consumer = consumerRepo.create({ @@ -57,6 +90,14 @@ export class SeedService { approvedAt: new Date(), }); await consumerRepo.save(consumer); + } else { + // Registry-level consumers are managed by this compatibility + // sync. Keep them deliverable even if an earlier subscription + // or consumer record was disabled. + consumer.status = "approved"; + consumer.webhookBaseUrl = platformUrl; + consumer.approvedAt ??= new Date(); + await consumerRepo.save(consumer); } const existing = await subRepo.findOne({ @@ -78,11 +119,45 @@ export class SeedService { }), ); seeded += 1; + } else if ( + !existing.active || + existing.ontologyFilter.length > 0 || + existing.evaultFilter.length > 0 + ) { + existing.active = true; + existing.ontologyFilter = []; + existing.evaultFilter = []; + await subRepo.save(existing); + seeded += 1; + } + } + + const managedSubscriptions = await subRepo + .createQueryBuilder("s") + .innerJoin(Consumer, "c", "c.id = s.consumerId") + .addSelect('c.ename', "consumerEname") + .where("s.isCatchAll = true") + .andWhere("s.active = true") + .andWhere("c.ename LIKE :prefix", { prefix: "catchall:%" }) + .getRawAndEntities(); + + for ( + let index = 0; + index < managedSubscriptions.entities.length; + index += 1 + ) { + const subscription = managedSubscriptions.entities[index]; + const ename = managedSubscriptions.raw[index] + .consumerEname as string; + if (currentTargets.get(ename) !== subscription.targetUrl) { + subscription.active = false; + await subRepo.save(subscription); + seeded += 1; } } console.log( - `[seed] catch-all seeding done: ${seeded} new of ${platforms.length} platforms`, + `[seed] catch-all reconciliation done: ${seeded} changed of ${platforms.length} platforms`, ); return { seeded, total: platforms.length }; }