fix(contacts): make concurrent first-contact upserts safe

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<void> {
+    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,

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<void>(resolve => {
+        releaseInitialLookups = resolve;
+      });
+      let releaseLosingCreates!: () => void;
+      const electedContactCreated = new Promise<void>(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<void>(resolve => {
+        releaseInitialLookups = resolve;
+      });
+      let releaseLosingCreates!: () => void;
+      const electedContactCreated = new Promise<void>(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<void>(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)', () => {