diff --git a/Dockerfile b/Dockerfile index 5332549..2635110 100644 --- a/Dockerfile +++ b/Dockerfile @@ -104,7 +104,8 @@ RUN test -s "/app/public/brand/header-logo.png" && \ test -f "/app/prisma/migrations/20260924_add_certificate_tested_on/migration.sql" RUN test -f "/app/prisma/migrations/20261008_permission_group_department_identity/migration.sql" && \ test -f "/app/prisma/migrations/20261008_team_function_configs/migration.sql" && \ - test -f "/app/prisma/migrations/20261009_material_purge_outbox/migration.sql" + test -f "/app/prisma/migrations/20261009_material_purge_outbox/migration.sql" && \ + test -f "/app/prisma/migrations/20261009_material_purge_reference_indexes/migration.sql" # 这两个动态路由负责兜底访问本地上传文件；如果它们在构建上下文里丢失， # 线上音频/视频就会直接 404。这里提前失败，避免再把缺路由的镜像发出去。 diff --git a/docs/回收站删除索引修复-2026-10-09.md b/docs/回收站删除索引修复-2026-10-09.md new file mode 100644 index 0000000..00b7e65 --- /dev/null +++ b/docs/回收站删除索引修复-2026-10-09.md @@ -0,0 +1,19 @@ +# 回收站删除引用索引修复 + +此前每条清理任务都现场重建全库文件引用。约 40 万条资产下，第一次优化仍需约 5 秒；这个耗时不是正常文件删除所需，而是全表扫描和路径转换造成的。原生后台任务、持久化清单和失败续跑保留，但生产清理不再调用全量扫描。 + +新增 `20261009_material_purge_reference_indexes` 迁移，在九个真实引用来源建立 GIN 表达式索引。文件地址、数组与 JSON 中的字符串生成检索键，由数据库在插入、更新、删除时自动维护，覆盖 Prisma 和直接 SQL 写入。历史记录在建索引时纳入，不需要应用双写、定期同步或引用缓存。 + +索引仅返回候选。应用仍用完整存储类型、桶及路径判定共享引用；哈希冲突只会增加候选。回收站素材、软删除 SKU、历史对象地址、白底派生文件和活动上传/审批引用继续受到保护。URL 点段和控制字符的少量特殊引用保留为索引候选，再交由原生 URL 规则核对；Windows Unicode 路径保守匹配。SQL 函数明确禁用并行执行，避免错误恢复子事务在并行查询内报错。 + +提交后立即唤醒已经启动的清理 worker。空队列检查与入队交错时保留唤醒信号，每轮满 20 条立即接着处理下一轮；5 秒定时器只用于恢复和到期重试。没有把排队或未完成的物理清理显示成成功。 + +## 验证 + +- 21 个相关 Jest 测试，包括入队与空队列检查交错时不会丢失唤醒。 +- 14 个真实 PostgreSQL/文件系统/MinIO 集成验证组；包括幂等、事务回滚、租约接管、真实删除、共享保护、直接 SQL 更新、队列状态变化、特殊地址和失败续跑。 +- 独立数据库 400,000 条额外资产：引用检查 59ms，提交 24ms，4 个文件从提交到实际删除及审计完成 122ms；资产查询计划实际使用 `material_assets_purge_references_idx`。 +- 已启动 worker 收到提交后 60ms 完成，不等待 5 秒轮询。 +- 类型检查、存储边界检查和 14 个安全迁移测试通过。迁移审计核对九个索引的表达式、状态谓词、有效性及三个函数的完整定义。 + +生产验收只读核对原先 18 个任务的 1,276 个文件和现存引用样本，不由自动化提交真实素材删除。隔离测试耗时不等于生产完整物理清理耗时；生产查询耗时和发布凭据另存于发布回执。 diff --git a/lib/__tests__/material-purge-files.test.ts b/lib/__tests__/material-purge-files.test.ts index c817a6b..6372588 100644 --- a/lib/__tests__/material-purge-files.test.ts +++ b/lib/__tests__/material-purge-files.test.ts @@ -87,7 +87,7 @@ test('候选筛选保留特殊文件名、编码 URL 和控制字符 URL 的等 const db: any = { material: empty, materialAsset: { findMany: jest.fn().mockResolvedValue(rows) }, materialSku: empty, brandKnowledge: empty, systemSettings: empty, performanceSurveyWorkItem: empty, workReviewVersion: empty, productPackageUploadJob: empty, taskDetailChangeRequest: empty, - $queryRaw: jest.fn().mockResolvedValue(rows) } + $queryRaw: jest.fn().mockResolvedValue(rows.map(row => ({ sources: [row.fileUrl], derived: true }))) } const full = await findPurgeFileReferences(db) const targeted = await findPurgeFileReferences(db, targets) expect(targeted.has('local::shared[1]+.bin')).toBe(true) diff --git a/lib/__tests__/material-purge-worker.test.ts b/lib/__tests__/material-purge-worker.test.ts new file mode 100644 index 0000000..29d5432 --- /dev/null +++ b/lib/__tests__/material-purge-worker.test.ts @@ -0,0 +1,33 @@ +/** @jest-environment node */ +import { ensureMaterialPurgeWorkerStarted } from '@/lib/material-purge' +import { prisma } from '@/lib/prisma' + +jest.mock('@/lib/prisma', () => ({ prisma: { $transaction: jest.fn() } })) +jest.mock('@/lib/material-purge-files', () => ({})) + +test('空队列检查期间收到入队信号，立即再检查，不丢失唤醒并等待轮询', async () => { + jest.useFakeTimers() + const global = globalThis as typeof globalThis & { __materialPurgeWorkerStarted?: boolean; __materialPurgePump?: () => Promise } + const oldBuild = process.env.NEXT_BUILD + delete process.env.NEXT_BUILD + delete global.__materialPurgeWorkerStarted + delete global.__materialPurgePump + let finish!: (rows: unknown[]) => void + const query = jest.fn().mockImplementationOnce(() => new Promise(resolve => { finish = resolve })).mockResolvedValue([]) + ;(prisma.$transaction as jest.Mock).mockImplementation(callback => callback({ $queryRaw: query })) + try { + ensureMaterialPurgeWorkerStarted() + expect(query).toHaveBeenCalledTimes(1) + ensureMaterialPurgeWorkerStarted() + finish([]) + await jest.advanceTimersByTimeAsync(0) + expect(query).toHaveBeenCalledTimes(2) + } finally { + jest.clearAllTimers() + jest.useRealTimers() + delete global.__materialPurgeWorkerStarted + delete global.__materialPurgePump + if (oldBuild === undefined) delete process.env.NEXT_BUILD + else process.env.NEXT_BUILD = oldBuild + } +}) diff --git a/lib/material-purge-files.ts b/lib/material-purge-files.ts index 424628d..a78ae74 100644 --- a/lib/material-purge-files.ts +++ b/lib/material-purge-files.ts @@ -4,6 +4,7 @@ import type { Prisma } from '@prisma/client' import { deleteFromS3, getTrustedS3MutationKey } from '@/lib/s3' import { getUploadRootDir } from '@/lib/upload-paths' import { buildPublicDisplayUrl, buildWhiteBackgroundPublicDisplayPathFromAssetPath } from '@/lib/public-display' +import { findIndexedPurgeReferenceSources } from '@/lib/material-purge-reference-query' export type PurgeFile = { source: string @@ -133,95 +134,56 @@ function stringsIn(value: unknown): string[] { return [] } -/** - * 常规路径按本次文件的完整地址匹配；编码、控制字符和 URL 点路径 - * 不能按字面排除，全部纳入候选。最终仍由原生路径校验核对完整身份。 - * 白底派生图的本地引用也单独纳入，不依赖源文件与公网文件的路径相同。 - * 包含回收站与 SKU 软删除记录，保留其他素材恢复时仍需要的共用文件。 - */ +/** 生产清理按目标文件查索引；无 files 参数的全量读取仅供只读审计核对。 */ export async function findPurgeFileReferences(db: Prisma.TransactionClient, files?: readonly PurgeFile[]): Promise> { if (files && !files.length) return new Set() - const caseFold = process.platform === 'win32' - const sources = files ? new Set() : null - const root = path.resolve(getUploadRootDir()) - for (const file of files || []) { - const values = file.storage === 's3' - ? [`/${file.key}`, `/${file.bucket}/${file.key}`] - : [file.key, `uploads/${file.key}`, `/uploads/${file.key}`, `./uploads/${file.key}`, - path.join(root, file.key), path.relative(process.cwd(), path.join(root, file.key)), - `./${path.relative(process.cwd(), path.join(root, file.key))}`] - for (const value of values) { - const normalized = value.replace(/\\/g, '/') - sources!.add(caseFold ? normalized.toLowerCase() : normalized) - } + let sources: string[] + let assets: { fileUrl: string }[] + if (files) { + const rows = await findIndexedPurgeReferenceSources(db, files) + sources = rows.flatMap(row => row.sources).filter((source): source is string => typeof source === 'string') + assets = rows.filter(row => row.derived).flatMap(row => row.sources.map(fileUrl => ({ fileUrl }))) + } else { + const rows = await Promise.all([ + db.material.findMany({ select: { previewUrl: true, zipFilePath: true, displayImages: true, + videoSourceUrl: true, videoPreview540Url: true, videoPreviewClipUrl: true, videoCoverUrl: true } }), + db.materialAsset.findMany({ select: { fileUrl: true } }), + db.materialSku.findMany({ select: { filePath: true, previewUrl: true } }), + db.brandKnowledge.findMany({ select: { filePath: true, previewImagePath: true, previewImageSlices: true } }), + db.systemSettings.findMany({ select: { headerLogoUrl: true, loginLogoUrl: true } }), + db.performanceSurveyWorkItem.findMany({ select: { imageUrls: true } }), + db.workReviewVersion.findMany({ select: { imageUrls: true } }), + db.productPackageUploadJob.findMany({ where: { status: { in: ['queued', 'processing'] } }, select: { assets: true } }), + db.taskDetailChangeRequest.findMany({ where: { status: { notIn: ['applied', 'rejected', 'expired', 'cancelled'] } }, select: { operations: true } }), + ]) + sources = stringsIn(rows) + assets = rows[1] as { fileUrl: string }[] } - const candidate = (source: string) => { - if (!sources || /[%\x00-\x1f\x7f]/.test(source)) return true - const value = source.trim().replace(/\\/g, '/') - const pathname = /^https?:\/\//i.test(value) ? value.split(/[?#]/, 1)[0] : value - const relative = /^https?:\/\//i.test(pathname) ? pathname.slice(pathname.indexOf('/', pathname.indexOf('://') + 3)) : pathname - if (relative.includes('/.') || relative.includes('//')) return true - return sources.has(caseFold ? relative.toLowerCase() : relative) - } - // 格式特殊的地址先全取，再用原生规则核对。Windows 非 ASCII 地址也全取， - // 避免数据库 locale 与 JavaScript 文件系统大小写规则不同。 - const edgeWhitespace = '[[:space:]\u00a0\u1680\u2000-\u200a\u2028\u2029\u202f\u205f\u3000\ufeff]' - const [materials, assets, skus, knowledge, settings, workItems, versions, uploads, changes] = await Promise.all([ - db.material.findMany({ select: { previewUrl: true, zipFilePath: true, displayImages: true, - videoSourceUrl: true, videoPreview540Url: true, videoPreviewClipUrl: true, videoCoverUrl: true } }), - sources - ? db.$queryRaw<{ fileUrl: string }[]>` - WITH paths AS ( - SELECT file_url, CASE - WHEN lower(left(file_url, 7)) = 'http://' - THEN substring(file_url FROM 7 + strpos(substring(file_url FROM 8), '/')) - WHEN lower(left(file_url, 8)) = 'https://' - THEN substring(file_url FROM 8 + strpos(substring(file_url FROM 9), '/')) - ELSE file_url END AS reference FROM material_assets - ) - SELECT asset.file_url AS "fileUrl" FROM paths AS asset - WHERE (CASE WHEN ${caseFold} THEN lower(asset.reference) ELSE asset.reference END) - IN (SELECT unnest(${Array.from(sources)}::text[])) - OR asset.file_url ~ '[%[:cntrl:]]' - OR strpos(asset.file_url, ${'\\'}) > 0 OR strpos(asset.file_url, '/.') > 0 - OR strpos(asset.file_url, '?') > 0 OR strpos(asset.file_url, '#') > 0 - OR asset.file_url ~ '(^|[^:])//' OR asset.file_url ~ ${`^${edgeWhitespace}|${edgeWhitespace}$`} - OR strpos(asset.file_url, '/white-background/') > 0 - OR (${caseFold} AND asset.file_url ~ ${'[^\u0001-\u007f]'})` - : db.materialAsset.findMany({ select: { fileUrl: true } }), - db.materialSku.findMany({ select: { filePath: true, previewUrl: true } }), - db.brandKnowledge.findMany({ select: { filePath: true, previewImagePath: true, previewImageSlices: true } }), - db.systemSettings.findMany({ select: { headerLogoUrl: true, loginLogoUrl: true } }), - db.performanceSurveyWorkItem.findMany({ select: { imageUrls: true } }), - db.workReviewVersion.findMany({ select: { imageUrls: true } }), - db.productPackageUploadJob.findMany({ where: { status: { in: ['queued', 'processing'] } }, select: { assets: true } }), - db.taskDetailChangeRequest.findMany({ where: { status: { notIn: ['applied', 'rejected', 'expired', 'cancelled'] } }, select: { operations: true } }), - ]) + const targets = files && new Set(files.map(purgeFileIdentity)) const identities = new Set() - for (const source of stringsIn([materials, assets, skus, knowledge, settings, workItems, versions, uploads, changes])) { - if (!candidate(source)) continue + const protect = (identity: string) => { if (!targets || targets.has(identity)) identities.add(identity) } + for (const source of sources) { try { const file = identifyPurgeFile(source) - if (file) identities.add(purgeFileIdentity(file)) + if (file) protect(purgeFileIdentity(file)) } catch { - // 外部只读引用不扩大写权限。路径相同的历史 HTTP 引用仍保护当前桶中的对象。 + // 外部只读引用不扩大写权限；历史域名的同路径仍保护当前桶中的对象。 if (/^https?:\/\//i.test(source)) { try { let key = decodeURIComponent(new URL(source).pathname.slice(1)) const bucket = process.env.S3_BUCKET || '' if (bucket && key.startsWith(`${bucket}/`)) key = key.slice(bucket.length + 1) - identities.add(purgeFileIdentity({ storage: 's3', bucket, key })) + protect(purgeFileIdentity({ storage: 's3', bucket, key })) } catch { /* 非文件字符串不构成引用 */ } } } } - // 白底图的公网派生文件也可能被其他 SKU 的本地源文件引用。 for (const asset of assets) { const derived = buildWhiteBackgroundPublicDisplayPathFromAssetPath(asset.fileUrl) const source = derived && buildPublicDisplayUrl(derived) - if (source && candidate(source)) { + if (source) { const file = identifyPurgeFile(source) - if (file) identities.add(purgeFileIdentity(file)) + if (file) protect(purgeFileIdentity(file)) } } return identities diff --git a/lib/material-purge-reference-query.ts b/lib/material-purge-reference-query.ts new file mode 100644 index 0000000..79807bc --- /dev/null +++ b/lib/material-purge-reference-query.ts @@ -0,0 +1,60 @@ +import { createHash } from 'node:crypto' +import path from 'node:path' +import type { Prisma } from '@prisma/client' +import { getUploadRootDir } from '@/lib/upload-paths' +import type { PurgeFile } from '@/lib/material-purge-files' + +type ReferenceRow = { sources: string[]; derived: boolean } +const hash = (value: string) => createHash('md5').update(value).digest('hex') + +/** 索引只找候选；存储归属、历史引用及共用文件仍由原生路径规则核对。 */ +export function purgeReferenceLookupKeys(files: readonly PurgeFile[]): string[] { + const keys = new Set(['url:resolve']) + const root = path.resolve(getUploadRootDir()) + const windows = process.platform === 'win32' + if (windows) keys.add('windows:unicode') + for (const file of files) { + const aliases = file.storage === 's3' + ? [`/${file.key}`, `/${file.bucket}/${file.key}`] + : [file.key, `uploads/${file.key}`, `/uploads/${file.key}`, `./uploads/${file.key}`, + path.join(root, file.key), path.relative(process.cwd(), path.join(root, file.key)), + `./${path.relative(process.cwd(), path.join(root, file.key))}`] + for (const alias of aliases) { + let normalized = alias.replace(/\\/g, '/').replace(/\/{2,}/g, '/') + while (/(^|\/)\.(\/|$)/.test(normalized)) normalized = normalized.replace(/(^|\/)\.(\/|$)/g, '$1') + keys.add(windows + ? `ascii:${hash(normalized.replace(/[A-Z]/g, char => char.toLowerCase()))}` + : `path:${hash(normalized)}`) + } + if (file.storage === 's3') { + const marker = 'public-display/' + const index = file.key.indexOf(marker) + if (index >= 0) keys.add(`derived:${hash(file.key.slice(index + marker.length))}`) + } + } + return Array.from(keys) +} + +export async function findIndexedPurgeReferenceSources(db: Prisma.TransactionClient, files: readonly PurgeFile[]): Promise { + const keys = purgeReferenceLookupKeys(files) + return db.$queryRaw` + SELECT ARRAY["previewUrl", "zipFilePath", "videoSourceUrl", "videoPreview540Url", "videoPreviewClipUrl", "videoCoverUrl"] || "displayImages" AS sources, false AS derived + FROM materials WHERE purge_reference_keys(ARRAY["previewUrl", "zipFilePath", "videoSourceUrl", "videoPreview540Url", "videoPreviewClipUrl", "videoCoverUrl"] || "displayImages", false) && ${keys}::text[] + UNION ALL + SELECT ARRAY[file_url], true FROM material_assets WHERE purge_reference_keys(ARRAY[file_url], true) && ${keys}::text[] + UNION ALL + SELECT ARRAY[file_path, preview_url], false FROM material_skus WHERE purge_reference_keys(ARRAY[file_path, preview_url], false) && ${keys}::text[] + UNION ALL + SELECT ARRAY[file_path, preview_image_path] || preview_image_slices, false FROM brand_knowledge WHERE purge_reference_keys(ARRAY[file_path, preview_image_path] || preview_image_slices, false) && ${keys}::text[] + UNION ALL + SELECT ARRAY["headerLogoUrl", "loginLogoUrl"], false FROM system_settings WHERE purge_reference_keys(ARRAY["headerLogoUrl", "loginLogoUrl"], false) && ${keys}::text[] + UNION ALL + SELECT purge_json_reference_strings(image_urls), false FROM performance_survey_work_items WHERE purge_reference_keys(purge_json_reference_strings(image_urls), false) && ${keys}::text[] + UNION ALL + SELECT purge_json_reference_strings(image_urls), false FROM work_review_versions WHERE purge_reference_keys(purge_json_reference_strings(image_urls), false) && ${keys}::text[] + UNION ALL + SELECT purge_json_reference_strings(assets), false FROM product_package_upload_jobs WHERE status IN ('queued', 'processing') AND purge_reference_keys(purge_json_reference_strings(assets), false) && ${keys}::text[] + UNION ALL + SELECT purge_json_reference_strings(operations), false FROM task_detail_change_requests WHERE status NOT IN ('applied', 'rejected', 'expired', 'cancelled') AND purge_reference_keys(purge_json_reference_strings(operations), false) && ${keys}::text[] + ` +} diff --git a/lib/material-purge.ts b/lib/material-purge.ts index b8ddbda..51e46a4 100644 --- a/lib/material-purge.ts +++ b/lib/material-purge.ts @@ -184,17 +184,28 @@ export async function retryMaterialPurge(jobId: string, operatorUserId: string) export function ensureMaterialPurgeWorkerStarted() { if (process.env.NEXT_BUILD === '1' || process.env.NEXT_PHASE === 'phase-production-build') return - const global = globalThis as typeof globalThis & { __materialPurgeWorkerStarted?: boolean } - if (global.__materialPurgeWorkerStarted) return + const global = globalThis as typeof globalThis & { __materialPurgeWorkerStarted?: boolean; __materialPurgePump?: () => Promise } + if (global.__materialPurgeWorkerStarted) { void global.__materialPurgePump?.(); return } global.__materialPurgeWorkerStarted = true let running = false + let wakeRequested = false const pump = async () => { - if (running) return + if (running) { wakeRequested = true; return } running = true - try { for (let count = 0; count < 20 && await processNextMaterialPurge(); count += 1) { /* 限定每轮批量 */ } } + wakeRequested = false + let more = false + try { + for (let count = 0; count < 20 && await processNextMaterialPurge(); count += 1) more = count === 19 + } catch (error) { console.error('[MaterialPurge] 清理队列暂不可用，将在下一轮重试:', error) } - finally { running = false } + finally { + running = false + // 入队与当前空队列检查交错时保留唤醒信号；满一轮继续处理下一批。 + if (wakeRequested || more) setImmediate(() => { void pump() }).unref?.() + } } + // 入队事务提交后直接唤醒；轮询仅用于重启恢复和延迟重试，不让新请求等下一轮。 + global.__materialPurgePump = pump setInterval(() => { void pump() }, 5_000).unref?.() void pump() } diff --git a/prisma/migrations/20261009_material_purge_reference_indexes/migration.sql b/prisma/migrations/20261009_material_purge_reference_indexes/migration.sql new file mode 100644 index 0000000..1d7cfdc --- /dev/null +++ b/prisma/migrations/20261009_material_purge_reference_indexes/migration.sql @@ -0,0 +1,134 @@ +BEGIN; + +-- 只建立引用检索索引，不改动业务记录。索引由 PostgreSQL 随每次写入维护， +-- 因而覆盖 ORM、历史导入及直接 SQL 写入，不依赖应用层双写或定时回填。 +CREATE FUNCTION purge_decode_reference(value text) RETURNS text +LANGUAGE plpgsql IMMUTABLE STRICT PARALLEL UNSAFE AS $$ +DECLARE result bytea := ''::bytea; position integer := 1; piece text; +BEGIN + IF strpos(value, '%') = 0 THEN RETURN value; END IF; + WHILE position <= length(value) LOOP + piece := substr(value, position, 1); + IF piece = '%' THEN + piece := substr(value, position + 1, 2); + IF piece !~ '^[0-9a-fA-F]{2}$' THEN RETURN value; END IF; + result := result || decode(piece, 'hex'); position := position + 3; + ELSE + result := result || convert_to(piece, 'UTF8'); position := position + 1; + END IF; + END LOOP; + RETURN convert_from(result, 'UTF8'); +EXCEPTION WHEN character_not_in_repertoire OR untranslatable_character THEN RETURN value; +END; +$$; + +CREATE FUNCTION purge_json_reference_strings(value jsonb) RETURNS text[] +LANGUAGE plpgsql IMMUTABLE STRICT PARALLEL UNSAFE AS $$ +DECLARE result text[] := '{}'; child jsonb; +BEGIN + CASE jsonb_typeof(value) + WHEN 'string' THEN RETURN ARRAY[value #>> '{}']; + WHEN 'array' THEN + FOR child IN SELECT jsonb_array_elements(value) LOOP + result := result || purge_json_reference_strings(child); + END LOOP; + WHEN 'object' THEN + FOR child IN SELECT v FROM jsonb_each(value) AS entry(k, v) LOOP + result := result || purge_json_reference_strings(child); + END LOOP; + ELSE NULL; + END CASE; + RETURN result; +END; +$$; + +-- 哈希仅用于索引候选匹配，最终删除授权仍使用应用的完整 storage/bucket/key。 +-- 哈希冲突只会增加候选，不会授权删除；内嵌 data:image 不进入文件索引。 +CREATE FUNCTION purge_reference_keys(sources text[], derive_white_background boolean) RETURNS text[] +LANGUAGE plpgsql IMMUTABLE STRICT PARALLEL UNSAFE AS $$ +DECLARE + result text[] := '{}'; source text; value text; local_source text; decoded text; derived text; + segments text[]; tail text[]; wb integer; round integer; is_http boolean; + whitespace text := U&'\0009\000A\000B\000C\000D\0020\00A0\1680\2000\2001\2002\2003\2004\2005\2006\2007\2008\2009\200A\2028\2029\202F\205F\3000\FEFF'; +BEGIN + FOREACH source IN ARRAY sources LOOP + value := btrim(source, whitespace); + IF value IS NULL OR value = '' OR value LIKE 'data:image/%' THEN CONTINUE; END IF; + is_http := value ~* '^https?://'; + value := replace(value, chr(92), '/'); + IF is_http THEN + -- URL 的点段及内嵌制表/换行由 WHATWG URL 处理。保留稀少的这类候选， + -- 不在 SQL 中另写一套 URL 授权规则；常规地址及编码地址走精确索引。 + IF value ~ E'[\\t\\r\\n]' OR value ~* '/(\.|%2e)' THEN + result := array_append(result, 'url:resolve'); + END IF; + value := regexp_replace(value, '[?#].*$', ''); + value := regexp_replace(value, '^https?://[^/]*', '', 'i'); + END IF; + IF NOT is_http THEN + value := regexp_replace(value, '/+', '/', 'g'); + WHILE value ~ '(^|/)\.(\/|$)' LOOP + value := regexp_replace(value, '(^|/)\.(\/|$)', '\1', 'g'); + END LOOP; + END IF; + local_source := value; + IF value ~ '[^\x01-\x7f]' THEN result := array_append(result, 'windows:unicode'); END IF; + FOR round IN 0..6 LOOP + IF is_http THEN value := regexp_replace(value, '^/+', '/'); END IF; + result := result || ARRAY['path:' || md5(value), + 'ascii:' || md5(translate(value, 'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz'))]; + decoded := purge_decode_reference(value); + EXIT WHEN decoded = value; + value := decoded; + END LOOP; + IF derive_white_background AND NOT is_http THEN + segments := array_remove(string_to_array(local_source, '/'), ''); + wb := array_position(segments, 'white-background'); + IF wb >= 3 AND array_length(segments, 1) >= wb + 2 THEN + tail := segments[wb + 2:array_length(segments, 1)]; + IF tail[1] = 'preview' THEN tail := tail[2:array_length(tail, 1)]; END IF; + IF cardinality(tail) > 0 THEN + derived := 'white-background/' || segments[wb - 1] || '/' || segments[wb + 1] || '/' || array_to_string(tail, '/'); + FOR round IN 0..6 LOOP + result := array_append(result, 'derived:' || md5(derived)); + decoded := purge_decode_reference(derived); + EXIT WHEN decoded = derived; + derived := decoded; + END LOOP; + END IF; + END IF; + END IF; + END LOOP; + RETURN ARRAY(SELECT DISTINCT token FROM unnest(result) token); +END; +$$; + +CREATE INDEX materials_purge_references_idx ON materials USING gin + (purge_reference_keys(ARRAY["previewUrl", "zipFilePath", "videoSourceUrl", "videoPreview540Url", "videoPreviewClipUrl", "videoCoverUrl"] || "displayImages", false)); +CREATE INDEX material_assets_purge_references_idx ON material_assets USING gin + (purge_reference_keys(ARRAY[file_url], true)); +CREATE INDEX material_skus_purge_references_idx ON material_skus USING gin + (purge_reference_keys(ARRAY[file_path, preview_url], false)); +CREATE INDEX brand_knowledge_purge_references_idx ON brand_knowledge USING gin + (purge_reference_keys(ARRAY[file_path, preview_image_path] || preview_image_slices, false)); +CREATE INDEX system_settings_purge_references_idx ON system_settings USING gin + (purge_reference_keys(ARRAY["headerLogoUrl", "loginLogoUrl"], false)); +CREATE INDEX performance_work_items_purge_references_idx ON performance_survey_work_items USING gin + (purge_reference_keys(purge_json_reference_strings(image_urls), false)); +CREATE INDEX work_review_versions_purge_references_idx ON work_review_versions USING gin + (purge_reference_keys(purge_json_reference_strings(image_urls), false)); +CREATE INDEX package_upload_jobs_purge_references_idx ON product_package_upload_jobs USING gin + (purge_reference_keys(purge_json_reference_strings(assets), false)) WHERE status IN ('queued', 'processing'); +CREATE INDEX task_detail_changes_purge_references_idx ON task_detail_change_requests USING gin + (purge_reference_keys(purge_json_reference_strings(operations), false)) WHERE status NOT IN ('applied', 'rejected', 'expired', 'cancelled'); + +ANALYZE materials; +ANALYZE material_assets; +ANALYZE material_skus; +ANALYZE brand_knowledge; +ANALYZE system_settings; +ANALYZE performance_survey_work_items; +ANALYZE work_review_versions; +ANALYZE product_package_upload_jobs; +ANALYZE task_detail_change_requests; +COMMIT; diff --git a/scripts/git-incremental-deploy.ps1 b/scripts/git-incremental-deploy.ps1 index 29d59d4..3c56386 100644 --- a/scripts/git-incremental-deploy.ps1 +++ b/scripts/git-incremental-deploy.ps1 @@ -1524,6 +1524,7 @@ const expectedMigrationNames = new Set([ '20261008_permission_group_department_identity', '20261008_team_function_configs', '20261009_material_purge_outbox', + '20261009_material_purge_reference_indexes', ]); if (migrations.length !== expectedMigrationNames.size || migrations.some((item) => !expectedMigrationNames.has(item && item.name) diff --git a/scripts/safe-production-migrate.mjs b/scripts/safe-production-migrate.mjs index 170b1c3..12b462c 100644 --- a/scripts/safe-production-migrate.mjs +++ b/scripts/safe-production-migrate.mjs @@ -1,6 +1,6 @@ import { createHash } from 'node:crypto' import { spawnSync } from 'node:child_process' -import { createReadStream, realpathSync } from 'node:fs' +import { createReadStream, realpathSync, readFileSync } from 'node:fs' import { lstat, mkdir, readFile, readdir, rename, unlink, writeFile } from 'node:fs/promises' import path from 'node:path' import { fileURLToPath } from 'node:url' @@ -40,6 +40,7 @@ const TARGET_MIGRATIONS = Object.freeze([ '20261008_permission_group_department_identity', '20261008_team_function_configs', '20261009_material_purge_outbox', + '20261009_material_purge_reference_indexes', ]) const VERIFIED_BACKUP_GENERATOR = 'scripts/create-verified-production-backup.mjs' const VERIFIED_BACKUP_MAINTENANCE_CONFIRMATION = 'NO_WRITES_DURING_SNAPSHOT' @@ -420,7 +421,10 @@ async function queryEvidence(prisma, localMigrations) { 'performance_conversion_sync_states_lease_idx' ,'material_permission_pack_dept_unit_key', 'material_permission_pack_departments_pack_id_idx', 'team_function_configs_organization_unit_id_key', 'team_function_config_audits_config_id_created_at_idx', - 'material_purge_jobs_material_id_key', 'material_purge_jobs_status_next_attempt_at_idx', 'material_purge_audits_job_id_created_at_idx' + 'material_purge_jobs_material_id_key', 'material_purge_jobs_status_next_attempt_at_idx', 'material_purge_audits_job_id_created_at_idx', + 'materials_purge_references_idx', 'material_assets_purge_references_idx', 'material_skus_purge_references_idx', + 'brand_knowledge_purge_references_idx', 'system_settings_purge_references_idx', 'performance_work_items_purge_references_idx', + 'work_review_versions_purge_references_idx', 'package_upload_jobs_purge_references_idx', 'task_detail_changes_purge_references_idx' ) ORDER BY idx.relname`, ) @@ -470,7 +474,13 @@ async function queryEvidence(prisma, localMigrations) { FROM "system_settings" ORDER BY id`, ) - return { migrationRows, identity: identityRows[0], columns, indexes, constraints, brandRows } + const referenceFunctions = await prisma.$queryRawUnsafe(` + SELECT proname, prosrc, provolatile::text, proparallel::text, proisstrict, + pg_get_function_identity_arguments(p.oid) AS arguments + FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace + WHERE n.nspname = current_schema() AND proname IN + ('purge_decode_reference', 'purge_json_reference_strings', 'purge_reference_keys')`) + return { migrationRows, identity: identityRows[0], columns, indexes, constraints, brandRows, referenceFunctions } } function completedMigrationMap(rows) { @@ -858,6 +868,47 @@ export function assertTargetMigrationEvidence(localMigrations, evidence) { } assertOrganizationConfigurationEvidence(evidence) assertMaterialPurgeEvidence(evidence) + assertMaterialPurgeReferenceEvidence(evidence) +} + +export function materialPurgeReferenceExpectations() { + const sql = readFileSync(path.join(MIGRATIONS_ROOT, '20261009_material_purge_reference_indexes/migration.sql'), 'utf8').replace(/\r\n/g, '\n') + const functions = [...sql.matchAll(/CREATE FUNCTION (\w+)\(([^)]*)\) RETURNS [^\n]+\nLANGUAGE plpgsql IMMUTABLE STRICT PARALLEL UNSAFE AS \$\$([\s\S]*?)\$\$;/g)] + .map(match => ({ proname: match[1], arguments: match[2], prosrc: match[3], provolatile: 'i', proparallel: 'u', proisstrict: true })) + const indexes = [...sql.matchAll(/CREATE INDEX (\w+) ON (\w+) USING gin\n \(([^;]+)\);/g)] + .map(match => { + const [expression, predicate = ''] = match[3].split(') WHERE ') + const normalizedPredicate = predicate + ? predicate.startsWith('status NOT IN') ? "status <> ALL (ARRAY['applied','rejected','expired','cancelled'])" : "status = ANY (ARRAY['queued','processing'])" + : '' + return { indexname: match[1], tablename: match[2], access_method: 'gin', indisunique: false, + indisprimary: false, indisvalid: true, indisready: true, columns: [expression], predicate: normalizedPredicate } + }) + if (functions.length !== 3 || indexes.length !== 9) fail('Purge reference migration definitions are incomplete') + return { referenceFunctions: functions, indexes } +} + +export function assertMaterialPurgeReferenceEvidence(evidence) { + const expected = materialPurgeReferenceExpectations() + for (const item of expected.indexes) { + const actual = evidence.indexes.find(index => index.indexname === item.indexname) + if (!actual || actual.tablename !== item.tablename || actual.access_method !== 'gin' + || actual.indisunique !== false || actual.indisprimary !== false || actual.indisvalid !== true || actual.indisready !== true + || actual.columns?.length !== 1 + || normalizeConstraintExpression(actual.columns[0]) !== normalizeConstraintExpression(item.columns[0]) + || normalizeConstraintExpression(actual.predicate) !== normalizeConstraintExpression(item.predicate)) { + fail(`Purge reference index mismatch: ${item.indexname}`) + } + } + for (const item of expected.referenceFunctions) { + const matches = (evidence.referenceFunctions || []).filter(fn => fn.proname === item.proname) + const actual = matches[0] + if (matches.length !== 1 || actual.arguments !== item.arguments || actual.provolatile !== 'i' + || actual.proparallel !== 'u' || actual.proisstrict !== true + || String(actual.prosrc).replace(/\r\n/g, '\n') !== item.prosrc) { + fail(`Purge reference function mismatch: ${item.proname}`) + } + } } export function assertMaterialPurgeEvidence(evidence) { diff --git a/scripts/safe-production-migrate.test.mjs b/scripts/safe-production-migrate.test.mjs index 8d762a8..7d8f61f 100644 --- a/scripts/safe-production-migrate.test.mjs +++ b/scripts/safe-production-migrate.test.mjs @@ -15,6 +15,8 @@ import { assertTargetMigrationEvidence, assertOrganizationConfigurationEvidence, assertMaterialPurgeEvidence, + assertMaterialPurgeReferenceEvidence, + materialPurgeReferenceExpectations, databaseFingerprintHash, normalizeColumnDefault, requireExactCommit, @@ -420,6 +422,7 @@ test('target evidence rejects nullable logo columns and structurally wrong index ['20261008_permission_group_department_identity', { checksum: 'ac' }], ['20261008_team_function_configs', { checksum: 'ad' }], ['20261009_material_purge_outbox', { checksum: 'ae' }], + ['20261009_material_purge_reference_indexes', { checksum: 'af' }], ]) const evidence = { migrationRows: [ @@ -691,6 +694,10 @@ test('target evidence rejects nullable logo columns and structurally wrong index ], } evidence.migrationRows.push({ migration_name: '20261009_material_purge_outbox', checksum: 'ae', finished_at: 'now', rolled_back_at: null }) + evidence.migrationRows.push({ migration_name: '20261009_material_purge_reference_indexes', checksum: 'af', finished_at: 'now', rolled_back_at: null }) + const references = materialPurgeReferenceExpectations() + evidence.indexes.push(...references.indexes) + evidence.referenceFunctions = references.referenceFunctions const organization = organizationEvidence() const purge = purgeEvidence() for (const key of ['columns', 'indexes', 'constraints']) evidence[key].push(...organization[key], ...purge[key]) @@ -1063,3 +1070,13 @@ test('container supply chain and migrator image are immutable while SOP keeps co assert.ok(compareIdentity > postIdentity) assert.ok(writeAudit > compareIdentity) }) + + +test('purge reference verification rejects missing GIN coverage, predicate drift and altered canonicalization', () => { + const evidence = materialPurgeReferenceExpectations() + assert.doesNotThrow(() => assertMaterialPurgeReferenceEvidence(evidence)) + assert.throws(() => assertMaterialPurgeReferenceEvidence({ ...evidence, indexes: evidence.indexes.slice(1) }), /index mismatch/) + assert.throws(() => assertMaterialPurgeReferenceEvidence({ ...evidence, indexes: evidence.indexes.map(index => ({ ...index, indisvalid: false })) }), /index mismatch/) + assert.throws(() => assertMaterialPurgeReferenceEvidence({ ...evidence, indexes: evidence.indexes.map(index => index.predicate ? { ...index, predicate: '' } : index) }), /index mismatch/) + assert.throws(() => assertMaterialPurgeReferenceEvidence({ ...evidence, referenceFunctions: evidence.referenceFunctions.map(fn => ({ ...fn, prosrc: fn.prosrc + ' ' })) }), /function mismatch/) +}) diff --git a/scripts/test-material-purge-integration.ts b/scripts/test-material-purge-integration.ts index c8fef94..d8d1e11 100644 --- a/scripts/test-material-purge-integration.ts +++ b/scripts/test-material-purge-integration.ts @@ -31,16 +31,18 @@ async function main() { assert.equal(pushed.status, 0, pushed.stderr) // 重建迁移前的日志级联关系，在空的专用数据库中执行真实 reviewed SQL。 const migrationSql = await fs.readFile('prisma/migrations/20261009_material_purge_outbox/migration.sql', 'utf8') + const referenceSql = await fs.readFile('prisma/migrations/20261009_material_purge_reference_indexes/migration.sql', 'utf8') const baselineSql = `DROP TABLE material_purge_audits; DROP TABLE material_purge_jobs; ALTER TABLE material_logs ADD CONSTRAINT "material_logs_materialId_fkey" FOREIGN KEY ("materialId") REFERENCES materials(id) ON DELETE CASCADE; ALTER TABLE material_download_logs ADD CONSTRAINT "material_download_logs_materialId_fkey" FOREIGN KEY ("materialId") REFERENCES materials(id) ON DELETE CASCADE;` const migrated = spawnSync('docker', ['exec', '-i', 'gongxiangp2-purge-qa-20261009', 'psql', '-U', 'postgres', '-d', database, '-v', 'ON_ERROR_STOP=1'], { - encoding: 'utf8', input: `${baselineSql}\n${migrationSql}`, + encoding: 'utf8', input: `${baselineSql}\n${migrationSql}\n${referenceSql}`, }) assert.equal(migrated.status, 0, migrated.stderr) const { prisma } = await import('../lib/prisma') - const { enqueueMaterialPurge, processNextMaterialPurge, retryMaterialPurge } = await import('../lib/material-purge') - const { collectPurgeFiles, findPurgeFileReferences, purgeFileIdentity } = await import('../lib/material-purge-files') + const { enqueueMaterialPurge, processNextMaterialPurge, retryMaterialPurge, ensureMaterialPurgeWorkerStarted } = await import('../lib/material-purge') + const { collectPurgeFiles, findPurgeFileReferences, purgeFileIdentity, identifyPurgeFile } = await import('../lib/material-purge-files') + const { purgeReferenceLookupKeys } = await import('../lib/material-purge-reference-query') const run = randomUUID() const passed: string[] = [] passed.push('真实升级 SQL 创建清理表、重试索引与审计表，并解除历史日志的素材级联删除') @@ -168,6 +170,29 @@ async function main() { assert(audits.some(audit => audit.event === 'manual_retry' && audit.operatorUserId === admin.id)) passed.push('部分文件失败保存进度，自动重试耗尽后人工续跑；失败/重试操作人审计完整') + const indexedSource = await file('indexed-变更[1]%.bin') + const indexedFiles = [{ ...identifyPurgeFile(indexedSource)!, outcome: 'pending' as const }] + const indexedKeeper = await make('index-maintenance') + const indexedAsset = await prisma.materialAsset.create({ data: { materialId: indexedKeeper.id, assetType: 'test', fileUrl: indexedSource } }) + assert((await findPurgeFileReferences(prisma, indexedFiles)).has(purgeFileIdentity(indexedFiles[0]))) + await prisma.$executeRaw`UPDATE material_assets SET file_url = 'uploads/unrelated.bin' WHERE id = ${indexedAsset.id}` + assert(!(await findPurgeFileReferences(prisma, indexedFiles)).has(purgeFileIdentity(indexedFiles[0]))) + const jsonJob = await prisma.productPackageUploadJob.create({ data: { materialId: indexedKeeper.id, operatorUserId: admin.id, + assets: [{ nested: { fileUrl: indexedSource } }], status: 'queued', stage: 'queued' } }) + assert((await findPurgeFileReferences(prisma, indexedFiles)).has(purgeFileIdentity(indexedFiles[0]))) + await prisma.productPackageUploadJob.update({ where: { id: jsonJob.id }, data: { status: 'completed' } }) + assert(!(await findPurgeFileReferences(prisma, indexedFiles)).has(purgeFileIdentity(indexedFiles[0]))) + const aliases = [indexedSource, `/uploads/./indexed-变更[1]%.bin`, path.join(root, 'indexed-变更[1]%.bin'), + '\u00a0/UPLOADS/INDEXED-变更[1]%.BIN\ufeff', 'uploads\\indexed-变更[1]%.bin'] + for (const previewUrl of aliases) { + await prisma.material.update({ where: { id: indexedKeeper.id }, data: { previewUrl } }) + const full = await findPurgeFileReferences(prisma) + const targeted = await findPurgeFileReferences(prisma, indexedFiles) + assert.equal(targeted.has(purgeFileIdentity(indexedFiles[0])), full.has(purgeFileIdentity(indexedFiles[0])), previewUrl) + } + await prisma.material.update({ where: { id: indexedKeeper.id }, data: { previewUrl: 'data:image/png;base64,Zm9v' } }) + passed.push('数据库索引随 ORM/直接 SQL 写入和队列状态变化同步；Unicode、百分号、大小写、绝对路径、点路径引用与全量审计一致') + if (s3Enabled) { const { S3Client, CreateBucketCommand, PutObjectCommand, HeadObjectCommand } = await import('@aws-sdk/client-s3') const client = new S3Client({ endpoint: process.env.S3_ENDPOINT, region: 'us-east-1', forcePathStyle: true, @@ -208,6 +233,20 @@ async function main() { await enqueueMaterialPurge([display.id], admin.id) await processNextMaterialPurge() await client.send(new HeadObjectCommand({ Bucket, Key: displayKey })) + const referenceTarget = [{ ...identifyPurgeFile(source(sharedKey))!, outcome: 'pending' as const }] + for (const previewUrl of [ + source(sharedKey).replace('shared', '%73hared'), source(sharedKey).replace('shared', '%2573hared'), + `https://legacy.test/${sharedKey.replace('shared', '%73hared')}?cache=1#fragment`, + `https://legacy.test/${sharedKey.replace('shared', 'sha\tred')}`, + source(sharedKey).replace('/shared', '/unused/../shared'), + source(sharedKey).replace('/shared', '/%2e/shared'), + source(sharedKey).replace(`/${Bucket}/`, `//${Bucket}/`), + ]) { + await prisma.material.update({ where: { id: objectKeeper.id }, data: { previewUrl } }) + const full = await findPurgeFileReferences(prisma) + const targeted = await findPurgeFileReferences(prisma, referenceTarget) + assert.equal(targeted.has(purgeFileIdentity(referenceTarget[0])), full.has(purgeFileIdentity(referenceTarget[0])), previewUrl) + } const objectForeign = await make('object-foreign', { zipFilePath: source(foreignKey) }) await assert.rejects(enqueueMaterialPurge([objectForeign.id], admin.id), /业务前缀之外/) await client.send(new HeadObjectCommand({ Bucket, Key: foreignKey })) @@ -216,6 +255,55 @@ async function main() { client.destroy() } + if (process.argv.includes('--scale')) { + const scaleKeeper = await make('400k-reference-keeper') + for (let offset = 0; offset < 400_000; offset += 80_000) { + await prisma.$executeRaw`INSERT INTO material_assets (id, material_id, asset_type, file_url, sort_order, updated_at) + SELECT ${run + '-scale-'} || i::text, ${scaleKeeper.id}, 'qa-scale', + 'https://isolated.test/project/scale/' || i::text || '/image.bin', 0, now() FROM generate_series(${offset}, ${offset + 79_999}) AS i` + console.log(JSON.stringify({ mode: 'isolated-scale-setup', assets: offset + 80_000 })) + } + await prisma.$executeRaw`ANALYZE material_assets` + const scaleSources = await Promise.all(Array.from({ length: 4 }, (_, i) => file(`scale-delete-${i}.bin`))) + const scaleMaterial = await make('scale-delete', { zipFilePath: scaleSources[0], displayImages: scaleSources.slice(1) }) + const scaleFiles = await collectPurgeFiles({ ...scaleMaterial, assets: [], skus: [] }) + const keys = purgeReferenceLookupKeys(scaleFiles) + const plan = await prisma.$queryRaw`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) + SELECT file_url FROM material_assets WHERE purge_reference_keys(ARRAY[file_url], true) && ${keys}::text[]` + assert(JSON.stringify(plan).includes('material_assets_purge_references_idx'), '40万行必须实际使用引用索引') + const referenceStarted = Date.now() + await findPurgeFileReferences(prisma, scaleFiles) + const referenceMs = Date.now() - referenceStarted + const submitted = Date.now() + const [scaleJob] = await enqueueMaterialPurge([scaleMaterial.id], admin.id) + const acceptedMs = Date.now() - submitted + await processNextMaterialPurge() + const completedMs = Date.now() - submitted + const result = await prisma.materialPurgeJob.findUniqueOrThrow({ where: { id: scaleJob.id } }) + assert.equal(result.status, 'completed') + assert((await Promise.all(scaleSources.map((_, i) => exists(`scale-delete-${i}.bin`)))).every(value => !value)) + assert(referenceMs < 500 && completedMs < 1500, `索引检查 ${referenceMs}ms / 实际删除 ${completedMs}ms 超出验收线`) + await fs.mkdir('tmp/material-purge-index-qa', { recursive: true }) + await fs.writeFile('tmp/material-purge-index-qa/scale.json', JSON.stringify({ assets: await prisma.materialAsset.count(), referenceMs, acceptedMs, completedMs, plan }, null, 2)) + passed.push(`40万资产下只查目标引用 ${referenceMs}ms；提交 ${acceptedMs}ms、4个文件真实删除及审计完成 ${completedMs}ms，执行计划使用 GIN 索引`) + } + + delete process.env.NEXT_BUILD + ensureMaterialPurgeWorkerStarted() + await new Promise(resolve => setTimeout(resolve, 100)) + const wake = await make('wake-up', { zipFilePath: await file('wake-up.bin') }) + const [wakeJob] = await enqueueMaterialPurge([wake.id], admin.id) + const wakeStarted = Date.now() + ensureMaterialPurgeWorkerStarted() + let wakeResult + do { + await new Promise(resolve => setTimeout(resolve, 25)) + wakeResult = await prisma.materialPurgeJob.findUniqueOrThrow({ where: { id: wakeJob.id } }) + } while (wakeResult.status !== 'completed' && Date.now() - wakeStarted < 1500) + assert.equal(wakeResult.status, 'completed', '新请求必须直接唤醒已启动的 worker，不能等 5 秒轮询') + process.env.NEXT_BUILD = '1' + passed.push(`已启动的 worker 收到提交立即清理，${Date.now() - wakeStarted}ms 完成，未等待5秒轮询`) + const output = { database: url.pathname.slice(1), passed, checks: passed.length, at: new Date().toISOString() } await fs.mkdir('tmp/material-purge-qa', { recursive: true }) await fs.writeFile('tmp/material-purge-qa/integration.json', JSON.stringify(output, null, 2))