fix(moodlog): acknowledge durable source writes

src/app/api/integrations/moodlog/webhook/__tests__/route.test.ts

@@ -192,6 +192,23 @@ describe("POST /api/integrations/moodlog/webhook", () => {
     expect(wheres[0]).toEqual(wheres[1]);
   });
 
+  it("returns retryable failure when the source record cannot be persisted", async () => {
+    vi.mocked(prisma.moodEntry.upsert).mockRejectedValueOnce(
+      new Error("database unavailable"),
+    );
+
+    const res = await POST(
+      jsonRequest(VALID_PAYLOAD, { "x-webhook-secret": "plaintext-secret" }),
+    );
+
+    expect(res.status).toBe(503);
+    expect(await res.json()).toMatchObject({
+      data: null,
+      error: expect.stringMatching(/persist/i),
+    });
+    expect(prisma.moodEntry.upsert).toHaveBeenCalledOnce();
+  });
+
   it("mood.deleted removes any matching entry rather than upserting", async () => {
     const deletedPayload = {
       ...VALID_PAYLOAD,

src/app/api/integrations/moodlog/webhook/route.ts

@@ -6,6 +6,7 @@ import { checkRateLimit } from "@/lib/rate-limit";
 import { getClientIp } from "@/lib/api-response";
 import { moodLogWebhookPayloadSchema } from "@/lib/validations/moodlog";
 import { readMoodLogSecret } from "@/lib/moodlog-secret";
+import { persistMoodLogSourceEntry } from "@/lib/moodlog/persistence";
 import { apiHandler } from "@/lib/api-handler";
 import { annotate, getEvent } from "@/lib/logging/context";
 import { recomputeMoodBucketsForEntry } from "@/lib/rollups/mood-rollups";
@@ -97,7 +98,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
   const { event, entry } = parsed.data;
   annotate({ meta: { webhook_event: event } });
 
-  // 5. Process event (non-blocking — fire and forget with error logging)
+  // 5. Process event and acknowledge only after the source write commits.
   const moodLoggedAt = new Date(entry.time);
   // v1.12.1 — when the upstream supplies a stable entry id, dedup on
   // `(userId, source, externalId)` so a re-emit with a re-rounded /
@@ -116,67 +117,14 @@ export const POST = apiHandler(async (request: NextRequest) => {
             : { date: entry.date, moodLoggedAt }),
         },
       });
-    } else if (externalId) {
-      // mood.created or mood.updated, keyed on the stable external id.
-      await prisma.moodEntry.upsert({
-        where: {
-          userId_source_externalId: {
-            userId: user.id,
-            source: "MOODLOG",
-            externalId,
-          },
-        },
-        update: {
-          mood: entry.mood,
-          score: entry.score,
-          tags: entry.tags ? JSON.stringify(entry.tags) : null,
-          source: "MOODLOG",
-          date: entry.date,
-          moodLoggedAt,
-          // No-op on a live row, a deliberate RESURRECTION on a tombstoned
-          // one — moodLog is the source of truth for the rows it minted; a
-          // created/updated event means the entry exists upstream (deletion
-          // arrives as its own `mood.deleted` event).
-          deletedAt: null,
-        },
-        create: {
-          userId: user.id,
-          date: entry.date,
-          mood: entry.mood,
-          score: entry.score,
-          tags: entry.tags ? JSON.stringify(entry.tags) : null,
-          source: "MOODLOG",
-          externalId,
-          moodLoggedAt,
-        },
-      });
     } else {
-      // mood.created or mood.updated, legacy wall-clock key.
-      await prisma.moodEntry.upsert({
-        where: {
-          userId_date_moodLoggedAt: {
-            userId: user.id,
-            date: entry.date,
-            moodLoggedAt,
-          },
-        },
-        update: {
-          mood: entry.mood,
-          score: entry.score,
-          tags: entry.tags ? JSON.stringify(entry.tags) : null,
-          source: "MOODLOG",
-          // Same resurrection rule as the externalId-keyed upsert above.
-          deletedAt: null,
-        },
-        create: {
-          userId: user.id,
-          date: entry.date,
-          mood: entry.mood,
-          score: entry.score,
-          tags: entry.tags ? JSON.stringify(entry.tags) : null,
-          source: "MOODLOG",
-          moodLoggedAt,
-        },
+      await persistMoodLogSourceEntry(user.id, {
+        externalId,
+        date: entry.date,
+        moodLoggedAt,
+        mood: entry.mood,
+        score: entry.score,
+        tags: entry.tags,
       });
     }
 
@@ -193,9 +141,11 @@ export const POST = apiHandler(async (request: NextRequest) => {
       );
     }
   } catch (err) {
-    getEvent()?.addWarning("DB operation failed: " + err);
+    const message = err instanceof Error ? err.message : String(err);
+    getEvent()?.addWarning(`MoodLog source persistence failed: ${message}`);
+    return apiError("Failed to persist moodLog event", 503);
   }
 
-  // 6. Return 200 immediately
+  // 6. The source record is durable; the provider can stop retrying.
   return new Response(null, { status: 200 });
 });

