Netpulse_SasS/server/cmd/netpulse-migrate/main.go

// Команда netpulse-migrate — накочування схеми.
//
// Бінарник, а не скрипт: у контейнері немає ані PowerShell, ані psql, а
// вимагати клієнт Postgres поруч із застосунком означає тягнути в образ
// половину дистрибутива заради одного запуску.
//
// Міграції вшиті в бінарник через embed: файл, який лежить поруч,
// рано чи пізно виявиться версією з іншого релізу.
//
// Крім схеми, ця команда відповідає ще за одне — і саме тому вона тут, а
// не в скрипті установника. На ЧИСТІЙ базі вона видає ролям netpulse_app
// і netpulse_worker паролі з DSN, які їй передали, і перевіряє, що ними
// справді можна зайти. Тобто нова інсталяція вмикає RLS сама, без
// окремого кроку й без вікна, у якому RLS вимкнено. На базі, де вже
// щось накочено, вона паролів не чіпає: перехід на роль без BYPASSRLS
// там — покрокова процедура з відкатом (deploy/RLS-EXISTING-INSTALL.md),
// а не побічний наслідок `docker compose up`. Як саме розрізняються ці
// два випадки — у коментарі до installKind.
package main

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"flag"
	"fmt"
	"io/fs"
	"os"
	"os/signal"
	"regexp"
	"sort"
	"strings"
	"syscall"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
	schema "github.com/netpulse/netpulse/server/migrations"
)

// Шлях відносний до кореня модуля не працює: embed бачить лише те, що
// лежить у каталозі пакета або нижче. Тому міграції тягнуться через
// окремий пакет, який стоїть поруч із ними.
var migrations = schema.Files

// migrateLockKey — advisory-блокування на час накочування.
//
// Два інстанси, що стартують одночасно (rolling update, docker compose
// зі скейлом), інакше накотили б ту саму міграцію двічі: перевірка
// «чи застосовано» і сам запис — різні моменти часу.
const migrateLockKey = 0x6e70_6d67 // "npmg"

// continuousRe ловить те, що не можна виконати в транзакції.
//
// TimescaleDB відмовляє: CREATE MATERIALIZED VIEW WITH
// (timescaledb.continuous) поза транзакційним блоком. Такі файли
// накочуються без обгортки — ціна в тому, що збій посеред файлу лишає
// половину змін, тому їх свідомо тримають короткими.
var continuousRe = regexp.MustCompile(`timescaledb\.continuous`)

func main() {
	if err := run(); err != nil {
		fmt.Fprintln(os.Stderr, "netpulse-migrate:", err)
		os.Exit(1)
	}
}

