feat(product-archive): 全量档案图片自动增量索引、换图失效与失败重试

This commit is contained in:
WuYongJie 2026-10-12 08:49:10 +08:00
parent 50062a4c3d
commit 7b9fd60f6e
15 changed files with 1061 additions and 117 deletions

View File

@ -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 或不可读取地址不能靠索引任务生成图片,需要按条码定位源图片问题。

View File

@ -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) => {

View File

@ -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" <<EOF
[Unit]
Description=Incremental product archive image index with retry
After=network-online.target
Wants=network-online.target
[Service]
Type=oneshot
WorkingDirectory=$APP_ROOT
Environment=NODE_ENV=production
Environment=OMP_NUM_THREADS=1
EnvironmentFile=$BASE/shared/.env
ExecStart=/usr/bin/env IMAGE_SEARCH_MODEL_PATH=$MODEL_DIR $APP_ROOT/node_modules/.bin/tsx scripts/index-product-archive-images.ts --limit=$INDEX_LIMIT --batch-size=10
Nice=10
TimeoutStartSec=30min
StandardOutput=journal
StandardError=journal
EOF
cat >"/etc/systemd/system/$SERVICE_NAME.timer" <<EOF
[Unit]
Description=Check product archive image index every five minutes
[Timer]
OnCalendar=*-*-* *:0/5:00
Persistent=true
RandomizedDelaySec=15
Unit=$SERVICE_NAME.service
[Install]
WantedBy=timers.target
EOF
# Add only our own post-sync hook. Existing source sync definitions and schedules stay intact.
for sync_unit in furniture-supply-chain-product-archive-sync furniture-supply-chain-product-archive-development-sync; do
if systemctl cat "$sync_unit.service" >/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" <<EOF
[Service]
ExecStartPost=-/usr/bin/systemctl --no-block start $SERVICE_NAME.service
EOF
fi
done
systemctl daemon-reload
systemctl enable --now "$SERVICE_NAME.timer"
systemctl start --no-block "$SERVICE_NAME.service"
systemctl list-timers --all "$SERVICE_NAME.timer" --no-pager

View File

@ -0,0 +1,21 @@
export const PRODUCT_ARCHIVE_IMAGE_INDEX_RETRY_MS = 6 * 60 * 60 * 1000
export type ProductArchiveImageIndexOutcome = {
status: 'INDEXED' | 'UNCHANGED' | 'FAILED' | 'SKIPPED' | 'SUPERSEDED'
message?: string
}
export type ProductArchiveImageIndexProgress = {
processed: number
indexed: number
unchanged: number
failed: number
skipped: number
superseded: number
invalidated: number
}
export type ProductArchiveImageIndexRunResult = ProductArchiveImageIndexProgress & {
status: 'COMPLETED' | 'LOCKED'
remaining: number
}

View File

@ -0,0 +1,149 @@
import type { Prisma } from '@prisma/client'
jest.mock('@/lib/prisma', () => ({
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<unknown>) => 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()
})
})

View File

@ -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' } }))
})
})

View File

@ -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<ProductArchiveImageCandidate[]>
countPendingCandidates(input: ProductArchiveImageIndexSelection): Promise<number>
invalidateStaleEmbeddings(): Promise<number>
saveIfCurrent(input: ProductArchiveImageIndexWrite): Promise<boolean>
}
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<ProductArchiveImageCandidate[]>(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<Array<{ count: number }>>(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<Array<{ image: string | null }>>(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<ArrayBuffer> }
await tx.productArchiveImageEmbedding.upsert({
where: { archiveId: input.archiveId },
create: data,
update: data,
})
return true
})
}
}
export const productArchiveImageIndexRepository = new ProductArchiveImageIndexRepositoryClass()

View File

@ -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<number>
}
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<ProductArchiveImageEmbeddingRecord[]>(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<Array<{ count: number }>>(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
}
}

View File

@ -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<string, ProductArchiveImageIndexOutcome>) {
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)
})
})

View File

@ -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<Pick<ProductArchiveImageCandidate, 'id' | 'image'>> = {}) {
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<typeof createService>
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' })
})
})

View File

@ -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' }))

View File

@ -0,0 +1,40 @@
import { Client } from 'pg'
const LOCK_NAMESPACE = 20261010
const LOCK_KEY = 1
export type ProductArchiveImageIndexLease = { release(): Promise<void> }
export interface ProductArchiveImageIndexLock {
acquire(): Promise<ProductArchiveImageIndexLease | null>
}
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()

View File

@ -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, 'indexCandidate'> = 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<ProductArchiveImageIndexRunResult> {
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()

View File

@ -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, 'saveIfCurrent'> = productArchiveImageIndexRepository,
private readonly searchRepository: Pick<ProductArchiveImageSearchRepository, 'getEmbeddingByArchiveId'> = productArchiveImageSearchRepository,
private readonly readImage = readArchiveImage,
private readonly hashImage = hashProductArchiveImage,
private readonly createEmbedding = createProductArchiveImageEmbedding,
) {}
async indexCandidate(candidate: ProductArchiveImageCandidate): Promise<ProductArchiveImageIndexOutcome> {
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()

View File

@ -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,