diff --git a/furniture-3d-system/src/app/api/supply-chain/product-archive/export/route.ts b/furniture-3d-system/src/app/api/supply-chain/product-archive/export/route.ts index 007415d3..2fe120ba 100644 --- a/furniture-3d-system/src/app/api/supply-chain/product-archive/export/route.ts +++ b/furniture-3d-system/src/app/api/supply-chain/product-archive/export/route.ts @@ -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[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, { + 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, }) }) } diff --git a/furniture-3d-system/src/app/supply-chain/database/product-archive/ProductArchiveExportProgress.tsx b/furniture-3d-system/src/app/supply-chain/database/product-archive/ProductArchiveExportProgress.tsx index 98b0761d..69fee832 100644 --- a/furniture-3d-system/src/app/supply-chain/database/product-archive/ProductArchiveExportProgress.tsx +++ b/furniture-3d-system/src/app/supply-chain/database/product-archive/ProductArchiveExportProgress.tsx @@ -65,9 +65,12 @@ export function ProductArchiveExportProgress({
- {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}
{progress.status === 'complete' && progress.totalRows > 0 && (
文件已开始下载
diff --git a/furniture-3d-system/src/app/supply-chain/database/product-archive/useProductArchiveExport.ts b/furniture-3d-system/src/app/supply-chain/database/product-archive/useProductArchiveExport.ts index 262fe602..884ccbc5 100644 --- a/furniture-3d-system/src/app/supply-chain/database/product-archive/useProductArchiveExport.ts +++ b/furniture-3d-system/src/app/supply-chain/database/product-archive/useProductArchiveExport.ts @@ -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) diff --git a/furniture-3d-system/src/server/repositories/supply-chain-product-archive-export.repository.ts b/furniture-3d-system/src/server/repositories/supply-chain-product-archive-export.repository.ts index 9f080d45..ccce5f4e 100644 --- a/furniture-3d-system/src/server/repositories/supply-chain-product-archive-export.repository.ts +++ b/furniture-3d-system/src/server/repositories/supply-chain-product-archive-export.repository.ts @@ -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 +} export interface SupplyChainProductArchiveExportRepository { - listRows(filters?: ProductArchiveFilters): Promise + openRowSource: (filters?: ProductArchiveFilters) => Promise +} + +async function* iterateBatches( + ids: string[], + dataset: PeriodDataset, + batchSize: number, +): AsyncGenerator { + 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 { 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), + } } } diff --git a/furniture-3d-system/src/server/services/__tests__/supply-chain-product-archive-export.service.test.ts b/furniture-3d-system/src/server/services/__tests__/supply-chain-product-archive-export.service.test.ts index 62349dde..584219ff 100644 --- a/furniture-3d-system/src/server/services/__tests__/supply-chain-product-archive-export.service.test.ts +++ b/furniture-3d-system/src/server/services/__tests__/supply-chain-product-archive-export.service.test.ts @@ -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 { + 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 }) { + 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>[] + 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) }) }) diff --git a/furniture-3d-system/src/server/services/supply-chain-product-archive-export.service.ts b/furniture-3d-system/src/server/services/supply-chain-product-archive-export.service.ts index ce1214a9..44af6790 100644 --- a/furniture-3d-system/src/server/services/supply-chain-product-archive-export.service.ts +++ b/furniture-3d-system/src/server/services/supply-chain-product-archive-export.service.ts @@ -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((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 + /** 释放并发额度，幂等；正常写完或客户端断开都会自动释放，调用方无需手动调。 */ + 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 { + 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 里连 都不会有，冻结首行丢失。 + // 必须走 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 } } }