feat(supply-chain): 商品库存快照落库并改读快照
- 新增 ProductInventorySnapshot / ProductInventorySnapshotRun 模型与正式迁移（条码×实际仓库） - 同步任务复用现有库存/云仓/档案读取，临时表+事务原子替换，advisory lock 防重叠，失败保留上一份快照 - 新增 15 分钟 systemd timer 安装脚本与同步脚本（身份校验、凭据走环境变量） - 列表/导出接口改读快照并展示 syncedAt，保留全零行过滤与导出字段选择
This commit is contained in:
WuYongJie
WuYongJie
parent
5cf84a19c8
commit
ef4e4e36ed
@ -75,6 +75,7 @@
|
||||
"materials:sync-standard": "tsx scripts/sync-standard-finishes.ts",
|
||||
"materials:sync-standard:apply": "tsx scripts/sync-standard-finishes.ts --apply",
|
||||
"sync:jst-sku-stock": "tsx scripts/sync-jst-sku-stock-products.ts",
|
||||
"sync:product-inventory-snapshot": "tsx scripts/sync-product-inventory-snapshot.ts",
|
||||
"image-search:model:prepare": "tsx scripts/prepare-product-archive-image-search-model.ts",
|
||||
"image-search:index": "tsx scripts/index-product-archive-images.ts",
|
||||
"repair:barcode-print:historical": "tsx scripts/repair-barcode-print-source-rows.ts",
|
||||
|
||||
@ -0,0 +1,48 @@
|
||||
-- 商品库存快照：条码 × 实际仓库粒度，由 15 分钟定时任务全量替换写入；失败保留上一份快照。
|
||||
CREATE TABLE "ProductInventorySnapshot" (
|
||||
"id" TEXT NOT NULL,
|
||||
"barcode" TEXT NOT NULL,
|
||||
"wmsCoId" INTEGER NOT NULL,
|
||||
"productName" TEXT NOT NULL DEFAULT '',
|
||||
"category" TEXT NOT NULL DEFAULT '',
|
||||
"sku" TEXT NOT NULL DEFAULT '',
|
||||
"productionStatus" TEXT NOT NULL DEFAULT '',
|
||||
"supplier" TEXT NOT NULL DEFAULT '',
|
||||
"purchaser" TEXT NOT NULL DEFAULT '',
|
||||
"stockQty" INTEGER NOT NULL DEFAULT 0,
|
||||
"orderLockQty" INTEGER NOT NULL DEFAULT 0,
|
||||
"availableQty" INTEGER NOT NULL DEFAULT 0,
|
||||
"cloudAvailableQty" INTEGER NOT NULL DEFAULT 0,
|
||||
"warehouseName" TEXT NOT NULL DEFAULT '',
|
||||
"warehouseAttribute" TEXT NOT NULL DEFAULT '',
|
||||
"warehouseOwnership" TEXT NOT NULL DEFAULT '',
|
||||
"warehouseHead" TEXT NOT NULL DEFAULT '',
|
||||
"syncedAt" TIMESTAMP(3) NOT NULL,
|
||||
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
|
||||
CONSTRAINT "ProductInventorySnapshot_pkey" PRIMARY KEY ("id")
|
||||
);
|
||||
|
||||
CREATE UNIQUE INDEX "ProductInventorySnapshot_barcode_wmsCoId_key" ON "ProductInventorySnapshot"("barcode", "wmsCoId");
|
||||
|
||||
CREATE INDEX "ProductInventorySnapshot_warehouseName_idx" ON "ProductInventorySnapshot"("warehouseName");
|
||||
|
||||
CREATE INDEX "ProductInventorySnapshot_syncedAt_idx" ON "ProductInventorySnapshot"("syncedAt");
|
||||
|
||||
-- 同步运行状态：每次定时任务一行，记录 RUNNING / COMPLETED / FAILED。
|
||||
CREATE TABLE "ProductInventorySnapshotRun" (
|
||||
"id" TEXT NOT NULL,
|
||||
"startedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
"completedAt" TIMESTAMP(3),
|
||||
"status" VARCHAR(24) NOT NULL,
|
||||
"rowsRead" INTEGER NOT NULL DEFAULT 0,
|
||||
"rowsWritten" INTEGER NOT NULL DEFAULT 0,
|
||||
"syncedAt" TIMESTAMP(3),
|
||||
"errorMessage" TEXT,
|
||||
|
||||
CONSTRAINT "ProductInventorySnapshotRun_pkey" PRIMARY KEY ("id")
|
||||
);
|
||||
|
||||
CREATE INDEX "ProductInventorySnapshotRun_startedAt_idx" ON "ProductInventorySnapshotRun"("startedAt");
|
||||
|
||||
CREATE INDEX "ProductInventorySnapshotRun_status_startedAt_idx" ON "ProductInventorySnapshotRun"("status", "startedAt");
|
||||
@ -5205,3 +5205,46 @@ model SupplyChainLimitFlowAuth {
|
||||
@@index([styleCode])
|
||||
@@index([applyType])
|
||||
}
|
||||
|
||||
/// 商品库存快照：条码 × 实际仓库粒度的库存数据，由 15 分钟定时任务全量替换写入。
|
||||
/// 数据来源于聚水潭库存明细/云仓链路与产品档案，失败时保留上一份快照。
|
||||
model ProductInventorySnapshot {
|
||||
id String @id @default(cuid())
|
||||
barcode String
|
||||
wmsCoId Int
|
||||
productName String @default("")
|
||||
category String @default("")
|
||||
sku String @default("")
|
||||
productionStatus String @default("")
|
||||
supplier String @default("")
|
||||
purchaser String @default("")
|
||||
stockQty Int @default(0)
|
||||
orderLockQty Int @default(0)
|
||||
availableQty Int @default(0)
|
||||
cloudAvailableQty Int @default(0)
|
||||
warehouseName String @default("")
|
||||
warehouseAttribute String @default("")
|
||||
warehouseOwnership String @default("")
|
||||
warehouseHead String @default("")
|
||||
syncedAt DateTime
|
||||
createdAt DateTime @default(now())
|
||||
|
||||
@@unique([barcode, wmsCoId])
|
||||
@@index([warehouseName])
|
||||
@@index([syncedAt])
|
||||
}
|
||||
|
||||
/// 商品库存快照同步运行记录：用于排查定时任务状态与失败原因，不记录连接串或 SQL 载荷。
|
||||
model ProductInventorySnapshotRun {
|
||||
id String @id @default(cuid())
|
||||
startedAt DateTime @default(now())
|
||||
completedAt DateTime?
|
||||
status String @db.VarChar(24)
|
||||
rowsRead Int @default(0)
|
||||
rowsWritten Int @default(0)
|
||||
syncedAt DateTime?
|
||||
errorMessage String? @db.Text
|
||||
|
||||
@@index([startedAt])
|
||||
@@index([status, startedAt])
|
||||
}
|
||||
|
||||
@ -0,0 +1,63 @@
|
||||
#!/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="${PRODUCT_INVENTORY_SNAPSHOT_SERVICE_NAME:-furniture-product-inventory-snapshot}"
|
||||
SYNC_ENV="$BASE/shared/product-inventory-snapshot.env"
|
||||
CALENDAR="${PRODUCT_INVENTORY_SNAPSHOT_CALENDAR:-*:0/15}"
|
||||
|
||||
if [ "$(id -u)" -ne 0 ]; then
|
||||
echo "This installer requires root." >&2
|
||||
exit 1
|
||||
fi
|
||||
if [ ! -f "$APP_ROOT/scripts/sync-product-inventory-snapshot.ts" ]; then
|
||||
echo "Missing deployed snapshot script: $APP_ROOT/scripts/sync-product-inventory-snapshot.ts" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
if [ ! -f "$SYNC_ENV" ]; then
|
||||
umask 077
|
||||
cat >"$SYNC_ENV" <<EOF
|
||||
# 商品库存管理 · 15 分钟库存快照
|
||||
# 身份校验值必须填生产实际值，否则同步脚本拒绝写入。
|
||||
# 取值方式：node_modules/.bin/tsx -e "import {loadEnvConfig} from '@next/env';loadEnvConfig(process.cwd());console.log(process.env.DATABASE_URL)"
|
||||
PRODUCT_INVENTORY_SNAPSHOT_EXPECTED_DATABASE_IDENTITY=
|
||||
PRODUCT_INVENTORY_SNAPSHOT_EXPECTED_SOURCE_DATABASE_IDENTITY=
|
||||
EOF
|
||||
echo "Created env template: $SYNC_ENV"
|
||||
fi
|
||||
|
||||
chmod 600 "$SYNC_ENV"
|
||||
|
||||
cat >"/etc/systemd/system/$SERVICE_NAME.service" <<EOF
|
||||
[Unit]
|
||||
Description=Materialize product inventory snapshot
|
||||
After=network-online.target
|
||||
Wants=network-online.target
|
||||
[Service]
|
||||
Type=oneshot
|
||||
WorkingDirectory=$APP_ROOT
|
||||
Environment=NODE_ENV=production
|
||||
EnvironmentFile=$BASE/shared/.env
|
||||
EnvironmentFile=$SYNC_ENV
|
||||
ExecStart=/bin/bash -lc 'exec ./node_modules/.bin/tsx scripts/sync-product-inventory-snapshot.ts --apply --confirm=sync-product-inventory-snapshot --expected-database-identity="\${PRODUCT_INVENTORY_SNAPSHOT_EXPECTED_DATABASE_IDENTITY:?Missing expected local identity}" --expected-source-database-identity="\${PRODUCT_INVENTORY_SNAPSHOT_EXPECTED_SOURCE_DATABASE_IDENTITY:?Missing expected source identity}"'
|
||||
TimeoutStartSec=30min
|
||||
Restart=on-failure
|
||||
RestartSec=15min
|
||||
EOF
|
||||
|
||||
cat >"/etc/systemd/system/$SERVICE_NAME.timer" <<EOF
|
||||
[Unit]
|
||||
Description=Product inventory snapshot every 15 minutes
|
||||
[Timer]
|
||||
OnCalendar=$CALENDAR
|
||||
Persistent=true
|
||||
Unit=$SERVICE_NAME.service
|
||||
[Install]
|
||||
WantedBy=timers.target
|
||||
EOF
|
||||
|
||||
systemctl daemon-reload
|
||||
systemctl enable --now "$SERVICE_NAME.timer"
|
||||
systemctl list-timers --all "$SERVICE_NAME.timer" --no-pager
|
||||
@ -0,0 +1,69 @@
|
||||
import { loadEnvConfig } from '@next/env'
|
||||
|
||||
loadEnvConfig(process.cwd())
|
||||
|
||||
const CONFIRM_TOKEN = 'sync-product-inventory-snapshot'
|
||||
|
||||
function arg(name: string) {
|
||||
const prefix = `--${name}=`
|
||||
return process.argv.slice(2).find((value) => value.startsWith(prefix))?.slice(prefix.length) ?? ''
|
||||
}
|
||||
|
||||
function required(name: string) {
|
||||
const value = process.env[name]?.trim()
|
||||
if (!value) throw new Error(`${name} is required.`)
|
||||
return value
|
||||
}
|
||||
|
||||
function identity(url: string) {
|
||||
const parsed = new URL(url)
|
||||
return [
|
||||
parsed.protocol.replace(':', ''),
|
||||
decodeURIComponent(parsed.username || ''),
|
||||
parsed.hostname,
|
||||
parsed.port || '',
|
||||
parsed.pathname.replace(/^\//, ''),
|
||||
parsed.searchParams.get('schema') ?? '',
|
||||
].join('|')
|
||||
}
|
||||
|
||||
function sourceIdentity() {
|
||||
return [
|
||||
'sqlserver',
|
||||
required('JST_REPORT_DB_USER'),
|
||||
required('JST_REPORT_DB_HOST'),
|
||||
process.env.JST_REPORT_DB_PORT?.trim() || '1433',
|
||||
required('JST_REPORT_DB_NAME'),
|
||||
process.env.JST_REPORT_DB_SCHEMA?.trim() || 'rpt',
|
||||
].join('|')
|
||||
}
|
||||
|
||||
async function main() {
|
||||
if (!process.argv.includes('--apply') || arg('confirm') !== CONFIRM_TOKEN) {
|
||||
throw new Error(`写入商品库存快照需要 --apply 和 --confirm=${CONFIRM_TOKEN}`)
|
||||
}
|
||||
const databaseUrl = required('DATABASE_URL')
|
||||
if (arg('expected-database-identity') !== identity(databaseUrl)) {
|
||||
throw new Error('expected database identity mismatch')
|
||||
}
|
||||
if (arg('expected-source-database-identity') !== sourceIdentity()) {
|
||||
throw new Error('expected source identity mismatch')
|
||||
}
|
||||
|
||||
const { prisma } = await import('../src/lib/prisma')
|
||||
const { productInventorySnapshotService } = await import(
|
||||
'../src/server/services/product-inventory-snapshot.service'
|
||||
)
|
||||
try {
|
||||
console.log(JSON.stringify({ stage: 'start' }))
|
||||
const result = await productInventorySnapshotService.sync()
|
||||
console.log(JSON.stringify({ stage: 'completed', ...result }, null, 2))
|
||||
} finally {
|
||||
await prisma.$disconnect()
|
||||
}
|
||||
}
|
||||
|
||||
void main().catch((error) => {
|
||||
console.error(error instanceof Error ? error.message : String(error))
|
||||
process.exitCode = 1
|
||||
})
|
||||
@ -9,7 +9,7 @@ import {
|
||||
sanitizeProductInventorySort,
|
||||
type ProductInventoryColumnKey,
|
||||
} from '@/lib/product-inventory'
|
||||
import { productInventoryService } from '@/server/services/product-inventory.service'
|
||||
import { productInventorySnapshotReadService } from '@/server/services/product-inventory-snapshot-read.service'
|
||||
import { supplyChainPermissionSettingsService } from '@/server/services/supply-chain-permission-settings.service'
|
||||
|
||||
// 商品库存导出：支持勾选字段，按当前筛选/排序导出全部匹配行。
|
||||
@ -76,7 +76,7 @@ export async function GET(request: NextRequest) {
|
||||
return jsonError(ErrorCode.VALIDATION_INVALID_FORMAT, '排序条件不正确', 400, ctx.requestId)
|
||||
}
|
||||
|
||||
const payload = await productInventoryService.createExportWorkbook({
|
||||
const payload = await productInventorySnapshotReadService.createExportWorkbook({
|
||||
columns,
|
||||
filters,
|
||||
sort,
|
||||
|
||||
@ -7,7 +7,7 @@ import {
|
||||
sanitizeProductInventoryFilters,
|
||||
sanitizeProductInventorySort,
|
||||
} from '@/lib/product-inventory'
|
||||
import { productInventoryService } from '@/server/services/product-inventory.service'
|
||||
import { productInventorySnapshotReadService } from '@/server/services/product-inventory-snapshot-read.service'
|
||||
import { supplyChainPermissionSettingsService } from '@/server/services/supply-chain-permission-settings.service'
|
||||
|
||||
// 商品库存管理列表：字段筛选（文本模糊/点选、数值大小）、排序、分页均在服务端完成。
|
||||
@ -79,7 +79,7 @@ export async function GET(request: NextRequest) {
|
||||
)
|
||||
}
|
||||
|
||||
const page = await productInventoryService.getPage({
|
||||
const page = await productInventorySnapshotReadService.getPage({
|
||||
page: parsed.data.page,
|
||||
pageSize: parsed.data.pageSize,
|
||||
filters,
|
||||
|
||||
@ -0,0 +1,97 @@
|
||||
/** @jest-environment node */
|
||||
jest.mock('pg', () => ({ Pool: jest.fn() }))
|
||||
|
||||
import { Pool } from 'pg'
|
||||
import type { ProductInventoryRow } from '@/lib/product-inventory'
|
||||
import { ProductInventorySnapshotRepositoryClass } from '../product-inventory-snapshot.repository'
|
||||
|
||||
const row: ProductInventoryRow = {
|
||||
category: '置物架',
|
||||
sku: 'SKU-1',
|
||||
barcode: 'BAR-1',
|
||||
productName: '商品一',
|
||||
productionStatus: '销售',
|
||||
supplier: '供应商甲',
|
||||
purchaser: '土元',
|
||||
stockQty: 10,
|
||||
orderLockQty: 2,
|
||||
availableQty: 8,
|
||||
publicAvailableQty: 5,
|
||||
cloudAvailableQty: 3,
|
||||
warehouseName: '仓一',
|
||||
warehouseAttribute: '成品发货仓-真实库存',
|
||||
warehouseOwnership: '简尚',
|
||||
warehouseHead: '土元',
|
||||
wmsCoId: 101,
|
||||
}
|
||||
|
||||
describe('商品库存快照分阶段发布', () => {
|
||||
const query = jest.fn()
|
||||
const release = jest.fn()
|
||||
const end = jest.fn()
|
||||
|
||||
beforeEach(() => {
|
||||
jest.clearAllMocks()
|
||||
query.mockImplementation(async (text: string) => {
|
||||
if (text.includes('pg_try_advisory_lock')) return { rows: [{ locked: true }] }
|
||||
if (text.startsWith('SELECT count(*)')) return { rows: [{ count: 1 }] }
|
||||
return { rows: [], rowCount: 1 }
|
||||
})
|
||||
;(Pool as unknown as jest.Mock).mockImplementation(() => ({
|
||||
connect: async () => ({ query, release }),
|
||||
end,
|
||||
}))
|
||||
})
|
||||
|
||||
it('暂存后在一个事务内整体替换快照，并持独占锁', async () => {
|
||||
await new ProductInventorySnapshotRepositoryClass().run(async (session) => {
|
||||
await session.stage([row], new Date('2026-10-19T01:00:00.000Z'))
|
||||
return session.publish(1, new Date('2026-10-19T01:00:00.000Z'))
|
||||
})
|
||||
|
||||
const calls = query.mock.calls as [string, unknown[]?][]
|
||||
const stage = calls.find(([text]) => text.startsWith('INSERT INTO product_inventory_snapshot_stage'))!
|
||||
const payload = JSON.parse(stage[1]![0] as string) as Array<Record<string, unknown>>
|
||||
expect(payload).toHaveLength(1)
|
||||
expect(payload[0]).toMatchObject({
|
||||
barcode: 'BAR-1',
|
||||
wmsCoId: 101,
|
||||
cloudAvailableQty: 3,
|
||||
syncedAt: '2026-10-19T01:00:00.000Z',
|
||||
})
|
||||
expect(String(payload[0].id)).toHaveLength(36)
|
||||
|
||||
const statements = calls.map(([text]) => text)
|
||||
expect(statements).toContain('CREATE UNIQUE INDEX ON product_inventory_snapshot_stage ("barcode", "wmsCoId")')
|
||||
expect(statements.some((text) => text.startsWith('CREATE TEMP TABLE product_inventory_snapshot_stage'))).toBe(true)
|
||||
const publishIndex = statements.findIndex((text) => text.startsWith('INSERT INTO "ProductInventorySnapshot"'))
|
||||
expect(publishIndex).toBeGreaterThan(-1)
|
||||
expect(statements.indexOf('BEGIN')).toBeLessThan(publishIndex)
|
||||
expect(statements.indexOf('DELETE FROM "ProductInventorySnapshot"')).toBeGreaterThan(statements.indexOf('BEGIN'))
|
||||
expect(statements.indexOf('COMMIT')).toBeGreaterThan(publishIndex)
|
||||
expect(statements).toContain('SELECT pg_advisory_unlock(90712, 1)')
|
||||
expect(release).toHaveBeenCalledTimes(1)
|
||||
expect(end).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('发布失败时回滚并保留上一份快照，运行记录标记失败', async () => {
|
||||
const original = query.getMockImplementation()!
|
||||
query.mockImplementation(async (text: string) => {
|
||||
if (text.startsWith('INSERT INTO "ProductInventorySnapshot"')) {
|
||||
throw new Error('publication failed')
|
||||
}
|
||||
return original(text)
|
||||
})
|
||||
|
||||
await expect(new ProductInventorySnapshotRepositoryClass().run(async (session) => {
|
||||
await session.stage([row], new Date())
|
||||
return session.publish(1, new Date())
|
||||
})).rejects.toThrow('publication failed')
|
||||
|
||||
const statements = (query.mock.calls as [string][]).map(([text]) => text)
|
||||
expect(statements).toContain('ROLLBACK')
|
||||
expect(statements.some((text) => text.includes(
|
||||
`UPDATE "ProductInventorySnapshotRun" SET status='FAILED'`,
|
||||
))).toBe(true)
|
||||
})
|
||||
})
|
||||
@ -0,0 +1,275 @@
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import { Pool, type PoolClient } from 'pg'
|
||||
import type { ProductInventoryRow } from '@/lib/product-inventory'
|
||||
|
||||
// 商品库存快照仓储：
|
||||
// - 全量替换写入，临时表暂存后在一个事务里发布；失败回滚并保留上一份快照。
|
||||
// - pg_try_advisory_lock 防止 15 分钟定时任务重叠执行。
|
||||
// - 状态写入 ProductInventorySnapshotRun，便于排查失败；不记录连接串或 SQL 载荷。
|
||||
|
||||
const TABLE = '"ProductInventorySnapshot"'
|
||||
const RUNS = '"ProductInventorySnapshotRun"'
|
||||
const STAGE = 'product_inventory_snapshot_stage'
|
||||
const ADVISORY_LOCK_KEY = 90712
|
||||
|
||||
const SNAPSHOT_COLUMNS = [
|
||||
'id',
|
||||
'barcode',
|
||||
'wmsCoId',
|
||||
'productName',
|
||||
'category',
|
||||
'sku',
|
||||
'productionStatus',
|
||||
'supplier',
|
||||
'purchaser',
|
||||
'stockQty',
|
||||
'orderLockQty',
|
||||
'availableQty',
|
||||
'cloudAvailableQty',
|
||||
'warehouseName',
|
||||
'warehouseAttribute',
|
||||
'warehouseOwnership',
|
||||
'warehouseHead',
|
||||
'syncedAt',
|
||||
] as const
|
||||
|
||||
const quotedColumns = SNAPSHOT_COLUMNS.map((name) => `"${name}"`).join(', ')
|
||||
const quotedColumnList = quotedColumns.split(', ')
|
||||
const recordTypes = SNAPSHOT_COLUMNS.map((name) => `"${name}" ${name === 'wmsCoId'
|
||||
|| name === 'stockQty'
|
||||
|| name === 'orderLockQty'
|
||||
|| name === 'availableQty'
|
||||
|| name === 'cloudAvailableQty'
|
||||
? 'integer'
|
||||
: name === 'syncedAt'
|
||||
? 'timestamp'
|
||||
: 'text'}`).join(', ')
|
||||
|
||||
export type ProductInventorySnapshotSyncResult = {
|
||||
rowsWritten: number
|
||||
syncedAt: string
|
||||
}
|
||||
|
||||
export interface ProductInventorySnapshotSyncSession {
|
||||
begin(): Promise<void>
|
||||
stage(rows: readonly ProductInventoryRow[], syncedAt: Date): Promise<void>
|
||||
publish(expected: number, syncedAt: Date): Promise<number>
|
||||
}
|
||||
|
||||
class Session implements ProductInventorySnapshotSyncSession {
|
||||
private runId: string | null = null
|
||||
|
||||
constructor(private readonly client: PoolClient) {}
|
||||
|
||||
private async ensureRun() {
|
||||
if (this.runId) return
|
||||
this.runId = randomUUID()
|
||||
await this.client.query(
|
||||
`INSERT INTO ${RUNS} (id, status) VALUES ($1, 'RUNNING')`,
|
||||
[this.runId],
|
||||
)
|
||||
await this.client.query(
|
||||
`CREATE TEMP TABLE ${STAGE} (LIKE ${TABLE} INCLUDING DEFAULTS INCLUDING CONSTRAINTS) ON COMMIT PRESERVE ROWS`,
|
||||
)
|
||||
await this.client.query('CREATE UNIQUE INDEX ON product_inventory_snapshot_stage ("barcode", "wmsCoId")')
|
||||
}
|
||||
|
||||
/** 同步开始时即写入 RUNNING 记录；后续数据源读取失败也能留下状态。 */
|
||||
async begin() {
|
||||
await this.ensureRun()
|
||||
}
|
||||
|
||||
async stage(rows: readonly ProductInventoryRow[], syncedAt: Date) {
|
||||
await this.ensureRun()
|
||||
const syncedAtIso = syncedAt.toISOString()
|
||||
const payload = rows.map((row) => ({
|
||||
id: randomUUID(),
|
||||
barcode: row.barcode,
|
||||
wmsCoId: row.wmsCoId,
|
||||
productName: row.productName,
|
||||
category: row.category,
|
||||
sku: row.sku,
|
||||
productionStatus: row.productionStatus,
|
||||
supplier: row.supplier,
|
||||
purchaser: row.purchaser,
|
||||
stockQty: row.stockQty,
|
||||
orderLockQty: row.orderLockQty,
|
||||
availableQty: row.availableQty,
|
||||
cloudAvailableQty: row.cloudAvailableQty ?? 0,
|
||||
warehouseName: row.warehouseName,
|
||||
warehouseAttribute: row.warehouseAttribute,
|
||||
warehouseOwnership: row.warehouseOwnership,
|
||||
warehouseHead: row.warehouseHead,
|
||||
syncedAt: syncedAtIso,
|
||||
}))
|
||||
await this.client.query(
|
||||
`INSERT INTO ${STAGE} (${quotedColumns})
|
||||
SELECT ${quotedColumns} FROM jsonb_to_recordset($1::jsonb) AS x(${recordTypes})`,
|
||||
[JSON.stringify(payload)],
|
||||
)
|
||||
await this.client.query(
|
||||
`UPDATE ${RUNS} SET "rowsRead" = "rowsRead" + $2 WHERE id = $1`,
|
||||
[this.runId, rows.length],
|
||||
)
|
||||
}
|
||||
|
||||
async publish(expected: number, syncedAt: Date) {
|
||||
if (!this.runId) throw new Error('Product inventory snapshot staging not initialized')
|
||||
const check = await this.client.query<{ count: number }>(
|
||||
`SELECT count(*)::int AS count FROM ${STAGE}`,
|
||||
)
|
||||
if (expected <= 0 || check.rows[0].count !== expected) {
|
||||
throw new Error('Product inventory snapshot staging count mismatch')
|
||||
}
|
||||
await this.client.query('BEGIN')
|
||||
try {
|
||||
await this.client.query(`DELETE FROM ${TABLE}`)
|
||||
const inserted = await this.client.query(
|
||||
`INSERT INTO ${TABLE} (${quotedColumns}) SELECT ${quotedColumns} FROM ${STAGE}`,
|
||||
)
|
||||
if (inserted.rowCount !== expected) {
|
||||
throw new Error('Product inventory snapshot publish count mismatch')
|
||||
}
|
||||
await this.client.query(
|
||||
`UPDATE ${RUNS} SET status='COMPLETED', "completedAt"=CURRENT_TIMESTAMP,
|
||||
"rowsWritten"=$2, "syncedAt"=$3::timestamp, "errorMessage"=NULL
|
||||
WHERE id=$1`,
|
||||
[this.runId, expected, syncedAt.toISOString()],
|
||||
)
|
||||
await this.client.query('COMMIT')
|
||||
return expected
|
||||
} catch (error) {
|
||||
await this.client.query('ROLLBACK')
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
async fail(error: unknown) {
|
||||
if (!this.runId) return
|
||||
// 不写入原始驱动错误，避免连接串或 SQL 细节进入运行记录。
|
||||
await this.client.query(
|
||||
`UPDATE ${RUNS} SET status='FAILED', "completedAt"=CURRENT_TIMESTAMP,
|
||||
"errorMessage"=$2 WHERE id=$1 AND status='RUNNING'`,
|
||||
[
|
||||
this.runId,
|
||||
error instanceof Error
|
||||
? `${error.name}: synchronization failed; previous published data retained`
|
||||
: 'Synchronization failed',
|
||||
],
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
export interface ProductInventorySnapshotRepository {
|
||||
/** 在独占锁下执行整轮同步（取数由调用方在回调内完成），失败会记录 FAILED 并保留旧快照。 */
|
||||
run<T>(work: (session: ProductInventorySnapshotSyncSession) => Promise<T>): Promise<T>
|
||||
/** 读取当前快照（页面/导出使用），附带快照同步时间。 */
|
||||
listRows(): Promise<{ rows: ProductInventoryRow[]; syncedAt: string | null }>
|
||||
}
|
||||
|
||||
type SnapshotReadPoolGlobal = typeof globalThis & {
|
||||
productInventorySnapshotReadPool?: Pool
|
||||
}
|
||||
|
||||
function getSnapshotReadPool() {
|
||||
const globalForPool = globalThis as SnapshotReadPoolGlobal
|
||||
if (globalForPool.productInventorySnapshotReadPool) {
|
||||
return globalForPool.productInventorySnapshotReadPool
|
||||
}
|
||||
const connectionString = process.env.DATABASE_URL?.trim()
|
||||
if (!connectionString) throw new Error('DATABASE_URL is required')
|
||||
const pool = new Pool({ connectionString, max: 1, idleTimeoutMillis: 30_000 })
|
||||
pool.on('error', (error) => {
|
||||
console.error('[product-inventory-snapshot] idle database connection error', error.message)
|
||||
})
|
||||
globalForPool.productInventorySnapshotReadPool = pool
|
||||
return pool
|
||||
}
|
||||
|
||||
export class ProductInventorySnapshotRepositoryClass implements ProductInventorySnapshotRepository {
|
||||
async listRows(): Promise<{ rows: ProductInventoryRow[]; syncedAt: string | null }> {
|
||||
const pool = getSnapshotReadPool()
|
||||
const rowsResult = await pool.query<{
|
||||
barcode: string
|
||||
wmsCoId: number
|
||||
productName: string
|
||||
category: string
|
||||
sku: string
|
||||
productionStatus: string
|
||||
supplier: string
|
||||
purchaser: string
|
||||
stockQty: number
|
||||
orderLockQty: number
|
||||
availableQty: number
|
||||
cloudAvailableQty: number
|
||||
warehouseName: string
|
||||
warehouseAttribute: string
|
||||
warehouseOwnership: string
|
||||
warehouseHead: string
|
||||
}>(`SELECT "barcode", "wmsCoId", "productName", "category", "sku", "productionStatus",
|
||||
"supplier", "purchaser", "stockQty", "orderLockQty", "availableQty", "cloudAvailableQty",
|
||||
"warehouseName", "warehouseAttribute", "warehouseOwnership", "warehouseHead"
|
||||
FROM ${TABLE}`)
|
||||
const syncedResult = await pool.query<{ synced: Date | null }>(
|
||||
`SELECT MAX("syncedAt") AT TIME ZONE 'UTC' AS synced FROM ${TABLE}`,
|
||||
)
|
||||
const syncedAt = syncedResult.rows[0]?.synced
|
||||
return {
|
||||
rows: rowsResult.rows.map((row) => ({
|
||||
barcode: row.barcode,
|
||||
wmsCoId: row.wmsCoId,
|
||||
productName: row.productName,
|
||||
category: row.category,
|
||||
sku: row.sku,
|
||||
productionStatus: row.productionStatus,
|
||||
supplier: row.supplier,
|
||||
purchaser: row.purchaser,
|
||||
stockQty: row.stockQty,
|
||||
orderLockQty: row.orderLockQty,
|
||||
availableQty: row.availableQty,
|
||||
cloudAvailableQty: row.cloudAvailableQty,
|
||||
publicAvailableQty: row.availableQty - row.cloudAvailableQty,
|
||||
warehouseName: row.warehouseName,
|
||||
warehouseAttribute: row.warehouseAttribute,
|
||||
warehouseOwnership: row.warehouseOwnership,
|
||||
warehouseHead: row.warehouseHead,
|
||||
})),
|
||||
syncedAt: syncedAt ? new Date(syncedAt).toISOString() : null,
|
||||
}
|
||||
}
|
||||
|
||||
async run<T>(work: (session: ProductInventorySnapshotSyncSession) => Promise<T>): Promise<T> {
|
||||
const pool = new Pool({ connectionString: process.env.DATABASE_URL, max: 1 })
|
||||
const client = await pool.connect()
|
||||
const session = new Session(client)
|
||||
let locked = false
|
||||
try {
|
||||
// 与 Prisma 的 timestamp(3) 取值口径保持一致：数据库内统一存 UTC 墙上时间。
|
||||
await client.query("SET TIME ZONE 'UTC'")
|
||||
const result = await client.query<{ locked: boolean }>(
|
||||
`SELECT pg_try_advisory_lock(${ADVISORY_LOCK_KEY}, 1) AS locked`,
|
||||
)
|
||||
locked = result.rows[0].locked
|
||||
if (!locked) {
|
||||
throw new Error('Another product inventory snapshot synchronization is already running')
|
||||
}
|
||||
await client.query(
|
||||
`UPDATE ${RUNS} SET status='FAILED', "completedAt"=CURRENT_TIMESTAMP,
|
||||
"errorMessage"='Previous synchronization interrupted before publication'
|
||||
WHERE status='RUNNING'`,
|
||||
)
|
||||
await session.begin()
|
||||
return await work(session)
|
||||
} catch (error) {
|
||||
await session.fail(error)
|
||||
throw error
|
||||
} finally {
|
||||
if (locked) await client.query(`SELECT pg_advisory_unlock(${ADVISORY_LOCK_KEY}, 1)`)
|
||||
client.release()
|
||||
await pool.end()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export const productInventorySnapshotRepository =
|
||||
new ProductInventorySnapshotRepositoryClass()
|
||||
@ -0,0 +1,43 @@
|
||||
import {
|
||||
PRODUCT_INVENTORY_COLUMNS,
|
||||
type ProductInventoryColumnKey,
|
||||
type ProductInventoryRow,
|
||||
} from '@/lib/product-inventory'
|
||||
|
||||
function exportFileStamp(date: Date) {
|
||||
const pad = (value: number) => String(value).padStart(2, '0')
|
||||
return `${date.getFullYear()}${pad(date.getMonth() + 1)}${pad(date.getDate())}-${pad(date.getHours())}${pad(date.getMinutes())}`
|
||||
}
|
||||
|
||||
/** 生成商品库存导出工作簿：列顺序按用户勾选，数据由调用方提供（实时或快照）。 */
|
||||
export async function createProductInventoryExportFile(input: {
|
||||
columns: ProductInventoryColumnKey[]
|
||||
rows: readonly ProductInventoryRow[]
|
||||
}): Promise<{ content: Buffer; fileName: string }> {
|
||||
const columns = input.columns
|
||||
.map((key) => PRODUCT_INVENTORY_COLUMNS.find((column) => column.key === key))
|
||||
.filter((column): column is NonNullable<typeof column> => Boolean(column))
|
||||
if (!columns.length) throw new Error('请至少选择一个导出字段')
|
||||
|
||||
const ExcelJS = (await import('exceljs')).default
|
||||
const workbook = new ExcelJS.Workbook()
|
||||
const worksheet = workbook.addWorksheet('商品库存')
|
||||
worksheet.columns = columns.map((column) => ({
|
||||
header: column.label,
|
||||
key: column.key,
|
||||
width: Math.min(40, Math.max(10, Math.round(column.width / 8))),
|
||||
}))
|
||||
worksheet.views = [{ state: 'frozen', ySplit: 1 }]
|
||||
worksheet.getRow(1).font = { bold: true }
|
||||
for (const row of input.rows) {
|
||||
worksheet.addRow(Object.fromEntries(
|
||||
columns.map((column) => [column.key, row[column.key] ?? '']),
|
||||
))
|
||||
}
|
||||
|
||||
const buffer = await workbook.xlsx.writeBuffer()
|
||||
return {
|
||||
content: Buffer.from(buffer as ArrayBuffer),
|
||||
fileName: `商品库存_${exportFileStamp(new Date())}.xlsx`,
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,151 @@
|
||||
import type { ProductInventoryRow } from '@/lib/product-inventory'
|
||||
import { productInventorySnapshotRepository } from '@/server/repositories/product-inventory-snapshot.repository'
|
||||
import {
|
||||
clearProductInventorySnapshotReadCache,
|
||||
ProductInventorySnapshotReadService,
|
||||
} from './product-inventory-snapshot-read.service'
|
||||
|
||||
jest.mock('@/server/repositories/product-inventory-snapshot.repository', () => ({
|
||||
productInventorySnapshotRepository: {
|
||||
listRows: jest.fn(),
|
||||
},
|
||||
}))
|
||||
|
||||
const mockedStore = productInventorySnapshotRepository as unknown as {
|
||||
listRows: jest.Mock
|
||||
}
|
||||
|
||||
function snapshotRow(overrides: Partial<ProductInventoryRow>): ProductInventoryRow {
|
||||
return {
|
||||
category: '置物架',
|
||||
sku: 'SKU-1',
|
||||
barcode: 'BAR-1',
|
||||
productName: '商品一',
|
||||
productionStatus: '销售',
|
||||
supplier: '供应商甲',
|
||||
purchaser: '土元',
|
||||
stockQty: 10,
|
||||
orderLockQty: 2,
|
||||
availableQty: 8,
|
||||
publicAvailableQty: 6,
|
||||
cloudAvailableQty: 2,
|
||||
warehouseName: '仓一',
|
||||
warehouseAttribute: '成品发货仓-真实库存',
|
||||
warehouseOwnership: '简尚',
|
||||
warehouseHead: '土元',
|
||||
wmsCoId: 101,
|
||||
...overrides,
|
||||
}
|
||||
}
|
||||
|
||||
describe('商品库存快照读取服务', () => {
|
||||
beforeEach(() => {
|
||||
clearProductInventorySnapshotReadCache()
|
||||
jest.clearAllMocks()
|
||||
mockedStore.listRows.mockResolvedValue({
|
||||
rows: [
|
||||
snapshotRow({}),
|
||||
snapshotRow({
|
||||
barcode: 'BAR-2',
|
||||
sku: 'SKU-2',
|
||||
stockQty: 5,
|
||||
orderLockQty: 0,
|
||||
availableQty: 5,
|
||||
cloudAvailableQty: 0,
|
||||
publicAvailableQty: 5,
|
||||
warehouseName: '仓二',
|
||||
}),
|
||||
],
|
||||
syncedAt: '2026-10-19T03:53:30.000Z',
|
||||
})
|
||||
})
|
||||
|
||||
it('读取快照并返回分页、小计与 syncedAt', async () => {
|
||||
const service = new ProductInventorySnapshotReadService()
|
||||
const page = await service.getPage({ page: 1, pageSize: 50, filters: {}, sort: null })
|
||||
|
||||
expect(page.total).toBe(2)
|
||||
expect(page.rows.map((row) => row.barcode)).toEqual(['BAR-1', 'BAR-2'])
|
||||
expect(page.subtotals).toEqual({
|
||||
stockQty: 15,
|
||||
orderLockQty: 2,
|
||||
availableQty: 13,
|
||||
publicAvailableQty: 11,
|
||||
cloudAvailableQty: 2,
|
||||
})
|
||||
expect(page.sourceStatus.builtAt).toBe('2026-10-19T03:53:30.000Z')
|
||||
expect(page.sourceStatus.sourceUpdatedAt).toBeNull()
|
||||
expect(page.sourceStatus.cloudReady).toBe(true)
|
||||
// 中文候选值按拼音排序：仓二(er) 在 仓一(yi) 之前。
|
||||
expect(page.facets.warehouseName).toEqual(['仓二', '仓一'])
|
||||
})
|
||||
|
||||
it('支持筛选、排序与分页', async () => {
|
||||
const service = new ProductInventorySnapshotReadService()
|
||||
|
||||
const filtered = await service.getPage({
|
||||
page: 1,
|
||||
pageSize: 50,
|
||||
filters: { warehouseName: { values: ['仓二'] } },
|
||||
sort: null,
|
||||
})
|
||||
expect(filtered.total).toBe(1)
|
||||
expect(filtered.rows[0].barcode).toBe('BAR-2')
|
||||
|
||||
const sorted = await service.getPage({
|
||||
page: 1,
|
||||
pageSize: 1,
|
||||
filters: {},
|
||||
sort: { key: 'availableQty', direction: 'desc' },
|
||||
})
|
||||
expect(sorted.total).toBe(2)
|
||||
expect(sorted.rows[0].barcode).toBe('BAR-1')
|
||||
const second = await service.getPage({
|
||||
page: 2,
|
||||
pageSize: 1,
|
||||
filters: {},
|
||||
sort: { key: 'availableQty', direction: 'desc' },
|
||||
})
|
||||
expect(second.rows[0].barcode).toBe('BAR-2')
|
||||
})
|
||||
|
||||
it('快照为空时提示先运行同步任务', async () => {
|
||||
mockedStore.listRows.mockResolvedValue({ rows: [], syncedAt: null })
|
||||
const service = new ProductInventorySnapshotReadService()
|
||||
|
||||
await expect(service.getPage({ page: 1, pageSize: 50, filters: {}, sort: null }))
|
||||
.rejects.toThrow('库存快照尚未生成')
|
||||
})
|
||||
|
||||
it('剔除同行各数值字段都为 0 的明细', async () => {
|
||||
mockedStore.listRows.mockResolvedValue({
|
||||
rows: [
|
||||
snapshotRow({}),
|
||||
snapshotRow({
|
||||
barcode: 'BAR-ZERO',
|
||||
stockQty: 0,
|
||||
orderLockQty: 0,
|
||||
availableQty: 0,
|
||||
cloudAvailableQty: 0,
|
||||
publicAvailableQty: 0,
|
||||
}),
|
||||
],
|
||||
syncedAt: '2026-10-19T03:53:30.000Z',
|
||||
})
|
||||
const service = new ProductInventorySnapshotReadService()
|
||||
const page = await service.getPage({ page: 1, pageSize: 50, filters: {}, sort: null })
|
||||
|
||||
expect(page.total).toBe(1)
|
||||
expect(page.rows[0].barcode).toBe('BAR-1')
|
||||
})
|
||||
|
||||
it('数据集缓存，refresh 时重新读取快照', async () => {
|
||||
const service = new ProductInventorySnapshotReadService()
|
||||
await service.getPage({ page: 1, pageSize: 50, filters: {}, sort: null })
|
||||
await service.getPage({ page: 1, pageSize: 50, filters: {}, sort: null })
|
||||
expect(mockedStore.listRows).toHaveBeenCalledTimes(1)
|
||||
|
||||
await service.getPage({ page: 1, pageSize: 50, filters: {}, sort: null, refresh: true })
|
||||
expect(mockedStore.listRows).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
})
|
||||
@ -0,0 +1,173 @@
|
||||
import { AppError, ErrorCode } from '@/server/core/errors'
|
||||
import {
|
||||
compareProductInventoryRows,
|
||||
computeProductInventoryFacets,
|
||||
computeProductInventorySubtotals,
|
||||
isProductInventoryZeroNumericRow,
|
||||
matchesProductInventoryFilters,
|
||||
PRODUCT_INVENTORY_MAX_PAGE_SIZE,
|
||||
type ProductInventoryColumnKey,
|
||||
type ProductInventoryFilters,
|
||||
type ProductInventoryPage,
|
||||
type ProductInventoryRow,
|
||||
type ProductInventorySort,
|
||||
} from '@/lib/product-inventory'
|
||||
import {
|
||||
productInventorySnapshotRepository,
|
||||
type ProductInventorySnapshotRepository,
|
||||
} from '@/server/repositories/product-inventory-snapshot.repository'
|
||||
import { createProductInventoryExportFile } from '@/server/services/product-inventory-export'
|
||||
|
||||
// 页面读快照：只读取本地 ProductInventorySnapshot，避免每次请求实时拉聚水潭。
|
||||
// 快照由 15 分钟定时任务全量替换；页面展示 syncedAt 作为数据截止时间。
|
||||
|
||||
const DATASET_TTL_MS = 60 * 1000
|
||||
|
||||
type SnapshotDataset = {
|
||||
rows: ProductInventoryRow[]
|
||||
facets: ProductInventoryPage['facets']
|
||||
sourceStatus: ProductInventoryPage['sourceStatus']
|
||||
}
|
||||
|
||||
type SnapshotCacheGlobal = typeof globalThis & {
|
||||
productInventorySnapshotDataset?: { builtAtMs: number; dataset: SnapshotDataset }
|
||||
productInventorySnapshotBuilding?: Promise<SnapshotDataset>
|
||||
}
|
||||
|
||||
export function clearProductInventorySnapshotReadCache() {
|
||||
const globalForCache = globalThis as SnapshotCacheGlobal
|
||||
delete globalForCache.productInventorySnapshotDataset
|
||||
delete globalForCache.productInventorySnapshotBuilding
|
||||
}
|
||||
|
||||
export type ProductInventorySnapshotPageInput = {
|
||||
page: number
|
||||
pageSize: number
|
||||
filters: ProductInventoryFilters
|
||||
sort: ProductInventorySort | null
|
||||
/** 「更新」按钮：强制重新读取快照表（不触发实时拉取）。 */
|
||||
refresh?: boolean
|
||||
}
|
||||
|
||||
export class ProductInventorySnapshotReadService {
|
||||
constructor(
|
||||
private readonly store: ProductInventorySnapshotRepository = productInventorySnapshotRepository,
|
||||
) {}
|
||||
|
||||
private async buildDataset(): Promise<SnapshotDataset> {
|
||||
const { rows, syncedAt } = await this.store.listRows()
|
||||
if (!rows.length || !syncedAt) {
|
||||
throw new AppError(
|
||||
ErrorCode.SYSTEM_INTERNAL,
|
||||
'库存快照尚未生成，请先运行商品库存快照同步任务。',
|
||||
)
|
||||
}
|
||||
// 与页面口径一致：同行各数值字段全为 0 的明细不展示（快照表保留完整数据）。
|
||||
const visibleRows = rows.filter((row) => !isProductInventoryZeroNumericRow(row))
|
||||
if (!visibleRows.length) {
|
||||
throw new AppError(
|
||||
ErrorCode.SYSTEM_INTERNAL,
|
||||
'库存快照没有可展示的有效明细，请检查同步任务。',
|
||||
)
|
||||
}
|
||||
return {
|
||||
rows: visibleRows,
|
||||
facets: computeProductInventoryFacets(visibleRows),
|
||||
sourceStatus: {
|
||||
builtAt: syncedAt,
|
||||
sourceUpdatedAt: null,
|
||||
inventoryReady: true,
|
||||
archiveReady: true,
|
||||
warehouseReady: true,
|
||||
cloudReady: true,
|
||||
cloudMessage: '',
|
||||
inventoryRowCount: visibleRows.length,
|
||||
archiveMatchedCount: visibleRows.filter((row) => row.category || row.sku).length,
|
||||
warnings: [],
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
private rebuild() {
|
||||
const globalForCache = globalThis as SnapshotCacheGlobal
|
||||
if (globalForCache.productInventorySnapshotBuilding) {
|
||||
return globalForCache.productInventorySnapshotBuilding
|
||||
}
|
||||
const building = this.buildDataset()
|
||||
.then((dataset) => {
|
||||
globalForCache.productInventorySnapshotDataset = { builtAtMs: Date.now(), dataset }
|
||||
globalForCache.productInventorySnapshotBuilding = undefined
|
||||
return dataset
|
||||
})
|
||||
.catch((error) => {
|
||||
globalForCache.productInventorySnapshotBuilding = undefined
|
||||
throw error
|
||||
})
|
||||
globalForCache.productInventorySnapshotBuilding = building
|
||||
return building
|
||||
}
|
||||
|
||||
private async ensureDataset(refresh: boolean): Promise<SnapshotDataset> {
|
||||
const globalForCache = globalThis as SnapshotCacheGlobal
|
||||
const cached = globalForCache.productInventorySnapshotDataset
|
||||
if (refresh || !cached) {
|
||||
await this.rebuild()
|
||||
} else if (Date.now() - cached.builtAtMs > DATASET_TTL_MS) {
|
||||
void this.rebuild().catch((error) => {
|
||||
console.error('[product-inventory-snapshot-read] background refresh failed', error)
|
||||
})
|
||||
}
|
||||
const dataset = (globalThis as SnapshotCacheGlobal).productInventorySnapshotDataset?.dataset
|
||||
if (!dataset) {
|
||||
throw new AppError(
|
||||
ErrorCode.SYSTEM_INTERNAL,
|
||||
'库存快照数据尚未就绪，请点击「更新」重试。',
|
||||
)
|
||||
}
|
||||
return dataset
|
||||
}
|
||||
|
||||
async getPage(input: ProductInventorySnapshotPageInput): Promise<ProductInventoryPage> {
|
||||
const dataset = await this.ensureDataset(Boolean(input.refresh))
|
||||
const pageSize = Math.min(PRODUCT_INVENTORY_MAX_PAGE_SIZE, Math.max(1, Math.floor(input.pageSize)))
|
||||
const filtered = dataset.rows.filter((row) => matchesProductInventoryFilters(row, input.filters))
|
||||
const sorted = input.sort
|
||||
? [...filtered].sort((left, right) => compareProductInventoryRows(left, right, input.sort!))
|
||||
: filtered
|
||||
const total = sorted.length
|
||||
const totalPages = Math.max(1, Math.ceil(total / pageSize))
|
||||
const page = Math.min(Math.max(1, Math.floor(input.page)), totalPages)
|
||||
return {
|
||||
rows: sorted.slice((page - 1) * pageSize, page * pageSize),
|
||||
page,
|
||||
pageSize,
|
||||
total,
|
||||
totalPages,
|
||||
subtotals: computeProductInventorySubtotals(filtered),
|
||||
facets: dataset.facets,
|
||||
sourceStatus: dataset.sourceStatus,
|
||||
}
|
||||
}
|
||||
|
||||
async listForExport(input: {
|
||||
filters: ProductInventoryFilters
|
||||
sort: ProductInventorySort | null
|
||||
}): Promise<ProductInventoryRow[]> {
|
||||
const dataset = await this.ensureDataset(false)
|
||||
const filtered = dataset.rows.filter((row) => matchesProductInventoryFilters(row, input.filters))
|
||||
return input.sort
|
||||
? [...filtered].sort((left, right) => compareProductInventoryRows(left, right, input.sort!))
|
||||
: filtered
|
||||
}
|
||||
|
||||
async createExportWorkbook(input: {
|
||||
columns: ProductInventoryColumnKey[]
|
||||
filters: ProductInventoryFilters
|
||||
sort: ProductInventorySort | null
|
||||
}): Promise<{ content: Buffer; fileName: string }> {
|
||||
const rows = await this.listForExport({ filters: input.filters, sort: input.sort })
|
||||
return createProductInventoryExportFile({ columns: input.columns, rows })
|
||||
}
|
||||
}
|
||||
|
||||
export const productInventorySnapshotReadService = new ProductInventorySnapshotReadService()
|
||||
@ -0,0 +1,147 @@
|
||||
import { productInventoryRepository } from '@/server/repositories/product-inventory.repository'
|
||||
import { productInventorySnapshotRepository } from '@/server/repositories/product-inventory-snapshot.repository'
|
||||
import { supplyChainProductArchiveWarehouseConfigService } from '@/server/services/supply-chain-product-archive-warehouse-config.service'
|
||||
import { ProductInventorySnapshotService } from './product-inventory-snapshot.service'
|
||||
|
||||
jest.mock('@/server/repositories/product-inventory.repository', () => ({
|
||||
productInventoryRepository: {
|
||||
listFinishedInventory: jest.fn(),
|
||||
getLatestInventoryModifiedAt: jest.fn(),
|
||||
loadArchiveProfiles: jest.fn(),
|
||||
loadCloudAvailableMap: jest.fn(),
|
||||
listCloudStockDetail: jest.fn(),
|
||||
},
|
||||
}))
|
||||
|
||||
jest.mock('@/server/repositories/product-inventory-snapshot.repository', () => ({
|
||||
productInventorySnapshotRepository: {
|
||||
run: jest.fn(),
|
||||
},
|
||||
}))
|
||||
|
||||
jest.mock('@/server/services/supply-chain-product-archive-warehouse-config.service', () => ({
|
||||
supplyChainProductArchiveWarehouseConfigService: {
|
||||
list: jest.fn(),
|
||||
},
|
||||
}))
|
||||
|
||||
const mockedSource = productInventoryRepository as unknown as {
|
||||
listFinishedInventory: jest.Mock
|
||||
loadArchiveProfiles: jest.Mock
|
||||
loadCloudAvailableMap: jest.Mock
|
||||
}
|
||||
const mockedStore = productInventorySnapshotRepository as unknown as {
|
||||
run: jest.Mock
|
||||
}
|
||||
const mockedWarehouseConfig = supplyChainProductArchiveWarehouseConfigService as unknown as {
|
||||
list: jest.Mock
|
||||
}
|
||||
|
||||
describe('商品库存快照同步服务', () => {
|
||||
let stage: jest.Mock
|
||||
let publish: jest.Mock
|
||||
|
||||
beforeEach(() => {
|
||||
jest.clearAllMocks()
|
||||
stage = jest.fn()
|
||||
publish = jest.fn(async (count: number) => count)
|
||||
mockedStore.run.mockImplementation(async (work) => work({
|
||||
begin: jest.fn(),
|
||||
stage,
|
||||
publish,
|
||||
}))
|
||||
|
||||
mockedSource.listFinishedInventory.mockResolvedValue([
|
||||
{ barcode: 'BAR-1', name: '商品一', qty: 10, orderLockQty: 2, wmsCoId: 101 },
|
||||
])
|
||||
mockedSource.loadArchiveProfiles.mockResolvedValue(new Map([[
|
||||
'BAR-1',
|
||||
{
|
||||
barcode: 'BAR-1',
|
||||
category: '置物架',
|
||||
sku: 'SKU-1',
|
||||
productionStatus: '销售',
|
||||
supplier: '供应商甲',
|
||||
purchaser: '土元',
|
||||
},
|
||||
]]))
|
||||
mockedSource.loadCloudAvailableMap.mockResolvedValue({
|
||||
available: true,
|
||||
message: '',
|
||||
map: new Map([['BAR-1|101', 3]]),
|
||||
})
|
||||
mockedWarehouseConfig.list.mockResolvedValue({
|
||||
items: [{
|
||||
wmsCoId: 101,
|
||||
warehouseName: '仓一',
|
||||
warehouseAttribute: '成品发货仓-真实库存',
|
||||
warehouseOwnership: 'JIANSHANG',
|
||||
warehouseHead: '土元',
|
||||
}],
|
||||
pageInfo: { page: 1, pageSize: 200, total: 1, totalPages: 1 },
|
||||
facets: { heads: [], wmsCoIds: [], names: [], attributes: [], createdDates: [], updatedDates: [], statuses: [] },
|
||||
})
|
||||
})
|
||||
|
||||
it('组装快照行并写入：云仓、仓库配置与档案字段完整', async () => {
|
||||
const service = new ProductInventorySnapshotService()
|
||||
const result = await service.sync()
|
||||
|
||||
expect(result.rowsWritten).toBe(1)
|
||||
expect(typeof result.syncedAt).toBe('string')
|
||||
const [rows, syncedAt] = stage.mock.calls[0]
|
||||
expect(syncedAt).toBeInstanceOf(Date)
|
||||
expect(rows).toHaveLength(1)
|
||||
expect(rows[0]).toMatchObject({
|
||||
barcode: 'BAR-1',
|
||||
wmsCoId: 101,
|
||||
category: '置物架',
|
||||
sku: 'SKU-1',
|
||||
stockQty: 10,
|
||||
orderLockQty: 2,
|
||||
availableQty: 8,
|
||||
cloudAvailableQty: 3,
|
||||
publicAvailableQty: 5,
|
||||
warehouseName: '仓一',
|
||||
warehouseAttribute: '成品发货仓-真实库存',
|
||||
warehouseOwnership: '简尚',
|
||||
warehouseHead: '土元',
|
||||
})
|
||||
expect(publish).toHaveBeenCalledWith(1, expect.any(Date))
|
||||
})
|
||||
|
||||
it('云仓数据源不可用时中止同步，不写入快照', async () => {
|
||||
mockedSource.loadCloudAvailableMap.mockResolvedValue({
|
||||
available: false,
|
||||
message: '运营云仓数据源读取失败，请稍后重试',
|
||||
map: new Map(),
|
||||
})
|
||||
const service = new ProductInventorySnapshotService()
|
||||
|
||||
await expect(service.sync()).rejects.toThrow('运营云仓数据源读取失败')
|
||||
expect(stage).not.toHaveBeenCalled()
|
||||
expect(publish).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('成品库存为空时中止同步，不写入快照', async () => {
|
||||
mockedSource.listFinishedInventory.mockResolvedValue([])
|
||||
const service = new ProductInventorySnapshotService()
|
||||
|
||||
await expect(service.sync()).rejects.toThrow('成品库存明细为空')
|
||||
expect(stage).not.toHaveBeenCalled()
|
||||
expect(publish).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('仓库配置匹配不足时中止同步，避免写入空仓库列', async () => {
|
||||
mockedWarehouseConfig.list.mockResolvedValue({
|
||||
items: [],
|
||||
pageInfo: { page: 1, pageSize: 200, total: 0, totalPages: 1 },
|
||||
facets: { heads: [], wmsCoIds: [], names: [], attributes: [], createdDates: [], updatedDates: [], statuses: [] },
|
||||
})
|
||||
const service = new ProductInventorySnapshotService()
|
||||
|
||||
await expect(service.sync()).rejects.toThrow('仓库配置匹配不足')
|
||||
expect(stage).not.toHaveBeenCalled()
|
||||
expect(publish).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
@ -0,0 +1,90 @@
|
||||
import {
|
||||
buildProductInventoryRow,
|
||||
type ProductInventoryWarehouseProfile,
|
||||
} from '@/lib/product-inventory'
|
||||
import { productArchiveWarehouseOwnershipLabel } from '@/lib/supply-chain-product-archive-warehouse-config'
|
||||
import {
|
||||
productInventoryRepository,
|
||||
type ProductInventoryRepository,
|
||||
} from '@/server/repositories/product-inventory.repository'
|
||||
import {
|
||||
productInventorySnapshotRepository,
|
||||
type ProductInventorySnapshotRepository,
|
||||
type ProductInventorySnapshotSyncResult,
|
||||
} from '@/server/repositories/product-inventory-snapshot.repository'
|
||||
import { supplyChainProductArchiveWarehouseConfigService } from '@/server/services/supply-chain-product-archive-warehouse-config.service'
|
||||
|
||||
// 商品库存快照同步：复用商品库存管理的读取方法，组装后一次性全量替换快照表。
|
||||
// 任一数据源不可用即中止写入，保留上一份快照，避免出现半新半旧或空列数据。
|
||||
|
||||
const SNAPSHOT_STAGE_BATCH_SIZE = 1000
|
||||
|
||||
export class ProductInventorySnapshotService {
|
||||
constructor(
|
||||
private readonly source: ProductInventoryRepository = productInventoryRepository,
|
||||
private readonly store: ProductInventorySnapshotRepository = productInventorySnapshotRepository,
|
||||
private readonly warehouseConfigService: typeof supplyChainProductArchiveWarehouseConfigService =
|
||||
supplyChainProductArchiveWarehouseConfigService,
|
||||
) {}
|
||||
|
||||
private async loadWarehouseIndex() {
|
||||
const byWmsCoId = new Map<number, ProductInventoryWarehouseProfile>()
|
||||
let page = 1
|
||||
for (;;) {
|
||||
const result = await this.warehouseConfigService.list({ page, pageSize: 200 })
|
||||
for (const item of result.items) {
|
||||
if (item.wmsCoId === null) continue
|
||||
byWmsCoId.set(item.wmsCoId, {
|
||||
wmsCoId: item.wmsCoId,
|
||||
name: item.warehouseName,
|
||||
attribute: item.warehouseAttribute ?? '',
|
||||
ownershipLabel: item.warehouseOwnership
|
||||
? productArchiveWarehouseOwnershipLabel(item.warehouseOwnership)
|
||||
: '',
|
||||
head: item.warehouseHead || '',
|
||||
})
|
||||
}
|
||||
const { page: currentPage, totalPages } = result.pageInfo
|
||||
if (currentPage >= totalPages || page > 20) break
|
||||
page += 1
|
||||
}
|
||||
return byWmsCoId
|
||||
}
|
||||
|
||||
async sync(): Promise<ProductInventorySnapshotSyncResult> {
|
||||
// 取数与写入放在同一个受锁会话里：取数失败也会写入 FAILED 运行记录，且不会与下一轮重叠。
|
||||
return this.store.run(async (session) => {
|
||||
const [inventory, archive, cloud, warehouseIndex] = await Promise.all([
|
||||
this.source.listFinishedInventory(),
|
||||
this.source.loadArchiveProfiles(),
|
||||
this.source.loadCloudAvailableMap(),
|
||||
this.loadWarehouseIndex(),
|
||||
])
|
||||
if (!cloud.available) {
|
||||
throw new Error(cloud.message || '运营云仓数据源不可用，已保留上一份库存快照')
|
||||
}
|
||||
if (!inventory.length) {
|
||||
throw new Error('成品库存明细为空，已保留上一份库存快照')
|
||||
}
|
||||
const warehouseIds = new Set(inventory.map((record) => record.wmsCoId))
|
||||
const matchedWarehouses = [...warehouseIds].filter((id) => warehouseIndex.has(id)).length
|
||||
if (!matchedWarehouses) {
|
||||
throw new Error('仓库配置匹配不足，已保留上一份库存快照')
|
||||
}
|
||||
|
||||
const syncedAt = new Date()
|
||||
const rows = inventory.map((record) => buildProductInventoryRow(record, {
|
||||
archive: archive.get(record.barcode),
|
||||
warehouse: warehouseIndex.get(record.wmsCoId),
|
||||
cloudAvailableQty: cloud.map.get(`${record.barcode}|${record.wmsCoId}`) ?? 0,
|
||||
}))
|
||||
for (let index = 0; index < rows.length; index += SNAPSHOT_STAGE_BATCH_SIZE) {
|
||||
await session.stage(rows.slice(index, index + SNAPSHOT_STAGE_BATCH_SIZE), syncedAt)
|
||||
}
|
||||
const rowsWritten = await session.publish(rows.length, syncedAt)
|
||||
return { rowsWritten, syncedAt: syncedAt.toISOString() }
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
export const productInventorySnapshotService = new ProductInventorySnapshotService()
|
||||
@ -6,7 +6,6 @@ import {
|
||||
computeProductInventorySubtotals,
|
||||
isProductInventoryZeroNumericRow,
|
||||
matchesProductInventoryFilters,
|
||||
PRODUCT_INVENTORY_COLUMNS,
|
||||
PRODUCT_INVENTORY_MAX_PAGE_SIZE,
|
||||
type ProductInventoryArchiveProfile,
|
||||
type ProductInventoryColumnKey,
|
||||
@ -18,6 +17,7 @@ import {
|
||||
type ProductInventoryWarehouseProfile,
|
||||
} from '@/lib/product-inventory'
|
||||
import { productArchiveWarehouseOwnershipLabel } from '@/lib/supply-chain-product-archive-warehouse-config'
|
||||
import { createProductInventoryExportFile } from '@/server/services/product-inventory-export'
|
||||
import {
|
||||
productInventoryRepository,
|
||||
type ProductInventoryRepository,
|
||||
@ -55,11 +55,6 @@ export type ProductInventoryPageInput = {
|
||||
refresh?: boolean
|
||||
}
|
||||
|
||||
function exportFileStamp(date: Date) {
|
||||
const pad = (value: number) => String(value).padStart(2, '0')
|
||||
return `${date.getFullYear()}${pad(date.getMonth() + 1)}${pad(date.getDate())}-${pad(date.getHours())}${pad(date.getMinutes())}`
|
||||
}
|
||||
|
||||
export class ProductInventoryService {
|
||||
constructor(
|
||||
private readonly repository: ProductInventoryRepository = productInventoryRepository,
|
||||
@ -249,36 +244,8 @@ export class ProductInventoryService {
|
||||
filters: ProductInventoryFilters
|
||||
sort: ProductInventorySort | null
|
||||
}): Promise<{ content: Buffer; fileName: string }> {
|
||||
const columns = input.columns
|
||||
.map((key) => PRODUCT_INVENTORY_COLUMNS.find((column) => column.key === key))
|
||||
.filter((column): column is NonNullable<typeof column> => Boolean(column))
|
||||
if (!columns.length) {
|
||||
throw new AppError(ErrorCode.VALIDATION_REQUIRED_FIELD, '请至少选择一个导出字段')
|
||||
}
|
||||
|
||||
const rows = await this.listForExport({ filters: input.filters, sort: input.sort })
|
||||
// 动态引入 exceljs：避免页面/接口冷启动时加载导出依赖，也便于测试环境隔离。
|
||||
const ExcelJS = (await import('exceljs')).default
|
||||
const workbook = new ExcelJS.Workbook()
|
||||
const worksheet = workbook.addWorksheet('商品库存')
|
||||
worksheet.columns = columns.map((column) => ({
|
||||
header: column.label,
|
||||
key: column.key,
|
||||
width: Math.min(40, Math.max(10, Math.round(column.width / 8))),
|
||||
}))
|
||||
worksheet.views = [{ state: 'frozen', ySplit: 1 }]
|
||||
worksheet.getRow(1).font = { bold: true }
|
||||
for (const row of rows) {
|
||||
worksheet.addRow(Object.fromEntries(
|
||||
columns.map((column) => [column.key, row[column.key] ?? '']),
|
||||
))
|
||||
}
|
||||
|
||||
const buffer = await workbook.xlsx.writeBuffer()
|
||||
return {
|
||||
content: Buffer.from(buffer as ArrayBuffer),
|
||||
fileName: `商品库存_${exportFileStamp(new Date())}.xlsx`,
|
||||
}
|
||||
return createProductInventoryExportFile({ columns: input.columns, rows })
|
||||
}
|
||||
|
||||
/** 点击「运营云仓可用数」下钻：该条码在该实际仓库下的各云仓库存明细。 */
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user