Миграции при выкате: одноразовая джоба и обратная совместимость

import pg from 'pg';

// адрес базы — первой строкой и без пароля: видно, куда ушли запросы
const url = new URL(process.env.DATABASE_URL);
url.password = '';
console.log('база:', url.href);

const подключить = async () => {
  const client = new pg.Client({ connectionString: process.env.DATABASE_URL });
  await client.connect();
  return client;
};

const ИМЯ = '0001_notes';
const ШАГ = 'CREATE TABLE notes (id serial PRIMARY KEY, body text NOT NULL)';

/*
 * Раннер главы 9: прочитать журнал, применить непомеченное, поставить отметку
 * в одной транзакции с шагом. Разница одна — таких раннеров здесь два, как их
 * получается два, когда миграцию применяет старт приложения.
 */
async function выкатить(имя, client, { замок, старт, перед } = {}) {
  if (старт) await старт();
  await client.query('BEGIN');
  // общий замок базы: число — просто ключ, важно, что он один у всех копий
  if (замок) await client.query('SELECT pg_advisory_xact_lock(4242)');
  const { rowCount } = await client.query('SELECT 1 FROM __migrations WHERE name = $1', [ИМЯ]);
  console.log(`  ${имя}: отметок про ${ИМЯ} в журнале — ${rowCount}`);
  if (перед) await перед();
  if (rowCount) {
    await client.query('ROLLBACK');
    console.log(`  ${имя}: применять нечего`);
    return;
  }
  try {
    await client.query(ШАГ);
    await client.query('INSERT INTO __migrations (name) VALUES ($1)', [ИМЯ]);
    await client.query('COMMIT');
    console.log(`  ${имя}: применил ${ИМЯ}`);
  } catch (err) {
    await client.query('ROLLBACK');
    console.log(`  ${имя}: ${err.code} ${err.message}`);
  }
}

/** Веха: одна сторона её отмечает, другая на ней ждёт. */
const веха = () => {
  let отметить;
  const пройдена = new Promise((готово) => (отметить = готово));
  return { пройдена, отметить };
};

const служебный = await подключить();
const A = await подключить();
const B = await подключить();

// журнал раннер заводит сам при первом выкате (глава 9); спор идёт не о нём
await служебный.query('CREATE TABLE __migrations (name text PRIMARY KEY, at timestamptz NOT NULL DEFAULT now())');

const состояние = async () => {
  const { rows: отметки } = await служебный.query('SELECT name FROM __migrations ORDER BY name');
  const { rows: таблицы } = await служебный.query(
    "SELECT table_name AS имя FROM information_schema.tables WHERE table_schema = 'public' ORDER BY 1",
  );
  console.log('  журнал:', отметки.map((строка) => строка.name).join(', ') || 'пусто');
  console.log('  таблицы:', таблицы.map((строка) => строка.имя).join(', '));
};

/*
 * Вехами задан порядок: в проде его решает случай, а записанный вывод обязан
 * совпадать при каждом прогоне. Обе копии успевают прочитать журнал прежде,
 * чем первая применит шаг, — и только это здесь от случая и требуется.
 */
console.log('--- два раннера, у каждого своя транзакция');
const прочитанA = веха();
const дошлаA = веха();
const первая = выкатить('A', A, { перед: прочитанA.отметить });
const вторая = выкатить('B', B, { старт: () => прочитанA.пройдена, перед: () => дошлаA.пройдена });
await первая;
дошлаA.отметить();
await вторая;
await состояние();

// вернём базу в то состояние, в котором её застали оба раннера
await служебный.query('DROP TABLE notes');
await служебный.query('DELETE FROM __migrations');

console.log('--- те же два раннера под общим замком базы');
const замокВзят = веха();
await Promise.all([
  выкатить('A', A, { замок: true, перед: замокВзят.отметить }),
  выкатить('B', B, { замок: true, старт: () => замокВзят.пройдена }),
]);
await состояние();

await служебный.end();
await A.end();
await B.end();

база: postgres://postgres@127.0.0.1:<порт-базы>/app
--- два раннера, у каждого своя транзакция
  A: отметок про 0001_notes в журнале — 0
  B: отметок про 0001_notes в журнале — 0
  A: применил 0001_notes
  B: 42P07 relation "notes" already exists
  журнал: 0001_notes
  таблицы: __migrations, notes
