From 0300858fdf0c3c0889e1d658103e258ad341c150 Mon Sep 17 00:00:00 2001 From: WuYongJie Date: Fri, 3 Jul 2026 14:46:59 +0800 Subject: [PATCH] feat(external): sync readonly views to jsreport --- .../workflow/外部只读数据库开放操作说明.md | 52 +- furniture-3d-system/package.json | 2 + .../install-external-jsreport-sync-timer.sh | 86 +++ .../sync-external-readonly-to-jsreport.ts | 712 ++++++++++++++++++ 4 files changed, 851 insertions(+), 1 deletion(-) create mode 100644 furniture-3d-system/scripts/install-external-jsreport-sync-timer.sh create mode 100644 furniture-3d-system/scripts/sync-external-readonly-to-jsreport.ts diff --git a/furniture-3d-system/docs/workflow/外部只读数据库开放操作说明.md b/furniture-3d-system/docs/workflow/外部只读数据库开放操作说明.md index 71b9c621..5fd1cf63 100644 --- a/furniture-3d-system/docs/workflow/外部只读数据库开放操作说明.md +++ b/furniture-3d-system/docs/workflow/外部只读数据库开放操作说明.md @@ -149,7 +149,57 @@ psql "postgresql://external_core_reader:@:5432/&2 + exit 1 +fi + +if [ ! -d "$APP_ROOT" ]; then + echo "Missing app root: $APP_ROOT" >&2 + exit 1 +fi + +if [ ! -f "$APP_ENV" ]; then + echo "Missing app env: $APP_ENV" >&2 + exit 1 +fi + +mkdir -p "$SHARED" +if [ ! -f "$SYNC_ENV" ]; then + umask 077 + cat >"$SYNC_ENV" <<'EOF' +# Furniture external read-only views -> JSReportReplica.app. +# Fill these values on the production server. Do not commit secrets to Git. +JSREPORT_REPLICA_HOST= +JSREPORT_REPLICA_PORT=1433 +JSREPORT_REPLICA_DATABASE=JSReportReplica +JSREPORT_REPLICA_USERNAME=app_writer +JSREPORT_REPLICA_PASSWORD= +JSREPORT_REPLICA_SCHEMA=app +JSREPORT_REPLICA_ENCRYPT=false +JSREPORT_REPLICA_TRUST_SERVER_CERTIFICATE=true +JSREPORT_REPLICA_CONNECTION_TIMEOUT_MS=15000 +JSREPORT_REPLICA_REQUEST_TIMEOUT_MS=120000 +EXTERNAL_JSREPORT_BATCH_SIZE=1000 +EXTERNAL_JSREPORT_VIEWS=product_catalog_wide_v1,intake_product_limits_v1,warehouse_locations_v1,sku_warehouse_stock_v1,packaged_items_v1,sku_quality_inspections_v1 +EOF + echo "Created env template: $SYNC_ENV" +fi + +chmod 600 "$SYNC_ENV" + +cat >"$SERVICE_FILE" <"$TIMER_FILE" < void +} + +type SyncSummary = { + view: string + targetTable: string + sourceRows: number + copiedRows: number + targetRows: number | null + recreated: boolean +} + +const DEFAULT_CORE_VIEWS = [ + 'product_catalog_wide_v1', + 'intake_product_limits_v1', + 'warehouse_locations_v1', + 'sku_warehouse_stock_v1', + 'packaged_items_v1', + 'sku_quality_inspections_v1', +] + +const KEY_TEXT_COLUMNS = new Set([ + 'product_id', + 'product_model_number', + 'product_business_key', + 'sku_id', + 'sku_code', + 'sku_business_key', + 'barcode', + 'barcode_business_key', + 'warehouse_location_id', + 'warehouse_location_code', + 'packaged_item_id', + 'packaged_item_code', + 'record_key', + 'material_id', + 'material_code', + 'asset_variant_id', + 'spec_code', + 'source_pre_sku_id', + 'source_asset_variant_id', + 'source_asset_variant_ids', + 'limit_source_product_business_key', + 'limit_source_sku_business_key', +]) + +const INDEX_CANDIDATE_COLUMNS = [ + 'barcode_business_key', + 'barcode', + 'sku_business_key', + 'sku_code', + 'product_business_key', + 'product_model_number', + 'warehouse_location_code', + 'record_key', + 'packaged_item_code', + 'material_code', + 'spec_code', +] + +function parseCliOptions(argv: string[]): CliOptions { + const apply = argv.includes('--apply') + const dryRun = argv.includes('--dry-run') || !apply + + if (apply && argv.includes('--dry-run')) { + throw new Error('Cannot pass --apply and --dry-run together.') + } + + const batchSize = Number(readArg(argv, '--batch-size') ?? env('EXTERNAL_JSREPORT_BATCH_SIZE', { defaultValue: '1000' })) + if (!Number.isInteger(batchSize) || batchSize <= 0 || batchSize > 10000) { + throw new Error('--batch-size must be an integer between 1 and 10000.') + } + + const sourceSchema = readArg(argv, '--source-schema') ?? env('EXTERNAL_JSREPORT_SOURCE_SCHEMA', { defaultValue: 'external' }) + const targetSchema = readArg(argv, '--target-schema') ?? env('JSREPORT_REPLICA_SCHEMA', { defaultValue: 'app' }) + const views = splitList(readArg(argv, '--views') ?? env('EXTERNAL_JSREPORT_VIEWS', { defaultValue: DEFAULT_CORE_VIEWS.join(',') })) + + assertIdentifier(sourceSchema, 'source schema') + assertIdentifier(targetSchema, 'target schema') + for (const view of views) assertIdentifier(view, 'view') + + if (views.length === 0) { + throw new Error('No views configured for sync.') + } + + return { apply, dryRun, batchSize, sourceSchema, targetSchema, views } +} + +function readArg(argv: string[], name: string): string | null { + const equalsPrefix = `${name}=` + const equalsValue = argv.find((item) => item.startsWith(equalsPrefix)) + if (equalsValue) return equalsValue.slice(equalsPrefix.length) + const index = argv.indexOf(name) + if (index >= 0 && argv[index + 1]) return argv[index + 1] + return null +} + +function splitList(value: string): string[] { + return value + .split(',') + .map((item) => item.trim()) + .filter(Boolean) +} + +function env(name: string, options?: { fallbackNames?: string[]; defaultValue?: string; required?: boolean }): string { + const names = [name, ...(options?.fallbackNames ?? [])] + for (const item of names) { + const value = process.env[item]?.trim() + if (value) return value + } + if (options?.defaultValue !== undefined) return options.defaultValue + if (options?.required) throw new Error(`Missing env: ${name}`) + return '' +} + +function assertIdentifier(value: string, label: string) { + if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(value)) { + throw new Error(`Invalid ${label}: ${value}`) + } +} + +function qPg(identifier: string): string { + assertIdentifier(identifier, 'PostgreSQL identifier') + return `"${identifier}"` +} + +function qSql(identifier: string): string { + return `[${identifier.replaceAll(']', ']]')}]` +} + +function sqlTableName(schema: string, table: string): string { + return `${qSql(schema)}.${qSql(table)}` +} + +function getPgConnectionString(): string { + return env('EXTERNAL_JSREPORT_SOURCE_DATABASE_URL', { + fallbackNames: ['DATABASE_URL'], + required: true, + }) +} + +function getSqlServerConfig(): sql.config { + return { + server: env('JSREPORT_REPLICA_HOST', { + fallbackNames: ['JSREPORT_REPLICA_SERVER'], + required: true, + }), + port: Number(env('JSREPORT_REPLICA_PORT', { defaultValue: '1433' })), + database: env('JSREPORT_REPLICA_DATABASE', { required: true }), + user: env('JSREPORT_REPLICA_USERNAME', { + fallbackNames: ['JSREPORT_REPLICA_USER'], + required: true, + }), + password: env('JSREPORT_REPLICA_PASSWORD', { required: true }), + options: { + encrypt: env('JSREPORT_REPLICA_ENCRYPT', { defaultValue: 'false' }) === 'true', + trustServerCertificate: env('JSREPORT_REPLICA_TRUST_SERVER_CERTIFICATE', { defaultValue: 'true' }) !== 'false', + }, + pool: { + max: Number(env('JSREPORT_REPLICA_POOL_MAX', { defaultValue: '4' })), + min: 0, + idleTimeoutMillis: Number(env('JSREPORT_REPLICA_IDLE_TIMEOUT_MS', { defaultValue: '30000' })), + }, + requestTimeout: Number(env('JSREPORT_REPLICA_REQUEST_TIMEOUT_MS', { defaultValue: '120000' })), + connectionTimeout: Number(env('JSREPORT_REPLICA_CONNECTION_TIMEOUT_MS', { defaultValue: '15000' })), + } +} + +function mapColumn(column: SourceColumn): SyncColumn { + const dataType = column.data_type.toLowerCase() + const udtName = column.udt_name.toLowerCase() + const columnName = column.column_name + + if (dataType === 'boolean' || udtName === 'bool') { + return withSql(column, 'bit', (table) => table.columns.add(columnName, sql.Bit, { nullable: true })) + } + if (dataType === 'smallint' || udtName === 'int2') { + return withSql(column, 'smallint', (table) => table.columns.add(columnName, sql.SmallInt, { nullable: true })) + } + if (dataType === 'integer' || udtName === 'int4') { + return withSql(column, 'int', (table) => table.columns.add(columnName, sql.Int, { nullable: true })) + } + if (dataType === 'bigint' || udtName === 'int8') { + return withSql(column, 'bigint', (table) => table.columns.add(columnName, sql.BigInt, { nullable: true })) + } + if (dataType === 'numeric' || dataType === 'decimal') { + const precision = clamp(column.numeric_precision ?? 38, 1, 38) + const scale = clamp(column.numeric_scale ?? 10, 0, Math.min(precision, 18)) + return withSql(column, `decimal(${precision}, ${scale})`, (table) => + table.columns.add(columnName, sql.Decimal(precision, scale), { nullable: true }), + ) + } + if (dataType === 'double precision' || udtName === 'float8') { + return withSql(column, 'float', (table) => table.columns.add(columnName, sql.Float, { nullable: true })) + } + if (dataType === 'real' || udtName === 'float4') { + return withSql(column, 'real', (table) => table.columns.add(columnName, sql.Real, { nullable: true })) + } + if (dataType === 'date') { + return withSql(column, 'date', (table) => table.columns.add(columnName, sql.Date, { nullable: true })) + } + if (dataType.startsWith('timestamp')) { + return withSql(column, 'datetime2(3)', (table) => table.columns.add(columnName, sql.DateTime2(3), { nullable: true })) + } + + if (dataType === 'character varying' || dataType === 'character') { + const length = column.character_maximum_length + if (length && length > 0 && length <= 4000) { + return withSql(column, `nvarchar(${length})`, (table) => + table.columns.add(columnName, sql.NVarChar(length), { nullable: true }), + ) + } + } + + if (udtName === 'uuid') { + return withSql(column, 'nvarchar(36)', (table) => table.columns.add(columnName, sql.NVarChar(36), { nullable: true })) + } + + if (KEY_TEXT_COLUMNS.has(columnName)) { + return withSql(column, 'nvarchar(512)', (table) => table.columns.add(columnName, sql.NVarChar(512), { nullable: true })) + } + + if (columnName.endsWith('_status') || columnName.endsWith('_type') || columnName.endsWith('_conclusion')) { + return withSql(column, 'nvarchar(128)', (table) => table.columns.add(columnName, sql.NVarChar(128), { nullable: true })) + } + + return withSql(column, 'nvarchar(max)', (table) => table.columns.add(columnName, sql.NVarChar(sql.MAX), { nullable: true })) +} + +function withSql(column: SourceColumn, sqlDeclaration: string, addBulkColumn: (table: sql.Table) => void): SyncColumn { + return { ...column, sqlDeclaration, addBulkColumn } +} + +function clamp(value: number, min: number, max: number): number { + return Math.min(Math.max(value, min), max) +} + +async function getSourceColumns(pg: Client, sourceSchema: string, view: string): Promise { + const result = await pg.query( + ` + SELECT + column_name, + ordinal_position, + data_type, + udt_name, + character_maximum_length, + numeric_precision, + numeric_scale + FROM information_schema.columns + WHERE table_schema = $1 + AND table_name = $2 + ORDER BY ordinal_position + `, + [sourceSchema, view], + ) + if (result.rowCount === 0) { + throw new Error(`Missing source view or columns: ${sourceSchema}.${view}`) + } + return result.rows.map(mapColumn) +} + +async function getSourceCount(pg: Client, sourceSchema: string, view: string): Promise { + const result = await pg.query<{ row_count: string }>( + `SELECT COUNT(*)::bigint AS row_count FROM ${qPg(sourceSchema)}.${qPg(view)}`, + ) + return Number(result.rows[0]?.row_count ?? 0) +} + +function chooseOrderColumn(columns: SyncColumn[]): string { + for (const candidate of ['barcode_business_key', 'sku_business_key', 'product_business_key', 'barcode', 'sku_code']) { + if (columns.some((column) => column.column_name === candidate)) return candidate + } + return columns[0]?.column_name ?? '' +} + +async function fetchSourceBatch( + pg: Client, + sourceSchema: string, + view: string, + columns: SyncColumn[], + limit: number, + offset: number, +): Promise[]> { + const columnList = columns.map((column) => qPg(column.column_name)).join(', ') + const orderColumn = chooseOrderColumn(columns) + const result = await pg.query>( + ` + SELECT ${columnList} + FROM ${qPg(sourceSchema)}.${qPg(view)} + ORDER BY ${qPg(orderColumn)} NULLS LAST + LIMIT $1 OFFSET $2 + `, + [limit, offset], + ) + return result.rows +} + +async function ensureSyncLogTable(pool: sql.ConnectionPool, targetSchema: string) { + await pool.request().query(` + IF OBJECT_ID(N'${targetSchema}.external_readonly_sync_runs', N'U') IS NULL + BEGIN + CREATE TABLE ${sqlTableName(targetSchema, 'external_readonly_sync_runs')} ( + run_id uniqueidentifier NOT NULL, + source_schema nvarchar(128) NOT NULL, + source_view nvarchar(128) NOT NULL, + target_schema nvarchar(128) NOT NULL, + target_table nvarchar(128) NOT NULL, + started_at datetime2(3) NOT NULL, + finished_at datetime2(3) NULL, + status nvarchar(32) NOT NULL, + source_row_count bigint NULL, + target_row_count bigint NULL, + error_message nvarchar(max) NULL, + CONSTRAINT ${qSql(`PK_${targetSchema}_external_readonly_sync_runs`)} PRIMARY KEY (run_id) + ); + END + `) +} + +async function startSyncRun( + pool: sql.ConnectionPool, + targetSchema: string, + payload: { + runId: string + sourceSchema: string + sourceView: string + targetTable: string + startedAt: Date + }, +) { + await pool + .request() + .input('runId', sql.UniqueIdentifier, payload.runId) + .input('sourceSchema', sql.NVarChar(128), payload.sourceSchema) + .input('sourceView', sql.NVarChar(128), payload.sourceView) + .input('targetSchema', sql.NVarChar(128), targetSchema) + .input('targetTable', sql.NVarChar(128), payload.targetTable) + .input('startedAt', sql.DateTime2(3), payload.startedAt) + .query(` + INSERT INTO ${sqlTableName(targetSchema, 'external_readonly_sync_runs')} ( + run_id, source_schema, source_view, target_schema, target_table, started_at, status + ) + VALUES (@runId, @sourceSchema, @sourceView, @targetSchema, @targetTable, @startedAt, N'RUNNING') + `) +} + +async function finishSyncRun( + pool: sql.ConnectionPool, + targetSchema: string, + payload: { + runId: string + status: 'SUCCEEDED' | 'FAILED' + sourceRows?: number + targetRows?: number | null + errorMessage?: string + }, +) { + await pool + .request() + .input('runId', sql.UniqueIdentifier, payload.runId) + .input('finishedAt', sql.DateTime2(3), new Date()) + .input('status', sql.NVarChar(32), payload.status) + .input('sourceRows', sql.BigInt, payload.sourceRows ?? null) + .input('targetRows', sql.BigInt, payload.targetRows ?? null) + .input('errorMessage', sql.NVarChar(sql.MAX), payload.errorMessage?.slice(0, 4000) ?? null) + .query(` + UPDATE ${sqlTableName(targetSchema, 'external_readonly_sync_runs')} + SET finished_at = @finishedAt, + status = @status, + source_row_count = @sourceRows, + target_row_count = @targetRows, + error_message = @errorMessage + WHERE run_id = @runId + `) +} + +async function tableExists(pool: sql.ConnectionPool, targetSchema: string, tableName: string): Promise { + const result = await pool + .request() + .input('schemaName', sql.NVarChar(128), targetSchema) + .input('tableName', sql.NVarChar(128), tableName) + .query(` + SELECT 1 AS exists_flag + FROM sys.tables t + JOIN sys.schemas s ON s.schema_id = t.schema_id + WHERE s.name = @schemaName + AND t.name = @tableName + `) + return result.recordset.length > 0 +} + +async function getTargetColumns(pool: sql.ConnectionPool, targetSchema: string, tableName: string) { + const result = await pool + .request() + .input('schemaName', sql.NVarChar(128), targetSchema) + .input('tableName', sql.NVarChar(128), tableName) + .query<{ + column_name: string + ordinal_position: number + type_name: string + max_length: number + precision: number + scale: number + }>(` + SELECT + c.name AS column_name, + c.column_id AS ordinal_position, + ty.name AS type_name, + c.max_length, + c.precision, + c.scale + FROM sys.columns c + JOIN sys.tables t ON t.object_id = c.object_id + JOIN sys.schemas s ON s.schema_id = t.schema_id + JOIN sys.types ty ON ty.user_type_id = c.user_type_id + WHERE s.name = @schemaName + AND t.name = @tableName + ORDER BY c.column_id + `) + return result.recordset +} + +function sameTargetShape(sourceColumns: SyncColumn[], targetColumns: Awaited>): boolean { + if (sourceColumns.length !== targetColumns.length) return false + return sourceColumns.every((source, index) => { + const target = targetColumns[index] + return target?.column_name === source.column_name && normalizeSqlType(target) === source.sqlDeclaration.toLowerCase() + }) +} + +function normalizeSqlType(column: { + type_name: string + max_length: number + precision: number + scale: number +}): string { + const typeName = column.type_name.toLowerCase() + if (typeName === 'nvarchar') { + if (column.max_length === -1) return 'nvarchar(max)' + return `nvarchar(${column.max_length / 2})` + } + if (typeName === 'decimal' || typeName === 'numeric') { + return `${typeName}(${column.precision}, ${column.scale})` + } + if (typeName === 'datetime2') { + return 'datetime2(3)' + } + return typeName +} + +async function createTargetTable(pool: sql.ConnectionPool, targetSchema: string, tableName: string, columns: SyncColumn[]) { + const columnSql = columns + .map((column) => `${qSql(column.column_name)} ${column.sqlDeclaration} NULL`) + .join(',\n ') + await pool.request().query(` + CREATE TABLE ${sqlTableName(targetSchema, tableName)} ( + ${columnSql} + ) + `) +} + +async function dropTableIfExists(pool: sql.ConnectionPool, targetSchema: string, tableName: string) { + await pool.request().query(` + IF OBJECT_ID(N'${targetSchema}.${tableName}', N'U') IS NOT NULL + DROP TABLE ${sqlTableName(targetSchema, tableName)} + `) +} + +async function createIndexes(pool: sql.ConnectionPool, targetSchema: string, tableName: string, columns: SyncColumn[]) { + const columnByName = new Map(columns.map((column) => [column.column_name, column])) + for (const columnName of INDEX_CANDIDATE_COLUMNS) { + const column = columnByName.get(columnName) + if (!column || column.sqlDeclaration.includes('max')) continue + const indexName = `IX_${tableName}_${columnName}`.slice(0, 120) + await pool.request().query(` + IF NOT EXISTS ( + SELECT 1 + FROM sys.indexes i + JOIN sys.tables t ON t.object_id = i.object_id + JOIN sys.schemas s ON s.schema_id = t.schema_id + WHERE s.name = N'${targetSchema}' + AND t.name = N'${tableName}' + AND i.name = N'${indexName}' + ) + CREATE INDEX ${qSql(indexName)} + ON ${sqlTableName(targetSchema, tableName)} (${qSql(columnName)}) + WHERE ${qSql(columnName)} IS NOT NULL + `) + } +} + +function makeBulkTable(targetSchema: string, tableName: string, columns: SyncColumn[]): sql.Table { + const table = new sql.Table(tableName) + table.schema = targetSchema + table.create = false + for (const column of columns) { + column.addBulkColumn(table) + } + return table +} + +function normalizeValue(value: unknown): unknown { + if (value === null || value === undefined) return null + if (Array.isArray(value)) return JSON.stringify(value) + if (value instanceof Date) return value + if (Buffer.isBuffer(value)) return value + if (typeof value === 'object') return JSON.stringify(value) + return value +} + +async function bulkInsertRows(pool: sql.ConnectionPool, targetSchema: string, tableName: string, columns: SyncColumn[], rows: Record[]) { + if (rows.length === 0) return + const table = makeBulkTable(targetSchema, tableName, columns) + for (const row of rows) { + table.rows.add(...columns.map((column) => normalizeValue(row[column.column_name]))) + } + await pool.request().bulk(table) +} + +async function replaceTargetFromStaging( + pool: sql.ConnectionPool, + targetSchema: string, + targetTable: string, + stagingTable: string, + columns: SyncColumn[], + recreate: boolean, +) { + const columnList = columns.map((column) => qSql(column.column_name)).join(', ') + const transaction = new sql.Transaction(pool) + await transaction.begin(sql.ISOLATION_LEVEL.SERIALIZABLE) + try { + const request = new sql.Request(transaction) + if (recreate) { + await request.query(` + IF OBJECT_ID(N'${targetSchema}.${targetTable}', N'U') IS NOT NULL + DROP TABLE ${sqlTableName(targetSchema, targetTable)}; + EXEC sp_rename N'${targetSchema}.${stagingTable}', N'${targetTable}', N'OBJECT'; + `) + } else { + await request.query(` + TRUNCATE TABLE ${sqlTableName(targetSchema, targetTable)}; + INSERT INTO ${sqlTableName(targetSchema, targetTable)} (${columnList}) + SELECT ${columnList} + FROM ${sqlTableName(targetSchema, stagingTable)}; + DROP TABLE ${sqlTableName(targetSchema, stagingTable)}; + `) + } + await transaction.commit() + } catch (error) { + await transaction.rollback() + throw error + } +} + +async function getTargetCount(pool: sql.ConnectionPool, targetSchema: string, tableName: string): Promise { + const result = await pool.request().query<{ row_count: number }>( + `SELECT COUNT_BIG(*) AS row_count FROM ${sqlTableName(targetSchema, tableName)}`, + ) + return Number(result.recordset[0]?.row_count ?? 0) +} + +async function syncView(pg: Client, pool: sql.ConnectionPool, options: CliOptions, view: string): Promise { + const runId = crypto.randomUUID() + const targetTable = view + const stagingTable = `__sync_${view}_${runId.replaceAll('-', '').slice(0, 10)}` + const startedAt = new Date() + const columns = await getSourceColumns(pg, options.sourceSchema, view) + const sourceRows = await getSourceCount(pg, options.sourceSchema, view) + + if (options.dryRun) { + return { + view, + targetTable: `${options.targetSchema}.${targetTable}`, + sourceRows, + copiedRows: 0, + targetRows: null, + recreated: false, + } + } + + await ensureSyncLogTable(pool, options.targetSchema) + await startSyncRun(pool, options.targetSchema, { + runId, + sourceSchema: options.sourceSchema, + sourceView: view, + targetTable, + startedAt, + }) + + let copiedRows = 0 + let targetRows: number | null = null + let recreated = false + + try { + await dropTableIfExists(pool, options.targetSchema, stagingTable) + await createTargetTable(pool, options.targetSchema, stagingTable, columns) + + for (let offset = 0; offset < sourceRows; offset += options.batchSize) { + const rows = await fetchSourceBatch(pg, options.sourceSchema, view, columns, options.batchSize, offset) + await bulkInsertRows(pool, options.targetSchema, stagingTable, columns, rows) + copiedRows += rows.length + console.log(`[${view}] copied ${copiedRows}/${sourceRows}`) + } + + const exists = await tableExists(pool, options.targetSchema, targetTable) + const sameShape = exists ? sameTargetShape(columns, await getTargetColumns(pool, options.targetSchema, targetTable)) : false + recreated = !exists || !sameShape + await replaceTargetFromStaging(pool, options.targetSchema, targetTable, stagingTable, columns, recreated) + await createIndexes(pool, options.targetSchema, targetTable, columns) + targetRows = await getTargetCount(pool, options.targetSchema, targetTable) + + await finishSyncRun(pool, options.targetSchema, { + runId, + status: 'SUCCEEDED', + sourceRows, + targetRows, + }) + + return { + view, + targetTable: `${options.targetSchema}.${targetTable}`, + sourceRows, + copiedRows, + targetRows, + recreated, + } + } catch (error) { + await dropTableIfExists(pool, options.targetSchema, stagingTable).catch(() => undefined) + await finishSyncRun(pool, options.targetSchema, { + runId, + status: 'FAILED', + sourceRows, + targetRows, + errorMessage: error instanceof Error ? error.message : String(error), + }).catch(() => undefined) + throw error + } +} + +async function main() { + const options = parseCliOptions(process.argv.slice(2)) + const startedAt = new Date() + console.log(`[${startedAt.toISOString()}] start external readonly JSReport sync`) + console.log( + JSON.stringify( + { + mode: options.apply ? 'apply' : 'dry-run', + sourceSchema: options.sourceSchema, + targetSchema: options.targetSchema, + views: options.views, + batchSize: options.batchSize, + }, + null, + 2, + ), + ) + + const pg = new Client({ connectionString: getPgConnectionString(), connectionTimeoutMillis: 15000, query_timeout: 120000 }) + const pool = await sql.connect(getSqlServerConfig()) + + try { + await pg.connect() + const summaries: SyncSummary[] = [] + for (const view of options.views) { + summaries.push(await syncView(pg, pool, options, view)) + } + + console.log( + JSON.stringify( + { + finishedAt: new Date().toISOString(), + summaries, + }, + null, + 2, + ), + ) + } finally { + await pg.end().catch(() => undefined) + await pool.close() + } +} + +main().catch((error) => { + console.error(`external readonly JSReport sync failed: ${error instanceof Error ? error.stack ?? error.message : String(error)}`) + process.exitCode = 1 +})