1package storage
2
3import (
4 "context"
5 "crypto/sha256"
6 "database/sql"
7 "embed"
8 "encoding/hex"
9 "errors"
10 "fmt"
11 "io/fs"
12 "path"
13 "regexp"
14 "strconv"
15 "time"
16)
17
18//go:embed migrations/*.sql
19var embeddedMigrations embed.FS
20
21var migrationName = regexp.MustCompile(`^([0-9]{4})_[a-z0-9_]+\.sql$`)
22
23type migration struct {
24 version int
25 name string
26 checksum string
27 sql string
28}
29
30func loadMigrations(fsys fs.FS, directory string) ([]migration, error) {
31 entries, err := fs.ReadDir(fsys, directory)
32 if err != nil {
33 return nil, fmt.Errorf("read migrations: %w", err)
34 }
35 migrations := make([]migration, 0, len(entries))
36 for _, entry := range entries {
37 if entry.IsDir() {
38 return nil, fmt.Errorf("migration %q is a directory", entry.Name())
39 }
40 matches := migrationName.FindStringSubmatch(entry.Name())
41 if matches == nil {
42 return nil, fmt.Errorf("migration %q has invalid name", entry.Name())
43 }
44 version, err := strconv.Atoi(matches[1])
45 if err != nil || version < 1 {
46 return nil, fmt.Errorf("migration %q has invalid version", entry.Name())
47 }
48 content, err := fs.ReadFile(fsys, path.Join(directory, entry.Name()))
49 if err != nil {
50 return nil, fmt.Errorf("read migration %q: %w", entry.Name(), err)
51 }
52 if len(content) == 0 {
53 return nil, fmt.Errorf("migration %q is empty", entry.Name())
54 }
55 digest := sha256.Sum256(content)
56 migrations = append(migrations, migration{
57 version: version,
58 name: entry.Name(),
59 checksum: hex.EncodeToString(digest[:]),
60 sql: string(content),
61 })
62 }
63 for i, item := range migrations {
64 want := i + 1
65 if item.version != want {
66 return nil, fmt.Errorf("migration %q has version %d, want %d", item.name, item.version, want)
67 }
68 }
69 return migrations, nil
70}
71
72func applyMigrations(ctx context.Context, db *sql.DB, migrations []migration, now time.Time) error {
73 conn, err := db.Conn(ctx)
74 if err != nil {
75 return classify("reserve migration connection", err)
76 }
77 defer conn.Close()
78 var foreignKeys int
79 if err := conn.QueryRowContext(ctx, `PRAGMA foreign_keys`).Scan(&foreignKeys); err != nil {
80 return classify("read foreign_keys pragma", err)
81 }
82 if _, err := conn.ExecContext(ctx, `PRAGMA foreign_keys = OFF`); err != nil {
83 return classify("turn foreign keys off for migrations", err)
84 }
85 err = applyMigrationsOn(ctx, conn, migrations, now)
86 if _, restore := conn.ExecContext(context.WithoutCancel(ctx), `PRAGMA foreign_keys = `+strconv.Itoa(foreignKeys)); restore != nil {
87 return errors.Join(err, classify("restore foreign_keys pragma", restore))
88 }
89 return err
90}
91
92func applyMigrationsOn(ctx context.Context, conn *sql.Conn, migrations []migration, now time.Time) error {
93 tx, err := conn.BeginTx(ctx, nil)
94 if err != nil {
95 return classify("begin migrations", err)
96 }
97 defer tx.Rollback()
98
99 if _, err := tx.ExecContext(ctx, `
100CREATE TABLE IF NOT EXISTS schema_migrations (
101 version INTEGER PRIMARY KEY CHECK (version > 0),
102 checksum TEXT NOT NULL CHECK (length(checksum) = 64),
103 applied_at INTEGER NOT NULL
104) STRICT`); err != nil {
105 return classify("create migration ledger", err)
106 }
107
108 rows, err := tx.QueryContext(ctx, `SELECT version, checksum FROM schema_migrations ORDER BY version`)
109 if err != nil {
110 return classify("read migration ledger", err)
111 }
112 applied := 0
113 for rows.Next() {
114 var version int
115 var checksum string
116 if err := rows.Scan(&version, &checksum); err != nil {
117 rows.Close()
118 return classify("scan migration ledger", err)
119 }
120 if applied >= len(migrations) {
121 rows.Close()
122 return fmt.Errorf("migration version %d: %w", version, ErrSchemaTooNew)
123 }
124 want := migrations[applied]
125 if version != want.version || checksum != want.checksum {
126 rows.Close()
127 return fmt.Errorf("migration ledger row %d does not match %s: %w", version, want.name, ErrSchemaInvalid)
128 }
129 applied++
130 }
131 if err := rows.Err(); err != nil {
132 rows.Close()
133 return classify("iterate migration ledger", err)
134 }
135 if err := rows.Close(); err != nil {
136 return classify("close migration ledger", err)
137 }
138
139 for _, item := range migrations[applied:] {
140 if _, err := tx.ExecContext(ctx, item.sql); err != nil {
141 return classify("apply migration "+item.name, err)
142 }
143 if _, err := tx.ExecContext(ctx,
144 `INSERT INTO schema_migrations (version, checksum, applied_at) VALUES (?, ?, ?)`,
145 item.version, item.checksum, now.UTC().UnixMicro()); err != nil {
146 return classify("record migration "+item.name, err)
147 }
148 }
149 if applied < len(migrations) {
150 if err := checkForeignKeys(ctx, tx); err != nil {
151 return err
152 }
153 }
154 if err := tx.Commit(); err != nil {
155 return classify("commit migrations", err)
156 }
157 return nil
158}
159
160func checkForeignKeys(ctx context.Context, tx *sql.Tx) error {
161 var table, parent string
162 var row sql.NullInt64
163 var key int
164 err := tx.QueryRowContext(ctx, `PRAGMA foreign_key_check`).Scan(&table, &row, &parent, &key)
165 switch {
166 case errors.Is(err, sql.ErrNoRows):
167 return nil
168 case err != nil:
169 return classify("check foreign keys after migrations", err)
170 }
171 return fmt.Errorf("migrations left a row of %s whose %s row is missing: %w", table, parent, ErrSchemaInvalid)
172}