diff --git a/furniture-3d-system/docs/workflow/产品全量档案图片索引操作说明.md b/furniture-3d-system/docs/workflow/产品全量档案图片索引操作说明.md new file mode 100644 index 00000000..b7d424a2 --- /dev/null +++ b/furniture-3d-system/docs/workflow/产品全量档案图片索引操作说明.md @@ -0,0 +1,54 @@ +# 产品全量档案图片索引操作说明 + +全量档案本地图片索引只读取 `SupplyChainProductArchive.image`。上游「我司 SKU 命中」继续使用现有独立图搜服务。 + +## 更新规则 + +- 有图片且没有索引：建立索引。 +- 图片地址变化、旧索引失效、模型版本或向量规格变化：重新索引。 +- 图片为空或地址已经变化：旧向量退出本地搜索。任务会把旧记录标为 `STALE`。 +- 下载或计算失败：保存 `FAILED` 和原因，六小时后再次尝试；图片地址再次变化则立即处理。 +- 图片内容相同、地址不同：复用向量并更新地址与状态，避免反复处理同一张图片。 +- 下载、计算期间图片改变或产品删除：提交前锁定并核对档案行，丢弃过期结果，下一轮处理最新图片。 +- 所有索引任务使用同一 PostgreSQL 会话锁，避免手动补索引、同步触发与定时任务并发计算。 +- 正常增量任务不会重新下载全部已就绪图片；同一 URL 下替换了图片内容时，使用完整内容核验命令。 + +## 本地命令 + +在主应用目录执行，模型沿用现有 `IMAGE_SEARCH_MODEL_PATH` 或 `.local-services/models/dinov2-small`。 + +```bash +# 每次最多处理 200 条缺失、换图或到期重试记录 +npm run image-search:index + +# 补齐当前全部待处理记录，并立即重试历史失败 +npm run image-search:index -- --limit=0 --retry-now + +# 下载所有带图档案，核验同一 URL 下的图片内容变化 +npm run image-search:index -- --all --limit=0 +``` + +命令持续输出已处理、成功、复用、失败、跳过、过期结果丢弃和失效条数；失败行输出条码与原因。最后输出本轮结果及待处理数量。失败仍会进入后续重试，本轮失败数必须单独关注，`remaining` 只计算已到重试时间的记录。 + +## 生产启用 + +代码必须先进入被授权的 `origin/main`，再经统一部署入口发布。安装定时器与首次补索引按用户授权执行，不能把本地开发完成等同于生产已启用。 + +安装入口：`scripts/install-product-archive-image-index-timer.sh`。它要求现有生产模型文件齐全，并让任务显式使用共享模型目录。当前应用启动命令已注入模型路径，无须为本功能重写应用环境。 + +- 服务：`furniture-product-archive-image-index.service`。 +- 定时器：`furniture-product-archive-image-index.timer`，每五分钟检查，单次最多处理 200 条，失败六小时后重试。 +- 两个现有全量同步服务分别增加本功能自己的 `image-index.conf`，同步成功后异步启动索引服务。 +- 前台保存、导入及其他来源的图片变化由五分钟定时检查覆盖；索引下载与推理不阻塞保存请求。 +- 通过 `PRODUCT_ARCHIVE_IMAGE_INDEX_LIMIT` 可在安装时调整单次处理上限。 +- 脚本和日志通过发布目录与 systemd 管理，不手工修改发布目录。 + +```bash +bash scripts/install-product-archive-image-index-timer.sh +systemctl status furniture-product-archive-image-index.timer --no-pager +journalctl -u furniture-product-archive-image-index.service -n 50 --no-pager +``` + +首次补齐历史缺口可等待定时任务分批完成；需要一次补完时，手动索引命令必须指向共享模型路径。运行中的另一索引任务会返回 `LOCKED`，稍后重试。 + +验收应核对有图档案与索引的关联，而不是只看 `READY` 行数：当前地址一致、状态 `READY`、模型版本与 768 维向量均符合要求才是可用索引。历史 404 或不可读取地址不能靠索引任务生成图片，需要按条码定位源图片问题。 diff --git a/furniture-3d-system/scripts/index-product-archive-images.ts b/furniture-3d-system/scripts/index-product-archive-images.ts index 49dac9c7..22ffff7d 100644 --- a/furniture-3d-system/scripts/index-product-archive-images.ts +++ b/furniture-3d-system/scripts/index-product-archive-images.ts @@ -9,54 +9,35 @@ function readIntegerFlag(name: string, fallback: number) { return Number.isInteger(parsed) && parsed >= 0 ? parsed : fallback } +function hasFlag(name: string) { + return process.argv.includes(name) +} + async function main() { const [ - { productArchiveImageSearchRepository }, - { - indexProductArchiveImageCandidate, - isProductArchiveImageSearchModelReady, - }, + { productArchiveImageIndexRunner }, + { prisma }, ] = await Promise.all([ - import('@/server/repositories/product-archive-image-search.repository'), - import('@/server/services/product-archive-image-search.service'), + import('@/server/services/product-archive-image-index-runner.service'), + import('@/lib/prisma'), ]) - if (!isProductArchiveImageSearchModelReady()) { - throw new Error('图片模型未就绪，请先运行 npm run image-search:model:prepare') - } - - const limit = readIntegerFlag('--limit', 100) + const limit = readIntegerFlag('--limit', 200) const batchSize = Math.min(50, Math.max(1, readIntegerFlag('--batch-size', 10))) - let cursor: string | undefined - let processed = 0 - let indexed = 0 - let unchanged = 0 - let failed = 0 - - while (limit === 0 || processed < limit) { - const candidates = await productArchiveImageSearchRepository.listCandidates({ - cursor, - take: Math.min(batchSize, limit === 0 ? batchSize : limit - processed), + try { + const result = await productArchiveImageIndexRunner.run({ + limit, + batchSize, + full: hasFlag('--all'), + retryNow: hasFlag('--retry-now'), + onProgress: (progress) => { + if (progress.error || progress.processed % 10 === 0 || progress.processed === 1) console.log(JSON.stringify({ progress })) + }, }) - if (candidates.length === 0) break - - for (const candidate of candidates) { - cursor = candidate.id - processed += 1 - const result = await indexProductArchiveImageCandidate(candidate) - if (result.status === 'INDEXED') indexed += 1 - else if (result.status === 'UNCHANGED') unchanged += 1 - else if (result.status === 'FAILED') failed += 1 - if (processed % 10 === 0 || processed === 1) { - console.log(JSON.stringify({ processed, indexed, unchanged, failed, cursor })) - } - if (limit > 0 && processed >= limit) break - } - - if (candidates.length < batchSize) break + console.log(JSON.stringify({ completed: result.status === 'COMPLETED', ...result })) + } finally { + await prisma.$disconnect() } - - console.log(JSON.stringify({ completed: true, processed, indexed, unchanged, failed })) } void main().catch((error) => { diff --git a/furniture-3d-system/scripts/install-product-archive-image-index-timer.sh b/furniture-3d-system/scripts/install-product-archive-image-index-timer.sh new file mode 100644 index 00000000..56f59010 --- /dev/null +++ b/furniture-3d-system/scripts/install-product-archive-image-index-timer.sh @@ -0,0 +1,74 @@ +#!/usr/bin/env bash +set -euo pipefail + +BASE="${FURNITURE_PROD_BASE:-/opt/furniture-3d-system-prod}" +APP_ROOT="$BASE/current/furniture-3d-system" +SERVICE_NAME=furniture-product-archive-image-index +MODEL_DIR="$BASE/shared/models/dinov2-small" +INDEX_LIMIT="${PRODUCT_ARCHIVE_IMAGE_INDEX_LIMIT:-200}" + +if [ "$(id -u)" -ne 0 ]; then + echo "This installer must run as root." >&2 + exit 1 +fi +if [ ! -f "$APP_ROOT/scripts/index-product-archive-images.ts" ] || [ ! -f "$BASE/shared/.env" ]; then + echo "Missing application script or shared environment." >&2 + exit 1 +fi +if [ ! -s "$MODEL_DIR/config.json" ] || [ ! -s "$MODEL_DIR/preprocessor_config.json" ] || [ ! -s "$MODEL_DIR/onnx/model_quantized.onnx" ]; then + echo "Image search model is not ready at $MODEL_DIR." >&2 + exit 1 +fi +if ! [[ "$INDEX_LIMIT" =~ ^[1-9][0-9]*$ ]]; then + echo "PRODUCT_ARCHIVE_IMAGE_INDEX_LIMIT must be a positive integer." >&2 + exit 1 +fi + +cat >"/etc/systemd/system/$SERVICE_NAME.service" <"/etc/systemd/system/$SERVICE_NAME.timer" </dev/null 2>&1; then + mkdir -p "/etc/systemd/system/$sync_unit.service.d" + cat >"/etc/systemd/system/$sync_unit.service.d/image-index.conf" < ({ + prisma: { + $queryRaw: jest.fn(), + $executeRaw: jest.fn(), + $transaction: jest.fn(), + }, +})) + +import { prisma } from '@/lib/prisma' +import { productArchiveImageIndexRepository } from '../product-archive-image-index.repository' + +const queryRaw = prisma.$queryRaw as jest.Mock +const executeRaw = prisma.$executeRaw as jest.Mock +const transaction = prisma.$transaction as jest.Mock + +function sqlText(sql: Prisma.Sql) { + return sql.strings.join('') +} + +const SELECTION = { + modelVersion: 'dinov2-small-shape-v1', + dimensions: 768, + retryBefore: new Date('2026-10-10T00:00:00.000Z'), +} + +describe('productArchiveImageIndexRepository.listPendingCandidates', () => { + beforeEach(() => jest.clearAllMocks()) + + it('增量模式：SQL 同时覆盖缺失、换图、失效、模型版本、向量规格与到期重试', async () => { + const rows = [{ id: 'a1', image: 'u1' }] + queryRaw.mockResolvedValue(rows) + + const result = await productArchiveImageIndexRepository.listPendingCandidates({ ...SELECTION, take: 5 }) + + expect(result).toBe(rows) + const sql = queryRaw.mock.calls[0][0] as Prisma.Sql + expect(sqlText(sql)).toContain('FROM "SupplyChainProductArchive" a') + expect(sqlText(sql)).toContain('LEFT JOIN "ProductArchiveImageEmbedding" e ON e."archiveId" = a.id') + expect(sqlText(sql)).toContain(`btrim(coalesce(a.image, '')) <> ''`) + expect(sqlText(sql)).toContain('e.id IS NULL') + expect(sqlText(sql)).toContain('e."imageUrl" <> btrim(a.image)') + expect(sqlText(sql)).toContain(`e.status = 'STALE'`) + expect(sqlText(sql)).toContain('e."modelVersion" <> ') + expect(sqlText(sql)).toContain('e.dimensions <> ') + expect(sqlText(sql)).toContain('octet_length(e.embedding) <> ') + expect(sqlText(sql)).toContain(`e.status <> 'READY' AND e."updatedAt" <= `) + expect(sqlText(sql)).toContain('ORDER BY a.id ASC') + expect(sql.values).toEqual( + expect.arrayContaining([SELECTION.modelVersion, SELECTION.dimensions, SELECTION.dimensions * 4, SELECTION.retryBefore, 5]), + ) + expect(sqlText(sql)).not.toContain('a.id >') + }) + + it('带 afterId 游标时追加主键范围条件', async () => { + queryRaw.mockResolvedValue([]) + await productArchiveImageIndexRepository.listPendingCandidates({ ...SELECTION, afterId: 'a9', take: 5 }) + const sql = queryRaw.mock.calls[0][0] as Prisma.Sql + expect(sqlText(sql)).toContain('a.id > ') + expect(sql.values).toContain('a9') + }) + + it('full 模式核验全部带图档案，不再套用增量过滤', async () => { + queryRaw.mockResolvedValue([]) + await productArchiveImageIndexRepository.listPendingCandidates({ ...SELECTION, full: true, take: 5 }) + const sql = queryRaw.mock.calls[0][0] as Prisma.Sql + expect(sqlText(sql)).toContain('TRUE') + expect(sqlText(sql)).not.toContain('e."imageUrl" <> btrim(a.image)') + }) +}) + +describe('productArchiveImageIndexRepository.countPendingCandidates', () => { + beforeEach(() => jest.clearAllMocks()) + + it('返回待处理条数，过滤条件与列表一致', async () => { + queryRaw.mockResolvedValue([{ count: 7 }]) + const count = await productArchiveImageIndexRepository.countPendingCandidates(SELECTION) + expect(count).toBe(7) + const sql = queryRaw.mock.calls[0][0] as Prisma.Sql + expect(sqlText(sql)).toContain('count(*)::int AS count') + expect(sqlText(sql)).toContain('e."imageUrl" <> btrim(a.image)') + }) + + it('空结果按 0 处理', async () => { + queryRaw.mockResolvedValue([]) + await expect(productArchiveImageIndexRepository.countPendingCandidates(SELECTION)).resolves.toBe(0) + }) +}) + +describe('productArchiveImageIndexRepository.invalidateStaleEmbeddings', () => { + beforeEach(() => jest.clearAllMocks()) + + it('把删图或换图的旧向量标为 STALE', async () => { + executeRaw.mockResolvedValue(3) + await expect(productArchiveImageIndexRepository.invalidateStaleEmbeddings()).resolves.toBe(3) + const sql = executeRaw.mock.calls[0][0] as Prisma.Sql + expect(sqlText(sql)).toContain(`SET status = 'STALE'`) + expect(sqlText(sql)).toContain(`e.status <> 'STALE'`) + expect(sqlText(sql)).toContain(`btrim(coalesce(a.image, '')) = '' OR e."imageUrl" <> btrim(a.image)`) + }) +}) + +describe('productArchiveImageIndexRepository.saveIfCurrent', () => { + const write = { + archiveId: 'a1', + imageUrl: 'u1', + imageHash: 'h1', + embedding: new Uint8Array(16), + dimensions: 4, + modelVersion: 'dinov2-small-shape-v1', + status: 'READY' as const, + errorMessage: null, + indexedAt: new Date('2026-10-10T00:00:00.000Z'), + } + + function mockTransaction(image: string | null | undefined) { + const tx = { + $queryRaw: jest.fn().mockResolvedValue(image === undefined ? [] : [{ image }]), + productArchiveImageEmbedding: { upsert: jest.fn().mockResolvedValue(undefined) }, + } + transaction.mockImplementation(async (fn: (tx: typeof tx) => Promise) => fn(tx)) + return tx + } + + beforeEach(() => jest.clearAllMocks()) + + it('档案行当前 URL 一致（含首尾空格）时提交 upsert', async () => { + const tx = mockTransaction(' u1 ') + await expect(productArchiveImageIndexRepository.saveIfCurrent(write)).resolves.toBe(true) + expect(tx.productArchiveImageEmbedding.upsert).toHaveBeenCalledWith({ + where: { archiveId: 'a1' }, + create: expect.objectContaining({ imageUrl: 'u1', status: 'READY', imageHash: 'h1' }), + update: expect.objectContaining({ imageUrl: 'u1', status: 'READY', imageHash: 'h1' }), + }) + }) + + it('档案图片已换成别的地址时丢弃结果，不写库', async () => { + const tx = mockTransaction('other') + await expect(productArchiveImageIndexRepository.saveIfCurrent(write)).resolves.toBe(false) + expect(tx.productArchiveImageEmbedding.upsert).not.toHaveBeenCalled() + }) + + it('档案行已删除（查不到行）时丢弃结果', async () => { + const tx = mockTransaction(undefined) + await expect(productArchiveImageIndexRepository.saveIfCurrent(write)).resolves.toBe(false) + expect(tx.productArchiveImageEmbedding.upsert).not.toHaveBeenCalled() + }) +}) diff --git a/furniture-3d-system/src/server/repositories/__tests__/product-archive-image-search.repository.test.ts b/furniture-3d-system/src/server/repositories/__tests__/product-archive-image-search.repository.test.ts new file mode 100644 index 00000000..3d141839 --- /dev/null +++ b/furniture-3d-system/src/server/repositories/__tests__/product-archive-image-search.repository.test.ts @@ -0,0 +1,110 @@ +import type { Prisma } from '@prisma/client' +import { + PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS, +} from '@/lib/product-archive-image-search' + +jest.mock('@/lib/prisma', () => ({ + prisma: { + $queryRaw: jest.fn(), + supplyChainProductArchive: { findMany: jest.fn() }, + productArchiveImageEmbedding: { findUnique: jest.fn() }, + }, +})) + +import { prisma } from '@/lib/prisma' +import { productArchiveImageSearchRepository } from '../product-archive-image-search.repository' + +const queryRaw = prisma.$queryRaw as jest.Mock +const archiveFindMany = prisma.supplyChainProductArchive.findMany as jest.Mock +const embeddingFindUnique = prisma.productArchiveImageEmbedding.findUnique as jest.Mock + +function sqlText(sql: Prisma.Sql) { + return sql.strings.join('') +} + +describe('productArchiveImageSearchRepository 就绪向量判定', () => { + beforeEach(() => jest.clearAllMocks()) + + it('listReadyEmbeddings 只取当前地址一致、规格正确的 READY 向量', async () => { + const rows = [{ archiveId: 'a1', imageUrl: 'u1' }] + queryRaw.mockResolvedValue(rows) + + const result = await productArchiveImageSearchRepository.listReadyEmbeddings('dinov2-small-shape-v1') + + expect(result).toBe(rows) + const sql = queryRaw.mock.calls[0][0] as Prisma.Sql + expect(sqlText(sql)).toContain('FROM "ProductArchiveImageEmbedding" e') + expect(sqlText(sql)).toContain('JOIN "SupplyChainProductArchive" a ON a.id = e."archiveId"') + expect(sqlText(sql)).toContain(`e.status = 'READY'`) + expect(sqlText(sql)).toContain(`btrim(coalesce(a.image, '')) <> '' AND e."imageUrl" = btrim(a.image)`) + expect(sqlText(sql)).toContain('e.dimensions = ') + expect(sqlText(sql)).toContain('octet_length(e.embedding) = ') + expect(sql.values).toEqual([ + 'dinov2-small-shape-v1', + PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS, + PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS * 4, + ]) + }) + + it('countReadyEmbeddings 用同一口径统计，缺行时返回 0', async () => { + queryRaw.mockResolvedValue([{ count: 5 }]) + await expect(productArchiveImageSearchRepository.countReadyEmbeddings('dinov2-small-shape-v1')).resolves.toBe(5) + + queryRaw.mockResolvedValue([]) + await expect(productArchiveImageSearchRepository.countReadyEmbeddings('dinov2-small-shape-v1')).resolves.toBe(0) + + const sql = queryRaw.mock.calls[1][0] as Prisma.Sql + expect(sqlText(sql)).toContain('count(*)::int AS count') + expect(sqlText(sql)).toContain(`e."imageUrl" = btrim(a.image)`) + }) +}) + +describe('productArchiveImageSearchRepository 档案候选查询', () => { + beforeEach(() => jest.clearAllMocks()) + + it('findCandidatesByCodes 去重、去空白后按条码与 SKU 不区分大小写匹配', async () => { + const rows = [{ id: 'a1' }] + archiveFindMany.mockResolvedValue(rows) + + const result = await productArchiveImageSearchRepository.findCandidatesByCodes({ + barcodes: [' 6901234500018 ', '6901234500018', ''], + skus: ['ZSJ008103N958V01'], + }) + + expect(result).toBe(rows) + expect(archiveFindMany).toHaveBeenCalledTimes(1) + const where = archiveFindMany.mock.calls[0][0].where + expect(where.OR).toEqual([ + { barcode: { in: ['6901234500018'], mode: 'insensitive' } }, + { sku: { in: ['ZSJ008103N958V01'], mode: 'insensitive' } }, + ]) + }) + + it('编码集合为空时不查库直接返回空', async () => { + await expect(productArchiveImageSearchRepository.findCandidatesByCodes({ barcodes: [' ', ''], skus: [] })) + .resolves.toEqual([]) + expect(archiveFindMany).not.toHaveBeenCalled() + }) + + it('listCandidates 带 cursor 时从游标后一条继续', async () => { + archiveFindMany.mockResolvedValue([]) + await productArchiveImageSearchRepository.listCandidates({ cursor: 'a1', take: 10 }) + const params = archiveFindMany.mock.calls[0][0] + expect(params.cursor).toEqual({ id: 'a1' }) + expect(params.skip).toBe(1) + expect(params.take).toBe(10) + expect(params.where).toEqual({ image: { not: null } }) + }) + + it('getCandidatesByIds 空 ID 集合不查库', async () => { + await expect(productArchiveImageSearchRepository.getCandidatesByIds([])).resolves.toEqual([]) + expect(archiveFindMany).not.toHaveBeenCalled() + }) + + it('getEmbeddingByArchiveId 按唯一键读取向量行', async () => { + const row = { archiveId: 'a1', imageUrl: 'u1' } + embeddingFindUnique.mockResolvedValue(row) + await expect(productArchiveImageSearchRepository.getEmbeddingByArchiveId('a1')).resolves.toBe(row) + expect(embeddingFindUnique).toHaveBeenCalledWith(expect.objectContaining({ where: { archiveId: 'a1' } })) + }) +}) diff --git a/furniture-3d-system/src/server/repositories/product-archive-image-index.repository.ts b/furniture-3d-system/src/server/repositories/product-archive-image-index.repository.ts new file mode 100644 index 00000000..f6b49a9a --- /dev/null +++ b/furniture-3d-system/src/server/repositories/product-archive-image-index.repository.ts @@ -0,0 +1,103 @@ +import { Prisma } from '@prisma/client' +import { prisma } from '@/lib/prisma' +import type { ProductArchiveImageCandidate } from './product-archive-image-search.repository' + +export type ProductArchiveImageIndexSelection = { + modelVersion: string + dimensions: number + retryBefore: Date + full?: boolean +} + +export type ProductArchiveImageIndexWrite = { + archiveId: string + imageUrl: string + imageHash: string + embedding: Uint8Array + dimensions: number + modelVersion: string + status: 'READY' | 'FAILED' + errorMessage: string | null + indexedAt: Date | null +} + +export interface ProductArchiveImageIndexRepository { + listPendingCandidates(input: ProductArchiveImageIndexSelection & { + afterId?: string + take: number + }): Promise + countPendingCandidates(input: ProductArchiveImageIndexSelection): Promise + invalidateStaleEmbeddings(): Promise + saveIfCurrent(input: ProductArchiveImageIndexWrite): Promise +} + +function pendingWhere(input: ProductArchiveImageIndexSelection) { + if (input.full) return Prisma.sql`TRUE` + return Prisma.sql`( + e.id IS NULL + OR e."imageUrl" <> btrim(a.image) + OR e.status = 'STALE' + OR e."modelVersion" <> ${input.modelVersion} + OR (e.status = 'READY' AND ( + e.dimensions <> ${input.dimensions} + OR octet_length(e.embedding) <> ${input.dimensions * 4} + )) + OR (e.status <> 'READY' AND e."updatedAt" <= ${input.retryBefore}) + )` +} + +export class ProductArchiveImageIndexRepositoryClass implements ProductArchiveImageIndexRepository { + async listPendingCandidates(input: ProductArchiveImageIndexSelection & { + afterId?: string + take: number + }) { + return prisma.$queryRaw(Prisma.sql` + SELECT a.id, a.image, a.spu, a.sku, a.barcode, a."productName" + FROM "SupplyChainProductArchive" a + LEFT JOIN "ProductArchiveImageEmbedding" e ON e."archiveId" = a.id + WHERE btrim(coalesce(a.image, '')) <> '' AND ${pendingWhere(input)} + ${input.afterId ? Prisma.sql`AND a.id > ${input.afterId}` : Prisma.empty} + ORDER BY a.id ASC + LIMIT ${input.take} + `) + } + + async countPendingCandidates(input: ProductArchiveImageIndexSelection) { + const rows = await prisma.$queryRaw>(Prisma.sql` + SELECT count(*)::int AS count + FROM "SupplyChainProductArchive" a + LEFT JOIN "ProductArchiveImageEmbedding" e ON e."archiveId" = a.id + WHERE btrim(coalesce(a.image, '')) <> '' AND ${pendingWhere(input)} + `) + return rows[0]?.count ?? 0 + } + + async invalidateStaleEmbeddings() { + return prisma.$executeRaw(Prisma.sql` + UPDATE "ProductArchiveImageEmbedding" e + SET status = 'STALE', "indexedAt" = NULL, "updatedAt" = NOW() + FROM "SupplyChainProductArchive" a + WHERE a.id = e."archiveId" AND e.status <> 'STALE' + AND (btrim(coalesce(a.image, '')) = '' OR e."imageUrl" <> btrim(a.image)) + `) + } + + async saveIfCurrent(input: ProductArchiveImageIndexWrite) { + // Hold the archive row only while committing. Downloads and inference happen outside this transaction. + return prisma.$transaction(async (tx) => { + const rows = await tx.$queryRaw>(Prisma.sql` + SELECT image FROM "SupplyChainProductArchive" WHERE id = ${input.archiveId} FOR UPDATE + `) + if (rows.length === 0 || rows[0].image?.trim() !== input.imageUrl) return false + const data = { ...input, embedding: input.embedding as Uint8Array } + await tx.productArchiveImageEmbedding.upsert({ + where: { archiveId: input.archiveId }, + create: data, + update: data, + }) + return true + }) + } +} + +export const productArchiveImageIndexRepository = new ProductArchiveImageIndexRepositoryClass() diff --git a/furniture-3d-system/src/server/repositories/product-archive-image-search.repository.ts b/furniture-3d-system/src/server/repositories/product-archive-image-search.repository.ts index 8fc2be77..69c0df73 100644 --- a/furniture-3d-system/src/server/repositories/product-archive-image-search.repository.ts +++ b/furniture-3d-system/src/server/repositories/product-archive-image-search.repository.ts @@ -1,5 +1,6 @@ import { Prisma } from '@prisma/client' import { prisma } from '@/lib/prisma' +import { PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS } from '@/lib/product-archive-image-search' const eligibleArchiveWhere = { image: { not: null }, @@ -66,7 +67,7 @@ export interface ProductArchiveImageSearchRepository { countReadyEmbeddings(modelVersion: string): Promise } -class ProductArchiveImageSearchRepositoryClass +export class ProductArchiveImageSearchRepositoryClass implements ProductArchiveImageSearchRepository { async listCandidates(params: { cursor?: string; take: number }) { @@ -139,10 +140,15 @@ class ProductArchiveImageSearchRepositoryClass } async listReadyEmbeddings(modelVersion: string) { - return prisma.productArchiveImageEmbedding.findMany({ - where: { status: 'READY', modelVersion }, - select: embeddingSelect, - }) + return prisma.$queryRaw(Prisma.sql` + SELECT e."archiveId", e."imageUrl", e."imageHash", e.embedding, e.dimensions, e."modelVersion" + FROM "ProductArchiveImageEmbedding" e + JOIN "SupplyChainProductArchive" a ON a.id = e."archiveId" + WHERE e.status = 'READY' AND e."modelVersion" = ${modelVersion} + AND btrim(coalesce(a.image, '')) <> '' AND e."imageUrl" = btrim(a.image) + AND e.dimensions = ${PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS} + AND octet_length(e.embedding) = ${PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS * 4} + `) } async getEmbeddingByArchiveId(archiveId: string) { @@ -161,9 +167,16 @@ class ProductArchiveImageSearchRepositoryClass } async countReadyEmbeddings(modelVersion: string) { - return prisma.productArchiveImageEmbedding.count({ - where: { status: 'READY', modelVersion }, - }) + const rows = await prisma.$queryRaw>(Prisma.sql` + SELECT count(*)::int AS count + FROM "ProductArchiveImageEmbedding" e + JOIN "SupplyChainProductArchive" a ON a.id = e."archiveId" + WHERE e.status = 'READY' AND e."modelVersion" = ${modelVersion} + AND btrim(coalesce(a.image, '')) <> '' AND e."imageUrl" = btrim(a.image) + AND e.dimensions = ${PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS} + AND octet_length(e.embedding) = ${PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS * 4} + `) + return rows[0]?.count ?? 0 } } diff --git a/furniture-3d-system/src/server/services/__tests__/product-archive-image-index-runner.service.test.ts b/furniture-3d-system/src/server/services/__tests__/product-archive-image-index-runner.service.test.ts new file mode 100644 index 00000000..a557c392 --- /dev/null +++ b/furniture-3d-system/src/server/services/__tests__/product-archive-image-index-runner.service.test.ts @@ -0,0 +1,151 @@ +import { + PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS, + PRODUCT_ARCHIVE_IMAGE_SEARCH_MODEL_VERSION, +} from '@/lib/product-archive-image-search' +import { PRODUCT_ARCHIVE_IMAGE_INDEX_RETRY_MS } from '@/lib/product-archive-image-index' + +jest.mock('@/server/repositories/product-archive-image-index.repository', () => ({ + productArchiveImageIndexRepository: {}, +})) +jest.mock('@/server/services/product-archive-image-index.service', () => ({ + productArchiveImageIndexService: {}, +})) +jest.mock('@/server/services/product-archive-image-index-lock.service', () => ({ + productArchiveImageIndexLock: {}, +})) +jest.mock('@/server/services/product-archive-image-shape', () => ({ + isProductArchiveImageSearchModelReady: jest.fn(), +})) + +import { ProductArchiveImageIndexRunner } from '@/server/services/product-archive-image-index-runner.service' +import type { ProductArchiveImageIndexOutcome } from '@/lib/product-archive-image-index' + +function candidate(id: string) { + return { + id, + image: `/api/storage/object?key=archive%2F${id}.jpg`, + spu: 'ZSJ0081', + sku: 'ZSJ008103N958V01', + barcode: `69012345000${id.length}`, + productName: '示例家具', + } +} + +function createRunner(candidatesByBatch: string[][], outcomes: Record) { + const repository = { + invalidateStaleEmbeddings: jest.fn().mockResolvedValue(2), + listPendingCandidates: jest.fn().mockImplementation(async () => candidatesByBatch.shift() ?? []), + countPendingCandidates: jest.fn().mockResolvedValue(4), + } + const indexer = { + indexCandidate: jest.fn().mockImplementation(async (item: { id: string }) => + outcomes[item.id] ?? { status: 'INDEXED' as const }), + } + const release = jest.fn().mockResolvedValue(undefined) + const lock = { acquire: jest.fn().mockResolvedValue({ release }) } + const modelReady = jest.fn().mockReturnValue(true) + const runner = new ProductArchiveImageIndexRunner(repository, indexer, lock, modelReady) + return { runner, repository, indexer, lock, release, modelReady } +} + +describe('ProductArchiveImageIndexRunner.run', () => { + beforeEach(() => jest.clearAllMocks()) + + it('模型未就绪时直接报错，不抢锁也不碰数据库', async () => { + const ctx = createRunner([], {}) + ctx.modelReady.mockReturnValue(false) + await expect(ctx.runner.run()).rejects.toThrow('图片模型未就绪') + expect(ctx.lock.acquire).not.toHaveBeenCalled() + expect(ctx.repository.invalidateStaleEmbeddings).not.toHaveBeenCalled() + }) + + it('已有索引任务在跑（抢锁失败）时返回 LOCKED', async () => { + const ctx = createRunner([], {}) + ctx.lock.acquire.mockResolvedValue(null) + await expect(ctx.runner.run()).resolves.toMatchObject({ status: 'LOCKED', remaining: 0 }) + expect(ctx.repository.listPendingCandidates).not.toHaveBeenCalled() + }) + + it('正常增量：失效标记 -> 分批处理 -> 游标推进 -> 汇总与剩余数', async () => { + const onProgress = jest.fn() + const ctx = createRunner( + [[candidate('a1'), candidate('a2')], [candidate('a3')]], + { + a1: { status: 'INDEXED' }, + a2: { status: 'FAILED', message: '下载超时' }, + a3: { status: 'UNCHANGED' }, + }, + ) + + const before = Date.now() + const result = await ctx.runner.run({ limit: 10, batchSize: 2, onProgress }) + + expect(result).toEqual({ + processed: 3, indexed: 1, unchanged: 1, failed: 1, skipped: 0, superseded: 0, invalidated: 2, + status: 'COMPLETED', remaining: 4, + }) + expect(ctx.repository.invalidateStaleEmbeddings).toHaveBeenCalledTimes(1) + + const first = ctx.repository.listPendingCandidates.mock.calls[0][0] + expect(first).toMatchObject({ + modelVersion: PRODUCT_ARCHIVE_IMAGE_SEARCH_MODEL_VERSION, + dimensions: PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS, + full: undefined, + afterId: undefined, + take: 2, + }) + expect(first.retryBefore.getTime()).toBeGreaterThanOrEqual(before - PRODUCT_ARCHIVE_IMAGE_INDEX_RETRY_MS) + expect(first.retryBefore.getTime()).toBeLessThanOrEqual(Date.now() - PRODUCT_ARCHIVE_IMAGE_INDEX_RETRY_MS + 60_000) + // 第二批从上一批最后一条之后继续 + expect(ctx.repository.listPendingCandidates.mock.calls[1][0]).toMatchObject({ afterId: 'a2', take: 2 }) + + expect(onProgress).toHaveBeenCalledTimes(3) + expect(onProgress.mock.calls[1][0]).toMatchObject({ processed: 2, failed: 1, barcode: '690123450002', error: '下载超时' }) + + // 剩余数只按增量口径统计 + expect(ctx.repository.countPendingCandidates).toHaveBeenCalledWith({ + modelVersion: PRODUCT_ARCHIVE_IMAGE_SEARCH_MODEL_VERSION, + dimensions: PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS, + retryBefore: expect.any(Date), + full: false, + }) + expect(ctx.release).toHaveBeenCalledTimes(1) + }) + + it('limit 小于 batchSize 时单批 take 收敛到 limit', async () => { + const ctx = createRunner([[candidate('a1')]], {}) + await ctx.runner.run({ limit: 1, batchSize: 10 }) + expect(ctx.repository.listPendingCandidates).toHaveBeenCalledTimes(1) + expect(ctx.repository.listPendingCandidates.mock.calls[0][0].take).toBe(1) + }) + + it('limit=0 表示补齐模式：持续取批直到取空', async () => { + const ctx = createRunner([[candidate('a1'), candidate('a2')], [candidate('a3')]], {}) + const result = await ctx.runner.run({ limit: 0, batchSize: 2 }) + expect(result.status).toBe('COMPLETED') + expect(result.processed).toBe(3) + expect(ctx.repository.listPendingCandidates.mock.calls.map((call) => call[0].take)).toEqual([2, 2, 2]) + }) + + it('retryNow=true 时失败记录立即进入待处理范围', async () => { + const ctx = createRunner([], {}) + const before = Date.now() + await ctx.runner.run({ retryNow: true }) + const { retryBefore } = ctx.repository.listPendingCandidates.mock.calls[0][0] + expect(retryBefore.getTime()).toBeGreaterThanOrEqual(before) + }) + + it('非法入参直接校验失败', async () => { + const ctx = createRunner([], {}) + await expect(ctx.runner.run({ limit: -1 })).rejects.toThrow('limit') + await expect(ctx.runner.run({ batchSize: 0 })).rejects.toThrow('batchSize') + await expect(ctx.runner.run({ batchSize: 51 })).rejects.toThrow('batchSize') + }) + + it('处理过程中抛错也会释放锁', async () => { + const ctx = createRunner([[candidate('a1')]], {}) + ctx.indexer.indexCandidate.mockRejectedValue(new Error('数据库连接中断')) + await expect(ctx.runner.run()).rejects.toThrow('数据库连接中断') + expect(ctx.release).toHaveBeenCalledTimes(1) + }) +}) diff --git a/furniture-3d-system/src/server/services/__tests__/product-archive-image-index.service.test.ts b/furniture-3d-system/src/server/services/__tests__/product-archive-image-index.service.test.ts new file mode 100644 index 00000000..a80404db --- /dev/null +++ b/furniture-3d-system/src/server/services/__tests__/product-archive-image-index.service.test.ts @@ -0,0 +1,137 @@ +import { + PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS, + PRODUCT_ARCHIVE_IMAGE_SEARCH_MODEL_VERSION, +} from '@/lib/product-archive-image-search' +import type { ProductArchiveImageCandidate } from '@/server/repositories/product-archive-image-search.repository' + +jest.mock('@/server/repositories/product-archive-image-index.repository', () => ({ + productArchiveImageIndexRepository: {}, +})) +jest.mock('@/server/repositories/product-archive-image-search.repository', () => ({ + productArchiveImageSearchRepository: { getEmbeddingByArchiveId: jest.fn() }, +})) +jest.mock('@/server/services/product-archive-image-shape', () => ({ + isProductArchiveImageSearchModelReady: jest.fn(), + createProductArchiveImageEmbedding: jest.fn(), + hashProductArchiveImage: jest.fn(), +})) + +import { ProductArchiveImageIndexService } from '@/server/services/product-archive-image-index.service' + +function candidate(overrides: Partial> = {}) { + return { + id: overrides.id ?? 'a1', + image: overrides.image ?? '/api/storage/object?key=archive%2Fa.jpg', + spu: 'ZSJ0081', + sku: 'ZSJ008103N958V01', + barcode: '6901234500018', + productName: '示例家具', + } as ProductArchiveImageCandidate +} + +function readyEmbedding(overrides: Partial<{ + imageHash: string + modelVersion: string + dimensions: number + embedding: Uint8Array +}> = {}) { + return { + archiveId: 'a1', + imageUrl: '/api/storage/object?key=archive%2Fa.jpg', + imageHash: overrides.imageHash ?? 'h1', + modelVersion: overrides.modelVersion ?? PRODUCT_ARCHIVE_IMAGE_SEARCH_MODEL_VERSION, + dimensions: overrides.dimensions ?? PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS, + embedding: overrides.embedding ?? new Uint8Array(PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS * 4), + } +} + +function createService() { + const repository = { saveIfCurrent: jest.fn().mockResolvedValue(true) } + const searchRepository = { getEmbeddingByArchiveId: jest.fn().mockResolvedValue(null) } + const readImage = jest.fn().mockResolvedValue({ buffer: Buffer.from('img'), contentType: 'image/jpeg' }) + const hashImage = jest.fn().mockResolvedValue('h1') + const createEmbedding = jest.fn().mockResolvedValue({ + buffer: Buffer.alloc(PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS * 4), + dimensions: PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS, + }) + const service = new ProductArchiveImageIndexService( + repository, searchRepository, readImage, hashImage, createEmbedding, + ) + return { service, repository, searchRepository, readImage, hashImage, createEmbedding } +} + +describe('ProductArchiveImageIndexService.indexCandidate', () => { + let ctx: ReturnType + beforeEach(() => { + ctx = createService() + }) + + it('空白图片直接跳过，不下载也不写库', async () => { + await expect(ctx.service.indexCandidate(candidate({ image: ' ' }))) + .resolves.toEqual({ status: 'SKIPPED' }) + expect(ctx.readImage).not.toHaveBeenCalled() + expect(ctx.repository.saveIfCurrent).not.toHaveBeenCalled() + }) + + it('新档案：下载、哈希、推理后写入 READY 向量', async () => { + await expect(ctx.service.indexCandidate(candidate())).resolves.toEqual({ status: 'INDEXED' }) + expect(ctx.readImage).toHaveBeenCalledWith(candidate().image) + expect(ctx.createEmbedding).toHaveBeenCalledWith({ buffer: Buffer.from('img'), contentType: 'image/jpeg' }) + expect(ctx.repository.saveIfCurrent).toHaveBeenCalledWith(expect.objectContaining({ + archiveId: 'a1', + imageUrl: candidate().image, + imageHash: 'h1', + status: 'READY', + errorMessage: null, + dimensions: PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS, + })) + }) + + it('内容、模型版本与向量规格都没变时复用旧向量并刷新地址状态', async () => { + const existing = readyEmbedding() + ctx.searchRepository.getEmbeddingByArchiveId.mockResolvedValue(existing) + + await expect(ctx.service.indexCandidate(candidate())).resolves.toEqual({ status: 'UNCHANGED' }) + expect(ctx.createEmbedding).not.toHaveBeenCalled() + expect(ctx.repository.saveIfCurrent).toHaveBeenCalledTimes(1) + expect(ctx.repository.saveIfCurrent).toHaveBeenCalledWith(expect.objectContaining({ + embedding: existing.embedding, + status: 'READY', + imageUrl: candidate().image, + })) + }) + + it('同图但模型版本已升级时重新推理', async () => { + ctx.searchRepository.getEmbeddingByArchiveId.mockResolvedValue(readyEmbedding({ modelVersion: 'old-v0' })) + await expect(ctx.service.indexCandidate(candidate())).resolves.toEqual({ status: 'INDEXED' }) + expect(ctx.createEmbedding).toHaveBeenCalledTimes(1) + }) + + it('同图同版本但旧向量字节数不合规时重新推理', async () => { + ctx.searchRepository.getEmbeddingByArchiveId.mockResolvedValue( + readyEmbedding({ embedding: new Uint8Array(8) }), + ) + await expect(ctx.service.indexCandidate(candidate())).resolves.toEqual({ status: 'INDEXED' }) + expect(ctx.createEmbedding).toHaveBeenCalledTimes(1) + }) + + it('下载或推理失败：写入 FAILED 与原因，向量为空、indexedAt 为空', async () => { + ctx.readImage.mockRejectedValue(new Error('下载超时')) + const outcome = await ctx.service.indexCandidate(candidate()) + expect(outcome).toEqual({ status: 'FAILED', message: '下载超时' }) + expect(ctx.repository.saveIfCurrent).toHaveBeenCalledWith(expect.objectContaining({ + archiveId: 'a1', + imageHash: 'failed', + embedding: expect.any(Buffer), + dimensions: 0, + status: 'FAILED', + errorMessage: '下载超时', + indexedAt: null, + })) + }) + + it('提交时档案地址已被换掉：结果过期返回 SUPERSEDED', async () => { + ctx.repository.saveIfCurrent.mockResolvedValue(false) + await expect(ctx.service.indexCandidate(candidate())).resolves.toEqual({ status: 'SUPERSEDED' }) + }) +}) diff --git a/furniture-3d-system/src/server/services/__tests__/product-archive-image-search.service.test.ts b/furniture-3d-system/src/server/services/__tests__/product-archive-image-search.service.test.ts index 04215de1..7849ee0e 100644 --- a/furniture-3d-system/src/server/services/__tests__/product-archive-image-search.service.test.ts +++ b/furniture-3d-system/src/server/services/__tests__/product-archive-image-search.service.test.ts @@ -54,9 +54,10 @@ function upstreamBody(overrides: { } } +// 向量行的 imageUrl 必须与档案当前 image 一致，本地命中才有效（换图后旧向量退出搜索） const READY_EMBEDDINGS = [ - { archiveId: 'arch-1', imageUrl: 'u1', imageHash: 'h1', embedding: new Uint8Array(16), dimensions: 4, modelVersion: 'dinov2-small-shape-v1' }, - { archiveId: 'arch-2', imageUrl: 'u2', imageHash: 'h2', embedding: new Uint8Array(16), dimensions: 4, modelVersion: 'dinov2-small-shape-v1' }, + { archiveId: 'arch-1', imageUrl: '/api/storage/object?key=archive%2Fa.jpg', imageHash: 'h1', embedding: new Uint8Array(16), dimensions: 4, modelVersion: 'dinov2-small-shape-v1' }, + { archiveId: 'arch-2', imageUrl: '/api/storage/object?key=archive%2Fa.jpg', imageHash: 'h2', embedding: new Uint8Array(16), dimensions: 4, modelVersion: 'dinov2-small-shape-v1' }, ] describe('ProductArchiveImageSearchService（双层混合检索）', () => { @@ -205,6 +206,25 @@ describe('ProductArchiveImageSearchService（双层混合检索）', () => { } }) + it('向量记录的图片地址与档案当前地址不一致时，本地结果里剔除该档案', async () => { + // 向量缓存 60s，拨时钟隔离本用例 + const realNow = Date.now + Date.now = () => realNow() + 120_000 + mockListReadyEmbeddings.mockResolvedValue([ + { archiveId: 'arch-1', imageUrl: '/api/storage/object?key=archive%2Fold.jpg', imageHash: 'h1', embedding: new Uint8Array(16), dimensions: 4, modelVersion: 'dinov2-small-shape-v1' }, + ]) + mockSearchUpstream.mockResolvedValue(upstreamBody({ ownProductMatches: [] })) + + try { + const service = new ProductArchiveImageSearchService() + const result = await service.search({ buffer: Buffer.from('img'), contentType: 'image/png' }) + + expect(result.archiveResults).toHaveLength(0) + } finally { + Date.now = realNow + } + }) + it('空图片直接校验失败，不调任何一路', async () => { const service = new ProductArchiveImageSearchService() await expect(service.search({ buffer: Buffer.alloc(0), contentType: 'image/png' })) diff --git a/furniture-3d-system/src/server/services/product-archive-image-index-lock.service.ts b/furniture-3d-system/src/server/services/product-archive-image-index-lock.service.ts new file mode 100644 index 00000000..27a209a9 --- /dev/null +++ b/furniture-3d-system/src/server/services/product-archive-image-index-lock.service.ts @@ -0,0 +1,40 @@ +import { Client } from 'pg' + +const LOCK_NAMESPACE = 20261010 +const LOCK_KEY = 1 + +export type ProductArchiveImageIndexLease = { release(): Promise } +export interface ProductArchiveImageIndexLock { + acquire(): Promise +} + +export class PostgresProductArchiveImageIndexLock implements ProductArchiveImageIndexLock { + async acquire() { + const client = new Client({ connectionString: process.env.DATABASE_URL, connectionTimeoutMillis: 10_000 }) + try { + await client.connect() + const result = await client.query<{ acquired: boolean }>( + 'SELECT pg_try_advisory_lock($1::integer, $2::integer) AS acquired', [LOCK_NAMESPACE, LOCK_KEY], + ) + if (!result.rows[0]?.acquired) { + await client.end() + return null + } + let released = false + return { release: async () => { + if (released) return + released = true + try { + await client.query('SELECT pg_advisory_unlock($1::integer, $2::integer)', [LOCK_NAMESPACE, LOCK_KEY]) + } finally { + await client.end() + } + } } + } catch (error) { + await client.end().catch(() => undefined) + throw error + } + } +} + +export const productArchiveImageIndexLock = new PostgresProductArchiveImageIndexLock() diff --git a/furniture-3d-system/src/server/services/product-archive-image-index-runner.service.ts b/furniture-3d-system/src/server/services/product-archive-image-index-runner.service.ts new file mode 100644 index 00000000..e90223f7 --- /dev/null +++ b/furniture-3d-system/src/server/services/product-archive-image-index-runner.service.ts @@ -0,0 +1,78 @@ +import { + PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS, + PRODUCT_ARCHIVE_IMAGE_SEARCH_MODEL_VERSION, +} from '@/lib/product-archive-image-search' +import { + PRODUCT_ARCHIVE_IMAGE_INDEX_RETRY_MS, + type ProductArchiveImageIndexProgress, + type ProductArchiveImageIndexRunResult, +} from '@/lib/product-archive-image-index' +import { + productArchiveImageIndexRepository, + type ProductArchiveImageIndexRepository, +} from '@/server/repositories/product-archive-image-index.repository' +import { productArchiveImageIndexService, type ProductArchiveImageIndexService } from './product-archive-image-index.service' +import { productArchiveImageIndexLock, type ProductArchiveImageIndexLock } from './product-archive-image-index-lock.service' +import { isProductArchiveImageSearchModelReady } from './product-archive-image-shape' + +export class ProductArchiveImageIndexRunner { + constructor( + private readonly repository: ProductArchiveImageIndexRepository = productArchiveImageIndexRepository, + private readonly indexer: Pick = productArchiveImageIndexService, + private readonly lock: ProductArchiveImageIndexLock = productArchiveImageIndexLock, + private readonly modelReady = isProductArchiveImageSearchModelReady, + ) {} + + async run(input: { + limit?: number + batchSize?: number + full?: boolean + retryNow?: boolean + onProgress?: (progress: ProductArchiveImageIndexProgress & { barcode?: string; error?: string }) => void + } = {}): Promise { + const limit = input.limit ?? 200 + const batchSize = input.batchSize ?? 10 + if (!Number.isSafeInteger(limit) || limit < 0) throw new Error('limit must be a nonnegative integer') + if (!Number.isSafeInteger(batchSize) || batchSize < 1 || batchSize > 50) throw new Error('batchSize must be between 1 and 50') + if (!this.modelReady()) throw new Error('图片模型未就绪，请检查 IMAGE_SEARCH_MODEL_PATH 或运行模型准备脚本') + const progress: ProductArchiveImageIndexProgress = { + processed: 0, indexed: 0, unchanged: 0, failed: 0, skipped: 0, superseded: 0, invalidated: 0, + } + const lease = await this.lock.acquire() + if (!lease) return { ...progress, status: 'LOCKED', remaining: 0 } + const selection = { + modelVersion: PRODUCT_ARCHIVE_IMAGE_SEARCH_MODEL_VERSION, + dimensions: PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS, + retryBefore: new Date(Date.now() - (input.retryNow ? 0 : PRODUCT_ARCHIVE_IMAGE_INDEX_RETRY_MS)), + full: input.full, + } + try { + progress.invalidated = await this.repository.invalidateStaleEmbeddings() + let afterId: string | undefined + while (limit === 0 || progress.processed < limit) { + const candidates = await this.repository.listPendingCandidates({ + ...selection, afterId, take: limit === 0 ? batchSize : Math.min(batchSize, limit - progress.processed), + }) + if (candidates.length === 0) break + for (const candidate of candidates) { + afterId = candidate.id + const outcome = await this.indexer.indexCandidate(candidate) + progress.processed += 1 + switch (outcome.status) { + case 'INDEXED': progress.indexed += 1; break + case 'UNCHANGED': progress.unchanged += 1; break + case 'FAILED': progress.failed += 1; break + case 'SKIPPED': progress.skipped += 1; break + case 'SUPERSEDED': progress.superseded += 1; break + } + input.onProgress?.({ ...progress, barcode: candidate.barcode, error: outcome.message }) + } + } + return { ...progress, status: 'COMPLETED', remaining: await this.repository.countPendingCandidates({ ...selection, full: false }) } + } finally { + await lease.release() + } + } +} + +export const productArchiveImageIndexRunner = new ProductArchiveImageIndexRunner() diff --git a/furniture-3d-system/src/server/services/product-archive-image-index.service.ts b/furniture-3d-system/src/server/services/product-archive-image-index.service.ts new file mode 100644 index 00000000..41ca5847 --- /dev/null +++ b/furniture-3d-system/src/server/services/product-archive-image-index.service.ts @@ -0,0 +1,75 @@ +import { + PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS, + PRODUCT_ARCHIVE_IMAGE_SEARCH_MODEL_VERSION, +} from '@/lib/product-archive-image-search' +import type { ProductArchiveImageIndexOutcome } from '@/lib/product-archive-image-index' +import { + productArchiveImageSearchRepository, + type ProductArchiveImageCandidate, + type ProductArchiveImageSearchRepository, +} from '@/server/repositories/product-archive-image-search.repository' +import { + productArchiveImageIndexRepository, + type ProductArchiveImageIndexRepository, +} from '@/server/repositories/product-archive-image-index.repository' +import { createProductArchiveImageEmbedding, hashProductArchiveImage } from './product-archive-image-shape' + +async function readArchiveImage(imageUrl: string) { + const [{ fetchImageBuffer }, { extractNonIntegratedAssetObjectStorageKey }, { createObjectReadUrl }] = + await Promise.all([ + import('./assisted-assembly-ai-render.temp-storage'), + import('@/app/intake/non-integrated/nonIntegratedAssetServer'), + import('@/lib/storage'), + ]) + const objectKey = extractNonIntegratedAssetObjectStorageKey(imageUrl) + return fetchImageBuffer(objectKey ? createObjectReadUrl(objectKey) : imageUrl) +} + +export class ProductArchiveImageIndexService { + constructor( + private readonly repository: Pick = productArchiveImageIndexRepository, + private readonly searchRepository: Pick = productArchiveImageSearchRepository, + private readonly readImage = readArchiveImage, + private readonly hashImage = hashProductArchiveImage, + private readonly createEmbedding = createProductArchiveImageEmbedding, + ) {} + + async indexCandidate(candidate: ProductArchiveImageCandidate): Promise { + const imageUrl = candidate.image?.trim() ?? '' + if (!imageUrl) return { status: 'SKIPPED' } + const base = { archiveId: candidate.id, imageUrl, modelVersion: PRODUCT_ARCHIVE_IMAGE_SEARCH_MODEL_VERSION } + try { + const fetched = await this.readImage(imageUrl) + const imageHash = await this.hashImage(fetched.buffer) + const existing = await this.searchRepository.getEmbeddingByArchiveId(candidate.id) + if ( + existing?.imageHash === imageHash && + existing.modelVersion === base.modelVersion && + existing.dimensions === PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS && + existing.embedding.byteLength === PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS * 4 + ) { + // The same image may acquire a new URL. Refresh the URL/status as well as reusing its vector. + const saved = await this.repository.saveIfCurrent({ + ...base, imageHash, embedding: existing.embedding, + dimensions: existing.dimensions, status: 'READY', errorMessage: null, indexedAt: new Date(), + }) + return { status: saved ? 'UNCHANGED' : 'SUPERSEDED' } + } + const embedding = await this.createEmbedding(fetched) + const saved = await this.repository.saveIfCurrent({ + ...base, imageHash, embedding: embedding.buffer, dimensions: embedding.dimensions, + status: 'READY', errorMessage: null, indexedAt: new Date(), + }) + return { status: saved ? 'INDEXED' : 'SUPERSEDED' } + } catch (error) { + const message = error instanceof Error ? error.message.slice(0, 500) : '图片索引失败' + const saved = await this.repository.saveIfCurrent({ + ...base, imageHash: 'failed', embedding: Buffer.alloc(0), dimensions: 0, + status: 'FAILED', errorMessage: message, indexedAt: null, + }) + return { status: saved ? 'FAILED' : 'SUPERSEDED', message } + } + } +} + +export const productArchiveImageIndexService = new ProductArchiveImageIndexService() diff --git a/furniture-3d-system/src/server/services/product-archive-image-search.service.ts b/furniture-3d-system/src/server/services/product-archive-image-search.service.ts index 501a5942..a210e771 100644 --- a/furniture-3d-system/src/server/services/product-archive-image-search.service.ts +++ b/furniture-3d-system/src/server/services/product-archive-image-search.service.ts @@ -3,7 +3,6 @@ import { type ImageSearchUpstreamOwnMatch, } from '@/lib/image-search-upstream' import { - PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS, PRODUCT_ARCHIVE_IMAGE_SEARCH_MODEL_VERSION, type ProductArchiveImageSearchResponse, type ProductArchiveImageSearchResult, @@ -208,10 +207,14 @@ export class ProductArchiveImageSearchService { ranked.map((item) => item.archiveId), ) const byId = new Map(candidates.map((candidate) => [candidate.id, candidate])) + const rankedIds = new Set(ranked.map((item) => item.archiveId)) + const indexedUrls = new Map(records.filter((record) => rankedIds.has(record.archiveId)) + .map((record) => [record.archiveId, record.imageUrl])) const results: ProductArchiveLocalSearchResult[] = [] for (const item of ranked) { const candidate = byId.get(item.archiveId) - if (!candidate) continue + // A different process may have changed the image while this process still has its 60s vector cache. + if (!candidate || !candidate.image?.trim() || candidate.image.trim() !== indexedUrls.get(item.archiveId)) continue results.push({ rank: results.length + 1, score: item.cosine, @@ -231,71 +234,6 @@ export class ProductArchiveImageSearchService { } } -/** 索引脚本用：为单条档案图片计算/更新本地向量（图未变化时跳过） */ -export async function indexProductArchiveImageCandidate( - candidate: ProductArchiveImageCandidate, - repository: ProductArchiveImageSearchRepository = productArchiveImageSearchRepository, -) { - const imageUrl = candidate.image?.trim() ?? '' - if (!imageUrl) return { status: 'SKIPPED' as const } - - let fetched: { buffer: Buffer; contentType: string } - try { - const { fetchImageBuffer } = await import('./assisted-assembly-ai-render.temp-storage') - const [{ extractNonIntegratedAssetObjectStorageKey }, { createObjectReadUrl }] = - await Promise.all([ - import('@/app/intake/non-integrated/nonIntegratedAssetServer'), - import('@/lib/storage'), - ]) - const objectKey = extractNonIntegratedAssetObjectStorageKey(imageUrl) - fetched = await fetchImageBuffer( - objectKey ? createObjectReadUrl(objectKey) : imageUrl, - ) - const imageHash = await hashProductArchiveImage(fetched.buffer) - const existing = await repository.getEmbeddingByArchiveId(candidate.id) - if ( - existing?.imageHash === imageHash && - existing.modelVersion === PRODUCT_ARCHIVE_IMAGE_SEARCH_MODEL_VERSION && - existing.dimensions === PRODUCT_ARCHIVE_IMAGE_SEARCH_DIMENSIONS && - existing.embedding.byteLength > 0 - ) { - return { status: 'UNCHANGED' as const } - } - - const embedding = await createProductArchiveImageEmbedding({ - buffer: fetched.buffer, - contentType: fetched.contentType, - }) - await repository.upsertEmbedding({ - archiveId: candidate.id, - imageUrl, - imageHash, - embedding: embedding.buffer, - dimensions: embedding.dimensions, - modelVersion: PRODUCT_ARCHIVE_IMAGE_SEARCH_MODEL_VERSION, - status: 'READY', - errorMessage: null, - indexedAt: new Date(), - }) - embeddingCache = null - return { status: 'INDEXED' as const } - } catch (error) { - const message = error instanceof Error ? error.message.slice(0, 500) : '图片索引失败' - await repository.upsertEmbedding({ - archiveId: candidate.id, - imageUrl, - imageHash: 'failed', - embedding: Buffer.alloc(0), - dimensions: 0, - modelVersion: PRODUCT_ARCHIVE_IMAGE_SEARCH_MODEL_VERSION, - status: 'FAILED', - errorMessage: message, - indexedAt: null, - }) - return { status: 'FAILED' as const, message } - } -} - function toUpstreamResult( match: ImageSearchUpstreamOwnMatch, candidate: ProductArchiveImageCandidate,