// 插件数据面 · 静态声明解析与校验（`DB-03 §三` 的落地件）回归测试。
//
// 覆盖三件事：
//   ① 库名换算与白名单复校（B 档"三点兜底"里的生成 + 正则两点）；
//   ② YAML 子集解析 + 声明校验（9 种中性类型 / 规模上限 / 归属列禁令）；
//   ③ diff 与「只增不减」禁令（加表/加列/加索引放行；删列/改类型/无默认值非空列拒绝）。
//
// 跑法：`npm test`（先 build 再跑 `lib/`）。

import { test } from 'node:test'
import assert from 'node:assert/strict'
import { mkdtempSync, mkdirSync, writeFileSync, rmSync, readdirSync, existsSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'

import {
  NEUTRAL_TYPES,
  PLUGIN_DB_RE,
  assertPluginDbName,
  declarationDdl,
  parseDeclFromDir,
  parseYamlSubset,
  pluginDbNameOf,
  pluginIdOf,
  validateDecl,
  PluginDataError,
} from '../lib/db/plugin-data/schema.js'
import {
  isDuplicateDatabase,
  isInsufficientPrivilege,
  metaTableDdl,
  pluginDbUrlOf,
  quoteIdent,
  reconcilePluginDatabases,
  backupSchemaOnly,
  pgDumpEnv,
} from '../lib/db/plugin-data/datastore.js'
import {
  FORBIDDEN_KINDS,
  dataPlaneStateOf,
  diffDecl,
  emptyCurrent,
  planHashOf,
  stateAllowsCreate,
  stateAllowsEnable,
  stateAllowsMigrate,
} from '../lib/db/plugin-data/diff.js'

// ── ① 库名换算与复校 ───────────────────────────────────────────────────────

test('plugin-data: pluginIdOf 去 scope / 小写 / 折非法字符', () => {
  assert.equal(pluginIdOf('@dsh-local/trpg-kit'), 'trpg_kit')
  assert.equal(pluginIdOf('mcn-suite'), 'mcn_suite')
  assert.equal(pluginIdOf('@scope/a.b'), 'a_b')
  assert.equal(pluginIdOf('Plain'), 'plain')
})

test('plugin-data: 库名 = dshs_pl_ 前缀 + pluginId，且通过正则', () => {
  assert.equal(pluginDbNameOf('@dsh-local/trpg-kit'), 'dshs_pl_trpg_kit')
  assert.match(pluginDbNameOf('mcn-suite'), PLUGIN_DB_RE)
  // 无 scope 的名字同样成立（候选池里既有带 scope 的也有不带的）。
  assert.match(pluginDbNameOf('storyforge'), PLUGIN_DB_RE)
})

test('plugin-data: assertPluginDbName 拦下非法库名（B 档白名单的第二道兜底）', () => {
  // ⚠️ 这是「PG 没有库名前缀级授权」的补偿：非法名字**必须在 DDL 之前**被拒。
  // 长度上限：标识部分 `[a-z][a-z0-9_]{0,40}` ⇒ 41 个字符是上限（40+1）。
  const maxIdent = 'a'.repeat(41)
  for (const bad of [
    'evil_probe',            // 无前缀
    'dshs_pl_9bad',          // 数字开头
    'dshs_pl_Bad',           // 大写
    'dshs_pl_',              // 空标识
    'dshs_pl_a-b',           // 连字符（换算是 `_`，非法字符进不来）
    'dshs_pl_' + 'a'.repeat(42), // 超长（42 > 41）
    'public',                // 保留名
    "dshs_pl_x'; DROP DATABASE dshs; --", // 注入形态
  ]) {
    assert.throws(() => assertPluginDbName(bad), PluginDataError, `应拒绝：${bad}`)
  }
  // 边界：41 位标识**通过**（上限之内）。
  assert.equal(assertPluginDbName('dshs_pl_' + maxIdent), 'dshs_pl_' + maxIdent)
  assert.equal(assertPluginDbName('dshs_pl_trpg_kit'), 'dshs_pl_trpg_kit')
})

// ── ② YAML 子集解析 ────────────────────────────────────────────────────────

test('plugin-data: YAML 子集能解出 DB-03 §三 的示例（块状序列 + 行内流式映射）', () => {
  const text = [
    'pluginId: trpg_kit',
    'schemaVersion: 3',
    'tables:',
    '  - name: character_sheet',
    '    scope: room',
    '    columns:',
    '      - { name: pc_name,  type: text,      notNull: true, maxBytes: 128 }',
    '      - { name: sheet,    type: json,      notNull: true, maxBytes: 16384 }',
    '      - { name: sheet_rev,type: integer,   default: 1 }',
    '    indexes:',
    '      - { columns: [room_id, pc_name], unique: true }',
  ].join('\n')
  const v = parseYamlSubset(text)
  assert.equal(v.schemaVersion, 3)
  assert.equal(v.tables.length, 1)
  assert.equal(v.tables[0].name, 'character_sheet')
  assert.equal(v.tables[0].scope, 'room')
  assert.equal(v.tables[0].columns.length, 3)
  assert.equal(v.tables[0].columns[0].name, 'pc_name')
  assert.equal(v.tables[0].columns[0].notNull, true)
  assert.equal(v.tables[0].columns[0].maxBytes, 128)
  assert.deepEqual(v.tables[0].indexes[0].columns, ['room_id', 'pc_name'])
  assert.equal(v.tables[0].indexes[0].unique, true)
})

test('plugin-data: YAML 子集拒绝超范围语法（⛔ 不静默猜）', () => {
  assert.throws(() => parseYamlSubset('---\na: 1'), /多文档/)
  assert.throws(() => parseYamlSubset('\tkey: 1'), /Tab/)
  assert.throws(() => parseYamlSubset('a: &anchor 1'), /不支持/)
})

// ── ③ 声明校验 ────────────────────────────────────────────────────────────

const GOOD = {
  schemaVersion: 3,
  tables: [
    {
      name: 'character_sheet',
      scope: 'room',
      columns: [
        { name: 'pc_name', type: 'text', notNull: true, maxBytes: 128, default: '' },
        { name: 'sheet', type: 'json', notNull: true, maxBytes: 16384, default: '{}' },
        { name: 'sheet_rev', type: 'integer', default: 1 },
      ],
      indexes: [{ columns: ['room_id', 'pc_name'], unique: true }],
    },
  ],
}

test('plugin-data: 合法声明通过，9 种中性类型齐备', () => {
  const r = validateDecl('@dsh-local/trpg-kit', GOOD, 'test')
  assert.equal(r.level, 'ok')
  assert.equal(r.decl.dbName, 'dshs_pl_trpg_kit')
  assert.equal(r.decl.tables[0].physicalName, 'p_trpg_kit_character_sheet')
  assert.deepEqual(r.decl.tables[0].columns.map((c) => c.type), ['text', 'json', 'integer'])
  assert.equal(NEUTRAL_TYPES.length, 9)
})

test('plugin-data: 声明里自带归属列 / 审计列 ⇒ 拒绝（DB-03 §一 归属列由内核补）', () => {
  for (const reserved of ['user_id', 'room_id', 'created_at', 'updated_at']) {
    const r = validateDecl('p', {
      schemaVersion: 1,
      tables: [{ name: 't', scope: 'user', columns: [{ name: reserved, type: 'text', default: '' }] }],
    }, 'test')
    assert.equal(r.level, 'invalid', reserved)
    assert.ok(r.findings.some((f) => f.code === 'column_reserved'), reserved)
  }
})

test('plugin-data: 每表 ≤20 列 / 每插件 ≤20 表（DB-03 §三 规模上限）', () => {
  const many = (n) => Array.from({ length: n }, (_, i) => ({ name: `c${i}`, type: 'text', default: '' }))
  const over = validateDecl('p', {
    schemaVersion: 1,
    tables: [{ name: 't', scope: 'user', columns: many(21) }],
  }, 'test')
  assert.ok(over.findings.some((f) => f.code === 'columns_over_limit'))

  const tables = Array.from({ length: 21 }, (_, i) => ({ name: `t${i}`, scope: 'user', columns: many(1) }))
  const overT = validateDecl('p', { schemaVersion: 1, tables }, 'test')
  assert.ok(overT.findings.some((f) => f.code === 'tables_over_limit'))
})

test('plugin-data: 未知类型 / json 缺 maxBytes / 无默认值非空列 均被拒', () => {
  const bad = validateDecl('p', {
    schemaVersion: 1,
    tables: [{
      name: 't',
      scope: 'user',
      columns: [
        { name: 'a', type: 'varchar(255)' },              // 非中性类型
        { name: 'b', type: 'json', default: '{}' },        // json 缺 maxBytes
        { name: 'c', type: 'text', notNull: true },        // notNull 无默认值
      ],
    }],
  }, 'test')
  assert.equal(bad.level, 'invalid')
  const codes = bad.findings.map((f) => f.code)
  assert.ok(codes.includes('column_type_unknown'))
  assert.ok(codes.includes('json_maxbytes_required'))
  assert.ok(codes.includes('notnull_no_default'))
  // 收集式：三条一次全报（不"改一条重传一次"）。
  assert.equal(bad.findings.length, 3)
})

test('plugin-data: scope 必填且只能是 user / room', () => {
  for (const scope of [undefined, 'plugin', 'USER']) {
    const r = validateDecl('p', {
      schemaVersion: 1,
      tables: [{ name: 't', scope, columns: [{ name: 'a', type: 'text', default: '' }] }],
    }, 'test')
    assert.ok(r.findings.some((f) => f.code === 'scope_invalid'), String(scope))
  }
})

// ── ④ 从包目录解析（package.json 的 dsh.data → 内联 / yaml 文件）───────────

function withPkgDir(pkg, files = {}) {
  const dir = mkdtempSync(join(tmpdir(), 'dsh-data-decl-'))
  mkdirSync(dir, { recursive: true })
  writeFileSync(join(dir, 'package.json'), JSON.stringify(pkg, null, 2))
  for (const [rel, text] of Object.entries(files)) {
    const abs = join(dir, rel)
    mkdirSync(join(abs, '..'), { recursive: true })
    writeFileSync(abs, text)
  }
  return dir
}

test('plugin-data: 无 dsh.data ⇒ level none（不建库、不受门禁）', () => {
  const dir = withPkgDir({ name: '@dsh-local/plain', version: '1.0.0' })
  try {
    const r = parseDeclFromDir(dir, '@dsh-local/plain')
    assert.equal(r.level, 'none')
    assert.equal(r.origin, 'none')
  } finally { rmSync(dir, { recursive: true, force: true }) }
})

test('plugin-data: dsh.data.schema 指向 YAML 文件 ⇒ 解析成功', () => {
  const yaml = [
    'schemaVersion: 2',
    'tables:',
    '  - name: card',
    '    scope: user',
    '    columns:',
    '      - { name: title, type: text, notNull: true, default: "" }',
  ].join('\n') + '\n'
  const dir = withPkgDir({ name: 'mcn-suite', version: '1.0.0', dsh: { data: { schema: './dsh.data.yaml' } } }, { 'dsh.data.yaml': yaml })
  try {
    const r = parseDeclFromDir(dir, 'mcn-suite')
    assert.equal(r.level, 'ok', JSON.stringify(r.findings))
    assert.equal(r.origin, 'yaml-file')
    assert.equal(r.decl.dbName, 'dshs_pl_mcn_suite')
    assert.equal(r.decl.tables[0].physicalName, 'p_mcn_suite_card')
  } finally { rmSync(dir, { recursive: true, force: true }) }
})

test('plugin-data: dsh.data.schema 内联对象 ⇒ 解析成功', () => {
  const dir = withPkgDir({
    name: 'inline-plug',
    version: '1.0.0',
    dsh: { data: { schema: { schemaVersion: 1, tables: [{ name: 't', scope: 'user', columns: [{ name: 'a', type: 'bigint', default: 0 }] }] } } },
  })
  try {
    const r = parseDeclFromDir(dir, 'inline-plug')
    assert.equal(r.level, 'ok')
    assert.equal(r.origin, 'package-json-inline')
  } finally { rmSync(dir, { recursive: true, force: true }) }
})

test('plugin-data: 声明文件逃出包目录 / 不存在 ⇒ 拒绝（路径穿越防护）', () => {
  const dir = withPkgDir({ name: 'x', version: '1.0.0', dsh: { data: { schema: '../../../etc/passwd' } } })
  try {
    const r = parseDeclFromDir(dir, 'x')
    assert.equal(r.level, 'invalid')
    assert.equal(r.findings[0].code, 'schema_escapes_package')
  } finally { rmSync(dir, { recursive: true, force: true }) }

  const dir2 = withPkgDir({ name: 'y', version: '1.0.0', dsh: { data: { schema: './missing.yaml' } } })
  try {
    const r = parseDeclFromDir(dir2, 'y')
    assert.equal(r.level, 'invalid')
    assert.equal(r.findings[0].code, 'schema_file_missing')
  } finally { rmSync(dir2, { recursive: true, force: true }) }
})

// ── ⑤ DDL 生成 ────────────────────────────────────────────────────────────

test('plugin-data: DDL 带上库内归属列 / 审计列，表名带 p_<pluginId>_ 前缀', () => {
  const r = validateDecl('@dsh-local/trpg-kit', GOOD, 'test')
  const ddl = declarationDdl(r.decl)
  assert.equal(ddl.length, 2, '一张表 + 一个索引')
  assert.match(ddl[0], /CREATE TABLE IF NOT EXISTS "p_trpg_kit_character_sheet"/)
  assert.match(ddl[0], /"room_id" TEXT NOT NULL/)
  assert.match(ddl[0], /"created_at" BIGINT NOT NULL/)
  assert.match(ddl[0], /"updated_at" BIGINT NOT NULL/)
  assert.match(ddl[0], /"sheet" JSONB NOT NULL DEFAULT '\{\}'/)
  assert.match(ddl[0], /"sheet" JSONB/)
  assert.match(ddl[1], /CREATE UNIQUE INDEX IF NOT EXISTS "p_trpg_kit_character_sheet_room_id_pc_name_uq_1"/)
})

test('plugin-data: 默认值里的单引号被转义（SQL 字面量）', () => {
  const r = validateDecl('p', {
    schemaVersion: 1,
    tables: [{ name: 't', scope: 'user', columns: [{ name: 'a', type: 'text', notNull: true, default: "it's" }] }],
  }, 'test')
  const ddl = declarationDdl(r.decl).join('\n')
  assert.match(ddl, /DEFAULT 'it''s'/)
})

// ── ⑥ diff 与「只增不减」禁令 ─────────────────────────────────────────────

const DECL_OK = validateDecl('@dsh-local/trpg-kit', GOOD, 'test').decl

test('plugin-data: 空库 ⇒ 计划含建库 + 建表 + 建索引，无禁令', () => {
  const plan = diffDecl(DECL_OK, emptyCurrent())
  assert.deepEqual(plan.items.map((i) => i.kind), ['create_database', 'create_table', 'add_index'])
  assert.equal(plan.forbidden.length, 0)
  assert.deepEqual(plan.summary, { tables: 1, columns: 0, indexes: 1 })
})

test('plugin-data: 库已就绪（结构一致）⇒ 零待做项、零禁令', () => {
  const current = {
    exists: true,
    schemaVersion: 3,
    tables: [{
      name: 'p_trpg_kit_character_sheet',
      columns: [
        { name: 'room_id', dataType: 'text' },
        { name: 'created_at', dataType: 'int8' },
        { name: 'updated_at', dataType: 'int8' },
        { name: 'pc_name', dataType: 'text' },
        { name: 'sheet', dataType: 'jsonb' },
        { name: 'sheet_rev', dataType: 'int4' },
      ],
      indexes: ['p_trpg_kit_character_sheet_room_id_pc_name_uq_1'],
    }],
  }
  const plan = diffDecl(DECL_OK, current)
  assert.equal(plan.items.length, 0)
  assert.equal(plan.forbidden.length, 0)
  assert.equal(dataPlaneStateOf({ hasDeclaration: true, current, plan }), 'ready')
})

test('plugin-data: 现状有、声明无的列 ⇒ forbidden drop_column（⛔ 只增不减）', () => {
  const current = {
    exists: true,
    schemaVersion: 3,
    tables: [{
      name: 'p_trpg_kit_character_sheet',
      columns: [
        { name: 'room_id', dataType: 'text' },
        { name: 'pc_name', dataType: 'text' },
        { name: 'sheet', dataType: 'jsonb' },
        { name: 'sheet_rev', dataType: 'int4' },
        { name: 'legacy_note', dataType: 'text' }, // ← 声明里没有它
      ],
      indexes: [],
    }],
  }
  const plan = diffDecl(DECL_OK, current)
  assert.ok(plan.forbidden.some((f) => f.kind === 'drop_column' && f.target.endsWith('legacy_note')))
  assert.equal(dataPlaneStateOf({ hasDeclaration: true, current, plan }), 'blocked')
})

test('plugin-data: 类型变化 ⇒ forbidden alter_type（int ↔ text 之类的静默转换被拦）', () => {
  const current = {
    exists: true,
    schemaVersion: 3,
    tables: [{
      name: 'p_trpg_kit_character_sheet',
      columns: [
        { name: 'room_id', dataType: 'text' },
        { name: 'pc_name', dataType: 'text' },
        { name: 'sheet', dataType: 'jsonb' },
        { name: 'sheet_rev', dataType: 'text' }, // 声明是 integer
      ],
      indexes: [],
    }],
  }
  const plan = diffDecl(DECL_OK, current)
  assert.ok(plan.forbidden.some((f) => f.kind === 'alter_type' && f.target.endsWith('sheet_rev')))
})

test('plugin-data: 声明新增列 ⇒ add_column 放行（必须带默认值）', () => {
  const decl2 = validateDecl('@dsh-local/trpg-kit', {
    ...GOOD,
    schemaVersion: 4,
    tables: [{
      ...GOOD.tables[0],
      columns: [...GOOD.tables[0].columns, { name: 'level', type: 'integer', default: 1 }],
    }],
  }, 'test').decl
  const current = {
    exists: true,
    schemaVersion: 3,
    tables: [{
      name: 'p_trpg_kit_character_sheet',
      columns: [
        { name: 'room_id', dataType: 'text' },
        { name: 'pc_name', dataType: 'text' },
        { name: 'sheet', dataType: 'jsonb' },
        { name: 'sheet_rev', dataType: 'int4' },
      ],
      indexes: ['p_trpg_kit_character_sheet_room_id_pc_name_uq_1'],
    }],
  }
  const plan = diffDecl(decl2, current)
  assert.equal(plan.forbidden.length, 0)
  assert.deepEqual(plan.items.map((i) => i.kind), ['add_column'])
  assert.match(plan.items[0].sql, /ADD COLUMN IF NOT EXISTS "level" INTEGER DEFAULT 1/)
  assert.equal(dataPlaneStateOf({ hasDeclaration: true, current, plan }), 'drift')
})

test('plugin-data: planHash 覆盖声明 / 现状 / 库名（防 TOCTOU）', () => {
  const cur = emptyCurrent()
  const plan = diffDecl(DECL_OK, cur)
  const h1 = planHashOf({ dbName: DECL_OK.dbName, schemaVersion: 3, items: plan.items, forbidden: [], current: cur })
  assert.equal(h1, plan.planHash, '同一份输入 ⇒ 同一指纹')

  // 现状变了（有人手工改过库）⇒ 指纹必须变（否则"确认过的计划"会失真）。
  const drifted = { exists: true, schemaVersion: 3, tables: [{ name: 'p_trpg_kit_x', columns: [{ name: 'a', dataType: 'text' }] }] }
  const h2 = planHashOf({ dbName: DECL_OK.dbName, schemaVersion: 3, items: plan.items, forbidden: [], current: drifted })
  assert.notEqual(h1, h2)

  // 库名变了 ⇒ 指纹变。
  const h3 = planHashOf({ dbName: 'dshs_pl_other', schemaVersion: 3, items: plan.items, forbidden: [], current: cur })
  assert.notEqual(h1, h3)
})

test('plugin-data: 状态机 —— 门禁判据（未建库不许开启 / 检测不过不许建库 / 未确认不许迁移）', () => {
  // 门禁 ① 未建库不许「开启」：只有 none / ready 放行。
  assert.equal(stateAllowsEnable('none'), true, '无数据面 ⇒ 不受约束')
  assert.equal(stateAllowsEnable('ready'), true)
  for (const s of ['pending', 'created', 'drift', 'blocked', 'error']) {
    assert.equal(stateAllowsEnable(s), false, `不许开启：${s}`)
  }
  // 门禁 ② 检测不通过不许建库：blocked 不在可建集合内。
  assert.equal(stateAllowsCreate('pending'), true)
  assert.equal(stateAllowsCreate('blocked'), false, '声明非法 ⇒ 不许建库')
  assert.equal(stateAllowsCreate('ready'), false)
  // 门禁 ③ 未确认 planHash 不许迁移：只有 drift 可执行（blocked 必须改声明）。
  assert.equal(stateAllowsMigrate('drift'), true)
  for (const s of ['pending', 'blocked', 'error', 'ready', 'none', 'created']) {
    assert.equal(stateAllowsMigrate(s), false, `不许迁移：${s}`)
  }
})

test('plugin-data: forbidden 清单只增不减（四种禁令齐备）', () => {
  // ⛔ 这条断言是给后人看的：要加"允许的操作"必须先来这里改这份清单。
  assert.deepEqual([...FORBIDDEN_KINDS], ['drop_column', 'alter_type', 'rename', 'notnull_no_default'])
})

test('plugin-data: 声明非法时 diff 不参与（level invalid ⇒ 无 decl）', () => {
  const bad = validateDecl('p', { schemaVersion: 0, tables: [] }, 'test')
  assert.equal(bad.level, 'invalid')
  assert.equal(bad.decl, null)
})

// ── ⑦ 建库口（`datastore.ts`）—— B 档"三点兜底"的可执行判据 ───────────────────
//
// ⚠️ 本节的判据**不连库**：库名复校 / 标识符引用 / 错误码映射 / 对账集合逻辑
// 全是纯函数或可用假池替代 ⇒ 能在 CI 里把 B 档最危险的那条（库名拼进 SQL）钉死。
// 真机取证（建库 ＋ 台账号双写一致）另见交付物，不进 `npm test`。

test('plugin-data: 标识符引用把双引号翻倍（⛔ 库名永不裸拼进 SQL）', () => {
  // 标准 PG 引用：标识符里的 `"` 翻倍。配合 `assertPluginDbName` 成两重防线。
  assert.equal(quoteIdent('dshs_pl_trpg_kit'), '"dshs_pl_trpg_kit"')
  assert.equal(quoteIdent('a"b'), '"a""b"')
  // 注入形态：引号被翻倍 ⇒ 整串仍是**单个标识符**，不可能逃出 `CREATE DATABASE "…"`。
  // 期望值用模板串写（含单引号，⛔ 不转义成 `\'` 那种易错形态）。
  const INJECT = "x'; DROP DATABASE dshs; --"
  assert.equal(quoteIdent(INJECT), `"x'; DROP DATABASE dshs; --"`)
  // 双引号形态：翻倍后仍是单标识符（PG 的转义规则）。
  assert.equal(quoteIdent('a"b"' + 'c'), '"a""b""c"')
})

test('plugin-data: pluginDbUrlOf 只改库名、拒绝非法库名（⛔ 不裸拼连接串）', () => {
  const base = 'postgres://dshs:pw@127.0.0.1:15432/dshs'
  // 正常：path 换成插件库，其余（user / host / port / 口令）逐字不变。
  assert.equal(pluginDbUrlOf(base, 'dshs_pl_trpg_kit'), 'postgres://dshs:pw@127.0.0.1:15432/dshs_pl_trpg_kit')
  // ⚠️ 带 query 的连接串也不能丢 query —— 用 URL 解析正是为此（字符串替换会丢）。
  assert.equal(
    pluginDbUrlOf('postgres://u:p@h:5432/dshs?sslmode=require', 'dshs_pl_x'),
    'postgres://u:p@h:5432/dshs_pl_x?sslmode=require',
  )
  // 🔴 非法库名 ⇒ 在建连接**之前**就抛（B 档第二点），⛔ 不下发。
  for (const bad of ['evil_probe', 'dshs_pl_9bad', 'dshs_pl_a-b', "dshs_pl_x'; --"]) {
    assert.throws(() => pluginDbUrlOf(base, bad), PluginDataError, `应拒绝：${bad}`)
  }
  // 连接串不是 URL 形态 ⇒ 抛（⛔ 不降级猜测成"指向控制面库"）。
  assert.throws(() => pluginDbUrlOf('host=127.0.0.1 dbname=dshs', 'dshs_pl_x'), PluginDataError)
})

test('plugin-data: PG 错误码分类（幂等 vs 未授 CREATEDB）', () => {
  // 42P04 duplicate_database ⇒ 幂等成功（§二-② 验收断言）。
  assert.equal(isDuplicateDatabase({ code: '42P04' }), true)
  assert.equal(isDuplicateDatabase({ code: '42501' }), false)
  assert.equal(isDuplicateDatabase(new Error('boom')), false)
  assert.equal(isDuplicateDatabase(null), false)
  // 42501 insufficient_privilege ⇒ 503 PG_CREATEDB_MISSING（未被 ALTER ROLE 授权 / 已回滚）。
  assert.equal(isInsufficientPrivilege({ code: '42501' }), true)
  assert.equal(isInsufficientPrivilege({ code: '42P04' }), false)
})

test('plugin-data: 库内台账表 DDL 幂等且带版本列（diff 判 ready 的依据）', () => {
  const ddl = metaTableDdl()
  assert.match(ddl, /CREATE TABLE IF NOT EXISTS "p_meta_schema"/)
  assert.match(ddl, /schema_version BIGINT NOT NULL/)
  // ⛔ 不带 `p_<pluginId>_` 前缀 —— 它是库级台账，不是某插件的业务表。
  assert.ok(!/p_[a-z]+_[a-z_]+ /.test(ddl.replace('p_meta_schema', '')))
})

test('plugin-data: 台账 × pg_database 对账 —— 绕过平台建的库必须被抓出来', () => {
  // 假池：只实现本函数用到的 `query`（`datastore.ts` 对本函数只调这一处）。
  const pool = {
    query: async () => ({ rows: [{ datname: 'dshs_pl_trpg_kit' }, { datname: 'dshs_pl_evil_probe' }, { datname: 'dshs_pl_gone' }] }),
  }
  return reconcilePluginDatabases(pool, ['dshs_pl_trpg_kit', 'dshs_pl_deleted']).then((r) => {
    // ⚠️ `databases` 的**顺序**由 SQL 的 `ORDER BY datname` 负责，假池不排序 ⇒
    //    这里只断言集合成员（顺序归真机取证管，⛔ 不让假池的返回序污染判据）。
    assert.deepEqual([...r.databases].sort(), ['dshs_pl_evil_probe', 'dshs_pl_gone', 'dshs_pl_trpg_kit'])
    assert.deepEqual(r.ledgered, ['dshs_pl_deleted', 'dshs_pl_trpg_kit'])
    // 🔴 orphans = 库里在、台账不在 ⇒ 绕过平台建的（B 档第三点要抓的正是这个）。
    assert.deepEqual([...r.orphans].sort(), ['dshs_pl_evil_probe', 'dshs_pl_gone'])
    // 反向：台账在、库里没了 ⇒ 也被平台外动过。
    assert.deepEqual(r.missing, ['dshs_pl_deleted'])
  })
})

test('plugin-data: 对账一致时两个差异集合都为空（clean 判据）', () => {
  const pool = { query: async () => ({ rows: [{ datname: 'dshs_pl_a' }] }) }
  return reconcilePluginDatabases(pool, ['dshs_pl_a']).then((r) => {
    assert.deepEqual(r.orphans, [])
    assert.deepEqual(r.missing, [])
  })
})

// ── ④ 迁移前结构备份（设计件 §七-3：备份失败 ⇒ ⛔ 不执行）────────────────────

test('plugin-data: 结构备份 —— 非法库名在下发 pg_dump 之前就被拒', async () => {
  const root = mkdtempSync(join(tmpdir(), 'bk-'))
  try {
    const r = await backupSchemaOnly('postgres://u:p@127.0.0.1:5432/x', 'evil_probe', root, 'pkg', 3)
    assert.equal(r.ok, false)
    assert.match(r.error, /invalid_plugin_db_name/)
    // 关键：⛔ 没落任何文件（连目录都不该建）。
    assert.equal(r.path, '')
    assert.deepEqual(readdirSync(root), [])
  } finally {
    rmSync(root, { recursive: true, force: true })
  }
})

test('plugin-data: 结构备份 —— 包名折叠成安全路径，⛔ 不越出 backupRoot', async () => {
  const root = mkdtempSync(join(tmpdir(), 'bk-'))
  try {
    // 用不存在的库名（合法）逼出 pg_dump 失败路径 —— 但仍要先看**目标目录**是否安全。
    const r = await backupSchemaOnly(
      'postgres://u:p@127.0.0.1:1/none',
      'dshs_pl_probe',
      root,
      '../../etc/passwd',
      2,
    )
    assert.equal(r.ok, false) // 连不上 ⇒ 失败，但路径必须已被折叠
    // 折叠后不含 `..` / 不含 `/`：目录层最多是 <root>/plugin-db/<flat>
    const top = readdirSync(root)
    assert.deepEqual(top, ['plugin-db'])
    const sub = readdirSync(join(root, 'plugin-db'))
    assert.equal(sub.length, 1)
    assert.ok(!sub[0].includes('..'), '⛔ 包名里的 `..` 未被折叠')
    assert.ok(!sub[0].includes('/'), '⛔ 包名里的 `/` 未被折叠')
  } finally {
    rmSync(root, { recursive: true, force: true })
  }
})

test('plugin-data: 结构备份 —— pg_dump 不可用/连不上时 ok=false 且不留半截文件', async () => {
  const root = mkdtempSync(join(tmpdir(), 'bk-'))
  try {
    // 指到一个必然连不上的端口 ⇒ 走失败分支。
    const r = await backupSchemaOnly('postgres://u:p@127.0.0.1:1/x', 'dshs_pl_probe', root, 'probe', 1)
    assert.equal(r.ok, false)
    assert.equal(r.bytes, 0)
    assert.ok(r.error !== null && r.error.length > 0)
    // 🔴 「⛔ 不留一个看起来像备份的空壳」= 目录里不能有 .sql
    const dir = join(root, 'plugin-db', 'probe')
    const files = existsSync(dir) ? readdirSync(dir) : []
    assert.equal(files.filter((f) => f.endsWith('.sql')).length, 0)
  } finally {
    rmSync(root, { recursive: true, force: true })
  }
})

/**
 * 🔴 回归钉：`pg_dump` 的**主机/端口/用户/口令必须真的进到子进程 env**。
 *
 * 起因（2026-09-23 第 8 棒真机 E2E 抓出的真缺陷）：`pgEnvFrom()` 的返回值被摊在
 * `execFileSync` 的 **options 层**（与 `env:` 平级）而不是并进 `env` ⇒ 子进程一个 `PGHOST`
 * 都拿不到 ⇒ `pg_dump` 回落 Unix socket `/var/run/postgresql/.s.PGSQL.5432`
 * ⇒ 47 上必报 `could not connect to server: No such file or directory`，
 * **两条真机 E2E 判据（备份落盘 / 反证 500）全被它挡住**。
 *
 * 本用例直接钉住**被测函数**的返回值形态：它必须是一份含 `PGHOST`/`PGPORT`/`PGUSER`/
 * `PGPASSWORD` 的**完整 env**（⇒ 调用处只能整体交给 `env:`，摊在 options 层就不成立）。
 * ⚠️ 不走"生成假 pg_dump 再 spawn"的路子：Windows 的 `execFileSync` 不从 PATH 解析
 *    `.cmd`/`.bat`（实测 `ENOENT`/`EINVAL`），跨平台会假红 —— 判据应钉在不变式上。
 */
test('plugin-data: 结构备份 —— pg_dump 的 env 必须含 PGHOST/PGPORT/PGUSER/PGPASSWORD', () => {
  const env = pgDumpEnv('postgres://dshs:secret@127.0.0.1:15432/dshs', 'secret')
  assert.equal(env.PGHOST, '127.0.0.1', '🔴 PGHOST 必须下发到 pg_dump 子进程')
  assert.equal(env.PGPORT, '15432', '🔴 PGPORT 必须下发到 pg_dump 子进程')
  assert.equal(env.PGUSER, 'dshs', '🔴 PGUSER 必须下发到 pg_dump 子进程')
  assert.equal(env.PGPASSWORD, 'secret', '口令必须经 PGPASSWORD 下发（⛔ 不进 argv）')
  assert.ok(env.PGCONNECT_TIMEOUT !== undefined, '应带连接超时，避免挂死')
  // ⛔ 口令不许出现在 argv 里 —— 本函数只负责 env，故 argv 由调用处的字面量保证；
  //    这里反向钉一下：返回值**只有** env 语义的键，不许混进 execFileSync 的选项名。
  for (const k of Object.keys(env)) {
    assert.ok(!['stdio', 'timeout', 'cwd', 'encoding'].includes(k), `⛔ env 里混进了选项名 ${k}`)
  }
})

test('plugin-data: 结构备份 —— 连接串不可解析时 env 仍可用（不抛）', () => {
  const env = pgDumpEnv('not-a-url', 'pw')
  assert.equal(env.PGPASSWORD, 'pw')
  assert.equal(env.PGHOST, undefined)
})