Mahakisore7 · Sep 19, 2026 · Sep 19, 2026
diff --git a/apps/api/package.json b/apps/api/package.json
@@ -17,8 +17,10 @@
     "@aws-sdk/client-cognito-identity-provider": "^3.700.0",
     "@aws-sdk/client-dynamodb": "^3.700.0",
     "@aws-sdk/client-secrets-manager": "^3.700.0",
+    "@aws-sdk/client-sesv2": "^3.1136.0",
     "@aws-sdk/client-sqs": "^3.700.0",
     "@aws-sdk/lib-dynamodb": "^3.700.0",
+    "@aws-sdk/util-dynamodb": "^3.996.9",
     "@stocksync/core": "*",
     "aws-jwt-verify": "^4.0.1"
   },

diff --git a/apps/api/src/handlers/notifyAlerts.ts b/apps/api/src/handlers/notifyAlerts.ts
@@ -0,0 +1,121 @@
+import {
+  CognitoIdentityProviderClient,
+  ListUsersInGroupCommand,
+  type UserType,
+} from "@aws-sdk/client-cognito-identity-provider";
+import { SESv2Client, SendEmailCommand } from "@aws-sdk/client-sesv2";
+import { unmarshall } from "@aws-sdk/util-dynamodb";
+import type { DynamoDBStreamEvent } from "aws-lambda";
+import { createLogger } from "../lib/logger";
+import type { InventoryRecordItem } from "../lib/inventoryRecord";
+
+const logger = createLogger("notifyAlerts");
+
+const cognito = new CognitoIdentityProviderClient({});
+const ses = new SESv2Client({});
+
+const USER_POOL_ID = process.env.USER_POOL_ID ?? "";
+const FROM_EMAIL = process.env.SES_FROM_EMAIL ?? "";
+const LOW_STOCK_THRESHOLD = process.env.LOW_STOCK_THRESHOLD ? Number(process.env.LOW_STOCK_THRESHOLD) : 5;
+
+function attr(user: UserType, name: string): string | undefined {
+  return user.Attributes?.find((a) => a.Name === name)?.Value;
+}
+
+/**
+ * Owners aren't grouped per-shop in Cognito (the "owner" group holds every
+ * shop's owner) — so finding the one for this shop_id means paging through
+ * the whole group and filtering client-side by custom:shop_id. Acceptable
+ * at hackathon/demo scale; would need a shop->owner index table to stay
+ * cheap at real scale.
+ */
+async function findShopOwnerEmail(shopId: string): Promise<string | undefined> {
+  let nextToken: string | undefined;
+  do {
+    const response = await cognito.send(
+      new ListUsersInGroupCommand({ UserPoolId: USER_POOL_ID, GroupName: "owner", NextToken: nextToken }),
+    );
+    const match = (response.Users ?? []).find((user) => attr(user, "custom:shop_id") === shopId);
+    if (match) return attr(match, "email");
+    nextToken = response.NextToken;
+  } while (nextToken);
+  return undefined;
+}
+
+async function sendAlertEmail(to: string, subject: string, body: string): Promise<void> {
+  try {
+    await ses.send(
+      new SendEmailCommand({
+        FromEmailAddress: FROM_EMAIL,
+        Destination: { ToAddresses: [to] },
+        Content: { Simple: { Subject: { Data: subject }, Body: { Text: { Data: body } } } },
+      }),
+    );
+  } catch (error) {
+    // Notifications are advisory, never correctness-critical — same
+    // degrade-gracefully principle as bedrock.ts's explainer. A failed
+    // send (e.g. SES sandbox rejecting an unverified recipient) must never
+    // fail the stream-triggered invocation.
+    logger.error("SES send failed", { error, to, subject });
+  }
+}
+
+function crossedIntoLowStock(oldItem: InventoryRecordItem | undefined, newItem: InventoryRecordItem): boolean {
+  return newItem.stock <= LOW_STOCK_THRESHOLD && (!oldItem || oldItem.stock > LOW_STOCK_THRESHOLD);
+}
+
+function enteredNeedsReview(oldItem: InventoryRecordItem | undefined, newItem: InventoryRecordItem): boolean {
+  return newItem.conflict_status === "needs_review" && oldItem?.conflict_status !== "needs_review";
+}
+
+/**
+ * Triggered by inventory_records' DynamoDB Stream (enabled since Phase 1,
+ * never actually consumed until now — real-time push to connected clients
+ * goes through wsPush.ts's direct in-process call from conflictResolver.ts
+ * instead, so this stream was sitting unused). Compares OLD vs NEW images
+ * to fire an email only on the edge that matters — crossing *into* low
+ * stock, or a conflict newly appearing — never on every write to an
+ * already-low or already-flagged item, which would spam the owner on each
+ * subsequent sale.
+ */
+export async function handler(event: DynamoDBStreamEvent): Promise<void> {
+  for (const record of event.Records) {
+    try {
+      if (record.eventName === "REMOVE" || !record.dynamodb?.NewImage) continue;
+
+      const newItem = unmarshall(record.dynamodb.NewImage as Record<string, never>) as InventoryRecordItem;
+      const oldItem = record.dynamodb.OldImage
+        ? (unmarshall(record.dynamodb.OldImage as Record<string, never>) as InventoryRecordItem)
+        : undefined;
+
+      const lowStock = crossedIntoLowStock(oldItem, newItem);
+      const newConflict = enteredNeedsReview(oldItem, newItem);
+      if (!lowStock && !newConflict) continue;
+
+      const ownerEmail = await findShopOwnerEmail(newItem.shop_id);
+      if (!ownerEmail) {
+        logger.warn("no owner found for shop, skipping alert", { shop_id: newItem.shop_id });
+        continue;
+      }
+
+      if (lowStock) {
+        await sendAlertEmail(
+          ownerEmail,
+          `Low stock: ${newItem.name ?? newItem.item_id}`,
+          `${newItem.name ?? newItem.item_id} is down to ${newItem.stock} unit(s), at or below your threshold of ${LOW_STOCK_THRESHOLD}.`,
+        );
+      }
+      if (newConflict) {
+        await sendAlertEmail(
+          ownerEmail,
+          `Conflict needs review: ${newItem.name ?? newItem.item_id}`,
+          `${newItem.name ?? newItem.item_id} has a data conflict that needs your review — two counters recorded different values for the same field.`,
+        );
+      }
+    } catch (error) {
+      // One malformed/unexpected record must never block the rest of the
+      // batch or cause the stream shard to stall retrying it forever.
+      logger.error("failed to process stream record", { error });
+    }
+  }
+}
diff --git a/apps/api/test/notifyAlerts.test.ts b/apps/api/test/notifyAlerts.test.ts
@@ -0,0 +1,186 @@
+import { CognitoIdentityProviderClient, ListUsersInGroupCommand } from "@aws-sdk/client-cognito-identity-provider";
+import { SESv2Client, SendEmailCommand } from "@aws-sdk/client-sesv2";
+import { marshall } from "@aws-sdk/util-dynamodb";
+import { mockClient } from "aws-sdk-client-mock";
+import type { DynamoDBStreamEvent } from "aws-lambda";
+import { beforeAll, beforeEach, describe, expect, it } from "vitest";
+import type { InventoryRecordItem } from "../src/lib/inventoryRecord";
+
+const cognitoMock = mockClient(CognitoIdentityProviderClient);
+const sesMock = mockClient(SESv2Client);
+
+// notifyAlerts.ts reads USER_POOL_ID/SES_FROM_EMAIL once at module load —
+// same static-import-hoisting reason staffInvite.test.ts uses a dynamic
+// import behind a beforeAll.
+let handler: typeof import("../src/handlers/notifyAlerts").handler;
+
+beforeAll(async () => {
+  process.env.USER_POOL_ID = "pool-1";
+  process.env.SES_FROM_EMAIL = "alerts@shop.com";
+  process.env.LOW_STOCK_THRESHOLD = "5";
+  ({ handler } = await import("../src/handlers/notifyAlerts"));
+});
+
+function baseItem(overrides: Partial<InventoryRecordItem> = {}): InventoryRecordItem {
+  return {
+    pk: "SHOP#shop-1#ITEM#parle-g",
+    sk: "CURRENT",
+    shop_id: "shop-1",
+    item_id: "parle-g",
+    name: "Parle-G 100g",
+    base_stock: 40,
+    pn_counter: { increments: {}, decrements: {} },
+    stock: 40,
+    vector_clock: {},
+    field_last_writer: {},
+    conflict_status: "none",
+    updated_at: "2026-09-19T00:00:00.000Z",
+    ...overrides,
+  };
+}
+
+function streamEvent(params: {
+  eventName?: "INSERT" | "MODIFY" | "REMOVE";
+  oldItem?: InventoryRecordItem;
+  newItem?: InventoryRecordItem;
+}): DynamoDBStreamEvent {
+  const { eventName = "MODIFY", oldItem, newItem } = params;
+  return {
+    Records: [
+      {
+        eventName,
+        dynamodb: {
+          OldImage: oldItem ? (marshall(oldItem) as never) : undefined,
+          NewImage: newItem ? (marshall(newItem) as never) : undefined,
+        },
+      },
+    ],
+  } as unknown as DynamoDBStreamEvent;
+}
+
+const OWNER_USER = {
+  Username: "owner-1",
+  Attributes: [
+    { Name: "email", Value: "owner@shop-1.com" },
+    { Name: "custom:shop_id", Value: "shop-1" },
+  ],
+};
+
+beforeEach(() => {
+  cognitoMock.reset();
+  sesMock.reset();
+  cognitoMock.on(ListUsersInGroupCommand).resolves({ Users: [OWNER_USER] });
+  sesMock.on(SendEmailCommand).resolves({});
+});
+
+describe("notifyAlerts (inventory_records stream consumer)", () => {
+  it("sends a low-stock alert when stock crosses at/under the threshold", async () => {
+    const event = streamEvent({
+      oldItem: baseItem({ stock: 10 }),
+      newItem: baseItem({ stock: 3 }),
+    });
+    await handler(event);
+
+    expect(sesMock.commandCalls(SendEmailCommand)).toHaveLength(1);
+    const call = sesMock.commandCalls(SendEmailCommand)[0].args[0].input;
+    expect(call.Destination?.ToAddresses).toEqual(["owner@shop-1.com"]);
+    expect(call.Content?.Simple?.Subject?.Data).toMatch(/low stock/i);
+  });
+
+  it("does not re-alert when stock was already at/under the threshold", async () => {
+    const event = streamEvent({
+      oldItem: baseItem({ stock: 3 }),
+      newItem: baseItem({ stock: 2 }),
+    });
+    await handler(event);
+
+    expect(sesMock.commandCalls(SendEmailCommand)).toHaveLength(0);
+  });
+
+  it("treats a brand-new item (no OldImage) at/under the threshold as crossing in", async () => {
+    const event = streamEvent({ eventName: "INSERT", newItem: baseItem({ stock: 1 }) });
+    await handler(event);
+
+    expect(sesMock.commandCalls(SendEmailCommand)).toHaveLength(1);
+  });
+
+  it("sends a conflict alert when conflict_status newly becomes needs_review", async () => {
+    const event = streamEvent({
+      oldItem: baseItem({ conflict_status: "none" }),
+      newItem: baseItem({ conflict_status: "needs_review" }),
+    });
+    await handler(event);
+
+    expect(sesMock.commandCalls(SendEmailCommand)).toHaveLength(1);
+    const call = sesMock.commandCalls(SendEmailCommand)[0].args[0].input;
+    expect(call.Content?.Simple?.Subject?.Data).toMatch(/conflict/i);
+  });
+
+  it("does not re-alert when conflict_status was already needs_review", async () => {
+    const event = streamEvent({
+      oldItem: baseItem({ conflict_status: "needs_review" }),
+      newItem: baseItem({ conflict_status: "needs_review" }),
+    });
+    await handler(event);
+
+    expect(sesMock.commandCalls(SendEmailCommand)).toHaveLength(0);
+  });
+
+  it("sends both alerts when a single write both drops stock low and flags a conflict", async () => {
+    const event = streamEvent({
+      oldItem: baseItem({ stock: 10, conflict_status: "none" }),
+      newItem: baseItem({ stock: 1, conflict_status: "needs_review" }),
+    });
+    await handler(event);
+
+    expect(sesMock.commandCalls(SendEmailCommand)).toHaveLength(2);
+  });
+
+  it("ignores REMOVE events entirely", async () => {
+    const event = streamEvent({ eventName: "REMOVE", oldItem: baseItem({ stock: 1 }) });
+    await handler(event);
+
+    expect(cognitoMock.commandCalls(ListUsersInGroupCommand)).toHaveLength(0);
+    expect(sesMock.commandCalls(SendEmailCommand)).toHaveLength(0);
+  });
+
+  it("skips gracefully when no owner is found for the shop, never throwing", async () => {
+    cognitoMock.on(ListUsersInGroupCommand).resolves({ Users: [] });
+    const event = streamEvent({ oldItem: baseItem({ stock: 10 }), newItem: baseItem({ stock: 1 }) });
+
+    await expect(handler(event)).resolves.toBeUndefined();
+    expect(sesMock.commandCalls(SendEmailCommand)).toHaveLength(0);
+  });
+
+  it("pages through ListUsersInGroup to find the matching shop's owner", async () => {
+    cognitoMock
+      .on(ListUsersInGroupCommand)
+      .resolvesOnce({
+        Users: [{ Username: "other", Attributes: [{ Name: "email", Value: "x@other.com" }, { Name: "custom:shop_id", Value: "shop-2" }] }],
+        NextToken: "page-2",
+      })
+      .resolvesOnce({ Users: [OWNER_USER] });
+
+    const event = streamEvent({ oldItem: baseItem({ stock: 10 }), newItem: baseItem({ stock: 1 }) });
+    await handler(event);
+
+    expect(cognitoMock.commandCalls(ListUsersInGroupCommand)).toHaveLength(2);
+    expect(sesMock.commandCalls(SendEmailCommand)[0].args[0].input.Destination?.ToAddresses).toEqual(["owner@shop-1.com"]);
+  });
+
+  it("never throws when SES rejects the send (e.g. sandbox-unverified recipient)", async () => {
+    sesMock.on(SendEmailCommand).rejects(new Error("MessageRejected: Email address is not verified"));
+    const event = streamEvent({ oldItem: baseItem({ stock: 10 }), newItem: baseItem({ stock: 1 }) });
+
+    await expect(handler(event)).resolves.toBeUndefined();
+  });
+
+  it("processes remaining records even if one record is malformed", async () => {
+    const good = streamEvent({ oldItem: baseItem({ stock: 10 }), newItem: baseItem({ stock: 1 }) }).Records[0];
+    const malformed = { eventName: "MODIFY", dynamodb: { NewImage: { bogus: "not-a-valid-attribute-value" } } };
+    const event = { Records: [malformed, good] } as unknown as DynamoDBStreamEvent;
+
+    await expect(handler(event)).resolves.toBeUndefined();
+    expect(sesMock.commandCalls(SendEmailCommand)).toHaveLength(1);
+  });
+});
diff --git a/infra/cdk/lib/config.ts b/infra/cdk/lib/config.ts
@@ -29,3 +29,12 @@ export const bedrockCrossAccountId = "368339042048";
 export const bedrockCrossAccountRegion = "us-east-1";
 export const bedrockInferenceProfileArn =
   `arn:aws:bedrock:${bedrockCrossAccountRegion}:${bedrockCrossAccountId}:inference-profile/us.${bedrockModelId}`;
+
+// Phase 4 (notifications): the verified SES sender identity for low-stock/
+// conflict alert emails. SES starts every new account in sandbox mode,
+// where both the sender AND every recipient must be verified individually
+// — using the same address for both keeps this deployable without a
+// production-access request. Real per-shop delivery (a different owner's
+// inbox) only succeeds once this account exits sandbox, same kind of
+// account-level gate as the Bedrock approval above.
+export const alertsFromEmail = "kisore2004@gmail.com";