func run() error {
	dsn := flag.String("dsn", os.Getenv("NETPULSE_DSN"), "postgres://user:pass@host:5432/db")
	dsnApp := flag.String("dsn-app", os.Getenv("NETPULSE_DSN_APP"),
		"DSN, яким ходитимуть api й collector; на чистій базі звідси береться пароль ролі")
	dsnWorker := flag.String("dsn-worker", os.Getenv("NETPULSE_DSN_WORKER"),
		"DSN фонових тактів; на чистій базі звідси береться пароль ролі")
	dryRun := flag.Bool("dry-run", false, "лише показати, що буде застосовано")
	timeout := flag.Duration("timeout", 10*time.Minute, "стеля на всі міграції")
	flag.Parse()

	if *dsn == "" {
		return errors.New("не вказано -dsn (або NETPULSE_DSN)")
	}

	files, err := listMigrations()
	if err != nil {
		return err
	}

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	ctx, cancel := context.WithTimeout(ctx, *timeout)
	defer cancel()

	pool, err := pgxpool.New(ctx, *dsn)
	if err != nil {
		return fmt.Errorf("підключення: %w", err)
	}
	defer pool.Close()

	// Одне з'єднання на весь запуск: advisory-блокування живе рівно
	// стільки, скільки з'єднання, яке його взяло.
	conn, err := pool.Acquire(ctx)
	if err != nil {
		return fmt.Errorf("підключення: %w", err)
	}
	defer conn.Release()

	if err := checkEncoding(ctx, conn.Conn()); err != nil {
		return err
	}

	if err := bootstrap(ctx, conn.Conn()); err != nil {
		return err
	}

	if _, err := conn.Exec(ctx, `SELECT pg_advisory_lock($1)`, int64(migrateLockKey)); err != nil {
		return fmt.Errorf("блокування: %w", err)
	}
	defer func() {
		_, _ = conn.Exec(context.WithoutCancel(ctx),
			`SELECT pg_advisory_unlock($1)`, int64(migrateLockKey))
	}()

	// Ознака інсталяції знімається ТУТ: під блокуванням і до першої
	// застосованої міграції. Пізніше вона вже неправдива — після
	// накочування будь-яка база виглядає як «наявна інсталяція».
	//
	// Під -dry-run не знімається взагалі: режим обіцяє «лише показати»,
	// а перший запис відповіді — теж запис, і він назавжди.
	var inst installInfo
	if !*dryRun {
		inst, err = installKind(ctx, conn.Conn())
		if err != nil {
			return err
		}
	}

	applied, err := appliedVersions(ctx, conn.Conn())
	if err != nil {
		return err
	}

	pending := 0
	for _, f := range files {
		version := strings.TrimSuffix(f, ".sql")
		body, err := migrations.ReadFile(f)
		if err != nil {
			return err
		}
		sum := sha256.Sum256(body)
		checksum := hex.EncodeToString(sum[:])

		if old, ok := applied[version]; ok {
			// Змінена вже застосована міграція — це майже завжди
			// помилка: у різних інсталяціях опиниться різна схема під
			// одним номером. Кажемо про це й зупиняємось.
			if old != checksum {
				return fmt.Errorf(
					"міграцію %s змінено після застосування (було %s…, стало %s…).\n"+
						"Не правте застосовані міграції — додайте нову",
					version, old[:8], checksum[:8])
			}
			continue
		}

		pending++
		if *dryRun {
			fmt.Printf("буде застосовано  %s\n", version)
			continue
		}

		fmt.Printf("застосовую  %s\n", version)
		if err := apply(ctx, conn.Conn(), version, checksum, string(body)); err != nil {
			return fmt.Errorf("міграція %s: %w", version, err)
		}
	}

	switch {
	case pending == 0:
		fmt.Println("схема актуальна")
	case *dryRun:
		fmt.Printf("непримінених міграцій: %d\n", pending)
	default:
		fmt.Printf("застосовано міграцій: %d\n", pending)
	}

	if *dryRun {
		return nil
	}
	// Ролі — після схеми, бо саме схема (0063) їх і заводить. І досі
	// під блокуванням: два мігратори, що стартували разом, інакше
	// виконали б ALTER ROLE ... PASSWORD одночасно.
	return ensureRoles(ctx, conn.Conn(), *dsn, *dsnApp, *dsnWorker, inst)
}

func listMigrations() ([]string, error) {
	entries, err := fs.ReadDir(migrations, ".")
	if err != nil {
		return nil, err
	}
	var out []string
	for _, e := range entries {
		if !e.IsDir() && strings.HasSuffix(e.Name(), ".sql") {
			out = append(out, e.Name())
		}
	}
	// Порядок — за іменем: номер на початку файлу і є версією.
	sort.Strings(out)
	if len(out) == 0 {
		return nil, errors.New("у бінарнику немає жодної міграції")
	}
	return out, nil
}

func bootstrap(ctx context.Context, conn *pgx.Conn) error {
	if _, err := conn.Exec(ctx, `
		CREATE TABLE IF NOT EXISTS public.schema_migrations (
		  version    text PRIMARY KEY,
		  checksum   text NOT NULL,
		  applied_at timestamptz NOT NULL DEFAULT now()
		)
	`); err != nil {
		return err
	}

	// Обидві таблиці заводить мігратор, а не міграція: вони мусять
	// існувати ДО того, як накотиться перша з них.
	if _, err := conn.Exec(ctx, `
		CREATE TABLE IF NOT EXISTS public.netpulse_install (
		  singleton   boolean PRIMARY KEY DEFAULT true CHECK (singleton),
		  fresh       boolean NOT NULL,
		  applied_was integer NOT NULL,
		  decided_at  timestamptz NOT NULL DEFAULT now()
		)
	`); err != nil {
		return err
	}
	_, err := conn.Exec(ctx, `
		COMMENT ON TABLE public.netpulse_install IS
		  'Чи була база порожня, коли netpulse-migrate уперше її побачив. '
		  'Від цього рядка залежить, чи вільно міграторові видавати паролі '
		  'ролям, тобто вмикати RLS. Пишеться один раз і не переглядається.'
	`)
	return err
}