src/lib/moodlog/__tests__/persistence.test.ts

@@ -0,0 +1,64 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+vi.mock("@/lib/db", () => ({
+  prisma: {
+    moodEntry: {
+      upsert: vi.fn(),
+      updateMany: vi.fn(),
+      findUnique: vi.fn(),
+      update: vi.fn(),
+    },
+  },
+}));
+
+import { prisma } from "@/lib/db";
+import { persistMoodLogSourceEntry } from "../persistence";
+
+describe("persistMoodLogSourceEntry concurrent legacy adoption", () => {
+  beforeEach(() => {
+    vi.resetAllMocks();
+  });
+
+  it("applies the losing replay payload to the row adopted by the winner", async () => {
+    const moodLoggedAt = new Date("2026-07-19T08:30:00.000Z");
+    vi.mocked(prisma.moodEntry.upsert).mockRejectedValueOnce(
+      Object.assign(new Error("natural-key collision"), { code: "P2002" }),
+    );
+    vi.mocked(prisma.moodEntry.updateMany).mockResolvedValueOnce({
+      count: 0,
+    });
+    vi.mocked(prisma.moodEntry.findUnique).mockResolvedValueOnce({
+      date: "2026-07-19",
+      moodLoggedAt,
+    } as never);
+    vi.mocked(prisma.moodEntry.update).mockResolvedValueOnce({} as never);
+
+    await persistMoodLogSourceEntry("user-1", {
+      externalId: "provider-event-1",
+      date: "2026-07-19",
+      moodLoggedAt,
+      mood: "SUPER_GUT",
+      score: 5,
+      tags: ["updated"],
+    });
+
+    expect(prisma.moodEntry.update).toHaveBeenCalledWith({
+      where: {
+        userId_source_externalId: {
+          userId: "user-1",
+          source: "MOODLOG",
+          externalId: "provider-event-1",
+        },
+      },
+      data: {
+        mood: "SUPER_GUT",
+        score: 5,
+        tags: JSON.stringify(["updated"]),
+        source: "MOODLOG",
+        deletedAt: null,
+        date: "2026-07-19",
+        moodLoggedAt,
+      },
+    });
+  });
+});

src/lib/moodlog/__tests__/sync.test.ts

