#Plan 05 — message-processor STOP cascade migration Implementation Plan

import { setDNC, closeAllOpenForContact } from '@retaintive/common/domain';

// tests/unit/persist-message-batch.test.ts(在现有 describe 块内 append)
import { setDNC, closeAllOpenForContact } from '@retaintive/common/domain';
vi.mock('@retaintive/common/domain', () => ({
  setDNC: vi.fn(() => ({ __fakeSQL: 'setDNC' })),
  closeAllOpenForContact: vi.fn(async () => ({
    status: 'allow' as const,
    statements: [{ __fakeSQL: 'close1' }, { __fakeSQL: 'close2' }, { __fakeSQL: 't1' }, { __fakeSQL: 't2' }],
    closedTaskIds: ['task-uuid-1', 'task-uuid-2'],
  })),
}));

it('STOP message closes all open tasks via closeAllOpenForContact', async () => {
  const repo = createNeonRepository('ssm-param');
  const mockClient = createMockDrizzleClient();
  // ... seed message entry with body='STOP', resolved customerPhone='+15551234567', storeId='STORE_A'
  // ... seed mockClient.batch() to capture statements

  await repo.persistMessageBatch([entry]);

  // Verify setDNC called once with correct args
  expect(setDNC).toHaveBeenCalledWith({
    phone: '+15551234567',
    storeId: 'STORE_A',
    updatedBy: 'system',
  });

  // Verify closeAllOpenForContact called once with correct args
  expect(closeAllOpenForContact).toHaveBeenCalledWith(
    {
      contactPhone: '+15551234567',
      storeId: 'STORE_A',
      closeResult: 'do_not_contact',
      closeNote: 'Auto-closed by DNC hard stop (SMS STOP keyword detected)',
      actor: { type: 'system', id: 'message-processor' },
    },
    { db: mockClient },
  );

  // Verify client.batch called with DNC stmt + 4 close-cascade statements (1 stmt setDNC + 4 from closeAllOpen)
  expect(mockClient.batch).toHaveBeenCalled();
  const lastBatchCall = mockClient.batch.mock.calls.at(-1)![0];
  expect(lastBatchCall).toHaveLength(5);
});

it('inbound SMS containing "STOP" as substring does NOT trigger DNC cascade', async () => {
  const repo = createNeonRepository('ssm-param');
  const mockClient = createMockDrizzleClient();
  const entry = makeEntry({ subject: 'just kidding STOP not really', direction: 'Inbound', type: 'SMS' });

  await repo.persistMessageBatch([entry]);

  // Cascade module NOT called
  expect(setDNC).not.toHaveBeenCalled();
  expect(closeAllOpenForContact).not.toHaveBeenCalled();

  // Main batch (message + contact + timeline) still ran
  expect(mockClient.batch).toHaveBeenCalledTimes(1);
});

import {
  createDrizzleClient,
  messages,
  contacts,
  contactTimeline,
  sql,
  buildTimelineValues,
  buildIdentityFields,
  lastActivityAtForward,
  NAME_TRUST,
} from '@retaintive/common/db';
import type { DrizzleClient, MessageParty, MessageAttachment } from '@retaintive/common/db';
import { resolveContactIdentity } from '@retaintive/common/phone-identity';

import { closeOpenTasksForDnc } from '../../../shared/utils/dnc-cascade';
import { createLogger } from '../../../shared/utils/logger';

import {
  createDrizzleClient,
  messages,
  contacts,
  contactTimeline,
  sql,
  buildTimelineValues,
  buildIdentityFields,
  lastActivityAtForward,
  NAME_TRUST,
} from '@retaintive/common/db';
import type { DrizzleClient, MessageParty, MessageAttachment } from '@retaintive/common/db';
import { resolveContactIdentity } from '@retaintive/common/phone-identity';
import { setDNC, closeAllOpenForContact } from '@retaintive/common/domain';

// REMOVED Plan 05: `closeOpenTasksForDnc` 不再直接 import — Phase 1 由 `closeAllOpenForContact`
// 内部调用 dnc-cascade.ts helper(Plan 02 内部 wire 好)。Phase 2 删 helper file。
import { createLogger } from '../../../shared/utils/logger';