// installKind — «чиста база» чи «оновлення наявної інсталяції».
//
// Питання не риторичне: від відповіді залежить, чи вільно міграторові
// видати ролям паролі й тим самим перевести застосунок на роль без
// BYPASSRLS. Помилка в один бік дає нову інсталяцію, у якій api не може
// підключитись; у другий — робочий стенд, який серед ночі перемкнувся
// сам, бо хтось дописав рядок у .env.
//
// Ознака одна, фактична й перевірна очима: чи була public.schema_migrations
// ПОРОЖНЯ в ту мить, коли мігратор уперше побачив цю базу.
//
//   - порожня — жодна міграція ще не котилась, тобто в базі немає ані
//     схеми, ані даних, ані клієнтів. Перемикати нічого, ламати нічого;
//     роль без BYPASSRLS — просто те, з чим ця інсталяція народжується.
//   - непорожня — інсталяція вже працює. Перехід на ній робить людина
//     покроково (deploy/RLS-EXISTING-INSTALL.md), а не `docker compose up`.
//
// Чому відповідь ЗАПИСУЄТЬСЯ, а не обчислюється щоразу. Бо обчислення
// дає різні відповіді в різні дні. Перший запуск, що впав на тридцятій
// міграції, лишає непорожню schema_migrations — і повторний запуск
// вирішив би, що це «наявна інсталяція», не видав би паролів, і нова
// система лишилась би без входу. Записаний рядок такої різниці не має:
// його пишуть раз, під тим самим advisory-блокуванням, що й накочування.
//
// Чому не «чи є роль netpulse_app без пароля» і не «чи застосовано 0063».
// Обидві ознаки не розрізняють двох випадків, які треба розрізнити: у
// наявної інсталяції ДО переходу роль теж є й теж без пароля. Тобто
// вгадування дало б рівно ту тиху зміну, проти якої все це написано.
func installKind(ctx context.Context, conn *pgx.Conn) (installInfo, error) {
	if _, err := conn.Exec(ctx, `
		INSERT INTO public.netpulse_install (fresh, applied_was)
		SELECT count(*) = 0, count(*) FROM public.schema_migrations
		ON CONFLICT (singleton) DO NOTHING
	`); err != nil {
		return installInfo{}, fmt.Errorf("ознака інсталяції: %w", err)
	}

	var out installInfo
	if err := conn.QueryRow(ctx,
		`SELECT fresh, applied_was, decided_at FROM public.netpulse_install`,
	).Scan(&out.fresh, &out.appliedWas, &out.decidedAt); err != nil {
		return installInfo{}, fmt.Errorf("ознака інсталяції: %w", err)
	}
	return out, nil
}

// installInfo — записана відповідь разом із доказом, з якого вона взялась.
//
// appliedWas зберігається не для коду, а для людини: питання «чому мій
// стенд вважається наявною інсталяцією» має відповідь в одному SELECT,
// а не в читанні цього файлу.
type installInfo struct {
	fresh      bool
	appliedWas int
	decidedAt  time.Time
}

