Merge pull request #393 from mooouuu/feat/tag-added-trigger

feat(automations): dispatch tag added triggers

src/app/api/contacts/[id]/tags/route.test.ts

@@ -0,0 +1,95 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+const mocks = vi.hoisted(() => ({
+  requireRole: vi.fn(),
+  add: vi.fn(),
+  remove: vi.fn(),
+}));
+
+vi.mock('@/lib/auth/account', () => ({
+  requireRole: mocks.requireRole,
+  toErrorResponse: vi.fn(() =>
+    Response.json({ error: 'auth failed' }, { status: 403 })
+  ),
+}));
+
+vi.mock('@/lib/contacts/tag-events', () => ({
+  addContactTagAndDispatch: mocks.add,
+}));
+
+vi.mock('@/lib/contacts/tag-write', () => ({
+  ContactTagWriteError: class ContactTagWriteError extends Error {
+    status: number;
+    constructor(message: string, status = 500) {
+      super(message);
+      this.status = status;
+    }
+  },
+  removeContactTag: mocks.remove,
+}));
+
+import { DELETE, POST } from './route';
+
+const context = {
+  supabase: { name: 'scoped-client' },
+  accountId: 'account-1',
+  userId: 'user-1',
+  role: 'agent',
+  account: { id: 'account-1', name: 'Acme' },
+};
+
+function request(method: 'POST' | 'DELETE', body: unknown) {
+  return new Request('http://localhost/api/contacts/contact-1/tags', {
+    method,
+    headers: { 'Content-Type': 'application/json' },
+    body: JSON.stringify(body),
+  });
+}
+
+const params = { params: Promise.resolve({ id: 'contact-1' }) };
+
+beforeEach(() => {
+  mocks.requireRole.mockReset();
+  mocks.add.mockReset();
+  mocks.remove.mockReset();
+  mocks.requireRole.mockResolvedValue(context);
+});
+
+describe('/api/contacts/[id]/tags', () => {
+  it('requires an agent and dispatches a newly-added tag', async () => {
+    mocks.add.mockResolvedValue({ added: true, dispatched: true });
+
+    const response = await POST(request('POST', { tag_id: 'tag-1' }), params);
+
+    expect(response.status).toBe(200);
+    expect(mocks.requireRole).toHaveBeenCalledWith('agent');
+    expect(mocks.add).toHaveBeenCalledWith({
+      db: context.supabase,
+      accountId: 'account-1',
+      contactId: 'contact-1',
+      tagId: 'tag-1',
+    });
+  });
+
+  it('rejects a missing tag id before writing', async () => {
+    const response = await POST(request('POST', {}), params);
+    expect(response.status).toBe(400);
+    expect(mocks.add).not.toHaveBeenCalled();
+  });
+
+  it('removes a tag through the same account-scoped route', async () => {
+    mocks.remove.mockResolvedValue(undefined);
+
+    const response = await DELETE(
+      request('DELETE', { tag_id: 'tag-1' }),
+      params
+    );
+
+    expect(response.status).toBe(200);
+    expect(mocks.remove).toHaveBeenCalledWith(context.supabase, {
+      accountId: 'account-1',
+      contactId: 'contact-1',
+      tagId: 'tag-1',
+    });
+  });
+});

src/app/api/contacts/[id]/tags/route.ts

