xinmi/scripts/wiki_recovery_enqueue_sql.test.cjs

const { test } = require('node:test')
const assert = require('node:assert/strict')
const fs = require('node:fs')
const path = require('node:path')
const vm = require('node:vm')
const { execFileSync } = require('node:child_process')
const { recoveryFingerprintsSQL } = require('./wiki_recovery_budget_sql.cjs')
const { fifthRecoveryScope } = require('./wiki_recovery_policy.cjs')

const literal = value => `'${String(value).replaceAll("'", "''")}'`
const scope = fifthRecoveryScope
const setup = `
SET search_path = pg_temp;
CREATE TEMP TABLE task_pending_ops (id bigserial PRIMARY KEY, tenant_id bigint, task_type text, scope text, scope_id text, op text, dedup_key text, payload jsonb, fail_count int, enqueued_at timestamptz);
CREATE TEMP TABLE task_dead_letters (id bigint PRIMARY KEY, tenant_id bigint, task_type text, scope text, scope_id text, related_id text, payload jsonb);
CREATE TEMP TABLE knowledges (id text, tenant_id bigint, knowledge_base_id text, title text, parse_status text, pending_subtasks_count int, deleted_at timestamptz);
CREATE TEMP TABLE knowledge_bases (id text, tenant_id bigint, wiki_config jsonb, indexing_strategy jsonb, embedding_model_id text, summary_model_id text, deleted_at timestamptz);
CREATE TEMP TABLE chunks (id text, tenant_id bigint, knowledge_base_id text, knowledge_id text, content text, content_revision int, chunk_index int, start_at int, end_at int, chunk_type text, parent_chunk_id text, image_info text, is_enabled boolean, deleted_at timestamptz);
CREATE TEMP TABLE models (id text, tenant_id bigint, name text, type text, source text, status text, parameters jsonb, deleted_at timestamptz);
CREATE TEMP TABLE wiki_pages (tenant_id bigint, knowledge_base_id text, deleted_at timestamptz, source_refs jsonb);
CREATE TEMP TABLE wiki_dead_letter_recoveries (tenant_id bigint NOT NULL, dead_letter_id bigint NOT NULL, knowledge_id varchar NOT NULL, knowledge_base_id varchar NOT NULL, model_id varchar NOT NULL, pending_op_id bigint NOT NULL, queued_at timestamptz NOT NULL, baseline_pages bigint NOT NULL, baseline_chunks bigint NOT NULL, chunk_chars bigint NOT NULL, dead_letter_highwater bigint NOT NULL, attempt_number integer NOT NULL, retry_reason text NOT NULL DEFAULT '', deployed_fix_review jsonb, PRIMARY KEY(tenant_id,dead_letter_id,attempt_number));
${fs.readFileSync(path.resolve(__dirname, '../migrations/versioned/000214_wiki_recovery_budgets.up.sql'), 'utf8').replace('CREATE TABLE IF NOT EXISTS', 'CREATE TEMP TABLE')}
INSERT INTO task_dead_letters VALUES (10,10059,'wiki:ingest','knowledge_base',${literal(scope.knowledge_base_id)},${literal(scope.knowledge_id)},jsonb_build_object('op','ingest','knowledge_id',${literal(scope.knowledge_id)}));
INSERT INTO knowledges VALUES (${literal(scope.knowledge_id)},10059,${literal(scope.knowledge_base_id)},'isolated fixture','completed',0,NULL);
INSERT INTO knowledge_bases VALUES (${literal(scope.knowledge_base_id)},10059,jsonb_build_object('synthesis_model_id',${literal(scope.model_id)}),'{"wiki_enabled":true}','embedding','',NULL);
INSERT INTO chunks VALUES ('chunk',10059,${literal(scope.knowledge_base_id)},${literal(scope.knowledge_id)},'only fixture source',1,0,0,19,'text','','',true,NULL);
INSERT INTO models VALUES (${literal(scope.model_id)},10059,'chat','KnowledgeQA','remote','ready','{}',NULL),('embedding',10059,'embedding','Embedding','remote','ready','{}',NULL);
INSERT INTO wiki_dead_letter_recoveries VALUES (10059,10,${literal(scope.knowledge_id)},${literal(scope.knowledge_base_id)},${literal(scope.model_id)},4143,'2026-09-19T01:00:00Z',0,1,19,33,4,'previous fixture',NULL);
`

function sql(input) {
  return execFileSync('docker', ['--context', 'colima-yawan', 'exec', '-i', 'WeKnora-postgres', 'psql', '-X', '-qAt',
    '-U', 'postgres', '-d', 'yawan_native_acceptance_20260915', '-v', 'ON_ERROR_STOP=1'],
  { input, encoding: 'utf8', stdio: ['pipe', 'pipe', 'pipe'] }).trim()
}

