176 lines
13 KiB
JavaScript
176 lines
13 KiB
JavaScript
/** One-shot, audited owner backfill. The upstream report database is SELECT-only. */
|
|
import { createHash, randomUUID } from 'node:crypto'
|
|
import { readFile, writeFile, lstat } from 'node:fs/promises'
|
|
import path from 'node:path'
|
|
import { fileURLToPath, pathToFileURL } from 'node:url'
|
|
import { createRequire } from 'node:module'
|
|
const require = createRequire(import.meta.url)
|
|
const { selectArtPlannerOwner } = require('../lib/art-workflow-owner.cjs')
|
|
export const ART_CODE = 'PROC-3571C915-0506-4490-A91A-8AA7EA33F838'
|
|
export const SYNC_ID = 'ding-workflow-hourly-v4-product-id'
|
|
const own = (value, key) => !!value && typeof value === 'object' && Object.hasOwn(value, key)
|
|
const canonical = value => value === null || typeof value !== 'object' ? JSON.stringify(value)
|
|
: Array.isArray(value) ? `[${value.map(canonical).join(',')}]`
|
|
: `{${Object.keys(value).sort().map(key => `${JSON.stringify(key)}:${canonical(value[key])}`).join(',')}}`
|
|
export const digest = value => createHash('sha256').update(canonical(value)).digest('hex')
|
|
|
|
export function buildOwnerPatches(snapshots, sourceRows) {
|
|
const sources = new Map()
|
|
for (const row of sourceRows) {
|
|
const rows = sources.get(row.processInstanceId) || []
|
|
rows.push(row); sources.set(row.processInstanceId, rows)
|
|
}
|
|
return snapshots.filter(row => !own(row.formFields, 'plannerOwner')).map(row => {
|
|
if (row.formFields !== null && (typeof row.formFields !== 'object' || Array.isArray(row.formFields))) throw new Error('INVALID_FORM_FIELDS')
|
|
const source = sources.get(row.processInstanceId)
|
|
if (!source) throw new Error('SOURCE_WORKFLOW_NOT_FOUND')
|
|
const owner = selectArtPlannerOwner(source)
|
|
const after = { ...(row.formFields || {}), plannerOwner: owner }
|
|
return { workflowId: row.processInstanceId, businessId: row.businessId, before: row.formFields, beforeSqlNull: row.formFieldsSqlNull === true,
|
|
beforeHash: digest(row.formFields), owner, afterHash: digest(after) }
|
|
})
|
|
}
|
|
|
|
/** Idempotent, all-or-nothing CAS; a concurrent refresh requires a new audit. */
|
|
export async function applyOwnerPatches(db, patches, leaseOwner, restore = false) {
|
|
return db.$transaction(async tx => {
|
|
const leases = await tx.$queryRawUnsafe('SELECT lease_owner, lease_expires_at FROM ding_workflow_sync_states WHERE id=$1 FOR UPDATE', SYNC_ID)
|
|
if (leases.length !== 1 || leases[0].lease_owner !== leaseOwner || new Date(leases[0].lease_expires_at).getTime() <= Date.now()) throw new Error('SYNC_LEASE_LOST')
|
|
let changed = 0, skipped = 0
|
|
for (const patch of patches) {
|
|
const row = await tx.dingWorkflowSnapshot.findUnique({ where: { processInstanceId: patch.workflowId }, select: { formFields: true, processCode: true } })
|
|
if (!row || row.processCode !== ART_CODE) throw new Error('SNAPSHOT_TARGET_CHANGED')
|
|
if (restore) {
|
|
if (digest(row.formFields) === patch.beforeHash) { skipped++; continue }
|
|
if (digest(row.formFields) !== patch.afterHash) throw new Error('RESTORE_CONFLICT_REVIEW_MANUALLY')
|
|
} else {
|
|
// A completed hourly refresh is authoritative; never overwrite its owner.
|
|
if (own(row.formFields, 'plannerOwner')) { skipped++; continue }
|
|
if (digest(row.formFields) !== patch.beforeHash) throw new Error('SNAPSHOT_CHANGED_REAUDIT')
|
|
}
|
|
const nullState = row.formFields === null ? await tx.$queryRawUnsafe('SELECT form_fields IS NULL AS "sqlNull" FROM ding_workflow_snapshots WHERE process_instance_id=$1', patch.workflowId) : []
|
|
const previous = row.formFields === null && nullState[0]?.sqlNull ? null : JSON.stringify(row.formFields)
|
|
const next = restore ? patch.before : { ...(row.formFields || {}), plannerOwner: patch.owner }
|
|
const count = await tx.$executeRawUnsafe(`UPDATE ding_workflow_snapshots SET form_fields=$1::jsonb, updated_at=NOW()
|
|
WHERE process_instance_id=$2 AND process_code=$3 AND form_fields IS NOT DISTINCT FROM $4::jsonb`, restore && patch.beforeSqlNull ? null : JSON.stringify(next), patch.workflowId, ART_CODE, previous)
|
|
if (count !== 1) throw new Error('SNAPSHOT_CHANGED_REAUDIT')
|
|
changed++
|
|
}
|
|
return { changed, skipped }
|
|
}, { timeout: 120000 })
|
|
}
|
|
|
|
function table(schema, name) {
|
|
if (![schema, name].every(value => /^[A-Za-z_][A-Za-z0-9_]*$/.test(value))) throw new Error('INVALID_SOURCE_TABLE')
|
|
return `[${schema}].[${name}]`
|
|
}
|
|
async function readSourceRows(ids) {
|
|
if (!ids.length) return []
|
|
const sql = require('mssql')
|
|
const env = process.env
|
|
if (!['REPORT_DB_SERVER', 'REPORT_DB_DATABASE', 'REPORT_DB_USER', 'REPORT_DB_PASSWORD'].every(key => env[key])) throw new Error('REPORT_DATABASE_NOT_CONFIGURED')
|
|
const pool = new sql.ConnectionPool({ server: env.REPORT_DB_SERVER, database: env.REPORT_DB_DATABASE, user: env.REPORT_DB_USER,
|
|
password: env.REPORT_DB_PASSWORD, port: Number(env.REPORT_DB_PORT || 1433), options: { encrypt: env.REPORT_DB_ENCRYPT === 'true', trustServerCertificate: env.REPORT_DB_TRUST_SERVER_CERTIFICATE !== 'false' },
|
|
connectionTimeout: 20000, requestTimeout: 120000, pool: { max: 1, min: 0 } })
|
|
const workflow = table(env.REPORT_DB_WORKFLOW_SCHEMA || 'rpt', env.REPORT_DB_WORKFLOW_TABLE || 'ding_Workflow')
|
|
const scheme = table(env.REPORT_DB_WORKFLOW_SCHEMA || 'rpt', env.REPORT_DB_WORKFLOW_SCHEME_TABLE || 'ding_WorkflowScheme')
|
|
try {
|
|
await pool.connect()
|
|
const result = await pool.request().input('ids', sql.NVarChar(sql.MAX), JSON.stringify(ids)).input('code', sql.NVarChar, ART_CODE).query(`
|
|
SELECT CONVERT(nvarchar(255),w.processInstanceId) AS processInstanceId,CONVERT(nvarchar(64),s.id) AS sourceRowId,s.name,
|
|
CONVERT(nvarchar(max),s.[value]) AS value,CONVERT(nvarchar(max),s.extValue) AS extValue
|
|
FROM ${workflow} w INNER JOIN OPENJSON(@ids) ids ON CONVERT(nvarchar(255),w.processInstanceId)=CONVERT(nvarchar(255),ids.[value])
|
|
LEFT JOIN ${scheme} s ON s.processInstanceId=w.processInstanceId AND s.name=N'策划案负责人'
|
|
WHERE w.processCode=@code`)
|
|
return result.recordset
|
|
} finally { await pool.close() }
|
|
}
|
|
|
|
async function assertTarget(db, writing) {
|
|
const url = new URL(process.env.DATABASE_URL || '')
|
|
if (process.env.NODE_ENV !== 'production') {
|
|
if (!['localhost', '127.0.0.1'].includes(url.hostname) || url.port !== '55439' || !/^\/salary_isolated_/.test(url.pathname)) throw new Error('ISOLATED_DATABASE_REQUIRED')
|
|
return { commit: 'local', database: { host: url.hostname, port: url.port, name: url.pathname.slice(1) } }
|
|
}
|
|
const { requireExactCommit, assertExpectedDatabaseIdentity, assertBackupReceipt } = await import('./safe-production-migrate.mjs')
|
|
const commit = requireExactCommit(process.env.EXPECTED_COMMIT_SHA, 'EXPECTED_COMMIT_SHA')
|
|
const root = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '..')
|
|
if ((await readFile(path.join(root, 'DEPLOYMENT_COMMIT'), 'utf8')).trim() !== commit) throw new Error('IMAGE_COMMIT_MISMATCH')
|
|
const identity = await db.$queryRawUnsafe('SELECT current_database() AS database,current_user AS "user",inet_server_addr()::text AS server_address,inet_server_port()::text AS server_port')
|
|
const database = assertExpectedDatabaseIdentity(url, identity[0])
|
|
if (writing) {
|
|
if (process.env.ALLOW_ART_OWNER_BACKFILL !== 'PATCH_MISSING_PLANNER_OWNER') throw new Error('EXPLICIT_WRITE_FLAG_REQUIRED')
|
|
await assertBackupReceipt({ expectedCommit: commit, database })
|
|
}
|
|
return { commit, database }
|
|
}
|
|
|
|
async function privateWrite(file, value) {
|
|
const parent = await lstat(path.dirname(file))
|
|
if (!parent.isDirectory() || parent.isSymbolicLink() || (parent.mode & 0o077) !== 0) throw new Error('PRIVATE_OUTPUT_DIRECTORY_REQUIRED')
|
|
await writeFile(file, JSON.stringify(value, null, 2) + '\n', { flag: 'wx', mode: 0o600 })
|
|
}
|
|
|
|
async function main(args) {
|
|
const modes = args.filter(arg => ['--audit', '--apply', '--restore'].includes(arg))
|
|
if (modes.length !== 1 || args.some(arg => !['--audit', '--apply', '--restore'].includes(arg) && !/^--(?:output|manifest)=\/.+/.test(arg))) throw new Error('USE_AUDIT_APPLY_OR_RESTORE_WITH_ABSOLUTE_PATH')
|
|
const mode = modes[0], output = args.find(arg => arg.startsWith('--output='))?.slice(9), manifestFile = args.find(arg => arg.startsWith('--manifest='))?.slice(11)
|
|
const { PrismaClient } = require('@prisma/client')
|
|
const db = new PrismaClient({ log: [] })
|
|
let leaseOwner
|
|
try {
|
|
const target = await assertTarget(db, mode !== '--audit')
|
|
if (mode === '--audit') {
|
|
if (!output || manifestFile) throw new Error('AUDIT_OUTPUT_REQUIRED')
|
|
const snapshots = await db.$queryRawUnsafe('SELECT process_instance_id AS "processInstanceId", business_id AS "businessId", form_fields AS "formFields", form_fields IS NULL AS "formFieldsSqlNull" FROM ding_workflow_snapshots WHERE process_code=$1 ORDER BY process_instance_id', ART_CODE)
|
|
const missing = snapshots.filter(row => !own(row.formFields, 'plannerOwner'))
|
|
const patches = buildOwnerPatches(missing, await readSourceRows(missing.map(row => row.processInstanceId)))
|
|
const people = await db.performanceSurveyEmployeeProfile.findMany({ select: { dingtalkUserId: true, realName: true, flowerName: true, displayName: true } })
|
|
const unresolved = patches.filter(patch => {
|
|
if (patch.owner === null) return false
|
|
let contacts; try { contacts = JSON.parse(patch.owner) } catch { /* name-only source */ }
|
|
if (Array.isArray(contacts)) return contacts.length !== 1 || !people.some(person => person.dingtalkUserId === contacts[0].emplId)
|
|
return people.filter(person => [person.realName, person.flowerName, person.displayName].includes(patch.owner)).length !== 1
|
|
}).map(patch => patch.workflowId)
|
|
const manifest = { schemaVersion: 1, target, generatedAt: new Date().toISOString(), total: snapshots.length, unresolved, patches }
|
|
await privateWrite(output, manifest)
|
|
console.log(JSON.stringify({ mode, total: snapshots.length, missing: patches.length, explicitEmpty: patches.filter(patch => patch.owner === null).length, unresolved: unresolved.length,
|
|
manifestSha256: createHash('sha256').update(await readFile(output)).digest('hex') }))
|
|
return
|
|
}
|
|
if (!manifestFile || output) throw new Error('APPLY_MANIFEST_REQUIRED')
|
|
const stat = await lstat(manifestFile)
|
|
if (!stat.isFile() || stat.isSymbolicLink() || (stat.mode & 0o077) !== 0) throw new Error('PRIVATE_MANIFEST_REQUIRED')
|
|
const bytes = await readFile(manifestFile)
|
|
if (createHash('sha256').update(bytes).digest('hex') !== process.env.EXPECTED_OWNER_MANIFEST_SHA256) throw new Error('MANIFEST_HASH_MISMATCH')
|
|
const manifest = JSON.parse(bytes.toString('utf8'))
|
|
if (manifest.schemaVersion !== 1 || digest(manifest.target) !== digest(target) || !Array.isArray(manifest.patches)) throw new Error('MANIFEST_TARGET_MISMATCH')
|
|
for (const patch of manifest.patches) {
|
|
if (own(patch.before, 'plannerOwner') || digest(patch.before) !== patch.beforeHash || digest({ ...(patch.before || {}), plannerOwner: patch.owner }) !== patch.afterHash) throw new Error('INVALID_PATCH_MANIFEST')
|
|
}
|
|
leaseOwner = `planner-owner-backfill:${process.pid}:${randomUUID()}`
|
|
await db.dingWorkflowSyncState.upsert({ where: { id: SYNC_ID }, create: { id: SYNC_ID }, update: {} })
|
|
const claimed = await db.dingWorkflowSyncState.updateMany({ where: { id: SYNC_ID, OR: [{ leaseExpiresAt: null }, { leaseExpiresAt: { lt: new Date() } }] }, data: { leaseOwner, leaseExpiresAt: new Date(Date.now() + 20 * 60 * 1000) } })
|
|
if (claimed.count !== 1) throw new Error('SYNC_BUSY_RETRY_AFTER_COMPLETION')
|
|
if (mode === '--apply') {
|
|
const source = await readSourceRows(manifest.patches.map(patch => patch.workflowId))
|
|
const refreshed = buildOwnerPatches(manifest.patches.map(patch => ({ processInstanceId: patch.workflowId, businessId: patch.businessId, formFields: patch.before })), source)
|
|
if (refreshed.some((patch, i) => patch.afterHash !== manifest.patches[i].afterHash)) throw new Error('SOURCE_CHANGED_REAUDIT')
|
|
}
|
|
const result = await applyOwnerPatches(db, manifest.patches, leaseOwner, mode === '--restore')
|
|
await privateWrite(`${manifestFile}.${mode.slice(2)}.${randomUUID()}.json`, { ...result, manifestSha256: process.env.EXPECTED_OWNER_MANIFEST_SHA256, completedAt: new Date().toISOString() })
|
|
console.log(JSON.stringify({ mode, ...result }))
|
|
} finally {
|
|
if (leaseOwner) await db.dingWorkflowSyncState.updateMany({ where: { id: SYNC_ID, leaseOwner }, data: { leaseOwner: null, leaseExpiresAt: null } }).catch(() => {})
|
|
await db.$disconnect()
|
|
}
|
|
}
|
|
|
|
if (process.argv[1] && import.meta.url === pathToFileURL(path.resolve(process.argv[1])).href) {
|
|
main(process.argv.slice(2)).catch(error => {
|
|
// Bound data, SQL, credentials and personal records must not enter stdout.
|
|
console.error(`负责人维护未完成：${/^[A-Z_]+$/.test(error.message) ? error.message : 'VERIFY_DATABASE_SOURCE_BACKUP_AND_PRIVATE_AUDIT'}`)
|
|
process.exitCode = 1
|
|
})
|
|
}
|