statements.push(
  client
    .insert(contacts)
    .values({
      ...buildIdentityFields({
        phone: customerPhone,
        franchiseId: entry.franchiseId,
        accountId: entry.accountId,
        storeId: resolvedStoreId,
      }),
      lastActivityAt: eventTime,
      ...(formattedName != null && {
        firstName: formattedName,
        firstNameTrustScore: trustScore,
        firstNameUpdatedAt: eventTime,
      }),
      /* SMS STOP (TCPA opt-out): mark DNC on the new contact row. */
      ...(isStop && { doNotContact: true, doNotContactUpdatedBy: 'system' }),
    })
    .onConflictDoUpdate({
      target: [contacts.phone, contacts.storeId],
      set: {
        lastActivityAt: lastActivityAtForward(eventTime),
        updatedAt: sql`NOW()`,
        ...(formattedName != null && {
          firstName: sql`CASE WHEN ${trustScore} >= COALESCE(${contacts.firstNameTrustScore}, 0) THEN ${formattedName} ELSE ${contacts.firstName} END`,
          firstNameTrustScore: sql`CASE WHEN ${trustScore} >= COALESCE(${contacts.firstNameTrustScore}, 0) THEN ${trustScore} ELSE ${contacts.firstNameTrustScore} END`,
          firstNameUpdatedAt: sql`CASE WHEN ${trustScore} >= COALESCE(${contacts.firstNameTrustScore}, 0) THEN ${eventTime} ELSE ${contacts.firstNameUpdatedAt} END`,
        }),
        /*
         * SMS STOP (TCPA opt-out): set DNC. doNotContact is set true; the
         * updatedBy attribution only stamps 'system' on a false→true transition
         * so a prior 'staff' marker is preserved (mirrors contacts-analyzer's
         * sticky DNC write).
         */
        ...(isStop && {
          doNotContact: true,
          doNotContactUpdatedBy: sql`CASE WHEN ${contacts.doNotContact} = true THEN ${contacts.doNotContactUpdatedBy} ELSE 'system' END`,
        }),
      },
    }),
);

statements.push(
  client
    .insert(contacts)
    .values({
      ...buildIdentityFields({
        phone: customerPhone,
        franchiseId: entry.franchiseId,
        accountId: entry.accountId,
        storeId: resolvedStoreId,
      }),
      lastActivityAt: eventTime,
      ...(formattedName != null && {
        firstName: formattedName,
        firstNameTrustScore: trustScore,
        firstNameUpdatedAt: eventTime,
      }),
      // Plan 05: 删去 isStop spread — DNC 写入由 setDNC() 接管(Task 2 Step 3.3)
    })
    .onConflictDoUpdate({
      target: [contacts.phone, contacts.storeId],
      set: {
        lastActivityAt: lastActivityAtForward(eventTime),
        updatedAt: sql`NOW()`,
        ...(formattedName != null && {
          firstName: sql`CASE WHEN ${trustScore} >= COALESCE(${contacts.firstNameTrustScore}, 0) THEN ${formattedName} ELSE ${contacts.firstName} END`,
          firstNameTrustScore: sql`CASE WHEN ${trustScore} >= COALESCE(${contacts.firstNameTrustScore}, 0) THEN ${trustScore} ELSE ${contacts.firstNameTrustScore} END`,
          firstNameUpdatedAt: sql`CASE WHEN ${trustScore} >= COALESCE(${contacts.firstNameTrustScore}, 0) THEN ${eventTime} ELSE ${contacts.firstNameUpdatedAt} END`,
        }),
        // Plan 05: 删去 isStop spread — DNC 写入由 setDNC() 接管
      },
    }),
);

/*
 * SMS STOP cascade: after the message+DNC batch commits, close every open task
 * for this contact (shared lambda/shared/utils/dnc-cascade — same writes as the
 * AI-detected DNC path in contacts-analyzer). Deliberately AFTER the message
 * batch and in its own try: the opt-out (doNotContact=true) is already durably
 * recorded; a task-close failure must not fail the message write. Idempotent —
 * a repeat STOP finds no pending tasks. Only when contact identity fully resolved.
 */
if (isStop && customerPhone && storeId != null && entry.franchiseId && entry.accountId) {
  try {
    const dnc = await closeOpenTasksForDnc(client, {
      phone: customerPhone,
      storeId,
      franchiseId: entry.franchiseId,
      accountId: entry.accountId,
      actor: {
        actorType: 'system',
        actorSubjectId: ACTOR_SUBJECT_ID,
        actorSourceType: ACTOR_SOURCE_TYPE,
        actorSourceSystem: ACTOR_SOURCE_SYSTEM,
        closeInitiator: 'system',
      },
      idempotencyKeyFor: (taskId) =>
        `task.status_changed:dnc_stop:${String(msg.id)}:${taskId}`,
    });
    logger.info('SMS STOP — set DNC and ran task-close cascade', {
      customerPhone,
      storeId,
      closedCount: dnc.closedCount,
      taskIds: dnc.closedTaskIds,
      messageId: String(msg.id),
    });
  } catch (err) {
    logger.error('SMS STOP — DNC flag set but task-close cascade failed', {
      customerPhone,
      storeId,
      messageId: String(msg.id),
      error: err,
    });
  }
}

/*
 * Plan 05: SMS STOP cascade now uses @retaintive/common/domain shared modules.
 *
 * Two SQL units run in ONE atomic batch (spec §3.1 line 218 — DNC set + task close
 * must succeed or fail together):
 *   ① setDNC()                    — 1 UPDATE contacts SET do_not_contact=true
 *   ② closeAllOpenForContact()   — 1 UPDATE tasks (bulk close) + N timeline INSERT
 *
 * Module signatures:
 *   - setDNC: sticky, only writes true (spec §3.4 line 377-381)
 *   - closeAllOpenForContact: SELECT open tasks INSIDE module, returns SQL[]
 *     + closedTaskIds; internally uses status IN ('pending','open') filter
 *     (transitional — spec §6.2 line 785-790). Timeline rows are built INSIDE
 *     the module — caller does NOT prepend its own timeline statements.
 *
 * Try/catch boundary unchanged: cascade failure logs error but does NOT fail
 * the main message persist batch (already committed above). Idempotent — a repeat
 * STOP on an already-DNC contact finds no open tasks, closeAllOpenForContact()
 * returns closedTaskIds=[], no-op.
 */
