Fix/four issues jobsunday002

backend/src/modules/notifications/webhook.service.ts

@@ -14,7 +14,7 @@
  * - Delivery timeout is 10 seconds per attempt.
  */
 
-import { createHash, createHmac, randomUUID } from "node:crypto";
+import { createHash, createHmac, randomUUID, timingSafeEqual } from "node:crypto";
 import { createLogger } from "../../shared/logging/logger.js";
 import type { NotificationEvent } from "./notification.types.js";
 import type { DeliveryRecord, WebhookRegistration } from "./notification.types.js";
@@ -58,19 +58,60 @@ function hashSecret(secret: string): string {
   return createHash("sha256").update(secret).digest("hex");
 }
 
-function signPayload(secret: string, body: string): string {
-  return `sha256=${createHmac("sha256", secret).update(body).digest("hex")}`;
+function signPayload(secret: string, body: string, timestamp: number): string {
+  const message = `${timestamp}.${body}`;
+  return `sha256=${createHmac("sha256", secret).update(message).digest("hex")}`;
+}
+
+/**
+ * Verify an incoming webhook signature. Receivers call this to confirm
+ * the payload originated from VaultDAO.
+ * Rejects payloads older than `maxAgeMs` (default 5 minutes) to prevent replay attacks.
+ */
+export function verifyWebhookSignature(
+  secret: string,
+  body: string,
+  signature: string,
+  timestamp: number,
+  maxAgeMs: number = 300_000,
+): boolean {
+  const age = Math.abs(Date.now() - timestamp);
+  if (age > maxAgeMs) return false;
+
+  const expected = signPayload(secret, body, timestamp);
+  if (expected.length !== signature.length) return false;
+
+  const a = Buffer.from(expected);
+  const b = Buffer.from(signature);
+  try {
+    return timingSafeEqual(a, b);
+  } catch {
+    return false;
+  }
 }
 
 function sleep(ms: number): Promise<void> {
   return new Promise((resolve) => setTimeout(resolve, ms));
 }
 
+// ── Dead-letter entry ─────────────────────────────────────────────────────────
+
+export interface DeadLetterEntry {
+  readonly id: string;
+  readonly webhookId: string;
+  readonly event: NotificationEvent;
+  readonly lastError: string;
+  readonly attempts: number;
+  readonly failedAt: string;
+  replayed: boolean;
+}
+
 // ── Service ───────────────────────────────────────────────────────────────────
 
 export class WebhookDeliveryService {
   private readonly webhooks = new Map<string, StoredWebhookRegistration>();
   private readonly deliveryStore: StorageAdapter<StoredDeliveryRecord & { id: string }>;
+  private readonly deadLetters: Map<string, DeadLetterEntry> = new Map();
 
   constructor(
     deliveryStore?: StorageAdapter<StoredDeliveryRecord & { id: string }>,
@@ -160,6 +201,79 @@ export class WebhookDeliveryService {
     );
   }
 
+  // ── Dead-letter queue ─────────────────────────────────────────────────────
+
+  /**
+   * Return all entries in the dead-letter queue.
+   */
+  public getDeadLetters(): DeadLetterEntry[] {
+    return Array.from(this.deadLetters.values());
+  }
+
+  /**
+   * Return dead-letter entries for a specific webhook.
+   */
+  public getDeadLettersForWebhook(webhookId: string): DeadLetterEntry[] {
+    return Array.from(this.deadLetters.values()).filter(
+      (dl) => dl.webhookId === webhookId,
+    );
+  }
+
+  /**
+   * Replay a dead-letter entry: re-deliver the original event to its webhook.
+   * Marks the entry as replayed on success, or updates the error on failure.
+   * @returns true if the replay delivery succeeded.
+   */
+  public async replayDeadLetter(deadLetterId: string): Promise<boolean> {
+    const entry = this.deadLetters.get(deadLetterId);
+    if (!entry) {
+      throw new Error(`Dead-letter entry not found: ${deadLetterId}`);
+    }
+
+    const webhook = this.webhooks.get(entry.webhookId);
+    if (!webhook) {
+      throw new Error(`Webhook no longer registered: ${entry.webhookId}`);
+    }
+
+    logger.info("replaying dead-letter entry", {
+      deadLetterId,
+      webhookId: webhook.id,
+      eventId: entry.event.id,
+    });
+
+    try {
+      await this.deliverToWebhook(entry.event, webhook);
+      entry.replayed = true;
+      this.deadLetters.delete(deadLetterId);
+      return true;
+    } catch (err) {
+      logger.warn("dead-letter replay failed", {
+        deadLetterId,
+        error: err instanceof Error ? err.message : String(err),
+      });
+      return false;
+    }
+  }
+
+  /**
+   * Remove all entries from the dead-letter queue.
+   * @returns the number of entries purged.
+   */
+  public purgeDeadLetters(): number {
+    const count = this.deadLetters.size;
+    this.deadLetters.clear();
+    if (count > 0) logger.info("dead-letter queue purged", { count });
+    return count;
+  }
+
+  /**
+   * Remove a single dead-letter entry.
+   * @returns true if found and removed.
+   */
+  public removeDeadLetter(deadLetterId: string): boolean {
+    return this.deadLetters.delete(deadLetterId);
+  }
+
   // ── Internal ──────────────────────────────────────────────────────────────
 
   private validateHttpsUrl(url: string): void {
@@ -181,7 +295,9 @@ export class WebhookDeliveryService {
     webhook: StoredWebhookRegistration,
   ): Promise<void> {
     const body = JSON.stringify(event);
-    const signature = signPayload(webhook.secretRaw, body);
+    const timestamp = Date.now();
+    const signature = signPayload(webhook.secretRaw, body, timestamp);
+    const deliveryId = randomUUID();
 
     let lastError: string | null = null;
 
@@ -208,6 +324,8 @@ export class WebhookDeliveryService {
             headers: {
               "Content-Type": "application/json",
               "X-VaultDAO-Signature": signature,
+              "X-VaultDAO-Timestamp": String(timestamp),
+              "X-VaultDAO-Delivery-Id": deliveryId,
               "X-VaultDAO-Event": event.topic,
             },
             body,
@@ -241,10 +359,23 @@ export class WebhookDeliveryService {
       }
     }
 
-    // All attempts exhausted
+    // All attempts exhausted — move to dead-letter queue
     await this.recordDelivery(webhook.id, event.id, "failed", MAX_ATTEMPTS, lastError);
-    logger.error("webhook delivery exhausted", {
+
+    const dlEntry: DeadLetterEntry = {
+      id: randomUUID(),
+      webhookId: webhook.id,
+      event,
+      lastError: lastError ?? "unknown error",
+      attempts: MAX_ATTEMPTS,
+      failedAt: new Date().toISOString(),
+      replayed: false,
+    };
+    this.deadLetters.set(dlEntry.id, dlEntry);
+
+    logger.error("webhook delivery exhausted, moved to dead-letter queue", {
       webhookId: webhook.id,
+      deadLetterId: dlEntry.id,
       url: webhook.url,
       eventId: event.id,
       error: lastError,

backend/src/modules/proposals/adapters/sqlite-adapter.ts

@@ -27,6 +27,35 @@ const CREATE_TABLE_SQL = `
     ON proposal_activity (contract_id, timestamp ASC);
 `;
 
+const CREATE_FTS_SQL = `
+  CREATE VIRTUAL TABLE IF NOT EXISTS proposal_activity_fts
+    USING fts5(
+      proposal_id,
+      activity_type,
+      actor,
+      data,
+      content='proposal_activity',
+      content_rowid='id'
+    );
+
+  CREATE TRIGGER IF NOT EXISTS proposal_activity_ai AFTER INSERT ON proposal_activity BEGIN
+    INSERT INTO proposal_activity_fts(rowid, proposal_id, activity_type, actor, data)
+    VALUES (new.id, new.proposal_id, new.activity_type, new.actor, new.data);
+  END;
+
+  CREATE TRIGGER IF NOT EXISTS proposal_activity_ad AFTER DELETE ON proposal_activity BEGIN
+    INSERT INTO proposal_activity_fts(proposal_activity_fts, rowid, proposal_id, activity_type, actor, data)
+    VALUES ('delete', old.id, old.proposal_id, old.activity_type, old.actor, old.data);
+  END;
+
+  CREATE TRIGGER IF NOT EXISTS proposal_activity_au AFTER UPDATE ON proposal_activity BEGIN
+    INSERT INTO proposal_activity_fts(proposal_activity_fts, rowid, proposal_id, activity_type, actor, data)
+    VALUES ('delete', old.id, old.proposal_id, old.activity_type, old.actor, old.data);
+    INSERT INTO proposal_activity_fts(rowid, proposal_id, activity_type, actor, data)
+    VALUES (new.id, new.proposal_id, new.activity_type, new.actor, new.data);
+  END;
+`;
+
 // ─── Row type (raw from SQLite) ───────────────────────────────────────────
 
 interface ActivityRow {
@@ -76,6 +105,8 @@ export class SqliteProposalActivityAdapter implements SyncProposalActivityPersis
   private readonly stmtGetByProposalId: any;
   private readonly stmtGetByContractId: any;
   private readonly stmtSummary: any;
+  private readonly stmtFtsSearch: any;
+  private readonly stmtFtsSearchByProposal: any;
 
   constructor(databasePath: string) {
     this.db = new Database(databasePath);
@@ -84,8 +115,9 @@ export class SqliteProposalActivityAdapter implements SyncProposalActivityPersis
     this.db.pragma("journal_mode = WAL");
     this.db.pragma("foreign_keys = ON");
 
-    // Create table and indexes (idempotent)
+    // Create table, indexes, and FTS virtual table (idempotent)
     this.db.exec(CREATE_TABLE_SQL);
+    this.db.exec(CREATE_FTS_SQL);
 
     // Prepare statements
     this.stmtInsert = this.db.prepare(`
@@ -123,6 +155,25 @@ export class SqliteProposalActivityAdapter implements SyncProposalActivityPersis
       WHERE proposal_id = ?
       GROUP BY proposal_id, contract_id
     `);
+
+    this.stmtFtsSearch = this.db.prepare(`
+      SELECT pa.*
+      FROM proposal_activity_fts fts
+      JOIN proposal_activity pa ON pa.id = fts.rowid
+      WHERE proposal_activity_fts MATCH ?
+      ORDER BY fts.rank
+      LIMIT ?
+    `);
+
+    this.stmtFtsSearchByProposal = this.db.prepare(`
+      SELECT pa.*
+      FROM proposal_activity_fts fts
+      JOIN proposal_activity pa ON pa.id = fts.rowid
+      WHERE proposal_activity_fts MATCH ?
+        AND pa.proposal_id = ?
+      ORDER BY fts.rank
+      LIMIT ?
+    `);
   }
 
   // ── save ─────────────────────────────────────────────────────────────────
@@ -196,6 +247,34 @@ export class SqliteProposalActivityAdapter implements SyncProposalActivityPersis
     return row ?? null;
   }
 
+  // ── search (full-text) ───────────────────────────────────────────────────
+
+  /**
+   * Full-text search across proposal activity records.
+   * Searches proposal_id, activity_type, actor, and data fields.
+   * Uses FTS5 ranking for relevance ordering.
+   */
+  search(query: string, limit: number = 50): ProposalActivity[] {
+    const rows = this.stmtFtsSearch.all(query, limit) as ActivityRow[];
+    return rows.map(rowToActivity);
+  }
+
+  /**
+   * Full-text search scoped to a single proposal.
+   */
+  searchByProposal(proposalId: string, query: string, limit: number = 50): ProposalActivity[] {
+    const rows = this.stmtFtsSearchByProposal.all(query, proposalId, limit) as ActivityRow[];
+    return rows.map(rowToActivity);
+  }
+
+  /**
+   * Rebuild the FTS index from scratch. Useful after bulk imports that
+   * bypass the trigger-based sync (e.g. restoring from a backup).
+   */
+  rebuildFtsIndex(): void {
+    this.db.exec(`INSERT INTO proposal_activity_fts(proposal_activity_fts) VALUES ('rebuild')`);
+  }
+
   // ── close ────────────────────────────────────────────────────────────────
 
   /**