feat(awareness-service): Neo4j backfill and catch-all seeding

Add the one-time migration path:
- backfill-neo4j.ts reads MetaEnvelopes directly from evault-core's Neo4j
  (same node), reconstructs each packet's data payload and upserts into the
  packets table. Seeds history only - no deliveries are queued.
- SeedService ensures every platform in the registry has an approved consumer
  and a catch-all subscription, so existing webhook receivers keep working
  unchanged. Runs on every API launch and via the seed:catchall script.

Author: coodos

services/awareness-service/api/src/index.ts

@@ -11,6 +11,7 @@ import { queryRouter } from "./controllers/QueryController";
 import { subscriptionRouter } from "./controllers/SubscriptionController";
 import { AppDataSource } from "./database/data-source";
 import { DeliveryEngine } from "./services/DeliveryEngine";
+import { SeedService } from "./services/SeedService";
 
 async function start(): Promise<void> {
     await AppDataSource.initialize();
@@ -32,6 +33,10 @@ async function start(): Promise<void> {
     app.use(applicationRouter());
     app.use(adminRouter());
 
+    // Backward compat: keep every currently-registered platform receiving
+    // everything, the same way evault-core's old fanout did.
+    await new SeedService().seedCatchAll();
+
     const deliveryEngine = new DeliveryEngine();
     deliveryEngine.start();
 

services/awareness-service/api/src/scripts/backfill-neo4j.ts

@@ -0,0 +1,115 @@
+import "reflect-metadata";
+import neo4j from "neo4j-driver";
+import { AppDataSource } from "../database/data-source";
+import { Packet } from "../database/entities/Packet";
+
+/**
+ * One-time backfill. AaaS runs on the same physical node as evault-core's Neo4j,
+ * so this script reads MetaEnvelopes straight from the graph and seeds the
+ * `packets` table. It is idempotent (upsert keyed on packet id) and re-runnable.
+ *
+ * It seeds the packet store ONLY - it deliberately does not create deliveries,
+ * which would spam subscribers with the entire history on go-live.
+ */
+
+const BATCH = 500;
+
+/** Mirrors evault-core's deserializeValue for backfilled envelope values. */
+function deserialize(value: unknown, type: string): unknown {
+    if (type === "object" && typeof value === "string") {
+        try {
+            return JSON.parse(value);
+        } catch {
+            return value;
+        }
+    }
+    if (type === "array" && typeof value === "string") {
+        try {
+            return JSON.parse(value);
+        } catch {
+            return value;
+        }
+    }
+    return value;
+}
+
+async function main(): Promise<void> {
+    const uri = process.env.AWARENESS_NEO4J_URI ?? "bolt://localhost:7687";
+    const user = process.env.AWARENESS_NEO4J_USER ?? "neo4j";
+    const password = process.env.AWARENESS_NEO4J_PASSWORD ?? "neo4j";
+    const evaultPublicKey = process.env.EVAULT_PUBLIC_KEY ?? null;
+
+    const driver = neo4j.driver(uri, neo4j.auth.basic(user, password));
+    await AppDataSource.initialize();
+    const packetRepo = AppDataSource.getRepository(Packet);
+    const backfillTs = new Date();
+
+    let skip = 0;
+    let total = 0;
+
+    try {
+        for (;;) {
+            const session = driver.session();
+            let rows: any[];
+            try {
+                const result = await session.run(
+                    `MATCH (m:MetaEnvelope)-[:LINKS_TO]->(e:Envelope)
+                     RETURN m.id AS id, m.ontology AS ontology, m.eName AS eName,
+                            collect({ontology: e.ontology, value: e.value, valueType: e.valueType}) AS envelopes
+                     SKIP $skip LIMIT $batch`,
+                    { skip: neo4j.int(skip), batch: neo4j.int(BATCH) },
+                );
+                rows = result.records.map((r) => ({
+                    id: r.get("id"),
+                    ontology: r.get("ontology"),
+                    eName: r.get("eName"),
+                    envelopes: r.get("envelopes"),
+                }));
+            } finally {
+                await session.close();
+            }
+
+            if (rows.length === 0) break;
+
+            const packets = rows
+                .filter((row) => row.id && row.ontology)
+                .map((row) => {
+                    const data: Record<string, unknown> = {};
+                    for (const env of row.envelopes ?? []) {
+                        if (env?.ontology) {
+                            data[env.ontology] = deserialize(
+                                env.value,
+                                env.valueType,
+                            );
+                        }
+                    }
+                    return packetRepo.create({
+                        id: row.id,
+                        ontology: row.ontology,
+                        w3id: row.eName ?? null,
+                        evaultPublicKey,
+                        data,
+                        operation: "create" as const,
+                        receivedAt: backfillTs,
+                    });
+                });
+
+            if (packets.length > 0) {
+                await packetRepo.upsert(packets, ["id"]);
+                total += packets.length;
+            }
+            console.log(`[backfill] processed ${total} packets...`);
+            skip += BATCH;
+        }
+
+        console.log(`[backfill] complete: ${total} packets seeded`);
+    } finally {
+        await driver.close();
+        await AppDataSource.destroy();
+    }
+}
+
+main().catch((err) => {
+    console.error("[backfill] failed:", err);
+    process.exit(1);
+});

services/awareness-service/api/src/scripts/seed-catchall.ts

@@ -0,0 +1,22 @@
+import "reflect-metadata";
+import { AppDataSource } from "../database/data-source";
+import { SeedService } from "../services/SeedService";
+
+/**
+ * Standalone runner for catch-all subscription seeding. The same logic also
+ * runs automatically on every API launch; this script is for manual re-runs
+ * (e.g. after new platforms register with the registry).
+ */
+async function main(): Promise<void> {
+    await AppDataSource.initialize();
+    const result = await new SeedService().seedCatchAll();
+    console.log(
+        `[seed-catchall] complete: ${result.seeded} new / ${result.total} platforms`,
+    );
+    await AppDataSource.destroy();
+}
+
+main().catch((err) => {
+    console.error("[seed-catchall] failed:", err);
+    process.exit(1);
+});

services/awareness-service/api/src/services/SeedService.ts

@@ -0,0 +1,89 @@
+import axios from "axios";
+import { AppDataSource } from "../database/data-source";
+import { Consumer } from "../database/entities/Consumer";
+import { Subscription } from "../database/entities/Subscription";
+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 `<platform>/api/webhook`.
+ *
+ * Idempotent: existing catch-all subscriptions are left untouched.
+ */
+export class SeedService {
+    async seedCatchAll(): Promise<{ seeded: number; total: number }> {
+        if (!config.registryUrl) {
+            console.warn("[seed] PUBLIC_REGISTRY_URL not set, skipping");
+            return { seeded: 0, total: 0 };
+        }
+
+        let platforms: string[] = [];
+        try {
+            const response = await axios.get(
+                new URL("/platforms", config.registryUrl).toString(),
+                { timeout: 10000 },
+            );
+            platforms = Array.isArray(response.data) ? response.data : [];
+        } catch (err) {
+            console.error("[seed] failed to fetch registry platforms:", err);
+            return { seeded: 0, total: 0 };
+        }
+
+        const consumerRepo = AppDataSource.getRepository(Consumer);
+        const subRepo = AppDataSource.getRepository(Subscription);
+        let seeded = 0;
+
+        for (const platformUrl of platforms) {
+            let host: string;
+            let targetUrl: string;
+            try {
+                host = new URL(platformUrl).host;
+                targetUrl = new URL("/api/webhook", platformUrl).toString();
+            } catch {
+                console.warn(`[seed] skipping invalid platform: ${platformUrl}`);
+                continue;
+            }
+
+            const ename = `catchall:${host}`;
+            let consumer = await consumerRepo.findOne({ where: { ename } });
+            if (!consumer) {
+                consumer = consumerRepo.create({
+                    ename,
+                    name: host,
+                    status: "approved",
+                    webhookBaseUrl: platformUrl,
+                    approvedAt: new Date(),
+                });
+                await consumerRepo.save(consumer);
+            }
+
+            const existing = await subRepo.findOne({
+                where: {
+                    consumerId: consumer.id,
+                    isCatchAll: true,
+                    targetUrl,
+                },
+            });
+            if (!existing) {
+                await subRepo.save(
+                    subRepo.create({
+                        consumerId: consumer.id,
+                        targetUrl,
+                        ontologyFilter: [],
+                        evaultFilter: [],
+                        isCatchAll: true,
+                        active: true,
+                    }),
+                );
+                seeded += 1;
+            }
+        }
+
+        console.log(
+            `[seed] catch-all seeding done: ${seeded} new of ${platforms.length} platforms`,
+        );
+        return { seeded, total: platforms.length };
+    }
+}