func appliedVersions(ctx context.Context, conn *pgx.Conn) (map[string]string, error) {
	rows, err := conn.Query(ctx, `SELECT version, checksum FROM public.schema_migrations`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	out := map[string]string{}
	for rows.Next() {
		var v, c string
		if err := rows.Scan(&v, &c); err != nil {
			return nil, err
		}
		out[v] = c
	}
	return out, rows.Err()
}

// apply накочує один файл.
//
// Зазвичай в одній транзакції разом із записом у schema_migrations: без
// цього збій посеред файлу лишає схему в стані, який ніхто не описував,
// а наступний запуск вважає міграцію незастосованою й повторює її.
func apply(ctx context.Context, conn *pgx.Conn, version, checksum, body string) error {
	if continuousRe.MatchString(body) {
		// TimescaleDB забороняє continuous aggregates у транзакції.
		// Позначку ставимо після успіху: інакше збій лишив би міграцію
		// «застосованою» без застосування.
		if _, err := conn.Exec(ctx, body); err != nil {
			return err
		}
		_, err := conn.Exec(ctx,
			`INSERT INTO public.schema_migrations (version, checksum) VALUES ($1, $2)`,
			version, checksum)
		return err
	}

	tx, err := conn.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback(ctx) }()

	if _, err := tx.Exec(ctx, body); err != nil {
		return err
	}
	if _, err := tx.Exec(ctx,
		`INSERT INTO public.schema_migrations (version, checksum) VALUES ($1, $2)`,
		version, checksum); err != nil {
		return err
	}
	return tx.Commit(ctx)
}

// ensureRoles — крок, після якого застосунок заходить у базу тією роллю,
// якою мав, і про це відомо ще до того, як api спробує піднятись.
//
// Робить рівно дві речі, і друга важливіша за першу.
//
//  1. На ЧИСТІЙ базі видає ролям паролі. Не міграція, бо пароль у
//     міграції — це пароль у git і в контрольній сумі
//     public.schema_migrations, тобто пароль, який уже не змінити, не
//     зачепивши перевірку цілісності. І не установник окремим кроком,
//     бо крок, який можна забути, забувають. Пароль береться з САМОГО
//     DSN застосунку: у базу лягає рівно той рядок, який потім надішле
//     pgx, — той самий текст, розібраний тим самим розбирачем.
//
//  2. Заходить кожним DSN і перевіряє, ким саме зайшов. Це та половина,
//     якої бракувало 0063: тоді перевірили, що роль не бачить ЧУЖОГО, і
//     не перевірили, що вона взагалі заходить і бачить СВОЄ. Зелена
//     перевірка доводить рівно те, що вона перевіряє, тому тут
//     перевіряється найдешевше й найгучніше: сам вхід і атрибут
//     BYPASSRLS на тому кінці.
//
// Порожній appDSN — робочий стан, а не недоглянута конфігурація: так
// ходять тести, `go run` на стенді розробника й кожен .env, написаний до
// цієї зміни. Тоді функція не робить нічого.
func ensureRoles(ctx context.Context, conn *pgx.Conn, ownerDSN, appDSN, workerDSN string, inst installInfo) error {
	if appDSN == "" {
		return nil
	}

	owner, err := pgx.ParseConfig(ownerDSN)
	if err != nil {
		return fmt.Errorf("розбір DSN власника: %w", err)
	}
	app, err := pgx.ParseConfig(appDSN)
	if err != nil {
		return fmt.Errorf("розбір DSN застосунку: %w%s", err, dsnHint)
	}

	// DSN згорнувся до ролі власника — тобто NETPULSE_APP_PASSWORD
	// порожній і стек працює як до 0063. Для наявної інсталяції це
	// звичайний стан, і мовчати про нього правильно. Для чистої — ні:
	// нова система, яка щойно народилась без другого рубежу, має про це
	// сказати вголос один раз, поки її ще ніхто не наповнив даними.
	if app.User == owner.User {
		if inst.fresh {
			fmt.Println("УВАГА: чиста база, а NETPULSE_APP_PASSWORD порожній.")
			fmt.Println("       Застосунок ходитиме роллю власника, і жодна з політик RLS")
			fmt.Println("       не діятиме — ізоляцію кабінетів триматиме лише предикат")
			fmt.Println("       tenant_id у коді. Заповніть NETPULSE_APP_PASSWORD і")
			fmt.Println("       NETPULSE_WORKER_PASSWORD у .env і накотіть ще раз.")
		}
		return nil
	}

	// Роль застосунку без ролі воркера — найгірший з можливих станів, і
	// саме тому це помилка, а не попередження. Інтерфейс при ньому
	// працює, а фонові такти мовчки нічого не знаходять: запити-шукачі
	// черг ходять поверх усіх кабінетів, і під NOBYPASSRLS повертають
	// нуль рядків без жодної помилки в журналі. Бекапи не запускаються,
	// алерти не розсилаються, події не доходять до браузера — і все це
	// виглядає як тиша, а не як поломка.
	if workerDSN == "" {
		return fmt.Errorf(
			"NETPULSE_APP_PASSWORD задано, а NETPULSE_WORKER_PASSWORD — ні.\n"+
				"Роль %s не має BYPASSRLS, тому фонові такти нічого не знайдуть і не "+
				"поскаржаться. Задайте обидва паролі або жодного", app.User)
	}
	worker, err := pgx.ParseConfig(workerDSN)
	if err != nil {
		return fmt.Errorf("розбір DSN воркера: %w%s", err, dsnHint)
	}

	switch {
	case inst.fresh:
		if err := setPassword(ctx, conn, app.User, app.Password); err != nil {
			return err
		}
		if err := setPassword(ctx, conn, worker.User, worker.Password); err != nil {
			return err
		}
		fmt.Printf("чиста база: видано паролі ролям %s і %s\n", app.User, worker.User)
	default:
		// Наявна інсталяція: паролі тут уже видані руками, за
		// deploy/RLS-EXISTING-INSTALL.md. Переписувати їх із .env
		// мігратор не береться — це означало б, що файл, який хтось
		// відредагував і не перечитав, мовчки змінює доступ до робочої
		// бази. Нижче лишається сама перевірка входу, і саме вона
		// скаже, якщо крок з паролями пропустили.
		fmt.Printf("наявна інсталяція (на момент %s застосованих міграцій: %d): "+
			"паролі ролей не чіпаю\n",
			inst.decidedAt.Format("2006-01-02"), inst.appliedWas)
	}

	if err := checkLogin(ctx, appDSN, app.User, false); err != nil {
		return err
	}
	if err := checkLogin(ctx, workerDSN, worker.User, true); err != nil {
		return err
	}
	fmt.Printf("вхід перевірено: api й collector — %s (без BYPASSRLS), "+
		"фонові такти — %s (з BYPASSRLS)\n", app.User, worker.User)
	return nil
}

