import { csv , now } from '@mds-core/mds-utils' ; import logger from '@mds-core/mds-logger' ; import schema , { COLUMN_NAME , TABLE_NAME } from './schema' ; import { SqlExecuter , MDSPostgresClient , cols_sql , SqlExecuterFunction } from './sql-utils' ; const MIGRATIONS = [ 'createMigrationsTable' , 'alterGeographiesColumns' , 'alterAuditEventsColumns' , 'alterPreviousGeographiesColumn' , 'dropDeprecatedProviderTables' , 'dropReadOnlyGeographyColumn' , 'dropAuditEventsColumns' , 'alterReportsTripsMigration' ] as const ; type MIGRATION = typeof MIGRATIONS [ number ] ; async function dropTables ( client ) { const exec = SqlExecuter ( client ) ; const drop = csv ( schema . DEPRECATED_PROVIDER_TABLES . concat ( schema . TABLES ) ) ; await exec ( `DROP TABLE IF EXISTS ${ drop };` ) ; logger . info ( `postgres drop table succeeded: ${ drop }` ) ; } async function addForeignKey ( client , from , to , column ) { const exec = SqlExecuter ( client ) ; const foreignKeyName = `fk_${ to }_${ column }` ; const sql = `DO $$
    BEGIN
      IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = '${ foreignKeyName }') THEN
        ALTER TABLE ${ from }
        ADD CONSTRAINT ${ foreignKeyName }
        FOREIGN KEY (${ column }) REFERENCES ${ to } (${ column });
      END IF;
    END;
    $$` ; await exec ( sql ) ; } async function addIndex ( client , table , column , options = { unique : false } ) { const exec = SqlExecuter ( client ) ; const indexName = `idx_${ column }_${ table }` ; const { rows : { length : hasColumn } } = await exec ( `SELECT column_name FROM information_schema.columns WHERE table_name='${ table }' AND column_name='${ column }' AND table_catalog=CURRENT_CATALOG AND table_schema=CURRENT_SCHEMA` ) ; if ( hasColumn ) { const { rows : { length : hasIndex } } = await exec ( `SELECT tablename FROM pg_indexes WHERE tablename='${ table }' AND indexname='${ indexName }'` ) ; if ( ! hasIndex ) { await exec ( `CREATE${ options . unique ? ' UNIQUE ' : ' ' }INDEX ${ indexName } ON ${ table }(${ column })` ) ; } } } /**
 * create tables from a list of table names
 */ async function createTables ( client ) { const exec = SqlExecuter ( client ) ; const existing = await exec ( 'SELECT table_name FROM information_schema.tables WHERE table_catalog = CURRENT_CATALOG AND table_schema = CURRENT_SCHEMA' ) ; const missing = schema . TABLES . filter ( ( table ) => ! existing . rows . find ( ( row ) => row . table_name === table ) ) ; if ( missing . length > 0 ) { logger . warn ( 'existing' , JSON . stringify ( existing . rows ) , 'missing' , JSON . stringify ( missing ) ) ; const create = missing . map ( table => `CREATE TABLE ${ table } (${ schema . TABLE_COLUMNS [ table ] . map ( column => `${ column } ${ schema . COLUMN_TYPE [ column ] }` ) . join ( ', ' ) }, PRIMARY KEY (${ csv ( schema . TABLE_KEY [ table ] ) }));` ) . join ( '\n' ) ; logger . warn ( create ) ; await exec ( create ) ; logger . info ( 'postgres create table suceeded' ) ; await Promise . all ( missing . map ( table => addIndex ( client , table , schema . COLUMN . recorded ) ) ) ; await Promise . all ( missing . map ( table => addIndex ( client , table , schema . COLUMN . id , { unique : true } ) ) ) ; await addForeignKey ( client , schema . TABLE . policy_metadata , schema . TABLE . policies , schema . COLUMN . policy_id ) ; await addForeignKey ( client , schema . TABLE . geography_metadata , schema . TABLE . geographies , schema . COLUMN . geography_id ) ; if ( create . includes ( schema . TABLE . migrations ) ) { await exec ( `INSERT INTO ${ schema . TABLE . migrations } (${ cols_sql ( schema . TABLE_COLUMNS . migrations ) }) VALUES ${ MIGRATIONS . map ( migration => `('${ migration }', ${ now ( ) })` ) . join ( ', ' ) }` ) ; } } } async function doMigration ( exec , migration , migrate ) { const { PG_MIGRATIONS , PG_DEBUG } = process . env ; const migrations = PG_MIGRATIONS ? PG_MIGRATIONS . split ( ',' ) : [ ] ; if ( migrations . includes ( 'true' ) || migrations . includes ( migration ) ) { const { rowCount } = await exec ( `SELECT * FROM ${ schema . TABLE . migrations } WHERE ${ schema . COLUMN . migration } = '${ migration }'` ) ; if ( rowCount === 0 ) { try { await exec ( `INSERT INTO ${ schema . TABLE . migrations } (${ cols_sql ( schema . TABLE_COLUMNS . migrations ) }) VALUES ('${ migration }', ${ now ( ) })` ) ; process . env . PG_DEBUG = 'true' ; try { logger . warn ( 'Running migration' , migration ) ; await migrate ( exec ) ; logger . warn ( 'Migration' , migration , 'succeeded' ) ; } catch ( err ) { logger . error ( 'Migration' , migration , 'failed' , err ) ; } process . env . PG_DEBUG = PG_DEBUG ; } catch { } } } } async function alterGeographiesColumnsMigration ( exec ) { await exec ( `ALTER TABLE ${ schema . TABLE . geographies } RENAME COLUMN previous_geography_ids TO previous_geographies` ) ; await exec ( `ALTER TABLE ${ schema . TABLE . geographies } ADD COLUMN ${ schema . COLUMN . publish_date } ${ schema . COLUMN_TYPE . publish_date }` ) ; await exec ( `ALTER TABLE ${ schema . TABLE . geographies } ADD COLUMN ${ schema . COLUMN . effective_date } ${ schema . COLUMN_TYPE . effective_date }` ) ; await exec ( `ALTER TABLE ${ schema . TABLE . geographies } ADD COLUMN ${ schema . COLUMN . description } ${ schema . COLUMN_TYPE . description }` ) ; } async function alterAuditEventsColumnsMigration ( exec ) { await exec ( `ALTER TABLE ${ schema . TABLE . audit_events } ADD COLUMN provider_event_id bigint` ) ; await exec ( `ALTER TABLE ${ schema . TABLE . audit_events } ADD COLUMN provider_event_type varchar(31)` ) ; await exec ( `ALTER TABLE ${ schema . TABLE . audit_events } ADD COLUMN provider_event_type_reason varchar(31)` ) ; } async function alterPreviousGeographiesColumnMigration ( exec ) { await exec ( `ALTER TABLE ${ schema . TABLE . geographies } RENAME COLUMN previous_geographies TO ${ schema . COLUMN . prev_geographies }` ) ; } async function dropDeprecatedProviderTablesMigration ( exec ) { await exec ( `DROP TABLE IF EXISTS ${ csv ( schema . DEPRECATED_PROVIDER_TABLES ) };` ) ; } async function dropReadOnlyGeographyColumnMigration ( exec ) { await exec ( `ALTER TABLE ${ schema . TABLE . geographies } DROP COLUMN read_only` ) ; } async function dropAuditEventsColumnsMigration ( exec ) { await exec ( `ALTER TABLE ${ schema . TABLE . audit_events } DROP COLUMN provider_event_id` ) ; await exec ( `ALTER TABLE ${ schema . TABLE . audit_events } DROP COLUMN provider_event_type` ) ; await exec ( `ALTER TABLE ${ schema . TABLE . audit_events } DROP COLUMN provider_event_type_reason` ) ; } async function doMigrations ( client ) { const exec = SqlExecuter ( client ) ; await doMigration ( exec , 'alterGeographiesColumns' , alterGeographiesColumnsMigration ) ; await doMigration ( exec , 'alterAuditEventsColumns' , alterAuditEventsColumnsMigration ) ; await doMigration ( exec , 'alterPreviousGeographiesColumn' , alterPreviousGeographiesColumnMigration ) ; await doMigration ( exec , 'dropDeprecatedProviderTables' , dropDeprecatedProviderTablesMigration ) ; await doMigration ( exec , 'dropReadOnlyGeographyColumn' , dropReadOnlyGeographyColumnMigration ) ; await doMigration ( exec , 'dropAuditEventsColumns' , dropAuditEventsColumnsMigration ) ; } async function updateSchema ( client ) { await createTables ( client ) ; await doMigrations ( client ) ; } export { updateSchema , dropTables , createTables } ;