diff --git a/apps/api/src/services/ContactService.ts b/apps/api/src/services/ContactService.ts index fbc81ca7..a8abc953 100644 --- a/apps/api/src/services/ContactService.ts +++ b/apps/api/src/services/ContactService.ts @@ -284,6 +284,28 @@ export class ContactService { }); } + /** + * Apply an explicit subscription transition once, even when callers observed + * the same stale contact state. The compare-and-set result owns event emission. + */ + private static async applySubscriptionChange( + projectId: string, + contactId: string, + subscribed: boolean, + ): Promise { + const changed = await prisma.contact.updateMany({ + where: {id: contactId, projectId, subscribed: !subscribed}, + data: {subscribed}, + }); + if (changed.count === 1) { + await EventService.trackEvent( + projectId, + subscribed ? 'contact.subscribed' : 'contact.unsubscribed', + contactId, + ); + } + } + /** * Upsert a contact (create or update) with metadata merging * Supports persistent and non-persistent data fields @@ -309,29 +331,18 @@ export class ContactService { const mergedData = ContactService.mergeContactData(existing?.data ?? null, data ?? {}); if (existing) { - // Track subscription status change - const isSubscriptionChanging = subscribed !== undefined && existing.subscribed !== subscribed; - const wasSubscribed = existing.subscribed; - try { const updated = await prisma.contact.update({ where: {id: existing.id}, data: { data: Object.keys(mergedData).length > 0 ? toPrismaJson(mergedData) : Prisma.JsonNull, - ...(subscribed !== undefined ? {subscribed} : {}), }, }); - // Track subscription event if status changed - if (isSubscriptionChanging) { - if (subscribed && !wasSubscribed) { - await EventService.trackEvent(projectId, 'contact.subscribed', updated.id); - } else if (!subscribed && wasSubscribed) { - await EventService.trackEvent(projectId, 'contact.unsubscribed', updated.id); - } - } + if (subscribed === undefined) return updated; - return updated; + await ContactService.applySubscriptionChange(projectId, updated.id, subscribed); + return prisma.contact.findUniqueOrThrow({where: {id: updated.id}}); } catch (error) { // Provide helpful error message for database/validation issues throw new HttpException( @@ -350,6 +361,30 @@ export class ContactService { }, }); } catch (error) { + if (error instanceof Error && 'code' in error && error.code === 'P2002') { + const elected = await prisma.contact.findUnique({ + where: { + projectId_email: { + projectId, + email: normalizedEmail, + }, + }, + }); + + if (elected) { + // This payload was derived from an empty contact before another + // request won the insert. An explicit opt-out is monotonic and may + // safely win the race, but never replay stale contact data or a + // true/default subscription value over the elected row. + if (subscribed === false) { + await ContactService.applySubscriptionChange(projectId, elected.id, false); + return prisma.contact.findUniqueOrThrow({where: {id: elected.id}}); + } + + return elected; + } + } + // Provide helpful error message for database/validation issues throw new HttpException( 500, diff --git a/apps/api/src/services/__tests__/ContactService.test.ts b/apps/api/src/services/__tests__/ContactService.test.ts index 67c39347..2cbbbfda 100644 --- a/apps/api/src/services/__tests__/ContactService.test.ts +++ b/apps/api/src/services/__tests__/ContactService.test.ts @@ -1,6 +1,8 @@ -import {beforeEach, describe, expect, it} from 'vitest'; -import {ContactService} from '../ContactService'; +import {beforeEach, describe, expect, it, vi} from 'vitest'; + import {factories, getPrismaClient} from '../../../../../test/helpers'; +import {prisma as servicePrisma} from '../../database/prisma'; +import {ContactService} from '../ContactService'; describe('ContactService - Duplicate Prevention & Data Merging', () => { let projectId: string; @@ -73,6 +75,173 @@ describe('ContactService - Duplicate Prevention & Data Merging', () => { }); expect(contacts).toHaveLength(1); }); + + it('should return the elected contact to concurrent first-seen upserts without overwriting it', async () => { + const email = 'first-seen-race@example.com'; + const concurrency = 8; + const originalFindFirst = servicePrisma.contact.findFirst.bind(servicePrisma.contact); + const originalCreate = servicePrisma.contact.create.bind(servicePrisma.contact); + let initialLookups = 0; + let releaseInitialLookups!: () => void; + const allInitialLookupsStarted = new Promise(resolve => { + releaseInitialLookups = resolve; + }); + let releaseLosingCreates!: () => void; + const electedContactCreated = new Promise(resolve => { + releaseLosingCreates = resolve; + }); + + const contactLookup = vi.spyOn(servicePrisma.contact, 'findFirst').mockImplementation(async args => { + if (args.where?.projectId === projectId && args.where?.email === email) { + initialLookups += 1; + if (initialLookups === concurrency) releaseInitialLookups(); + await allInitialLookupsStarted; + return null; + } + + return originalFindFirst(args); + }); + const contactCreate = vi.spyOn(servicePrisma.contact, 'create').mockImplementation(async args => { + if (args.data.projectId === projectId && args.data.email === email && args.data.subscribed !== false) { + await electedContactCreated; + } + + const contact = await originalCreate(args); + if (args.data.projectId === projectId && args.data.email === email && args.data.subscribed === false) { + releaseLosingCreates(); + } + return contact; + }); + + try { + const contacts = await Promise.all([ + ContactService.upsert(projectId, email, {source: 'unsubscribe'}, false), + ...Array.from({length: concurrency - 1}, (_, index) => + ContactService.upsert(projectId, ` FIRST-SEEN-RACE@EXAMPLE.COM `, {attempt: index}), + ), + ]); + + expect(new Set(contacts.map(contact => contact.id))).toEqual(new Set([contacts[0]!.id])); + } finally { + contactLookup.mockRestore(); + contactCreate.mockRestore(); + } + + const stored = await prisma.contact.findUniqueOrThrow({ + where: {projectId_email: {projectId, email}}, + }); + expect(stored.subscribed).toBe(false); + expect(stored.data).toEqual({source: 'unsubscribe'}); + expect(await prisma.contact.count({where: {projectId, email}})).toBe(1); + }); + + it('should honor a losing explicit unsubscribe without overwriting the elected contact data', async () => { + const email = 'losing-unsubscribe@example.com'; + const concurrency = 8; + const originalFindFirst = servicePrisma.contact.findFirst.bind(servicePrisma.contact); + const originalCreate = servicePrisma.contact.create.bind(servicePrisma.contact); + let initialLookups = 0; + let releaseInitialLookups!: () => void; + const allInitialLookupsStarted = new Promise(resolve => { + releaseInitialLookups = resolve; + }); + let releaseLosingCreates!: () => void; + const electedContactCreated = new Promise(resolve => { + releaseLosingCreates = resolve; + }); + + const contactLookup = vi.spyOn(servicePrisma.contact, 'findFirst').mockImplementation(async args => { + if (args.where?.projectId === projectId && args.where?.email === email) { + initialLookups += 1; + if (initialLookups === concurrency) releaseInitialLookups(); + await allInitialLookupsStarted; + return null; + } + + return originalFindFirst(args); + }); + const contactCreate = vi.spyOn(servicePrisma.contact, 'create').mockImplementation(async args => { + const contactData = args.data.data; + const shouldWin = + contactData !== null && + typeof contactData === 'object' && + !Array.isArray(contactData) && + 'source' in contactData && + contactData.source === 'winner'; + + if (args.data.projectId === projectId && args.data.email === email && !shouldWin) { + await electedContactCreated; + } + + const contact = await originalCreate(args); + if (args.data.projectId === projectId && args.data.email === email && shouldWin) { + releaseLosingCreates(); + } + return contact; + }); + + try { + const contacts = await Promise.all([ + ContactService.upsert(projectId, email, {source: 'winner'}, true), + ContactService.upsert(projectId, email, {source: 'losing unsubscribe'}, false), + ...Array.from({length: concurrency - 2}, (_, index) => + ContactService.upsert(projectId, ` LOSING-UNSUBSCRIBE@EXAMPLE.COM `, {attempt: index}), + ), + ]); + + expect(new Set(contacts.map(contact => contact.id))).toEqual(new Set([contacts[0]!.id])); + } finally { + contactLookup.mockRestore(); + contactCreate.mockRestore(); + } + + const stored = await prisma.contact.findUniqueOrThrow({ + where: {projectId_email: {projectId, email}}, + }); + expect(stored.subscribed).toBe(false); + expect(stored.data).toEqual({source: 'winner'}); + expect( + await prisma.event.count({ + where: {projectId, contactId: stored.id, name: 'contact.unsubscribed'}, + }), + ).toBe(1); + }); + + it('should emit one unsubscribe event when concurrent upserts observe the same subscribed contact', async () => { + const email = 'concurrent-unsubscribe@example.com'; + const contact = await ContactService.upsert(projectId, email, {source: 'initial'}, true); + const originalFindFirst = servicePrisma.contact.findFirst.bind(servicePrisma.contact); + let lookups = 0; + let releaseLookups!: () => void; + const bothLookupsCompleted = new Promise(resolve => { + releaseLookups = resolve; + }); + const contactLookup = vi.spyOn(servicePrisma.contact, 'findFirst').mockImplementation(async args => { + const found = await originalFindFirst(args); + if (args.where?.projectId === projectId && args.where?.email === email) { + lookups += 1; + if (lookups === 2) releaseLookups(); + await bothLookupsCompleted; + } + return found; + }); + + try { + await Promise.all([ + ContactService.upsert(projectId, email, {first: true}, false), + ContactService.upsert(projectId, email, {second: true}, false), + ]); + } finally { + contactLookup.mockRestore(); + } + + expect((await prisma.contact.findUniqueOrThrow({where: {id: contact.id}})).subscribed).toBe(false); + expect( + await prisma.event.count({ + where: {projectId, contactId: contact.id, name: 'contact.unsubscribed'}, + }), + ).toBe(1); + }); }); describe('Email Normalization (case-insensitive find-or-create)', () => {