M5: watchlist/alerts/admin routes as mountable module (2-line wire after broker-track merge)

439b0e11022b34ebfc320f2a52b0244f26e84670 · 2026-07-21 13:19:19 -0700 · Steve Abrams
Files touched

A src/server/watchlist_admin.ts
Diff

commit 439b0e11022b34ebfc320f2a52b0244f26e84670
Author: Steve Abrams <steve@designerwallcoverings.com>
Date:   Tue Jul 21 13:19:19 2026 -0700

    M5: watchlist/alerts/admin routes as mountable module (2-line wire after broker-track merge)
---
 src/server/watchlist_admin.ts | 112 ++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 112 insertions(+)

diff --git a/src/server/watchlist_admin.ts b/src/server/watchlist_admin.ts
new file mode 100644
index 0000000..87496ed
--- /dev/null
+++ b/src/server/watchlist_admin.ts
@@ -0,0 +1,112 @@
+/**
+ * M5 routes: watchlist CRUD, alerts feed, admin ingest-runs provenance + run-now.
+ * Mounted from index.ts via mountWatchlistAdmin(app) — kept separate so the
+ * broker-track and M5 workstreams never edit the same file.
+ */
+import type { Express } from 'express';
+import { execFile } from 'node:child_process';
+import { join, dirname } from 'node:path';
+import { fileURLToPath } from 'node:url';
+import { query } from '../../db/pool.ts';
+
+const __dirname = dirname(fileURLToPath(import.meta.url));
+const ROOT = join(__dirname, '..', '..');
+
+// whitelist: job name -> script (never run user-supplied paths)
+const RUN_JOBS: Record<string, string> = {
+  zillow: 'src/ingest/zillow_research.ts',
+  redfin: 'src/ingest/redfin_tracker.ts',
+  fhfa: 'src/ingest/fhfa_hpi.ts',
+  acs: 'src/ingest/census_acs.ts',
+  derive: 'src/ingest/derive_metrics.ts',
+  score: 'src/score/opportunity.ts',
+  alerts: 'src/jobs/alerts_check.ts',
+  brokers: 'src/ingest/brokers/engine.ts',
+};
+let jobRunning: string | null = null; // one at a time — these are heavyweight
+
+export function mountWatchlistAdmin(app: Express): void {
+  app.get('/api/watchlist', async (_req, res) => {
+    try {
+      const r = await query(
+        `SELECT w.id, w.label, w.thresholds, w.created_at,
+                rg.canonical_key AS key, rg.name, rg.state_code AS state, rg.region_type,
+                rs.opportunity_score,
+                (SELECT value FROM metric_series WHERE region_id = rg.id AND metric='zhvi' ORDER BY period DESC LIMIT 1) AS zhvi,
+                (SELECT value FROM metric_series WHERE region_id = rg.id AND metric='zhvi_yoy' ORDER BY period DESC LIMIT 1) AS zhvi_yoy,
+                (SELECT value FROM metric_series WHERE region_id = rg.id AND metric='dom' ORDER BY period DESC LIMIT 1) AS dom
+           FROM watchlist w
+           JOIN region rg ON rg.id = w.region_id
+           LEFT JOIN region_scores rs ON rs.region_id = rg.id
+                AND rs.score_date = (SELECT MAX(score_date) FROM region_scores)
+          ORDER BY w.created_at DESC`,
+      );
+      res.json({ items: r.rows });
+    } catch (e: any) { res.status(500).json({ error: String(e.message || e) }); }
+  });
+
+  app.post('/api/watchlist', async (req, res) => {
+    try {
+      const { key, thresholds, label } = req.body || {};
+      const rg = await query<{ id: number; name: string }>(
+        `SELECT id, name FROM region WHERE canonical_key = $1`, [String(key || '')],
+      );
+      if (!rg.rows.length) return res.status(400).json({ error: 'unknown region key' });
+      const r = await query(
+        `INSERT INTO watchlist (region_id, label, thresholds) VALUES ($1, $2, $3) RETURNING id`,
+        [rg.rows[0].id, label || rg.rows[0].name, thresholds ? JSON.stringify(thresholds) : null],
+      );
+      res.json({ ok: true, id: r.rows[0].id });
+    } catch (e: any) { res.status(500).json({ error: String(e.message || e) }); }
+  });
+
+  app.delete('/api/watchlist/:id', async (req, res) => {
+    try {
+      const id = Number(req.params.id);
+      if (!Number.isInteger(id)) return res.status(400).json({ error: 'bad id' });
+      await query(`DELETE FROM alerts_log WHERE watchlist_id = $1`, [id]);
+      await query(`DELETE FROM watchlist WHERE id = $1`, [id]);
+      res.json({ ok: true });
+    } catch (e: any) { res.status(500).json({ error: String(e.message || e) }); }
+  });
+
+  app.get('/api/alerts', async (_req, res) => {
+    try {
+      const r = await query(
+        `SELECT a.id, a.metric, a.observed, a.fired_at, a.detail,
+                rg.name, rg.state_code AS state, rg.canonical_key AS key
+           FROM alerts_log a
+           LEFT JOIN watchlist w ON w.id = a.watchlist_id
+           LEFT JOIN region rg ON rg.id = w.region_id
+          ORDER BY a.fired_at DESC LIMIT 50`,
+      );
+      res.json({ alerts: r.rows });
+    } catch (e: any) { res.status(500).json({ error: String(e.message || e) }); }
+  });
+
+  app.get('/api/admin/runs', async (_req, res) => {
+    try {
+      const r = await query(
+        `SELECT id, source, url, file_hash, started_at, finished_at, status,
+                rows_upserted, rows_skipped, notes
+           FROM ingest_runs ORDER BY id DESC LIMIT 60`,
+      );
+      res.json({ runs: r.rows });
+    } catch (e: any) { res.status(500).json({ error: String(e.message || e) }); }
+  });
+
+  app.post('/api/admin/run/:job', (req, res) => {
+    const job = String(req.params.job);
+    const script = RUN_JOBS[job];
+    if (!script) return res.status(400).json({ error: 'unknown job' });
+    if (jobRunning) return res.status(409).json({ error: `job already running: ${jobRunning}` });
+    jobRunning = job;
+    execFile('npx', ['tsx', script], { cwd: ROOT, timeout: 45 * 60_000, maxBuffer: 16 * 1024 * 1024 },
+      (err, stdout, stderr) => {
+        jobRunning = null;
+        const tail = (stdout || '').trim().split('\n').slice(-2).join(' | ');
+        if (err) return res.status(500).json({ error: String(err.message), tail: (stderr || '').slice(-500) });
+        res.json({ ok: true, summary: tail });
+      });
+  });
+}
← f42ca12 M5 (partial): refresh orchestrator, alerts checker, monthly · back to Nationalrealestate · M-B1 national broker registry: 5 open-state adapters (TX/FL/ 1af53fa →