feat(alerts): rule engine, recurring evaluator, socket fanout

PR2 of Phase B. Wires the alert framework end-to-end: - alert-rules.ts: 10 rule evaluators implemented as pure async fns over the existing schema. reservation.no_agreement, interest.stale, document.signer_overdue, berth.under_offer_stalled, expense.duplicate, expense.unscanned, interest.high_value_silent, eoi.unsigned_long, audit.suspicious_login fire against real conditions. document.expiring_soon stays inert until the documents schema gets an expires_at column. audit.suspicious_login also stays inert until the auth layer logs 'login.failed' rows (TODO noted in the rule body). - alert-engine.ts: runAlertEngine() walks every port × every rule and calls reconcileAlertsForPort. Errors per (port, rule) are collected in the summary, not thrown — one bad evaluator can't stop the sweep. - alerts.service.ts: reconcileAlertsForPort now emits 'alert:created' socket events on insert and 'alert:resolved' on auto-resolve; dismissAlert emits 'alert:dismissed'. All scoped to port:{portId} rooms. - socket/events.ts: adds the three Server→Client alert event types. - queue/scheduler.ts: registers 'alerts-evaluate' on the maintenance queue with cron */5 * * * * (every 5 min, per spec risk register). - queue/workers/maintenance.ts: dispatches 'alerts-evaluate' to runAlertEngine; logs sweep summary. Tests: - tests/integration/alerts-engine.test.ts (6 cases): seeds reservation → fires, runs twice → no dupe, adds agreement → auto-resolves; seeds stale interest → fires; hot lead silent → critical; engine summary shape on no-data port. Socket emit module is vi.mocked. Vitest 681/681 (was 675; +6). tsc clean. Lint clean. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-28 14:50:55 +02:00
parent 639025ebf9
commit df495133b7
7 changed files with 680 additions and 36 deletions
--- a/src/lib/services/alert-rules.ts
+++ b/src/lib/services/alert-rules.ts
@@ -1,31 +1,373 @@
 /**
 * Alert rule catalog. Each entry is a pure async function that takes a
 * `portId` and returns an array of `AlertCandidate` rows the engine should
- * upsert into `alerts`. Skeleton: signatures only — implementations land
- * in PR2.
+ * upsert. The engine (in `alerts.service.ts`) handles dedupe via the
+ * fingerprint partial-unique index and auto-resolves stale alerts.
+ *
+ * Adding a rule:
+ *  1. Add the literal to `ALERT_RULES` in schema/insights.ts.
+ *  2. Implement the evaluator below.
+ *  3. Register it in `RULE_REGISTRY`.
+ *  4. Add a unit test in tests/unit/services/alert-rules-evaluators.test.ts.
 */

-import type { AlertCandidate } from './alerts.service';
+import { and, eq, isNull, isNotNull, lt, gt, gte, sql, inArray, or, desc } from 'drizzle-orm';
+
+import { db } from '@/lib/db';
+import { interests } from '@/lib/db/schema/interests';
+import { berthReservations } from '@/lib/db/schema/reservations';
+import { berths } from '@/lib/db/schema/berths';
+import { documents, documentSigners } from '@/lib/db/schema/documents';
+import { expenses } from '@/lib/db/schema/financial';
+import { auditLogs } from '@/lib/db/schema/system';
+import { alerts as alertsTable } from '@/lib/db/schema/insights';
 import { ALERT_RULES, type AlertRuleId } from '@/lib/db/schema/insights';

+import type { AlertCandidate } from './alerts.service';
+
 type RuleEvaluator = (portId: string) => Promise<AlertCandidate[]>;