// dsnHint — про пароль, який зіпсував DSN.
//
// Урок цього проєкту, і коштував він півдня. Пароль ролі їде всередині
// URL, тому не кожен рядок у ньому лишається собою:
//
//   - `/` обриває користувацьку частину, і хвіст пароля стає шляхом,
//     тобто іменем бази. З'єднання не встановлюється, у журналі —
//     «database ... does not exist»;
//   - `%` читається як початок %XX. Пароль при цьому мовчки стає іншим
//     рядком, і помилка виглядає як «authentication failed» без причини.
//
// Обидва символи трапляються в base64 (`/`) і в тому, що люди набирають
// руками (`%`). Виглядає це не як зіпсований рядок у .env, а як «RLS усе
// зламав», і шукають не там. Тому hex — не порада, а вимога.
const dsnHint = "\n" +
	"Найчастіша причина — пароль ролі не в hex. Пароль їде всередині URL:\n" +
	"`/` обриває його й перетворює хвіст на ім'я бази, `%` читається як\n" +
	"початок %XX і мовчки змінює пароль. Генеруйте так: openssl rand -hex 24"

// setPassword видає роль пароль.
//
// Лапки ставить сам Postgres через format(%I/%L): ALTER ROLE не приймає
// параметрів, а команда, яка ЗАДАЄ пароль, — найгірше місце для склеювання
// рядків руками. Ціна відома й прийнята: якщо на сервері ввімкнено
// log_statement = all, готова команда потрапить у журнал БД разом із
// паролем. Так само поводиться будь-який ALTER ROLE з psql.
func setPassword(ctx context.Context, conn *pgx.Conn, role, password string) error {
	if password == "" {
		return fmt.Errorf("порожній пароль ролі %s: DSN зібрано неправильно", role)
	}
	var stmt string
	if err := conn.QueryRow(ctx,
		`SELECT format('ALTER ROLE %I PASSWORD %L', $1::text, $2::text)`,
		role, password).Scan(&stmt); err != nil {
		return fmt.Errorf("скласти команду для ролі %s: %w", role, err)
	}
	if _, err := conn.Exec(ctx, stmt); err != nil {
		return fmt.Errorf("видати пароль ролі %s: %w", role, err)
	}
	return nil
}

