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

'use strict';

/**
 * ops 存储 · PostgreSQL 驱动
 * ----------------------------------------------------------------------------
 * 与 SQLite 驱动同接口。`pg` 为可选依赖（本期不安装）：未安装时 health() 返回明确原因，
 * 只有「店铺运营」功能不可用，服务与其它功能不受影响。
 *
 * 上线切换步骤见设计文档 §5.6：改 config.storage.driver = "postgres" + 填连接串 + npm i pg。
 */

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 timestamptz NOT NULL,
     year integer NOT NULL,
     parser_kind text NOT NULL,
     parser_version integer NOT NULL,
     report jsonb NOT NULL DEFAULT '{}'::jsonb,
     content bytea 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 timestamptz NOT NULL,
     last_seen timestamptz NOT NULL
   )`,
  `CREATE TABLE IF NOT EXISTS ops_row (
     date char(8) 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 numeric(14,2),
     exposure integer,
     click_rate numeric(8,6),
     raw jsonb NOT NULL DEFAULT '{}'::jsonb,
     quality text[] NOT NULL DEFAULT '{}',
     updated_at timestamptz NOT NULL,
     PRIMARY KEY (date, shop_key)
   )`,
  `CREATE INDEX IF NOT EXISTS ops_row_school_date_idx ON ops_row (school, date)`,
  `CREATE TABLE IF NOT EXISTS ops_platform_row (
     date char(8) NOT NULL,
     shop_id text NOT NULL,
     upload_id text NOT NULL REFERENCES ops_upload(id) ON DELETE CASCADE,
     school text NOT NULL DEFAULT '',
     shop_name text NOT NULL DEFAULT '',
     data jsonb NOT NULL DEFAULT '{}'::jsonb,
     updated_at timestamptz NOT NULL,
     PRIMARY KEY (date, shop_id)
   )`,
  `CREATE INDEX IF NOT EXISTS ops_platform_row_date_idx ON ops_platform_row (date)`,
];

function rowToApi(row) {
  if (!row) return null;
  return {
    date: String(row.date || '').trim(),
    shopKey: row.shop_key,
    shopId: row.shop_id,
    shopName: row.shop_name,
    school: row.school,
    status: row.status,
    order_cnt: row.order_cnt == null ? null : Number(row.order_cnt),
    revenue: row.revenue == null ? null : Number(row.revenue),
    exposure: row.exposure == null ? null : Number(row.exposure),
    click_rate: row.click_rate == null ? null : Number(row.click_rate),
    raw: row.raw || {},
    quality: row.quality || [],
    uploadId: row.upload_id,
    updatedAt: row.updated_at instanceof Date ? row.updated_at.toISOString() : row.updated_at,
  };
}

function createPgStore({ connectionString, ssl = true, max = 5, schema = 'public', connectTimeoutMs = 10000 }) {
  let pool = null;
  let initPromise = null;

  function getPool() {
    if (pool) return pool;
    let pg;
    try {
      pg = require('pg');
    } catch {
      const err = new Error('未安装 pg（npm i pg）');
      err.reason = 'pg-not-installed';
      throw err;
    }
    pool = new pg.Pool({
      connectionString,
      max: Number(max) || 5,
      ssl: ssl ? { rejectUnauthorized: false } : false,
      connectionTimeoutMillis: Number(connectTimeoutMs) || 10000,
      options: `-c search_path=${schema || 'public'}`,
    });
    return pool;
  }

  async function init() {
    if (!initPromise) {
      initPromise = (async () => {
        const client = await getPool().connect();
        try {
          for (const sql of DDL) await client.query(sql);
          await client.query(
            'INSERT INTO ops_meta (key, value) VALUES ($1, $2) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value',
            ['schema_version', SCHEMA_VERSION],
          );
        } finally {
          client.release();
        }
        return true;
      })().catch((e) => {
        initPromise = null;
        throw e;
      });
    }
    return initPromise;
  }

  async function health() {
    try {
      await getPool().query('SELECT 1 AS ok');
      return { ok: true, kind: 'postgres', detail: (schema || 'public') };
    } catch (e) {
      return { ok: false, kind: 'postgres', detail: e.message || String(e) };
    }
  }

  async function withTransaction(fn) {
    const client = await getPool().connect();
    try {
      await client.query('BEGIN');
      const result = await fn(makeTx(client));
      await client.query('COMMIT');
      return result;
    } catch (e) {
      try { await client.query('ROLLBACK'); } catch { /* 已结束 */ }
      throw e;
    } finally {
      client.release();
    }
  }

  function makeTx(client) {
    return {
      async insertUpload(upload) {
        await client.query(
          `INSERT INTO ops_upload (id, original_name, sha256, size_bytes, uploaded_at, year,
             parser_kind, parser_version, report, content)
           VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)`,
          [upload.id, upload.originalName, upload.sha256, upload.sizeBytes, upload.uploadedAt,
            upload.year, upload.parserKind, upload.parserVersion, JSON.stringify(upload.report || {}), upload.content],
        );
      },
      async upsertRows(rows) {
        const now = new Date().toISOString();
        for (const r of rows) {
          await client.query(
            `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 ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)
             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`,
            [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 || {}), r.quality || [], now],
          );
        }
      },
      async upsertShops(shops) {
        const now = new Date().toISOString();
        for (const s of shops) {
          await client.query(
            `INSERT INTO ops_shop (key, shop_id, shop_name, school, status, last_upload_id, first_seen, last_seen)
             VALUES ($1,$2,$3,$4,$5,$6,$7,$8)
             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`,
            [s.key, s.shopId || null, s.shopName, s.school, s.status || null, s.lastUploadId || null, now, now],
          );
        }
      },
      async deleteRowsByUpload(uploadId) {
        await client.query('DELETE FROM ops_row WHERE upload_id = $1', [uploadId]);
      },
      async upsertPlatformRows(rows) {
        const now = new Date().toISOString();
        for (const r of rows) {
          await client.query(
            `INSERT INTO ops_platform_row (date, shop_id, upload_id, school, shop_name, data, updated_at)
             VALUES ($1,$2,$3,$4,$5,$6,$7)
             ON CONFLICT (date, shop_id) DO UPDATE SET
               upload_id = EXCLUDED.upload_id, school = EXCLUDED.school,
               shop_name = EXCLUDED.shop_name, data = EXCLUDED.data, updated_at = EXCLUDED.updated_at`,
            [r.date, r.shopId, r.uploadId, r.school || '', r.shopName || '', JSON.stringify(r.data || {}), now],
          );
        }
      },
      async deletePlatformRowsByUpload(uploadId) {
        await client.query('DELETE FROM ops_platform_row WHERE upload_id = $1', [uploadId]);
      },
      async deletePlatformRowsBySha(sha256) {
        await client.query('DELETE FROM ops_platform_row WHERE upload_id IN (SELECT id FROM ops_upload WHERE sha256 = $1)', [sha256]);
      },
      async deleteRowsBySha(sha256) {
        await client.query('DELETE FROM ops_row WHERE upload_id IN (SELECT id FROM ops_upload WHERE sha256 = $1)', [sha256]);
      },
      async pruneOrphanShops() {
        await client.query('DELETE FROM ops_shop s WHERE NOT EXISTS (SELECT 1 FROM ops_row r WHERE r.shop_key = s.key)');
      },
      async updateReport(id, report) {
        await client.query('UPDATE ops_upload SET report = $1 WHERE id = $2', [JSON.stringify(report || {}), id]);
      },
    };
  }

  async function findUploadBySha(sha256) {
    const r = await getPool().query('SELECT id FROM ops_upload WHERE sha256 = $1 ORDER BY uploaded_at DESC LIMIT 1', [sha256]);
    return r.rows.length ? r.rows[0].id : null;
  }

  function uploadRowToApi(row) {
    if (!row) return null;
    const report = row.report || {};
    return {
      id: row.id,
      originalName: row.original_name,
      sha256: row.sha256,
      sizeBytes: row.size_bytes,
      uploadedAt: row.uploaded_at instanceof Date ? row.uploaded_at.toISOString() : 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() {
    const r = await getPool().query(`SELECT id, original_name, sha256, size_bytes, uploaded_at, year,
      parser_kind, parser_version, report FROM ops_upload ORDER BY uploaded_at DESC`);
    return r.rows.map(uploadRowToApi);
  }

  async function getUpload(id) {
    const r = await getPool().query(`SELECT id, original_name, sha256, size_bytes, uploaded_at, year,
      parser_kind, parser_version, report FROM ops_upload WHERE id = $1`, [id]);
    return uploadRowToApi(r.rows[0]);
  }

  async function getContent(id) {
    const r = await getPool().query('SELECT content, original_name FROM ops_upload WHERE id = $1', [id]);
    if (!r.rows.length) return null;
    return { content: r.rows[0].content, originalName: r.rows[0].original_name };
  }

  async function queryRows({ from, to, schools, shops } = {}) {
    const where = [];
    const params = [];
    if (from) { params.push(String(from)); where.push(`date >= $${params.length}`); }
    if (to) { params.push(String(to)); where.push(`date <= $${params.length}`); }
    if (Array.isArray(schools) && schools.length) { params.push(schools.map(String)); where.push(`school = ANY($${params.length})`); }
    if (Array.isArray(shops) && shops.length) { params.push(shops.map(String)); where.push(`shop_key = ANY($${params.length})`); }
    const sql = `SELECT * FROM ops_row ${where.length ? 'WHERE ' + where.join(' AND ') : ''}
      ORDER BY school ASC, shop_name ASC, date ASC`;
    const r = await getPool().query(sql, params);
    return r.rows.map(rowToApi);
  }

  async function queryPlatformRows({ from, to } = {}) {
    const where = [];
    const params = [];
    if (from) { params.push(String(from)); where.push(`date >= ${params.length}`); }
    if (to) { params.push(String(to)); where.push(`date <= ${params.length}`); }
    const sql = `SELECT * FROM ops_platform_row ${where.length ? 'WHERE ' + where.join(' AND ') : ''} ORDER BY date ASC, shop_id ASC`;
    const r = await getPool().query(sql, params);
    return r.rows.map((row) => ({
      date: String(row.date).trim(),
      shopId: row.shop_id,
      uploadId: row.upload_id,
      school: row.school,
      shopName: row.shop_name,
      data: row.data || {},
      updatedAt: row.updated_at instanceof Date ? row.updated_at.toISOString() : row.updated_at,
    }));
  }

  async function summary() {
    const poolRef = getPool();
    const [uploadCount, rowCount, platformRowCount, range, matrix, schools, shops, last] = await Promise.all([
      poolRef.query('SELECT COUNT(*)::int AS c FROM ops_upload'),
      poolRef.query('SELECT COUNT(*)::int AS c FROM ops_row'),
      poolRef.query('SELECT COUNT(*)::int AS c FROM ops_platform_row'),
      poolRef.query(`SELECT MIN(d) AS f, MAX(d) AS t FROM (
        SELECT date AS d FROM ops_row UNION SELECT date AS d FROM ops_platform_row)`),
      poolRef.query(`SELECT COUNT(*)::int AS c FROM (
        SELECT date, shop_key FROM ops_row UNION SELECT date, shop_id FROM ops_platform_row)`),
      poolRef.query(`SELECT school, COUNT(DISTINCT shop)::int AS c FROM (
        SELECT school, shop_id AS shop FROM ops_platform_row WHERE school <> ''
        UNION SELECT school, shop_key AS shop FROM ops_row WHERE school <> ''
        UNION SELECT school, key AS shop FROM ops_shop WHERE school <> ''
      ) AS t GROUP BY school ORDER BY school ASC`),
      poolRef.query(`SELECT key, MAX(shop_id) AS shop_id, MAX(shop_name) AS shop_name,
        MAX(school) AS school, MAX(status) AS status FROM (
        SELECT key, shop_id, shop_name, school, status FROM ops_shop
        UNION ALL
        SELECT shop_id AS key, shop_id, shop_name, school, NULL AS status FROM ops_platform_row
      ) AS t GROUP BY key ORDER BY school ASC, shop_name ASC`),
      poolRef.query('SELECT id, original_name, uploaded_at, report FROM ops_upload ORDER BY uploaded_at DESC LIMIT 1'),
    ]);
    return {
      schools: schools.rows.map((r) => ({ name: r.school, shopCount: r.c })),
      shops: shops.rows.map((r) => ({ key: r.key, shopId: r.shop_id, shopName: r.shop_name, school: r.school, status: r.status })),
      uploadCount: uploadCount.rows[0].c,
      rowCount: rowCount.rows[0].c,
      platformRowCount: platformRowCount.rows[0].c,
      matrixRowCount: matrix.rows[0].c,
      coverage: range.rows[0] && range.rows[0].f ? { from: String(range.rows[0].f).replace(/-/g, ''), to: String(range.rows[0].t).replace(/-/g, '') } : null,
      lastUpload: last.rows.length ? {
        id: last.rows[0].id,
        at: last.rows[0].uploaded_at instanceof Date ? last.rows[0].uploaded_at.toISOString() : last.rows[0].uploaded_at,
        originalName: last.rows[0].original_name,
        rows: (last.rows[0].report || {}).rows || 0,
        warningCount: (last.rows[0].report || {}).warningCount || 0,
      } : null,
    };
  }

  async function close() {
    if (pool) {
      try { await pool.end(); } catch { /* ignore */ }
      pool = null;
      initPromise = null;
    }
  }

  return {
    kind: 'postgres',
    init,
    health,
    withTransaction,
    findUploadBySha,
    queryPlatformRows,
    listUploads,
    getUpload,
    getContent,
    queryRows,
    summary,
    close,
  };
}

module.exports = { createPgStore };