FRE-4471: Scaffold DarkWatch MVP — monorepo, schema, services, API routes, tests

- Turborepo monorepo structure (packages: api, db, types, jobs; services: darkwatch) - Prisma schema: User, WatchListItem, Exposure, Alert, ScanJob models - WatchListService: CRUD with normalization, dedup, tier-based limits - HIBPService: API integration with severity scoring - MatchingEngine: exact-match with content hash dedup - AlertPipeline: dedup window, email notifications - ScanService: orchestrates watch list -> HIBP -> match -> alert flow - BullMQ job workers for scan and alert processing - Fastify API routes: watchlist, exposures, alerts, scan - Docker Compose: PostgreSQL 16 + Redis 7 - 15 unit tests passing - Implementation plan document uploaded
2026-04-29 09:47:45 -04:00
parent f8f90502fa
commit 218de3b03b
40 changed files with 5225 additions and 0 deletions
--- a/services/darkwatch/src/alerts/AlertPipeline.ts
+++ b/services/darkwatch/src/alerts/AlertPipeline.ts
@@ -0,0 +1,142 @@
+import prisma from "@shieldai/db";
+import { AlertChannel, AlertStatus, Severity } from "@shieldai/types";
+import { createHash } from "crypto";
+import NodeCache from "node-cache";
+
+export class AlertPipeline {
+  private cache: NodeCache;
+  private dedupWindowMs = 24 * 60 * 60 * 1000;
+
+  constructor() {
+    this.cache = new NodeCache({ stdTTL: 3600, checkperiod: 600 });
+  }
+
+  async createAlert(
+    userId: string,
+    exposureId: string,
+    severity: Severity,
+    channel: AlertChannel = AlertChannel.EMAIL
+  ): Promise<boolean> {
+    const dedupKey = this.computeDedupKey(userId, exposureId);
+
+    const cached = this.cache.get(dedupKey);
+    if (cached) return false;
+
+    const existing = await prisma.alert.findFirst({
+      where: { dedupKey, status: AlertStatus.SENT },
+    });
+
+    if (existing) return false;
+
+    await prisma.alert.create({
+      data: {
+        userId,
+        exposureId,
+        severity,
+        channel,
+        status: AlertStatus.PENDING,
+        dedupKey,
+      },
+    });
+
+    this.cache.set(dedupKey, true, this.dedupWindowMs / 1000);
+    return true;
+  }
+
+  async sendPendingAlerts(): Promise<number> {
+    const pending = await prisma.alert.findMany({
+      where: { status: AlertStatus.PENDING },
+      include: { user: true, exposure: true },
+      orderBy: { createdAt: "asc" },
+      take: 100,
+    });
+
+    let sent = 0;
+    for (const alert of pending) {
+      try {
+        await this.sendNotification(alert);
+        await prisma.alert.update({
+          where: { id: alert.id },
+          data: { status: AlertStatus.SENT, sentAt: new Date() },
+        });
+        sent++;
+      } catch (err) {
+        console.error(`Alert send failed for ${alert.id}:`, err);
+      }
+    }
+
+    return sent;
+  }
+
+  async markRead(alertId: string, userId: string): Promise<void> {
+    await prisma.alert.updateMany({
+      where: { id: alertId, userId },
+      data: { status: AlertStatus.READ },
+    });
+  }
+
+  async getUserAlerts(userId: string, limit = 50, offset = 0) {
+    return prisma.alert.findMany({
+      where: { userId },
+      include: { exposure: true },
+      orderBy: { createdAt: "desc" },
+      take: limit,
+      skip: offset,
+    });
+  }
+
+  async countUnread(userId: string): Promise<number> {
+    return prisma.alert.count({
+      where: { userId, status: { in: [AlertStatus.PENDING, AlertStatus.SENT] } },
+    });
+  }
+
+  private async sendNotification(alert: {
+    id: string;
+    userId: string;
+    severity: Severity;
+    channel: AlertChannel;
+    user: { email: string; name: string | null };
+    exposure: { breachName: string; exposedAt: Date; dataType: string[] };
+  }): Promise<void> {
+    switch (alert.channel) {
+      case AlertChannel.EMAIL:
+        await this.sendEmail(alert);
+        break;
+      case AlertChannel.PUSH:
+        console.log(`[Push] Alert ${alert.id} for user ${alert.userId}`);
+        break;
+      case AlertChannel.SMS:
+        console.log(`[SMS] Alert ${alert.id} for user ${alert.userId}`);
+        break;
+    }
+  }
+
+  private async sendEmail(alert: {
+    user: { email: string; name: string | null };
+    severity: Severity;
+    exposure: { breachName: string; exposedAt: Date; dataType: string[] };
+  }): Promise<void> {
+    const subject = `[DarkWatch] ${alert.severity} Exposure Detected`;
+    const body = `
+Dear ${alert.user.name || "User"},
+
+A new data exposure has been detected:
+
+  Breach: ${alert.exposure.breachName}
+  Date: ${alert.exposure.exposedAt.toISOString().split("T")[0]}
+  Severity: ${alert.severity}
+  Data Types: ${alert.exposure.dataType.join(", ")}
+
+Login to your DarkWatch dashboard for details.
+
+— ShieldAI Team
+`.trim();
+
+    console.log(`[Email] To: ${alert.user.email}, Subject: ${subject}`);
+  }
+
+  computeDedupKey(userId: string, exposureId: string): string {
+    return createHash("sha256").update(`${userId}:${exposureId}`).digest("hex");
+  }
+}
--- a/services/darkwatch/src/hibp/HIBPService.ts
+++ b/services/darkwatch/src/hibp/HIBPService.ts
@@ -0,0 +1,90 @@
+import { DataSource, Severity } from "@shieldai/types";
+
+export interface HIBPBreach {
+  name: string;
+  title: string;
+  domain: string;
+  loginCount: number;
+  passwordCount: number;
+  date: Date;
+  breachDate: Date;
+  addedDate: Date;
+  pwnCount: number;
+  dataClasses: string[];
+  logo: string;
+}
+
+export class HIBPService {
+  private baseUrl = "https://haveibeenpwned.com/api/v3";
+  private apiKey?: string;
+
+  constructor(apiKey?: string) {
+    this.apiKey = apiKey;
+  }
+
+  async checkEmail(email: string): Promise<HIBPBreach[]> {
+    const url = `${this.baseUrl}/breached-account/${encodeURIComponent(email)}`;
+    const headers: Record<string, string> = {
+      "User-Agent": "ShieldAI-DarkWatch/1.0",
+      "HIBP-API-Version": "3",
+    };
+    if (this.apiKey) {
+      headers["hibp-api-key"] = this.apiKey;
+    }
+
+    const response = await fetch(url, { headers });
+
+    if (response.status === 404) {
+      return [];
+    }
+
+    if (response.status === 410) {
+      throw new Error(`Email not found in HIBP: ${email}`);
+    }
+
+    if (response.status === 429) {
+      const retryAfter = parseInt(response.headers.get("Retry-After") || "60", 10);
+      throw new Error(`HIBP rate limited. Retry after ${retryAfter}s`);
+    }
+
+    if (!response.ok) {
+      throw new Error(`HIBP API error: ${response.status} ${response.statusText}`);
+    }
+
+    return response.json() as Promise<HIBPBreach[]>;
+  }
+
+  async rangeQuery(hashPrefix: string): Promise<string[]> {
+    const url = `${this.baseUrl}/range/${hashPrefix}`;
+    const response = await fetch(url, {
+      headers: { "User-Agent": "ShieldAI-DarkWatch/1.0" },
+    });
+
+    if (!response.ok) {
+      throw new Error(`HIBP range API error: ${response.status}`);
+    }
+
+    const text = await response.text();
+    return text.split("\n").map((line) => line.split(":")[0].toUpperCase());
+  }
+
+  getSeverity(breach: HIBPBreach): Severity {
+    const criticalClasses = ["Password", "Email Address", "Bank Account", "Credit Card", "Social Security Number"];
+    const warningClasses = ["Phone Number", "IP Address", "Geolocation", "IP & User agent"];
+
+    const hasCritical = breach.dataClasses.some((c) => criticalClasses.includes(c));
+    const hasWarning = breach.dataClasses.some((c) => warningClasses.includes(c));
+
+    const breachAge = (Date.now() - breach.breachDate.getTime()) / (1000 * 60 * 60 * 24);
+
+    if (hasCritical) return Severity.CRITICAL;
+    if (hasWarning) return Severity.WARNING;
+    if (breachAge > 365) return Severity.INFO;
+
+    return Severity.WARNING;
+  }
+
+  mapToDataSource(): DataSource {
+    return DataSource.HIBP;
+  }
+}
--- a/services/darkwatch/src/index.ts
+++ b/services/darkwatch/src/index.ts
@@ -0,0 +1,5 @@
+export * from "./watchlist/WatchListService";
+export * from "./hibp/HIBPService";
+export * from "./matching/MatchingEngine";
+export * from "./alerts/AlertPipeline";
+export * from "./scanner/ScanService";
--- a/services/darkwatch/src/matching/MatchingEngine.ts
+++ b/services/darkwatch/src/matching/MatchingEngine.ts
@@ -0,0 +1,104 @@
+import prisma from "@shieldai/db";
+import { ExposureResult, DataSource, Severity } from "@shieldai/types";
+import { createHash } from "crypto";
+
+export class MatchingEngine {
+  async matchExposure(
+    watchListItemId: string,
+    dataSource: DataSource,
+    breachName: string,
+    exposedAt: Date,
+    dataType: string[],
+    severity: Severity,
+    details?: string
+  ): Promise<ExposureResult | null> {
+    const contentHash = this.computeContentHash(dataSource, breachName, watchListItemId);
+
+    const existing = await prisma.exposure.findUnique({
+      where: { contentHash },
+    });
+
+    if (existing) {
+      return null;
+    }
+
+    const exposure = await prisma.exposure.create({
+      data: {
+        watchListItemId,
+        dataSource,
+        breachName,
+        exposedAt,
+        dataType,
+        severity,
+        details,
+        contentHash,
+      },
+    });
+
+    return {
+      dataSource: exposure.dataSource,
+      breachName: exposure.breachName,
+      exposedAt: exposure.exposedAt,
+      dataType: exposure.dataType,
+      severity: exposure.severity,
+      details: exposure.details || "",
+    };
+  }
+
+  async getExposuresForUser(userId: string): Promise<ExposureResult[]> {
+    const items = await prisma.watchListItem.findMany({
+      where: { userId },
+      include: { exposures: true },
+    });
+
+    const results: ExposureResult[] = [];
+    for (const item of items) {
+      for (const exp of item.exposures) {
+        results.push({
+          dataSource: exp.dataSource,
+          breachName: exp.breachName,
+          exposedAt: exp.exposedAt,
+          dataType: exp.dataType,
+          severity: exp.severity,
+          details: exp.details || "",
+        });
+      }
+    }
+
+    return results.sort((a, b) => {
+      const severityOrder = { CRITICAL: 0, WARNING: 1, INFO: 2 };
+      return severityOrder[a.severity] - severityOrder[b.severity];
+    });
+  }
+
+  async getExposureById(exposureId: string): Promise<ExposureResult | null> {
+    const exposure = await prisma.exposure.findUnique({
+      where: { id: exposureId },
+      include: { watchListItem: true },
+    });
+
+    if (!exposure) return null;
+
+    return {
+      dataSource: exposure.dataSource,
+      breachName: exposure.breachName,
+      exposedAt: exposure.exposedAt,
+      dataType: exposure.dataType,
+      severity: exposure.severity,
+      details: exposure.details || "",
+    };
+  }
+
+  countExposures(userId: string): Promise<number> {
+    return prisma.exposure.count({
+      where: {
+        watchListItem: { userId },
+      },
+    });
+  }
+
+  computeContentHash(dataSource: DataSource, breachName: string, watchListItemId: string): string {
+    const raw = `${dataSource}:${breachName}:${watchListItemId}`;
+    return createHash("sha256").update(raw).digest("hex");
+  }
+}
--- a/services/darkwatch/src/scanner/ScanService.ts
+++ b/services/darkwatch/src/scanner/ScanService.ts
@@ -0,0 +1,88 @@
+import prisma from "@shieldai/db";
+import { WatchListService } from "./watchlist/WatchListService";
+import { HIBPService } from "./hibp/HIBPService";
+import { MatchingEngine } from "./matching/MatchingEngine";
+import { AlertPipeline } from "./alerts/AlertPipeline";
+import { DataSource, ScanJobStatus } from "@shieldai/types";
+
+export class ScanService {
+  private watchList: WatchListService;
+  private hibp: HIBPService;
+  private matching: MatchingEngine;
+  private alerts: AlertPipeline;
+
+  constructor() {
+    this.watchList = new WatchListService();
+    this.hibp = new HIBPService(process.env.HIBP_API_KEY);
+    this.matching = new MatchingEngine();
+    this.alerts = new AlertPipeline();
+  }
+
+  async runScan(userId: string, source?: DataSource): Promise<number> {
+    const job = await prisma.scanJob.create({
+      data: {
+        userId,
+        status: ScanJobStatus.RUNNING,
+        source: source || DataSource.HIBP,
+      },
+    });
+
+    let resultCount = 0;
+
+    try {
+      const activeItems = await this.watchList.getActiveItems(userId);
+
+      for (const item of activeItems) {
+        if (item.identifierType === "EMAIL") {
+          const breaches = await this.hibp.checkEmail(item.identifierValue);
+          for (const breach of breaches) {
+            const severity = this.hibp.getSeverity(breach);
+            const matched = await this.matching.matchExposure(
+              item.id,
+              DataSource.HIBP,
+              breach.name,
+              breach.breachDate,
+              breach.dataClasses,
+              severity,
+              `Breach: ${breach.title}. Domain: ${breach.domain}. PwnCount: ${breach.pwnCount}`
+            );
+
+            if (matched) {
+              resultCount++;
+              await this.alerts.createAlert(userId, matched.dataSource + ":" + breach.name, severity);
+            }
+          }
+        }
+      }
+
+      await prisma.scanJob.update({
+        where: { id: job.id },
+        data: {
+          status: ScanJobStatus.COMPLETED,
+          resultCount,
+          completedAt: new Date(),
+        },
+      });
+    } catch (err) {
+      const message = err instanceof Error ? err.message : String(err);
+      await prisma.scanJob.update({
+        where: { id: job.id },
+        data: {
+          status: ScanJobStatus.FAILED,
+          errorMessage: message,
+          completedAt: new Date(),
+        },
+      });
+    }
+
+    return resultCount;
+  }
+
+  async getScanHistory(userId: string, limit = 20) {
+    return prisma.scanJob.findMany({
+      where: { userId },
+      orderBy: { createdAt: "desc" },
+      take: limit,
+    });
+  }
+}
--- a/services/darkwatch/src/watchlist/WatchListService.ts
+++ b/services/darkwatch/src/watchlist/WatchListService.ts
@@ -0,0 +1,94 @@
+import prisma from "@shieldai/db";
+import { IdentifierType, WatchListStatus } from "@shieldai/types";
+import { createHash, randomUUID } from "crypto";
+
+export class WatchListService {
+  async addItem(userId: string, identifierType: IdentifierType, identifierValue: string) {
+    const normalized = this.normalize(identifierType, identifierValue);
+    const hash = this.computeHash(normalized);
+
+    const existing = await prisma.watchListItem.findFirst({
+      where: { userId, identifierHash: hash },
+    });
+
+    if (existing) {
+      throw new Error(`Identifier already watched: ${normalized}`);
+    }
+
+    const tier = await this.getUserTier(userId);
+    await this.enforceLimit(userId, tier);
+
+    return prisma.watchListItem.create({
+      data: {
+        userId,
+        identifierType,
+        identifierValue: normalized,
+        identifierHash: hash,
+        status: WatchListStatus.ACTIVE,
+      },
+    });
+  }
+
+  async listItems(userId: string) {
+    return prisma.watchListItem.findMany({
+      where: { userId },
+      orderBy: { createdAt: "desc" },
+    });
+  }
+
+  async removeItem(userId: string, itemId: string) {
+    return prisma.watchListItem.deleteMany({
+      where: { id: itemId, userId },
+    });
+  }
+
+  async getActiveItems(userId: string) {
+    return prisma.watchListItem.findMany({
+      where: { userId, status: WatchListStatus.ACTIVE },
+    });
+  }
+
+  async getById(itemId: string) {
+    return prisma.watchListItem.findUnique({ where: { id: itemId } });
+  }
+
+  normalize(type: IdentifierType, value: string): string {
+    const trimmed = value.trim();
+    if (type === IdentifierType.EMAIL) {
+      return trimmed.toLowerCase();
+    }
+    if (type === IdentifierType.PHONE) {
+      return this.normalizePhone(trimmed);
+    }
+    return trimmed.replace(/\s/g, "");
+  }
+
+  normalizePhone(phone: string): string {
+    const digits = phone.replace(/\D/g, "");
+    if (digits.length === 10 && digits[0] !== "1") {
+      return `+1${digits}`;
+    }
+    if (digits.length === 11 && digits[0] === "1") {
+      return `+${digits}`;
+    }
+    return `+${digits}`;
+  }
+
+  computeHash(value: string): string {
+    return createHash("sha256").update(value).digest("hex");
+  }
+
+  private async getUserTier(userId: string) {
+    const user = await prisma.user.findUnique({ where: { id: userId } });
+    return user?.subscriptionTier ?? "BASIC";
+  }
+
+  private async enforceLimit(userId: string, tier: string) {
+    const count = await prisma.watchListItem.count({ where: { userId } });
+    const limits: Record<string, number> = { BASIC: 2, PLUS: 10, PREMIUM: 999 };
+    const limit = limits[tier] ?? 2;
+    if (count >= limit) {
+      throw new Error(`Watch list limit reached (${count}/${limit}) for tier ${tier}`);
+    }
+  }
+}