diff --git a/.env.server.example b/.env.server.example index b5119f41..1666e79b 100644 --- a/.env.server.example +++ b/.env.server.example @@ -40,6 +40,12 @@ LOG_LEVEL=info LOG_INTEGRATION_LEVEL=warn LOG_RETENTION_DAYS=30 +# 数据库敏感报文/事件保留策略（天）。清理任务每天执行一次。 +RAW_PAYLOAD_RETENTION_DAYS=180 +TASK_EVENT_RETENTION_DAYS=180 +WEBHOOK_EVENT_RETENTION_DAYS=180 +AUDIT_LOG_RETENTION_DAYS=365 + # 文件存储：生产使用阿里云 OSS（生产编排已移除 MinIO，仅本地开发使用）。 # STORAGE_MODE 取值 minio | dual | oss；dual 仅在 MinIO→OSS 迁移/回滚期间使用。 STORAGE_MODE=oss @@ -74,6 +80,9 @@ DB_BACKUP_KEEP=7 ADMIN_SESSION_SECRET=change-me-long-random-session-secret ADMIN_DEFAULT_USERS_JSON=[{"username":"admin","password":"change-me-admin-password","role":"admin"},{"username":"operator","password":"change-me-operator-password","role":"operator"}] +# 生产环境必须使用显式来源白名单，多个来源用逗号分隔。 +CORS_ALLOWED_ORIGINS=https://order.khhao.com,https://worker.khhao.com + # 91Kaquan KAQUAN91_USER_ID= KAQUAN91_SECRET= diff --git a/apps/backend/scripts/check-sql-guard.ts b/apps/backend/scripts/check-sql-guard.ts index 0488d6ff..5817b7e9 100644 --- a/apps/backend/scripts/check-sql-guard.ts +++ b/apps/backend/scripts/check-sql-guard.ts @@ -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[] = [] diff --git a/apps/backend/src/app.integration.test.ts b/apps/backend/src/app.integration.test.ts new file mode 100644 index 00000000..06b8be0d --- /dev/null +++ b/apps/backend/src/app.integration.test.ts @@ -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') + } +}) diff --git a/apps/backend/src/config/defaults.ts b/apps/backend/src/config/defaults.ts index 74e5766e..29b59b27 100644 --- a/apps/backend/src/config/defaults.ts +++ b/apps/backend/src/config/defaults.ts @@ -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, diff --git a/apps/backend/src/config/env-documentation.test.ts b/apps/backend/src/config/env-documentation.test.ts index d97944c9..92b75868 100644 --- a/apps/backend/src/config/env-documentation.test.ts +++ b/apps/backend/src/config/env-documentation.test.ts @@ -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', diff --git a/apps/backend/src/config/env-overrides.ts b/apps/backend/src/config/env-overrides.ts index 222387ee..cb2ee68f 100644 --- a/apps/backend/src/config/env-overrides.ts +++ b/apps/backend/src/config/env-overrides.ts @@ -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']), diff --git a/apps/backend/src/config/runtime-validation.test.ts b/apps/backend/src/config/runtime-validation.test.ts index ebff99b8..e31f33b6 100644 --- a/apps/backend/src/config/runtime-validation.test.ts +++ b/apps/backend/src/config/runtime-validation.test.ts @@ -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({ diff --git a/apps/backend/src/config/runtime-validation.ts b/apps/backend/src/config/runtime-validation.ts index 355a6064..133b2eb8 100644 --- a/apps/backend/src/config/runtime-validation.ts +++ b/apps/backend/src/config/runtime-validation.ts @@ -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 diff --git a/apps/backend/src/db/migrations/067_data_retention_indexes.sql b/apps/backend/src/db/migrations/067_data_retention_indexes.sql new file mode 100644 index 00000000..78a3e294 --- /dev/null +++ b/apps/backend/src/db/migrations/067_data_retention_indexes.sql @@ -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 '支持任务事件按创建时间清理'; diff --git a/apps/backend/src/db/postgres.integration.test.ts b/apps/backend/src/db/postgres.integration.test.ts new file mode 100644 index 00000000..c9d28eeb --- /dev/null +++ b/apps/backend/src/db/postgres.integration.test.ts @@ -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() + } + }, +) diff --git a/apps/backend/src/middleware/cors.ts b/apps/backend/src/middleware/cors.ts index b514bf5a..42ab8d3b 100644 --- a/apps/backend/src/middleware/cors.ts +++ b/apps/backend/src/middleware/cors.ts @@ -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) diff --git a/apps/backend/src/repositories/worker-platform/work-order-settlement-repo.ts b/apps/backend/src/repositories/worker-platform/work-order-settlement-repo.ts index 75219e77..71a67898 100644 --- a/apps/backend/src/repositories/worker-platform/work-order-settlement-repo.ts +++ b/apps/backend/src/repositories/worker-platform/work-order-settlement-repo.ts @@ -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) diff --git a/apps/backend/src/repositories/worker-platform/work-order-share-repo.ts b/apps/backend/src/repositories/worker-platform/work-order-share-repo.ts index 0f06d21a..e85d7cdb 100644 --- a/apps/backend/src/repositories/worker-platform/work-order-share-repo.ts +++ b/apps/backend/src/repositories/worker-platform/work-order-share-repo.ts @@ -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( diff --git a/apps/backend/src/services/file-storage/file-storage-service.ts b/apps/backend/src/services/file-storage/file-storage-service.ts index 8ce2aa50..bbc89f7e 100644 --- a/apps/backend/src/services/file-storage/file-storage-service.ts +++ b/apps/backend/src/services/file-storage/file-storage-service.ts @@ -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 } } diff --git a/apps/backend/src/services/file-storage/object-storage.ts b/apps/backend/src/services/file-storage/object-storage.ts index c9d060f0..a1e6c465 100644 --- a/apps/backend/src/services/file-storage/object-storage.ts +++ b/apps/backend/src/services/file-storage/object-storage.ts @@ -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 { + await this.ensureBucketReady() + await this.client.send( + new DeleteObjectCommand({ Bucket: this.bucket, Key: normalizeObjectKey(key) }), + ) + } + async getObject(key: string): Promise { await this.ensureBucketReady() const objectKey = normalizeObjectKey(key) @@ -233,6 +241,18 @@ export class StorageRouter { return this.oss!.getObject(key) } } + + async deleteObject(key: string): Promise { + 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 diff --git a/apps/backend/src/services/scheduler/config-service.test.ts b/apps/backend/src/services/scheduler/config-service.test.ts index f7949b37..a68f4b7a 100644 --- a/apps/backend/src/services/scheduler/config-service.test.ts +++ b/apps/backend/src/services/scheduler/config-service.test.ts @@ -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: {}, + }) +}) diff --git a/apps/backend/src/services/scheduler/config-service.ts b/apps/backend/src/services/scheduler/config-service.ts index 465c8d21..f42c62bc 100644 --- a/apps/backend/src/services/scheduler/config-service.ts +++ b/apps/backend/src/services/scheduler/config-service.ts @@ -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, diff --git a/apps/backend/src/services/scheduler/data-retention-job.ts b/apps/backend/src/services/scheduler/data-retention-job.ts new file mode 100644 index 00000000..71980e86 --- /dev/null +++ b/apps/backend/src/services/scheduler/data-retention-job.ts @@ -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 { + 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 +} diff --git a/apps/backend/src/services/scheduler/scheduler-service.ts b/apps/backend/src/services/scheduler/scheduler-service.ts index 622f6007..5174dadf 100644 --- a/apps/backend/src/services/scheduler/scheduler-service.ts +++ b/apps/backend/src/services/scheduler/scheduler-service.ts @@ -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() const jobStates = new Map() @@ -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, diff --git a/apps/backend/src/services/worker-platform/admin-worker-user-service.ts b/apps/backend/src/services/worker-platform/admin-worker-user-service.ts index 251b38a3..6fcd742b 100644 --- a/apps/backend/src/services/worker-platform/admin-worker-user-service.ts +++ b/apps/backend/src/services/worker-platform/admin-worker-user-service.ts @@ -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) { diff --git a/apps/backend/src/services/worker-platform/worker-permission-utils.ts b/apps/backend/src/services/worker-platform/worker-permission-utils.ts index 95b6707e..68c22656 100644 --- a/apps/backend/src/services/worker-platform/worker-permission-utils.ts +++ b/apps/backend/src/services/worker-platform/worker-permission-utils.ts @@ -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), diff --git a/apps/backend/src/types/runtime-config.ts b/apps/backend/src/types/runtime-config.ts index 7f1b9667..f4f5aa43 100644 --- a/apps/backend/src/types/runtime-config.ts +++ b/apps/backend/src/types/runtime-config.ts @@ -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 diff --git a/apps/frontend/src/components/WorkOrderEventTimeline.tsx b/apps/frontend/src/components/WorkOrderEventTimeline.tsx index edc93fa3..f3489181 100644 --- a/apps/frontend/src/components/WorkOrderEventTimeline.tsx +++ b/apps/frontend/src/components/WorkOrderEventTimeline.tsx @@ -169,7 +169,10 @@ export function formatWorkOrderEvent(event: WorkOrderEvent): EventNodeStyle { color: 'green', } case 'gift_material_reset': - return { title: note ? `客服重置打手好友赠送资料：${note}` : '客服重置打手好友赠送资料', color: 'orange' } + return { + title: note ? `客服重置打手好友赠送资料：${note}` : '客服重置打手好友赠送资料', + color: 'orange', + } // —— 验收节点 —— case 'acceptance_submitted': { diff --git a/apps/frontend/src/pages/admin/panels/WorkerDetailDrawer.tsx b/apps/frontend/src/pages/admin/panels/WorkerDetailDrawer.tsx index 9af0cc29..c184b61a 100644 --- a/apps/frontend/src/pages/admin/panels/WorkerDetailDrawer.tsx +++ b/apps/frontend/src/pages/admin/panels/WorkerDetailDrawer.tsx @@ -54,7 +54,11 @@ function resolveLedgerTagColor(ledgerType: string) { ].includes(ledgerType) ) return 'gold' - if (['deposit_release', 'deposit_unfreeze', 'reward_unfreeze', 'after_sales_refund'].includes(ledgerType)) + if ( + ['deposit_release', 'deposit_unfreeze', 'reward_unfreeze', 'after_sales_refund'].includes( + ledgerType, + ) + ) return 'green' if ( [ diff --git a/apps/frontend/src/pages/admin/panels/WorkersPanel.tsx b/apps/frontend/src/pages/admin/panels/WorkersPanel.tsx index 8ea43cb1..24e34ccf 100644 --- a/apps/frontend/src/pages/admin/panels/WorkersPanel.tsx +++ b/apps/frontend/src/pages/admin/panels/WorkersPanel.tsx @@ -371,9 +371,9 @@ export default function WorkersPanel() { render: (_, row) => row.personalDepositFreeAmount !== null ? ( - {formatMoney(row.effectiveDepositFreeAmount)}（等级 {formatMoney( - row.levelDepositFreeAmount, - )} + 个人 {formatMoney(row.personalDepositFreeAmount)}） + {formatMoney(row.effectiveDepositFreeAmount)}（等级{' '} + {formatMoney(row.levelDepositFreeAmount)} + 个人{' '} + {formatMoney(row.personalDepositFreeAmount)}） ) : (