feat: migrate default postgres schema

- Migrate schema from public to stacks-blockchain-api. Permission to start migration
requires affirmative consent from user

Author: rmottley-hiro

.env

@@ -3,8 +3,14 @@ PG_PORT=5490
 PG_USER=postgres
 PG_PASSWORD=postgres
 PG_DATABASE=stacks_blockchain_api
-PG_SCHEMA=public
+PG_SCHEMA=stacks_blockchain_api
+PG_OLD_SCHEMA=public
 PG_SSL=false
+
+# Enable to migrate to new schema on startup, must be set
+# Set to false after migration to avoid the check on every startup
+# PG_MIGRATE_SCHEMA_ON_STARTUP=true
+
 # Idle connection timeout in seconds, defaults to 30
 # PG_IDLE_TIMEOUT=30
 # Max connection lifetime in seconds, defaults to 60

src/datastore/pg-write-store.ts

@@ -92,11 +92,13 @@ import {
   connectPostgres,
   isProdEnv,
   isTestEnv,
+  parseBoolean,
   runMigrations,
 } from '@hirosystems/api-toolkit';
 import { PgServer, getConnectionArgs, getConnectionConfig } from './connection';
 import { BigNumber } from 'bignumber.js';
 import { RedisNotifier } from './redis-notifier';
+import { log } from 'console';
 
 const MIGRATIONS_TABLE = 'pgmigrations';
 const INSERT_BATCH_SIZE = 500;
