← back to AbramsOS
tick 3 (scaffold): CPSC recall_event + recall_match tables + matcher + fetcher
600c717969e9f05bf44ec8149c628b36a4bfd2a2 · 2026-05-10 00:30:43 -0700 · Steve
- db/migrations/0002_recalls.sql: recall_event, recall_match, recall_pull_log
- lib/cpsc-fetcher.js: hits saferproducts.gov public API + normalizes to our shape
- lib/recall-matcher.js: canonical scoring pseudocode from docs/SPEC.md
(gtin/upc 0.45, model fuzzy 0.25, brand 0.05, post-purchase 0.05, UDI 0.40)
- scripts/ingest-recalls.js: idempotent backfill (run: node scripts/ingest-recalls.js [days])
- No live data ingested yet; ingest is opt-in
Files touched
A db/migrations/0002_recalls.sqlA lib/cpsc-fetcher.jsA lib/recall-matcher.jsA scripts/ingest-recalls.js
Diff
commit 600c717969e9f05bf44ec8149c628b36a4bfd2a2
Author: Steve <steve@designerwallcoverings.com>
Date: Sun May 10 00:30:43 2026 -0700
tick 3 (scaffold): CPSC recall_event + recall_match tables + matcher + fetcher
- db/migrations/0002_recalls.sql: recall_event, recall_match, recall_pull_log
- lib/cpsc-fetcher.js: hits saferproducts.gov public API + normalizes to our shape
- lib/recall-matcher.js: canonical scoring pseudocode from docs/SPEC.md
(gtin/upc 0.45, model fuzzy 0.25, brand 0.05, post-purchase 0.05, UDI 0.40)
- scripts/ingest-recalls.js: idempotent backfill (run: node scripts/ingest-recalls.js [days])
- No live data ingested yet; ingest is opt-in
---
db/migrations/0002_recalls.sql | 50 ++++++++++++++++++
lib/cpsc-fetcher.js | 59 +++++++++++++++++++++
lib/recall-matcher.js | 114 +++++++++++++++++++++++++++++++++++++++++
scripts/ingest-recalls.js | 78 ++++++++++++++++++++++++++++
4 files changed, 301 insertions(+)
diff --git a/db/migrations/0002_recalls.sql b/db/migrations/0002_recalls.sql
new file mode 100644
index 0000000..b74a34b
--- /dev/null
+++ b/db/migrations/0002_recalls.sql
@@ -0,0 +1,50 @@
+-- 0002_recalls.sql — recall_event + recall_match tables
+-- Apply: psql -d abrams_os -f db/migrations/0002_recalls.sql
+
+BEGIN;
+
+CREATE TABLE IF NOT EXISTS recall_event (
+ id text PRIMARY KEY,
+ authority text NOT NULL, -- 'CPSC', 'NHTSA', 'FDA_drug', 'FDA_device', etc.
+ external_id text NOT NULL, -- e.g. CPSC RecallID
+ published_at timestamptz,
+ title text,
+ hazard text,
+ remedy text,
+ url text,
+ product_keys_jsonb jsonb NOT NULL DEFAULT '{}'::jsonb, -- gtins, models, brand, serial_ranges, udi_di
+ raw_jsonb jsonb, -- full upstream payload for re-parsing
+ ingested_at timestamptz NOT NULL DEFAULT now(),
+ UNIQUE (authority, external_id)
+);
+CREATE INDEX IF NOT EXISTS recall_event_published_idx ON recall_event (published_at DESC);
+CREATE INDEX IF NOT EXISTS recall_event_authority_idx ON recall_event (authority, published_at DESC);
+CREATE INDEX IF NOT EXISTS recall_event_keys_idx ON recall_event USING gin (product_keys_jsonb);
+
+CREATE TABLE IF NOT EXISTS recall_match (
+ id text PRIMARY KEY,
+ recall_id text NOT NULL REFERENCES recall_event(id) ON DELETE CASCADE,
+ asset_table text NOT NULL, -- 'purchase' | 'purchase_item' (later: medical_asset)
+ asset_id text NOT NULL,
+ user_id text NOT NULL REFERENCES user_account(id) ON DELETE CASCADE,
+ confidence numeric(3,2) NOT NULL,
+ status text NOT NULL DEFAULT 'pending_review', -- pending_review | confirmed | dismissed
+ matched_at timestamptz NOT NULL DEFAULT now(),
+ reviewed_at timestamptz,
+ metadata_jsonb jsonb NOT NULL DEFAULT '{}'::jsonb,
+ UNIQUE (recall_id, asset_table, asset_id)
+);
+CREATE INDEX IF NOT EXISTS recall_match_user_idx ON recall_match (user_id, matched_at DESC);
+CREATE INDEX IF NOT EXISTS recall_match_status_idx ON recall_match (status, matched_at DESC);
+
+CREATE TABLE IF NOT EXISTS recall_pull_log (
+ id bigserial PRIMARY KEY,
+ authority text NOT NULL,
+ pulled_at timestamptz NOT NULL DEFAULT now(),
+ fetched integer NOT NULL,
+ inserted integer NOT NULL,
+ updated integer NOT NULL,
+ error text
+);
+
+COMMIT;
diff --git a/lib/cpsc-fetcher.js b/lib/cpsc-fetcher.js
new file mode 100644
index 0000000..464384e
--- /dev/null
+++ b/lib/cpsc-fetcher.js
@@ -0,0 +1,59 @@
+// CPSC Recall API client.
+// Public, unauthenticated. Docs: https://www.cpsc.gov/Recalls/CPSC-Recalls-Application-Programming-Interface-API
+// Endpoint: https://www.saferproducts.gov/RestWebServices/Recall
+// Filters: RecallDateStart, RecallDateEnd, RecallNumber, RecallTitle, RecallURL, etc.
+
+const BASE = 'https://www.saferproducts.gov/RestWebServices/Recall';
+
+async function fetchRecalls({ since = null, until = null, format = 'json', timeoutMs = 30_000 } = {}) {
+ const params = new URLSearchParams({ format });
+ if (since) params.set('RecallDateStart', toCpscDate(since));
+ if (until) params.set('RecallDateEnd', toCpscDate(until));
+ const url = `${BASE}?${params.toString()}`;
+ const ctrl = new AbortController();
+ const t = setTimeout(() => ctrl.abort(), timeoutMs);
+ try {
+ const r = await fetch(url, { signal: ctrl.signal });
+ if (!r.ok) throw new Error(`CPSC ${r.status}: ${url}`);
+ return r.json();
+ } finally {
+ clearTimeout(t);
+ }
+}
+
+function toCpscDate(d) {
+ if (typeof d === 'string') return d;
+ return d.toISOString().slice(0, 10); // YYYY-MM-DD
+}
+
+// Normalize one CPSC recall into our recall_event shape
+function normalize(raw) {
+ // CPSC API field names use Pascal case
+ const products = raw.Products || [];
+ const manufacturers = raw.Manufacturers || [];
+ const hazards = raw.Hazards || [];
+ const remedies = raw.Remedies || [];
+ const images = raw.Images || [];
+
+ const product_keys = {
+ models: products.flatMap(p => [p.Name, p.Model].filter(Boolean)),
+ brand: manufacturers[0]?.Name || null,
+ descriptions: products.map(p => p.Description).filter(Boolean),
+ upcs: products.map(p => p.UPC).filter(Boolean),
+ image_urls: images.map(i => i.URL).filter(Boolean),
+ };
+
+ return {
+ authority: 'CPSC',
+ external_id: String(raw.RecallNumber || raw.RecallID),
+ published_at: raw.RecallDate ? new Date(raw.RecallDate) : null,
+ title: raw.Title || raw.RecallTitle || null,
+ hazard: hazards.map(h => h.Name).filter(Boolean).join('; ') || null,
+ remedy: remedies.map(r => r.Name).filter(Boolean).join('; ') || null,
+ url: raw.URL || (raw.RecallNumber ? `https://www.cpsc.gov/Recalls/${raw.RecallNumber}` : null),
+ product_keys,
+ raw,
+ };
+}
+
+module.exports = { fetchRecalls, normalize };
diff --git a/lib/recall-matcher.js b/lib/recall-matcher.js
new file mode 100644
index 0000000..56fcbe6
--- /dev/null
+++ b/lib/recall-matcher.js
@@ -0,0 +1,114 @@
+// Implements the canonical recall-matching pseudocode from docs/SPEC.md.
+// Compares an asset (purchase or purchase_item) against a recall_event's
+// product_keys; emits a confidence score 0..1 and a status.
+
+const db = require('./db');
+const audit = require('./audit');
+const { id } = require('./ids');
+
+function fuzzy(a, b) {
+ if (!a || !b) return 0;
+ const A = String(a).toLowerCase().replace(/[^a-z0-9]/g, '');
+ const B = String(b).toLowerCase().replace(/[^a-z0-9]/g, '');
+ if (!A || !B) return 0;
+ if (A === B) return 1;
+ if (A.includes(B) || B.includes(A)) return 0.95;
+ // crude similarity: longest common token
+ const tokens = (s) => s.match(/[a-z0-9]{3,}/g) || [];
+ const tA = new Set(tokens(A));
+ const tB = new Set(tokens(B));
+ let hits = 0;
+ for (const t of tA) if (tB.has(t)) hits += 1;
+ if (!tA.size) return 0;
+ return Math.min(0.9, hits / tA.size);
+}
+
+function bestModelMatch(modelStr, modelList) {
+ if (!modelStr || !modelList?.length) return 0;
+ let best = 0;
+ for (const m of modelList) {
+ const s = fuzzy(modelStr, m);
+ if (s > best) best = s;
+ }
+ return best;
+}
+
+/**
+ * @param asset { id, gtin, model, brand, serial, category, purchase_date }
+ * @param recall recall_event row with product_keys_jsonb already parsed
+ */
+function score(asset, recall) {
+ let s = 0;
+ const pk = recall.product_keys_jsonb || recall.product_keys || {};
+
+ if (asset.gtin && (pk.gtins || []).includes(asset.gtin)) s += 0.45;
+ if (asset.gtin && (pk.upcs || []).includes(asset.gtin)) s += 0.45;
+
+ if (asset.model) {
+ const ms = bestModelMatch(asset.model, pk.models);
+ if (ms > 0.92) s += 0.25;
+ else if (ms > 0.6) s += 0.10;
+ }
+
+ // Serial range matching is rare for CPSC consumer recalls; skip unless we have explicit ranges
+ if (asset.serial && Array.isArray(pk.serial_ranges)) {
+ for (const r of pk.serial_ranges) {
+ if (r.start && r.end && String(asset.serial) >= r.start && String(asset.serial) <= r.end) {
+ s += 0.20; break;
+ }
+ }
+ }
+
+ if (asset.brand && pk.brand && fuzzy(asset.brand, pk.brand) > 0.8) s += 0.05;
+
+ if (asset.purchase_date && recall.published_at && new Date(recall.published_at) >= new Date(asset.purchase_date)) {
+ s += 0.05;
+ }
+
+ // Medical-device UDI bonus (deferred — medical_asset isn't implemented yet)
+ if (asset.category === 'medical_device' && asset.udi_di && pk.udi_di === asset.udi_di) s += 0.40;
+
+ s = Math.min(1, Math.round(s * 1000) / 1000);
+ let status;
+ if (s >= 0.85) status = 'positive_match';
+ else if (s >= 0.60) status = 'needs_user_confirmation';
+ else status = 'no_match';
+ return { score: s, status };
+}
+
+/**
+ * Walk all unmatched purchases for a user against a recall.
+ * Inserts recall_match rows for any score >= 0.60.
+ */
+async function matchRecallAgainstUser(recall, userId) {
+ const purchases = await db.query(
+ `SELECT p.id, p.merchant_name AS brand, p.purchase_date, NULL::text AS gtin, NULL::text AS model, NULL::text AS serial, NULL::text AS category
+ FROM purchase p WHERE p.user_id = $1`,
+ [userId]
+ );
+ let inserted = 0;
+ for (const a of purchases.rows) {
+ const r = score(a, recall);
+ if (r.score < 0.60) continue;
+ const matchId = id('purchase'); // re-use ulid prefix
+ try {
+ await db.query(
+ `INSERT INTO recall_match (id, recall_id, asset_table, asset_id, user_id, confidence, status, metadata_jsonb)
+ VALUES ($1, $2, 'purchase', $3, $4, $5, $6, $7)
+ ON CONFLICT (recall_id, asset_table, asset_id) DO NOTHING`,
+ [matchId, recall.id, a.id, userId, r.score, r.status === 'positive_match' ? 'pending_review' : 'pending_review', { score: r.score, status: r.status }]
+ );
+ inserted += 1;
+ await audit.log({
+ actorType: 'system',
+ objectType: 'recall_match',
+ objectId: matchId,
+ eventType: 'recall_match_created',
+ metadata: { recall_id: recall.id, asset_id: a.id, score: r.score, status: r.status },
+ });
+ } catch (_) { /* dup */ }
+ }
+ return inserted;
+}
+
+module.exports = { score, fuzzy, bestModelMatch, matchRecallAgainstUser };
diff --git a/scripts/ingest-recalls.js b/scripts/ingest-recalls.js
new file mode 100644
index 0000000..9320f9e
--- /dev/null
+++ b/scripts/ingest-recalls.js
@@ -0,0 +1,78 @@
+#!/usr/bin/env node
+// Backfill CPSC recalls for the last N days (default 365). Idempotent: ON CONFLICT skips.
+// Usage: node scripts/ingest-recalls.js [days]
+
+require('dotenv').config();
+const db = require('../lib/db');
+const { id } = require('../lib/ids');
+const cpsc = require('../lib/cpsc-fetcher');
+const matcher = require('../lib/recall-matcher');
+
+const DEV_USER_ID = 'user_steve';
+
+async function main() {
+ const days = parseInt(process.argv[2] || '365', 10);
+ const since = new Date(Date.now() - days * 24 * 60 * 60 * 1000);
+ console.log(`[ingest] CPSC recalls since ${since.toISOString().slice(0, 10)} (${days} days)`);
+
+ let raw;
+ try {
+ raw = await cpsc.fetchRecalls({ since });
+ } catch (err) {
+ console.error('[ingest] CPSC fetch failed:', err.message);
+ await db.query(`INSERT INTO recall_pull_log (authority, fetched, inserted, updated, error) VALUES ('CPSC', 0, 0, 0, $1)`, [err.message]);
+ process.exit(1);
+ }
+
+ const list = Array.isArray(raw) ? raw : raw?.results || [];
+ console.log(`[ingest] received ${list.length} recalls from CPSC`);
+
+ let inserted = 0;
+ let updated = 0;
+ let totalMatches = 0;
+
+ for (const r of list) {
+ const norm = cpsc.normalize(r);
+ if (!norm.external_id) continue;
+ const eventId = id('document'); // reuse ulid; we're storing in recall_event
+
+ try {
+ const result = await db.query(
+ `INSERT INTO recall_event (id, authority, external_id, published_at, title, hazard, remedy, url, product_keys_jsonb, raw_jsonb)
+ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
+ ON CONFLICT (authority, external_id) DO UPDATE SET
+ published_at = EXCLUDED.published_at,
+ title = EXCLUDED.title,
+ hazard = EXCLUDED.hazard,
+ remedy = EXCLUDED.remedy,
+ url = EXCLUDED.url,
+ product_keys_jsonb = EXCLUDED.product_keys_jsonb,
+ raw_jsonb = EXCLUDED.raw_jsonb
+ RETURNING id, (xmax = 0) AS was_inserted`,
+ [eventId, norm.authority, norm.external_id, norm.published_at, norm.title, norm.hazard, norm.remedy, norm.url, norm.product_keys, norm.raw]
+ );
+ const row = result.rows[0];
+ if (row.was_inserted) inserted += 1; else updated += 1;
+
+ // Try to match this recall against the user's purchases
+ try {
+ const fullRow = await db.query(`SELECT * FROM recall_event WHERE id = $1`, [row.id]);
+ const matches = await matcher.matchRecallAgainstUser(fullRow.rows[0], DEV_USER_ID);
+ totalMatches += matches;
+ } catch (e) {
+ console.warn('[match]', norm.external_id, e.message);
+ }
+ } catch (err) {
+ console.warn('[insert]', norm.external_id, err.message);
+ }
+ }
+
+ await db.query(
+ `INSERT INTO recall_pull_log (authority, fetched, inserted, updated) VALUES ('CPSC', $1, $2, $3)`,
+ [list.length, inserted, updated]
+ );
+ console.log(`[ingest] done. inserted=${inserted} updated=${updated} matches=${totalMatches}`);
+ await db.pool.end();
+}
+
+main().catch((e) => { console.error(e); process.exit(1); });
← 9e812a4 tick 2: Google Drive sync (receipt-shaped PDFs/images)
·
back to AbramsOS
·
tick 4: PDF parsing for receipt attachments (tier-3) d95d5e5 →