refactor(storage): extract WebhookRepository across all three storage adapters (#125)

refactor(storage): extract WebhookRepository from storage adapters

Addresses the TODO on postgres.adapter.ts L49. Extracts all webhook and delivery methods from PostgresAdapter, SqliteAdapter, and MemoryAdapter into dedicated WebhookRepository classes under apps/api/src/storage/adapters/repositories/.

Each adapter becomes a thin facade that delegates to its repository. Zero behavior changes - existing webhook-storage test suite (25 tests) passes unchanged.

Establishes the pattern for extracting the remaining domains (ACL, anomaly, slowlog, commandlog, latency, memory, hotkeys, etc.) in follow-up PRs.

Made-with: Cursor

Author: amitkojha05

apps/api/src/storage/adapters/memory.adapter.ts

@@ -42,6 +42,7 @@ import type {
   MetricForecastSettings,
   MetricKind,
 } from '@betterdb/shared';
+import { WebhookMemoryRepository } from './repositories/webhook.memory.repository';
 
 export class MemoryAdapter implements StoragePort {
   private aclEntries: StoredAclEntry[] = [];
@@ -56,11 +57,10 @@ export class MemoryAdapter implements StoragePort {
   private vectorIndexSnapshots: VectorIndexSnapshot[] = [];
   private metricForecastSettings: Map<string, MetricForecastSettings> = new Map();
   private settings: AppSettings | null = null;
-  private webhooks: Map<string, Webhook> = new Map();
-  private deliveries: Map<string, WebhookDelivery> = new Map();
+  private readonly MAX_DELIVERIES_PER_WEBHOOK = 1000;
+  private readonly webhookRepo = new WebhookMemoryRepository(this.MAX_DELIVERIES_PER_WEBHOOK);
   private idCounter = 1;
   private ready: boolean = false;
-  private readonly MAX_DELIVERIES_PER_WEBHOOK = 1000;
 
   async initialize(): Promise<void> {
     this.ready = true;
@@ -755,167 +755,66 @@ export class MemoryAdapter implements StoragePort {
   }
 
   async createWebhook(webhook: Omit<Webhook, 'id' | 'createdAt' | 'updatedAt'>): Promise<Webhook> {
-    const id = randomUUID();
-    const now = Date.now();
-    const newWebhook: Webhook = {
-      id,
-      ...webhook,
-      createdAt: now,
-      updatedAt: now,
-    };
-    this.webhooks.set(id, newWebhook);
-    return { ...newWebhook };
+    return this.webhookRepo.createWebhook(webhook);
   }
 
   async getWebhook(id: string): Promise<Webhook | null> {
-    const webhook = this.webhooks.get(id);
-    return webhook ? { ...webhook } : null;
+    return this.webhookRepo.getWebhook(id);
   }
 
   async getWebhooksByInstance(connectionId?: string): Promise<Webhook[]> {
-    let webhooks = Array.from(this.webhooks.values());
-    if (connectionId) {
-      webhooks = webhooks.filter((w) => w.connectionId === connectionId || !w.connectionId);
-    } else {
-      // No connectionId provided - only return global webhooks (not scoped to any connection)
-      webhooks = webhooks.filter((w) => !w.connectionId);
-    }
-    return webhooks.sort((a, b) => b.createdAt - a.createdAt).map((w) => ({ ...w }));
+    return this.webhookRepo.getWebhooksByInstance(connectionId);
   }
 
   async getWebhooksByEvent(event: WebhookEventType, connectionId?: string): Promise<Webhook[]> {
-    let webhooks = Array.from(this.webhooks.values()).filter(
-      (w) => w.enabled && w.events.includes(event),
-    );
-    if (connectionId) {
-      // Return webhooks scoped to this connection OR global webhooks (no connectionId)
-      webhooks = webhooks.filter((w) => w.connectionId === connectionId || !w.connectionId);
-    } else {
-      // No connectionId provided - only return global webhooks (not scoped to any connection)
-      webhooks = webhooks.filter((w) => !w.connectionId);
-    }
-    return webhooks.map((w) => ({ ...w }));
+    return this.webhookRepo.getWebhooksByEvent(event, connectionId);
   }
 
   async updateWebhook(
     id: string,
     updates: Partial<Omit<Webhook, 'id' | 'createdAt' | 'updatedAt'>>,
   ): Promise<Webhook | null> {
-    const webhook = this.webhooks.get(id);
-    if (!webhook) return null;
-
-    // Filter out undefined values to prevent overwriting existing fields
-    const definedUpdates = Object.fromEntries(
-      Object.entries(updates).filter(([_, value]) => value !== undefined),
-    );
-
-    const updated: Webhook = {
-      ...webhook,
-      ...definedUpdates,
-      id,
-      createdAt: webhook.createdAt,
-      updatedAt: Date.now(),
-    };
-    this.webhooks.set(id, updated);
-    return { ...updated };
+    return this.webhookRepo.updateWebhook(id, updates);
   }
 
   async deleteWebhook(id: string): Promise<boolean> {
-    const deleted = this.webhooks.delete(id);
-    if (deleted) {
-      const deliveriesToDelete = Array.from(this.deliveries.entries())
-        .filter(([_, d]) => d.webhookId === id)
-        .map(([id]) => id);
-      deliveriesToDelete.forEach((deliveryId) => this.deliveries.delete(deliveryId));
-    }
-    return deleted;
+    return this.webhookRepo.deleteWebhook(id);
   }
 
   async createDelivery(
     delivery: Omit<WebhookDelivery, 'id' | 'createdAt'>,
   ): Promise<WebhookDelivery> {
-    const now = Date.now();
-    const id = randomUUID();
-    const newDelivery: WebhookDelivery = {
-      id,
-      ...delivery,
-      createdAt: now,
-    };
-    this.deliveries.set(id, newDelivery);
-
-    const webhookDeliveries = Array.from(this.deliveries.values())
-      .filter((d) => d.webhookId === delivery.webhookId)
-      .sort((a, b) => b.createdAt - a.createdAt);
-
-    if (webhookDeliveries.length > this.MAX_DELIVERIES_PER_WEBHOOK) {
-      const toDelete = webhookDeliveries.slice(this.MAX_DELIVERIES_PER_WEBHOOK);
-      toDelete.forEach((d) => this.deliveries.delete(d.id));
-    }
-
-    return { ...newDelivery };
+    return this.webhookRepo.createDelivery(delivery);
   }
 
   async getDelivery(id: string): Promise<WebhookDelivery | null> {
-    const delivery = this.deliveries.get(id);
-    return delivery ? { ...delivery } : null;
+    return this.webhookRepo.getDelivery(id);
   }
 
   async getDeliveriesByWebhook(
     webhookId: string,
     limit: number = 50,
     offset: number = 0,
   ): Promise<WebhookDelivery[]> {
-    return Array.from(this.deliveries.values())
-      .filter((d) => d.webhookId === webhookId)
-      .sort((a, b) => b.createdAt - a.createdAt)
-      .slice(offset, offset + limit)
-      .map((d) => ({ ...d }));
+    return this.webhookRepo.getDeliveriesByWebhook(webhookId, limit, offset);
   }
 
   async updateDelivery(
     id: string,
     updates: Partial<Omit<WebhookDelivery, 'id' | 'webhookId' | 'createdAt'>>,
   ): Promise<boolean> {
-    const delivery = this.deliveries.get(id);
-    if (!delivery) return false;
-
-    const updated: WebhookDelivery = {
-      ...delivery,
-      ...updates,
-      id,
-      webhookId: delivery.webhookId,
-      createdAt: delivery.createdAt,
-    };
-    this.deliveries.set(id, updated);
-    return true;
+    return this.webhookRepo.updateDelivery(id, updates);
   }
 
   async getRetriableDeliveries(
     limit: number = 100,
     connectionId?: string,
   ): Promise<WebhookDelivery[]> {
-    const now = Date.now();
-    let deliveries = Array.from(this.deliveries.values()).filter(
-      (d) => d.status === 'retrying' && d.nextRetryAt && d.nextRetryAt <= now,
-    );
-    if (connectionId) {
-      deliveries = deliveries.filter((d) => d.connectionId === connectionId);
-    }
-    return deliveries
-      .sort((a, b) => (a.nextRetryAt || 0) - (b.nextRetryAt || 0))
-      .slice(0, limit)
-      .map((d) => ({ ...d }));
+    return this.webhookRepo.getRetriableDeliveries(limit, connectionId);
   }
 
   async pruneOldDeliveries(cutoffTimestamp: number, connectionId?: string): Promise<number> {
-    const before = this.deliveries.size;
-    Array.from(this.deliveries.entries())
-      .filter(
-        ([_, d]) =>
-          d.createdAt < cutoffTimestamp && (!connectionId || d.connectionId === connectionId),
-      )
-      .forEach(([id]) => this.deliveries.delete(id));
-    return before - this.deliveries.size;
+    return this.webhookRepo.pruneOldDeliveries(cutoffTimestamp, connectionId);
   }
 
   // Slow Log Methods