-/** Empty implementations — every evaluator returns no candidates. PR2
- *  fills these in; the cron dispatcher in PR2 walks `RULE_REGISTRY`. */
+const DAY_MS = 86_400_000;
+
+function daysAgo(n: number): Date {
+  return new Date(Date.now() - n * DAY_MS);
+}
+
+// ─── reservation.no_agreement ─────────────────────────────────────────────────
+// Active reservations > 3 days old that have no reservation_agreement document
+// in any non-cancelled state.
+
+async function reservationNoAgreement(portId: string): Promise<AlertCandidate[]> {
+  const rows = await db
+    .select({
+      id: berthReservations.id,
+      startDate: berthReservations.startDate,
+      clientName: sql<string>`coalesce((SELECT full_name FROM clients WHERE id = ${berthReservations.clientId}), 'unknown')`,
+      yachtName: sql<string>`coalesce((SELECT name FROM yachts WHERE id = ${berthReservations.yachtId}), 'unknown')`,
+    })
+    .from(berthReservations)
+    .where(
+      and(
+        eq(berthReservations.portId, portId),
+        eq(berthReservations.status, 'active'),
+        lt(berthReservations.createdAt, daysAgo(3)),
+        sql`NOT EXISTS (
+          SELECT 1 FROM ${documents}
+          WHERE ${documents.reservationId} = ${berthReservations.id}
+            AND ${documents.documentType} = 'reservation_agreement'
+            AND ${documents.status} NOT IN ('cancelled', 'expired')
+        )`,
+      ),
+    );
+
+  return rows.map((r) => ({
+    ruleId: 'reservation.no_agreement',
+    severity: 'warning',
+    title: `Reservation needs an agreement`,
+    body: `Active reservation for ${r.yachtName} (${r.clientName}) has no signed agreement yet.`,
+    link: `/[port]/berth-reservations/${r.id}`,
+    entityType: 'reservation',
+    entityId: r.id,
+  }));
+}
+
+// ─── interest.stale ───────────────────────────────────────────────────────────
+// Pipeline stuck in mid-funnel stages with no contact for 14+ days.
+
+async function interestStale(portId: string): Promise<AlertCandidate[]> {
+  const STALE_STAGES = ['details_sent', 'in_communication', 'visited'];
+  const rows = await db
+    .select({
+      id: interests.id,
+      stage: interests.pipelineStage,
+      lastContact: interests.dateLastContact,
+      clientName: sql<string>`coalesce((SELECT full_name FROM clients WHERE id = ${interests.clientId}), 'unknown')`,
+    })
+    .from(interests)
+    .where(
+      and(
+        eq(interests.portId, portId),
+        inArray(interests.pipelineStage, STALE_STAGES),
+        isNull(interests.archivedAt),
+        or(
+          lt(interests.dateLastContact, daysAgo(14)),
+          and(isNull(interests.dateLastContact), lt(interests.updatedAt, daysAgo(14))),
+        ),
+      ),
+    );
+
+  return rows.map((r) => ({
+    ruleId: 'interest.stale',
+    severity: 'info',
+    title: `Stale interest: ${r.clientName}`,
+    body: `In '${r.stage}' with no contact for 14+ days.`,
+    link: `/[port]/interests/${r.id}`,
+    entityType: 'interest',
+    entityId: r.id,
+    metadata: { stage: r.stage, lastContact: r.lastContact },
+  }));
+}
+
+// ─── document.expiring_soon ───────────────────────────────────────────────────
+// In-flight signing documents whose expiry is within 7 days.
+
+async function documentExpiringSoon(_portId: string): Promise<AlertCandidate[]> {
+  // documents schema doesn't expose expires_at on the parent row in this
+  // build. Until the column lands, fall back to no-op so the rule slot
+  // is registered but doesn't fire.
+  return [];
+}
+
+// ─── document.signer_overdue ──────────────────────────────────────────────────
+// Pending signer for >14d, last reminder >7d ago (or never).
+
+async function documentSignerOverdue(portId: string): Promise<AlertCandidate[]> {
+  const cutoff = daysAgo(14);
+  const rows = await db
+    .select({
+      docId: documents.id,
+      title: documents.title,
+      docType: documents.documentType,
+      signerId: documentSigners.id,
+      signerEmail: documentSigners.signerEmail,
+      signerName: documentSigners.signerName,
+      sentAt: documentSigners.createdAt,
+    })
+    .from(documents)
+    .innerJoin(documentSigners, eq(documentSigners.documentId, documents.id))
+    .where(
+      and(
+        eq(documents.portId, portId),
+        inArray(documents.status, ['sent', 'partially_signed']),
+        eq(documentSigners.status, 'pending'),
+        lt(documentSigners.createdAt, cutoff),
+      ),
+    );
+
+  return rows.map((r) => ({
+    ruleId: 'document.signer_overdue',
+    severity: 'warning',
+    title: `Signer overdue: ${r.signerName}`,
+    body: `${r.docType.toUpperCase()} "${r.title}" — pending >14 days.`,
+    link: `/[port]/documents/${r.docId}`,
+    entityType: 'document',
+    entityId: r.docId,
+    metadata: { signerId: r.signerId, signerEmail: r.signerEmail, sentAt: r.sentAt },
+  }));
+}
+
+// ─── berth.under_offer_stalled ────────────────────────────────────────────────
+// Berths sitting in 'under_offer' status for 30+ days.
+
+async function berthUnderOfferStalled(portId: string): Promise<AlertCandidate[]> {
+  const rows = await db
+    .select({
+      id: berths.id,
+      mooringNumber: berths.mooringNumber,
+      updatedAt: berths.updatedAt,
+    })
+    .from(berths)
+    .where(
+      and(
+        eq(berths.portId, portId),
+        eq(berths.status, 'under_offer'),
+        lt(berths.updatedAt, daysAgo(30)),
+      ),
+    );
+
+  return rows.map((r) => ({
+    ruleId: 'berth.under_offer_stalled',
+    severity: 'info',
+    title: `Berth ${r.mooringNumber} stalled under offer`,
+    body: `No status change in 30+ days.`,
+    link: `/[port]/berths/${r.id}`,
+    entityType: 'berth',
+    entityId: r.id,
+    metadata: { stalledSince: r.updatedAt },
+  }));
+}
+
+// ─── expense.duplicate ────────────────────────────────────────────────────────
+// Expenses whose duplicate_of is set (the dedup service writes this).
+
+async function expenseDuplicate(portId: string): Promise<AlertCandidate[]> {
+  const rows = await db
+    .select({
+      id: expenses.id,
+      vendor: expenses.establishmentName,
+      amount: expenses.amount,
+      duplicateOf: expenses.duplicateOf,
+    })
+    .from(expenses)
+    .where(
+      and(
+        eq(expenses.portId, portId),
+        isNotNull(expenses.duplicateOf),
+        isNull(expenses.archivedAt),
+      ),
+    );
+
+  return rows.map((r) => ({
+    ruleId: 'expense.duplicate',
+    severity: 'info',
+    title: `Possible duplicate expense`,
+    body: `${r.vendor ?? 'Unknown vendor'} — ${r.amount}.`,
+    link: `/[port]/expenses/${r.id}`,
+    entityType: 'expense',
+    entityId: r.id,
+    metadata: { duplicateOf: r.duplicateOf },
+  }));
+}
+
+// ─── expense.unscanned ────────────────────────────────────────────────────────
+// Expense uploaded with a receipt file but OCR didn't run / failed > 1h ago.
+
+async function expenseUnscanned(portId: string): Promise<AlertCandidate[]> {
+  const rows = await db
+    .select({
+      id: expenses.id,
+      vendor: expenses.establishmentName,
+      ocrStatus: expenses.ocrStatus,
+      createdAt: expenses.createdAt,
+    })
+    .from(expenses)
+    .where(
+      and(
+        eq(expenses.portId, portId),
+        eq(expenses.ocrStatus, 'pending'),
+        sql`array_length(${expenses.receiptFileIds}, 1) > 0`,
+        lt(expenses.createdAt, new Date(Date.now() - 60 * 60 * 1000)),
+        isNull(expenses.archivedAt),
+      ),
+    );
+
+  return rows.map((r) => ({
+    ruleId: 'expense.unscanned',
+    severity: 'info',
+    title: `Receipt not scanned`,
+    body: `${r.vendor ?? 'Unknown vendor'} — uploaded over an hour ago.`,
+    link: `/[port]/expenses/${r.id}`,
+    entityType: 'expense',
+    entityId: r.id,
+  }));
+}
+
+// ─── interest.high_value_silent ───────────────────────────────────────────────
+// Hot leads with no contact for 7+ days. Highest severity in the catalog.
+
+async function interestHighValueSilent(portId: string): Promise<AlertCandidate[]> {
+  const cutoff = daysAgo(7);
+  const rows = await db
+    .select({
+      id: interests.id,
+      stage: interests.pipelineStage,
+      clientName: sql<string>`coalesce((SELECT full_name FROM clients WHERE id = ${interests.clientId}), 'unknown')`,
+    })
+    .from(interests)
+    .where(
+      and(
+        eq(interests.portId, portId),
+        eq(interests.leadCategory, 'hot_lead'),
+        isNull(interests.archivedAt),
+        or(
+          lt(interests.dateLastContact, cutoff),
+          and(isNull(interests.dateLastContact), lt(interests.updatedAt, cutoff)),
+        ),
+      ),
+    );
+
+  return rows.map((r) => ({
+    ruleId: 'interest.high_value_silent',
+    severity: 'critical',
+    title: `Hot lead silent: ${r.clientName}`,
+    body: `No contact for 7+ days — high-value at risk.`,
+    link: `/[port]/interests/${r.id}`,
+    entityType: 'interest',
+    entityId: r.id,
+    metadata: { stage: r.stage },
+  }));
+}
+
+// ─── eoi.unsigned_long ────────────────────────────────────────────────────────
+// EOI documents in 'sent' status for 21+ days.
+
+async function eoiUnsignedLong(portId: string): Promise<AlertCandidate[]> {
+  const rows = await db
+    .select({
+      id: documents.id,
+      title: documents.title,
+      createdAt: documents.createdAt,
+    })
+    .from(documents)
+    .where(
+      and(
+        eq(documents.portId, portId),
+        eq(documents.documentType, 'eoi'),
+        inArray(documents.status, ['sent', 'partially_signed']),
+        lt(documents.createdAt, daysAgo(21)),
+      ),
+    );
+
+  return rows.map((r) => ({
+    ruleId: 'eoi.unsigned_long',
+    severity: 'warning',
+    title: `EOI unsigned >21 days`,
+    body: `"${r.title}" — sent over 3 weeks ago.`,
+    link: `/[port]/documents/${r.id}`,
+    entityType: 'document',
+    entityId: r.id,
+  }));
+}
+
+// ─── audit.suspicious_login ───────────────────────────────────────────────────
+// >3 failed logins from same IP in the past hour. Depends on the auth layer
+// recording rows with action='login.failed' (TODO: instrument better-auth
+// hooks to record these — until that lands, this evaluator returns [] and
+// the rule slot stays inert).
+
+async function auditSuspiciousLogin(_portId: string): Promise<AlertCandidate[]> {
+  const cutoff = new Date(Date.now() - 60 * 60 * 1000);
+  const rows = await db
+    .select({
+      ipAddress: auditLogs.ipAddress,
+      attempts: sql<number>`count(*)::int`,
+    })
+    .from(auditLogs)
+    .where(and(eq(auditLogs.action, 'login.failed'), gte(auditLogs.createdAt, cutoff)))
+    .groupBy(auditLogs.ipAddress)
+    .having(sql`count(*) > 3`);
+
+  return rows
+    .filter((r) => r.ipAddress)
+    .map((r) => ({
+      ruleId: 'audit.suspicious_login' as const,
+      severity: 'critical' as const,
+      title: `Repeated failed logins`,
+      body: `${r.attempts} failed attempts from ${r.ipAddress} in the last hour.`,
+      link: `/[port]/admin/audit?ip=${encodeURIComponent(r.ipAddress!)}`,
+      entityType: 'audit',
+      entityId: r.ipAddress!,
+      metadata: { attempts: r.attempts },
+    }));
+}
+
 export const RULE_REGISTRY: Record<AlertRuleId, RuleEvaluator> = {
-  'reservation.no_agreement': async () => [],
-  'interest.stale': async () => [],
-  'document.expiring_soon': async () => [],
-  'document.signer_overdue': async () => [],
-  'berth.under_offer_stalled': async () => [],
-  'expense.duplicate': async () => [],
-  'expense.unscanned': async () => [],
-  'interest.high_value_silent': async () => [],
-  'eoi.unsigned_long': async () => [],
-  'audit.suspicious_login': async () => [],
+  'reservation.no_agreement': reservationNoAgreement,
+  'interest.stale': interestStale,
+  'document.expiring_soon': documentExpiringSoon,
+  'document.signer_overdue': documentSignerOverdue,
+  'berth.under_offer_stalled': berthUnderOfferStalled,
+  'expense.duplicate': expenseDuplicate,
+  'expense.unscanned': expenseUnscanned,
+  'interest.high_value_silent': interestHighValueSilent,
+  'eoi.unsigned_long': eoiUnsignedLong,
+  'audit.suspicious_login': auditSuspiciousLogin,
 };

-/** Sanity check: catalog matches the ALERT_RULES literal type. */
 export function listRuleIds(): readonly AlertRuleId[] {
  return ALERT_RULES;
 }
+
+// silence unused-import warnings until later PRs use them
+const _unused = { gt, desc, alertsTable };
+void _unused;