yml2213
1217da7393
1. 扩展商品匹配配置支持 maxMatchLogs (条数上限) 与 matchLogRetentionDays (保留天数)； 2. 新增分批物理清理逻辑 pruneWorkProductMatchLogs，通过索引高效分批删除超出限制的记录； 3. 集成定时任务数据保留清理 data-retention-job，并在保存配置与写入时自动滚动清理； 4. 匹配诊断日志面板新增可视化保留策略设置栏，支持管理员自定义上限、一键保存并提供一键清理超出记录按钮。
113 lines
3.6 KiB
TypeScript
113 lines
3.6 KiB
TypeScript
import { query } from "../../db/client.js"
|
|
import { runtimeConfig } from "../../config/runtime.js"
|
|
import { getWorkerProductMatchConfig } from "../worker-platform/worker-product-match-config-service.js"
|
|
import { pruneWorkProductMatchLogs } from "../../repositories/worker-platform/product-match-repo.js"
|
|
|
|
export type DataRetentionResult = {
|
|
status: "ok"
|
|
message: string
|
|
rawPayloadRows: number
|
|
taskEventRows: number
|
|
webhookEventRows: number
|
|
auditLogRows: number
|
|
matchLogRows: number
|
|
matchLogDeletedRows: number
|
|
}
|
|
|
|
/** 清理敏感原始报文并删除已过保留期的事件/审计记录。 */
|
|
export async function runDataRetentionJob(): Promise<DataRetentionResult> {
|
|
const retention = runtimeConfig.retention || {}
|
|
const rawPayloadDays = positiveDays(retention.rawPayloadDays, 180)
|
|
const taskEventDays = positiveDays(retention.taskEventDays, 180)
|
|
const webhookEventDays = positiveDays(retention.webhookEventDays, 180)
|
|
const auditLogDays = positiveDays(retention.auditLogDays, 365)
|
|
|
|
// 1. 按配置清理超出上限或过期的打手商品匹配日志
|
|
const matchConfig = getWorkerProductMatchConfig()
|
|
const matchLogPruneResult = await pruneWorkProductMatchLogs({
|
|
maxRows: matchConfig.maxMatchLogs,
|
|
retentionDays: matchConfig.matchLogRetentionDays,
|
|
})
|
|
|
|
// 2. 清理其他报文与日志
|
|
const rawPayloadRows =
|
|
(await scrubJsonColumnInBatches("orders", "updated_at", rawPayloadDays)) +
|
|
(await scrubJsonColumnInBatches("kuaishou_industry_vouchers", "updated_at", rawPayloadDays))
|
|
const matchLogRows = await scrubJsonColumnInBatches(
|
|
"work_product_match_logs",
|
|
"created_at",
|
|
rawPayloadDays,
|
|
)
|
|
const taskEventRows = await deleteOldRowsInBatches("task_events", taskEventDays)
|
|
const webhookEventRows = await deleteOldRowsInBatches("webhook_events", webhookEventDays)
|
|
const auditLogRows = await deleteOldRowsInBatches("admin_audit_logs", auditLogDays)
|
|
|
|
return {
|
|
status: "ok",
|
|
message: "数据保留清理完成",
|
|
rawPayloadRows,
|
|
matchLogRows,
|
|
matchLogDeletedRows: matchLogPruneResult.totalDeleted,
|
|
taskEventRows,
|
|
webhookEventRows,
|
|
auditLogRows,
|
|
}
|
|
}
|
|
async function scrubJsonColumnInBatches(
|
|
table: 'orders' | 'kuaishou_industry_vouchers' | 'work_product_match_logs',
|
|
dateColumn: 'created_at' | 'updated_at',
|
|
retentionDays: number,
|
|
) {
|
|
let total = 0
|
|
while (true) {
|
|
const result = await query(
|
|
`
|
|
WITH candidates AS (
|
|
SELECT ctid
|
|
FROM ${table}
|
|
WHERE ${dateColumn} < NOW() - make_interval(days => $1::int)
|
|
AND raw_payload_json <> '{}'::jsonb
|
|
LIMIT 1000
|
|
)
|
|
UPDATE ${table} AS target
|
|
SET raw_payload_json = '{}'::jsonb
|
|
FROM candidates
|
|
WHERE target.ctid = candidates.ctid
|
|
`,
|
|
[retentionDays],
|
|
)
|
|
total += result.rowCount || 0
|
|
if ((result.rowCount || 0) < 1000) return total
|
|
}
|
|
}
|
|
|
|
async function deleteOldRowsInBatches(
|
|
table: 'task_events' | 'webhook_events' | 'admin_audit_logs',
|
|
retentionDays: number,
|
|
) {
|
|
let total = 0
|
|
while (true) {
|
|
const result = await query(
|
|
`
|
|
WITH candidates AS (
|
|
SELECT ctid
|
|
FROM ${table}
|
|
WHERE created_at < NOW() - make_interval(days => $1::int)
|
|
LIMIT 1000
|
|
)
|
|
DELETE FROM ${table} AS target
|
|
USING candidates
|
|
WHERE target.ctid = candidates.ctid
|
|
`,
|
|
[retentionDays],
|
|
)
|
|
total += result.rowCount || 0
|
|
if ((result.rowCount || 0) < 1000) return total
|
|
}
|
|
}
|
|
|
|
function positiveDays(value: unknown, fallback: number) {
|
|
const days = Number(value)
|
|
return Number.isInteger(days) && days > 0 ? days : fallback
|
|
}
|