Function applyOne

internal/identity/migrate.go:350–397

applyOne reads a migration file and applies it. For migrations without the no-transaction directive: SQL + tracker insert run in a single transaction so a partial-failure migration is safe to retry. Tx isolation defaults to READ COMMITTED, which is correct here (we don't want SERIALIZABLE forcing r

(ctx context.Context, conn *pgxpool.Conn, fsys fs.FS, name string)

// itself; the directive is opt-in for the specific Postgres features
// that require it.
func applyOne(ctx context.Context, conn *pgxpool.Conn, fsys fs.FS, name string) error {
	body, err := fs.ReadFile(fsys, name)
	if err != nil {
		return fmt.Errorf("read %s: %w", name, err)
	}

	if hasNoTransactionDirective(string(body)) {
		// pgx's simple protocol sends multi-statement strings to the
		// server as a single Query message, which Postgres wraps in an
		// implicit transaction — defeating the directive's purpose and
		// surfacing as a confusing "cannot run inside a transaction
		// block" error from the actual offending statement. Catch it
		// here with a clear message.
		if looksMultiStatement(string(body)) {
			return fmt.Errorf("migration %s uses -- e2a:no-transaction but contains multiple statements; split into separate files (one per statement)", name)
		}
		if _, err := conn.Exec(ctx, string(body)); err != nil {
			return fmt.Errorf("exec migration (no-transaction): %w", err)
		}
		if _, err := conn.Exec(ctx,
			"INSERT INTO schema_migrations (filename) VALUES ($1) ON CONFLICT DO NOTHING",
			name,
		); err != nil {
			return fmt.Errorf("record migration: %w", err)
		}
		return nil
	}

	tx, err := conn.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return fmt.Errorf("begin tx: %w", err)
	}
	defer func() { _ = tx.Rollback(ctx) }()

	if _, err := tx.Exec(ctx, string(body)); err != nil {
		return fmt.Errorf("exec migration: %w", err)
	}
	if _, err := tx.Exec(ctx,
		"INSERT INTO schema_migrations (filename) VALUES ($1) ON CONFLICT DO NOTHING",
		name,
	); err != nil {
		return fmt.Errorf("record migration: %w", err)
	}
	if err := tx.Commit(ctx); err != nil {
		return fmt.Errorf("commit: %w", err)
	}
	return nil
}