`scripts/queue-slo-snapshot/queue-slo-snapshot.mjs`

96 lines
// queue-slo-snapshot — read-only SLO snapshot of the Shopify prod write queue
// (shopify_api_queue). Every DW prod write (product create/update/archive,
// GMC-exclude, April restore, draft reactivation, body-leak strip) drains
// through this queue via the shopify-queue-worker on Kamatera. Before Steve
// fires any ≤1k/day batched apply, he needs to know the live queue is healthy
// and not already backed up.
//
// ⚠️ STALENESS GATE (cycle-21 lesson: the instrument must detect its own
// blindness). The Mac2 mirror copy of shopify_api_queue is NOT in the 2-table
// logical-replication set (see memory dw-unified-kamatera-canonical), so on Mac2
// it is a FROZEN snapshot — empirically last_completed froze 2026-04-14. This
// script REFUSES to report SLO numbers as "live" when the newest completed_at is
// older than --max-stale-hours (default 6). On Mac2 it will (correctly) report
// STALE / UNRELIABLE. To get a real read, point DW_UNIFIED_URL at the canonical
// Kamatera DB (a read-only SELECT; prod-DB read is Steve-gated from Mac2).
//
//   node queue-slo-snapshot.mjs [--max-stale-hours N]
// READ-ONLY. $0. No writes.
import { execFileSync } from 'node:child_process';
import fs from 'node:fs';

const PSQL = [ '/opt/homebrew/opt/postgresql@14/bin/psql', '/usr/local/opt/postgresql@14/bin/psql', 'psql' ]
  .find(p => { try { execFileSync(p, ['--version'], { stdio: 'ignore' }); return true; } catch { return false; } }) || 'psql';
const DB = process.env.DW_UNIFIED_URL || 'postgresql:///dw_unified?host=/tmp';
const args = process.argv.slice(2);
const MAX_STALE_H = parseFloat(args.find((_,i,a)=>a[i-1]==='--max-stale-hours') || '6') || 6;
const today = new Date().toISOString().slice(0,10); // dynamic — a scheduled run must not clobber a prior day's report
const OUT = `${process.env.HOME}/.claude/yolo-queue/queue-slo-snapshot-${today}.json`;
const MD  = `${process.env.HOME}/.claude/yolo-queue/queue-slo-snapshot-${today}.md`;

function q(sql) {
  const out = execFileSync(PSQL, [DB, '-At', '-F', '|', '-c', sql], { encoding: 'utf8', maxBuffer: 64*1024*1024 });
  return out.trim() ? out.trim().split('\n').map(r => r.split('|')) : [];
}

// freshness probe
const fresh = q(`select extract(epoch from (now()-max(completed_at)))/3600.0,
                        max(completed_at), max(created_at),
                        count(*) filter (where completed_at > now()-interval '24 hours')
                 from shopify_api_queue`)[0] || [];
const hoursSinceCompleted = fresh[0] === '' || fresh[0] == null ? null : parseFloat(fresh[0]);
const lastCompleted = fresh[1] || '(none)';
const lastCreated   = fresh[2] || '(none)';
const tput24 = parseInt(fresh[3] || '0', 10);
const STALE = hoursSinceCompleted == null || hoursSinceCompleted > MAX_STALE_H;

const byStatus = q(`select status, count(*) from shopify_api_queue group by status order by 2 desc`);
const pendingAge = q(`select min(created_at), extract(epoch from (now()-min(created_at)))/3600.0
                      from shopify_api_queue where status in ('pending','queued','pending_product')`)[0] || [];
const errAgg = q(`select count(*) filter (where error is not null),
                         count(*) filter (where retries>0),
                         count(*) filter (where retries>=max_retries and max_retries>0)
                  from shopify_api_queue`)[0] || [];

const report = {
  generated_at: new Date().toISOString(), db: DB.replace(/:[^:@/]*@/, ':***@'),
  freshness: { hours_since_last_completed: hoursSinceCompleted, last_completed: lastCompleted,
    last_created: lastCreated, throughput_24h: tput24, max_stale_hours: MAX_STALE_H, stale: STALE },
  by_status: Object.fromEntries(byStatus.map(r => [r[0], parseInt(r[1],10)])),
  oldest_pending: pendingAge[0] || null, oldest_pending_age_h: pendingAge[1] ? parseFloat(pendingAge[1]) : null,
  errors: { with_error: parseInt(errAgg[0]||'0',10), retried: parseInt(errAgg[1]||'0',10), exhausted: parseInt(errAgg[2]||'0',10) },
};
fs.writeFileSync(OUT, JSON.stringify(report, null, 2));

let md = `# Shopify write-queue SLO snapshot — ${new Date().toISOString().slice(0,16)}\n\n`;
md += `DB: \`${report.db}\` · **READ-ONLY, $0.**\n\n`;
if (STALE) {
  md += `## 🔴 STALE / UNRELIABLE — do NOT treat these numbers as live\n`;
  md += `Newest \`completed_at\` is **${hoursSinceCompleted==null?'unknown':hoursSinceCompleted.toFixed(1)+'h'} old** (last: ${lastCompleted}), throughput-24h = **${tput24}**.\n`;
  md += `This copy is not being drained — on Mac2 the queue table is a frozen snapshot (not in the 2-table logical-replication set, per \`dw-unified-kamatera-canonical\`). `;
  md += `**A real SLO read must run against the canonical Kamatera DB.** Below is the frozen-snapshot content for reference only.\n\n`;
} else {
  md += `## 🟢 LIVE — drained within ${MAX_STALE_H}h\nLast completed ${hoursSinceCompleted.toFixed(2)}h ago · throughput-24h = ${tput24}.\n\n`;
}
md += `**Status distribution:** ${byStatus.map(r=>`${r[0]}=${r[1]}`).join(' · ')}\n\n`;
md += `**Oldest pending:** ${report.oldest_pending||'—'} (${report.oldest_pending_age_h?report.oldest_pending_age_h.toFixed(1)+'h':'—'})\n\n`;
md += `**Errors:** with_error=${report.errors.with_error} · retried=${report.errors.retried} · retry-exhausted=${report.errors.exhausted}\n`;
fs.writeFileSync(MD, md);

console.log(`[queue-slo] ${STALE?'🔴 STALE':'🟢 LIVE'} · last_completed=${lastCompleted} · tput24=${tput24} · pending=${report.by_status.pending||0}`);
console.log(`Report: ${MD}`);

// Alert ONLY on a genuine LIVE backup/error — NEVER on the by-design Mac2 frozen-mirror
// STALE (that's expected here and would cry-wolf nightly). best-effort, never throws.
const liveBackup = !STALE && ((report.oldest_pending_age_h ?? 0) > 12 || report.errors.exhausted > 0);
if (liveBackup) {
  try { const ac = new AbortController(); const t = setTimeout(()=>ac.abort(), 6000);
    await fetch(`${process.env.CNCP_URL||'http://localhost:3333'}/api/parking-lot`,
      { method:'POST', headers:{'Content-Type':'application/json'},
        body: JSON.stringify({ url:'file://queue-slo-snapshot',
          note:`[QUEUE-SLO 🔴] LIVE write-queue backed up: oldest_pending=${report.oldest_pending_age_h?.toFixed(1)}h · retry-exhausted=${report.errors.exhausted} · pending=${report.by_status.pending||0}` }),
        signal: ac.signal });
    clearTimeout(t); } catch { /* dashboard down — report still written */ }
}
process.exit(STALE ? 2 : 0);