fix(external): stream jsreport sync from stable cursor

furniture-3d-system/scripts/sync-external-readonly-to-jsreport.ts

@@ -308,35 +308,52 @@ async function getSourceCount(pg: Client, sourceSchema: string, view: string): P
   return Number(result.rows[0]?.row_count ?? 0)
 }
 
-function chooseOrderColumn(columns: SyncColumn[]): string {
+function chooseOrderColumns(columns: SyncColumn[]): string[] {
+  const columnNames = new Set(columns.map((column) => column.column_name))
+  const orderColumns: 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
+    if (columnNames.has(candidate)) orderColumns.push(candidate)
   }
-  return columns[0]?.column_name ?? ''
+  if (orderColumns.length === 0 && columns[0]) orderColumns.push(columns[0].column_name)
+  return orderColumns
 }
 
-async function fetchSourceBatch(
+async function declareSourceCursor(
   pg: Client,
+  cursorName: string,
   sourceSchema: string,
   view: string,
   columns: SyncColumn[],
-  limit: number,
-  offset: number,
-): Promise<Record<string, unknown>[]> {
+): Promise<void> {
+  assertIdentifier(cursorName, 'cursor name')
   const columnList = columns.map((column) => qPg(column.column_name)).join(', ')
-  const orderColumn = chooseOrderColumn(columns)
-  const result = await pg.query<Record<string, unknown>>(
+  const orderColumns = chooseOrderColumns(columns)
+  const orderClause = orderColumns.map((columnName) => `${qPg(columnName)} NULLS LAST`).join(', ')
+  await pg.query(
     `
+    DECLARE ${qPg(cursorName)} NO SCROLL CURSOR FOR
     SELECT ${columnList}
     FROM ${qPg(sourceSchema)}.${qPg(view)}
-    ORDER BY ${qPg(orderColumn)} NULLS LAST
-    LIMIT $1 OFFSET $2
+    ORDER BY ${orderClause}
     `,
-    [limit, offset],
   )
+}
+
+async function fetchSourceCursorBatch(
+  pg: Client,
+  cursorName: string,
+  limit: number,
+): Promise<Record<string, unknown>[]> {
+  assertIdentifier(cursorName, 'cursor name')
+  const result = await pg.query<Record<string, unknown>>(`FETCH FORWARD ${limit} FROM ${qPg(cursorName)}`)
   return result.rows
 }
 
+async function closeSourceCursor(pg: Client, cursorName: string): Promise<void> {
+  assertIdentifier(cursorName, 'cursor name')
+  await pg.query(`CLOSE ${qPg(cursorName)}`)
+}
+
 async function ensureSyncLogTable(pool: sql.ConnectionPool, targetSchema: string, targetTablePrefix: string) {
   const syncRunsTable = getSyncRunsTableName(targetTablePrefix)
   await pool.request().query(`
@@ -609,19 +626,32 @@ async function syncView(pg: Client, pool: sql.ConnectionPool, options: CliOption
   const targetTable = toTargetTableName(view, options.targetTablePrefix)
   const syncRunsTable = getSyncRunsTableName(options.targetTablePrefix)
   const stagingTable = `__sync_${view}_${runId.replaceAll('-', '').slice(0, 10)}`
+  const sourceCursor = `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)
+  await pg.query('BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY')
+  let pgTransactionOpen = true
+  let sourceCursorOpen = false
+  let sourceRows = 0
 
-  if (options.dryRun) {
-    return {
-      view,
-      targetTable: `${options.targetSchema}.${targetTable}`,
-      sourceRows,
-      copiedRows: 0,
-      targetRows: null,
-      recreated: false,
+  try {
+    sourceRows = await getSourceCount(pg, options.sourceSchema, view)
+
+    if (options.dryRun) {
+      await pg.query('COMMIT')
+      pgTransactionOpen = false
+      return {
+        view,
+        targetTable: `${options.targetSchema}.${targetTable}`,
+        sourceRows,
+        copiedRows: 0,
+        targetRows: null,
+        recreated: false,
+      }
     }
+  } catch (error) {
+    if (pgTransactionOpen) await pg.query('ROLLBACK').catch(() => undefined)
+    throw error
   }
 
   await ensureSyncLogTable(pool, options.targetSchema, options.targetTablePrefix)
@@ -640,13 +670,20 @@ async function syncView(pg: Client, pool: sql.ConnectionPool, options: CliOption
   try {
     await dropTableIfExists(pool, options.targetSchema, stagingTable)
     await createTargetTable(pool, options.targetSchema, stagingTable, columns)
+    await declareSourceCursor(pg, sourceCursor, options.sourceSchema, view, columns)
+    sourceCursorOpen = true
 
-    for (let offset = 0; offset < sourceRows; offset += options.batchSize) {
-      const rows = await fetchSourceBatch(pg, options.sourceSchema, view, columns, options.batchSize, offset)
+    while (true) {
+      const rows = await fetchSourceCursorBatch(pg, sourceCursor, options.batchSize)
+      if (rows.length === 0) break
       await bulkInsertRows(pool, options.targetSchema, stagingTable, columns, rows)
       copiedRows += rows.length
       console.log(`[${view}] copied ${copiedRows}/${sourceRows}`)
     }
+    await closeSourceCursor(pg, sourceCursor)
+    sourceCursorOpen = false
+    await pg.query('COMMIT')
+    pgTransactionOpen = false
 
     const exists = await tableExists(pool, options.targetSchema, targetTable)
     const sameShape = exists ? sameTargetShape(columns, await getTargetColumns(pool, options.targetSchema, targetTable)) : false
@@ -671,6 +708,8 @@ async function syncView(pg: Client, pool: sql.ConnectionPool, options: CliOption
       recreated,
     }
   } catch (error) {
+    if (sourceCursorOpen) await closeSourceCursor(pg, sourceCursor).catch(() => undefined)
+    if (pgTransactionOpen) await pg.query('ROLLBACK').catch(() => undefined)
     await dropTableIfExists(pool, options.targetSchema, stagingTable).catch(() => undefined)
     await finishSyncRun(pool, options.targetSchema, syncRunsTable, {
       runId,