hyreal-gen-pic/server/stores/ops-sqlite.js

'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 });
    const { DatabaseSync } = require('node:sqlite');
    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);
        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 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, 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 };