'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, shop_id, shop_name, school, status FROM ops_shop 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 };