xinmi/scripts/rebuild_grounded_graph.cjs

const fs = require('node:fs')
const path = require('node:path')
const assert = require('node:assert/strict')
const { execFileSync } = require('node:child_process')
const root = path.resolve(__dirname, '../..')
const docker = (...args) => execFileSync('docker', ['--context', 'colima-yawan', ...args], { encoding: 'utf8', maxBuffer: 32 * 1024 * 1024 }).trim()
const allIDs = ['16c3e59c-4bb6-41cf-ae0a-6d0941b3d032', '20b9280e-ed73-4700-8fa5-944d6ae6dd5f', 'bb86b44a-576b-4bf8-a6bf-8f1baca3a845', 'badd084b-f229-4005-8fb3-196fbe139fe0', '5d28a7e5-e00e-44a7-a0e1-c72342cdcf68', '3ae9257d-928a-4886-8101-7cfd0ed8c27c']
const args = process.argv.slice(2)
const revisionArg = args.find(arg => arg.startsWith('--backup-revision='))
const chunksArg = args.find(arg => arg.startsWith('--chunks='))
const chunks = chunksArg ? [...new Set(chunksArg.slice('--chunks='.length).split(','))] : []
assert.ok(chunks.length <= 51, 'At most 51 explicitly reviewed chunks may be rebuilt')
for (const chunk of chunks) assert.match(chunk, /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/)
assert.ok(!chunksArg || (chunks.length > 0 && args.includes('--product-only')), 'Targeted repair requires product-only scope')
const revision = revisionArg?.slice('--backup-revision='.length) || '20260919'
assert.match(revision, /^[a-z0-9_]{1,32}$/, 'Backup revision must be an SQL-safe identifier')
assert.ok(args.every(arg => ['--apply', '--product-only'].includes(arg) || arg === revisionArg || arg === chunksArg), 'Unknown repair option')
const ids = args.includes('--product-only') ? allIDs.slice(0, 1) : allIDs
const recordsBackup = `native_graph_grounding_backup_${revision}`
const sourcesBackup = `native_graph_source_grounding_backup_${revision}`
const scope = `tenant_id=10059 AND knowledge_base_id IN (${ids.map(id => `'${id}'`).join(',')})`
const chunkValues = chunks.map(id => `'${id}'`).join(',')
const recordFilter = chunks.length ? ` AND (source_table='knowledgegraphchunkstate' AND payload->>'native_chunk_id' IN (${chunkValues}) OR source_table='knowledgegraphevidence' AND payload->>'source_chunk_id' IN (${chunks.map(id => `'native:${id}'`).join(',')}))` : ''
const sourceFilter = chunks.length ? ` AND native_chunk_id IN (${chunkValues})` : ''
const sql = query => docker('exec', 'WeKnora-postgres', 'psql', '-XAt', '-v', 'ON_ERROR_STOP=1', '-U', 'postgres', '-d', 'yawan_native_acceptance_20260915', '-c', query)
function queueStatus() {
  const inspected = JSON.parse(docker('inspect', 'yawan-native-acceptance'))[0]
  const env = Object.fromEntries(inspected.Config.Env.map(value => { const i = value.indexOf('='); return [value.slice(0, i), value.slice(i + 1)] }))
  const previous = process.env.REDISCLI_AUTH
  process.env.REDISCLI_AUTH = env.REDIS_PASSWORD || ''
  try {
    const result = {}
    for (const [state, command] of [['active', 'LLEN'], ['pending', 'LLEN'], ['scheduled', 'ZCARD'], ['retry', 'ZCARD']]) {
      const value = docker('exec', '-e', 'REDISCLI_AUTH', 'WeKnora-redis', 'redis-cli', '-n', env.REDIS_DB || '0', command, `asynq:{graph}:${state}`)
      assert.match(value, /^\d+$/)
      result[state] = Number(value)
    }
    return result
  } finally {
    if (previous === undefined) delete process.env.REDISCLI_AUTH
    else process.env.REDISCLI_AUTH = previous
  }
}
function main() {
  const queue = queueStatus()
  const count = Number(sql(`SELECT count(*) FROM native_graph_records WHERE ${scope} AND payload->>'origin'='native_extraction_v1'`))
  if (chunks.length) {
    const matched = Number(sql(`SELECT count(*) FROM chunks WHERE ${scope} AND id IN (${chunkValues}) AND deleted_at IS NULL AND is_enabled AND chunk_type='text'`))
    assert.equal(matched, chunks.length, 'Every selected source must belong to the active product knowledge base')
  }
  const directCount = Number(sql(`SELECT count(*) FROM native_graph_records WHERE ${scope} AND payload->>'origin'='native_extraction_v1'${recordFilter}`))
  if (!process.argv.includes('--apply')) { console.log(JSON.stringify({ queue, derived_records_to_backup: count, direct_records_to_rebuild: directCount, selected_chunks: chunks })); return }
  assert.equal(docker('inspect', 'yawan-native-graph-recovery', '--format', '{{.State.Running}}'), 'false', 'Stop graph enqueuer before reset')
  assert.ok(Object.values(queue).every(n => n === 0), 'Wait for graph queue to drain before rebuilding')
  assert.ok(count > 0 && count < 20000, 'Unexpected graph repair scope')
  const backup = sql(`SELECT json_build_object('records', (SELECT json_agg(r) FROM native_graph_records r WHERE ${scope} AND payload->>'origin'='native_extraction_v1'), 'sources', (SELECT json_agg(s) FROM native_graph_sources s WHERE ${scope} AND mapping_state='native_extraction'))`)
  const backupPath = path.join(root, 'acceptance', `native-graph-before-grounding-${Date.now()}.json`)
  const parsed = JSON.parse(backup)
  assert.equal(parsed.records.length, count)
  fs.writeFileSync(backupPath, backup, { flag: 'wx', mode: 0o600 })
  // A transactional database copy also keeps rollback possible if the host disk disappears.
  sql(`BEGIN;
    ${ids.map(id => `SELECT pg_advisory_xact_lock(hashtextextended('native-graph:10059:${id}',0));`).join('\n')}
    CREATE TABLE ${recordsBackup} AS SELECT * FROM native_graph_records WHERE ${scope} AND payload->>'origin'='native_extraction_v1';
    CREATE TABLE ${sourcesBackup} AS SELECT * FROM native_graph_sources WHERE ${scope} AND mapping_state='native_extraction';
    DO $$ BEGIN IF (SELECT count(*) FROM ${recordsBackup}) <> ${count} THEN RAISE EXCEPTION 'Graph changed after backup'; END IF; END $$;
    DELETE FROM native_graph_records WHERE ${scope} AND payload->>'origin'='native_extraction_v1'${recordFilter};
    DELETE FROM native_graph_sources WHERE ${scope} AND mapping_state='native_extraction'${sourceFilter};
    ${chunks.length ? `DELETE FROM native_graph_records r WHERE ${scope} AND r.payload->>'origin'='native_extraction_v1'
      AND r.source_table IN ('knowledgeentity','knowledgerelation')
      AND NOT EXISTS (SELECT 1 FROM native_graph_records e WHERE e.tenant_id=r.tenant_id AND e.knowledge_base_id=r.knowledge_base_id
        AND e.source_table='knowledgegraphevidence' AND e.payload->>'subject_id'=r.legacy_id::text
        AND e.payload->>'subject_type'=CASE r.source_table WHEN 'knowledgeentity' THEN 'entity' ELSE 'relation' END);` : ''}
    COMMIT;`)
  const remaining = Number(sql(`SELECT count(*) FROM native_graph_records WHERE ${scope} AND payload->>'origin'='native_extraction_v1'`))
  console.log(JSON.stringify({ removed_derived_records: count - remaining, backed_up_sources: parsed.sources?.length || 0, selected_chunks: chunks, backup: backupPath, database_backups: [recordsBackup, sourcesBackup], source_documents_and_indexes_preserved: true }))
}
try { main() } catch (error) { console.error(error.message); process.exitCode = 1 }