Merge remote-tracking branch 'origin/codex/xuxiake-mainline' into codex/wuyongjie-bom-estimate-agent-phase0
This commit is contained in:
WuYongJie
commit
f30866530a
@ -1,4 +1,5 @@
|
||||
import { NextRequest, NextResponse } from 'next/server'
|
||||
import { Readable } from 'node:stream'
|
||||
import { ErrorCode } from '@/server/core/errors'
|
||||
import { handleRequest, jsonError } from '@/server/core/response'
|
||||
import { supplyChainPermissionSettingsService } from '@/server/services/supply-chain-permission-settings.service'
|
||||
@ -9,6 +10,51 @@ function buildContentDisposition(fileName: string) {
|
||||
return `attachment; filename*=UTF-8''${encodeURIComponent(fileName)}`
|
||||
}
|
||||
|
||||
/**
|
||||
* 把服务层的生成流包成响应。
|
||||
*
|
||||
* 导出改成流式后有两个必须显式处理的差异：
|
||||
* 1. 体积在写完之前未知，所以不能再发 Content-Length，由 chunked 编码收尾
|
||||
* （前端已支持「无 Content-Length → 不定量进度」这条分支）；
|
||||
* 2. 响应头一旦发出，后面生成失败就只能掐断流，拿不到 JSON 错误，所以要在这里
|
||||
* 记日志，不能让失败静默。
|
||||
*/
|
||||
async function createExportResponse(params: {
|
||||
filters: Parameters<typeof supplyChainProductArchiveExportService.openWorkbook>[0]['filters']
|
||||
requestId: string
|
||||
signal: AbortSignal
|
||||
}) {
|
||||
const result = await supplyChainProductArchiveExportService.openWorkbook({
|
||||
filters: params.filters,
|
||||
})
|
||||
|
||||
result.completion.catch((error) => {
|
||||
console.error(`[${params.requestId}] product archive export stream failed:`, error)
|
||||
})
|
||||
|
||||
// 客户端断开（关页面 / 取消下载）时销毁生成流，停止继续读库和压 zip
|
||||
const abort = () => {
|
||||
result.stream.destroy()
|
||||
}
|
||||
if (params.signal.aborted) {
|
||||
abort()
|
||||
} else {
|
||||
params.signal.addEventListener('abort', abort, { once: true })
|
||||
}
|
||||
|
||||
return new NextResponse(Readable.toWeb(result.stream) as ReadableStream<Uint8Array>, {
|
||||
status: 200,
|
||||
headers: {
|
||||
'Content-Type': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
||||
'Content-Disposition': buildContentDisposition(result.fileName),
|
||||
'Cache-Control': 'no-store',
|
||||
'X-Exported-Row-Count': String(result.total),
|
||||
'X-Exported-Field-Count': String(result.fieldCount),
|
||||
'X-Request-Id': params.requestId,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
export async function GET(request: NextRequest) {
|
||||
return handleRequest(request, async (ctx) => {
|
||||
const access = await supplyChainPermissionSettingsService.getUserAccess(ctx.user, 'productArchive')
|
||||
@ -21,18 +67,10 @@ export async function GET(request: NextRequest) {
|
||||
return jsonError(ErrorCode.VALIDATION_INVALID_FORMAT, '筛选条件格式不正确', 400, ctx.requestId)
|
||||
}
|
||||
|
||||
const payload = await supplyChainProductArchiveExportService.createWorkbook({ filters: filters.data })
|
||||
return new NextResponse(payload.bytes, {
|
||||
status: 200,
|
||||
headers: {
|
||||
'Content-Type': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
||||
'Content-Length': String(payload.bytes.byteLength),
|
||||
'Content-Disposition': buildContentDisposition(payload.fileName),
|
||||
'Cache-Control': 'no-store',
|
||||
'X-Exported-Row-Count': String(payload.total),
|
||||
'X-Exported-Field-Count': String(payload.fieldCount),
|
||||
'X-Request-Id': ctx.requestId,
|
||||
},
|
||||
return createExportResponse({
|
||||
filters: filters.data,
|
||||
requestId: ctx.requestId,
|
||||
signal: request.signal,
|
||||
})
|
||||
})
|
||||
}
|
||||
@ -52,18 +90,10 @@ export async function POST(request: NextRequest) {
|
||||
return jsonError(ErrorCode.VALIDATION_INVALID_FORMAT, '筛选条件格式不正确', 400, ctx.requestId)
|
||||
}
|
||||
|
||||
const payload = await supplyChainProductArchiveExportService.createWorkbook({ filters: filters.data })
|
||||
return new NextResponse(payload.bytes, {
|
||||
status: 200,
|
||||
headers: {
|
||||
'Content-Type': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
||||
'Content-Length': String(payload.bytes.byteLength),
|
||||
'Content-Disposition': buildContentDisposition(payload.fileName),
|
||||
'Cache-Control': 'no-store',
|
||||
'X-Exported-Row-Count': String(payload.total),
|
||||
'X-Exported-Field-Count': String(payload.fieldCount),
|
||||
'X-Request-Id': ctx.requestId,
|
||||
},
|
||||
return createExportResponse({
|
||||
filters: filters.data,
|
||||
requestId: ctx.requestId,
|
||||
signal: request.signal,
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
@ -65,9 +65,12 @@ export function ProductArchiveExportProgress({
|
||||
<span style={{ width: hasPercent ? `${progress.percent}%` : '36%' }} />
|
||||
</div>
|
||||
<div className="product-export-detail">
|
||||
{progress.totalBytes > 0 && progress.status === 'downloading'
|
||||
{progress.status === 'downloading' && progress.totalBytes > 0
|
||||
? `已下载 ${formatBytes(progress.processedBytes)} / ${formatBytes(progress.totalBytes)}`
|
||||
: progress.message}
|
||||
: progress.status === 'downloading' && progress.processedBytes > 0
|
||||
// 流式导出发不了 Content-Length，进度条走不定量，这里退化成「已接收多少」
|
||||
? `已接收 ${formatBytes(progress.processedBytes)}`
|
||||
: progress.message}
|
||||
</div>
|
||||
{progress.status === 'complete' && progress.totalRows > 0 && (
|
||||
<div className="product-export-complete">文件已开始下载</div>
|
||||
|
||||
@ -49,14 +49,23 @@ async function readExportBlob(
|
||||
const chunks: ArrayBuffer[] = []
|
||||
let processedBytes = 0
|
||||
|
||||
while (true) {
|
||||
const { done, value } = await reader.read()
|
||||
if (done) break
|
||||
const chunk = new Uint8Array(value.byteLength)
|
||||
chunk.set(value)
|
||||
chunks.push(chunk.buffer)
|
||||
processedBytes += value.byteLength
|
||||
onProgress(processedBytes, totalBytes)
|
||||
try {
|
||||
while (true) {
|
||||
const { done, value } = await reader.read()
|
||||
if (done) break
|
||||
const chunk = new Uint8Array(value.byteLength)
|
||||
chunk.set(value)
|
||||
chunks.push(chunk.buffer)
|
||||
processedBytes += value.byteLength
|
||||
onProgress(processedBytes, totalBytes)
|
||||
}
|
||||
} catch (error) {
|
||||
// 服务端是边生成边下发的：响应头已经发出去之后，生成失败只能掐断响应体，
|
||||
// 浏览器这里只会报一个网络错误。已经收到过数据就说明是中途断的，换成可读提示。
|
||||
if (processedBytes > 0) {
|
||||
throw new Error('导出文件中途中断，请稍后重试')
|
||||
}
|
||||
throw error
|
||||
}
|
||||
|
||||
return new Blob(chunks, {
|
||||
@ -99,7 +108,7 @@ export function useProductArchiveExport(filters: ProductArchiveFilters) {
|
||||
totalBytes: responseBytes,
|
||||
totalRows,
|
||||
fileName,
|
||||
message: '文件已生成，正在下载...',
|
||||
message: '正在生成并下载文件...',
|
||||
})
|
||||
const blob = await readExportBlob(response, (processedBytes, totalBytes) => {
|
||||
setProgress({
|
||||
@ -111,7 +120,7 @@ export function useProductArchiveExport(filters: ProductArchiveFilters) {
|
||||
totalBytes,
|
||||
totalRows,
|
||||
fileName,
|
||||
message: '文件已生成，正在下载...',
|
||||
message: '正在生成并下载文件...',
|
||||
})
|
||||
})
|
||||
const url = URL.createObjectURL(blob)
|
||||
|
||||
@ -1,22 +1,83 @@
|
||||
import type { Prisma } from '@prisma/client'
|
||||
import { prisma } from '@/lib/prisma'
|
||||
import type { ProductArchiveFilters } from '@/lib/supply-chain-product-archive'
|
||||
import {
|
||||
productArchiveSelect,
|
||||
type ProductArchiveRecord,
|
||||
} from './supply-chain-product-archive.repository'
|
||||
import { listPeriodPerformance, overlayPerformance, preparePeriodArchiveQuery, readSelectedArchives } from './supply-chain-product-archive-sales-period.repository'
|
||||
import {
|
||||
listPeriodPerformance,
|
||||
overlayPerformance,
|
||||
preparePeriodArchiveQuery,
|
||||
type PeriodDataset,
|
||||
} from './supply-chain-product-archive-sales-period.repository'
|
||||
|
||||
/** 单批取多少行。流式导出内存与总量无关，批大小只影响数据库往返次数。 */
|
||||
export const PRODUCT_ARCHIVE_EXPORT_BATCH_SIZE = 2000
|
||||
|
||||
/** 与页面、导出保持一致的稳定排序（barcode 唯一，排序结果确定）。 */
|
||||
const EXPORT_ORDER_BY: Prisma.SupplyChainProductArchiveOrderByWithRelationInput[] = [
|
||||
{ businessCreatedAt: 'desc' },
|
||||
{ barcode: 'asc' },
|
||||
]
|
||||
|
||||
export type ProductArchiveExportRowSource = {
|
||||
/** 本次导出的总行数。在打开数据源时就已冻结，用于响应头的 X-Exported-Row-Count。 */
|
||||
total: number
|
||||
/** 按导出顺序分批产出完整行（已叠加业绩）。 */
|
||||
batches: (batchSize?: number) => AsyncGenerator<ProductArchiveRecord[]>
|
||||
}
|
||||
|
||||
export interface SupplyChainProductArchiveExportRepository {
|
||||
listRows(filters?: ProductArchiveFilters): Promise<ProductArchiveRecord[]>
|
||||
openRowSource: (filters?: ProductArchiveFilters) => Promise<ProductArchiveExportRowSource>
|
||||
}
|
||||
|
||||
async function* iterateBatches(
|
||||
ids: string[],
|
||||
dataset: PeriodDataset,
|
||||
batchSize: number,
|
||||
): AsyncGenerator<ProductArchiveRecord[]> {
|
||||
for (let index = 0; index < ids.length; index += batchSize) {
|
||||
const batchIds = ids.slice(index, index + batchSize)
|
||||
const records = await prisma.supplyChainProductArchive.findMany({
|
||||
where: { id: { in: batchIds } },
|
||||
select: productArchiveSelect,
|
||||
orderBy: EXPORT_ORDER_BY,
|
||||
})
|
||||
// `id in (...)` 不保证返回顺序，按冻结的 id 顺序重排，保证导出顺序与旧实现一致
|
||||
const byId = new Map(records.map((record) => [record.id, record]))
|
||||
const ordered = batchIds
|
||||
.map((id) => byId.get(id))
|
||||
.filter((record): record is ProductArchiveRecord => Boolean(record))
|
||||
yield overlayPerformance(ordered, dataset)
|
||||
}
|
||||
}
|
||||
|
||||
class SupplyChainProductArchiveExportRepositoryClass
|
||||
implements SupplyChainProductArchiveExportRepository
|
||||
{
|
||||
async listRows(filters?: ProductArchiveFilters) {
|
||||
/**
|
||||
* 打开一次导出的行数据源。
|
||||
*
|
||||
* 这里先把「本次要导哪些 id」一次性冻结，再按 id 分批取完整行。
|
||||
* 原因：旧实现是一条 SQL 把全部行取回，天然是数据库的隐式快照；改成流式后如果
|
||||
* 直接用 skip/offset 翻页，导出期间上游往档案表同步新数据会让行集合错位
|
||||
* （重复行或漏行），所以先取有序 id 列表、再按 id 分批，行集合全程不变。
|
||||
*/
|
||||
async openRowSource(filters?: ProductArchiveFilters): Promise<ProductArchiveExportRowSource> {
|
||||
const dataset = await listPeriodPerformance(filters)
|
||||
const query = await preparePeriodArchiveQuery(filters, undefined, dataset)
|
||||
const records = await readSelectedArchives(query, productArchiveSelect)
|
||||
return overlayPerformance(records, dataset)
|
||||
const ids = query.ids ?? (await prisma.supplyChainProductArchive.findMany({
|
||||
where: query.where,
|
||||
select: { id: true },
|
||||
orderBy: EXPORT_ORDER_BY,
|
||||
})).map((row) => row.id)
|
||||
|
||||
return {
|
||||
total: ids.length,
|
||||
batches: (batchSize = PRODUCT_ARCHIVE_EXPORT_BATCH_SIZE) =>
|
||||
iterateBatches(ids, dataset, batchSize),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -1,48 +1,170 @@
|
||||
import ExcelJS from 'exceljs'
|
||||
import type { Readable } from 'node:stream'
|
||||
import { ARCHIVE_COLUMNS } from '@/app/supply-chain/database/product-archive/columns'
|
||||
import { SupplyChainProductArchiveExportService } from '../supply-chain-product-archive-export.service'
|
||||
import { ErrorCode } from '@/server/core/errors'
|
||||
import {
|
||||
PRODUCT_ARCHIVE_EXPORT_CONCURRENCY_LIMIT,
|
||||
PRODUCT_ARCHIVE_EXPORT_SHEET_NAME,
|
||||
SupplyChainProductArchiveExportService,
|
||||
getActiveProductArchiveExportCount,
|
||||
} from '../supply-chain-product-archive-export.service'
|
||||
|
||||
const ARCHIVE_ROW = {
|
||||
id: 'row-1',
|
||||
barcode: 'BAR-001', spu: 'SPU-001', sku: 'SKU-001', image: null,
|
||||
firstSourceCode: 'PRODUCT_DEVELOPMENT', businessCreatedAt: new Date('2026-08-01T00:00:00Z'),
|
||||
categoryL1: '桌类', categoryL2: '电脑桌', productName: '测试桌', designer: '',
|
||||
opsDirector: '', opsDeptHead: '', opsHead: '', productStatus: '销售', material: '', color: '',
|
||||
weight: null, length: null, width: null, height: null, volume: null, packType: '', productTag: '',
|
||||
installDocId: '', hasDrawing: '', hasVideo: '', spec: '', brand: '', orderType: '', inOutWarehouse: '',
|
||||
storageType: 'INBOUND', isSeasonal: '', replaceCount: '', region: '', supplierName: '', supplierType: '',
|
||||
supplierStatus: '', purchaseOwner: '', settlementMethod: '', boardProvider: '', penaltyStatus: '',
|
||||
warehouseType: '', warehouseHead: '', minimumOrderQuantity: '', firstLeadTime: '', stableLeadTime: '',
|
||||
palletSpec: '', backupSupplier: '', dailySales: '', sales3d: '', sales7d: '', sales15d: '', stockJianshang: '', stockClearance: '',
|
||||
purchasePrice: '', costPrice: '', costPriceSecondary: '', fieldSourceMap: {}, createdAt: new Date(), updatedAt: new Date(),
|
||||
imageCache: null, salesPerformance: null,
|
||||
}
|
||||
|
||||
function createRepository(rowCount = 1) {
|
||||
return {
|
||||
openRowSource: jest.fn().mockImplementation(async () => ({
|
||||
total: rowCount,
|
||||
batches: async function* batches() {
|
||||
for (let index = 0; index < rowCount; index += 1) {
|
||||
yield [{ ...ARCHIVE_ROW, id: `row-${index + 1}` }]
|
||||
}
|
||||
},
|
||||
})),
|
||||
}
|
||||
}
|
||||
|
||||
function createService(rowCount = 1) {
|
||||
const repository = createRepository(rowCount)
|
||||
const warehouseConfigRepository = { listActive: jest.fn().mockResolvedValue([]) }
|
||||
const service = new SupplyChainProductArchiveExportService(
|
||||
repository as never,
|
||||
warehouseConfigRepository,
|
||||
() => new Date('2026-08-31T12:00:00Z'),
|
||||
)
|
||||
return { repository, service }
|
||||
}
|
||||
|
||||
/** 把生成流收成完整字节；必须先开始读，否则 writer.commit() 等不到流排空。 */
|
||||
async function collect(stream: Readable): Promise<Buffer> {
|
||||
const chunks: Buffer[] = []
|
||||
for await (const chunk of stream) {
|
||||
chunks.push(Buffer.from(chunk as Buffer))
|
||||
}
|
||||
return Buffer.concat(chunks)
|
||||
}
|
||||
|
||||
async function drain(result: { stream: Readable; completion: Promise<void> }) {
|
||||
const bytesPromise = collect(result.stream)
|
||||
await result.completion
|
||||
return bytesPromise
|
||||
}
|
||||
|
||||
describe('product archive full export', () => {
|
||||
it('exports every archive column and every repository row', async () => {
|
||||
const repository = {
|
||||
listRows: jest.fn().mockResolvedValue([
|
||||
{
|
||||
id: 'row-1',
|
||||
barcode: 'BAR-001', spu: 'SPU-001', sku: 'SKU-001', image: null,
|
||||
firstSourceCode: 'PRODUCT_DEVELOPMENT', businessCreatedAt: new Date('2026-08-01T00:00:00Z'),
|
||||
categoryL1: '桌类', categoryL2: '电脑桌', productName: '测试桌', designer: '',
|
||||
opsDirector: '', opsDeptHead: '', opsHead: '', productStatus: '销售', material: '', color: '',
|
||||
weight: null, length: null, width: null, height: null, volume: null, packType: '', productTag: '',
|
||||
installDocId: '', hasDrawing: '', hasVideo: '', spec: '', brand: '', orderType: '', inOutWarehouse: '',
|
||||
storageType: 'INBOUND', isSeasonal: '', replaceCount: '', region: '', supplierName: '', supplierType: '',
|
||||
supplierStatus: '', purchaseOwner: '', settlementMethod: '', boardProvider: '', penaltyStatus: '',
|
||||
warehouseType: '', warehouseHead: '', minimumOrderQuantity: '', firstLeadTime: '', stableLeadTime: '',
|
||||
palletSpec: '', backupSupplier: '', dailySales: '', sales3d: '', sales7d: '', sales15d: '', stockJianshang: '', stockClearance: '',
|
||||
purchasePrice: '', costPrice: '', costPriceSecondary: '', fieldSourceMap: {}, createdAt: new Date(), updatedAt: new Date(),
|
||||
imageCache: null, salesPerformance: null,
|
||||
},
|
||||
]),
|
||||
}
|
||||
const warehouseConfigRepository = { listActive: jest.fn().mockResolvedValue([]) }
|
||||
const service = new SupplyChainProductArchiveExportService(
|
||||
repository as never,
|
||||
warehouseConfigRepository,
|
||||
() => new Date('2026-08-31T12:00:00Z'),
|
||||
)
|
||||
const { repository, service } = createService(1)
|
||||
|
||||
const payload = await service.createWorkbook({ filters: { productName: ['测试桌'] } })
|
||||
const result = await service.openWorkbook({ filters: { productName: ['测试桌'] } })
|
||||
const bytes = await drain(result)
|
||||
const workbook = new ExcelJS.Workbook()
|
||||
await workbook.xlsx.load(payload.bytes)
|
||||
const worksheet = workbook.getWorksheet('产品档案资料')!
|
||||
await workbook.xlsx.load(bytes)
|
||||
const worksheet = workbook.getWorksheet(PRODUCT_ARCHIVE_EXPORT_SHEET_NAME)!
|
||||
const headers = (worksheet.getRow(1).values as unknown[]).slice(1)
|
||||
|
||||
expect(repository.listRows).toHaveBeenCalledWith({ productName: ['测试桌'] })
|
||||
expect(payload.total).toBe(1)
|
||||
expect(payload.fieldCount).toBe(89)
|
||||
expect(repository.openRowSource).toHaveBeenCalledWith({ productName: ['测试桌'] })
|
||||
expect(result.total).toBe(1)
|
||||
expect(result.fieldCount).toBe(89)
|
||||
expect(headers).toHaveLength(89)
|
||||
expect(headers).toContain('商品名称')
|
||||
expect(headers).toContain('配件率')
|
||||
expect(worksheet.getCell(2, headers.indexOf('商品名称') + 1).value).toBe('测试桌')
|
||||
expect(payload.fileName).toMatch(/^产品全量档案资料-20260831-/)
|
||||
expect(result.fileName).toMatch(/^产品全量档案资料-20260831-/)
|
||||
})
|
||||
|
||||
it('keeps the frozen header row', async () => {
|
||||
// 流式 worksheet 的 views 只有 getter，赋值会被静默忽略；这个断言专门守住这条回归。
|
||||
const { service } = createService(1)
|
||||
|
||||
const result = await service.openWorkbook({ filters: {} })
|
||||
const bytes = await drain(result)
|
||||
const workbook = new ExcelJS.Workbook()
|
||||
await workbook.xlsx.load(bytes)
|
||||
const worksheet = workbook.getWorksheet(PRODUCT_ARCHIVE_EXPORT_SHEET_NAME)!
|
||||
|
||||
expect(worksheet.views).toEqual([
|
||||
expect.objectContaining({ state: 'frozen', ySplit: 1 }),
|
||||
])
|
||||
expect(ARCHIVE_COLUMNS.flatMap((group) => group.columns)).toHaveLength(89)
|
||||
})
|
||||
|
||||
it('streams rows in batches instead of materialising them all', async () => {
|
||||
const batchCalls: number[] = []
|
||||
const repository = {
|
||||
openRowSource: jest.fn().mockResolvedValue({
|
||||
total: 4,
|
||||
batches: (batchSize?: number) => {
|
||||
batchCalls.push(batchSize ?? 0)
|
||||
return (async function* batches() {
|
||||
for (let index = 0; index < 4; index += 1) {
|
||||
yield [{ ...ARCHIVE_ROW, id: `row-${index + 1}` }]
|
||||
}
|
||||
})()
|
||||
},
|
||||
}),
|
||||
}
|
||||
const service = new SupplyChainProductArchiveExportService(
|
||||
repository as never,
|
||||
{ listActive: jest.fn().mockResolvedValue([]) },
|
||||
() => new Date('2026-08-31T12:00:00Z'),
|
||||
)
|
||||
|
||||
const result = await service.openWorkbook({ filters: {} })
|
||||
const bytes = await drain(result)
|
||||
const workbook = new ExcelJS.Workbook()
|
||||
await workbook.xlsx.load(bytes)
|
||||
|
||||
expect(batchCalls).toEqual([2000])
|
||||
expect(result.total).toBe(4)
|
||||
expect(workbook.getWorksheet(PRODUCT_ARCHIVE_EXPORT_SHEET_NAME)!.rowCount).toBe(5)
|
||||
})
|
||||
|
||||
it('caps concurrent exports and releases the slot when the stream finishes', async () => {
|
||||
const { service } = createService(1)
|
||||
|
||||
const active = [] as Awaited<ReturnType<typeof service.openWorkbook>>[]
|
||||
for (let index = 0; index < PRODUCT_ARCHIVE_EXPORT_CONCURRENCY_LIMIT; index += 1) {
|
||||
active.push(await service.openWorkbook({ filters: {} }))
|
||||
}
|
||||
expect(getActiveProductArchiveExportCount()).toBe(PRODUCT_ARCHIVE_EXPORT_CONCURRENCY_LIMIT)
|
||||
|
||||
await expect(service.openWorkbook({ filters: {} })).rejects.toMatchObject({
|
||||
code: ErrorCode.SYSTEM_RATE_LIMIT,
|
||||
statusCode: 429,
|
||||
})
|
||||
|
||||
await Promise.all(active.map(drain))
|
||||
expect(getActiveProductArchiveExportCount()).toBe(0)
|
||||
|
||||
// 额度还回来之后可以继续导出
|
||||
const afterRelease = await service.openWorkbook({ filters: {} })
|
||||
await drain(afterRelease)
|
||||
expect(getActiveProductArchiveExportCount()).toBe(0)
|
||||
})
|
||||
|
||||
it('releases the slot when the client disconnects mid-stream', async () => {
|
||||
// 关页面 / 取消下载会让生成流被销毁。如果额度不还，攒够 3 次中断就会把导出永久锁死。
|
||||
const { service } = createService(50)
|
||||
|
||||
const result = await service.openWorkbook({ filters: {} })
|
||||
expect(getActiveProductArchiveExportCount()).toBe(1)
|
||||
|
||||
result.stream.destroy()
|
||||
await new Promise((resolve) => setImmediate(resolve))
|
||||
|
||||
expect(getActiveProductArchiveExportCount()).toBe(0)
|
||||
})
|
||||
})
|
||||
|
||||
@ -1,10 +1,15 @@
|
||||
import { PassThrough } from 'node:stream'
|
||||
import type { Readable } from 'node:stream'
|
||||
import ExcelJS from 'exceljs'
|
||||
import { ARCHIVE_COLUMNS } from '@/app/supply-chain/database/product-archive/columns'
|
||||
import type { ProductArchiveFilters, SupplyChainProductArchiveRow } from '@/lib/supply-chain-product-archive'
|
||||
import { AppError, ErrorCode } from '@/server/core/errors'
|
||||
import {
|
||||
PRODUCT_ARCHIVE_EXPORT_BATCH_SIZE,
|
||||
supplyChainProductArchiveExportRepository,
|
||||
type SupplyChainProductArchiveExportRepository,
|
||||
} from '@/server/repositories/supply-chain-product-archive-export.repository'
|
||||
import type { ProductArchiveRecord } from '@/server/repositories/supply-chain-product-archive.repository'
|
||||
import {
|
||||
supplyChainProductArchiveWarehouseConfigRepository,
|
||||
type ProductArchiveWarehouseConfigRecord,
|
||||
@ -17,6 +22,66 @@ import {
|
||||
normalizeWarehouseName,
|
||||
} from './supply-chain-product-archive.mapper'
|
||||
|
||||
/** 导出工作表名。改这里会改变产物的 sheet 名，前端无依赖，但用户会看到。 */
|
||||
export const PRODUCT_ARCHIVE_EXPORT_SHEET_NAME = '产品档案资料'
|
||||
|
||||
/**
|
||||
* 同机并发导出上限。
|
||||
*
|
||||
* 生产 Node 堆被限到 768MB，流式导出单实例只占几十 MB，但每次导出还要额外占
|
||||
* 一条数据库连接 + 一份全量 id 快照；不限并发会让多个导出同时抢连接、互相拖慢，
|
||||
* 叠加后仍可能踩到堆上限。超限直接返回 429，让用户稍后重试。
|
||||
*/
|
||||
export const PRODUCT_ARCHIVE_EXPORT_CONCURRENCY_LIMIT = 3
|
||||
|
||||
let activeExportCount = 0
|
||||
|
||||
/** 当前进行中的导出数，供测试与排查观察。 */
|
||||
export function getActiveProductArchiveExportCount() {
|
||||
return activeExportCount
|
||||
}
|
||||
|
||||
function acquireExportSlot() {
|
||||
if (activeExportCount >= PRODUCT_ARCHIVE_EXPORT_CONCURRENCY_LIMIT) {
|
||||
throw new AppError(
|
||||
ErrorCode.SYSTEM_RATE_LIMIT,
|
||||
'当前导出任务较多，请稍后重试',
|
||||
{ activeExports: activeExportCount, limit: PRODUCT_ARCHIVE_EXPORT_CONCURRENCY_LIMIT },
|
||||
)
|
||||
}
|
||||
activeExportCount += 1
|
||||
|
||||
let released = false
|
||||
return () => {
|
||||
if (released) return
|
||||
released = true
|
||||
activeExportCount = Math.max(0, activeExportCount - 1)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 单批写完后如果客户端读得慢，等下游排空再继续读库。
|
||||
*
|
||||
* ExcelJS 流式 worksheet 的 `addRow(...).commit()` 是同步把 XML 写进内部缓冲，
|
||||
* 不会把下游背压传回调用方；如果客户端拉得很慢而这边全速读库写盘，zip 输出会在
|
||||
* 内存里一直堆。这里显式等一次 drain，把生产速度和客户端消费速度解耦开。
|
||||
*/
|
||||
function relieveBackpressure(stream: PassThrough) {
|
||||
if (!stream.writableNeedDrain) return Promise.resolve()
|
||||
return new Promise<void>((resolve) => {
|
||||
const done = () => {
|
||||
stream.off('drain', done)
|
||||
stream.off('close', done)
|
||||
stream.off('error', done)
|
||||
resolve()
|
||||
}
|
||||
stream.once('drain', done)
|
||||
// 流被销毁（客户端断开）时 drain 不会再来，靠 close/error 兜底，避免卡死
|
||||
stream.once('close', done)
|
||||
stream.once('error', done)
|
||||
})
|
||||
}
|
||||
|
||||
function fileTimestamp(now: Date) {
|
||||
const pad = (value: number) => String(value).padStart(2, '0')
|
||||
return [
|
||||
@ -41,6 +106,19 @@ function columnName(index: number) {
|
||||
return name
|
||||
}
|
||||
|
||||
export type ProductArchiveExportStream = {
|
||||
/** 已开始产出的 xlsx 字节流。边生成边吐，服务端不整包驻留。 */
|
||||
stream: Readable
|
||||
fileName: string
|
||||
/** 本次导出的总行数（打开数据源时已冻结）。 */
|
||||
total: number
|
||||
fieldCount: number
|
||||
/** 全部行写完（含 zip 收尾）的信号。调用方只需要 catch 它的错误。 */
|
||||
completion: Promise<void>
|
||||
/** 释放并发额度，幂等；正常写完或客户端断开都会自动释放，调用方无需手动调。 */
|
||||
release: () => void
|
||||
}
|
||||
|
||||
export class SupplyChainProductArchiveExportService {
|
||||
constructor(
|
||||
private readonly repository: SupplyChainProductArchiveExportRepository =
|
||||
@ -52,8 +130,13 @@ export class SupplyChainProductArchiveExportService {
|
||||
private readonly now: () => Date = () => new Date(),
|
||||
) {}
|
||||
|
||||
private async mapRows(filters?: ProductArchiveFilters) {
|
||||
const records = await this.repository.listRows(filters)
|
||||
/**
|
||||
* 建一次导出用的行映射器。
|
||||
*
|
||||
* 仓库配置只在开头取一次（和旧实现一致），不能每行去查；映射器本身无状态，
|
||||
* 可以跨批复用。`now` 也在这里固定，保证同一份导出里所有相对时间字段口径一致。
|
||||
*/
|
||||
private async createRowMapper() {
|
||||
let warehouseConfigs: ProductArchiveWarehouseConfigRecord[] = []
|
||||
try {
|
||||
warehouseConfigs = await this.warehouseConfigRepository.listActive()
|
||||
@ -65,7 +148,7 @@ export class SupplyChainProductArchiveExportService {
|
||||
)
|
||||
|
||||
const now = this.now()
|
||||
return records.map((record) => {
|
||||
return (record: ProductArchiveRecord): SupplyChainProductArchiveRow => {
|
||||
const row = mapProductArchiveRow({ record, now })
|
||||
const config = warehouseConfigByName.get(normalizeWarehouseName(record.inOutWarehouse))
|
||||
const warehouseFields = resolveProductArchiveWarehouseFields({
|
||||
@ -78,40 +161,90 @@ export class SupplyChainProductArchiveExportService {
|
||||
...row,
|
||||
...warehouseFields,
|
||||
} satisfies SupplyChainProductArchiveRow
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
async createWorkbook(params: { filters?: ProductArchiveFilters }) {
|
||||
const rows = await this.mapRows(params.filters)
|
||||
const workbook = new ExcelJS.Workbook()
|
||||
workbook.creator = '产品全量档案'
|
||||
workbook.created = this.now()
|
||||
const worksheet = workbook.addWorksheet('产品档案资料')
|
||||
const columns = ARCHIVE_COLUMNS.flatMap((group) => group.columns)
|
||||
worksheet.columns = columns.map((column) => ({
|
||||
header: column.label,
|
||||
key: column.prop,
|
||||
width: Math.max(12, Math.min(32, Math.round(column.width / 7))),
|
||||
}))
|
||||
worksheet.views = [{ state: 'frozen', ySplit: 1 }]
|
||||
worksheet.autoFilter = { from: 'A1', to: `${columnName(columns.length - 1)}1` }
|
||||
worksheet.getRow(1).font = { bold: true }
|
||||
worksheet.getRow(1).alignment = { vertical: 'middle' }
|
||||
worksheet.getRow(1).height = 24
|
||||
/**
|
||||
* 打开一次流式导出。
|
||||
*
|
||||
* 返回时只做完了「确定要导哪些行」这步（查 id 快照、装载仓库配置），xlsx 字节
|
||||
* 还没开始产；调用方拿到 stream 立刻响应，边生成边下发，堆占用与总行数无关。
|
||||
*
|
||||
* 这条路径取代了原来的 `createWorkbook`（整包 writeBuffer）——旧写法在 89 列下
|
||||
* 约 60MB / 千行，生产 768MB 堆实际只能撑约 5000 行，99434 行量级的全量导出必崩。
|
||||
*/
|
||||
async openWorkbook(params: { filters?: ProductArchiveFilters }): Promise<ProductArchiveExportStream> {
|
||||
const release = acquireExportSlot()
|
||||
try {
|
||||
const rowSource = await this.repository.openRowSource(params.filters)
|
||||
const mapRow = await this.createRowMapper()
|
||||
const columns = ARCHIVE_COLUMNS.flatMap((group) => group.columns)
|
||||
|
||||
for (const row of rows) {
|
||||
const values = Object.fromEntries(
|
||||
columns.map((column) => [column.prop, row[column.prop as keyof SupplyChainProductArchiveRow] ?? '']),
|
||||
)
|
||||
worksheet.addRow(values)
|
||||
}
|
||||
const stream = new PassThrough()
|
||||
const writer = new ExcelJS.stream.xlsx.WorkbookWriter({
|
||||
stream,
|
||||
// 共享字符串表会按行累积所有唯一文本且无法回收，流式导出必须关掉
|
||||
useSharedStrings: false,
|
||||
useStyles: true,
|
||||
})
|
||||
writer.creator = '产品全量档案'
|
||||
writer.created = this.now()
|
||||
|
||||
const buffer = await workbook.xlsx.writeBuffer()
|
||||
return {
|
||||
bytes: buffer,
|
||||
fileName: `产品全量档案资料-${fileTimestamp(this.now())}.xlsx`,
|
||||
total: rows.length,
|
||||
fieldCount: columns.length,
|
||||
// ⚠️ 流式 worksheet 的 views 只有 getter 没有 setter，`worksheet.views = [...]`
|
||||
// 会被静默忽略（不报错），导出的 xlsx 里连 <sheetViews> 都不会有，冻结首行丢失。
|
||||
// 必须走 addWorksheet 的 options。
|
||||
const worksheet = writer.addWorksheet(PRODUCT_ARCHIVE_EXPORT_SHEET_NAME, {
|
||||
views: [{ state: 'frozen', ySplit: 1 }],
|
||||
})
|
||||
// 注意顺序：columns 的 header 是写进第 1 行单元格的，必须先设 columns 再 commit 表头行，
|
||||
// 否则表头行会以空值先落盘。
|
||||
worksheet.columns = columns.map((column) => ({
|
||||
header: column.label,
|
||||
key: column.prop,
|
||||
width: Math.max(12, Math.min(32, Math.round(column.width / 7))),
|
||||
}))
|
||||
worksheet.autoFilter = { from: 'A1', to: `${columnName(columns.length - 1)}1` }
|
||||
const headerRow = worksheet.getRow(1)
|
||||
headerRow.font = { bold: true }
|
||||
headerRow.alignment = { vertical: 'middle' }
|
||||
headerRow.height = 24
|
||||
headerRow.commit()
|
||||
|
||||
const completion = (async () => {
|
||||
for await (const batch of rowSource.batches(PRODUCT_ARCHIVE_EXPORT_BATCH_SIZE)) {
|
||||
// 客户端已断开就没必要继续读库、继续压 zip
|
||||
if (stream.destroyed) return
|
||||
for (const record of batch) {
|
||||
const row = mapRow(record)
|
||||
const values = Object.fromEntries(
|
||||
columns.map((column) => [column.prop, row[column.prop as keyof SupplyChainProductArchiveRow] ?? '']),
|
||||
)
|
||||
worksheet.addRow(values).commit()
|
||||
}
|
||||
await relieveBackpressure(stream)
|
||||
}
|
||||
if (stream.destroyed) return
|
||||
await worksheet.commit()
|
||||
await writer.commit()
|
||||
})()
|
||||
// 这里先挂一次兜底，保证无论路由层是否接 completion，都不会产生 unhandled rejection
|
||||
completion.then(release, release)
|
||||
|
||||
// 正常写完、或客户端中断销毁流，都要把并发额度还回去（release 幂等）
|
||||
stream.once('close', release)
|
||||
|
||||
return {
|
||||
stream,
|
||||
fileName: `产品全量档案资料-${fileTimestamp(this.now())}.xlsx`,
|
||||
total: rowSource.total,
|
||||
fieldCount: columns.length,
|
||||
completion,
|
||||
release,
|
||||
}
|
||||
} catch (error) {
|
||||
// 还没响应出字节，直接把额度还掉并交回统一错误处理（例如 429）
|
||||
release()
|
||||
throw error
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user