@@ -183,6 +185,11 @@ export class PgWriteStore extends PgStore {
       connectionConfig: getConnectionConfig(PgServer.primary),
     });
     if (!skipMigrations) {
+      // Before running new migrations, check if we need to migrate default schema
+      // PG_OLD_SCHEMA must be set to the current schema name (usually 'public')
+      if (parseBoolean(process.env['PG_MIGRATE_SCHEMA_ON_STARTUP'])) {
+        await this.migrateSchema(sql);
+      }
       await runMigrations(MIGRATIONS_DIR, 'up', getConnectionArgs(PgServer.primary), {
         logger: {
           debug: _ => {},
@@ -203,6 +210,22 @@ export class PgWriteStore extends PgStore {
     return store;
   }
 
+  static async migrateSchema(sql: PgSqlClient): Promise<void> {
+    const oldSchema = process.env['PG_OLD_SCHEMA'] || 'public';
+    const newSchema = process.env['PG_SCHEMA'] || 'stacks_blockchain_api';
+
+    // Fetch schemas to see if new schema already exists; skip migration if it exists
+    const schemas = await sql<{ schema_name: string }[]>`
+      SELECT schema_name FROM information_schema.schemata
+    `;
+    if (schemas.find(s => s.schema_name === newSchema)) {
+      logger.warn(`Schema ${newSchema} already exists, skipping migration.`);
+      logger.warn(`Set PG_MIGRATE_SCHEMA_ON_STARTUP=false to disable this check on startup.`);
+      return;
+    }
+    await sql`ALTER SCHEMA ${sql(oldSchema)} RENAME TO ${sql(newSchema)};`;
+  }
+
   async storeRawEventRequest(eventPath: string, payload: any): Promise<void> {
     if (eventPath === '/new_block' && typeof payload === 'object') {
       for (const tx of payload.transactions) {
@@ -3209,15 +3232,15 @@ export class PgWriteStore extends PgStore {
           RETURNING tx_id, sender_address, nonce, sponsor_address, fee_rate, sponsored, canonical
         ),
         affected_addresses AS (
-            SELECT 
+            SELECT
               sender_address AS address,
               fee_rate AS fee_change,
               canonical,
               sponsored
             FROM updated_txs
             WHERE sponsored = false
           UNION ALL
-            SELECT 
+            SELECT
               sponsor_address AS address,
               fee_rate AS fee_change,
               canonical,
@@ -3277,12 +3300,12 @@ export class PgWriteStore extends PgStore {
           RETURNING recipient, coinbase_amount, tx_fees_anchored, tx_fees_streamed_confirmed, tx_fees_streamed_produced, canonical
         ),
         reward_changes AS (
-          SELECT 
+          SELECT
             recipient AS address,
-            SUM(CASE WHEN canonical THEN 
-                (coinbase_amount + tx_fees_anchored + tx_fees_streamed_confirmed + tx_fees_streamed_produced) 
-              ELSE 
-                -(coinbase_amount + tx_fees_anchored + tx_fees_streamed_confirmed + tx_fees_streamed_produced) 
+            SUM(CASE WHEN canonical THEN
+                (coinbase_amount + tx_fees_anchored + tx_fees_streamed_confirmed + tx_fees_streamed_produced)
+              ELSE
+                -(coinbase_amount + tx_fees_anchored + tx_fees_streamed_confirmed + tx_fees_streamed_produced)
               END) AS balance_change
           FROM updated_rewards
           GROUP BY recipient
@@ -3296,7 +3319,7 @@ export class PgWriteStore extends PgStore {
           SET balance = ft_balances.balance + EXCLUDED.balance
           RETURNING ft_balances.address
         )
-        SELECT 
+        SELECT
           (SELECT COUNT(*)::int FROM updated_rewards) AS updated_rewards_count
       `;
       const updateCount = minerRewardResults[0]?.updated_rewards_count ?? 0;
@@ -3327,18 +3350,18 @@ export class PgWriteStore extends PgStore {
           RETURNING sender, recipient, amount, asset_event_type_id, canonical
         ),
         event_changes AS (
-          SELECT 
+          SELECT
             address,
             SUM(balance_change) AS balance_change
           FROM (
-              SELECT 
+              SELECT
                 sender AS address,
                 SUM(CASE WHEN canonical THEN -amount ELSE amount END) AS balance_change
               FROM updated_events
               WHERE asset_event_type_id IN (1, 3) -- Transfers and Burns affect the sender's balance
               GROUP BY sender
             UNION ALL
-              SELECT 
+              SELECT
                 recipient AS address,
                 SUM(CASE WHEN canonical THEN amount ELSE -amount END) AS balance_change
               FROM updated_events
@@ -3356,7 +3379,7 @@ export class PgWriteStore extends PgStore {
           SET balance = ft_balances.balance + EXCLUDED.balance
           RETURNING ft_balances.address
         )
-        SELECT 
+        SELECT
           (SELECT COUNT(*)::int FROM updated_events) AS updated_events_count
       `;
       const updateCount = stxResults[0]?.updated_events_count ?? 0;
@@ -3400,7 +3423,7 @@ export class PgWriteStore extends PgStore {
           SET balance = ft_balances.balance + EXCLUDED.balance
           RETURNING ft_balances.address
         )
-        SELECT 
+        SELECT
           (SELECT COUNT(*)::int FROM updated_events) AS updated_events_count
       `;
       const updateCount = ftResult[0]?.updated_events_count ?? 0;

src/index.ts

@@ -115,6 +115,11 @@ async function init(): Promise<void> {
         '`/extended/` endpoint. Please execute `npm run build` to regenerate it.'
     );
   }
+  if (!process.env['PG_MIGRATE_SCHEMA_ON_STARTUP']) {
+    throw new Error(
+      'Env variable PG_MIGRATE_SCHEMA_ON_STARTUP must be set to affirmatively enable or disable schema migration on startup'
+    );
+  }
   promClient.collectDefaultMetrics();
   chainIdConfigurationCheck();
   const apiMode = getApiMode();

tests/event-replay/import-export.test.ts

@@ -28,6 +28,7 @@ describe('import/export tests', () => {
 
   test('event import and export cycle - remote', async () => {
     // Import from mocknet TSV
+    await migrate('up', true);
     await importEventsFromTsv('tests/event-replay/tsv/mocknet.tsv', 'archival', true, true);
     const chainTip = await db.getChainTip(db.sql);
     expect(chainTip.block_height).toBe(28);
@@ -61,6 +62,7 @@ describe('import/export tests', () => {
 
   test('event import and export cycle - local', async () => {
     // Import from mocknet TSV
+    await migrate('up', true);
     await importEventsFromTsv('tests/event-replay/tsv/mocknet.tsv', 'archival', true, true);
     const chainTip = await db.getChainTip(db.sql);
     expect(chainTip.block_height).toBe(28);

tests/utils/test-helpers.ts

@@ -51,7 +51,7 @@ import { getConnectionArgs } from '../../src/datastore/connection';
 import { AddressStxBalance } from '../../src/api/schemas/entities/addresses';
 import { ServerStatusResponse } from '../../src/api/schemas/responses/responses';
 
-export async function migrate(direction: 'up' | 'down') {
+export async function migrate(direction: 'up' | 'down', skipMigrations: boolean = false) {
   const connArgs = getConnectionArgs();
   if (typeof connArgs !== 'string' && connArgs.schema) {
     const sql = await connectPostgres({
@@ -61,7 +61,7 @@ export async function migrate(direction: 'up' | 'down') {
     await sql`CREATE SCHEMA IF NOT EXISTS ${sql(connArgs.schema)}`;
     await sql.end();
   }
-  await runMigrations(MIGRATIONS_DIR, direction, connArgs);
+  if (!skipMigrations) await runMigrations(MIGRATIONS_DIR, direction, connArgs);
 }
 
 export interface TestEnvContext {