fix(marketplace): consume the agreement-event dead-letter queue

apps/api/src/services/marketplace-event.ts

@@ -0,0 +1,140 @@
+/**
+ * Shared AWS Marketplace event correlation.
+ *
+ * Sized to be importable by the lightweight DLQ consumer: it touches
+ * `@wraps/db` only, not the SES/React-Email stack the main worker pulls in for
+ * the welcome email. Both `marketplace-events` and `marketplace-dlq-consumer`
+ * correlate events through the same rules so the two cannot drift.
+ */
+
+import {
+  and,
+  awsMarketplaceSubscription,
+  db,
+  desc,
+  eq,
+  ilike,
+} from "@wraps/db";
+
+export type MarketplaceEvent = {
+  id?: string;
+  "detail-type"?: string;
+  source?: string;
+  time?: string;
+  detail?: {
+    agreement?: { id?: string; status?: string };
+    product?: { code?: string; id?: string };
+    license?: { arn?: string };
+    acceptor?: { accountId?: string };
+    offer?: { id?: string };
+  };
+};
+
+/**
+ * Detail types the worker acts on. Everything else is acknowledged and
+ * dropped — the rule is deliberately broader than this map so a new event
+ * type shows up in logs instead of vanishing.
+ *
+ * `Purchase Agreement Created` doubles as the activation signal because it is
+ * unverified whether a Free-pricing product emits `License Updated` at all
+ * (entitlements are a contract-pricing concept). Whichever arrives first wins;
+ * the second is a no-op.
+ */
+export const ACTIVATING = new Set([
+  "License Updated - Manufacturer",
+  "Purchase Agreement Created - Proposer",
+  "Purchase Agreement Created - Manufacturer",
+]);
+
+export const DEACTIVATING = new Set([
+  "License Deprovisioned - Manufacturer",
+  "Purchase Agreement Ended - Proposer",
+  "Purchase Agreement Ended - Manufacturer",
+]);
+
+export const ADVISORY = new Set([
+  "Purchase Agreement Advisory Issued - Manufacturer",
+]);
+
+/**
+ * AWS's own documentation prints licence ARNs inconsistently — the License
+ * event example omits the leading `arn:` that ResolveCustomer returns. Compare
+ * on the licence id (`l-…`) so a formatting difference cannot silently prevent
+ * every correlation.
+ */
+export function licenseKey(arn: string | undefined): string | null {
+  if (!arn) {
+    return null;
+  }
+  const match = arn.match(/l-[0-9a-f]+/i);
+  return match ? match[0].toLowerCase() : arn.toLowerCase();
+}
+
+/**
+ * Correlation, most specific first. Licence ARN is the only truly unique key
+ * under Concurrent Agreements; account + product is a last resort that can
+ * legitimately match several rows, so it takes the newest deliberately rather
+ * than whichever Postgres happens to return.
+ */
+export async function findSubscription(keys: {
+  arn?: string;
+  agreementId?: string;
+  accountId?: string;
+  productCode?: string;
+}) {
+  if (keys.arn) {
+    // Exact hit on the unique index covers the normal case.
+    const [exact] = await db
+      .select()
+      .from(awsMarketplaceSubscription)
+      .where(eq(awsMarketplaceSubscription.licenseArn, keys.arn))
+      .limit(1);
+    if (exact) {
+      return exact;
+    }
+
+    // Formatting mismatch fallback — match on the `l-…` id alone. Bounded by
+    // LIMIT rather than loading the table into memory.
+    const key = licenseKey(keys.arn);
+    if (key?.startsWith("l-")) {
+      const [suffix] = await db
+        .select()
+        .from(awsMarketplaceSubscription)
+        .where(ilike(awsMarketplaceSubscription.licenseArn, `%${key}%`))
+        .limit(1);
+      if (suffix) {
+        return suffix;
+      }
+    }
+  }
+
+  if (keys.agreementId) {
+    const [byAgreement] = await db
+      .select()
+      .from(awsMarketplaceSubscription)
+      .where(eq(awsMarketplaceSubscription.agreementId, keys.agreementId))
+      .limit(1);
+    if (byAgreement) {
+      return byAgreement;
+    }
+  }
+
+  if (keys.accountId && keys.productCode) {
+    const [byAccount] = await db
+      .select()
+      .from(awsMarketplaceSubscription)
+      .where(
+        and(
+          eq(awsMarketplaceSubscription.customerAwsAccountId, keys.accountId),
+          eq(awsMarketplaceSubscription.productCode, keys.productCode)
+        )
+      )
+      .orderBy(desc(awsMarketplaceSubscription.resolvedAt))
+      .limit(1);
+    if (byAccount) {
+      return byAccount;
+    }
+  }
+
+  return null;
+}

