优化数据库

2026-07-10 08:47:08 +08:00
parent 3318ed18ee
commit ce01c2aa34
9 changed files with 790 additions and 102 deletions
@@ -1,42 +1,122 @@
-import fs from 'node:fs'
 import path from 'node:path'
 import process from 'node:process'
 import { fileURLToPath } from 'node:url'

-import { closeDb, query, withTransaction } from './client.js'
+import { runner, type RunnerOption } from 'node-pg-migrate'
+import type { ClientConfig } from 'pg'
+
+import { runtimeConfig } from '../config/runtime.js'
+import { closeDb, query } from './client.js'

 const CURRENT_DIR = path.dirname(fileURLToPath(import.meta.url))
 const MIGRATIONS_DIR = path.join(CURRENT_DIR, 'migrations')
+const MIGRATIONS_TABLE = 'schema_migrations'
+const SHORT_NUMERIC_PREFIX_MESSAGE = /^Can't determine timestamp for \d+$/

 export async function runDatabaseMigrations() {
-  const files = fs.readdirSync(MIGRATIONS_DIR)
-    .filter((name) => name.endsWith('.sql'))
-    .sort()
+  await ensureNodePgMigrateMetadataCompatibility()

+  await runner({
+    databaseUrl: createMigrationDatabaseConfig(),
+    dir: MIGRATIONS_DIR,
+    direction: 'up',
+    migrationsTable: MIGRATIONS_TABLE,
+    migrationsSchema: 'public',
+    schema: 'public',
+    singleTransaction: true,
+    checkOrder: true,
+    logger: {
+      debug: () => undefined,
+      info: (message) => {
+        console.info(`[db:migrate] ${message}`)
+      },
+      warn: (message) => {
+        console.warn(`[db:migrate] ${message}`)
+      },
+      error: (message) => {
+        if (SHORT_NUMERIC_PREFIX_MESSAGE.test(message)) {
+          return
+        }
+        console.error(`[db:migrate] ${message}`)
+      },
+    },
+  } satisfies RunnerOption)
+
+  await syncLegacyMigrationColumns()
+}
+
+async function ensureNodePgMigrateMetadataCompatibility() {
  await query(`
-    CREATE TABLE IF NOT EXISTS schema_migrations (
+    CREATE TABLE IF NOT EXISTS ${MIGRATIONS_TABLE} (
      id BIGSERIAL PRIMARY KEY,
-      filename TEXT NOT NULL UNIQUE,
+      name VARCHAR(255),
+      run_on TIMESTAMPTZ,
+      filename TEXT,
      applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
    )
  `)

-  const appliedRows = await query('SELECT filename FROM schema_migrations')
-  const applied = new Set(appliedRows.rows.map((row) => String(row.filename)))
+  await query(`
+    ALTER TABLE ${MIGRATIONS_TABLE}
+      ADD COLUMN IF NOT EXISTS name VARCHAR(255),
+      ADD COLUMN IF NOT EXISTS run_on TIMESTAMPTZ,
+      ADD COLUMN IF NOT EXISTS filename TEXT,
+      ADD COLUMN IF NOT EXISTS applied_at TIMESTAMPTZ
+  `)

-  for (const filename of files) {
-    if (applied.has(filename)) {
-      continue
-    }
+  await query(`
+    ALTER TABLE ${MIGRATIONS_TABLE}
+      ALTER COLUMN filename DROP NOT NULL,
+      ALTER COLUMN applied_at SET DEFAULT NOW()
+  `)

-    const sql = fs.readFileSync(path.join(MIGRATIONS_DIR, filename), 'utf8')
-    await withTransaction(async (client) => {
-      await client.query(sql)
-      await client.query(
-        'INSERT INTO schema_migrations (filename, applied_at) VALUES ($1, NOW())',
-        [filename],
-      )
-    })
+  await query(`
+    UPDATE ${MIGRATIONS_TABLE}
+    SET
+      name = COALESCE(
+        NULLIF(name, ''),
+        NULLIF(REGEXP_REPLACE(COALESCE(filename, ''), '\\.[^.]+$', ''), ''),
+        CONCAT('legacy_', id)
+      ),
+      run_on = COALESCE(run_on, applied_at, NOW())
+    WHERE name IS NULL
+      OR name = ''
+      OR run_on IS NULL
+  `)
+
+  await query(`
+    ALTER TABLE ${MIGRATIONS_TABLE}
+      ALTER COLUMN name SET NOT NULL,
+      ALTER COLUMN run_on SET NOT NULL
+  `)
+
+  await query(`
+    CREATE UNIQUE INDEX IF NOT EXISTS idx_schema_migrations_name
+      ON ${MIGRATIONS_TABLE}(name)
+  `)
+}
+
+async function syncLegacyMigrationColumns() {
+  await query(`
+    UPDATE ${MIGRATIONS_TABLE}
+    SET
+      filename = COALESCE(NULLIF(filename, ''), CONCAT(name, '.sql')),
+      applied_at = COALESCE(applied_at, run_on, NOW())
+    WHERE filename IS NULL
+      OR filename = ''
+      OR applied_at IS NULL
+  `)
+}
+
+function createMigrationDatabaseConfig(): ClientConfig {
+  const connectionString = String(runtimeConfig.database?.url || '').trim()
+  if (!connectionString) {
+    throw new Error('DATABASE_URL 未配置，无法执行数据库迁移')
+  }
+
+  return {
+    connectionString,
+    ssl: runtimeConfig.database?.ssl ? { rejectUnauthorized: false } : false,
  }
 }

@@ -50,6 +130,6 @@ if (process.argv[1] && path.resolve(process.argv[1]) === currentFilePath) {
    .catch(async (error) => {
      console.error(error)
      await closeDb()
-      process.exit(1)
+      process.exitCode = 1
    })
 }
@@ -1,5 +1,9 @@
+-- 初始化完整业务库结构。
+-- 本项目当前不维护历史测试库的逐步升级链路；新环境只需要执行这一份 init 迁移。
+
 CREATE EXTENSION IF NOT EXISTS pgcrypto;

+-- 后台账号与审计日志。
 CREATE TABLE IF NOT EXISTS admin_users (
  id BIGSERIAL PRIMARY KEY,
  username TEXT NOT NULL UNIQUE,
@@ -23,9 +27,16 @@ CREATE TABLE IF NOT EXISTS admin_audit_logs (
  created_at TIMESTAMPTZ NOT NULL
 );

-CREATE INDEX IF NOT EXISTS idx_admin_audit_logs_created_at ON admin_audit_logs(created_at DESC);
-CREATE INDEX IF NOT EXISTS idx_admin_audit_logs_target ON admin_audit_logs(target_type, target_id);
+CREATE INDEX IF NOT EXISTS idx_admin_audit_logs_created_at
+  ON admin_audit_logs(created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_admin_audit_logs_target
+  ON admin_audit_logs(target_type, target_id);

+COMMENT ON TABLE admin_users IS '后台管理用户';
+COMMENT ON TABLE admin_audit_logs IS '后台操作审计日志';
+COMMENT ON COLUMN admin_audit_logs.payload_json IS '审计上下文快照，保留请求关键字段和变更内容';
+
+-- 履约档案与履约资源要求。
 CREATE TABLE IF NOT EXISTS fulfillment_profiles (
  id BIGSERIAL PRIMARY KEY,
  profile_key TEXT NOT NULL UNIQUE,
@@ -52,6 +63,12 @@ CREATE TABLE IF NOT EXISTS fulfillment_profile_requirements (
  UNIQUE(profile_id, role_key)
 );

+COMMENT ON TABLE fulfillment_profiles IS '履约档案，定义不同履约执行器的基础能力';
+COMMENT ON COLUMN fulfillment_profiles.config_json IS '履约执行器配置快照，适合存放低频变化的扩展配置';
+COMMENT ON TABLE fulfillment_profile_requirements IS '履约档案需要的账号、角色或凭证要求';
+COMMENT ON COLUMN fulfillment_profile_requirements.config_json IS '履约资源要求的扩展配置';
+
+-- 订单主数据与订单项。
 CREATE TABLE IF NOT EXISTS orders (
  id BIGSERIAL PRIMARY KEY,
  provider TEXT NOT NULL,
@@ -90,9 +107,18 @@ CREATE TABLE IF NOT EXISTS order_items (
  updated_at TIMESTAMPTZ NOT NULL
 );

-CREATE INDEX IF NOT EXISTS idx_order_items_order_id ON order_items(order_id);
-CREATE INDEX IF NOT EXISTS idx_order_items_sku_code ON order_items(sku_code);
+CREATE INDEX IF NOT EXISTS idx_order_items_order_id
+  ON order_items(order_id);
+CREATE INDEX IF NOT EXISTS idx_order_items_sku_code
+  ON order_items(sku_code);

+COMMENT ON TABLE orders IS '外部平台订单主表';
+COMMENT ON COLUMN orders.raw_payload_json IS '外部平台原始订单报文，便于追溯和兼容字段变化';
+COMMENT ON TABLE order_items IS '订单商品明细';
+COMMENT ON COLUMN order_items.spec_json IS '商品规格扩展字段';
+COMMENT ON COLUMN order_items.item_snapshot_json IS '订单项履约规划时的商品、路由和平台配置快照';
+
+-- 履约任务、领取令牌和任务事件。
 CREATE TABLE IF NOT EXISTS fulfillment_tasks (
  id BIGSERIAL PRIMARY KEY,
  order_id BIGINT NOT NULL REFERENCES orders(id) ON DELETE CASCADE,
@@ -126,9 +152,12 @@ CREATE TABLE IF NOT EXISTS fulfillment_tasks (
  UNIQUE(order_item_id, unit_index)
 );

-CREATE INDEX IF NOT EXISTS idx_fulfillment_tasks_status ON fulfillment_tasks(task_status, delivery_status);
-CREATE INDEX IF NOT EXISTS idx_fulfillment_tasks_order ON fulfillment_tasks(order_id, order_item_id);
-CREATE INDEX IF NOT EXISTS idx_fulfillment_tasks_claim_token ON fulfillment_tasks(claim_token);
+CREATE INDEX IF NOT EXISTS idx_fulfillment_tasks_status
+  ON fulfillment_tasks(task_status, delivery_status);
+CREATE INDEX IF NOT EXISTS idx_fulfillment_tasks_order
+  ON fulfillment_tasks(order_id, order_item_id);
+CREATE INDEX IF NOT EXISTS idx_fulfillment_tasks_claim_token
+  ON fulfillment_tasks(claim_token);
 CREATE INDEX IF NOT EXISTS idx_fulfillment_tasks_platform_order_lookup
  ON fulfillment_tasks(provider, platform, platform_order_id, id DESC);
 CREATE INDEX IF NOT EXISTS idx_fulfillment_tasks_created_at_desc
@@ -147,7 +176,8 @@ CREATE TABLE IF NOT EXISTS claim_tokens (
  updated_at TIMESTAMPTZ NOT NULL
 );

-CREATE INDEX IF NOT EXISTS idx_claim_tokens_status ON claim_tokens(status, expired_at);
+CREATE INDEX IF NOT EXISTS idx_claim_tokens_status
+  ON claim_tokens(status, expired_at);

 CREATE TABLE IF NOT EXISTS task_events (
  id BIGSERIAL PRIMARY KEY,
@@ -157,8 +187,16 @@ CREATE TABLE IF NOT EXISTS task_events (
  created_at TIMESTAMPTZ NOT NULL
 );

-CREATE INDEX IF NOT EXISTS idx_task_events_task_created_at ON task_events(task_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_task_events_task_created_at
+  ON task_events(task_id, created_at DESC);

+COMMENT ON TABLE fulfillment_tasks IS '履约任务，一件商品按数量拆成独立履约单元';
+COMMENT ON COLUMN fulfillment_tasks.context_json IS '任务执行上下文，保存履约过程中的平台返回、链接、角色和中间状态';
+COMMENT ON TABLE claim_tokens IS '用户领取链接令牌';
+COMMENT ON TABLE task_events IS '任务事件流水';
+COMMENT ON COLUMN task_events.payload_json IS '事件 payload 快照';
+
+-- 运行时上下文和快手云履约索引态。
 CREATE TABLE IF NOT EXISTS task_runtime_contexts (
  id BIGSERIAL PRIMARY KEY,
  task_id BIGINT NOT NULL UNIQUE REFERENCES fulfillment_tasks(id) ON DELETE CASCADE,
@@ -200,3 +238,79 @@ CREATE INDEX IF NOT EXISTS idx_kuaishou_cloud_task_states_consume_status
  ON kuaishou_cloud_task_states(consume_status);
 CREATE INDEX IF NOT EXISTS idx_kuaishou_cloud_task_states_updated_at_desc
  ON kuaishou_cloud_task_states(updated_at DESC);
+
+COMMENT ON TABLE task_runtime_contexts IS '任务运行时领取和截图上下文';
+COMMENT ON COLUMN task_runtime_contexts.artifacts_json IS '运行时产物集合，如截图、日志或外部资源引用';
+COMMENT ON COLUMN task_runtime_contexts.state_json IS '运行时状态快照';
+COMMENT ON TABLE kuaishou_cloud_task_states IS '从任务上下文同步出来的快手云履约可查询状态';
+
+-- 快手行业电子凭证。
+CREATE TABLE IF NOT EXISTS kuaishou_industry_vouchers (
+  id BIGSERIAL PRIMARY KEY,
+  voucher_code TEXT NOT NULL UNIQUE,
+  oid TEXT NOT NULL,
+  order_id BIGINT REFERENCES orders(id) ON DELETE SET NULL,
+  task_id BIGINT REFERENCES fulfillment_tasks(id) ON DELETE SET NULL,
+  unit_index INTEGER NOT NULL,
+  seller_id TEXT NOT NULL DEFAULT '',
+  token TEXT NOT NULL DEFAULT '',
+  status TEXT NOT NULL DEFAULT 'UNUSED',
+  valid_start_time BIGINT NOT NULL DEFAULT 0,
+  valid_end_time BIGINT NOT NULL DEFAULT 0,
+  consume_serial_num TEXT NOT NULL DEFAULT '',
+  consume_details_json JSONB NOT NULL DEFAULT '[]'::jsonb,
+  consumed_at TIMESTAMPTZ,
+  destroyed_at TIMESTAMPTZ,
+  raw_payload_json JSONB NOT NULL DEFAULT '{}'::jsonb,
+  send_callback_status TEXT NOT NULL DEFAULT 'success',
+  send_callback_attempt_count INTEGER NOT NULL DEFAULT 0,
+  send_callback_last_error TEXT NOT NULL DEFAULT '',
+  send_callback_response_json JSONB NOT NULL DEFAULT '{}'::jsonb,
+  send_callback_sent_at TIMESTAMPTZ,
+  created_at TIMESTAMPTZ NOT NULL,
+  updated_at TIMESTAMPTZ NOT NULL,
+  UNIQUE(oid, unit_index)
+);
+
+CREATE INDEX IF NOT EXISTS idx_kuaishou_industry_vouchers_oid
+  ON kuaishou_industry_vouchers(oid, unit_index);
+CREATE INDEX IF NOT EXISTS idx_kuaishou_industry_vouchers_order_id
+  ON kuaishou_industry_vouchers(order_id);
+CREATE INDEX IF NOT EXISTS idx_kuaishou_industry_vouchers_task_id
+  ON kuaishou_industry_vouchers(task_id);
+CREATE INDEX IF NOT EXISTS idx_kuaishou_industry_vouchers_status
+  ON kuaishou_industry_vouchers(status);
+CREATE INDEX IF NOT EXISTS idx_kuaishou_industry_vouchers_seller_id
+  ON kuaishou_industry_vouchers(seller_id);
+CREATE INDEX IF NOT EXISTS idx_kuaishou_industry_vouchers_send_callback_status
+  ON kuaishou_industry_vouchers(send_callback_status);
+
+COMMENT ON TABLE kuaishou_industry_vouchers IS '快手行业电子凭证码实例';
+COMMENT ON COLUMN kuaishou_industry_vouchers.consume_details_json IS '快手核销详情列表';
+COMMENT ON COLUMN kuaishou_industry_vouchers.raw_payload_json IS '快手行业电子凭证原始请求或回调报文';
+COMMENT ON COLUMN kuaishou_industry_vouchers.send_callback_response_json IS '发码回调响应内容';
+
+-- 短链。
+CREATE TABLE IF NOT EXISTS short_links (
+  id BIGSERIAL PRIMARY KEY,
+  code TEXT NOT NULL UNIQUE,
+  target_url TEXT NOT NULL,
+  target_url_hash TEXT NOT NULL,
+  source TEXT NOT NULL DEFAULT '',
+  task_id BIGINT REFERENCES fulfillment_tasks(id) ON DELETE SET NULL,
+  expires_at TIMESTAMPTZ,
+  visit_count INTEGER NOT NULL DEFAULT 0,
+  last_visited_at TIMESTAMPTZ,
+  created_at TIMESTAMPTZ NOT NULL,
+  updated_at TIMESTAMPTZ NOT NULL,
+  UNIQUE(source, task_id, target_url_hash)
+);
+
+CREATE INDEX IF NOT EXISTS idx_short_links_code
+  ON short_links(code);
+CREATE INDEX IF NOT EXISTS idx_short_links_task_id
+  ON short_links(task_id);
+CREATE INDEX IF NOT EXISTS idx_short_links_expires_at
+  ON short_links(expires_at);
+
+COMMENT ON TABLE short_links IS '内部短链映射，用于隐藏较长领取链接或外部平台链接';
@@ -1,29 +0,0 @@
-CREATE TABLE IF NOT EXISTS kuaishou_industry_vouchers (
-  id BIGSERIAL PRIMARY KEY,
-  voucher_code TEXT NOT NULL UNIQUE,
-  oid TEXT NOT NULL,
-  order_id BIGINT REFERENCES orders(id) ON DELETE SET NULL,
-  task_id BIGINT REFERENCES fulfillment_tasks(id) ON DELETE SET NULL,
-  unit_index INTEGER NOT NULL,
-  token TEXT NOT NULL DEFAULT '',
-  status TEXT NOT NULL DEFAULT 'UNUSED',
-  valid_start_time BIGINT NOT NULL DEFAULT 0,
-  valid_end_time BIGINT NOT NULL DEFAULT 0,
-  consume_serial_num TEXT NOT NULL DEFAULT '',
-  consume_details_json JSONB NOT NULL DEFAULT '[]'::jsonb,
-  consumed_at TIMESTAMPTZ,
-  destroyed_at TIMESTAMPTZ,
-  raw_payload_json JSONB NOT NULL DEFAULT '{}'::jsonb,
-  created_at TIMESTAMPTZ NOT NULL,
-  updated_at TIMESTAMPTZ NOT NULL,
-  UNIQUE(oid, unit_index)
-);
-
-CREATE INDEX IF NOT EXISTS idx_kuaishou_industry_vouchers_oid
-  ON kuaishou_industry_vouchers(oid, unit_index);
-CREATE INDEX IF NOT EXISTS idx_kuaishou_industry_vouchers_order_id
-  ON kuaishou_industry_vouchers(order_id);
-CREATE INDEX IF NOT EXISTS idx_kuaishou_industry_vouchers_task_id
-  ON kuaishou_industry_vouchers(task_id);
-CREATE INDEX IF NOT EXISTS idx_kuaishou_industry_vouchers_status
-  ON kuaishou_industry_vouchers(status);
@@ -1,23 +0,0 @@
-CREATE TABLE IF NOT EXISTS short_links (
-  id BIGSERIAL PRIMARY KEY,
-  code TEXT NOT NULL UNIQUE,
-  target_url TEXT NOT NULL,
-  target_url_hash TEXT NOT NULL,
-  source TEXT NOT NULL DEFAULT '',
-  task_id BIGINT REFERENCES fulfillment_tasks(id) ON DELETE SET NULL,
-  expires_at TIMESTAMPTZ,
-  visit_count INTEGER NOT NULL DEFAULT 0,
-  last_visited_at TIMESTAMPTZ,
-  created_at TIMESTAMPTZ NOT NULL,
-  updated_at TIMESTAMPTZ NOT NULL,
-  UNIQUE(source, task_id, target_url_hash)
-);
-
-CREATE INDEX IF NOT EXISTS idx_short_links_code
-  ON short_links(code);
-
-CREATE INDEX IF NOT EXISTS idx_short_links_task_id
-  ON short_links(task_id);
-
-CREATE INDEX IF NOT EXISTS idx_short_links_expires_at
-  ON short_links(expires_at);
@@ -1,5 +0,0 @@
-ALTER TABLE kuaishou_industry_vouchers
-  ADD COLUMN IF NOT EXISTS seller_id TEXT NOT NULL DEFAULT '';
-
-CREATE INDEX IF NOT EXISTS idx_kuaishou_industry_vouchers_seller_id
-  ON kuaishou_industry_vouchers(seller_id);
@@ -1,9 +0,0 @@
-ALTER TABLE kuaishou_industry_vouchers
-  ADD COLUMN IF NOT EXISTS send_callback_status TEXT NOT NULL DEFAULT 'success',
-  ADD COLUMN IF NOT EXISTS send_callback_attempt_count INTEGER NOT NULL DEFAULT 0,
-  ADD COLUMN IF NOT EXISTS send_callback_last_error TEXT NOT NULL DEFAULT '',
-  ADD COLUMN IF NOT EXISTS send_callback_response_json JSONB NOT NULL DEFAULT '{}'::jsonb,
-  ADD COLUMN IF NOT EXISTS send_callback_sent_at TIMESTAMPTZ;
-
-CREATE INDEX IF NOT EXISTS idx_kuaishou_industry_vouchers_send_callback_status
-  ON kuaishou_industry_vouchers(send_callback_status);