'use strict'; /** * ops 存储 · SQLite 驱动(node:sqlite,零依赖) * ---------------------------------------------------------------------------- * 只负责「表和读写」,不含业务判断。接口与 PG 驱动保持一致,路由层不感知方言。 * 需要 Node >= 22.5(node:sqlite 内置模块)。 */ const fs = require('fs'); const path = require('path'); const SCHEMA_VERSION = '1'; const DDL = [ `CREATE TABLE IF NOT EXISTS ops_meta ( key TEXT PRIMARY KEY, value TEXT NOT NULL )`, `CREATE TABLE IF NOT EXISTS ops_upload ( id TEXT PRIMARY KEY, original_name TEXT NOT NULL, sha256 TEXT NOT NULL, size_bytes INTEGER NOT NULL, uploaded_at TEXT NOT NULL, year INTEGER NOT NULL, parser_kind TEXT NOT NULL, parser_version INTEGER NOT NULL, report TEXT NOT NULL DEFAULT '{}', content BLOB NOT NULL )`, `CREATE INDEX IF NOT EXISTS ops_upload_sha256_idx ON ops_upload (sha256, uploaded_at DESC)`, `CREATE TABLE IF NOT EXISTS ops_shop ( key TEXT PRIMARY KEY, shop_id TEXT, shop_name TEXT NOT NULL, school TEXT NOT NULL, status TEXT, last_upload_id TEXT REFERENCES ops_upload(id) ON DELETE SET NULL, first_seen TEXT NOT NULL, last_seen TEXT NOT NULL )`, `CREATE TABLE IF NOT EXISTS ops_row ( date TEXT NOT NULL, shop_key TEXT NOT NULL, upload_id TEXT NOT NULL REFERENCES ops_upload(id) ON DELETE CASCADE, shop_id TEXT, shop_name TEXT NOT NULL, school TEXT NOT NULL, status TEXT, order_cnt INTEGER, revenue REAL, exposure INTEGER, click_rate REAL, raw TEXT NOT NULL DEFAULT '{}', quality TEXT NOT NULL DEFAULT '[]', updated_at TEXT NOT NULL, PRIMARY KEY (date, shop_key) )`, `CREATE INDEX IF NOT EXISTS ops_row_school_date_idx ON ops_row (school, date)`, ]; function rowToApi(row) { if (!row) return null; return { date: row.date, shopKey: row.shop_key, shopId: row.shop_id, shopName: row.shop_name, school: row.school, status: row.status, order_cnt: row.order_cnt, revenue: row.revenue, exposure: row.exposure, click_rate: row.click_rate, raw: safeJson(row.raw, {}), quality: safeJson(row.quality, []), uploadId: row.upload_id, updatedAt: row.updated_at, }; } function safeJson(text, fallback) { try { const value = JSON.parse(text); return value == null ? fallback : value; } catch { return fallback; } } function createSqliteStore({ file }) { const dbFile = path.resolve(String(file || 'data/ops.db')); let db = null; let initPromise = null; function open() { if (db) return db; fs.mkdirSync(path.dirname(dbFile), { recursive: true }); let DatabaseSync; try { ({ DatabaseSync } = require('node:sqlite')); } catch (e) { // Node < 22.5 没有内置 sqlite:单独给一个可读原因,前端据此提示升级 Node const err = new Error(`当前 Node 版本不支持内置 SQLite(需要 22.5 及以上,当前 ${process.version})`); err.reason = 'sqlite-unavailable'; err.cause = e; throw err; } db = new DatabaseSync(dbFile); db.exec('PRAGMA journal_mode = WAL'); db.exec('PRAGMA busy_timeout = 5000'); db.exec('PRAGMA foreign_keys = ON'); return db; } async function init() { if (!initPromise) { initPromise = (async () => { const handle = open(); for (const sql of DDL) handle.exec(sql); handle.prepare('INSERT INTO ops_meta (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value') .run('schema_version', SCHEMA_VERSION); // 建表后 checkpoint:空库阶段 data/ops.db 也是自包含的 try { handle.exec('PRAGMA wal_checkpoint(TRUNCATE)'); } catch { /* ignore */ } return true; })().catch((e) => { initPromise = null; // 允许下次重试 throw e; }); } return initPromise; } async function health() { try { open().prepare('SELECT 1 AS ok').get(); return { ok: true, kind: 'sqlite', detail: dbFile }; } catch (e) { return { ok: false, kind: 'sqlite', detail: e.message || String(e) }; } } function insertUpload(upload) { open().prepare(`INSERT INTO ops_upload (id, original_name, sha256, size_bytes, uploaded_at, year, parser_kind, parser_version, report, content) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`).run( upload.id, upload.originalName, upload.sha256, upload.sizeBytes, upload.uploadedAt, upload.year, upload.parserKind, upload.parserVersion, JSON.stringify(upload.report || {}), upload.content, ); } function upsertRows(rows) { const stmt = open().prepare(`INSERT INTO ops_row (date, shop_key, upload_id, shop_id, shop_name, school, status, order_cnt, revenue, exposure, click_rate, raw, quality, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(date, shop_key) DO UPDATE SET upload_id = excluded.upload_id, shop_id = excluded.shop_id, shop_name = excluded.shop_name, school = excluded.school, status = excluded.status, order_cnt = excluded.order_cnt, revenue = excluded.revenue, exposure = excluded.exposure, click_rate = excluded.click_rate, raw = excluded.raw, quality = excluded.quality, updated_at = excluded.updated_at`); const now = new Date().toISOString(); for (const r of rows) { stmt.run( r.date, r.shopKey, r.uploadId, r.shopId || null, r.shopName, r.school, r.status || null, r.order_cnt, r.revenue, r.exposure, r.click_rate, JSON.stringify(r.raw || {}), JSON.stringify(r.quality || []), now, ); } } function upsertShops(shops) { const stmt = open().prepare(`INSERT INTO ops_shop (key, shop_id, shop_name, school, status, last_upload_id, first_seen, last_seen) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(key) DO UPDATE SET shop_id = excluded.shop_id, shop_name = excluded.shop_name, school = excluded.school, status = excluded.status, last_upload_id = excluded.last_upload_id, last_seen = excluded.last_seen`); const now = new Date().toISOString(); for (const s of shops) { stmt.run(s.key, s.shopId || null, s.shopName, s.school, s.status || null, s.lastUploadId || null, now, now); } } function deleteRowsByUpload(uploadId) { open().prepare('DELETE FROM ops_row WHERE upload_id = ?').run(uploadId); } function deleteRowsBySha(sha256) { open().prepare('DELETE FROM ops_row WHERE upload_id IN (SELECT id FROM ops_upload WHERE sha256 = ?)').run(sha256); } function pruneOrphanShops() { open().prepare('DELETE FROM ops_shop WHERE key NOT IN (SELECT DISTINCT shop_key FROM ops_row)').run(); } function updateReport(id, report) { open().prepare('UPDATE ops_upload SET report = ? WHERE id = ?').run(JSON.stringify(report || {}), id); } async function withTransaction(fn) { const handle = open(); handle.exec('BEGIN IMMEDIATE'); try { const result = await fn({ insertUpload, upsertRows, upsertShops, deleteRowsByUpload, deleteRowsBySha, pruneOrphanShops, updateReport }); handle.exec('COMMIT'); // 写事务结束后做一次 checkpoint:让 data/ops.db 自包含,备份时不必关心 -wal try { handle.exec('PRAGMA wal_checkpoint(TRUNCATE)'); } catch { /* 有读事务时跳过 */ } return result; } catch (e) { try { handle.exec('ROLLBACK'); } catch { /* 事务可能已结束 */ } throw e; } } async function findUploadBySha(sha256) { const row = open().prepare('SELECT id FROM ops_upload WHERE sha256 = ? ORDER BY uploaded_at DESC LIMIT 1').get(sha256); return row ? row.id : null; } function uploadRowToApi(row) { if (!row) return null; const report = safeJson(row.report, {}); return { id: row.id, originalName: row.original_name, sha256: row.sha256, sizeBytes: row.size_bytes, uploadedAt: row.uploaded_at, year: row.year, parserKind: row.parser_kind, parserVersion: row.parser_version, report, rows: report.rows || 0, shops: report.shops || 0, warningCount: report.warningCount || 0, }; } async function listUploads() { return open().prepare(`SELECT id, original_name, sha256, size_bytes, uploaded_at, year, parser_kind, parser_version, report FROM ops_upload ORDER BY uploaded_at DESC`).all().map(uploadRowToApi); } async function getUpload(id) { const row = open().prepare(`SELECT id, original_name, sha256, size_bytes, uploaded_at, year, parser_kind, parser_version, report FROM ops_upload WHERE id = ?`).get(id); return uploadRowToApi(row); } async function getContent(id) { const row = open().prepare('SELECT content, original_name FROM ops_upload WHERE id = ?').get(id); if (!row) return null; return { content: Buffer.from(row.content), originalName: row.original_name }; } async function queryRows({ from, to, schools, shops } = {}) { const where = []; const params = []; if (from) { where.push('date >= ?'); params.push(String(from)); } if (to) { where.push('date <= ?'); params.push(String(to)); } if (Array.isArray(schools) && schools.length) { where.push(`school IN (${schools.map(() => '?').join(',')})`); params.push(...schools.map(String)); } if (Array.isArray(shops) && shops.length) { where.push(`shop_key IN (${shops.map(() => '?').join(',')})`); params.push(...shops.map(String)); } const sql = `SELECT * FROM ops_row ${where.length ? 'WHERE ' + where.join(' AND ') : ''} ORDER BY school ASC, shop_name ASC, date ASC`; return open().prepare(sql).all(...params).map(rowToApi); } async function summary() { const handle = open(); const uploadCount = handle.prepare('SELECT COUNT(*) AS c FROM ops_upload').get().c; const rowCount = handle.prepare('SELECT COUNT(*) AS c FROM ops_row').get().c; const range = handle.prepare('SELECT MIN(date) AS f, MAX(date) AS t FROM ops_row').get(); const schools = handle.prepare('SELECT school, COUNT(*) AS c FROM ops_shop GROUP BY school ORDER BY school ASC') .all().map((r) => ({ name: r.school, shopCount: r.c })); const shops = handle.prepare('SELECT key, shop_id, shop_name, school, status FROM ops_shop ORDER BY school ASC, shop_name ASC') .all().map((r) => ({ key: r.key, shopId: r.shop_id, shopName: r.shop_name, school: r.school, status: r.status })); const last = handle.prepare(`SELECT id, original_name, uploaded_at, report FROM ops_upload ORDER BY uploaded_at DESC LIMIT 1`).get(); return { schools, shops, uploadCount, rowCount, coverage: range && range.f ? { from: range.f, to: range.t } : null, lastUpload: last ? { id: last.id, at: last.uploaded_at, originalName: last.original_name, rows: (safeJson(last.report, {}) || {}).rows || 0, warningCount: (safeJson(last.report, {}) || {}).warningCount || 0, } : null, }; } async function close() { if (db) { try { db.exec('PRAGMA wal_checkpoint(TRUNCATE)'); } catch { /* ignore */ } try { db.close(); } catch { /* ignore */ } db = null; initPromise = null; } } return { kind: 'sqlite', file: dbFile, init, health, withTransaction, insertUpload, upsertRows, upsertShops, deleteRowsByUpload, updateReport, findUploadBySha, listUploads, getUpload, getContent, queryRows, summary, close, }; } module.exports = { createSqliteStore };