@@ -0,0 +1,98 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+vi.mock("@/lib/db", () => ({
+  prisma: {
+    user: {
+      findUnique: vi.fn(),
+      update: vi.fn(),
+    },
+    moodEntry: {
+      upsert: vi.fn(),
+    },
+  },
+}));
+
+vi.mock("@/lib/crypto", () => ({
+  decrypt: vi.fn((value: string) => value),
+}));
+
+vi.mock("@/lib/validations/notifications", () => ({
+  isPublicUrl: vi.fn(() => true),
+}));
+
+vi.mock("@/lib/safe-fetch", () => ({
+  safeFetch: vi.fn(),
+}));
+
+vi.mock("@/lib/integrations/status", () => ({
+  isReauthRequired: vi.fn(),
+  recordSyncFailure: vi.fn(),
+  recordSyncSuccess: vi.fn(),
+}));
+
+vi.mock("@/lib/rollups/mood-rollups", () => ({
+  recomputeUserMoodRollups: vi.fn(),
+}));
+
+vi.mock("@/lib/logging/context", () => ({
+  getEvent: vi.fn(() => ({
+    addExternalCall: vi.fn(),
+    addWarning: vi.fn(),
+  })),
+}));
+
+import { prisma } from "@/lib/db";
+import {
+  isReauthRequired,
+  recordSyncFailure,
+  recordSyncSuccess,
+} from "@/lib/integrations/status";
+import { safeFetch } from "@/lib/safe-fetch";
+import { syncMoodLogEntries } from "../sync";
+
+const SOURCE_ENTRY = {
+  id: "provider-event-1",
+  date: "2026-07-19",
+  time: "2026-07-19T08:30:00.000Z",
+  mood: "GUT",
+  score: 4,
+  tags: ["walk"],
+};
+
+describe("syncMoodLogEntries persistence completion", () => {
+  beforeEach(() => {
+    vi.resetAllMocks();
+    vi.mocked(prisma.user.findUnique).mockResolvedValue({
+      moodLogEnabled: true,
+      moodLogUrlEncrypted: "https://moodlog.example.test",
+      moodLogApiKeyEncrypted: "api-key",
+      moodLogLastSyncedAt: new Date("2026-07-18T08:30:00.000Z"),
+    } as never);
+    vi.mocked(isReauthRequired).mockResolvedValue(false);
+    vi.mocked(safeFetch).mockResolvedValue(
+      new Response(JSON.stringify({ version: "1", entries: [SOURCE_ENTRY] }), {
+        status: 200,
+      }),
+    );
+    vi.mocked(prisma.user.update).mockResolvedValue({} as never);
+  });
+
+  it("holds the watermark and success marker when a source upsert fails", async () => {
+    const persistenceError = new Error("database unavailable");
+    vi.mocked(prisma.moodEntry.upsert).mockRejectedValueOnce(persistenceError);
+
+    await expect(syncMoodLogEntries("user-1")).rejects.toThrow(
+      "database unavailable",
+    );
+
+    expect(prisma.user.update).not.toHaveBeenCalled();
+    expect(recordSyncSuccess).not.toHaveBeenCalled();
+    expect(recordSyncFailure).toHaveBeenCalledWith({
+      userId: "user-1",
+      integration: "moodlog",
+      kind: "transient",
+      message: "database unavailable",
+      errorCode: "persistence_failed",
+    });
+  });
+});

src/lib/moodlog/persistence.ts

@@ -0,0 +1,132 @@
+import { prisma } from "@/lib/db";
+import { isP2002 } from "@/lib/prisma-errors";
+
+export interface MoodLogSourceEntry {
+  externalId: string | null;
+  date: string;
+  moodLoggedAt: Date;
+  mood: string;
+  score: number;
+  tags?: readonly string[];
+}
+
+/**
+ * Persist one moodLog-owned source row using the provider id when available
+ * and the legacy wall-clock key otherwise.
+ *
+ * A provider may add an id after an older delivery already created the row by
+ * wall-clock key. In that case the external-id upsert collides with the legacy
+ * unique constraint; adopt that durable row and attach the provider id rather
+ * than creating a duplicate or retrying forever.
+ */
+export async function persistMoodLogSourceEntry(
+  userId: string,
+  entry: MoodLogSourceEntry,
+): Promise<void> {
+  const tags = entry.tags ? JSON.stringify(entry.tags) : null;
+  const update = {
+    mood: entry.mood,
+    score: entry.score,
+    tags,
+    source: "MOODLOG",
+    deletedAt: null,
+  } as const;
+
+  if (!entry.externalId) {
+    await prisma.moodEntry.upsert({
+      where: {
+        userId_date_moodLoggedAt: {
+          userId,
+          date: entry.date,
+          moodLoggedAt: entry.moodLoggedAt,
+        },
+      },
+      update,
+      create: {
+        userId,
+        date: entry.date,
+        moodLoggedAt: entry.moodLoggedAt,
+        ...update,
+      },
+    });
+    return;
+  }
+
+  let collisionError: unknown;
+  try {
+    await prisma.moodEntry.upsert({
+      where: {
+        userId_source_externalId: {
+          userId,
+          source: "MOODLOG",
+          externalId: entry.externalId,
+        },
+      },
+      update: {
+        ...update,
+        date: entry.date,
+        moodLoggedAt: entry.moodLoggedAt,
+      },
+      create: {
+        userId,
+        date: entry.date,
+        moodLoggedAt: entry.moodLoggedAt,
+        externalId: entry.externalId,
+        ...update,
+      },
+    });
+    return;
+  } catch (err) {
+    if (!isP2002(err)) throw err;
+    collisionError = err;
+  }
+
+  const adopted = await prisma.moodEntry.updateMany({
+    where: {
+      userId,
+      date: entry.date,
+      moodLoggedAt: entry.moodLoggedAt,
+      source: "MOODLOG",
+      externalId: null,
+    },
+    data: {
+      ...update,
+      externalId: entry.externalId,
+    },
+  });
+  if (adopted.count === 1) return;
+
+  // A concurrent replay may have adopted the same legacy row first.
+  const winner = await prisma.moodEntry.findUnique({
+    where: {
+      userId_source_externalId: {
+        userId,
+        source: "MOODLOG",
+        externalId: entry.externalId,
+      },
+    },
+    select: { date: true, moodLoggedAt: true },
+  });
+  if (
+    winner?.date === entry.date &&
+    winner.moodLoggedAt.getTime() === entry.moodLoggedAt.getTime()
+  ) {
+    await prisma.moodEntry.update({
+      where: {
+        userId_source_externalId: {
+          userId,
+          source: "MOODLOG",
+          externalId: entry.externalId,
+        },
+      },
+      data: {
+        ...update,
+        date: entry.date,
+        moodLoggedAt: entry.moodLoggedAt,
+      },
+    });
+    return;
+  }
+
+  throw collisionError;
+}