--- те же два раннера под общим замком базы
  A: отметок про 0001_notes в журнале — 0
  A: применил 0001_notes
  B: отметок про 0001_notes в журнале — 1
  B: применять нечего
  журнал: 0001_notes
  таблицы: __migrations, notes

import pg from 'pg';

// адрес базы — первой строкой и без пароля: видно, куда ушли запросы
const url = new URL(process.env.DATABASE_URL);
url.password = '';
console.log('база:', url.href);

const client = new pg.Client({ connectionString: process.env.DATABASE_URL });
await client.connect();

/*
 * Три версии одного обработчика заметок. Старая знает колонку text; переходная
 * пишет в обе и читает ту, в которой что-то есть; новая знает только body.
 * В момент выката на одной схеме работают две версии из трёх.
 */
const версии = {
  'старый код': {
    читает: 'SELECT text AS видно FROM notes ORDER BY id',
    пишет: 'INSERT INTO notes (text) VALUES ($1)',
  },
  'переходный код': {
    читает: 'SELECT COALESCE(body, text) AS видно FROM notes ORDER BY id',
    пишет: 'INSERT INTO notes (text, body) VALUES ($1, $1)',
  },
  'новый код': {
    читает: 'SELECT body AS видно FROM notes ORDER BY id',
    пишет: 'INSERT INTO notes (body) VALUES ($1)',
  },
};

// каждый запрос сам себе транзакция: отказ одного не мешает следующему
const попытка = async (запрос, значения) => {
  try {
    const { rows } = await client.query(запрос, значения);
    return { ok: true, rows };
  } catch (err) {
    return { ok: false, код: err.code, сообщение: err.message };
  }
};

/** Шаг миграции: удачный молчит, неудачный называет код SQLSTATE. */
const шаг = async (запрос) => {
  const итог = await попытка(запрос);
  if (!итог.ok) console.log(`  миграция: ${итог.код} ${итог.сообщение}`);
};

/** Что видит и что может записать версия кода на нынешней схеме. */
async function проверить(имя, метка) {
  const { читает, пишет } = версии[имя];
  const чтение = await попытка(читает);
  const запись = await попытка(пишет, [метка]);
  if (чтение.ok) {
    const пустых = чтение.rows.filter((строка) => строка.видно === null).length;
    console.log(`  ${имя} читает: строк ${чтение.rows.length}, пустых ${пустых}`);
  } else {
    console.log(`  ${имя} читает: ${чтение.код} ${чтение.сообщение}`);
  }
  console.log(`  ${имя} пишет: ${запись.ok ? 'записал' : `${запись.код} ${запись.сообщение}`}`);
}

const колонкиИСтроки = async () => {
  const { rows } = await client.query(
    `SELECT attname AS имя FROM pg_attribute
     WHERE attrelid = 'notes'::regclass AND attnum > 0 AND NOT attisdropped ORDER BY attnum`,
  );
  const колонки = rows.map((строка) => строка.имя);
  const { rows: данные } = await client.query(`SELECT ${колонки.join(', ')} FROM notes ORDER BY id`);
  console.log(`  колонки: ${колонки.join(', ')}`);
  for (const строка of данные) console.log(`    ${колонки.map((к) => `${к}=${строка[к] ?? 'null'}`).join(' ')}`);
};

const сНачала = async () => {
  await шаг('DROP TABLE IF EXISTS notes');
  await шаг('CREATE TABLE notes (id serial PRIMARY KEY, text text NOT NULL)');
  await шаг("INSERT INTO notes (text) VALUES ('была до выката')");
};

await сНачала();
console.log('--- схема, которая в проде сейчас');
await проверить('старый код', 'старый до выката');
await проверить('новый код', 'новый до выката');

console.log('--- в лоб, попытка 1: ALTER TABLE notes ADD COLUMN body text NOT NULL');
await шаг('ALTER TABLE notes ADD COLUMN body text NOT NULL');

console.log('--- в лоб, попытка 2: ALTER TABLE notes RENAME COLUMN text TO body');
await шаг('ALTER TABLE notes RENAME COLUMN text TO body');
await проверить('старый код', 'старый после переименования');
await проверить('новый код', 'новый после переименования');

await сНачала();
console.log('--- выкат 1, расширение начато: ADD COLUMN body и перенос значений');
await шаг('ALTER TABLE notes ADD COLUMN body text');
await шаг('UPDATE notes SET body = text WHERE body IS NULL');
await проверить('новый код', 'новый на половине расширения');