@@ -0,0 +1,76 @@
+import { NextResponse } from 'next/server';
+
+import { requireRole, toErrorResponse } from '@/lib/auth/account';
+import { addContactTagAndDispatch } from '@/lib/contacts/tag-events';
+import {
+  ContactTagWriteError,
+  removeContactTag,
+} from '@/lib/contacts/tag-write';
+
+function tagWriteErrorResponse(error: ContactTagWriteError): NextResponse {
+  return NextResponse.json({ error: error.message }, { status: error.status });
+}
+
+async function readTagId(request: Request): Promise<string | null> {
+  const body = (await request.json().catch(() => null)) as {
+    tag_id?: unknown;
+  } | null;
+  return typeof body?.tag_id === 'string' && body.tag_id.trim()
+    ? body.tag_id.trim()
+    : null;
+}
+
+export async function POST(
+  request: Request,
+  { params }: { params: Promise<{ id: string }> }
+) {
+  try {
+    const ctx = await requireRole('agent');
+    const { id: contactId } = await params;
+    const tagId = await readTagId(request);
+    if (!tagId) {
+      return NextResponse.json({ error: 'tag_id required' }, { status: 400 });
+    }
+
+    const result = await addContactTagAndDispatch({
+      db: ctx.supabase,
+      accountId: ctx.accountId,
+      contactId,
+      tagId,
+    });
+
+    return NextResponse.json({ ok: true, ...result });
+  } catch (error) {
+    if (error instanceof ContactTagWriteError) {
+      return tagWriteErrorResponse(error);
+    }
+    return toErrorResponse(error);
+  }
+}
+
+export async function DELETE(
+  request: Request,
+  { params }: { params: Promise<{ id: string }> }
+) {
+  try {
+    const ctx = await requireRole('agent');
+    const { id: contactId } = await params;
+    const tagId = await readTagId(request);
+    if (!tagId) {
+      return NextResponse.json({ error: 'tag_id required' }, { status: 400 });
+    }
+
+    await removeContactTag(ctx.supabase, {
+      accountId: ctx.accountId,
+      contactId,
+      tagId,
+    });
+
+    return NextResponse.json({ ok: true });
+  } catch (error) {
+    if (error instanceof ContactTagWriteError) {
+      return tagWriteErrorResponse(error);
+    }
+    return toErrorResponse(error);
+  }
+}

src/components/contacts/contact-detail-view.tsx

@@ -2,6 +2,7 @@
 
 import { useState, useEffect, useCallback } from 'react';
 import { createClient } from '@/lib/supabase/client';
+import { addContactTag, deleteContactTag } from '@/lib/contacts/tag-api';
 import { useAuth } from '@/hooks/use-auth';
 import { formatCurrency } from '@/lib/currency';
 import { toast } from 'sonner';