function captureTransaction(source, review) {
  const previous = { ...scope, attempt_number: 4, pending_op_id: 4143, pending: false, new_failures: 1,
    span_status: 'failed', error_code: 'EXTRACT_FAILED', queued_at: '2026-09-19T01:00:00Z' }
  const pilot = { ...scope, dead_letter_id: 31, attempt_number: 1, pending: false, new_failures: 0, span_status: 'done',
    pages_dropped: 0, failed_slug_writes: 0, missing_expected_pages: 0, failed_page_spans: 0,
    current_pages: 1, published_summaries: 1, expected_pages: 1, pages_written: 1, pages_total: 1 }
  let transaction
  const context = {
    __dirname, process: { argv: ['node', 'recover_wiki_deadletters.cjs', '--apply', '--ids=10', '--attempt=5',
      '--reason=Isolated durable budget transaction fixture', '--review-file=/isolated/review.json'] },
    console: { log() {}, error() {} },
    require(name) {
      if (name === 'node:fs') return {
        readFileSync(file) { return JSON.stringify(file === '/isolated/review.json' ? review : {
          stage: 'published', release: review.fix_release, api_image_id: review.api_image_id, published_at: '2026-09-20T01:00:00Z' }) },
        mkdirSync() {}, writeFileSync() {},
      }
      if (name === 'node:child_process') return { execFileSync(command, args, options = {}) {
        assert.equal(command, 'docker')
        if (args.includes('inspect')) return JSON.stringify({ running: true, release: review.fix_release, image_id: review.api_image_id })
        assert.ok(args.includes('psql'), 'No trigger, model or container mutation is allowed')
        if (options.input.startsWith('BEGIN;')) {
          transaction = options.input
          throw new Error('Captured isolated transaction')
        }
        if (options.input.includes('to_regclass')) return 't'
        if (options.input.includes('FROM (SELECT ledger.*')) return JSON.stringify([pilot, previous])
        if (options.input.includes('SELECT coalesce(json_agg(row_to_json(d))')) return '[]'
        return JSON.stringify([source])
      } }
      return require(name)
    },
  }
  vm.runInNewContext(fs.readFileSync(path.join(__dirname, 'recover_wiki_deadletters.cjs'), 'utf8'), context)
  assert.ok(transaction, 'The reviewed fixture must reach the actual enqueue transaction')
  return transaction
}

test('actual enqueue transaction commits all three records once and rolls all back on receipt failure', { skip: process.env.VERIFY_RECOVERY_SQL !== '1' }, () => {
  const fingerprint = recoveryFingerprintsSQL()
  const digests = JSON.parse(sql(`${setup}
    SELECT json_build_object('source_fingerprint',${fingerprint.source},'model_fingerprint',${fingerprint.model})
    FROM task_dead_letters d JOIN knowledges n ON n.id=d.related_id JOIN knowledge_bases k ON k.id=d.scope_id;`))
  const source = { ...scope, ...digests, id: 10, embedding_model_id: 'embedding', eligible: true, chunks: 1, pending: 0 }
  const review = { ...source, budget_version: 1, max_attempt: 5, previous_attempt: 4, previous_pending_op_id: 4143,
    previous_error_code: 'EXTRACT_FAILED', fix_release: 'public-20260920-180000', api_image_id: `sha256:${'a'.repeat(64)}`,
    max_provider_requests: 200, deadline_at: new Date(Date.now() + 3600000).toISOString() }
  const transaction = captureTransaction(source, review)
  const counts = `SELECT json_build_object('ops',(SELECT count(*) FROM task_pending_ops),'budgets',(SELECT count(*) FROM wiki_recovery_budgets),'receipts',(SELECT count(*) FROM wiki_dead_letter_recoveries WHERE attempt_number=5));`
  const committed = sql(`${setup}\n${transaction}\n${transaction}\n${counts}`)
  assert.deepEqual(JSON.parse(committed.split('\n').at(-1)), { ops: 1, budgets: 1, receipts: 1 })
  const failure = sql(`${setup}
    ALTER TABLE wiki_dead_letter_recoveries ADD CONSTRAINT fail_fixture_receipt CHECK (attempt_number<>5);
    \\set ON_ERROR_STOP off
    ${transaction}
    \\set ON_ERROR_STOP on
    ${counts}`)
  assert.deepEqual(JSON.parse(failure.split('\n').at(-1)), { ops: 0, budgets: 0, receipts: 0 })
})