console.log('--- расширение доведено: ALTER TABLE notes ALTER COLUMN text DROP NOT NULL');
await шаг('ALTER TABLE notes ALTER COLUMN text DROP NOT NULL');
await проверить('старый код', 'старый в выкат 1');
await проверить('переходный код', 'переходный в выкат 1');
await проверить('новый код', 'новый в выкат 1');
await колонкиИСтроки();

console.log('--- выкат 2: догоняющий перенос, дальше пишет только body');
await шаг('UPDATE notes SET body = text WHERE body IS NULL');
await проверить('переходный код', 'переходный в выкат 2');
await проверить('новый код', 'новый в выкат 2');

console.log('--- выкат 3, сужение: ALTER TABLE notes DROP COLUMN text');
await шаг('ALTER TABLE notes DROP COLUMN text');
await проверить('новый код', 'новый в выкат 3');
await проверить('старый код', 'старый в выкат 3');

console.log('--- другой способ расширения: ADD COLUMN tag text NOT NULL DEFAULT \'\' — и что легло в строки');
await шаг("ALTER TABLE notes ADD COLUMN tag text NOT NULL DEFAULT ''");
const { rows: описание } = await client.query(
  `SELECT atthasmissing, attmissingval::text AS значение FROM pg_attribute
   WHERE attrelid = 'notes'::regclass AND attname = 'tag'`,
);
console.log(`  pg_attribute про tag: atthasmissing ${описание[0].atthasmissing}, attmissingval ${описание[0].значение}`);
await проверить('новый код', 'новый не знает про tag');

console.log('--- колонка, которую считает сама база, и попытка кода в неё записать');
await шаг('ALTER TABLE notes ADD COLUMN body_mirror text GENERATED ALWAYS AS (body) STORED');
const { rows: зеркало } = await client.query('SELECT count(*)::int AS заполнено FROM notes WHERE body_mirror IS NOT NULL');
console.log(`  строк с непустым body_mirror: ${зеркало[0].заполнено}`);
await шаг("UPDATE notes SET body_mirror = 'своё'");

await client.end();

база: postgres://postgres@127.0.0.1:<порт-базы>/app
--- схема, которая в проде сейчас
  старый код читает: строк 1, пустых 0
  старый код пишет: записал
  новый код читает: 42703 column "body" does not exist
  новый код пишет: 42703 column "body" of relation "notes" does not exist
--- в лоб, попытка 1: ALTER TABLE notes ADD COLUMN body text NOT NULL
  миграция: 23502 column "body" of relation "notes" contains null values
--- в лоб, попытка 2: ALTER TABLE notes RENAME COLUMN text TO body
  старый код читает: 42703 column "text" does not exist
  старый код пишет: 42703 column "text" of relation "notes" does not exist
  новый код читает: строк 2, пустых 0
  новый код пишет: записал
--- выкат 1, расширение начато: ADD COLUMN body и перенос значений
  новый код читает: строк 1, пустых 0
  новый код пишет: 23502 null value in column "text" of relation "notes" violates not-null constraint
--- расширение доведено: ALTER TABLE notes ALTER COLUMN text DROP NOT NULL
  старый код читает: строк 1, пустых 0
  старый код пишет: записал
  переходный код читает: строк 2, пустых 0
  переходный код пишет: записал
  новый код читает: строк 3, пустых 1
  новый код пишет: записал
  колонки: id, text, body
    id=1 text=была до выката body=была до выката
    id=3 text=старый в выкат 1 body=null
    id=4 text=переходный в выкат 1 body=переходный в выкат 1
    id=5 text=null body=новый в выкат 1
--- выкат 2: догоняющий перенос, дальше пишет только body
  переходный код читает: строк 4, пустых 0
  переходный код пишет: записал
  новый код читает: строк 5, пустых 0
  новый код пишет: записал
--- выкат 3, сужение: ALTER TABLE notes DROP COLUMN text
  новый код читает: строк 6, пустых 0
  новый код пишет: записал
  старый код читает: 42703 column "text" does not exist
  старый код пишет: 42703 column "text" of relation "notes" does not exist
--- другой способ расширения: ADD COLUMN tag text NOT NULL DEFAULT '' — и что легло в строки
  pg_attribute про tag: atthasmissing true, attmissingval {""}
  новый код читает: строк 7, пустых 0
  новый код пишет: записал
--- колонка, которую считает сама база, и попытка кода в неё записать
  строк с непустым body_mirror: 8
  миграция: 428C9 column "body_mirror" can only be updated to DEFAULT

