Implement TODO stubs: webhook secret lookup, alert→incident wiring, catalog upsert/stage

- P3: getWebhookSecret() now queries DB; ingestAlert() creates/attaches incidents, auto-resolves on resolved status
- P4: stageUpdates() writes to staged_updates table; upsertService() with ON CONFLICT; getService/updateOwner implemented

products/03-alert-intelligence/src/api/webhooks.ts

@@ -133,17 +133,62 @@ function mapGrafanaSeverity(s: string | undefined): CanonicalAlert['severity'] {
 }
 
 async function getWebhookSecret(tenantSlug: string, provider: string): Promise<{ tenantId: string; secret: string } | null> {
-  // TODO: SELECT ws.secret, t.id FROM webhook_secrets ws JOIN tenants t ON ws.tenant_id = t.id WHERE t.slug = $1 AND ws.provider = $2
+  const { pool } = await import('../data/db.js');
-  return null;
+  const result = await pool.query(
+    `SELECT ws.secret, t.id as tenant_id
+     FROM webhook_secrets ws
+     JOIN tenants t ON ws.tenant_id = t.id
+     WHERE t.slug = $1 AND ws.provider = $2`,
+    [tenantSlug, provider],
+  );
+  if (!result.rows[0]) return null;
+  return { tenantId: result.rows[0].tenant_id, secret: result.rows[0].secret };
 }
 
 async function ingestAlert(tenantId: string, alert: CanonicalAlert): Promise<void> {
   await withTenant(tenantId, async (client) => {
-    await client.query(
+    // Persist raw alert
+    const alertResult = await client.query(
       `INSERT INTO alerts (tenant_id, source_provider, source_id, fingerprint, title, severity, status, service, environment, tags, raw_payload)
-       VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)`,
+       VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
+       RETURNING id`,
       [tenantId, alert.sourceProvider, alert.sourceId, alert.fingerprint, alert.title, alert.severity, alert.status, alert.service, alert.environment, JSON.stringify(alert.tags), JSON.stringify(alert.rawPayload)],
     );
-    // TODO: Feed into correlation engine
+    const alertId = alertResult.rows[0].id;
+
+    // Check for existing open incident with same fingerprint
+    const existing = await client.query(
+      `SELECT id, alert_count FROM incidents
+       WHERE fingerprint = $1 AND status IN ('open', 'acknowledged')
+       ORDER BY created_at DESC LIMIT 1`,
+      [alert.fingerprint],
+    );
+
+    if (existing.rows[0]) {
+      // Attach to existing incident
+      await client.query('UPDATE alerts SET incident_id = $1 WHERE id = $2', [existing.rows[0].id, alertId]);
+      await client.query(
+        `UPDATE incidents SET alert_count = alert_count + 1, last_alert_at = now() WHERE id = $1`,
+        [existing.rows[0].id],
+      );
+    } else if (alert.status === 'firing') {
+      // Create new incident
+      const incident = await client.query(
+        `INSERT INTO incidents (tenant_id, incident_key, fingerprint, service, title, severity, alert_count, first_alert_at, last_alert_at)
+         VALUES ($1, $2, $3, $4, $5, $6, 1, now(), now())
+         RETURNING id`,
+        [tenantId, `inc_${crypto.randomUUID().slice(0, 8)}`, alert.fingerprint, alert.service ?? 'unknown', alert.title, alert.severity],
+      );
+      await client.query('UPDATE alerts SET incident_id = $1 WHERE id = $2', [incident.rows[0].id, alertId]);
+    }
+
+    // Auto-resolve if alert status is resolved
+    if (alert.status === 'resolved') {
+      await client.query(
+        `UPDATE incidents SET status = 'resolved', resolved_at = now()
+         WHERE fingerprint = $1 AND status IN ('open', 'acknowledged')`,
+        [alert.fingerprint],
+      );
+    }
   });
 }

products/04-lightweight-idp/src/catalog/service.ts

@@ -131,22 +131,50 @@ export class CatalogService {
   }
 
   private async stageUpdates(tenantId: string, updates: StagedUpdate[]): Promise<number> {
-    // Write to staging table — admin reviews before committing
+    const { withTenant } = await import('../data/db.js');
-    // TODO: INSERT INTO staged_updates
+    await withTenant(tenantId, async (client) => {
+      for (const update of updates) {
+        await client.query(
+          `INSERT INTO staged_updates (tenant_id, service_name, source, changes)
+           VALUES ($1, $2, $3, $4)`,
+          [tenantId, update.serviceName, update.source, JSON.stringify(update.changes)],
+        );
+      }
+    });
     logger.info({ tenantId, count: updates.length }, 'Staged updates for review');
     return updates.length;
   }
 
   private async upsertService(tenantId: string, data: Partial<CatalogEntry>): Promise<void> {
-    // TODO: INSERT ... ON CONFLICT (tenant_id, name) DO UPDATE
+    const { withTenant } = await import('../data/db.js');
+    await withTenant(tenantId, async (client) => {
+      await client.query(
+        `INSERT INTO services (tenant_id, name, type, owner, owner_source, tags, metadata, last_discovered_at)
+         VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
+         ON CONFLICT (tenant_id, name) DO UPDATE SET
+           type = EXCLUDED.type, owner = EXCLUDED.owner, owner_source = EXCLUDED.owner_source,
+           tags = EXCLUDED.tags, metadata = EXCLUDED.metadata,
+           last_discovered_at = EXCLUDED.last_discovered_at, updated_at = now()`,
+        [tenantId, data.name, data.type, data.owner, data.ownerSource, JSON.stringify(data.tags ?? {}), JSON.stringify(data.metadata ?? {}), data.lastDiscoveredAt],
+      );
+    });
   }
 
   private async getService(tenantId: string, name: string): Promise<CatalogEntry | null> {
-    // TODO: SELECT from catalog
+    const { withTenant } = await import('../data/db.js');
-    return null;
+    return withTenant(tenantId, async (client) => {
+      const result = await client.query('SELECT * FROM services WHERE name = $1', [name]);
+      return result.rows[0] ?? null;
+    });
   }
 
   private async updateOwner(tenantId: string, name: string, owner: string, source: OwnerSource): Promise<void> {
-    // TODO: UPDATE catalog SET owner, owner_source
+    const { withTenant } = await import('../data/db.js');
+    await withTenant(tenantId, async (client) => {
+      await client.query(
+        'UPDATE services SET owner = $1, owner_source = $2, updated_at = now() WHERE name = $3',
+        [owner, source, name],
+      );
+    });
   }
 }