apps/api/src/workers/__tests__/marketplace-dlq-consumer.test.ts

@@ -0,0 +1,202 @@
+/**
+ * AWS Marketplace DLQ consumer.
+ *
+ * Pins the behaviours that make the dead-letter path safe: a dead-lettered
+ * lifecycle event marks its subscription `failed` (terminal and visible, so it
+ * no longer expires silently), a row that already left `pending` keeps its
+ * status (a whole-batch timeout dead-letters records that were applied on an
+ * earlier attempt), anything without a row or without an actionable
+ * detail type is surfaced but writes nothing, and the handler never throws —
+ * a DLQ consumer has no DLQ of its own to catch it.
+ */
+
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const updateSet = vi.fn();
+const updateWhere = vi.fn();
+const updateReturning = vi.fn();
+const selectWhere = vi.fn();
+const captureException = vi.fn();
+
+vi.mock("../../lib/sentry", () => ({}));
+vi.mock("@sentry/aws-serverless", () => ({
+  captureException: (...args: unknown[]) => captureException(...args),
+  wrapHandler: (fn: unknown) => fn,
+}));
+vi.mock("../../lib/logger", () => ({
+  log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() },
+  flushLogger: () => Promise.resolve(),
+}));
+
+vi.mock("@wraps/db", () => ({
+  awsMarketplaceSubscription: {
+    id: "id",
+    licenseArn: "license_arn",
+    agreementId: "agreement_id",
+    customerAwsAccountId: "customer_aws_account_id",
+    productCode: "product_code",
+    resolvedAt: "resolved_at",
+    status: "status",
+  },
+  and: (...a: unknown[]) => ({ and: a }),
+  desc: (col: unknown) => ({ desc: col }),
+  eq: (col: unknown, val: unknown) => ({ eq: [col, val] }),
+  ilike: (col: unknown, val: unknown) => ({ ilike: [col, val] }),
+  db: {
+    select: () => ({
+      from: () => ({
+        where: (w: unknown) => {
+          const tail = { limit: () => Promise.resolve(selectWhere(w)) };
+          // The account+product fallback orders before limiting.
+          return { ...tail, orderBy: () => tail };
+        },
+      }),
+    }),
+    update: () => ({
+      set: (v: unknown) => ({
+        where: (w: unknown) => {
+          updateSet(v);
+          updateWhere(w);
+          const result = Promise.resolve([]);
+          return Object.assign(result, {
+            returning: () => Promise.resolve(updateReturning()),
+          });
+        },
+      }),
+    }),
+  },
+}));
+
+const { handler } = await import("../marketplace-dlq-consumer");
+
+// The SQSHandler signature takes (event, context, callback); the test only
+// needs the event, and the handler ignores the rest.
+const invoke = (event: unknown) =>
+  handler(event as never, {} as never, undefined as never);
+
+const LICENSE_ARN =
+  "arn:aws:license-manager:us-east-1:905130073023:l-e52ca6f38bf84d0fafb8802ca15ac11a";
+
+function sqsEvent(detailType: string, detail: unknown) {
+  return {
+    Records: [
+      {
+        messageId: "m1",
+        body: JSON.stringify({
+          id: "evt-1",
+          "detail-type": detailType,
+          source: "aws.agreement-marketplace",
+          time: "2026-09-08T12:00:00Z",
+          detail,
+        }),
+      },
+    ],
+    // biome-ignore lint/suspicious/noExplicitAny: minimal SQS shape for the handler
+  } as any;
+}
+
+const matchedRow = {
+  id: "row-1",
+  licenseArn: LICENSE_ARN,
+  status: "pending",
+};
+
+describe("marketplace DLQ consumer", () => {
+  beforeEach(() => {
+    updateSet.mockReset();
+    updateWhere.mockReset();
+    updateReturning.mockReset();
+    selectWhere.mockReset();
+    captureException.mockReset();
+    selectWhere.mockReturnValue([matchedRow]);
+    updateReturning.mockReturnValue([{ id: matchedRow.id }]);
+  });
+
+  it("marks a correlated subscription failed", async () => {
+    await invoke(
+      sqsEvent("License Updated - Manufacturer", {
+        license: { arn: LICENSE_ARN },
+      })
+    );
+
+    expect(updateSet).toHaveBeenCalledTimes(1);
+    expect(updateSet.mock.calls[0][0]).toMatchObject({ status: "failed" });
+    expect(captureException).toHaveBeenCalledTimes(1);
+  });
+
+  it("only fails a row that is still pending", async () => {
+    await invoke(
+      sqsEvent("License Updated - Manufacturer", {
+        license: { arn: LICENSE_ARN },
+      })
+    );
+
+    expect(updateWhere).toHaveBeenCalledWith({
+      and: [{ eq: ["id", "row-1"] }, { eq: ["status", "pending"] }],
+    });
+  });
+
+  it("leaves an already-settled row alone but still surfaces it", async () => {
+    // A batch timeout redelivers records that were already applied; after the
+    // retries they dead-letter alongside the one that actually failed.
+    selectWhere.mockReturnValue([{ ...matchedRow, status: "active" }]);
+    updateReturning.mockReturnValue([]);
+
+    await invoke(
+      sqsEvent("Purchase Agreement Ended - Manufacturer", {
+        agreement: { id: "agr-1" },
+      })
+    );
+
+    expect(captureException).toHaveBeenCalledTimes(1);
+    expect(captureException.mock.calls[0][1]).toMatchObject({
+      tags: { stage: "settled" },
+      extra: { subscriptionId: "row-1", status: "active" },
+    });
+  });
+
+  it("marks failed for a deactivation event too", async () => {
+    await invoke(
+      sqsEvent("Purchase Agreement Ended - Manufacturer", {
+        agreement: { id: "agr-1" },
+      })
+    );
+
+    expect(updateSet).toHaveBeenCalledTimes(1);
+    expect(updateSet.mock.calls[0][0]).toMatchObject({ status: "failed" });
+  });
+
+  it("surfaces but writes nothing when no subscription matches", async () => {
+    selectWhere.mockReturnValue([]);
+
+    await invoke(
+      sqsEvent("License Updated - Manufacturer", {
+        license: { arn: LICENSE_ARN },
+      })
+    );
+
+    expect(updateSet).not.toHaveBeenCalled();
+    expect(captureException).toHaveBeenCalledTimes(1);
+  });
+
+  it("writes nothing for a non-actionable detail type", async () => {
+    await invoke(
+      sqsEvent("Purchase Agreement Advisory Issued - Manufacturer", {})
+    );
+
+    expect(selectWhere).not.toHaveBeenCalled();
+    expect(updateSet).not.toHaveBeenCalled();
+    expect(captureException).toHaveBeenCalledTimes(1);
+  });
+
+  it("never throws on a malformed record", async () => {
+    const event = {
+      Records: [{ messageId: "m2", body: "not json" }],
+      // biome-ignore lint/suspicious/noExplicitAny: minimal SQS shape
+    } as any;
+
+    await expect(invoke(event)).resolves.toBeUndefined();
+    expect(updateSet).not.toHaveBeenCalled();
+    expect(captureException).toHaveBeenCalledTimes(1);
+  });
+});