import pg from 'pg';

// адрес базы — первой строкой и без пароля: видно, куда ушли запросы
const url = new URL(process.env.DATABASE_URL);
url.password = '';
console.log('база:', url.href);

// роль соединения в обе стороны: в выводе имена, а не pid этой машины
const кто = new Map();
const pidРоли = new Map();
const подключить = async (роль) => {
  const client = new pg.Client({ connectionString: process.env.DATABASE_URL });
  await client.connect();
  const { rows } = await client.query('SELECT pg_backend_pid() AS pid');
  кто.set(rows[0].pid, роль);
  pidРоли.set(роль, rows[0].pid);
  return client;
};

const наблюдатель = await подключить('наблюдатель');
const отчёт = await подключить('отчёт');
const миграция = await подключить('миграция');
const читатель = await подключить('читатель');

await наблюдатель.query('CREATE TABLE notes (id serial PRIMARY KEY, body text NOT NULL, slug text)');
await наблюдатель.query("INSERT INTO notes (body, slug) VALUES ('первая', 'odna'), ('вторая', 'odna')");

// шаг миграции: удачный молчит, неудачный называет код SQLSTATE. Подробности
// отказа драйвер кладёт отдельным полем — печатаем и его, если оно есть
const шаг = async (client, запрос) => {
  try {
    await client.query(запрос);
  } catch (err) {
    const ещё = err.detail ? ` | detail: ${err.detail}` : '';
    console.log(`  миграция: ${err.code} ${err.message}${ещё}`);
  }
};

const пауза = (мс) => new Promise((готово) => setTimeout(готово, мс));

/*
 * Запрос уходит на сервер не мгновенно, а вывод обязан совпадать при каждом
 * прогоне: прежде чем идти дальше, ждём, пока база сама скажет, что этот
 * сеанс чего-то ждёт.
 */
const дождатьсяСеанса = async (роль) => {
  for (;;) {
    const { rows } = await отчёт.query('SELECT wait_event_type AS чего FROM pg_stat_activity WHERE pid = $1', [
      pidРоли.get(роль),
    ]);
    if (rows[0]?.чего === 'Lock') return;
    await пауза(20);
  }
};

const замки = async () => {
  const { rows } = await отчёт.query(
    `SELECT pid, mode, granted FROM pg_locks
     WHERE relation = 'notes'::regclass ORDER BY granted DESC, pid`,
  );
  for (const строка of rows) {
    console.log(`  ${кто.get(строка.pid) ?? 'кто-то ещё'}: ${строка.mode} ${строка.granted ? 'выдан' : 'ждёт'}`);
  }
};

console.log('--- долгий отчёт открыл транзакцию и читает таблицу');
await отчёт.query('BEGIN');
await отчёт.query('SELECT count(*) FROM notes');

console.log('--- миграция с lock_timeout = 250ms');
await миграция.query("SET lock_timeout = '250ms'");
try {
  await миграция.query('ALTER TABLE notes ADD COLUMN tag text');
  console.log('  миграция: применена');
} catch (err) {
  console.log(`  миграция: ${err.code} ${err.message}`);
}

console.log('--- та же миграция без таймаута, а сразу за ней — обычное чтение');
await миграция.query('SET lock_timeout = 0');
const ждётМиграция = миграция
  .query('ALTER TABLE notes ADD COLUMN tag text')
  .then(() => 'применена')
  .catch((err) => `${err.code} ${err.message}`);
await дождатьсяСеанса('миграция');
const ждётЧитатель = читатель
  .query('SELECT count(*) FROM notes')
  .then(() => 'прочитал')
  .catch((err) => `${err.code} ${err.message}`);
await дождатьсяСеанса('читатель');
await замки();

console.log('--- отчёт закрыл транзакцию');
await отчёт.query('COMMIT');
console.log(`  миграция: ${await ждётМиграция}`);
console.log(`  читатель: ${await ждётЧитатель}`);

console.log('--- CREATE INDEX CONCURRENTLY внутри транзакции, как её открывает раннер');
await миграция.query('BEGIN');
await шаг(миграция, 'CREATE INDEX CONCURRENTLY notes_slug_idx ON notes (slug)');
await миграция.query('ROLLBACK');

console.log('--- уникальный индекс на колонке, куда старый код успел положить дубли');
await шаг(миграция, 'CREATE UNIQUE INDEX notes_slug_key ON notes (slug)');