// checkLogin підключається наданим DSN і питає базу, ким він зайшов.
//
// Окреме з'єднання, а не SELECT з наявного: перевіряється саме те, що
// ламається, — автентифікація рядком, який зібрав docker compose.
//
// wantBypass — не прикраса. Роль застосунку з BYPASSRLS означає, що все
// нижче зелене й нічого не діє (рівно стан до 0063). Роль воркера БЕЗ
// BYPASSRLS означає протилежну поломку: черги мовчки порожні. Обидві
// беззвучні, тому обидві тут — помилка.
func checkLogin(ctx context.Context, dsn, wantUser string, wantBypass bool) error {
	c, err := pgx.Connect(ctx, dsn)
	if err != nil {
		return fmt.Errorf("вхід роллю %s не вдався: %w%s\n"+
			"Якщо це інсталяція, зроблена до 0063, — паролі ролям видає\n"+
			"deploy/RLS-EXISTING-INSTALL.md, крок 2; мігратор на непорожній базі\n"+
			"робить це навмисно не сам", wantUser, err, dsnHint)
	}
	defer func() { _ = c.Close(context.WithoutCancel(ctx)) }()

	var got string
	var bypass bool
	if err := c.QueryRow(ctx,
		`SELECT current_user, rolbypassrls FROM pg_roles WHERE rolname = current_user`,
	).Scan(&got, &bypass); err != nil {
		return fmt.Errorf("перевірка ролі %s: %w", wantUser, err)
	}
	if got != wantUser {
		return fmt.Errorf("DSN обіцяв роль %s, а з'єднання відкрито як %s", wantUser, got)
	}
	if bypass != wantBypass {
		if wantBypass {
			return fmt.Errorf(
				"роль %s не має BYPASSRLS: фонові такти нічого не знайдуть і не "+
					"поскаржаться (ALTER ROLE %s BYPASSRLS)", got, got)
		}
		return fmt.Errorf(
			"роль %s має BYPASSRLS: політики RLS на неї не діють, і все нижче "+
				"виглядатиме справним, лишаючись без другого рубежу "+
				"(ALTER ROLE %s NOBYPASSRLS)", got, got)
	}
	return nil
}

// checkEncoding зупиняє накочування на базі не в UTF-8.
//
// SQL_ASCII не кодування, а його відсутність: сервер просто пропускає
// байти. Українські тексти в колонках при цьому виглядають цілими, і
// саме тому проблему помічають пізно — коли ламається щось інше:
//
//   - jsonb із не-ASCII, надісланим у вигляді \uXXXX, не вставляється
//     взагалі («unsupported Unicode escape sequence»), а так шле JSON
//     половина HTTP-бібліотек;
//   - lower(), upper() і сортування працюють як над латиницею, тож
//     пошук за іменем хоста мовчки не знаходить очевидне.
//
// Виправити це після наливання даних можна лише повним перестворенням
// бази, тому перевірка стоїть перед першою ж міграцією, а не в README.
func checkEncoding(ctx context.Context, conn *pgx.Conn) error {
	var enc string
	if err := conn.QueryRow(ctx,
		`SELECT pg_encoding_to_char(encoding) FROM pg_database WHERE datname = current_database()`,
	).Scan(&enc); err != nil {
		return fmt.Errorf("кодування бази: %w", err)
	}
	if enc != "UTF8" {
		return fmt.Errorf(
			"база в кодуванні %s, потрібне UTF8.\n"+
				"Перестворіть її: CREATE DATABASE ... ENCODING 'UTF8' TEMPLATE template0;\n"+
				"У docker-compose за це відповідає POSTGRES_INITDB_ARGS", enc)
	}
	return nil
}