wulf-pulse/lib/services/phishing-sweep-service.ts

/**
 * Phishing Sweep Service
 *
 * Bounded cron reconciliation sweep over recently-modified tickets, mirroring
 * ticket-reconciliation-service.ts's structure. The webhook path (Plan 03,
 * webhook-service.ts) is the primary near-real-time trigger; this sweep
 * catches anything the webhook missed (dropped events, backfilled tickets,
 * tickets whose content changed after the webhook already fired) by
 * re-running the same shared detectPhishingTicket core.
 *
 * No duplicated pattern-matching/hashing logic here — this file only
 * queries candidate tickets and delegates detection to phishing-detector.ts.
 */

import { postgresClient } from './postgres-client';
import { detectPhishingTicket, type DetectableTicket } from './phishing-detector';
import { groupReportIntoCampaign } from './campaign-grouping-service';
import { createSyncLogger } from '../utils/sync-logger';

export interface PhishingSweepResult {
  scanned: number;
  flagged: number;
  skippedUnchanged: number;
  errors: number;
}

const SWEEP_WINDOW_DAYS = 7;
const SCAN_LIMIT = 500;

/**
 * Re-scan up to SCAN_LIMIT recently-modified tickets through the shared
 * phishing detector. Bounded and idempotent — safe to run on a recurring
 * schedule alongside the webhook trigger.
 */
export async function sweepPhishingTickets(): Promise<PhishingSweepResult> {
  const logger = createSyncLogger({ component: 'PhishingSweep' });
  const startedAt = Date.now();
  const result: PhishingSweepResult = {
    scanned: 0,
    flagged: 0,
    skippedUnchanged: 0,
    errors: 0,
  };

  const candidatesQuery = `
    SELECT id, ticket_number, title, description, company_id, contact_id, created_by_contact_id
      FROM tickets
     WHERE is_deleted = false
       AND last_activity_date > NOW() - INTERVAL '${SWEEP_WINDOW_DAYS} days'
     ORDER BY last_activity_date DESC
     LIMIT $1
  `;
  const candidates = await postgresClient.query<{
    id: string;
    ticket_number: string | null;
    title: string | null;
    description: string | null;
    company_id: number | null;
    contact_id: number | null;
    created_by_contact_id: number | null;
  }>(candidatesQuery, [SCAN_LIMIT]);

  logger.info(
    `Sweeping ${candidates.rows.length} recently-modified tickets (>${SWEEP_WINDOW_DAYS}d, limit=${SCAN_LIMIT})`
  );

  for (const row of candidates.rows) {
    result.scanned += 1;
    const ticket: DetectableTicket = {
      id: Number(row.id),
      ticket_number: row.ticket_number,
      title: row.title,
      description: row.description,
      company_id: row.company_id,
      contact_id: row.contact_id,
      created_by_contact_id: row.created_by_contact_id,
    };

    try {
      const detection = await detectPhishingTicket(ticket);
      if (detection.skippedUnchanged) {
        result.skippedUnchanged += 1;
      } else if (detection.flagged) {
        result.flagged += 1;
      }
      // D-01/D-08: grouping runs regardless of skippedUnchanged (a report
      // could have been created by a previous sweep pass and still lack a
      // campaign_id if grouping failed transiently that time); short-circuits
      // internally if already grouped.
      if (detection.flagged && detection.reportId) {
        await groupReportIntoCampaign(detection.reportId, { skipIfAlreadyGrouped: true });
      }
    } catch (err) {
      result.errors += 1;
      logger.warn(
        `Phishing detection failed for ticket ${ticket.id}`,
        { ticketId: ticket.id },
        err instanceof Error ? err : new Error(String(err))
      );
    }
  }

  logger.info(
    `Phishing sweep complete: scanned=${result.scanned} flagged=${result.flagged} skippedUnchanged=${result.skippedUnchanged} errors=${result.errors}`,
    { duration: Date.now() - startedAt }
  );

  return result;
}