src/lib/moodlog/sync.ts

@@ -8,6 +8,7 @@ import {
   recordSyncFailure,
   recordSyncSuccess,
 } from "@/lib/integrations/status";
+import { persistMoodLogSourceEntry } from "@/lib/moodlog/persistence";
 import { recomputeUserMoodRollups } from "@/lib/rollups/mood-rollups";
 
 /**
@@ -235,63 +236,26 @@ export async function syncMoodLogEntries(
           ? entry.id.slice(0, 120)
           : null;
 
-      if (externalId) {
-        await prisma.moodEntry.upsert({
-          where: {
-            userId_source_externalId: {
-              userId,
-              source: "MOODLOG",
-              externalId,
-            },
-          },
-          update: {
-            mood: entry.mood,
-            score: entry.score,
-            tags: entry.tags ? JSON.stringify(entry.tags) : null,
-            source: "MOODLOG",
-            date,
-            moodLoggedAt,
-          },
-          create: {
-            userId,
-            date,
-            mood: entry.mood,
-            score: entry.score,
-            tags: entry.tags ? JSON.stringify(entry.tags) : null,
-            source: "MOODLOG",
-            externalId,
-            moodLoggedAt,
-          },
-        });
-      } else {
-        await prisma.moodEntry.upsert({
-          where: {
-            userId_date_moodLoggedAt: {
-              userId,
-              date,
-              moodLoggedAt,
-            },
-          },
-          update: {
-            mood: entry.mood,
-            score: entry.score,
-            tags: entry.tags ? JSON.stringify(entry.tags) : null,
-            source: "MOODLOG",
-          },
-          create: {
-            userId,
-            date,
-            mood: entry.mood,
-            score: entry.score,
-            tags: entry.tags ? JSON.stringify(entry.tags) : null,
-            source: "MOODLOG",
-            moodLoggedAt,
-          },
-        });
-      }
+      await persistMoodLogSourceEntry(userId, {
+        externalId,
+        date,
+        moodLoggedAt,
+        mood: entry.mood,
+        score: entry.score,
+        tags: entry.tags,
+      });
       imported++;
     } catch (err) {
-      getEvent()?.addWarning(`Failed to upsert entry: ${err}`);
+      const message = err instanceof Error ? err.message : String(err);
+      getEvent()?.addWarning(`Failed to upsert moodLog entry: ${message}`);
+      await recordSyncFailure({
+        userId,
+        integration: "moodlog",
+        kind: "transient",
+        message,
+        errorCode: "persistence_failed",
+      });
+      throw err;
     }
   }
 

tests/integration/partial-unique-live-rows.test.ts

@@ -199,6 +199,157 @@ describe("MoodEntry — external-id re-import idempotency (v1.12.1 / 0122)", ()
     expect(rows[0].score).toBe(5);
   });
 
+  it("acknowledges an exact provider replay with one stable durable row", async () => {
+    const prisma = getPrismaClient();
+    const secret = "mood-secret-replay";
+    const user = await prisma.user.create({
+      data: {
+        username: "mood-provider-replay",
+        email: "mood-provider-replay@example.test",
+        role: "USER",
+        moodLogEnabled: true,
+        moodLogWebhookSecret: secret,
+      },
+    });
+    const { POST } =
+      await import("@/app/api/integrations/moodlog/webhook/route");
+    const payload = {
+      event: "mood.created",
+      timestamp: "2026-06-04T11:00:00.000Z",
+      entry: {
+        id: "provider-event-replay-1",
+        date: "2026-06-04",
+        time: "2026-06-04T11:00:00.000Z",
+        mood: "GUT",
+        score: 4,
+      },
+    };
+
+    const first = await POST(webhookRequest(secret, payload));
+    expect(first.status).toBe(200);
+    const firstRow = await prisma.moodEntry.findUniqueOrThrow({
+      where: {
+        userId_source_externalId: {
+          userId: user.id,
+          source: "MOODLOG",
+          externalId: "provider-event-replay-1",
+        },
+      },
+    });
+
+    const replay = await POST(webhookRequest(secret, payload));
+    expect(replay.status).toBe(200);
+    const rows = await prisma.moodEntry.findMany({
+      where: { userId: user.id, source: "MOODLOG" },
+    });
+    expect(rows).toHaveLength(1);
+    expect(rows[0].id).toBe(firstRow.id);
+  });
+
+  it("adopts a legacy natural-key row when a replay gains a provider id", async () => {
+    const prisma = getPrismaClient();
+    const secret = "mood-secret-key-upgrade";
+    const user = await prisma.user.create({
+      data: {
+        username: "mood-key-upgrade",
+        email: "mood-key-upgrade@example.test",
+        role: "USER",
+        moodLogEnabled: true,
+        moodLogWebhookSecret: secret,
+      },
+    });
+    const { POST } =
+      await import("@/app/api/integrations/moodlog/webhook/route");
+    const entry = {
+      date: "2026-06-04",
+      time: "2026-06-04T12:00:00.000Z",
+      mood: "GUT",
+      score: 4,
+    };
+
+    const legacy = await POST(
+      webhookRequest(secret, {
+        event: "mood.created",
+        timestamp: "2026-06-04T12:00:00.000Z",
+        entry,
+      }),
+    );
+    expect(legacy.status).toBe(200);
+    const legacyRow = await prisma.moodEntry.findFirstOrThrow({
+      where: { userId: user.id, source: "MOODLOG" },
+    });
+    expect(legacyRow.externalId).toBeNull();
+
+    const upgraded = await POST(
+      webhookRequest(secret, {
+        event: "mood.updated",
+        timestamp: "2026-06-04T12:05:00.000Z",
+        entry: { ...entry, id: "provider-key-upgrade-1" },
+      }),
+    );
+    expect(upgraded.status).toBe(200);
+
+    const rows = await prisma.moodEntry.findMany({
+      where: { userId: user.id, source: "MOODLOG" },
+    });
+    expect(rows).toHaveLength(1);
+    expect(rows[0].id).toBe(legacyRow.id);
+    expect(rows[0].externalId).toBe("provider-key-upgrade-1");
+  });
+
+  it("does not adopt an unrelated row that occupies the natural key", async () => {
+    const prisma = getPrismaClient();
+    const secret = "mood-secret-unrelated-collision";
+    const user = await prisma.user.create({
+      data: {
+        username: "mood-unrelated-collision",
+        email: "mood-unrelated-collision@example.test",
+        role: "USER",
+        moodLogEnabled: true,
+        moodLogWebhookSecret: secret,
+      },
+    });
+    const moodLoggedAt = new Date("2026-06-04T13:00:00.000Z");
+    const manual = await prisma.moodEntry.create({
+      data: {
+        userId: user.id,
+        date: "2026-06-04",
+        moodLoggedAt,
+        mood: "OKAY",
+        score: 3,
+        source: "MANUAL",
+      },
+    });
+    const { POST } =
+      await import("@/app/api/integrations/moodlog/webhook/route");
+
+    const response = await POST(
+      webhookRequest(secret, {
+        event: "mood.created",
+        timestamp: "2026-06-04T13:00:00.000Z",
+        entry: {
+          id: "provider-unrelated-collision-1",
+          date: "2026-06-04",
+          time: moodLoggedAt.toISOString(),
+          mood: "GUT",
+          score: 4,
+        },
+      }),
+    );
+    expect(response.status).toBe(503);
+
+    const row = await prisma.moodEntry.findUniqueOrThrow({
+      where: { id: manual.id },
+    });
+    expect(row).toMatchObject({
+      source: "MANUAL",
+      externalId: null,
+      mood: "OKAY",
+      score: 3,
+    });
+    expect(await prisma.moodEntry.count({ where: { userId: user.id } })).toBe(1);
+  });
+
   it("entries without an id still dedup on the legacy wall-clock key", async () => {
     const prisma = getPrismaClient();
     const secret = "mood-secret-legacy";