yml/order_site
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(ops): 数据保留清理、CORS 白名单与上传失败回滚

Browse Source 
- 新增每日数据保留清理任务:按保留期清空原始报文、删除过期
  事件/审计记录,批量幂等可重跑;067 迁移补齐清理索引并清理
  冗余索引,check-sql-guard 增加防线
- 生产 CORS 改为显式来源白名单(CORS_ALLOWED_ORIGINS),禁用
  通配符,非通配响应补充 Vary: Origin
- 文件上传任一步失败时回滚已上传对象,清理失败记录 warn 日志
- 新增 HTTP/CORS 与 PostgreSQL 集成测试;若干文件仅 prettier 重排
This commit is contained in:
yml2213
2026-08-30 13:59:01 +08:00
parent dcb0fa4c76
commit 02c9de8b9c
25 changed files with 489 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/scripts/check-sql-guard.ts
+11
View File
@@ -34,6 +34,17 @@ const checks: Array<{ name: string; file: string; patterns: string[] }> = [
file: path.join(BACKEND_ROOT, 'src/repositories/worker-platform/work-order-share-repo.ts'),
patterns: ['workOrderIds: number[]', 'wos.work_order_id = ANY($2::bigint[])'],
},
{
name: '数据保留清理扫描索引',
file: path.join(BACKEND_ROOT, 'src/db/migrations/067_data_retention_indexes.sql'),
patterns: [
'idx_orders_updated_at',
'idx_kuaishou_industry_vouchers_updated_at',
'idx_task_events_created_at',
'idx_webhook_events_created_at',
'DROP INDEX IF EXISTS idx_work_product_match_logs_created_at',
],
},
]
const failures: string[] = []
apps/backend/src/app.integration.test.ts
+40
View File
@@ -0,0 +1,40 @@
import assert from 'node:assert/strict'
import { once } from 'node:events'
import test from 'node:test'
import { createApp } from './app.js'
import { createDefaultRuntimeConfig } from './config/defaults.js'
import { createStartupState } from './startup/state.js'
test('HTTP health route and explicit CORS preflight', async () => {
const startupState = createStartupState()
startupState.core.ready = true
const config = createDefaultRuntimeConfig('/tmp/order-site')
config.cors.allowedOrigins = ['https://order.example.test']
const server = createApp({
startupState,
isShutdownStarted: () => false,
config,
}).listen(0, '127.0.0.1')
await once(server, 'listening')
try {
const address = server.address()
assert.ok(address && typeof address === 'object')
const baseUrl = `http://127.0.0.1:${address.port}`
const health = await fetch(`${baseUrl}/health/live`)
assert.equal(health.status, 200)
assert.equal((await health.json()).data.status, 'alive')
const preflight = await fetch(`${baseUrl}/api/v1/health`, {
method: 'OPTIONS',
headers: { Origin: 'https://order.example.test' },
})
assert.equal(preflight.status, 204)
assert.equal(preflight.headers.get('access-control-allow-origin'), 'https://order.example.test')
assert.equal(preflight.headers.get('vary'), 'Origin')
} finally {
server.close()
await once(server, 'close')
}
})
apps/backend/src/config/defaults.ts
+8
View File
@@ -21,6 +21,14 @@ export function createDefaultRuntimeConfig(projectRoot: string): RuntimeConfig {
retentionDays: 30,
},
retention: {
// 原始报文只保留追溯所需窗口，清理后保留订单/事件摘要。
rawPayloadDays: 180,
taskEventDays: 180,
webhookEventDays: 180,
auditLogDays: 365,
},
database: {
url: 'postgres://postgres:postgres@127.0.0.1:5432/order_site',
ssl: false,
apps/backend/src/config/env-documentation.test.ts
+10
View File
@@ -27,6 +27,16 @@ const DEPLOYMENT_ONLY_ENV_NAMES = [
'NPM_CONFIG_REGISTRY',
'POSTGRES_DB',
'POSTGRES_LOG_MIN_DURATION_STATEMENT_MS',
'POSTGRES_MEM_LIMIT',
'POSTGRES_CPUS',
'POSTGRES_SHARED_BUFFERS',
'POSTGRES_EFFECTIVE_CACHE_SIZE',
'POSTGRES_WORK_MEM',
'POSTGRES_MAINTENANCE_WORK_MEM',
'BACKEND_MEM_LIMIT',
'BACKEND_CPUS',
'WEB_MEM_LIMIT',
'WEB_CPUS',
'POSTGRES_PASSWORD',
'POSTGRES_PORT',
'POSTGRES_USER',
apps/backend/src/config/env-overrides.ts
+4
View File
@@ -33,6 +33,10 @@ export const ENV_OVERRIDES: readonly EnvOverride[] = [
stringEnv('LOG_LEVEL', ['logging', 'level']),
stringEnv('LOG_INTEGRATION_LEVEL', ['logging', 'integrationLevel']),
integerEnv('LOG_RETENTION_DAYS', ['logging', 'retentionDays']),
integerEnv('RAW_PAYLOAD_RETENTION_DAYS', ['retention', 'rawPayloadDays']),
integerEnv('TASK_EVENT_RETENTION_DAYS', ['retention', 'taskEventDays']),
integerEnv('WEBHOOK_EVENT_RETENTION_DAYS', ['retention', 'webhookEventDays']),
integerEnv('AUDIT_LOG_RETENTION_DAYS', ['retention', 'auditLogDays']),
stringEnv('DATABASE_URL', ['database', 'url']),
booleanEnv('DATABASE_SSL', ['database', 'ssl']),
integerEnv('DATABASE_MAX_CONNECTIONS', ['database', 'maxConnections']),
apps/backend/src/config/runtime-validation.test.ts
+11
View File
@@ -77,6 +77,17 @@ test('validateRuntimeConfig accepts complete production config', () => {
assert.deepEqual(issues, [])
})
test('validateRuntimeConfig rejects wildcard CORS in production', () => {
const issues = validateRuntimeConfig(createRuntimeConfig({ cors: { allowedOrigins: ['*'] } }), {
env: { NODE_ENV: 'production' },
})
assert.deepEqual(
issues.map((issue) => issue.path),
['cors.allowedOrigins'],
)
})
test('validateRuntimeConfig catches invalid numeric and url values', () => {
const issues = validateRuntimeConfig(
createRuntimeConfig({
apps/backend/src/config/runtime-validation.ts
+26
View File
@@ -79,6 +79,14 @@ export function validateRuntimeConfig(
min: 1,
max: 100,
})
if (config.retention) {
requireInteger(issues, 'retention.rawPayloadDays', config.retention.rawPayloadDays, { min: 1 })
requireInteger(issues, 'retention.taskEventDays', config.retention.taskEventDays, { min: 1 })
requireInteger(issues, 'retention.webhookEventDays', config.retention.webhookEventDays, {
min: 1,
})
requireInteger(issues, 'retention.auditLogDays', config.retention.auditLogDays, { min: 1 })
}
const storageMode = normalizeStorageMode(config.storage?.mode)
if (!storageMode) {
issues.push({
@@ -137,6 +145,24 @@ export function validateRuntimeConfig(
validateRequiredHttpUrl(issues, 'orders.claimBaseUrl', config.orders?.claimBaseUrl)
validateSecret(issues, 'admin.sessionSecret', config.admin?.sessionSecret, { minLength: 32 })
validateAdminDefaultUsers(issues, config.admin?.defaultUsers)
const allowedOrigins = config.cors?.allowedOrigins
if (Array.isArray(allowedOrigins)) {
if (allowedOrigins.length === 0 || allowedOrigins.includes('*')) {
issues.push({
path: 'cors.allowedOrigins',
message: '生产环境必须配置显式 CORS 域名白名单，不能使用 *',
})
} else {
allowedOrigins.forEach((origin, index) => {
if (!isHttpUrl(String(origin || '').trim())) {
issues.push({
path: `cors.allowedOrigins[${index}]`,
message: '必须是 http 或 https 来源 URL',
})
}
})
}
}
}
return issues
apps/backend/src/db/migrations/067_data_retention_indexes.sql
+20
View File
@@ -0,0 +1,20 @@
-- 067_data_retention_indexes.sql —— 数据保留清理所需索引。
CREATE INDEX IF NOT EXISTS idx_webhook_events_created_at
ON webhook_events(created_at DESC);
CREATE INDEX IF NOT EXISTS idx_orders_updated_at
ON orders(updated_at DESC);
CREATE INDEX IF NOT EXISTS idx_kuaishou_industry_vouchers_updated_at
ON kuaishou_industry_vouchers(updated_at DESC);
CREATE INDEX IF NOT EXISTS idx_task_events_created_at
ON task_events(created_at DESC);
DROP INDEX IF EXISTS idx_work_product_match_logs_created_at;
COMMENT ON INDEX idx_webhook_events_created_at IS '支持 webhook 回调去重数据按创建时间清理';
COMMENT ON INDEX idx_orders_updated_at IS '支持订单原始报文按更新时间清理';
COMMENT ON INDEX idx_kuaishou_industry_vouchers_updated_at IS '支持电子凭证原始报文按更新时间清理';
COMMENT ON INDEX idx_task_events_created_at IS '支持任务事件按创建时间清理';
apps/backend/src/db/postgres.integration.test.ts
+60
View File
@@ -0,0 +1,60 @@
import assert from 'node:assert/strict'
import test from 'node:test'
import { Pool } from 'pg'
const connectionString = process.env.TEST_POSTGRES_URL
test('real PostgreSQL transaction rolls back atomically', { skip: !connectionString }, async () => {
const pool = new Pool({ connectionString })
const client = await pool.connect()
const tableName = `integration_tx_${Date.now()}`
try {
await client.query(`CREATE TEMP TABLE ${tableName} (id integer primary key, value text)`)
await client.query('BEGIN')
await client.query(`INSERT INTO ${tableName} (id, value) VALUES (1, 'before-error')`)
try {
await client.query(`INSERT INTO ${tableName} (id, value) VALUES (1, 'duplicate')`)
} catch {
await client.query('ROLLBACK')
}
const result = await client.query(`SELECT COUNT(*)::int AS count FROM ${tableName}`)
assert.equal(result.rows[0].count, 0)
} finally {
client.release()
await pool.end()
}
})
test(
'real PostgreSQL enforces callback replay idempotency under concurrency',
{
skip: !connectionString,
},
async () => {
const pool = new Pool({ connectionString })
const tableName = `integration_webhook_${Date.now()}`
try {
await pool.query(
`CREATE TEMP TABLE ${tableName} (
id serial primary key,
provider text not null,
event_id text not null,
unique(provider, event_id)
)`,
)
await Promise.all(
Array.from({ length: 8 }, () =>
pool.query(
`INSERT INTO ${tableName} (provider, event_id)
VALUES ('integration', 'replay-1')
ON CONFLICT (provider, event_id) DO NOTHING`,
),
),
)
const result = await pool.query(`SELECT COUNT(*)::int AS count FROM ${tableName}`)
assert.equal(result.rows[0].count, 1)
} finally {
await pool.end()
}
},
)
apps/backend/src/middleware/cors.ts
+1
View File
@@ -11,6 +11,7 @@ export function createCorsMiddleware(
if (isWildcard) {
res.setHeader('Access-Control-Allow-Origin', '*')
} else {
res.setHeader('Vary', 'Origin')
const origin = req.headers.origin
if (origin && allowedOrigins.includes(origin)) {
res.setHeader('Access-Control-Allow-Origin', origin)
apps/backend/src/repositories/worker-platform/work-order-settlement-repo.ts
+3 -1
View File
@@ -837,7 +837,9 @@ export async function cancelWorkOrderSharing(input: {
}
}
const cancellableShares = shares.filter((share) => ['joined', 'submitted'].includes(share.status))
const cancellableShares = shares.filter((share) =>
['joined', 'submitted'].includes(share.status),
)
let releasedDepositAmount = 0
for (const share of cancellableShares) {
const workerId = Number(share.worker_id || 0)
apps/backend/src/repositories/worker-platform/work-order-share-repo.ts
+2 -1
View File
@@ -108,7 +108,8 @@ export function resolveWorkOrderShareDepositAmount(input: {
}) {
const totalQuantity = Math.max(1, Number(input.totalQuantity || 0))
const nextRequiredDeposit = Math.round(
(Math.max(0, Number(input.requiredDepositAmount || 0)) * Math.max(0, Number(input.nextQuantity || 0))) /
(Math.max(0, Number(input.requiredDepositAmount || 0)) *
Math.max(0, Number(input.nextQuantity || 0))) /
totalQuantity,
)
const requiredShareDeposit = Math.max(
apps/backend/src/services/file-storage/file-storage-service.ts
+86 -61
View File
@@ -8,6 +8,7 @@ import { createFileAsset, type FileAssetRow } from '../../repositories/file-asse
import { createHttpError } from '../../utils/http.js'
import { nowIso } from '../../utils/time.js'
import { getObjectStorage, normalizeObjectKey } from './object-storage.js'
import { logWarn } from '../../utils/logger.js'
const IMAGE_VARIANT_THUMB = 'thumb'
const IMAGE_VARIANT_MEDIUM = 'medium'
@@ -64,74 +65,98 @@ export async function uploadFileAsset(input: UploadFileAssetInput) {
const scene = normalizeScene(input.scene)
const objectKey = createObjectKey(scene, file.originalname, contentType)
const storage = getObjectStorage()
await storage.putObject({
key: objectKey,
content: file.buffer,
contentType,
metadata: {
'original-filename': file.originalname,
},
})
const uploadedKeys: string[] = []
try {
uploadedKeys.push(objectKey)
await storage.putObject({
key: objectKey,
content: file.buffer,
contentType,
metadata: {
'original-filename': file.originalname,
},
})
const variants = await generateImageVariants(objectKey, file.buffer, contentType)
let thumbnailObjectKey = ''
let mediumObjectKey = ''
const variants = await generateImageVariants(objectKey, file.buffer, contentType)
let thumbnailObjectKey = ''
let mediumObjectKey = ''
for (const variant of variants) {
try {
await storage.putObject({
key: variant.key,
content: variant.content,
contentType: variant.contentType,
metadata: {
'source-object': objectKey,
},
})
if (variant.name === IMAGE_VARIANT_THUMB) {
thumbnailObjectKey = variant.key
for (const variant of variants) {
try {
uploadedKeys.push(variant.key)
await storage.putObject({
key: variant.key,
content: variant.content,
contentType: variant.contentType,
metadata: {
'source-object': objectKey,
},
})
if (variant.name === IMAGE_VARIANT_THUMB) {
thumbnailObjectKey = variant.key
}
if (variant.name === IMAGE_VARIANT_MEDIUM) {
mediumObjectKey = variant.key
}
} catch {
// 缩略图失败不阻断原图上传；预览接口会回退到原图。
}
if (variant.name === IMAGE_VARIANT_MEDIUM) {
mediumObjectKey = variant.key
}
} catch {
// 缩略图失败不阻断原图上传；预览接口会回退到原图。
}
}
const now = nowIso()
const fileUrl = buildUnsignedFileUrl(objectKey)
const thumbnailUrl = thumbnailObjectKey ? buildUnsignedFileUrl(thumbnailObjectKey) : fileUrl
const mediumUrl = mediumObjectKey ? buildUnsignedFileUrl(mediumObjectKey) : fileUrl
const asset = await createFileAsset({
objectKey,
scene,
originalFilename: file.originalname,
contentType,
sizeBytes: file.buffer.length,
thumbnailObjectKey,
mediumObjectKey,
url: fileUrl,
thumbnailUrl,
mediumUrl,
uploaderType: String(input.uploaderType || '').trim(),
uploaderId: String(input.uploaderId || '').trim(),
metadataJson: JSON.stringify({
originalMimeType: file.mimetype,
originalSize: file.size,
}),
now,
})
return {
file: mapUploadedFile(asset, {
const now = nowIso()
const fileUrl = buildUnsignedFileUrl(objectKey)
const thumbnailUrl = thumbnailObjectKey ? buildUnsignedFileUrl(thumbnailObjectKey) : fileUrl
const mediumUrl = mediumObjectKey ? buildUnsignedFileUrl(mediumObjectKey) : fileUrl
const asset = await createFileAsset({
objectKey,
fileUrl,
scene,
originalFilename: file.originalname,
contentType,
sizeBytes: file.buffer.length,
thumbnailObjectKey,
mediumObjectKey,
url: fileUrl,
thumbnailUrl,
mediumUrl,
filename: file.originalname,
contentType,
size: file.buffer.length,
}),
uploaderType: String(input.uploaderType || '').trim(),
uploaderId: String(input.uploaderId || '').trim(),
metadataJson: JSON.stringify({
originalMimeType: file.mimetype,
originalSize: file.size,
}),
now,
})
if (!asset) {
throw new Error('文件元数据写入失败')
}
return {
file: mapUploadedFile(asset, {
objectKey,
fileUrl,
thumbnailUrl,
mediumUrl,
filename: file.originalname,
contentType,
size: file.buffer.length,
}),
}
} catch (error) {
const cleanupResults = await Promise.allSettled(
uploadedKeys.map((key) => storage.deleteObject(key)),
)
const cleanupFailures = cleanupResults.filter(
(result): result is PromiseRejectedResult => result.status === 'rejected',
)
if (cleanupFailures.length > 0) {
void logWarn('[file-storage]', '上传失败后的对象清理未完全成功', {
objectKeys: uploadedKeys,
failedCount: cleanupFailures.length,
errors: cleanupFailures.map((result) =>
result.reason instanceof Error ? result.reason.message : String(result.reason || ''),
),
})
}
throw error
}
}
apps/backend/src/services/file-storage/object-storage.ts
+20
View File
@@ -2,6 +2,7 @@ import { Readable } from 'node:stream'
import {
CreateBucketCommand,
DeleteObjectCommand,
GetObjectCommand,
HeadBucketCommand,
HeadObjectCommand,
@@ -88,6 +89,13 @@ export class ObjectStorage {
)
}
async deleteObject(key: string): Promise<void> {
await this.ensureBucketReady()
await this.client.send(
new DeleteObjectCommand({ Bucket: this.bucket, Key: normalizeObjectKey(key) }),
)
}
async getObject(key: string): Promise<StoredObject> {
await this.ensureBucketReady()
const objectKey = normalizeObjectKey(key)
@@ -233,6 +241,18 @@ export class StorageRouter {
return this.oss!.getObject(key)
}
}
async deleteObject(key: string): Promise<void> {
if (this.mode === 'minio') {
await this.legacy.deleteObject(key)
return
}
if (this.mode === 'oss') {
await this.oss!.deleteObject(key)
return
}
await Promise.allSettled([this.legacy.deleteObject(key), this.oss!.deleteObject(key)])
}
}
let storageInstance: StorageRouter | null = null
apps/backend/src/services/scheduler/config-service.test.ts
+13
View File
@@ -92,3 +92,16 @@ test('normalizeScheduledJobsConfig 支持多账号去重并保留 0 阈值', ()
assert.equal(config.jobs[0].intervalSeconds, 60)
assert.equal(config.jobs[0].cooldownSeconds, 60)
})
test('normalizeScheduledJobsConfig 自动补齐数据保留清理任务', () => {
const config = normalizeScheduledJobsConfig({ enabled: true, jobs: [] })
const retentionJob = config.jobs.find((job) => job.type === 'data_retention')
assert.deepEqual(retentionJob, {
id: 'data-retention',
type: 'data_retention',
enabled: true,
intervalSeconds: 86400,
config: {},
})
})
apps/backend/src/services/scheduler/config-service.ts
+29
View File
@@ -9,6 +9,7 @@ const SCHEDULED_JOBS_FILE_PATH = path.join(PROJECT_ROOT, 'data', 'scheduled-jobs
const CLOUDTENTACLES_HEALTH_JOB_ID = 'cloudtentacles-health'
const WORK_ORDER_TIMEOUT_JOB_ID = 'work-order-timeout'
const DEPOSIT_UNFREEZE_JOB_ID = 'deposit-unfreeze'
const DATA_RETENTION_JOB_ID = 'data-retention'
export function getScheduledJobsFilePath() {
return SCHEDULED_JOBS_FILE_PATH
@@ -73,6 +74,7 @@ export function normalizeScheduledJobsConfig(rawValue: unknown) {
const hasCloudtentaclesHealth = jobs.some((item) => item.id === CLOUDTENTACLES_HEALTH_JOB_ID)
const hasWorkOrderTimeout = jobs.some((item) => item.id === WORK_ORDER_TIMEOUT_JOB_ID)
const hasDepositUnfreeze = jobs.some((item) => item.id === DEPOSIT_UNFREEZE_JOB_ID)
const hasDataRetention = jobs.some((item) => item.id === DATA_RETENTION_JOB_ID)
if (!hasCloudtentaclesHealth) {
jobs.push(createDefaultCloudtentaclesHealthJob())
@@ -83,6 +85,9 @@ export function normalizeScheduledJobsConfig(rawValue: unknown) {
if (!hasDepositUnfreeze) {
jobs.push(createDefaultDepositUnfreezeJob())
}
if (!hasDataRetention) {
jobs.push(createDefaultDataRetentionJob())
}
return {
enabled: typeof source.enabled === 'boolean' ? source.enabled : true,
@@ -105,6 +110,9 @@ function normalizeScheduledJob(rawValue: unknown) {
if (type === 'deposit_unfreeze') {
return normalizeDepositUnfreezeJob(rawValue)
}
if (type === 'data_retention') {
return normalizeDataRetentionJob(rawValue)
}
return null
}
@@ -123,6 +131,16 @@ function normalizeDepositUnfreezeJob(rawValue: JsonObject) {
}
}
function normalizeDataRetentionJob(rawValue: JsonObject) {
return {
id: DATA_RETENTION_JOB_ID,
type: 'data_retention',
enabled: rawValue.enabled !== false,
intervalSeconds: normalizeRangeInteger(rawValue.intervalSeconds, 86400, 3600, 604800),
config: {},
}
}
function normalizeWorkOrderTimeoutJob(rawValue: JsonObject) {
const config = isPlainObject(rawValue.config) ? rawValue.config : {}
@@ -211,10 +229,21 @@ function createDefaultScheduledJobsConfig() {
createDefaultCloudtentaclesHealthJob(),
createDefaultWorkOrderTimeoutJob(),
createDefaultDepositUnfreezeJob(),
createDefaultDataRetentionJob(),
],
}
}
function createDefaultDataRetentionJob() {
return {
id: DATA_RETENTION_JOB_ID,
type: 'data_retention',
enabled: true,
intervalSeconds: 86400,
config: {},
}
}
function createDefaultDepositUnfreezeJob() {
return {
id: DEPOSIT_UNFREEZE_JOB_ID,
apps/backend/src/services/scheduler/data-retention-job.ts
+101
View File
@@ -0,0 +1,101 @@
import { query } from '../../db/client.js'
import { runtimeConfig } from '../../config/runtime.js'
export type DataRetentionResult = {
status: 'ok'
message: string
rawPayloadRows: number
taskEventRows: number
webhookEventRows: number
auditLogRows: number
matchLogRows: 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)
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,
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
}
apps/backend/src/services/scheduler/scheduler-service.ts
+13
View File
@@ -5,6 +5,7 @@ import { getScheduledJobById, getScheduledJobsConfig } from './config-service.js
import { runCloudtentaclesHealthJob } from './cloudtentacles-health-job.js'
import { runDepositUnfreezeJob } from './deposit-unfreeze-job.js'
import { runWorkOrderTimeoutJob } from './work-order-timeout-job.js'
import { runDataRetentionJob } from './data-retention-job.js'
const timers = new Map<string, NodeJS.Timeout>()
const jobStates = new Map<string, JsonObject>()
@@ -150,6 +151,15 @@ async function runJob(job: JsonObject, { manual = false }: { manual?: boolean }
jobId,
type: job.type,
status: String(summary.status || 'ok'),
...(job.type === 'data_retention'
? {
rawPayloadRows: Number(summary.rawPayloadRows || 0),
matchLogRows: Number(summary.matchLogRows || 0),
taskEventRows: Number(summary.taskEventRows || 0),
webhookEventRows: Number(summary.webhookEventRows || 0),
auditLogRows: Number(summary.auditLogRows || 0),
}
: {}),
})
return result
} catch (error) {
@@ -184,6 +194,9 @@ function dispatchJob(job: JsonObject) {
if (job.type === 'deposit_unfreeze') {
return runDepositUnfreezeJob(job)
}
if (job.type === 'data_retention') {
return runDataRetentionJob()
}
throw createHttpError(`不支持的定时任务类型：${job.type}`, {
statusCode: 400,
apps/backend/src/services/worker-platform/admin-worker-user-service.ts
+2 -1
View File
@@ -97,7 +97,8 @@ export async function setAdminWorkerPersonalDepositFreeAmount(
) {
const worker = await getRequiredWorker(workerId)
const rawAmount = payload.depositFreeAmount ?? payload.deposit_free_amount
const clearPersonalAmount = rawAmount === null || rawAmount === undefined || String(rawAmount).trim() === ''
const clearPersonalAmount =
rawAmount === null || rawAmount === undefined || String(rawAmount).trim() === ''
if (!clearPersonalAmount) {
const amount = Number(rawAmount)
if (!Number.isFinite(amount) || amount < 0) {
apps/backend/src/services/worker-platform/worker-permission-utils.ts
+2 -2
View File
@@ -17,8 +17,8 @@ export function resolveWorkerPermissions(worker: WorkerUserRow) {
personalDepositFreeAmount >= 0
return {
// 个人额度是在等级额度基础上的追加额度，两者叠加生效。
depositFreeAmount: levelDepositFreeAmount +
(hasPersonalDepositFreeAmount ? personalDepositFreeAmount : 0),
depositFreeAmount:
levelDepositFreeAmount + (hasPersonalDepositFreeAmount ? personalDepositFreeAmount : 0),
levelDepositFreeAmount,
personalDepositFreeAmount: hasPersonalDepositFreeAmount ? personalDepositFreeAmount : null,
maxActiveOrders: normalizePositiveInteger(permission.maxActiveOrders, 1),
apps/backend/src/types/runtime-config.ts
+6
View File
@@ -32,6 +32,12 @@ export type RuntimeConfig = {
integrationLevel: string
retentionDays: number
}
retention: {
rawPayloadDays: number
taskEventDays: number
webhookEventDays: number
auditLogDays: number
}
database: {
url: string
ssl: boolean