增加数据库慢查询观测与索引防线并收窄打手拼单分页查询

2026-08-22 18:40:11 +08:00
parent 0dfe363d35
commit 1b8e02dd17
16 changed files with 221 additions and 7 deletions
@@ -9,6 +9,7 @@
    "db:migrate": "tsx src/db/migrate.ts",
    "db:migrate:create": "tsx scripts/create-migration.ts",
    "db:migrate:status": "tsx scripts/migration-status.ts",
+    "check:sql": "tsx scripts/check-sql-guard.ts",
    "dev": "tsx watch --clear-screen=false src/index.ts",
    "format": "prettier --config ../../.prettierrc.json --ignore-path ../../.prettierignore --write \"src/**/*.{ts,js,json}\" \"scripts/**/*.{ts,js}\" eslint.config.js package.json tsconfig.json tsconfig.build.json tsconfig.eslint.json",
    "format:check": "prettier --config ../../.prettierrc.json --ignore-path ../../.prettierignore --check \"src/**/*.{ts,js,json}\" \"scripts/**/*.{ts,js}\" eslint.config.js package.json tsconfig.json tsconfig.build.json tsconfig.eslint.json",
@@ -23,7 +24,7 @@
    "test:industry:gen": "tsx scripts/gen-test-cases.ts",
    "test:industry:curl": "tsx scripts/curl-send-callback.ts",
    "typecheck": "tsc -p tsconfig.json --noEmit",
-    "check": "npm run format:check && npm run lint:check && npm run typecheck && npm test && npm run build",
+    "check": "npm run format:check && npm run check:sql && npm run lint:check && npm run typecheck && npm test && npm run build",
    "start": "node dist/index.js",
    "start:src": "tsx src/index.ts",
    "mock:feifei": "tsx scripts/mock-feifei-claim.ts"