@@ -230,24 +231,17 @@ export function ContactDetailView({
 
     const isSelected = contactTagIds.includes(tagId);
 
+    try {
       if (isSelected) {
-      const { error } = await supabase
+        await deleteContactTag(contactId, tagId);
-        .from('contact_tags')
-        .delete()
-        .eq('contact_id', contactId)
-        .eq('tag_id', tagId);
-      if (!error) {
         setContactTagIds((prev) => prev.filter((id) => id !== tagId));
-        onUpdated();
-      }
       } else {
-      const { error } = await supabase
+        await addContactTag(contactId, tagId);
-        .from('contact_tags')
-        .insert({ contact_id: contactId, tag_id: tagId });
-      if (!error) {
         setContactTagIds((prev) => [...prev, tagId]);
-        onUpdated();
       }
+      onUpdated();
+    } catch (error) {
+      toast.error(error instanceof Error ? error.message : t('toastUpdateFailed'));
     }
     setSavingTags(false);
   }

src/components/contacts/contact-form.tsx

@@ -3,6 +3,7 @@
 import { useState, useEffect } from 'react';
 import { createClient } from '@/lib/supabase/client';
 import { useAuth } from '@/hooks/use-auth';
+import { addContactTag, deleteContactTag } from '@/lib/contacts/tag-api';
 import { toast } from 'sonner';
 import type { Contact, Tag, ContactTag } from '@/types';
 import {
@@ -179,20 +180,16 @@ export function ContactForm({
 
       // Sync tags
       if (contactId) {
-        await supabase
+        const existingTagIds = new Set(contactTags.map((tag) => tag.tag_id));
-          .from('contact_tags')
+        const desiredTagIds = new Set(selectedTagIds);
-          .delete()
+        const toRemove = [...existingTagIds].filter((id) => !desiredTagIds.has(id));
-          .eq('contact_id', contactId);
+        const toAdd = [...desiredTagIds].filter((id) => !existingTagIds.has(id));
 
-        if (selectedTagIds.length > 0) {
+        for (const tagId of toRemove) {
-          const tagRows = selectedTagIds.map((tag_id) => ({
+          await deleteContactTag(contactId, tagId);
-            contact_id: contactId!,
+        }
-            tag_id,
+        for (const tagId of toAdd) {
-          }));
+          await addContactTag(contactId, tagId);
-          const { error: tagError } = await supabase
-            .from('contact_tags')
-            .insert(tagRows);
-          if (tagError) throw tagError;
         }
       }
 

src/lib/api/v1/contacts.ts

@@ -11,6 +11,7 @@ import type { SupabaseClient } from '@supabase/supabase-js';
 
 import { findExistingContact, isUniqueViolation } from '@/lib/contacts/dedupe';
 import { resolveImportTagIds } from '@/lib/contacts/resolve-import-tags';
+import { addContactTagAndDispatch } from '@/lib/contacts/tag-events';
 import { sanitizePhoneForMeta, isValidE164 } from '@/lib/whatsapp/phone-utils';
 
 /** Row select that embeds the contact's tags for serialization. */
@@ -199,10 +200,19 @@ export async function setContactTags(
     if (error) throw new ContactError('Failed to update contact tags', 500);
   }
   if (toAdd.length > 0) {
-    const { error } = await db
+    for (const tagId of toAdd) {
-      .from('contact_tags')
+      try {
-      .insert(toAdd.map((tag_id) => ({ contact_id: contactId, tag_id })));
+        await addContactTagAndDispatch({
-    if (error) throw new ContactError('Failed to update contact tags', 500);
+          db,
+          accountId,
+          contactId,
+          tagId,
+        });
+      } catch (error) {
+        console.error('[api/v1/contacts] tag add failed:', error);
+        throw new ContactError('Failed to update contact tags', 500);
+      }
+    }
   }
 }
 

src/lib/automations/engine.test.ts

@@ -11,6 +11,7 @@ const h = vi.hoisted(() => ({
     fromCalls: [] as string[],
     updateCalls: [] as { table: string; filters: [string, string, unknown][] }[],
     upsertCalls: [] as { table: string; payload: unknown }[],
+    logUpdates: [] as Record<string, unknown>[],
   },
 }));
 
@@ -46,7 +47,10 @@ vi.mock("./admin-client", () => {
     if (table === "automations") return { data: state.automations, error: null };
     if (table === "automation_logs") {
       if (type === "insert") return { data: { id: "log1" }, error: null };
-      if (type === "update") return { data: null, error: null };
+      if (type === "update") {
+        state.logUpdates.push(ops.payload as Record<string, unknown>);
+        return { data: null, error: null };
+      }
       return { data: { steps_executed: [], status: "success" }, error: null };
     }
     if (table === "automation_steps") return { data: state.steps, error: null };
@@ -109,6 +113,7 @@ beforeEach(() => {
   h.state.fromCalls = [];
   h.state.updateCalls = [];
   h.state.upsertCalls = [];
+  h.state.logUpdates = [];
 });
 
 describe("runAutomationsForTrigger — tenant isolation", () => {
@@ -336,3 +341,66 @@ describe("triggerMatches — interactive_reply", () => {
     expect(triggerMatches(automation([]), { interactive_reply_id: "yes" })).toBe(false);
   });
 });
+
+describe("triggerMatches — tag_added", () => {
+  function automation(tagId?: string): Automation {
+    return {
+      id: "a1",
+      account_id: ACCOUNT,
+      user_id: "u1",
+      name: "tag follow-up",
+      trigger_type: "tag_added",
+      trigger_config: tagId ? { tag_id: tagId } : {},
+      is_active: true,
+      execution_count: 0,
+      created_at: "",
+      updated_at: "",
+    };
+  }
+
+  it("matches only the exact tag id", () => {
+    expect(triggerMatches(automation("tag-a"), { tag_id: "tag-a" })).toBe(true);
+    expect(triggerMatches(automation("tag-a"), { tag_id: "tag-ab" })).toBe(false);
+  });
+
+  it("fails closed when the config or event tag is missing", () => {
+    expect(triggerMatches(automation(), { tag_id: "tag-a" })).toBe(false);
+    expect(triggerMatches(automation("tag-a"), {})).toBe(false);
+    expect(triggerMatches(automation("tag-a"), undefined)).toBe(false);
+  });
+});
+
+describe("tag_added — conversation policy", () => {
+  it("records a clear failed step when the contact has no conversation", async () => {
+    h.state.owned = { id: "c1" };
+    h.state.automations = [{
+      id: "a1",
+      account_id: ACCOUNT,
+      user_id: "u1",
+      name: "tag outreach",
+      trigger_type: "tag_added",
+      trigger_config: { tag_id: "tag-a" },
+      is_active: true,
+    }];
+    h.state.steps = [{
+      id: "s1",
+      automation_id: "a1",
+      step_type: "send_message",
+      position: 0,
+      parent_step_id: null,
+      step_config: { text: "Hello" },
+    }];
+
+    await runAutomationsForTrigger({
+      accountId: ACCOUNT,
+      triggerType: "tag_added",
+      contactId: "c1",
+      context: { tag_id: "tag-a" },
+    });
+
+    expect(h.state.logUpdates).toContainEqual(expect.objectContaining({
+      status: "failed",
+      error_message: "tag_added automation cannot send: contact has no existing conversation",
+    }));
+  });
+});

src/lib/automations/engine.ts

@@ -6,6 +6,7 @@ import type {
   ConditionStepConfig,
   KeywordMatchTriggerConfig,
   InteractiveReplyTriggerConfig,
+  TagTriggerConfig,
   SendMessageStepConfig,
   SendButtonsStepConfig,
   SendListStepConfig,
@@ -18,6 +19,8 @@ import type {
   AssignConversationStepConfig,
 } from '@/types'
 import { supabaseAdmin } from './admin-client'
+import { addContactTagIfAbsent } from '@/lib/contacts/tag-write'
+import { MAX_TAG_CHAIN_DEPTH, getTagChainDepth } from '@/lib/contacts/tag-chain'
 import { engineSendText, engineSendTemplate, engineSendInteractive } from './meta-send'
 import { validateInteractivePayload } from '@/lib/whatsapp/interactive'
 import { isDeliverableUrl } from '@/lib/webhooks/ssrf'
@@ -420,18 +423,40 @@ async function runStep(step: AutomationStep, args: ExecuteArgs): Promise<string>
     }
 
     case 'add_tag': {
-      // contact_tags has no account_id column; cross-tenant protection for
-      // the attacker-supplied contactId comes from the ownership guard in
-      // runAutomationsForTrigger.
       const cfg = step.step_config as TagStepConfig
       if (!args.contactId || !cfg.tag_id) throw new Error('add_tag needs contact + tag_id')
-      await db
+      const added = await addContactTagIfAbsent(db, {
-        .from('contact_tags')
+        accountId: args.automation.account_id,
-        .upsert(
+        contactId: args.contactId,
-          { contact_id: args.contactId, tag_id: cfg.tag_id },
+        tagId: cfg.tag_id,
-          { onConflict: 'contact_id,tag_id', ignoreDuplicates: true },
+      })
-        )
+      if (!added) return `tag ${cfg.tag_id} already present`
-      return `tag ${cfg.tag_id} added`
+
+      const depth = getTagChainDepth(args.context)
+      if (depth >= MAX_TAG_CHAIN_DEPTH) {
+        console.warn('[automations] tag_added chain depth limit reached', {
+          automationId: args.automation.id,
+          contactId: args.contactId,
+          tagId: cfg.tag_id,
+          depth,
+        })
+        return `tag ${cfg.tag_id} added; tag_added dispatch skipped at depth ${depth}`
+      }
+
+      await runAutomationsForTrigger({
+        accountId: args.automation.account_id,
+        triggerType: 'tag_added',
+        contactId: args.contactId,
+        context: {
+          ...args.context,
+          tag_id: cfg.tag_id,
+          vars: {
+            ...(args.context.vars ?? {}),
+            _tag_chain_depth: depth + 1,
+          },
+        },
+      })
+      return `tag ${cfg.tag_id} added and tag_added dispatched`
     }
 
     case 'remove_tag': {
@@ -613,7 +638,12 @@ async function resolveConversationId(args: ExecuteArgs): Promise<string> {
     .eq('contact_id', args.contactId)
     .maybeSingle()
   if (error) throw new Error(`conversation lookup failed: ${error.message}`)
-  if (!data?.id) throw new Error('no conversation for contact')
+  if (!data?.id) {
+    const prefix = args.triggerEvent === 'tag_added'
+      ? 'tag_added automation cannot send'
+      : 'cannot send'
+    throw new Error(`${prefix}: contact has no existing conversation`)
+  }
   return data.id as string
 }
 
@@ -642,6 +672,12 @@ export function triggerMatches(automation: Automation, ctx: AutomationContext |
     return cfg.reply_ids.includes(replyId)
   }
 
+  if (automation.trigger_type === 'tag_added') {
+    const cfg = automation.trigger_config as TagTriggerConfig
+    const tagId = ctx?.tag_id
+    return Boolean(tagId && cfg?.tag_id && cfg.tag_id === tagId)
+  }
+
   return true
 }
 

src/lib/contacts/tag-api.ts

@@ -0,0 +1,32 @@
+interface ContactTagMutationResult {
+  added?: boolean;
+  dispatched?: boolean;
+  reason?: 'duplicate' | 'max_depth';
+}
+
+async function mutateContactTag(
+  contactId: string,
+  tagId: string,
+  method: 'POST' | 'DELETE'
+): Promise<ContactTagMutationResult> {
+  const response = await fetch(`/api/contacts/${contactId}/tags`, {
+    method,
+    headers: { 'Content-Type': 'application/json' },
+    body: JSON.stringify({ tag_id: tagId }),
+  });
+  const body = (await response.json().catch(() => ({}))) as {
+    error?: string;
+  } & ContactTagMutationResult;
+  if (!response.ok) {
+    throw new Error(body.error ?? 'Failed to update contact tag');
+  }
+  return body;
+}
+
+export function addContactTag(contactId: string, tagId: string) {
+  return mutateContactTag(contactId, tagId, 'POST');
+}
+
+export function deleteContactTag(contactId: string, tagId: string) {
+  return mutateContactTag(contactId, tagId, 'DELETE');
+}

src/lib/contacts/tag-chain.ts

@@ -0,0 +1,10 @@
+export const MAX_TAG_CHAIN_DEPTH = 3;
+
+export function getTagChainDepth(context?: {
+  vars?: Record<string, unknown>;
+}): number {
+  const raw = context?.vars?._tag_chain_depth;
+  return typeof raw === 'number' && Number.isFinite(raw) && raw >= 0
+    ? Math.floor(raw)
+    : 0;
+}

src/lib/contacts/tag-events.test.ts

@@ -0,0 +1,108 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+const mocks = vi.hoisted(() => ({
+  add: vi.fn(),
+  dispatch: vi.fn(),
+}));
+
+vi.mock('./tag-write', () => ({
+  addContactTagIfAbsent: mocks.add,
+}));
+
+vi.mock('@/lib/automations/engine', () => ({
+  runAutomationsForTrigger: mocks.dispatch,
+}));
+
+import {
+  addContactTagAndDispatch,
+  getTagChainDepth,
+  MAX_TAG_CHAIN_DEPTH,
+} from './tag-events';
+
+const base = {
+  db: {} as never,
+  accountId: 'account-1',
+  contactId: 'contact-1',
+  tagId: 'tag-1',
+};
+
+beforeEach(() => {
+  mocks.add.mockReset();
+  mocks.dispatch.mockReset();
+  mocks.dispatch.mockResolvedValue(undefined);
+});
+
+describe('addContactTagAndDispatch', () => {
+  it('dispatches once for a newly inserted tag and propagates depth', async () => {
+    mocks.add.mockResolvedValue(true);
+
+    const result = await addContactTagAndDispatch({
+      ...base,
+      context: { vars: { source: 'flow', _tag_chain_depth: 1 } },
+    });
+
+    expect(result).toEqual({ added: true, dispatched: true });
+    expect(mocks.dispatch).toHaveBeenCalledWith({
+      accountId: 'account-1',
+      triggerType: 'tag_added',
+      contactId: 'contact-1',
+      context: {
+        tag_id: 'tag-1',
+        vars: { source: 'flow', _tag_chain_depth: 2 },
+      },
+    });
+  });
+
+  it('does not dispatch when the tag already exists', async () => {
+    mocks.add.mockResolvedValue(false);
+
+    await expect(addContactTagAndDispatch(base)).resolves.toEqual({
+      added: false,
+      dispatched: false,
+      reason: 'duplicate',
+    });
+    expect(mocks.dispatch).not.toHaveBeenCalled();
+  });
+
+  it('adds the tag but cuts a chain at the configured depth limit', async () => {
+    mocks.add.mockResolvedValue(true);
+
+    await expect(
+      addContactTagAndDispatch({
+        ...base,
+        context: { vars: { _tag_chain_depth: MAX_TAG_CHAIN_DEPTH } },
+      })
+    ).resolves.toEqual({
+      added: true,
+      dispatched: false,
+      reason: 'max_depth',
+    });
+    expect(mocks.dispatch).not.toHaveBeenCalled();
+  });
+
+  it('cuts an A-to-B-to-A tag chain before it can loop forever', async () => {
+    mocks.add.mockResolvedValue(true);
+    mocks.dispatch.mockImplementation(async (event) => {
+      const nextTag = event.context.tag_id === 'tag-a' ? 'tag-b' : 'tag-a';
+      await addContactTagAndDispatch({
+        ...base,
+        tagId: nextTag,
+        context: event.context,
+      });
+    });
+
+    await addContactTagAndDispatch({ ...base, tagId: 'tag-a' });
+
+    expect(mocks.dispatch).toHaveBeenCalledTimes(MAX_TAG_CHAIN_DEPTH);
+    expect(mocks.add).toHaveBeenCalledTimes(MAX_TAG_CHAIN_DEPTH + 1);
+  });
+});
+
+describe('getTagChainDepth', () => {
+  it('normalizes missing, invalid and fractional values', () => {
+    expect(getTagChainDepth()).toBe(0);
+    expect(getTagChainDepth({ vars: { _tag_chain_depth: '3' } })).toBe(0);
+    expect(getTagChainDepth({ vars: { _tag_chain_depth: -1 } })).toBe(0);
+    expect(getTagChainDepth({ vars: { _tag_chain_depth: 2.8 } })).toBe(2);
+  });
+});

src/lib/contacts/tag-events.ts

@@ -0,0 +1,67 @@
+import type { SupabaseClient } from '@supabase/supabase-js';
+
+import {
+  runAutomationsForTrigger,
+  type AutomationContext,
+} from '@/lib/automations/engine';
+import { addContactTagIfAbsent } from './tag-write';
+import { MAX_TAG_CHAIN_DEPTH, getTagChainDepth } from './tag-chain';
+
+export { MAX_TAG_CHAIN_DEPTH, getTagChainDepth } from './tag-chain';
+
+interface AddContactTagAndDispatchInput {
+  db: SupabaseClient;
+  accountId: string;
+  contactId: string;
+  tagId: string;
+  context?: AutomationContext;
+}
+
+export interface AddContactTagResult {
+  added: boolean;
+  dispatched: boolean;
+  reason?: 'duplicate' | 'max_depth';
+}
+
+/**
+ * Central server-side tag writer. It dispatches tag_added only for a
+ * newly-created join and caps chained tag automations to avoid loops.
+ */
+export async function addContactTagAndDispatch(
+  input: AddContactTagAndDispatchInput
+): Promise<AddContactTagResult> {
+  const added = await addContactTagIfAbsent(input.db, {
+    accountId: input.accountId,
+    contactId: input.contactId,
+    tagId: input.tagId,
+  });
+
+  if (!added) return { added: false, dispatched: false, reason: 'duplicate' };
+
+  const depth = getTagChainDepth(input.context);
+  if (depth >= MAX_TAG_CHAIN_DEPTH) {
+    console.warn('[automations] tag_added chain depth limit reached', {
+      accountId: input.accountId,
+      contactId: input.contactId,
+      tagId: input.tagId,
+      depth,
+    });
+    return { added: true, dispatched: false, reason: 'max_depth' };
+  }
+
+  await runAutomationsForTrigger({
+    accountId: input.accountId,
+    triggerType: 'tag_added',
+    contactId: input.contactId,
+    context: {
+      ...input.context,
+      tag_id: input.tagId,
+      vars: {
+        ...(input.context?.vars ?? {}),
+        _tag_chain_depth: depth + 1,
+      },
+    },
+  });
+
+  return { added: true, dispatched: true };
+}

src/lib/contacts/tag-write.test.ts

@@ -0,0 +1,97 @@
+import { describe, expect, it } from 'vitest';
+import type { SupabaseClient } from '@supabase/supabase-js';
+
+import { addContactTagIfAbsent } from './tag-write';
+
+interface FakeOptions {
+  contact?: { id: string } | null;
+  tag?: { id: string } | null;
+  insertData?: { id: string } | null;
+  insertError?: { code?: string; message: string } | null;
+}
+
+function fakeDb(options: FakeOptions = {}): SupabaseClient {
+  const contact =
+    options.contact === undefined ? { id: 'contact-1' } : options.contact;
+  const tag = options.tag === undefined ? { id: 'tag-1' } : options.tag;
+
+  return {
+    from(table: string) {
+      const state = { operation: 'select' };
+      const builder = {
+        select() {
+          return builder;
+        },
+        insert() {
+          state.operation = 'insert';
+          return builder;
+        },
+        eq() {
+          return builder;
+        },
+        maybeSingle() {
+          if (table === 'contacts')
+            return Promise.resolve({ data: contact, error: null });
+          if (table === 'tags')
+            return Promise.resolve({ data: tag, error: null });
+          if (table === 'contact_tags' && state.operation === 'insert') {
+            return Promise.resolve({
+              data:
+                options.insertData === undefined
+                  ? { id: 'join-1' }
+                  : options.insertData,
+              error: options.insertError ?? null,
+            });
+          }
+          return Promise.resolve({ data: null, error: null });
+        },
+      };
+      return builder;
+    },
+  } as unknown as SupabaseClient;
+}
+
+const input = {
+  accountId: 'account-1',
+  contactId: 'contact-1',
+  tagId: 'tag-1',
+};
+
+describe('addContactTagIfAbsent', () => {
+  it('returns true only when the join row was inserted', async () => {
+    await expect(addContactTagIfAbsent(fakeDb(), input)).resolves.toBe(true);
+  });
+
+  it('treats an error-free insert as successful even without a returned row', async () => {
+    await expect(
+      addContactTagIfAbsent(fakeDb({ insertData: null }), input)
+    ).resolves.toBe(true);
+  });
+
+  it('treats a unique violation as an idempotent duplicate', async () => {
+    const db = fakeDb({
+      insertData: null,
+      insertError: { code: '23505', message: 'duplicate key' },
+    });
+    await expect(addContactTagIfAbsent(db, input)).resolves.toBe(false);
+  });
+
+  it('refuses contacts and tags outside the account', async () => {
+    await expect(
+      addContactTagIfAbsent(fakeDb({ contact: null }), input)
+    ).rejects.toMatchObject({ status: 404 });
+    await expect(
+      addContactTagIfAbsent(fakeDb({ tag: null }), input)
+    ).rejects.toMatchObject({ status: 404 });
+  });
+
+  it('surfaces non-duplicate insert failures', async () => {
+    const db = fakeDb({
+      insertData: null,
+      insertError: { code: '42501', message: 'permission denied' },
+    });
+    await expect(addContactTagIfAbsent(db, input)).rejects.toThrow(
+      'Failed to add contact tag: permission denied'
+    );
+  });
+});

src/lib/contacts/tag-write.ts

@@ -0,0 +1,92 @@
+import type { SupabaseClient } from '@supabase/supabase-js';
+
+export class ContactTagWriteError extends Error {
+  readonly status: number;
+
+  constructor(message: string, status = 500) {
+    super(message);
+    this.name = 'ContactTagWriteError';
+    this.status = status;
+  }
+}
+
+interface ContactTagWriteInput {
+  accountId: string;
+  contactId: string;
+  tagId: string;
+}
+
+async function assertContactAndTagOwnership(
+  db: SupabaseClient,
+  input: ContactTagWriteInput
+): Promise<void> {
+  const [contactResult, tagResult] = await Promise.all([
+    db
+      .from('contacts')
+      .select('id')
+      .eq('id', input.contactId)
+      .eq('account_id', input.accountId)
+      .maybeSingle(),
+    db
+      .from('tags')
+      .select('id')
+      .eq('id', input.tagId)
+      .eq('account_id', input.accountId)
+      .maybeSingle(),
+  ]);
+
+  if (contactResult.error || tagResult.error) {
+    throw new ContactTagWriteError('Could not verify contact tag ownership');
+  }
+  if (!contactResult.data) {
+    throw new ContactTagWriteError('Contact not found', 404);
+  }
+  if (!tagResult.data) {
+    throw new ContactTagWriteError('Tag not found', 404);
+  }
+}
+
+/**
+ * Add a tag exactly once. The unique constraint on
+ * (contact_id, tag_id) is the concurrency-safe source of truth: a
+ * duplicate insert is a no-op and must not emit a tag_added event.
+ */
+export async function addContactTagIfAbsent(
+  db: SupabaseClient,
+  input: ContactTagWriteInput
+): Promise<boolean> {
+  await assertContactAndTagOwnership(db, input);
+
+  const { error } = await db
+    .from('contact_tags')
+    .insert({ contact_id: input.contactId, tag_id: input.tagId })
+    .select('id')
+    .maybeSingle();
+
+  if (error?.code === '23505') return false;
+  if (error) {
+    throw new ContactTagWriteError(
+      `Failed to add contact tag: ${error.message}`
+    );
+  }
+  return true;
+}
+
+export async function removeContactTag(
+  db: SupabaseClient,
+  input: ContactTagWriteInput
+): Promise<void> {
+  await assertContactAndTagOwnership(db, input);
+
+  const { error } = await db
+    .from('contact_tags')
+    .delete()
+    .eq('contact_id', input.contactId)
+    .eq('tag_id', input.tagId);
+
+  if (error) {
+    throw new ContactTagWriteError(
+      `Failed to remove contact tag: ${error.message}`
+    );
+  }
+}

src/lib/flows/engine.ts

@@ -40,6 +40,8 @@ import {
   engineSendText,
 } from "./meta-send";
 import { decideFallback, resolveFallbackPolicy } from "./fallback";
+import { addContactTagAndDispatch } from "@/lib/contacts/tag-events";
+import { removeContactTag } from "@/lib/contacts/tag-write";
 import {
   type CollectInputNodeConfig,
   type ConditionNodeConfig,
@@ -708,18 +710,22 @@ async function advanceFromNodeKey(
       const cfg = node.config as unknown as SetTagNodeConfig;
       try {
         if (cfg.mode === "add") {
-          await db
+          await addContactTagAndDispatch({
-            .from("contact_tags")
+            db,
-            .upsert(
+            accountId: run.account_id,
-              { contact_id: run.contact_id!, tag_id: cfg.tag_id },
+            contactId: run.contact_id!,
-              { onConflict: "contact_id,tag_id" },
+            tagId: cfg.tag_id,
-            );
+            context: {
+              conversation_id: run.conversation_id ?? undefined,
+              vars: run.vars,
+            },
+          });
         } else {
-          await db
+          await removeContactTag(db, {
-            .from("contact_tags")
+            accountId: run.account_id,
-            .delete()
+            contactId: run.contact_id!,
-            .eq("contact_id", run.contact_id!)
+            tagId: cfg.tag_id,
-            .eq("tag_id", cfg.tag_id);
+          });
         }
       } catch (err) {
         // Non-fatal — log + advance. A tag-write failure shouldn't