if (isStop && customerPhone && storeId != null && entry.franchiseId && entry.accountId) {
  try {
    const dncStmt = setDNC({
      phone: customerPhone,
      storeId,
      updatedBy: 'system',
    });

    const closeResult = await closeAllOpenForContact(
      {
        contactPhone: customerPhone,
        storeId,
        closeResult: 'do_not_contact',
        closeNote: 'Auto-closed by DNC hard stop (SMS STOP keyword detected)',
        actor: { type: 'system', id: ACTOR_SUBJECT_ID },
      },
      { db: client },
    );

    if (closeResult.status === 'reject') {
      // Module-level reject is unexpected for this caller (no DNC contact / hallucinated taskId
      // possible here — caller only triggers on inbound SMS STOP). Log and continue.
      logger.error('SMS STOP — closeAllOpenForContact returned reject', {
        customerPhone,
        storeId,
        reason: closeResult.reason,
        details: closeResult.details,
        messageId: String(msg.id),
      });
    } else {
      // Atomic batch: DNC set + bulk close + N timeline. Drizzle batch supports 1+ stmts.
      // eslint-disable-next-line @typescript-eslint/no-explicit-any -- same Drizzle batch tuple workaround as main batch above
      await client.batch([dncStmt, ...closeResult.statements] as any);
      logger.info('SMS STOP — DNC set + task-close cascade committed', {
        customerPhone,
        storeId,
        closedCount: closeResult.closedTaskIds.length,
        taskIds: closeResult.closedTaskIds,
        messageId: String(msg.id),
      });
    }
  } catch (err) {
    logger.error('SMS STOP — DNC flag set but task-close cascade failed', {
      customerPhone,
      storeId,
      messageId: String(msg.id),
      error: err,
    });
  }
}

// tests/integration/stop-cascade.test.ts
import { describe, it, expect, beforeAll, afterEach } from 'vitest';
import { createNeonRepository } from '../../src/infrastructure/neon-repository';
import { createDrizzleClient, tasks, contacts, contactTimeline } from '@retaintive/common/db';
import { and, eq } from 'drizzle-orm';

const NEON_URL = process.env['NEON_TEST_DATABASE_URL'];
const SSM_PARAM = process.env['NEON_DATABASE_URL_SSM_PARAM'] ?? '/test/neon/url';

const STORE_X = 'STORE_X_PLAN05_TEST';
const CONTACT_PHONE = '+15555550001';

describe.skipIf(!NEON_URL)('STOP cascade integration', () => {
  let db: ReturnType<typeof createDrizzleClient>;
  beforeAll(() => {
    db = createDrizzleClient({ databaseUrl: NEON_URL! });
  });

  afterEach(async () => {
    // cleanup test rows
    await db.delete(contactTimeline).where(eq(contactTimeline.storeId, STORE_X));
    await db.delete(tasks).where(eq(tasks.storeId, STORE_X));
    await db.delete(contacts).where(and(eq(contacts.phone, CONTACT_PHONE), eq(contacts.storeId, STORE_X)));
  });

  it('INT1: STOP closes 2 open tasks + sets DNC + writes 2 timeline rows', async () => {
    // seed (omitted for brevity — INSERT contact + 2 tasks)
    // ...

    const repo = createNeonRepository(SSM_PARAM);
    const entry = makeEntry({ subject: 'STOP', direction: 'Inbound', type: 'SMS', customerPhone: CONTACT_PHONE });
    await repo.persistMessageBatch([entry]);

    // Assert contact.do_not_contact = true
    const [c] = await db.select().from(contacts).where(and(eq(contacts.phone, CONTACT_PHONE), eq(contacts.storeId, STORE_X)));
    expect(c.doNotContact).toBe(true);

    // Assert 2 tasks closed
    const closedTasks = await db.select().from(tasks).where(and(eq(tasks.contactPhone, CONTACT_PHONE), eq(tasks.storeId, STORE_X)));
    expect(closedTasks).toHaveLength(2);
    closedTasks.forEach(t => {
      expect(t.status).toBe('closed');
      expect(t.closeResult).toBe('do_not_contact');
      expect(t.closeNote).toBe('Auto-closed by DNC hard stop (SMS STOP keyword detected)');
      expect(t.closeType).toBe('auto_closed');
    });

    // Assert timeline: 2 task.status_changed + 1 message.created
    const timeline = await db.select().from(contactTimeline).where(eq(contactTimeline.storeId, STORE_X));
    expect(timeline.filter(t => t.eventType === 'task.status_changed')).toHaveLength(2);
    expect(timeline.filter(t => t.eventType === 'message.created')).toHaveLength(1);
  });

  // INT2 / INT3 / INT4 implementer 自己按 case map 写
});