/** * Campaign Grouping Service * * Shared grouping core called by all 3 trigger sites: the webhook path * (`webhook-service.ts`'s `triggerPhishingDetection()`), the cron sweep * (`phishing-sweep-service.ts`), and the on-demand * `POST /api/phishing/tickets/{id}/analyze` route. One deterministic, * testable function computes a tiered campaign key (CAMP-01) and * finds-or-creates a `campaigns` row (CAMP-02) — no duplicated matching * logic between callers, mirroring `phishing-detector.ts`'s shared-core * architecture. * * D-07 limitation (load-bearing, stated explicitly): `groupReportIntoCampaign` * always runs BEFORE `parseAndStoreMessage` on the automatic webhook path * (see `webhook-service.ts`'s `triggerPhishingDetection()` — grouping happens * first, parsing happens afterward inside `runGatedPhishingStages()`). That * means at grouping time the CURRENT report never has its own `messages`/ * `indicators` row yet, so Tier 1 (Message-ID) and Tier 2 (attachment-hash/ * URL-domain) — both of which require the report's OWN signal to search * for candidates — can never match on the automatic path's one-and-only * grouping call (subsequent webhook events short-circuit via * `skipIfAlreadyGrouped`). Automatic grouping therefore always resolves via * Tier 3 (normalized subject + company + 24h window). * * Bug fix (debug session phishing-recipient-seubert): Tier 3 previously * scoped its match to `reports.requester_contact_id` — i.e. it only ever * merged reports filed by the SAME reporting employee. Since Tier 3 is the * only tier automatic grouping can ever reach (see D-07 above), that meant * the same phishing campaign sent to and reported by MULTIPLE different * employees at the same company could never be consolidated into one * campaign — each recipient's report silently became its own single-report * campaign, so any single ticket's evidence/blast-radius view under-reported * the campaign's true recipient list. Tier 3 now scopes to company + subject * only (no contact/requester restriction), matching its `sender_subject_ * client` name's original intent of grouping the same external campaign * across a company, independent of who reported it. (`sender` isn't * literally available yet at this point — see D-07 — so "client" scoping is * company-wide, deliberately wider than a single reporter.) */ import type { PoolClient } from 'pg'; import { postgresClient } from './postgres-client'; // ============================================================================= // Pure logic — tier-key helpers (mirrors matchesPhishingPatterns / computePhishingContentHash) // ============================================================================= /** * D-03: strip leading Re:/Fwd:/Fw: (repeated, case-insensitive), lowercase, trim. */ export function normalizeSubject(subject: string | null): string { let s = (subject ?? '').trim(); const prefixRe = /^(re|fwd|fw):\s*/i; while (prefixRe.test(s)) { s = s.replace(prefixRe, '').trim(); } return s.toLowerCase(); } /** * Pitfall 4: indicators.value for indicator_type='url' is a bare URL string, * not a domain — extract at read time, guarded (malformed/relative URLs are * possible in real-world phishing emails). */ export function extractUrlDomain(url: string): string | null { try { return new URL(url).hostname || null; } catch { return null; } } // ============================================================================= // Orchestration — groupReportIntoCampaign (mirrors detectPhishingTicket) // ============================================================================= export interface GroupReportResult { campaignId: string; groupMethod: 'message_id' | 'attachment_or_url' | 'sender_subject_client'; created: boolean; } interface OwnReportRow { title: string | null; company_id: number | null; created_at: string; campaign_id: string | null; } interface OwnMessageRow { id: string; message_id: string | null; } interface IndicatorRow { indicator_type: string; value: string; } interface CandidateIndicatorRow extends IndicatorRow { message_id: string; } interface Tier2CandidateRow { report_id: string; campaign_id: string; title: string | null; message_id: string; } /** Builds a `message_id:...` key — the strongest available signal (Tier 1). */ function computeTier1Key(messageId: string | null): string | null { return messageId ? `message_id:${messageId}` : null; } /** Builds an `attachment_or_url:...` key from own indicators + subject + sender (Tier 2). */ function computeTier2Key( attachmentHashes: string[], urlDomains: string[], normalizedSubject: string, senderValue: string | null ): string | null { if (!senderValue || !normalizedSubject) return null; if (attachmentHashes.length === 0 && urlDomains.length === 0) return null; const keyParts = [...attachmentHashes, ...urlDomains].sort(); return `attachment_or_url:${keyParts.join(',')}:${normalizedSubject}:${senderValue}`; } /** * Builds a `sender_subject_client:...` key from normalized subject + company * (Tier 3) — deliberately company-wide, NOT scoped to a single reporting * contact, so the same campaign reported by different employees at the same * company still consolidates into one campaign (see file-level bug-fix note). */ function computeTier3Key(normalizedSubject: string, companyId: number | null): string | null { if (!normalizedSubject || !companyId) return null; return `sender_subject_client:${normalizedSubject}:${companyId}`; } /** * CR-02: decrements a now-abandoned origin campaign's report_count. Shared * by both places a report can be abandoned from its origin campaign — the * sibling-migration branch (moving to a DIFFERENT existing campaign) and the * signal-diverged create-new fall-through (moving to a brand-new campaign) — * so neither is left with a stale, inflated count. Decrement only; never * deletes the origin row even if its count reaches 0 (out of scope — see * CR-02 scope note at the call sites). */ async function decrementOriginCampaign(client: PoolClient, originCampaignId: string): Promise { await client.query( `UPDATE campaigns SET report_count = report_count - 1, updated_at = NOW() WHERE id = $1`, [originCampaignId] ); } /** * Shared entry point called by the webhook path, the cron sweep, and the * on-demand `/analyze` route. Looks up the report's own signal (message, * indicators, requester/company), tries Tier 1 -> Tier 2 -> Tier 3 matching * queries in order inside a single transaction (Pitfall 2 — `campaigns. * campaign_key` has no UNIQUE constraint), and either links the report to an * existing campaign (bumping report_count/last_seen_at, CAMP-02) or inserts a * new campaigns row. * * D-08: `opts.skipIfAlreadyGrouped` short-circuits (returns null) when the * report is already linked — used by the automatic webhook/cron path to * avoid redundant work on every re-fire/sweep pass. The `/analyze` route * omits this option so it always re-runs full tiered matching (it may * upgrade a Tier-3-only report to Tier 1 after `parseAndStoreMessage` has * just populated `messages`/`indicators` for the first time). * * D-04: never merges campaigns — attaches to the first/best tiered match * found, or creates a new campaign. Multi-campaign-merge logic is explicitly * out of scope this phase. */ export async function groupReportIntoCampaign( reportId: string, opts?: { skipIfAlreadyGrouped?: boolean } ): Promise { try { if (opts?.skipIfAlreadyGrouped) { const existing = await postgresClient.query<{ campaign_id: string | null }>( `SELECT campaign_id::text AS campaign_id FROM reports WHERE id = $1`, [reportId] ); if (existing.rows[0]?.campaign_id) { return null; // already grouped, short-circuit (D-08) } } return await postgresClient.transaction(async (client) => { const ownReportRes = await client.query( `SELECT title, company_id, created_at, campaign_id::text AS campaign_id FROM reports WHERE id = $1`, [reportId] ); const ownReport = ownReportRes.rows[0]; if (!ownReport) { throw new Error(`groupReportIntoCampaign: report ${reportId} not found`); } const normalizedSubject = normalizeSubject(ownReport.title); // Own messages row, if `parseAndStoreMessage` has already run for this // report (D-07 — not the case for the automatic webhook/cron path // until an explicit /analyze call has happened at least once). const ownMessageRes = await client.query( `SELECT id::text AS id, message_id FROM messages WHERE report_id = $1 LIMIT 1`, [reportId] ); const ownMessage = ownMessageRes.rows[0] ?? null; let matchCampaignId: string | null = null; let matchGroupMethod: GroupReportResult['groupMethod'] | null = null; // --------------------------------------------------------------------- // Tier 1: Message-ID. Self-exclusion (`r.id != $2`) is REQUIRED — without // it this query would trivially match the report's own messages row // against itself, double-incrementing its own already-linked campaign // on every re-run (plan-checker finding). // --------------------------------------------------------------------- if (ownMessage?.message_id) { const tier1 = await client.query<{ campaign_id: string }>( `SELECT r.campaign_id::text AS campaign_id FROM messages m JOIN reports r ON r.id = m.report_id WHERE m.message_id = $1 AND r.campaign_id IS NOT NULL AND r.id != $2 ORDER BY r.created_at ASC LIMIT 1`, [ownMessage.message_id, reportId] ); if (tier1.rows[0]?.campaign_id) { matchCampaignId = tier1.rows[0].campaign_id; matchGroupMethod = 'message_id'; } } // --------------------------------------------------------------------- // Tier 2: attachment-hash / URL-domain + normalized subject + sender, // within a 24h window (D-02). Only reachable if the own report has its // own indicators (i.e. parseAndStoreMessage already ran for it). // Self-exclusion (`r.id != $1`) required for the same reason as Tier 1. // --------------------------------------------------------------------- let ownAttachmentHashes: string[] = []; let ownUrlDomains: string[] = []; let ownSenderValue: string | null = null; if (!matchCampaignId && ownMessage) { const ownIndicatorsRes = await client.query( `SELECT indicator_type, value FROM indicators WHERE message_id = $1`, [ownMessage.id] ); ownAttachmentHashes = ownIndicatorsRes.rows .filter((i) => i.indicator_type === 'attachment_hash') .map((i) => i.value); ownUrlDomains = ownIndicatorsRes.rows .filter((i) => i.indicator_type === 'url') .map((i) => extractUrlDomain(i.value)) .filter((d): d is string => d !== null); ownSenderValue = ownIndicatorsRes.rows.find((i) => i.indicator_type === 'sender')?.value ?? null; if ( (ownAttachmentHashes.length > 0 || ownUrlDomains.length > 0) && ownSenderValue && normalizedSubject ) { const tier2Candidates = await client.query( `SELECT r.id::text AS report_id, r.campaign_id::text AS campaign_id, r.title, m.id::text AS message_id FROM messages m JOIN reports r ON r.id = m.report_id WHERE r.campaign_id IS NOT NULL AND r.id != $1 AND r.created_at BETWEEN $2::timestamptz - INTERVAL '24 hours' AND $2::timestamptz + INTERVAL '24 hours'`, [reportId, ownReport.created_at] ); if (tier2Candidates.rows.length > 0) { const candidateMessageIds = tier2Candidates.rows.map((c) => c.message_id); const candidateIndicatorsRes = await client.query( `SELECT message_id::text AS message_id, indicator_type, value FROM indicators WHERE message_id = ANY($1::uuid[])`, [candidateMessageIds] ); const indicatorsByMessage = new Map(); for (const ind of candidateIndicatorsRes.rows) { const arr = indicatorsByMessage.get(ind.message_id) ?? []; arr.push(ind); indicatorsByMessage.set(ind.message_id, arr); } for (const candidate of tier2Candidates.rows) { if (normalizeSubject(candidate.title) !== normalizedSubject) continue; const candidateIndicators = indicatorsByMessage.get(candidate.message_id) ?? []; const candidateSender = candidateIndicators.find( (i) => i.indicator_type === 'sender' )?.value; if (candidateSender !== ownSenderValue) continue; const candidateHashes = candidateIndicators .filter((i) => i.indicator_type === 'attachment_hash') .map((i) => i.value); const candidateDomains = candidateIndicators .filter((i) => i.indicator_type === 'url') .map((i) => extractUrlDomain(i.value)) .filter((d): d is string => d !== null); const hashMatch = ownAttachmentHashes.some((h) => candidateHashes.includes(h)); const domainMatch = ownUrlDomains.some((d) => candidateDomains.includes(d)); if (hashMatch || domainMatch) { matchCampaignId = candidate.campaign_id; matchGroupMethod = 'attachment_or_url'; break; } } } } } // --------------------------------------------------------------------- // Tier 3: normalizeSubject(title) + company + 24h window (D-02, fixed // per file-level bug-fix note). Company-wide — deliberately NOT scoped // to `reports.requester_contact_id` — so the same campaign reported by // different employees at the same company still consolidates into one // campaign. Self-exclusion (`r.id != $2`) required for the same reason // as Tiers 1-2. // --------------------------------------------------------------------- if (!matchCampaignId && ownReport.company_id && normalizedSubject) { const tier3 = await client.query<{ campaign_id: string; title: string | null }>( `SELECT r.campaign_id::text AS campaign_id, r.title FROM reports r WHERE r.company_id = $1 AND r.campaign_id IS NOT NULL AND r.id != $2 AND r.created_at BETWEEN $3::timestamptz - INTERVAL '24 hours' AND $3::timestamptz + INTERVAL '24 hours' ORDER BY r.created_at ASC`, [ownReport.company_id, reportId, ownReport.created_at] ); const match = tier3.rows.find((r) => normalizeSubject(r.title) === normalizedSubject); if (match) { matchCampaignId = match.campaign_id; matchGroupMethod = 'sender_subject_client'; } } // --------------------------------------------------------------------- // Own-signal tier keys, computed once here and reused by both the // own-campaign revalidation guard (CR-03, below) and the create-new // fall-through further down — avoids computing them twice. // --------------------------------------------------------------------- const tier1Key = computeTier1Key(ownMessage?.message_id ?? null); const tier2Key = computeTier2Key( ownAttachmentHashes, ownUrlDomains, normalizedSubject, ownSenderValue ); const tier3Key = computeTier3Key(normalizedSubject, ownReport.company_id); const currentKeys = [tier1Key, tier2Key, tier3Key, `report:${reportId}`].filter( (k): k is string => k !== null ); // --------------------------------------------------------------------- // Find-or-create against `campaigns` (CAMP-02). // --------------------------------------------------------------------- if (matchCampaignId && matchGroupMethod) { // CR-01/CAMP-02 no-op guard: tier queries only exclude the report's // OWN row (`r.id != $x`), not its siblings already in the same // campaign. Since `/analyze` re-runs grouping unconditionally // (D-08), a report already linked to campaign X can re-match X via // a sibling row on every repeat call, inflating `report_count` // without bound. If the match resolves to the report's own current // campaign, it's already correctly linked — skip both UPDATEs. if (matchCampaignId === ownReport.campaign_id) { return { campaignId: matchCampaignId, groupMethod: matchGroupMethod, created: false }; } await client.query( `UPDATE campaigns SET report_count = report_count + 1, last_seen_at = NOW(), updated_at = NOW() WHERE id = $1`, [matchCampaignId] ); await client.query( `UPDATE reports SET campaign_id = $1, updated_at = NOW() WHERE id = $2`, [matchCampaignId, reportId] ); // CR-02: if the report had a DIFFERENT prior campaign (the // same-campaign case returned above; a never-grouped report has no // origin — D-08), decrement it so it doesn't keep a permanently // stale report_count for a report it no longer holds. Scope: // decrement only, never delete the origin row even if its count // reaches 0 (deleting introduces FK-cascade/detail-route risk out of // scope for this gap fix; an empty campaign is a strictly lesser // problem than a stale count). if (ownReport.campaign_id) { await decrementOriginCampaign(client, ownReport.campaign_id); } return { campaignId: matchCampaignId, groupMethod: matchGroupMethod, created: false }; } // --------------------------------------------------------------------- // CR-03: no sibling matched on any tier (matchCampaignId is null). // Every tier query above self-excludes the report's own row // (`r.id != $x`), so a single-report campaign (no sibling report // exists yet — the common case for any brand-new phishing lure before // a second person reports it) always reaches here with matchCampaignId // still null. Pre-fix, control fell straight into "create new // campaign" below, abandoning the report's still-valid campaign and // producing a SECOND campaigns row with the identical campaign_key on // every repeat /analyze call. This guard sits OUTSIDE the sibling- // match branch above precisely because it must only run when NO // sibling matched — it never suppresses a legitimate D-08 sibling // upgrade, which returns from the branch above instead. // --------------------------------------------------------------------- if (!matchCampaignId && ownReport.campaign_id) { const ownCampaignRes = await client.query<{ campaign_key: string; group_method: string }>( `SELECT campaign_key, group_method FROM campaigns WHERE id = $1`, [ownReport.campaign_id] ); const ownCampaign = ownCampaignRes.rows[0]; if (ownCampaign && currentKeys.includes(ownCampaign.campaign_key)) { // The report's own campaign still satisfies its freshly-recomputed // current tier key — it's still validly linked, reuse it as-is. return { campaignId: ownReport.campaign_id, groupMethod: ownCampaign.group_method as GroupReportResult['groupMethod'], created: false, }; } // CR-02: own campaign row is gone, or the report's signal has // genuinely diverged from its stored key — this report is about to // be abandoned from its origin campaign in favor of a brand-new one // below (the same class of abandonment as the sibling-migration // branch above, just landing on a new row instead of an existing // sibling's). Decrement now so the origin doesn't show a stale // count. Falls through to create-new unchanged. await decrementOriginCampaign(client, ownReport.campaign_id); } // No match on any tier — create a new campaign. Prefer the strongest // available key (Tier 1 > Tier 2 > Tier 3) so future duplicates of THIS // report can match it; do NOT set/transition campaigns.status (leave // migration default 'open' — Phase 19/20's concern). let newCampaignKey: string; let newGroupMethod: GroupReportResult['groupMethod']; if (tier1Key) { newCampaignKey = tier1Key; newGroupMethod = 'message_id'; } else if (tier2Key) { newCampaignKey = tier2Key; newGroupMethod = 'attachment_or_url'; } else if (tier3Key) { newCampaignKey = tier3Key; newGroupMethod = 'sender_subject_client'; } else { newCampaignKey = `report:${reportId}`; newGroupMethod = 'sender_subject_client'; } const newCampaign = await client.query<{ id: string }>( `INSERT INTO campaigns (campaign_key, group_method, first_seen_at, last_seen_at, report_count) VALUES ($1, $2, NOW(), NOW(), 1) RETURNING id::text AS id`, [newCampaignKey, newGroupMethod] ); const newCampaignId = newCampaign.rows[0].id; await client.query( `UPDATE reports SET campaign_id = $1, updated_at = NOW() WHERE id = $2`, [newCampaignId, reportId] ); return { campaignId: newCampaignId, groupMethod: newGroupMethod, created: true }; }); } catch (error) { console.error('[CAMPAIGN-GROUPING] Failed to group report into campaign', reportId, error); throw error; } }