fix: index file references for recycle-bin cleanup
This commit is contained in:
parent
de2d7595ce
commit
477ff64b07
@ -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。这里提前失败，避免再把缺路由的镜像发出去。
|
||||
|
||||
19
docs/回收站删除索引修复-2026-10-09.md
Normal file
19
docs/回收站删除索引修复-2026-10-09.md
Normal file
@ -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 个文件和现存引用样本，不由自动化提交真实素材删除。隔离测试耗时不等于生产完整物理清理耗时；生产查询耗时和发布凭据另存于发布回执。
|
||||
@ -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)
|
||||
|
||||
33
lib/__tests__/material-purge-worker.test.ts
Normal file
33
lib/__tests__/material-purge-worker.test.ts
Normal file
@ -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<void> }
|
||||
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
|
||||
}
|
||||
})
|
||||
@ -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,62 +134,20 @@ function stringsIn(value: unknown): string[] {
|
||||
return []
|
||||
}
|
||||
|
||||
/**
|
||||
* 常规路径按本次文件的完整地址匹配；编码、控制字符和 URL 点路径
|
||||
* 不能按字面排除，全部纳入候选。最终仍由原生路径校验核对完整身份。
|
||||
* 白底派生图的本地引用也单独纳入，不依赖源文件与公网文件的路径相同。
|
||||
* 包含回收站与 SKU 软删除记录，保留其他素材恢复时仍需要的共用文件。
|
||||
*/
|
||||
/** 生产清理按目标文件查索引；无 files 参数的全量读取仅供只读审计核对。 */
|
||||
export async function findPurgeFileReferences(db: Prisma.TransactionClient, files?: readonly PurgeFile[]): Promise<Set<string>> {
|
||||
if (files && !files.length) return new Set()
|
||||
const caseFold = process.platform === 'win32'
|
||||
const sources = files ? new Set<string>() : 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)
|
||||
}
|
||||
}
|
||||
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([
|
||||
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 } }),
|
||||
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.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 } }),
|
||||
@ -197,31 +156,34 @@ export async function findPurgeFileReferences(db: Prisma.TransactionClient, file
|
||||
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 targets = files && new Set(files.map(purgeFileIdentity))
|
||||
const identities = new Set<string>()
|
||||
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
|
||||
|
||||
60
lib/material-purge-reference-query.ts
Normal file
60
lib/material-purge-reference-query.ts
Normal file
@ -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<string>(['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<ReferenceRow[]> {
|
||||
const keys = purgeReferenceLookupKeys(files)
|
||||
return db.$queryRaw<ReferenceRow[]>`
|
||||
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[]
|
||||
`
|
||||
}
|
||||
@ -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<void> }
|
||||
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) { /* 限定每轮批量 */ } }
|
||||
catch (error) { console.error('[MaterialPurge] 清理队列暂不可用，将在下一轮重试:', error) }
|
||||
finally { running = false }
|
||||
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
|
||||
// 入队与当前空队列检查交错时保留唤醒信号；满一轮继续处理下一批。
|
||||
if (wakeRequested || more) setImmediate(() => { void pump() }).unref?.()
|
||||
}
|
||||
}
|
||||
// 入队事务提交后直接唤醒；轮询仅用于重启恢复和延迟重试，不让新请求等下一轮。
|
||||
global.__materialPurgePump = pump
|
||||
setInterval(() => { void pump() }, 5_000).unref?.()
|
||||
void pump()
|
||||
}
|
||||
|
||||
@ -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;
|
||||
@ -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)
|
||||
|
||||
@ -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) {
|
||||
|
||||
@ -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/)
|
||||
})
|
||||
|
||||
@ -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<any[]>`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))
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user