fix(awareness-service): dedupe deliveries by content hash so envelope updates re-deliver (#984) (#988)

Author: coodos

services/awareness-service/api/src/database/entities/Delivery.ts

@@ -16,11 +16,16 @@ export type DeliveryStatus =
 
 /**
  * A queued webhook delivery of one packet to one subscription. The unique
- * (subscriptionId, packetId) constraint makes ingest idempotent if evault-core
- * retries a POST.
+ * (subscriptionId, packetId, contentHash) constraint dedupes by content: a
+ * retried POST of the same payload is idempotent, while re-ingesting an updated
+ * envelope (new contentHash) queues a fresh delivery.
  */
 @Entity("deliveries")
-@Unique("uq_delivery_subscription_packet", ["subscriptionId", "packetId"])
+@Unique("uq_delivery_subscription_packet_content", [
+    "subscriptionId",
+    "packetId",
+    "contentHash",
+])
 export class Delivery {
     @PrimaryGeneratedColumn("uuid")
     id!: string;
@@ -32,6 +37,10 @@ export class Delivery {
     @Column({ type: "varchar" })
     packetId!: string;
 
+    /** SHA-256 of the packet payload at ingest; dedupes deliveries by content. */
+    @Column({ type: "varchar" })
+    contentHash!: string;
+
     @Column({ type: "varchar", default: "pending" })
     status!: DeliveryStatus;
 

services/awareness-service/api/src/database/migrations/1780000000000-AddDeliveryContentHash.ts

@@ -0,0 +1,47 @@
+import { MigrationInterface, QueryRunner } from "typeorm";
+
+/**
+ * Dedupe deliveries by payload content, not packet id alone. Adds a
+ * `contentHash` column to `deliveries` and swaps the unique key from
+ * (subscriptionId, packetId) to (subscriptionId, packetId, contentHash) so that
+ * re-ingesting an updated MetaEnvelope queues a fresh delivery instead of being
+ * silently dropped by the old constraint.
+ */
+export class AddDeliveryContentHash1780000000000
+    implements MigrationInterface
+{
+    name = "AddDeliveryContentHash1780000000000";
+
+    public async up(queryRunner: QueryRunner): Promise<void> {
+        await queryRunner.query(
+            `ALTER TABLE "deliveries" ADD COLUMN "contentHash" varchar`,
+        );
+        // Backfill legacy rows with a single constant: this preserves the old
+        // one-delivery-per-(subscription, packet) semantics for rows that
+        // predate content-based dedup.
+        await queryRunner.query(
+            `UPDATE "deliveries" SET "contentHash" = '' WHERE "contentHash" IS NULL`,
+        );
+        await queryRunner.query(
+            `ALTER TABLE "deliveries" ALTER COLUMN "contentHash" SET NOT NULL`,
+        );
+        await queryRunner.query(
+            `ALTER TABLE "deliveries" DROP CONSTRAINT "uq_delivery_subscription_packet"`,
+        );
+        await queryRunner.query(
+            `ALTER TABLE "deliveries" ADD CONSTRAINT "uq_delivery_subscription_packet_content" UNIQUE ("subscriptionId", "packetId", "contentHash")`,
+        );
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<void> {
+        await queryRunner.query(
+            `ALTER TABLE "deliveries" DROP CONSTRAINT "uq_delivery_subscription_packet_content"`,
+        );
+        await queryRunner.query(
+            `ALTER TABLE "deliveries" ADD CONSTRAINT "uq_delivery_subscription_packet" UNIQUE ("subscriptionId", "packetId")`,
+        );
+        await queryRunner.query(
+            `ALTER TABLE "deliveries" DROP COLUMN "contentHash"`,
+        );
+    }
+}

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

@@ -2,6 +2,7 @@ import { AppDataSource } from "../database/data-source";
 import { Delivery } from "../database/entities/Delivery";
 import { Packet } from "../database/entities/Packet";
 import type { AwarenessPayload } from "../types";
+import { contentHash } from "../utils/contentHash";
 import { SubscriptionMatcher } from "./SubscriptionMatcher";
 
 /** Returns the normalised origin of a URL, or null if it cannot be parsed. */
@@ -56,19 +57,27 @@ export class IngestService {
             return { packetId: packet.id, deliveriesQueued: 0 };
         }
 
+        // 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);
+
         const deliveryRepo = AppDataSource.getRepository(Delivery);
         const rows = subscriptions.map((sub) =>
             deliveryRepo.create({
                 subscriptionId: sub.id,
                 packetId: packet.id,
+                contentHash: hash,
                 status: "pending",
                 attempts: 0,
                 nextAttemptAt: new Date(),
             }),
         );
 
-        // orIgnore skips deliveries that already exist for this packet/sub pair,
-        // so an evault-core retry of POST /ingest does not double-deliver.
+        // orIgnore skips deliveries that already exist for this
+        // (subscription, packet, content) triple, so an evault-core retry of
+        // POST /ingest with unchanged data does not double-deliver. An updated
+        // payload has a different contentHash and is inserted as a new delivery.
         const result = await deliveryRepo
             .createQueryBuilder()
             .insert()

services/awareness-service/api/src/utils/contentHash.ts

@@ -0,0 +1,37 @@
+import crypto from "crypto";
+
+/**
+ * Deterministic JSON serialisation: object keys are sorted recursively so that
+ * two payloads with the same content but different key ordering serialise
+ * identically. Array order is preserved (it is semantically significant) and
+ * primitives/null pass through unchanged.
+ */
+export function stableStringify(value: unknown): string {
+    if (value === null || typeof value !== "object") {
+        return JSON.stringify(value) ?? "null";
+    }
+    if (Array.isArray(value)) {
+        return `[${value.map(stableStringify).join(",")}]`;
+    }
+    const entries = Object.keys(value as Record<string, unknown>)
+        .sort()
+        .map(
+            (key) =>
+                `${JSON.stringify(key)}:${stableStringify(
+                    (value as Record<string, unknown>)[key],
+                )}`,
+        );
+    return `{${entries.join(",")}}`;
+}
+
+/**
+ * SHA-256 of a packet's payload, computed over a stable serialisation so the
+ * hash depends only on content, not key ordering. Used to dedupe deliveries by
+ * content: identical re-ingests share a hash, a changed payload yields a new one.
+ */
+export function contentHash(data: unknown): string {
+    return crypto
+        .createHash("sha256")
+        .update(stableStringify(data ?? null))
+        .digest("hex");
+}