feat(external): sync readonly views to jsreport
This commit is contained in:
WuYongJie
parent
611b7842c0
commit
0300858fdf
@ -149,7 +149,57 @@ psql "postgresql://external_core_reader:<REDACTED_SECRET>@<DB_HOST>:5432/<DB_NAM
|
||||
psql "$DATABASE_URL" -f scripts/sql/check-external-readonly-contract.sql
|
||||
```
|
||||
|
||||
## 7. 规范审查
|
||||
## 7. 同步到 JSReportReplica.app 中间库
|
||||
|
||||
如果外部系统不直接连本项目 PostgreSQL，而是统一从 `JSReportReplica` 中间库读取，本项目维护一份 `external` 核心视图到 `app` schema 的同名表同步。
|
||||
|
||||
默认同步表：
|
||||
|
||||
- `app.product_catalog_wide_v1`
|
||||
- `app.intake_product_limits_v1`
|
||||
- `app.warehouse_locations_v1`
|
||||
- `app.sku_warehouse_stock_v1`
|
||||
- `app.packaged_items_v1`
|
||||
- `app.sku_quality_inspections_v1`
|
||||
- `app.external_readonly_sync_runs`：同步运行日志。
|
||||
|
||||
主宽表仍然是：
|
||||
|
||||
```sql
|
||||
SELECT *
|
||||
FROM app.product_catalog_wide_v1;
|
||||
```
|
||||
|
||||
同步入口：
|
||||
|
||||
```bash
|
||||
npm run sync:external-jsreport
|
||||
npm run sync:external-jsreport:apply
|
||||
```
|
||||
|
||||
第一个命令只演练并输出源视图行数；第二个命令才会写入 `JSReportReplica.app`。脚本会先写临时表，再替换正式表，避免同步中途留下半截数据。
|
||||
|
||||
生产定时器安装入口：
|
||||
|
||||
```bash
|
||||
bash scripts/install-external-jsreport-sync-timer.sh
|
||||
```
|
||||
|
||||
安装脚本只会在生产服务器本地生成 `/opt/furniture-3d-system-prod/shared/external-jsreport-sync.env` 模板。真实连接信息只允许写在这个本地 env 文件里，并保持 `600` 权限，不要写入 Git、文档或群公告。
|
||||
|
||||
必要环境变量：
|
||||
|
||||
- `DATABASE_URL` 或 `EXTERNAL_JSREPORT_SOURCE_DATABASE_URL`：源 PostgreSQL 连接。
|
||||
- `JSREPORT_REPLICA_HOST`
|
||||
- `JSREPORT_REPLICA_PORT`
|
||||
- `JSREPORT_REPLICA_DATABASE`
|
||||
- `JSREPORT_REPLICA_USERNAME`
|
||||
- `JSREPORT_REPLICA_PASSWORD`
|
||||
- `JSREPORT_REPLICA_SCHEMA`
|
||||
|
||||
默认定时频率是每 30 分钟同步一次，可通过 `EXTERNAL_JSREPORT_SYNC_ON_CALENDAR` 调整 systemd timer 的 `OnCalendar` 表达式。
|
||||
|
||||
## 8. 规范审查
|
||||
|
||||
结论：`passed`
|
||||
|
||||
|
||||
@ -56,6 +56,8 @@
|
||||
"report:oa-board-activities": "tsx scripts/report-oa-board-activities.ts",
|
||||
"sync:employee-directory": "tsx scripts/sync-employee-directory.ts",
|
||||
"sync:employee-directory:apply": "tsx scripts/sync-employee-directory.ts --apply",
|
||||
"sync:external-jsreport": "tsx scripts/sync-external-readonly-to-jsreport.ts",
|
||||
"sync:external-jsreport:apply": "tsx scripts/sync-external-readonly-to-jsreport.ts --apply",
|
||||
"db:verify-master-data:truth": "node scripts/verify-master-data-truth-sources.js",
|
||||
"db:verify-master-data:code": "node scripts/verify-master-data-code-boundaries.js",
|
||||
"db:verify-master-data:all": "npm run db:verify-master-data:truth && npm run db:verify-master-data:code && npm run db:verify-master-data:migration-status && npm run db:verify-master-data",
|
||||
|
||||
@ -0,0 +1,86 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
BASE="${FURNITURE_PROD_BASE:-/opt/furniture-3d-system-prod}"
|
||||
SERVICE_NAME="${EXTERNAL_JSREPORT_SYNC_SERVICE_NAME:-furniture-external-jsreport-sync}"
|
||||
APP_ROOT="$BASE/current/furniture-3d-system"
|
||||
SHARED="$BASE/shared"
|
||||
APP_ENV="$SHARED/.env"
|
||||
SYNC_ENV="$SHARED/external-jsreport-sync.env"
|
||||
SERVICE_FILE="/etc/systemd/system/$SERVICE_NAME.service"
|
||||
TIMER_FILE="/etc/systemd/system/$SERVICE_NAME.timer"
|
||||
SYNC_CALENDAR="${EXTERNAL_JSREPORT_SYNC_ON_CALENDAR:-*:0/30}"
|
||||
|
||||
if [ "$(id -u)" -ne 0 ]; then
|
||||
echo "This installer must run as root." >&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" <<EOF
|
||||
[Unit]
|
||||
Description=Furniture external readonly views to JSReportReplica sync
|
||||
After=network-online.target
|
||||
Wants=network-online.target
|
||||
|
||||
[Service]
|
||||
Type=oneshot
|
||||
WorkingDirectory=$APP_ROOT
|
||||
Environment=NODE_ENV=production
|
||||
EnvironmentFile=$APP_ENV
|
||||
EnvironmentFile=$SYNC_ENV
|
||||
ExecStart=/bin/bash -lc 'cd "$APP_ROOT" && ./node_modules/.bin/tsx scripts/sync-external-readonly-to-jsreport.ts --apply'
|
||||
StandardOutput=journal
|
||||
StandardError=journal
|
||||
EOF
|
||||
|
||||
cat >"$TIMER_FILE" <<EOF
|
||||
[Unit]
|
||||
Description=Run furniture external readonly JSReport sync
|
||||
|
||||
[Timer]
|
||||
OnCalendar=$SYNC_CALENDAR
|
||||
Persistent=true
|
||||
RandomizedDelaySec=60
|
||||
Unit=$SERVICE_NAME.service
|
||||
|
||||
[Install]
|
||||
WantedBy=timers.target
|
||||
EOF
|
||||
|
||||
systemctl daemon-reload
|
||||
systemctl enable --now "$SERVICE_NAME.timer"
|
||||
systemctl list-timers --all "$SERVICE_NAME.timer" --no-pager
|
||||
@ -0,0 +1,712 @@
|
||||
import crypto from 'node:crypto'
|
||||
import { Client } from 'pg'
|
||||
import sql from 'mssql'
|
||||
|
||||
type CliOptions = {
|
||||
apply: boolean
|
||||
dryRun: boolean
|
||||
batchSize: number
|
||||
sourceSchema: string
|
||||
targetSchema: string
|
||||
views: string[]
|
||||
}
|
||||
|
||||
type SourceColumn = {
|
||||
column_name: string
|
||||
ordinal_position: number
|
||||
data_type: string
|
||||
udt_name: string
|
||||
character_maximum_length: number | null
|
||||
numeric_precision: number | null
|
||||
numeric_scale: number | null
|
||||
}
|
||||
|
||||
type SyncColumn = SourceColumn & {
|
||||
sqlDeclaration: string
|
||||
addBulkColumn: (table: sql.Table) => 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<SyncColumn[]> {
|
||||
const result = await pg.query<SourceColumn>(
|
||||
`
|
||||
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<number> {
|
||||
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<Record<string, unknown>[]> {
|
||||
const columnList = columns.map((column) => qPg(column.column_name)).join(', ')
|
||||
const orderColumn = chooseOrderColumn(columns)
|
||||
const result = await pg.query<Record<string, unknown>>(
|
||||
`
|
||||
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<boolean> {
|
||||
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<ReturnType<typeof getTargetColumns>>): 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<string, unknown>[]) {
|
||||
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<number> {
|
||||
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<SyncSummary> {
|
||||
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
|
||||
})
|
||||
Loading…
x
Reference in New Issue
Block a user