diff --git a/furniture-3d-system/.env.example b/furniture-3d-system/.env.example index 6048edc4..253279b9 100644 --- a/furniture-3d-system/.env.example +++ b/furniture-3d-system/.env.example @@ -15,7 +15,9 @@ JST_REPORT_DB_PASSWORD= JST_REPORT_DB_ENCRYPT=true JST_REPORT_DB_TRUST_SERVER_CERTIFICATE=true -# OA 新品流程看板快照来源（MySQL） +# OA 新品流程看板数据源。默认 mysql；切换到 JSReportReplica 时设为 mssql。 +OA_BOARD_SOURCE_DRIVER=mysql +# MySQL 来源兼容配置。 OA_BOARD_SOURCE_DB_HOST= OA_BOARD_SOURCE_DB_PORT=3306 OA_BOARD_SOURCE_DB_NAME= @@ -24,6 +26,30 @@ OA_BOARD_SOURCE_DB_PASSWORD= OA_BOARD_SOURCE_DB_CHARSET=utf8mb4 OA_BOARD_SOURCE_DB_SSL=false +# SQL Server 来源：JSReportReplica.rpt 的 ding_Workflow、ding_WorkflowScheme、ding_WorkflowTask 只读视图。 +# 使用 mssql 来源时，快照仍写入下方配置的企划 MySQL 库。 +OA_BOARD_MSSQL_HOST= +OA_BOARD_MSSQL_PORT=1433 +OA_BOARD_MSSQL_DATABASE=JSReportReplica +OA_BOARD_MSSQL_SCHEMA=rpt +OA_BOARD_MSSQL_USERNAME= +OA_BOARD_MSSQL_PASSWORD= +OA_BOARD_MSSQL_ENCRYPT=false +OA_BOARD_MSSQL_TRUST_SERVER_CERTIFICATE=true +OA_BOARD_MSSQL_CONNECTION_TIMEOUT_MS=15000 +OA_BOARD_MSSQL_REQUEST_TIMEOUT_MS=120000 + +# OA 看板快照目标（MySQL）。留空时兼容复用 PLANNING_DATABASE_URL / PLANNING_DB_*。 +# 配置此组变量后，可保证切换 SQL Server 来源时所有快照、超时跟进和日报数据仍保存在企划库。 +OA_BOARD_SNAPSHOT_DATABASE_URL= +OA_BOARD_SNAPSHOT_DB_HOST= +OA_BOARD_SNAPSHOT_DB_PORT=3306 +OA_BOARD_SNAPSHOT_DB_NAME= +OA_BOARD_SNAPSHOT_DB_USER= +OA_BOARD_SNAPSHOT_DB_PASSWORD= +OA_BOARD_SNAPSHOT_DB_CHARSET=utf8mb4 +OA_BOARD_SNAPSHOT_DB_SSL=false + # HR 员工目录（SQL Server）。OA 快照仅在本地员工目录无法匹配时使用， # 用于将 ding_WorkflowTask.userId 解析成当前负责人的姓名。 HR_SQLSERVER_HOST= diff --git a/furniture-3d-system/scripts/sync-oa-board.ts b/furniture-3d-system/scripts/sync-oa-board.ts index c2148d2a..5c738ac8 100644 --- a/furniture-3d-system/scripts/sync-oa-board.ts +++ b/furniture-3d-system/scripts/sync-oa-board.ts @@ -62,9 +62,9 @@ Usage: Default mode is --dry-run. 当前 OA 新品流程看板默认优先读取 planning_database.oa_board_flow_detail_snapshot 快照表； ---apply 会从企划库直读并刷新快照表。 +--apply 会从配置的数据源读取并刷新企划库快照表。 -Preferred source environment variable: +MySQL source environment variable (OA_BOARD_SOURCE_DRIVER=mysql): OA_BOARD_SOURCE_DATABASE_URL=mysql://user:password@host:port/database Alternative source environment variables: @@ -81,6 +81,22 @@ Compatible fallback source environment variables: PLANNING_DB_NAME PLANNING_DB_USER PLANNING_DB_PASSWORD + +SQL Server source environment variables (OA_BOARD_SOURCE_DRIVER=mssql): + OA_BOARD_MSSQL_HOST + OA_BOARD_MSSQL_PORT + OA_BOARD_MSSQL_DATABASE + OA_BOARD_MSSQL_SCHEMA # default rpt + OA_BOARD_MSSQL_USERNAME + OA_BOARD_MSSQL_PASSWORD + +Snapshot target environment variables (always MySQL): + OA_BOARD_SNAPSHOT_DATABASE_URL + OA_BOARD_SNAPSHOT_DB_HOST + OA_BOARD_SNAPSHOT_DB_PORT + OA_BOARD_SNAPSHOT_DB_NAME + OA_BOARD_SNAPSHOT_DB_USER + OA_BOARD_SNAPSHOT_DB_PASSWORD `.trim()) } diff --git a/furniture-3d-system/src/app/intake/non-integrated/core/nonIntegratedStorage.ts b/furniture-3d-system/src/app/intake/non-integrated/core/nonIntegratedStorage.ts index 068739fb..c928dea1 100644 --- a/furniture-3d-system/src/app/intake/non-integrated/core/nonIntegratedStorage.ts +++ b/furniture-3d-system/src/app/intake/non-integrated/core/nonIntegratedStorage.ts @@ -586,6 +586,13 @@ export type NonIntegratedPricingRow = { jstVirtualInventoryLastAttemptAt?: string; jstVirtualInventoryError?: string; jstVirtualInventorySoId?: string; + /** Read-only status from the separate fixed virtual-inventory task queue. */ + fixedVirtualInventoryStatus?: "pending" | "pushed"; + fixedVirtualInventoryScheduledAt?: string; + fixedVirtualInventoryPushedAt?: string; + fixedVirtualInventoryRetryCount?: string; + fixedVirtualInventoryLastAttemptAt?: string; + fixedVirtualInventoryError?: string; }; export type NonIntegratedProductLimitRow = { diff --git a/furniture-3d-system/src/app/intake/non-integrated/workspaces/ProductPricingWorkspace.test.tsx b/furniture-3d-system/src/app/intake/non-integrated/workspaces/ProductPricingWorkspace.test.tsx new file mode 100644 index 00000000..104d6fa5 --- /dev/null +++ b/furniture-3d-system/src/app/intake/non-integrated/workspaces/ProductPricingWorkspace.test.tsx @@ -0,0 +1,57 @@ +import { render, screen } from "@testing-library/react"; +import { ProductPricingWorkspace } from "./ProductPricingWorkspace"; +import type { ProductPricingDisplayRow } from "./product-workflow/productWorkflowRowHelpers"; + +function createPricingRow(): ProductPricingDisplayRow { + return { + skuId: "sku-1", + model: "MODEL-001", + skuCode: "SKU-001", + barcode: "BAR-001", + setCount: "1", + specName: "规格一", + supplierName: "供应商", + costPrice: "88", + productTag: "", + systemEnteredAt: "2026-08-10T08:00:00.000Z", + jstVirtualInventoryStatus: "failed", + jstVirtualInventoryScheduledAt: "", + jstVirtualInventoryPushedAt: "", + jstVirtualInventoryRetryCount: "0", + jstVirtualInventoryLastAttemptAt: "", + jstVirtualInventoryError: "聚水潭未找到对应仓库", + jstVirtualInventorySoId: "", + fixedVirtualInventoryStatus: "pushed", + fixedVirtualInventoryScheduledAt: "2026-08-10T09:00:00.000Z", + fixedVirtualInventoryPushedAt: "2026-08-10T09:00:01.000Z", + fixedVirtualInventoryRetryCount: "0", + fixedVirtualInventoryLastAttemptAt: "2026-08-10T09:00:01.000Z", + fixedVirtualInventoryError: "", + }; +} + +describe("ProductPricingWorkspace", () => { + it("shows the fixed virtual inventory result before a failed legacy result", () => { + const row = createPricingRow(); + + render( + true} + onUpdateRow={jest.fn()} + onConfirmPushRows={jest.fn()} + />, + ); + + expect(screen.getByText("库存已推")).toHaveAttribute( + "title", + "固定虚拟库存已于 2026-08-10T09:00:01.000Z 推送", + ); + expect(screen.queryByText("库存失败")).not.toBeInTheDocument(); + }); +}); diff --git a/furniture-3d-system/src/app/intake/non-integrated/workspaces/ProductPricingWorkspace.tsx b/furniture-3d-system/src/app/intake/non-integrated/workspaces/ProductPricingWorkspace.tsx index 1372d7b8..6d9c15dc 100644 --- a/furniture-3d-system/src/app/intake/non-integrated/workspaces/ProductPricingWorkspace.tsx +++ b/furniture-3d-system/src/app/intake/non-integrated/workspaces/ProductPricingWorkspace.tsx @@ -38,6 +38,29 @@ function ReadOnlyCell({ value }: { value: string }) { } function getVirtualInventoryStatusMeta(row: ProductPricingDisplayRow) { + if (row.fixedVirtualInventoryStatus === "pending") { + const retryCount = Number(row.fixedVirtualInventoryRetryCount || "0"); + return { + label: retryCount > 0 ? "库存重试中" : "库存待推", + tone: "bg-[#e7f1ff] text-[#1764ad]", + title: row.fixedVirtualInventoryError + ? `固定虚拟库存推送失败，将于 ${row.fixedVirtualInventoryScheduledAt || "稍后"} 重试：${row.fixedVirtualInventoryError}` + : row.fixedVirtualInventoryScheduledAt + ? `固定虚拟库存已入队，预计 ${row.fixedVirtualInventoryScheduledAt} 发起` + : "固定虚拟库存已入推送队列", + }; + } + + if (row.fixedVirtualInventoryStatus === "pushed") { + return { + label: "库存已推", + tone: "bg-[#dff7e8] text-[#137a38]", + title: row.fixedVirtualInventoryPushedAt + ? `固定虚拟库存已于 ${row.fixedVirtualInventoryPushedAt} 推送` + : "固定虚拟库存已推送", + }; + } + const virtualInventorySkipTitle = row.jstVirtualInventoryError || "无需推虚拟库存"; const normalizedVirtualInventoryError = diff --git a/furniture-3d-system/src/app/intake/non-integrated/workspaces/product-workflow/productWorkflowRowHelpers.test.ts b/furniture-3d-system/src/app/intake/non-integrated/workspaces/product-workflow/productWorkflowRowHelpers.test.ts index e0c9d412..99452178 100644 --- a/furniture-3d-system/src/app/intake/non-integrated/workspaces/product-workflow/productWorkflowRowHelpers.test.ts +++ b/furniture-3d-system/src/app/intake/non-integrated/workspaces/product-workflow/productWorkflowRowHelpers.test.ts @@ -6,6 +6,7 @@ import type { import { applyProductLimitWorkbookRows, buildProductLimitRows, + normalizePricingRowsForPayload, } from "./productWorkflowRowHelpers" function createSkuItem( @@ -111,4 +112,26 @@ describe("productWorkflowRowHelpers", () => { }, ]) }) + + it("does not include fixed virtual inventory task fields in pricing saves", () => { + expect( + normalizePricingRowsForPayload([ + { + skuId: "sku-test-1", + costPrice: "88", + productTag: "", + fixedVirtualInventoryStatus: "pushed", + fixedVirtualInventoryPushedAt: "2026-08-10T09:00:01.000Z", + fixedVirtualInventoryRetryCount: "2", + fixedVirtualInventoryError: "previous failure", + }, + ]), + ).toEqual([ + { + skuId: "sku-test-1", + costPrice: "88", + productTag: "", + }, + ]) + }) }) diff --git a/furniture-3d-system/src/app/intake/non-integrated/workspaces/product-workflow/productWorkflowRowHelpers.ts b/furniture-3d-system/src/app/intake/non-integrated/workspaces/product-workflow/productWorkflowRowHelpers.ts index 9ce6cab5..62bef855 100644 --- a/furniture-3d-system/src/app/intake/non-integrated/workspaces/product-workflow/productWorkflowRowHelpers.ts +++ b/furniture-3d-system/src/app/intake/non-integrated/workspaces/product-workflow/productWorkflowRowHelpers.ts @@ -130,6 +130,17 @@ export function createProductPricingRowFromSku( jstVirtualInventoryError: existingRow?.jstVirtualInventoryError?.trim() ?? "", jstVirtualInventorySoId: existingRow?.jstVirtualInventorySoId?.trim() ?? "", + fixedVirtualInventoryStatus: existingRow?.fixedVirtualInventoryStatus, + fixedVirtualInventoryScheduledAt: + existingRow?.fixedVirtualInventoryScheduledAt?.trim() ?? "", + fixedVirtualInventoryPushedAt: + existingRow?.fixedVirtualInventoryPushedAt?.trim() ?? "", + fixedVirtualInventoryRetryCount: + existingRow?.fixedVirtualInventoryRetryCount?.trim() ?? "", + fixedVirtualInventoryLastAttemptAt: + existingRow?.fixedVirtualInventoryLastAttemptAt?.trim() ?? "", + fixedVirtualInventoryError: + existingRow?.fixedVirtualInventoryError?.trim() ?? "", }; } diff --git a/furniture-3d-system/src/server/repositories/jushuitan-fixed-virtual-inventory-task.repository.ts b/furniture-3d-system/src/server/repositories/jushuitan-fixed-virtual-inventory-task.repository.ts index 4bad84b6..f1a71ca0 100644 --- a/furniture-3d-system/src/server/repositories/jushuitan-fixed-virtual-inventory-task.repository.ts +++ b/furniture-3d-system/src/server/repositories/jushuitan-fixed-virtual-inventory-task.repository.ts @@ -54,6 +54,22 @@ class JushuitanFixedVirtualInventoryTaskRepository { }); } + async findByProductId(productId: string) { + return prisma.intakeJushuitanFixedVirtualInventoryTask.findMany({ + where: { productId }, + orderBy: [{ skuItemId: "asc" }, { createdAt: "asc" }], + select: { + skuItemId: true, + status: true, + scheduledAt: true, + retryCount: true, + lastAttemptAt: true, + pushedAt: true, + error: true, + }, + }); + } + async markSucceeded(input: FixedVirtualInventorySuccessInput) { return prisma.intakeJushuitanFixedVirtualInventoryTask.update({ where: { id: input.id }, @@ -86,7 +102,11 @@ export type JushuitanFixedVirtualInventoryTaskRecord = IntakeJushuitanFixedVirtualInventoryTask; export type JushuitanFixedVirtualInventoryTaskRepositoryContract = Pick< JushuitanFixedVirtualInventoryTaskRepository, - "createMissing" | "findDue" | "markSucceeded" | "scheduleRetry" + | "createMissing" + | "findDue" + | "findByProductId" + | "markSucceeded" + | "scheduleRetry" >; export const jushuitanFixedVirtualInventoryTaskRepository = diff --git a/furniture-3d-system/src/server/services/__tests__/intake-non-integrated-product.service.test.ts b/furniture-3d-system/src/server/services/__tests__/intake-non-integrated-product.service.test.ts index c9c7abe3..99a80bdf 100644 --- a/furniture-3d-system/src/server/services/__tests__/intake-non-integrated-product.service.test.ts +++ b/furniture-3d-system/src/server/services/__tests__/intake-non-integrated-product.service.test.ts @@ -94,6 +94,12 @@ jest.mock('../../repositories/draft-group.repository', () => ({ }, })) +jest.mock('../../repositories/jushuitan-fixed-virtual-inventory-task.repository', () => ({ + jushuitanFixedVirtualInventoryTaskRepository: { + findByProductId: jest.fn(), + }, +})) + jest.mock('../spu-number-reservation.service', () => ({ spuNumberReservationService: { reserveInTransaction: jest.fn(), @@ -136,6 +142,7 @@ import { intakeNonIntegratedProductRepository } from '../../repositories/intake- import { userRepository } from '../../repositories/user.repository' import { assetSkuGroupRepository, assetVariantRepository } from '../../repositories/asset-center.repository' import { draftGroupRepository, preSkuRepository } from '../../repositories/draft-group.repository' +import { jushuitanFixedVirtualInventoryTaskRepository } from '../../repositories/jushuitan-fixed-virtual-inventory-task.repository' import { prisma } from '@/lib/prisma' import { ErrorCode } from '@/server/core/errors' import { @@ -157,6 +164,9 @@ const mockedAssetSkuGroupRepository = jest.mocked(assetSkuGroupRepository) const mockedAssetVariantRepository = jest.mocked(assetVariantRepository) const mockedDraftGroupRepository = jest.mocked(draftGroupRepository) const mockedPreSkuRepository = jest.mocked(preSkuRepository) +const mockedFixedVirtualInventoryTaskRepository = jest.mocked( + jushuitanFixedVirtualInventoryTaskRepository, +) const mockedSpuNumberReservationService = jest.mocked(spuNumberReservationService) const mockedPrisma = jest.mocked(prisma) const mockedNodePermissionSettingsService = jest.mocked(intakeNodePermissionSettingsService) @@ -447,6 +457,7 @@ describe('intakeNonIntegratedProductService', () => { mockedRepository.findActiveRecordsBySpuInTransaction.mockResolvedValue([] as never) mockedNodePermissionSettingsService.getSettings.mockResolvedValue({ overrides: {} }) mockedUserRepository.findIdsByBusinessRole.mockResolvedValue([]) + mockedFixedVirtualInventoryTaskRepository.findByProductId.mockResolvedValue([]) }) it('lists all shared records when permissions are relaxed', async () => { @@ -1237,6 +1248,53 @@ describe('intakeNonIntegratedProductService', () => { ]) }) + it('returns fixed virtual inventory task status alongside the legacy status', async () => { + mockedRepository.findActiveById.mockResolvedValue( + createRecord({ + payload: createPayload({ + skuItems: [createSkuItem({ id: 'sku-1' })], + pricingRows: [ + { + skuId: 'sku-1', + costPrice: '199.00', + productTag: '', + jstVirtualInventoryStatus: 'failed', + jstVirtualInventoryError: '聚水潭未找到对应仓库', + }, + ], + }), + }) as never, + ) + mockedFixedVirtualInventoryTaskRepository.findByProductId.mockResolvedValue([ + { + skuItemId: 'sku-1', + status: 'pushed', + scheduledAt: new Date('2026-08-10T09:00:00.000Z'), + retryCount: 0, + lastAttemptAt: new Date('2026-08-10T09:00:00.000Z'), + pushedAt: new Date('2026-08-10T09:00:01.000Z'), + error: '', + }, + ] as never) + + const product = await intakeNonIntegratedProductService.getById( + 'non-integrated-1', + baseUser, + ) + + expect(mockedFixedVirtualInventoryTaskRepository.findByProductId).toHaveBeenCalledWith( + 'non-integrated-1', + ) + expect(product?.pricingRows).toEqual([ + expect.objectContaining({ + skuId: 'sku-1', + jstVirtualInventoryStatus: 'failed', + fixedVirtualInventoryStatus: 'pushed', + fixedVirtualInventoryPushedAt: '2026-08-10T09:00:01.000Z', + }), + ]) + }) + it('preserves the legacy modular tag after all pricing rows are already entered', async () => { mockedRepository.findActiveById.mockResolvedValue( createRecord({ diff --git a/furniture-3d-system/src/server/services/intake-non-integrated-product.service.ts b/furniture-3d-system/src/server/services/intake-non-integrated-product.service.ts index 53b10af3..a1ef946a 100644 --- a/furniture-3d-system/src/server/services/intake-non-integrated-product.service.ts +++ b/furniture-3d-system/src/server/services/intake-non-integrated-product.service.ts @@ -70,6 +70,7 @@ import { draftGroupRepository, preSkuRepository, } from '../repositories/draft-group.repository' +import { jushuitanFixedVirtualInventoryTaskRepository } from '../repositories/jushuitan-fixed-virtual-inventory-task.repository' import { assertNonIntegratedProductMasterDataReadyForFormalUse, resolveNonIntegratedSkuQuality, @@ -1979,6 +1980,51 @@ function buildStructuredProductRecordFromRecord(record: IntakeNonIntegratedRecor return applyPersistedRecordIdentity(applySharedProductTag(hydratedProduct), record) } +async function hydrateFixedVirtualInventoryStatuses( + product: SharedNonIntegratedProductRecord, +): Promise { + const tasks = await jushuitanFixedVirtualInventoryTaskRepository.findByProductId( + product.id, + ) + if (tasks.length === 0 || product.pricingRows.length === 0) return product + + const taskBySkuItemId = new Map(tasks.map((task) => [task.skuItemId, task])) + let changed = false + const pricingRows = product.pricingRows.map((row) => { + const task = taskBySkuItemId.get(row.skuId) + const fixedVirtualInventoryStatus: NonIntegratedPricingRow['fixedVirtualInventoryStatus'] = + task?.status === 'pending' + ? 'pending' + : task?.status === 'pushed' + ? 'pushed' + : undefined + if (!task || !fixedVirtualInventoryStatus) { + return row + } + + changed = true + return { + ...row, + fixedVirtualInventoryStatus, + ...(task.scheduledAt + ? { fixedVirtualInventoryScheduledAt: task.scheduledAt.toISOString() } + : {}), + ...(task.pushedAt + ? { fixedVirtualInventoryPushedAt: task.pushedAt.toISOString() } + : {}), + ...(task.retryCount > 0 + ? { fixedVirtualInventoryRetryCount: String(task.retryCount) } + : {}), + ...(task.lastAttemptAt + ? { fixedVirtualInventoryLastAttemptAt: task.lastAttemptAt.toISOString() } + : {}), + ...(task.error ? { fixedVirtualInventoryError: task.error } : {}), + } + }) + + return changed ? { ...product, pricingRows } : product +} + function buildBarcodePrintProductFromPricingRecord(record: BarcodePrintPricingRecord) { return buildStructuredProductRecordFromRecord({ ...record.product, @@ -6265,7 +6311,9 @@ class IntakeNonIntegratedProductService { if (!scopedProduct) return null const hydratedProduct = await hydrateSeriesProductDisplayName(scopedProduct) - return hydrateProductSkuModelSnapshots(hydratedProduct) + const productWithInventoryStatus = + await hydrateFixedVirtualInventoryStatuses(hydratedProduct) + return hydrateProductSkuModelSnapshots(productWithInventoryStatus) } async getDownloadableById( diff --git a/furniture-3d-system/src/server/services/jushuitan-fixed-virtual-inventory.service.test.ts b/furniture-3d-system/src/server/services/jushuitan-fixed-virtual-inventory.service.test.ts index 1f74b6b6..a6605ba8 100644 --- a/furniture-3d-system/src/server/services/jushuitan-fixed-virtual-inventory.service.test.ts +++ b/furniture-3d-system/src/server/services/jushuitan-fixed-virtual-inventory.service.test.ts @@ -43,6 +43,7 @@ function createRepository(tasks: JushuitanFixedVirtualInventoryTaskRecord[]) { return { createMissing: jest.fn(async () => 1), findDue: jest.fn(async () => tasks), + findByProductId: jest.fn(async () => []), markSucceeded: jest.fn(async () => undefined), scheduleRetry: jest.fn(async () => undefined), } satisfies JushuitanFixedVirtualInventoryTaskRepositoryContract; diff --git a/furniture-3d-system/src/server/services/oa-board-sync.service.ts b/furniture-3d-system/src/server/services/oa-board-sync.service.ts index 833b7a56..34c7ea65 100644 --- a/furniture-3d-system/src/server/services/oa-board-sync.service.ts +++ b/furniture-3d-system/src/server/services/oa-board-sync.service.ts @@ -3,6 +3,7 @@ import { type ConnectionOptions, type RowDataPacket, } from 'mysql2/promise' +import type sql from 'mssql' import { Prisma } from '@prisma/client' import { prisma } from '@/lib/prisma' import { externalEmployeeDirectoryService } from './external-employee-directory.service' @@ -322,60 +323,41 @@ function readFirstEnv(names: string[]) { return names.map(readEnv).find(Boolean) ?? '' } -export function createPlanningConnectionConfig(): string | ConnectionOptions { - const databaseUrl = readFirstEnv([ - 'OA_BOARD_SOURCE_DATABASE_URL', - 'PLANNING_DATABASE_URL', - ]) +type OaBoardSourceDriver = 'mysql' | 'mssql' - if (databaseUrl) { - return databaseUrl - } +function readOaBoardSourceDriver(): OaBoardSourceDriver { + const configured = readEnv('OA_BOARD_SOURCE_DRIVER').toLowerCase() + if (!configured || configured === 'mysql') return 'mysql' + if (configured === 'mssql' || configured === 'sqlserver') return 'mssql' + throw new Error(`Unsupported OA_BOARD_SOURCE_DRIVER: ${configured}`) +} - const host = readFirstEnv([ - 'OA_BOARD_SOURCE_DB_HOST', - 'PLANNING_DB_HOST', - 'JST_PLANNING_DB_HOST', - ]) - const database = readFirstEnv([ - 'OA_BOARD_SOURCE_DB_NAME', - 'PLANNING_DB_NAME', - 'JST_PLANNING_DB_NAME', - ]) - const user = readFirstEnv([ - 'OA_BOARD_SOURCE_DB_USER', - 'PLANNING_DB_USER', - 'JST_PLANNING_DB_USER', - ]) - const password = readFirstEnv([ - 'OA_BOARD_SOURCE_DB_PASSWORD', - 'PLANNING_DB_PASSWORD', - 'JST_PLANNING_DB_PASSWORD', - ]) +function createMysqlConnectionConfig(options: { + databaseUrlNames: string[] + hostNames: string[] + portNames: string[] + databaseNames: string[] + userNames: string[] + passwordNames: string[] + charsetNames: string[] + sslNames: string[] + missingMessage: string +}): string | ConnectionOptions { + const databaseUrl = readFirstEnv(options.databaseUrlNames) + if (databaseUrl) return databaseUrl + + const host = readFirstEnv(options.hostNames) + const database = readFirstEnv(options.databaseNames) + const user = readFirstEnv(options.userNames) + const password = readFirstEnv(options.passwordNames) if (!host || !database || !user || !password) { - throw new Error('OA 看板来源数据库未配置') + throw new Error(options.missingMessage) } - const port = Number( - readFirstEnv([ - 'OA_BOARD_SOURCE_DB_PORT', - 'PLANNING_DB_PORT', - 'JST_PLANNING_DB_PORT', - ]) || '3306', - ) - const charset = - readFirstEnv([ - 'OA_BOARD_SOURCE_DB_CHARSET', - 'PLANNING_DB_CHARSET', - 'JST_PLANNING_DB_CHARSET', - ]) || 'utf8mb4' - const sslEnabled = - readFirstEnv([ - 'OA_BOARD_SOURCE_DB_SSL', - 'PLANNING_DB_SSL', - 'JST_PLANNING_DB_SSL', - ]) === 'true' + const port = Number(readFirstEnv(options.portNames) || '3306') + const charset = readFirstEnv(options.charsetNames) || 'utf8mb4' + const sslEnabled = readFirstEnv(options.sslNames) === 'true' return { host, @@ -390,6 +372,83 @@ export function createPlanningConnectionConfig(): string | ConnectionOptions { } } +export function createOaBoardSnapshotConnectionConfig(): string | ConnectionOptions { + return createMysqlConnectionConfig({ + databaseUrlNames: ['OA_BOARD_SNAPSHOT_DATABASE_URL', 'PLANNING_DATABASE_URL'], + hostNames: ['OA_BOARD_SNAPSHOT_DB_HOST', 'PLANNING_DB_HOST', 'JST_PLANNING_DB_HOST'], + portNames: ['OA_BOARD_SNAPSHOT_DB_PORT', 'PLANNING_DB_PORT', 'JST_PLANNING_DB_PORT'], + databaseNames: ['OA_BOARD_SNAPSHOT_DB_NAME', 'PLANNING_DB_NAME', 'JST_PLANNING_DB_NAME'], + userNames: ['OA_BOARD_SNAPSHOT_DB_USER', 'PLANNING_DB_USER', 'JST_PLANNING_DB_USER'], + passwordNames: ['OA_BOARD_SNAPSHOT_DB_PASSWORD', 'PLANNING_DB_PASSWORD', 'JST_PLANNING_DB_PASSWORD'], + charsetNames: ['OA_BOARD_SNAPSHOT_DB_CHARSET', 'PLANNING_DB_CHARSET', 'JST_PLANNING_DB_CHARSET'], + sslNames: ['OA_BOARD_SNAPSHOT_DB_SSL', 'PLANNING_DB_SSL', 'JST_PLANNING_DB_SSL'], + missingMessage: 'OA 看板快照数据库未配置', + }) +} + +export function createPlanningConnectionConfig(): string | ConnectionOptions { + return createOaBoardSnapshotConnectionConfig() +} + +function createMysqlSourceConnectionConfig(): string | ConnectionOptions { + return createMysqlConnectionConfig({ + databaseUrlNames: ['OA_BOARD_SOURCE_DATABASE_URL', 'PLANNING_DATABASE_URL'], + hostNames: ['OA_BOARD_SOURCE_DB_HOST', 'PLANNING_DB_HOST', 'JST_PLANNING_DB_HOST'], + portNames: ['OA_BOARD_SOURCE_DB_PORT', 'PLANNING_DB_PORT', 'JST_PLANNING_DB_PORT'], + databaseNames: ['OA_BOARD_SOURCE_DB_NAME', 'PLANNING_DB_NAME', 'JST_PLANNING_DB_NAME'], + userNames: ['OA_BOARD_SOURCE_DB_USER', 'PLANNING_DB_USER', 'JST_PLANNING_DB_USER'], + passwordNames: ['OA_BOARD_SOURCE_DB_PASSWORD', 'PLANNING_DB_PASSWORD', 'JST_PLANNING_DB_PASSWORD'], + charsetNames: ['OA_BOARD_SOURCE_DB_CHARSET', 'PLANNING_DB_CHARSET', 'JST_PLANNING_DB_CHARSET'], + sslNames: ['OA_BOARD_SOURCE_DB_SSL', 'PLANNING_DB_SSL', 'JST_PLANNING_DB_SSL'], + missingMessage: 'OA 看板来源数据库未配置', + }) +} + +function assertSqlServerIdentifier(value: string, label: string) { + if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(value)) { + throw new Error(`Invalid ${label}: ${value}`) + } +} + +function quoteSqlServerIdentifier(value: string, label: string) { + assertSqlServerIdentifier(value, label) + return `[${value}]` +} + +function createSqlServerSourceConfig(): sql.config { + const host = readEnv('OA_BOARD_MSSQL_HOST') + const database = readEnv('OA_BOARD_MSSQL_DATABASE') + const user = readEnv('OA_BOARD_MSSQL_USERNAME') + const password = readEnv('OA_BOARD_MSSQL_PASSWORD') + + if (!host || !database || !user || !password) { + throw new Error('OA 看板 SQL Server 来源数据库未配置') + } + + return { + server: host, + port: Number(readEnv('OA_BOARD_MSSQL_PORT') || '1433'), + database, + user, + password, + options: { + encrypt: readEnv('OA_BOARD_MSSQL_ENCRYPT') === 'true', + trustServerCertificate: readEnv('OA_BOARD_MSSQL_TRUST_SERVER_CERTIFICATE') !== 'false', + }, + connectionTimeout: Number(readEnv('OA_BOARD_MSSQL_CONNECTION_TIMEOUT_MS') || '15000'), + requestTimeout: Number(readEnv('OA_BOARD_MSSQL_REQUEST_TIMEOUT_MS') || '120000'), + } +} + +async function loadSqlServer() { + const module = await import('mssql') + return module.default +} + +function getSqlServerSourceSchema() { + return quoteSqlServerIdentifier(readEnv('OA_BOARD_MSSQL_SCHEMA') || 'rpt', 'OA_BOARD_MSSQL_SCHEMA') +} + function normalizeText(value: unknown): string | null { const text = String(value ?? '').trim() return text || null @@ -2474,171 +2533,129 @@ function buildProjectGraph( } } -async function fetchSourceData(limit?: number): Promise { - const connectionConfig = createPlanningConnectionConfig() +type OaBoardSourceRows = { + workflowRows: SourceWorkflowRow[] + schemeRows: SourceSchemeRow[] + taskRows: SourceTaskRow[] +} + +function sourceWorkflowColumns(prefix = '') { + const column = (name: string) => `${prefix}${name}` + return [ + `${column('finishTime')} AS finishTime`, + `${column('businessId')} AS businessId`, + `${column('title')} AS title`, + `${column('result')} AS result`, + `${column('originatorUserId')} AS originatorUserId`, + `${column('originatorDeptId')} AS originatorDeptId`, + `${column('originatorDeptName')} AS originatorDeptName`, + `${column('status')} AS status`, + `${column('bizAction')} AS bizAction`, + `${column('processInstanceId')} AS processInstanceId`, + `${column('joinAt')} AS joinAt`, + `${column('updateAt')} AS updateAt`, + `${column('createAt')} AS createAt`, + `${column('processCode')} AS processCode`, + `${column('flowType')} AS flowType`, + ].join(', ') +} + +function buildSourceLinkFieldNames(limit?: number) { + return Array.from( + new Set([ + ...(limit ? OA_BOARD_JOIN_FIELD_NAMES.openingCode : []), + ...OA_BOARD_JOIN_FIELD_NAMES.orderCode, + ...OA_BOARD_JOIN_FIELD_NAMES.pricingCode, + ...OA_BOARD_JOIN_FIELD_NAMES.artCode, + ]), + ) +} + +function extractOpeningCodes(rows: SourceSchemeRow[]) { + return Array.from( + new Set( + rows + .filter((row) => matchesFieldName(row.name, OA_BOARD_JOIN_FIELD_NAMES.openingCode)) + .map((row) => normalizeText(row.value)) + .filter((value): value is string => Boolean(value)), + ), + ) +} + +function uniqueSourceWorkflowRows(rows: SourceWorkflowRow[]) { + const workflowMap = new Map() + rows.forEach((row) => workflowMap.set(row.processInstanceId, row)) + return Array.from(workflowMap.values()) +} + +function getSourceProcessInstanceIds(rows: SourceWorkflowRow[]) { + return Array.from( + new Set( + rows + .map((workflowRow) => normalizeText(workflowRow.processInstanceId)) + .filter((value): value is string => Boolean(value)), + ), + ) +} + +async function fetchMysqlSourceRows(limit?: number): Promise { + const connectionConfig = createMysqlSourceConnectionConfig() const connection = typeof connectionConfig === 'string' ? await mysql.createConnection(connectionConfig) : await mysql.createConnection(connectionConfig) try { - const processCodes = Object.values(OA_BOARD_PROCESS_CODES) const reviewProcessCode = OA_BOARD_PROCESS_CODES.review + const processCodes = Object.values(OA_BOARD_PROCESS_CODES) const workflowLimitSql = limit ? `LIMIT ${Math.max(1, Math.floor(limit))}` : '' - let workflowRows: SourceWorkflowRow[] = [] - - if (limit) { - const [reviewWorkflowRows] = await connection.query( - ` - SELECT finishTime, businessId, title, result, originatorUserId, originatorDeptId, - originatorDeptName, status, bizAction, processInstanceId, joinAt, - updateAt, createAt, processCode, flowType - FROM ding_Workflow - WHERE processCode = ? - ORDER BY updateAt DESC - ${workflowLimitSql} - `, - [reviewProcessCode], - ) - - const reviewProcessInstanceIds = reviewWorkflowRows.map((row) => row.processInstanceId) - const [reviewSchemeRows] = - reviewProcessInstanceIds.length > 0 - ? await connection.query( - ` - SELECT processInstanceId, componentType, name, CAST(value AS CHAR) AS value, CAST(extValue AS CHAR) AS extValue - FROM ding_WorkflowScheme - WHERE processInstanceId IN (${reviewProcessInstanceIds.map(() => '?').join(', ')}) - `, - reviewProcessInstanceIds, - ) - : [[] as SourceSchemeRow[]] - - const openingCodes = Array.from( - new Set( - reviewSchemeRows - .filter((row) => - matchesFieldName(row.name, OA_BOARD_JOIN_FIELD_NAMES.openingCode), - ) - .map((row) => normalizeText(row.value)) - .filter((value): value is string => Boolean(value)), - ), - ) - - const linkedFieldNames = Array.from( - new Set([ - ...OA_BOARD_JOIN_FIELD_NAMES.openingCode, - ...OA_BOARD_JOIN_FIELD_NAMES.orderCode, - ...OA_BOARD_JOIN_FIELD_NAMES.pricingCode, - ...OA_BOARD_JOIN_FIELD_NAMES.artCode, - ]), - ) - - const otherProcessCodes = processCodes.filter((code) => code !== reviewProcessCode) - const [linkedWorkflowRows] = - openingCodes.length > 0 - ? await connection.query( - ` - SELECT DISTINCT w.finishTime, w.businessId, w.title, w.result, w.originatorUserId, w.originatorDeptId, - w.originatorDeptName, w.status, w.bizAction, w.processInstanceId, w.joinAt, - w.updateAt, w.createAt, w.processCode, w.flowType - FROM ding_Workflow w - JOIN ding_WorkflowScheme s - ON s.processInstanceId = w.processInstanceId - WHERE w.processCode IN (${otherProcessCodes.map(() => '?').join(', ')}) - AND s.name IN (${linkedFieldNames.map(() => '?').join(', ')}) - AND CAST(s.value AS CHAR) IN (${openingCodes.map(() => '?').join(', ')}) - ORDER BY w.updateAt DESC - `, - [...otherProcessCodes, ...linkedFieldNames, ...openingCodes], - ) - : [[] as SourceWorkflowRow[]] - - const workflowMap = new Map() - reviewWorkflowRows.forEach((row) => workflowMap.set(row.processInstanceId, row)) - linkedWorkflowRows.forEach((row) => workflowMap.set(row.processInstanceId, row)) - workflowRows = Array.from(workflowMap.values()) - } else { - const [reviewWorkflowRows] = await connection.query( - ` - SELECT finishTime, businessId, title, result, originatorUserId, originatorDeptId, - originatorDeptName, status, bizAction, processInstanceId, joinAt, - updateAt, createAt, processCode, flowType - FROM ding_Workflow - WHERE processCode = ? - ORDER BY updateAt DESC - `, - [reviewProcessCode], - ) - - const reviewProcessInstanceIds = reviewWorkflowRows.map((row) => row.processInstanceId) - const [reviewSchemeRows] = - reviewProcessInstanceIds.length > 0 - ? await connection.query( - ` - SELECT processInstanceId, componentType, name, CAST(value AS CHAR) AS value, CAST(extValue AS CHAR) AS extValue - FROM ding_WorkflowScheme - WHERE processInstanceId IN (${reviewProcessInstanceIds.map(() => '?').join(', ')}) - `, - reviewProcessInstanceIds, - ) - : [[] as SourceSchemeRow[]] - - const openingCodes = Array.from( - new Set( - reviewSchemeRows - .filter((row) => - matchesFieldName(row.name, OA_BOARD_JOIN_FIELD_NAMES.openingCode), - ) - .map((row) => normalizeText(row.value)) - .filter((value): value is string => Boolean(value)), - ), - ) - - const linkedFieldNames = Array.from( - new Set([ - ...OA_BOARD_JOIN_FIELD_NAMES.orderCode, - ...OA_BOARD_JOIN_FIELD_NAMES.pricingCode, - ...OA_BOARD_JOIN_FIELD_NAMES.artCode, - ]), - ) - - const otherProcessCodes = processCodes.filter((code) => code !== reviewProcessCode) - const [linkedWorkflowRows] = - openingCodes.length > 0 - ? await connection.query( - ` - SELECT DISTINCT w.finishTime, w.businessId, w.title, w.result, w.originatorUserId, w.originatorDeptId, - w.originatorDeptName, w.status, w.bizAction, w.processInstanceId, w.joinAt, - w.updateAt, w.createAt, w.processCode, w.flowType - FROM ding_Workflow w - JOIN ding_WorkflowScheme s - ON s.processInstanceId = w.processInstanceId - WHERE w.processCode IN (${otherProcessCodes.map(() => '?').join(', ')}) - AND s.name IN (${linkedFieldNames.map(() => '?').join(', ')}) - AND CAST(s.value AS CHAR) IN (${openingCodes.map(() => '?').join(', ')}) - ORDER BY w.updateAt DESC - `, - [...otherProcessCodes, ...linkedFieldNames, ...openingCodes], - ) - : [[] as SourceWorkflowRow[]] - - const workflowMap = new Map() - reviewWorkflowRows.forEach((row) => workflowMap.set(row.processInstanceId, row)) - linkedWorkflowRows.forEach((row) => workflowMap.set(row.processInstanceId, row)) - workflowRows = Array.from(workflowMap.values()) - } - - const processInstanceIds = Array.from( - new Set( - workflowRows - .map((workflowRow) => normalizeText(workflowRow.processInstanceId)) - .filter((value): value is string => Boolean(value)), - ), + const [reviewWorkflowRows] = await connection.query( + ` + SELECT ${sourceWorkflowColumns()} + FROM ding_Workflow + WHERE processCode = ? + ORDER BY updateAt DESC + ${workflowLimitSql} + `, + [reviewProcessCode], ) + const reviewProcessInstanceIds = getSourceProcessInstanceIds(reviewWorkflowRows) + const [reviewSchemeRows] = + reviewProcessInstanceIds.length > 0 + ? await connection.query( + ` + SELECT processInstanceId, componentType, name, CAST(value AS CHAR) AS value, CAST(extValue AS CHAR) AS extValue + FROM ding_WorkflowScheme + WHERE processInstanceId IN (${reviewProcessInstanceIds.map(() => '?').join(', ')}) + `, + reviewProcessInstanceIds, + ) + : [[] as SourceSchemeRow[]] + + const openingCodes = extractOpeningCodes(reviewSchemeRows) + const otherProcessCodes = processCodes.filter((code) => code !== reviewProcessCode) + const linkedFieldNames = buildSourceLinkFieldNames(limit) + const [linkedWorkflowRows] = + openingCodes.length > 0 + ? await connection.query( + ` + SELECT DISTINCT ${sourceWorkflowColumns('w.')} + FROM ding_Workflow w + JOIN ding_WorkflowScheme s ON s.processInstanceId = w.processInstanceId + WHERE w.processCode IN (${otherProcessCodes.map(() => '?').join(', ')}) + AND s.name IN (${linkedFieldNames.map(() => '?').join(', ')}) + AND CAST(s.value AS CHAR) IN (${openingCodes.map(() => '?').join(', ')}) + ORDER BY w.updateAt DESC + `, + [...otherProcessCodes, ...linkedFieldNames, ...openingCodes], + ) + : [[] as SourceWorkflowRow[]] + + const workflowRows = uniqueSourceWorkflowRows([...reviewWorkflowRows, ...linkedWorkflowRows]) + const processInstanceIds = getSourceProcessInstanceIds(workflowRows) if (processInstanceIds.length === 0) { - return [] + return { workflowRows: [], schemeRows: [], taskRows: [] } } const [schemeRows] = await connection.query( @@ -2649,68 +2666,181 @@ async function fetchSourceData(limit?: number): Promise `, processInstanceIds, ) - const [taskRows] = await connection.query( ` SELECT t.id, t.processInstanceId, t.result, t.activityId, t.finishTime, t.createTime, t.userId, t.taskId, t.status, w.processCode AS processCode FROM ding_WorkflowTask t - JOIN ding_Workflow w - ON w.processInstanceId = t.processInstanceId + JOIN ding_Workflow w ON w.processInstanceId = t.processInstanceId WHERE t.processInstanceId IN (${processInstanceIds.map(() => '?').join(', ')}) `, processInstanceIds, ) - const schemeMap = new Map() - schemeRows.forEach((schemeRow) => { - const current = schemeMap.get(schemeRow.processInstanceId) ?? [] - current.push(schemeRow) - schemeMap.set(schemeRow.processInstanceId, current) - }) - - const taskMap = new Map() - taskRows.forEach((taskRow) => { - const current = taskMap.get(taskRow.processInstanceId) ?? [] - current.push(taskRow) - taskMap.set(taskRow.processInstanceId, current) - }) - - const [mappingLookup, nodeRuleLookup] = await Promise.all([ - loadActivityMappingLookup(), - loadNodeRuleLookup(), - ]) - - return workflowRows - .map((workflowRow) => { - const flowRecord = buildNormalizedFlowRecord( - workflowRow, - buildFieldSnapshots(schemeMap.get(workflowRow.processInstanceId) ?? []), - buildTaskRecordsWithMappings( - taskMap.get(workflowRow.processInstanceId) ?? [], - OA_BOARD_FLOW_LABELS[inferFlowType(workflowRow.processCode) as OaBoardFlowType] ?? - workflowRow.processCode, - mappingLookup, - ), - ) - - if (!flowRecord) return null - - return { - ...flowRecord, - taskRecords: applyNodeRuleTargets( - flowRecord.taskRecords, - flowRecord.productType, - nodeRuleLookup, - ), - } - }) - .filter((record): record is NormalizedFlowRecord => Boolean(record)) + return { workflowRows, schemeRows, taskRows } } finally { await connection.end() } } +function addSqlServerTextParameters( + sqlServer: typeof import('mssql').default, + request: sql.Request, + prefix: string, + values: string[], +) { + return values.map((value, index) => { + const name = `${prefix}${index}` + request.input(name, sqlServer.NVarChar(4000), value) + return `@${name}` + }) +} + +async function fetchSqlServerSourceRows(limit?: number): Promise { + const sqlServer = await loadSqlServer() + const pool = await sqlServer.connect(createSqlServerSourceConfig()) + const schema = getSqlServerSourceSchema() + const workflowTable = `${schema}.[ding_Workflow]` + const schemeTable = `${schema}.[ding_WorkflowScheme]` + const taskTable = `${schema}.[ding_WorkflowTask]` + + try { + const reviewProcessCode = OA_BOARD_PROCESS_CODES.review + const processCodes = Object.values(OA_BOARD_PROCESS_CODES) + const reviewRequest = pool.request().input('reviewProcessCode', sqlServer.NVarChar(100), reviewProcessCode) + if (limit) reviewRequest.input('workflowLimit', sqlServer.Int, Math.max(1, Math.floor(limit))) + const reviewResult = await reviewRequest.query(` + SELECT ${limit ? 'TOP (@workflowLimit)' : ''} ${sourceWorkflowColumns('[w].')} + FROM ${workflowTable} AS [w] + WHERE [w].[processCode] = @reviewProcessCode + ORDER BY [w].[updateAt] DESC + `) + const reviewWorkflowRows = reviewResult.recordset as SourceWorkflowRow[] + const reviewProcessInstanceIds = getSourceProcessInstanceIds(reviewWorkflowRows) + + let reviewSchemeRows: SourceSchemeRow[] = [] + if (reviewProcessInstanceIds.length > 0) { + const request = pool.request() + const ids = addSqlServerTextParameters(sqlServer, request, 'reviewProcessInstanceId', reviewProcessInstanceIds) + const result = await request.query(` + SELECT [processInstanceId] AS processInstanceId, [componentType] AS componentType, [name] AS name, + [value] AS value, [extValue] AS extValue + FROM ${schemeTable} + WHERE [processInstanceId] IN (${ids.join(', ')}) + `) + reviewSchemeRows = result.recordset as SourceSchemeRow[] + } + + const openingCodes = extractOpeningCodes(reviewSchemeRows) + const otherProcessCodes = processCodes.filter((code) => code !== reviewProcessCode) + const linkedFieldNames = buildSourceLinkFieldNames(limit) + let linkedWorkflowRows: SourceWorkflowRow[] = [] + if (openingCodes.length > 0) { + const request = pool.request() + const processCodeParameters = addSqlServerTextParameters(sqlServer, request, 'processCode', otherProcessCodes) + const fieldNameParameters = addSqlServerTextParameters(sqlServer, request, 'fieldName', linkedFieldNames) + const openingCodeParameters = addSqlServerTextParameters(sqlServer, request, 'openingCode', openingCodes) + const result = await request.query(` + SELECT DISTINCT ${sourceWorkflowColumns('[w].')} + FROM ${workflowTable} AS [w] + JOIN ${schemeTable} AS [s] ON [s].[processInstanceId] = [w].[processInstanceId] + WHERE [w].[processCode] IN (${processCodeParameters.join(', ')}) + AND [s].[name] IN (${fieldNameParameters.join(', ')}) + AND [s].[value] IN (${openingCodeParameters.join(', ')}) + ORDER BY [w].[updateAt] DESC + `) + linkedWorkflowRows = result.recordset as SourceWorkflowRow[] + } + + const workflowRows = uniqueSourceWorkflowRows([...reviewWorkflowRows, ...linkedWorkflowRows]) + const processInstanceIds = getSourceProcessInstanceIds(workflowRows) + if (processInstanceIds.length === 0) { + return { workflowRows: [], schemeRows: [], taskRows: [] } + } + + const schemeRequest = pool.request() + const schemeIds = addSqlServerTextParameters(sqlServer, schemeRequest, 'schemeProcessInstanceId', processInstanceIds) + const schemeResult = await schemeRequest.query(` + SELECT [processInstanceId] AS processInstanceId, [componentType] AS componentType, [name] AS name, + [value] AS value, [extValue] AS extValue + FROM ${schemeTable} + WHERE [processInstanceId] IN (${schemeIds.join(', ')}) + `) + + const taskRequest = pool.request() + const taskIds = addSqlServerTextParameters(sqlServer, taskRequest, 'taskProcessInstanceId', processInstanceIds) + const taskResult = await taskRequest.query(` + SELECT [t].[id] AS id, [t].[processInstanceId] AS processInstanceId, [t].[result] AS result, + [t].[activityId] AS activityId, [t].[finishTime] AS finishTime, [t].[createTime] AS createTime, + [t].[userId] AS userId, [t].[taskId] AS taskId, [t].[status] AS status, + [w].[processCode] AS processCode + FROM ${taskTable} AS [t] + JOIN ${workflowTable} AS [w] ON [w].[processInstanceId] = [t].[processInstanceId] + WHERE [t].[processInstanceId] IN (${taskIds.join(', ')}) + `) + + return { + workflowRows, + schemeRows: schemeResult.recordset as SourceSchemeRow[], + taskRows: taskResult.recordset as SourceTaskRow[], + } + } finally { + await pool.close() + } +} + +async function fetchSourceData(limit?: number): Promise { + const sourceRows = + readOaBoardSourceDriver() === 'mssql' + ? await fetchSqlServerSourceRows(limit) + : await fetchMysqlSourceRows(limit) + + const schemeMap = new Map() + sourceRows.schemeRows.forEach((schemeRow) => { + const current = schemeMap.get(schemeRow.processInstanceId) ?? [] + current.push(schemeRow) + schemeMap.set(schemeRow.processInstanceId, current) + }) + + const taskMap = new Map() + sourceRows.taskRows.forEach((taskRow) => { + const current = taskMap.get(taskRow.processInstanceId) ?? [] + current.push(taskRow) + taskMap.set(taskRow.processInstanceId, current) + }) + + const [mappingLookup, nodeRuleLookup] = await Promise.all([ + loadActivityMappingLookup(), + loadNodeRuleLookup(), + ]) + + return sourceRows.workflowRows + .map((workflowRow) => { + const flowRecord = buildNormalizedFlowRecord( + workflowRow, + buildFieldSnapshots(schemeMap.get(workflowRow.processInstanceId) ?? []), + buildTaskRecordsWithMappings( + taskMap.get(workflowRow.processInstanceId) ?? [], + OA_BOARD_FLOW_LABELS[inferFlowType(workflowRow.processCode) as OaBoardFlowType] ?? + workflowRow.processCode, + mappingLookup, + ), + ) + + if (!flowRecord) return null + + return { + ...flowRecord, + taskRecords: applyNodeRuleTargets( + flowRecord.taskRecords, + flowRecord.productType, + nodeRuleLookup, + ), + } + }) + .filter((record): record is NormalizedFlowRecord => Boolean(record)) +} + function buildProjectGroups(records: NormalizedFlowRecord[]): PreparedProjectGroup[] { const groups = new Map() const aliasToBoardKey = new Map()