@@ -0,0 +1,59 @@
+/**
+ * 数据库 SQL 约束检查。
+ * 该脚本检查关键防线是否被后续重构移除，不尝试替代 EXPLAIN 和生产监控。
+ */
+import fs from 'node:fs'
+import path from 'node:path'
+import { fileURLToPath } from 'node:url'
+
+const SCRIPT_DIR = path.dirname(fileURLToPath(import.meta.url))
+const BACKEND_ROOT = path.resolve(SCRIPT_DIR, '..')
+
+const checks: Array<{ name: string; file: string; patterns: string[] }> = [
+  {
+    name: '慢查询阈值配置',
+    file: path.join(BACKEND_ROOT, 'src/config/env-overrides.ts'),
+    patterns: ['DATABASE_SLOW_QUERY_THRESHOLD_MS', 'slowQueryThresholdMs'],
+  },
+  {
+    name: '慢查询日志与参数脱敏',
+    file: path.join(BACKEND_ROOT, 'src/db/client.ts'),
+    patterns: ['logSlowQuery', 'fingerprintSql', '参数值始终不写入日志'],
+  },
+  {
+    name: '高频 pending 查询索引',
+    file: path.join(BACKEND_ROOT, 'src/db/migrations/058_database_query_guardrails.sql'),
+    patterns: [
+      'idx_worker_cancel_requests_pending_worker_order',
+      'idx_worker_feedbacks_pending_worker_order',
+      'idx_work_order_events_order_type_id',
+    ],
+  },
+  {
+    name: '打手订单拼单查询限定当前页',
+    file: path.join(BACKEND_ROOT, 'src/repositories/worker-platform/work-order-share-repo.ts'),
+    patterns: ['workOrderIds: number[]', 'wos.work_order_id = ANY($2::bigint[])'],
+  },
+]
+
+const failures: string[] = []
+for (const check of checks) {
+  if (!fs.existsSync(check.file)) {
+    failures.push(`${check.name}: 文件不存在 ${check.file}`)
+    continue
+  }
+  const source = fs.readFileSync(check.file, 'utf8')
+  for (const pattern of check.patterns) {
+    if (!source.includes(pattern)) {
+      failures.push(`${check.name}: 缺少约束 ${pattern} (${check.file})`)
+    }
+  }
+}
+
+if (failures.length > 0) {
+  console.error('[sql-guard] SQL 约束检查失败')
+  for (const failure of failures) console.error(`[sql-guard] ${failure}`)
+  process.exitCode = 1
+} else {
+  console.info('[sql-guard] SQL 约束检查通过')
+}
@@ -26,6 +26,7 @@ export function createDefaultRuntimeConfig(projectRoot: string): RuntimeConfig {
      idleTimeoutMs: 30_000,
      connectionTimeoutMs: 5_000,
      statementTimeoutMs: 15_000,
+      slowQueryThresholdMs: 500,
    },

    storage: {
@@ -24,6 +24,7 @@ const DEPLOYMENT_ONLY_ENV_NAMES = [
  'MINIO_ROOT_USER',
  'NPM_CONFIG_REGISTRY',
  'POSTGRES_DB',
+  'POSTGRES_LOG_MIN_DURATION_STATEMENT_MS',
  'POSTGRES_PASSWORD',
  'POSTGRES_PORT',
  'POSTGRES_USER',
@@ -38,6 +38,7 @@ export const ENV_OVERRIDES: readonly EnvOverride[] = [
  integerEnv('DATABASE_IDLE_TIMEOUT_MS', ['database', 'idleTimeoutMs']),
  integerEnv('DATABASE_CONNECTION_TIMEOUT_MS', ['database', 'connectionTimeoutMs']),
  integerEnv('DATABASE_STATEMENT_TIMEOUT_MS', ['database', 'statementTimeoutMs']),
+  integerEnv('DATABASE_SLOW_QUERY_THRESHOLD_MS', ['database', 'slowQueryThresholdMs']),
  stringEnv('STORAGE_ENDPOINT', ['storage', 'endpoint']),
  stringEnv('STORAGE_BUCKET', ['storage', 'bucket']),
  stringEnv('STORAGE_ACCESS_KEY_ID', ['storage', 'accessKeyId']),
@@ -62,6 +62,12 @@ export function validateRuntimeConfig(
      min: 1,
    },
  )
+  requireOptionalInteger(
+    issues,
+    'database.slowQueryThresholdMs',
+    config.database?.slowQueryThresholdMs,
+    { min: 0 },
+  )
  requireInteger(issues, 'storage.maxUploadSizeMb', config.storage?.maxUploadSizeMb, {
    min: 1,
    max: 100,
@@ -2,12 +2,13 @@ import { Pool } from 'pg'
 import type { PoolClient, QueryResult, QueryResultRow } from 'pg'

 import { runtimeConfig } from '../config/runtime.js'
+import { logWarn } from '../utils/logger.js'

 let poolInstance: Pool | null = null

 export function getDb(): Pool {
  if (!poolInstance) {
-    poolInstance = new Pool({
+    const pool = new Pool({
      connectionString: String(runtimeConfig.database?.url || '').trim(),
      ssl: runtimeConfig.database?.ssl ? { rejectUnauthorized: false } : false,
      max: Number(runtimeConfig.database?.maxConnections || 10),
@@ -15,6 +16,12 @@ export function getDb(): Pool {
      connectionTimeoutMillis: Number(runtimeConfig.database?.connectionTimeoutMs || 5_000),
      statement_timeout: Number(runtimeConfig.database?.statementTimeoutMs || 15_000),
    })
+    pool.on('error', (error) => {
+      void logWarn('[db/pool]', '数据库连接池出现空闲连接错误', {
+        error: error instanceof Error ? error.message : String(error),
+      })
+    })
+    poolInstance = pool
  }

  return poolInstance
@@ -30,7 +37,40 @@ export async function query<T extends QueryResultRow>(
  params: unknown[] = [],
 ): Promise<QueryResult<T>> {
  const pool = getDb()
-  return pool.query<T>(text, params)
+  const startedAt = process.hrtime.bigint()
+  try {
+    const result = await pool.query<T>(text, params)
+    logSlowQuery(text, startedAt, result.rowCount ?? 0)
+    return result
+  } catch (error) {
+    logSlowQuery(text, startedAt, 0, error)
+    throw error
+  }
+}
+
+function logSlowQuery(text: string, startedAt: bigint, rowCount: number, error?: unknown) {
+  const thresholdMs = Number(runtimeConfig.database?.slowQueryThresholdMs ?? 500)
+  if (thresholdMs <= 0) return
+  const durationMs = Number(process.hrtime.bigint() - startedAt) / 1_000_000
+  if (durationMs < thresholdMs && !error) return
+  void logWarn('[db/query]', error ? '数据库 SQL 执行失败' : '数据库慢查询', {
+    durationMs: Math.round(durationMs * 100) / 100,
+    thresholdMs,
+    rowCount,
+    sql: fingerprintSql(text),
+    error: error instanceof Error ? error.message : undefined,
+  })
+}
+
+/** 只保留可聚合的 SQL 形状，参数值始终不写入日志。 */
+function fingerprintSql(text: string) {
+  return String(text || '')
+    .replace(/--[^\r\n]*/g, ' ')
+    .replace(/\/\*[\s\S]*?\*\//g, ' ')
+    .replace(/\$\d+/g, '?')
+    .replace(/\s+/g, ' ')
+    .trim()
+    .slice(0, 500)
 }

 export async function withTransaction<T>(fn: (client: PoolClient) => Promise<T>): Promise<T> {
@@ -0,0 +1,32 @@
+-- 数据库可观测性与高频查询索引保护。
+-- PostgreSQL 容器通过 shared_preload_libraries 预加载；托管数据库若无权限，跳过扩展创建并由平台侧开启。
+DO $$
+BEGIN
+  CREATE EXTENSION IF NOT EXISTS pg_stat_statements;
+EXCEPTION
+  WHEN insufficient_privilege OR undefined_file THEN
+    RAISE NOTICE '无法创建 pg_stat_statements，请在数据库平台侧启用该扩展';
+END
+$$;
+
+-- 打手订单状态计算中的 pending 撤单/反馈 EXISTS。
+CREATE INDEX IF NOT EXISTS idx_worker_cancel_requests_pending_worker_order
+  ON worker_order_cancel_requests(worker_id, work_order_id)
+  WHERE status = 'pending';
+
+CREATE INDEX IF NOT EXISTS idx_worker_feedbacks_pending_worker_order
+  ON worker_order_feedbacks(worker_id, work_order_id)
+  WHERE status = 'pending';
+
+-- 工单公共详情和资料更新角标按工单、事件类型取最新事件。
+CREATE INDEX IF NOT EXISTS idx_work_order_events_order_type_id
+  ON work_order_events(work_order_id, event_type, id DESC);
+
+-- 排行榜按打手和验收时间筛选；全部周期仍需汇总表或缓存，索引不能替代聚合。
+CREATE INDEX IF NOT EXISTS idx_work_orders_accepted_worker_at
+  ON work_orders(assigned_worker_id, accepted_at DESC)
+  WHERE status = 'accepted' AND assigned_worker_id IS NOT NULL;
+
+CREATE INDEX IF NOT EXISTS idx_work_order_shares_accepted_worker_at
+  ON work_order_shares(worker_id, accepted_at DESC)
+  WHERE status = 'accepted';
@@ -46,12 +46,16 @@ export async function listWorkOrderSharesByOrderIds(

 export async function listWorkerSharesByWorker(
  workerId: number | string,
+  workOrderIds: number[],
 ): Promise<WorkOrderShareRow[]> {
+  if (workOrderIds.length === 0) return []
  const result = await query<WorkOrderShareRow>(
    `${WORK_ORDER_SHARE_SELECT}
-     WHERE wos.worker_id = $1 AND wos.status != 'cancelled'
+     WHERE wos.worker_id = $1
+       AND wos.work_order_id = ANY($2::bigint[])
+       AND wos.status != 'cancelled'
     ORDER BY wos.id DESC`,
-    [Number(workerId)],
+    [Number(workerId), workOrderIds],
  )
  return result.rows
 }
@@ -61,9 +61,9 @@ export async function listWorkerMyOrders(query: JsonObject = {}, session: Worker
    sort: 'worker_claimed_at_desc',
  })
  const permissions = resolveWorkerPermissions(worker)
-  const myShares = await listWorkerSharesByWorker(worker.id)
-  const myShareByOrderId = new Map(myShares.map((share) => [Number(share.work_order_id), share]))
  const workOrderIds = items.map((item) => Number(item.id))
+  const myShares = await listWorkerSharesByWorker(worker.id, workOrderIds)
+  const myShareByOrderId = new Map(myShares.map((share) => [Number(share.work_order_id), share]))
  const [notes, views, pendingCancelOrderIds, pendingFeedbackOrderIds, materialEventAtByOrderId] =
    await Promise.all([
      listWorkerWorkOrderNotes(worker.id, workOrderIds),
@@ -37,6 +37,8 @@ export type RuntimeConfig = {
    idleTimeoutMs?: number
    connectionTimeoutMs?: number
    statementTimeoutMs?: number
+    /** 超过该时长的 SQL 记录慢查询告警；0 表示关闭。 */
+    slowQueryThresholdMs?: number
  }
  storage: {
    endpoint: string