wip(migration): progress towards seeded flat migration

packages/generator/src/latestmigration.ts

@@ -1,9 +1,9 @@
 import { PostgreSqlContainer } from '@testcontainers/postgresql';
 import { spawn } from 'node:child_process';
-import { writeFileSync } from 'node:fs';
+import { mkdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
 import { resolve } from 'node:path';
-import { Pool } from 'pg';
-import { migrate } from '../../server/src/migrations/migrations';
+import { MedplumServerConfig } from '../../server/src/config';
+import { main as serverMain } from '../../server/src/index';
 import { FileBuilder } from './filebuilder';
 
 const SCHEMA_DIR = resolve(__dirname, '../../server/src/migrations/schema');
@@ -22,13 +22,33 @@ async function main(): Promise<void> {
  // Start clean database...
  console.info('Starting Postgres container...');
  const container = await new PostgreSqlContainer().start();
- const pool = new Pool({ connectionString: container.getConnectionUri() });
- const client = await pool.connect();
- console.info('Connected to pool... Running migrations...');
 
- // Run migrations
- await migrate(client, true);
+ // Load existing `medplum.config.json`, change port
+ const baseConfig = JSON.parse(
+ readFileSync(resolve(__dirname, '../../server/medplum.config.json'), { encoding: 'utf-8' })
+ ) as MedplumServerConfig;
 
+ baseConfig.database.port = container.getPort();
+ baseConfig.database.dbname = container.getDatabase();
+ baseConfig.database.username = container.getUsername();
+ baseConfig.database.password = container.getPassword();
+ baseConfig.database.runMigrations = 'full';
+
+ // Write latestmigration config
+ writeFileSync(resolve(__dirname, '../../server/latestmigration.config.json'), JSON.stringify(baseConfig));
+
+ // Startup server, triggering migration
+ console.info('Starting server... Migrations will run...');
+ const { shutdown } = await serverMain('file:latestmigration.config.json');
+
+ // Cleanup config file
+ rmSync(resolve(__dirname, '../../server/latestmigration.config.json'));
+
+ // Shutdown when done
+ console.info('Shutting down Medplum server...');
+ await shutdown();
+
+ console.info('Dumping database via pg_dump...');
  // Dump resulting schema
  const { stdout } = spawn(
  'pg_dump',
@@ -39,8 +59,9 @@ async function main(): Promise<void> {
  container.getPort().toString(),
  '-U',
  container.getUsername(),
- '--schema-only',
  '--no-owner',
+ '--no-acl',
+ '--no-comments',
  container.getDatabase(),
  ],
  {
@@ -63,31 +84,157 @@ async function main(): Promise<void> {
 
  console.info('Database dump successful. Writing dump to file...');
 
- const builder = new FileBuilder();
- buildLatestMigration(builder, dump);
- writeFileSync(`${SCHEMA_DIR}/latest.ts`, builder.toString(), 'utf8');
+ mkdirSync(`${SCHEMA_DIR}/data`, { recursive: true });
+
+ // TODO: Remove this
+ writeFileSync(`${SCHEMA_DIR}/test.sql`, dump, 'utf-8');
+
+ const builder = new FlatMigrationBuilder(dump);
+ const migration = builder.buildMigration();
+ writeFileSync(`${SCHEMA_DIR}/latest.ts`, migration, { encoding: 'utf-8' });
 
  console.info('Migration file successfully created.');
 
- client.release();
- await pool.end();
  await container.stop();
 }
 
-function buildLatestMigration(b: FileBuilder, sqlDump: string): void {
- b.append("import { PoolClient } from 'pg';");
- b.newLine();
- b.append('export async function run(client: PoolClient): Promise<void> {');
- b.indentCount++;
+class FlatMigrationBuilder {
+ readonly builder: FileBuilder;
+ readonly sourceLines: string[];
+ copyDataParser: CopyStatementDataParser | undefined;
+ currentLine = 0;
+ bufferedStatement: string[] | undefined;
+
+ constructor(sqlDump: string) {
+ this.builder = new FileBuilder();
+ this.sourceLines = sqlDump.split('\n').map((line) => line.trim());
+ this.bufferedStatement = undefined;
+ this.copyDataParser = undefined;
+ }
+
+ private processCurrentLine(): boolean {
+ // If we processed the last line already, return false signalling we are done
+ if (this.currentLine === this.sourceLines.length) {
+ return false;
+ }
+
+ // If line is empty, skip it
+ const line = this.sourceLines[this.currentLine];
+ if (line === '' || line.startsWith('--')) {
+ this.currentLine++;
+ return true;
+ }
+
+ // If the copyDataParser exists, that means we need to use it
+ if (this.copyDataParser) {
+ // If we find the copy terminal character sequence, finalize the file for this copy statement
+ if (line.startsWith('\\.')) {
+ // Finalize the data file
+ this.copyDataParser.writeDataFile(`${SCHEMA_DIR}/data`);
+ // Remove copyParser
+ this.copyDataParser = undefined;
+
+ // Go to next line
+ this.currentLine++;
+ return true;
+ }
+
+ // Otherwise if we don't see the terminal character sequence,
+ // Parse the data line
+ this.copyDataParser.parseDataLine(line);
+
+ // Go to next line
+ this.currentLine++;
+ return true;
+ }
+
+ if (line.startsWith('COPY')) {
+ const nextLine = this.peekNextLine();
+ if (!nextLine) {
+ throw new Error('Invalid SQL file: COPY statement not followed by data to copy or `.` like expected');
+ }
+ // Skip COPY statements that don't contain any data
+ if (nextLine === '\\.') {
+ this.currentLine += 2;
+ return true;
+ }
+ // start building copy
+ this.copyDataParser = new CopyStatementDataParser(line);
+ this.currentLine++;
+ return true;
+ }
 
- b.newLine();
- b.appendNoWrap('await client.query(`');
- b.indentCount--;
+ this.bufferPartialStatement(line);
+ this.currentLine++;
+ return true;
+ }
 
- b.appendNoWrap(sqlDump);
+ private bufferPartialStatement(partialStatement: string): void {
+ if (!this.bufferedStatement) {
+ this.bufferedStatement = ["await client.query('"];
+ } else {
+ this.bufferedStatement.push(' ');
+ }
+ this.bufferedStatement.push(partialStatement.replaceAll("'", "\\'"));
+ if (partialStatement.endsWith(';')) {
+ this.endStatement();
+ }
+ }
 
- b.append('`);');
- b.append('}');
+ private endStatement(): void {
+ if (!this.bufferedStatement) {
+ throw new Error('No buffered statement to end');
+ }
+ this.bufferedStatement.push("');");
+ this.builder.appendNoWrap(this.bufferedStatement.join(''));
+ this.bufferedStatement = undefined;
+ }
+
+ private peekNextLine(): string | undefined {
+ if (this.currentLine === this.sourceLines.length - 1) {
+ return undefined;
+ }
+ return this.sourceLines[this.currentLine + 1];
+ }
+
+ buildMigration(): string {
+ const b = this.builder;
+
+ b.append("import { PoolClient } from 'pg';");
+ b.newLine();
+ b.append('export async function run(client: PoolClient): Promise<void> {');
+ b.indentCount++;
+
+ b.newLine();
+
+ let line = false;
+ do {
+ line = this.processCurrentLine();
+ } while (line);
+
+ b.append('}');
+
+ return b.toString();
+ }
+}
+
+class CopyStatementDataParser {
+ private dataFile = '';
+ constructor(readonly copyStatement: string) {}
+
+ parseDataLine(line: string): void {
+ this.dataFile += `${line}\n`;
+ }
+
+ writeDataFile(dirPath: string): void {
+ const filename = this.copyStatement.match(/public\."(.+?)"/)?.[1];
+ if (!filename) {
+ throw new Error(
+ 'Invalid migration file. Failed to match table name with the following pattern: `public."TABLE_NAME"`'
+ );
+ }
+ writeFileSync(`${dirPath}/${filename}.tsv`, this.dataFile, { encoding: 'utf-8' });
+ }
 }
 
 main().catch(console.error);

packages/server/package.json

@@ -64,6 +64,7 @@
  "nodemailer": "6.9.13",
  "otplib": "12.0.1",
  "pg": "8.11.5",
+ "pg-copy-streams": "6.0.6",
  "rate-limit-redis": "4.2.0",
  "rfc6902": "5.1.1",
  "ua-parser-js": "1.0.37",

packages/server/src/config.ts

@@ -98,7 +98,7 @@ export interface MedplumDatabaseConfig {
  password?: string;
  ssl?: MedplumDatabaseSslConfig;
  queryTimeout?: number;
- runMigrations?: boolean;
+ runMigrations?: 'full' | boolean;
 }
 
 export interface MedplumRedisConfig {

packages/server/src/database.ts

@@ -53,7 +53,7 @@ export async function initDatabase(serverConfig: MedplumServerConfig): Promise<v
  try {
  client = await pool.connect();
  await client.query('SELECT pg_advisory_lock($1)', [locks.migration]);
- await migrate(client);
+ await migrate(client, config.runMigrations === 'full');
  } finally {
  if (client) {
  await client.query('SELECT pg_advisory_unlock($1)', [locks.migration]);

packages/server/src/index.ts

@@ -5,7 +5,7 @@ import { initApp, shutdownApp } from './app';
 import { loadConfig } from './config';
 import { globalLogger } from './logger';
 
-export async function main(configName: string): Promise<void> {
+export async function main(configName: string): Promise<{ shutdown: () => Promise<void> }> {
  process.on('unhandledRejection', (err: any) => {
  globalLogger.error('Unhandled promise rejection', err);
  });
@@ -30,8 +30,10 @@ export async function main(configName: string): Promise<void> {
  globalLogger.level = parseLogLevel(config.logLevel);
  }
 
- const app = await initApp(express(), config);
- const server = app.listen(config.port);
+ const server = await initApp(express(), config);
+ await new Promise<void>((resolve) => {
+ server.listen(config.port, resolve);
+ });
  server.keepAliveTimeout = config.keepAliveTimeout ?? 90000;
  globalLogger.info('Server started', { port: config.port });
  gracefulShutdown(server, {
@@ -48,6 +50,8 @@ export async function main(configName: string): Promise<void> {
  globalLogger.info('Shutdown complete');
  },
  });
+
+ return { shutdown: shutdownApp };
 }
 
 if (require.main === module) {