diff --git a/src/app/api/integrations/moodlog/webhook/__tests__/route.test.ts b/src/app/api/integrations/moodlog/webhook/__tests__/route.test.ts index a54435f6d..1cf3a6519 100644 --- a/src/app/api/integrations/moodlog/webhook/__tests__/route.test.ts +++ b/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, diff --git a/src/app/api/integrations/moodlog/webhook/route.ts b/src/app/api/integrations/moodlog/webhook/route.ts index 64e699731..7d5445811 100644 --- a/src/app/api/integrations/moodlog/webhook/route.ts +++ b/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 }); }); diff --git a/src/lib/moodlog/__tests__/persistence.test.ts b/src/lib/moodlog/__tests__/persistence.test.ts new file mode 100644 index 000000000..f470ab1ff --- /dev/null +++ b/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, + }, + }); + }); +}); diff --git a/src/lib/moodlog/__tests__/sync.test.ts b/src/lib/moodlog/__tests__/sync.test.ts new file mode 100644 index 000000000..ad59e1393 --- /dev/null +++ b/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", + }); + }); +}); diff --git a/src/lib/moodlog/persistence.ts b/src/lib/moodlog/persistence.ts new file mode 100644 index 000000000..b9ac8f6b2 --- /dev/null +++ b/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 { + 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; +} diff --git a/src/lib/moodlog/sync.ts b/src/lib/moodlog/sync.ts index 6d2d85a57..6de5ab246 100644 --- a/src/lib/moodlog/sync.ts +++ b/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; } } diff --git a/tests/integration/partial-unique-live-rows.test.ts b/tests/integration/partial-unique-live-rows.test.ts index b8f4524d5..145c95052 100644 --- a/tests/integration/partial-unique-live-rows.test.ts +++ b/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";