| 83 | }; |
| 84 | |
| 85 | async function copyTable(tableName: string, whereCondition: string, joins: JoinCondition[] = []) { |
| 86 | let joinClause = ""; |
| 87 | for (const join of joins) { |
| 88 | const joinType = join.type || "INNER JOIN"; |
| 89 | joinClause += ` ${joinType} "${join.table}" ON ${join.condition}`; |
| 90 | } |
| 91 | // Step 1: Log the number of rows that match the WHERE clause in the remote database |
| 92 | const countQuery = `SELECT COUNT(*) FROM "${tableName}" ${joinClause} WHERE ${whereCondition}`; |
| 93 | const countResult = await remoteClient.query(countQuery); |
| 94 | const rowCount = countResult.rows[0].count as number; |
| 95 | console.log(`Copying ${rowCount} rows from table ${tableName}.`); |
| 96 | |
| 97 | const selectColumns = `"${tableName}".*`; |
| 98 | |
| 99 | const copyFromQuery = `COPY (SELECT ${selectColumns} FROM "${tableName}" ${joinClause} WHERE ${whereCondition}) TO STDOUT`; |
| 100 | const copyToQuery = `COPY "${tableName}" FROM STDIN`; |
| 101 | |
| 102 | const remoteStream = remoteClient.query(copyStreams.to(copyFromQuery)); |
| 103 | const localStream = localClient.query(copyStreams.from(copyToQuery)); |
| 104 | |
| 105 | remoteStream.pipe(localStream); |
| 106 | |
| 107 | await new Promise((resolve, reject) => { |
| 108 | localStream.on("error", reject); |
| 109 | localStream.on("finish", resolve); |
| 110 | }); |
| 111 | |
| 112 | // Step 3: Log when the table has finished copying |
| 113 | console.log(`Finished copying rows for table ${tableName}.`); |
| 114 | } |
| 115 | |
| 116 | await Promise.all([ |
| 117 | copyTable("Project", `id = '${projectId}'`), |