await наблюдатель.end();
await отчёт.end();
await миграция.end();
await читатель.end();

база: postgres://postgres@127.0.0.1:<порт-базы>/app
--- долгий отчёт открыл транзакцию и читает таблицу
--- миграция с lock_timeout = 250ms
  миграция: 55P03 canceling statement due to lock timeout
--- та же миграция без таймаута, а сразу за ней — обычное чтение
  отчёт: AccessShareLock выдан
  миграция: AccessExclusiveLock ждёт
  читатель: AccessShareLock ждёт
--- отчёт закрыл транзакцию
  миграция: применена
  читатель: прочитал
--- CREATE INDEX CONCURRENTLY внутри транзакции, как её открывает раннер
  миграция: 25001 CREATE INDEX CONCURRENTLY cannot run inside a transaction block
--- уникальный индекс на колонке, куда старый код успел положить дубли
  миграция: 23505 could not create unique index "notes_slug_key" | detail: Key (slug)=(odna) is duplicated.

import { execFileSync } from 'node:child_process';
import { readFileSync, rmSync, writeFileSync } from 'node:fs';
import pg from 'pg';

// адрес базы — первой строкой и без пароля: видно, куда ушли запросы
const url = new URL(process.env.DATABASE_URL!);
url.password = '';
console.log('база:', url.href);

const client = new pg.Client({ connectionString: process.env.DATABASE_URL });
await client.connect();

// так одноразовая джоба и запускается: CMD образа, код выхода — приговор выкату
const джоба = () => {
  try {
    execFileSync('drizzle-kit', ['migrate'], { stdio: 'inherit' });
    // drizzle-kit не переводит строку после последнего сообщения
    console.log();
    console.log('  код выхода 0');
  } catch (err) {
    console.log();
    console.log(`  код выхода ${(err as { status?: number }).status}`);
  }
};

const состояние = async () => {
  const { rows: таблицы } = await client.query(
    `SELECT table_schema || '.' || table_name AS имя FROM information_schema.tables
     WHERE table_schema IN ('public', 'drizzle') ORDER BY 1`,
  );
  console.log('  таблицы базы:', таблицы.map((строка) => строка.имя).join(', ') || 'ни одной');
  const { rows: журнал } = await client.query(
    `SELECT to_regclass('drizzle.__drizzle_migrations') IS NOT NULL AS есть`,
  );
  if (журнал[0].есть) {
    const { rows: отметки } = await client.query('SELECT count(*)::int AS сколько FROM drizzle.__drizzle_migrations');
    console.log('  отметок в журнале drizzle:', отметки[0].сколько);
  } else {
    console.log('  журнала drizzle в базе нет');
  }
  const { rows: индексы } = await client.query(
    `SELECT indexname AS имя FROM pg_indexes WHERE tablename = 'notes' ORDER BY 1`,
  );
  console.log('  индексы notes:', индексы.map((строка) => строка.имя).join(', ') || 'таблицы нет');
};

console.log('--- две непримененные миграции, во второй CREATE INDEX CONCURRENTLY');
джоба();
await состояние();

console.log('--- второй шаг убран из каталога миграций');
rmSync('drizzle/0001_notes_slug_idx.sql');
const путь = 'drizzle/meta/_journal.json';
const журнал = JSON.parse(readFileSync(путь, 'utf8'));
журнал.entries = журнал.entries.filter((запись: { tag: string }) => запись.tag !== '0001_notes_slug_idx');
writeFileSync(путь, JSON.stringify(журнал, null, 2));
джоба();
await состояние();

await client.end();

база: postgres://postgres@127.0.0.1:<порт-базы>/app
--- две непримененные миграции, во второй CREATE INDEX CONCURRENTLY
No config path provided, using default 'drizzle.config.ts'
Reading config file '<пример>/drizzle.config.ts'
Using 'pg' driver for database querying
[~] applying migrations...
[~] applying migrations...
  код выхода 1
  таблицы базы: drizzle.__drizzle_migrations
  отметок в журнале drizzle: 0
  индексы notes: таблицы нет
--- второй шаг убран из каталога миграций
No config path provided, using default 'drizzle.config.ts'
Reading config file '<пример>/drizzle.config.ts'
Using 'pg' driver for database querying
[~] applying migrations...
[✓] migrations applied successfully!
  код выхода 0
  таблицы базы: drizzle.__drizzle_migrations, public.notes
  отметок в журнале drizzle: 1
  индексы notes: notes_pkey