// Package migrations содержит Go-миграции goose (SQL-миграции лежат рядом // *.sql-файлами и прогоняются из embed FS пакета store). Регистрация — в // init(); чтобы она сработала, пакет blank-импортируется из store. package migrations import ( "context" "database/sql" "fmt" "strings" "time" "github.com/pressly/goose/v3" "git.vakhrushev.me/av/jellybit/internal/ident" ) func init() { goose.AddMigrationContext(upUlidIdentity, downUlidIdentity) } // upUlidIdentity переводит все таблицы на ULID-идентификаторы (TEXT PK), // разносит download.infohash в download_infohash и убирает idempotency_key // (см. openspec/changes/ulid-identity/design.md, D6). // // Работает при включённых foreign_keys (PRAGMA внутри транзакции — no-op), // поэтому порядок жёсткий: новые таблицы и данные — родители первыми, DROP // старых — дети первыми, затем RENAME (SQLite ≥ 3.25 переписывает REFERENCES // в ссылающихся таблицах). func upUlidIdentity(ctx context.Context, tx *sql.Tx) error { if err := createNewTables(ctx, tx); err != nil { return err } downloadIDs, err := migrateDownloads(ctx, tx) if err != nil { return err } recognitionIDs, err := migrateRecognitions(ctx, tx, downloadIDs) if err != nil { return err } if err := migrateHints(ctx, tx, downloadIDs); err != nil { return err } if err := migrateOverrides(ctx, tx, downloadIDs); err != nil { return err } if err := migrateCandidates(ctx, tx, recognitionIDs); err != nil { return err } if err := migrateFileLinks(ctx, tx, downloadIDs); err != nil { return err } // Старые таблицы: дети первыми, родитель последним (FK включены). for _, stmt := range []string{ `DROP TABLE file_link`, `DROP TABLE metadata_candidate`, `DROP TABLE override`, `DROP TABLE hint`, `DROP TABLE recognition`, `DROP TABLE download`, `ALTER TABLE download_new RENAME TO download`, `ALTER TABLE recognition_new RENAME TO recognition`, `ALTER TABLE hint_new RENAME TO hint`, `ALTER TABLE override_new RENAME TO override`, `ALTER TABLE metadata_candidate_new RENAME TO metadata_candidate`, `ALTER TABLE file_link_new RENAME TO file_link`, // Индексы — после переименований, с каноническими именами (старые // одноимённые ушли вместе со старыми таблицами). `CREATE INDEX idx_download_state ON download (state)`, `CREATE INDEX idx_download_infohash_download ON download_infohash (download_id)`, `CREATE INDEX idx_recognition_download ON recognition (download_id)`, `CREATE INDEX idx_hint_download ON hint (download_id)`, `CREATE INDEX idx_candidate_recognition ON metadata_candidate (recognition_id)`, `CREATE INDEX idx_file_link_download ON file_link (download_id)`, `CREATE INDEX idx_file_link_batch ON file_link (apply_batch_id)`, } { if _, err := tx.ExecContext(ctx, stmt); err != nil { return fmt.Errorf("ulid migration: %q: %w", stmt, err) } } return checkForeignKeys(ctx, tx) } // downUlidIdentity: обратной миграции нет — ULID → числовые id невосстановимы. // Откат — восстановление файла БД из копии (см. design.md, Migration Plan). func downUlidIdentity(context.Context, *sql.Tx) error { return fmt.Errorf("ulid identity migration is irreversible; restore the database file from a backup") } func createNewTables(ctx context.Context, tx *sql.Tx) error { for _, stmt := range []string{ `CREATE TABLE download_new ( id TEXT PRIMARY KEY, source_type TEXT NOT NULL, source_ref TEXT NOT NULL, display_name TEXT NOT NULL DEFAULT '', context TEXT NOT NULL DEFAULT '', state TEXT NOT NULL, error_code TEXT, error_msg TEXT, source_miss_count INTEGER NOT NULL DEFAULT 0, source_added_at TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')), updated_at TEXT NOT NULL DEFAULT (datetime('now')) )`, `CREATE TABLE download_infohash ( download_id TEXT NOT NULL REFERENCES download_new (id) ON DELETE CASCADE, infohash TEXT NOT NULL, kind TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT (datetime('now')), PRIMARY KEY (infohash, download_id) )`, `CREATE TABLE recognition_new ( id TEXT PRIMARY KEY, download_id TEXT NOT NULL REFERENCES download_new (id) ON DELETE CASCADE, attempt_no INTEGER NOT NULL DEFAULT 1, is_current INTEGER NOT NULL DEFAULT 1, media_type TEXT, title TEXT, original_title TEXT, year INTEGER, provider TEXT, provider_id TEXT, confidence REAL, reasons TEXT NOT NULL DEFAULT '[]', raw_llm TEXT, plan TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')) )`, `CREATE TABLE hint_new ( id TEXT PRIMARY KEY, download_id TEXT NOT NULL REFERENCES download_new (id) ON DELETE CASCADE, text TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT (datetime('now')) )`, `CREATE TABLE override_new ( id TEXT PRIMARY KEY, download_id TEXT NOT NULL REFERENCES download_new (id) ON DELETE CASCADE, field TEXT NOT NULL, value TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT (datetime('now')), UNIQUE (download_id, field) )`, `CREATE TABLE metadata_candidate_new ( id TEXT PRIMARY KEY, recognition_id TEXT NOT NULL REFERENCES recognition_new (id) ON DELETE CASCADE, provider TEXT NOT NULL, provider_id TEXT NOT NULL, title TEXT, year INTEGER, chosen INTEGER NOT NULL DEFAULT 0, url TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')) )`, `CREATE TABLE file_link_new ( id TEXT PRIMARY KEY, download_id TEXT NOT NULL REFERENCES download_new (id) ON DELETE CASCADE, apply_batch_id TEXT NOT NULL, src_path TEXT NOT NULL, dst_path TEXT NOT NULL, kind TEXT NOT NULL, status TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT (datetime('now')) )`, } { if _, err := tx.ExecContext(ctx, stmt); err != nil { return fmt.Errorf("ulid migration: create tables: %w", err) } } return nil } // idMap строит маппинг «старый int id → ULID» для таблицы: строки читаются в // порядке старого id (хронология), timestamp-часть ULID — из created_at, так // что лексикографический порядок новых id сохраняет исторический. Равные // секунды created_at упорядочивает monotonic-энтропия по порядку обхода. func idMap(ctx context.Context, tx *sql.Tx, table string) (map[int64]string, error) { rows, err := tx.QueryContext(ctx, `SELECT id, created_at FROM `+table+` ORDER BY id`) //nolint:gosec // имена таблиц — константы этого файла if err != nil { return nil, fmt.Errorf("ulid migration: read %s ids: %w", table, err) } defer func() { _ = rows.Close() }() out := map[int64]string{} for rows.Next() { var id int64 var createdAt string if err := rows.Scan(&id, &createdAt); err != nil { return nil, fmt.Errorf("ulid migration: scan %s id: %w", table, err) } out[id] = ident.NewIDAt(parseCreatedAt(createdAt)) } return out, rows.Err() } // parseCreatedAt разбирает метку datetime('now') (UTC); непарсибельная метка // → текущее время (порядок в пределах таблицы всё равно монотонен). func parseCreatedAt(s string) time.Time { t, err := time.ParseInLocation("2006-01-02 15:04:05", s, time.UTC) if err != nil { return time.Now() } return t } func migrateDownloads(ctx context.Context, tx *sql.Tx) (map[int64]string, error) { ids, err := idMap(ctx, tx, "download") if err != nil { return nil, err } // Сначала вычитываем всё и закрываем курсор, потом вставляем: Tx держит // одно соединение, Exec при открытых Rows на нём невозможен. type downloadRow struct { id int64 sourceType, sourceRef, displayName string contextText, state string infohash, errorCode, errorMsg sql.NullString sourceMissCount int sourceAddedAt sql.NullString createdAt, updatedAt string } var all []downloadRow rows, err := tx.QueryContext(ctx, ` SELECT id, source_type, source_ref, display_name, context, infohash, state, error_code, error_msg, source_miss_count, source_added_at, created_at, updated_at FROM download ORDER BY id`) if err != nil { return nil, fmt.Errorf("ulid migration: read downloads: %w", err) } for rows.Next() { var r downloadRow if err := rows.Scan(&r.id, &r.sourceType, &r.sourceRef, &r.displayName, &r.contextText, &r.infohash, &r.state, &r.errorCode, &r.errorMsg, &r.sourceMissCount, &r.sourceAddedAt, &r.createdAt, &r.updatedAt); err != nil { _ = rows.Close() return nil, fmt.Errorf("ulid migration: scan download: %w", err) } all = append(all, r) } if err := rows.Err(); err != nil { _ = rows.Close() return nil, fmt.Errorf("ulid migration: iterate downloads: %w", err) } _ = rows.Close() for _, r := range all { if _, err := tx.ExecContext(ctx, ` INSERT INTO download_new (id, source_type, source_ref, display_name, context, state, error_code, error_msg, source_miss_count, source_added_at, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ids[r.id], r.sourceType, r.sourceRef, r.displayName, r.contextText, r.state, r.errorCode, r.errorMsg, r.sourceMissCount, r.sourceAddedAt, r.createdAt, r.updatedAt); err != nil { return nil, fmt.Errorf("ulid migration: insert download %d: %w", r.id, err) } if r.infohash.Valid && r.infohash.String != "" { h := strings.ToLower(r.infohash.String) if _, err := tx.ExecContext(ctx, ` INSERT INTO download_infohash (download_id, infohash, kind, created_at) VALUES (?, ?, ?, ?)`, ids[r.id], h, hashKind(h), r.createdAt); err != nil { return nil, fmt.Errorf("ulid migration: insert infohash for %d: %w", r.id, err) } } } return ids, nil } // hashKind — вид инфохэша по длине hex: 40 — v1 (SHA-1), 64 — v2 (SHA-256). func hashKind(h string) string { if len(h) == 64 { return "v2" } return "v1" } func migrateRecognitions(ctx context.Context, tx *sql.Tx, downloads map[int64]string) (map[int64]string, error) { ids, err := idMap(ctx, tx, "recognition") if err != nil { return nil, err } if err := copyRows(ctx, tx, copySpec{ from: "recognition", to: "recognition_new", cols: []string{"attempt_no", "is_current", "media_type", "title", "original_title", "year", "provider", "provider_id", "confidence", "reasons", "raw_llm", "plan", "created_at"}, ids: ids, parent: parentRef{col: "download_id", ids: downloads}, }); err != nil { return nil, err } return ids, nil } func migrateHints(ctx context.Context, tx *sql.Tx, downloads map[int64]string) error { ids, err := idMap(ctx, tx, "hint") if err != nil { return err } return copyRows(ctx, tx, copySpec{ from: "hint", to: "hint_new", cols: []string{"text", "created_at"}, ids: ids, parent: parentRef{col: "download_id", ids: downloads}, }) } func migrateOverrides(ctx context.Context, tx *sql.Tx, downloads map[int64]string) error { ids, err := idMap(ctx, tx, "override") if err != nil { return err } return copyRows(ctx, tx, copySpec{ from: "override", to: "override_new", cols: []string{"field", "value", "created_at"}, ids: ids, parent: parentRef{col: "download_id", ids: downloads}, }) } func migrateCandidates(ctx context.Context, tx *sql.Tx, recognitions map[int64]string) error { ids, err := idMap(ctx, tx, "metadata_candidate") if err != nil { return err } return copyRows(ctx, tx, copySpec{ from: "metadata_candidate", to: "metadata_candidate_new", cols: []string{"provider", "provider_id", "title", "year", "chosen", "url", "created_at"}, ids: ids, parent: parentRef{col: "recognition_id", ids: recognitions}, }) } func migrateFileLinks(ctx context.Context, tx *sql.Tx, downloads map[int64]string) error { ids, err := idMap(ctx, tx, "file_link") if err != nil { return err } return copyRows(ctx, tx, copySpec{ from: "file_link", to: "file_link_new", cols: []string{"apply_batch_id", "src_path", "dst_path", "kind", "status", "created_at"}, ids: ids, parent: parentRef{col: "download_id", ids: downloads}, }) } // copySpec описывает перенос таблицы: собственный маппинг id, FK-родитель и // прочие столбцы, копируемые как есть. type copySpec struct { from, to string cols []string ids map[int64]string parent parentRef } type parentRef struct { col string ids map[int64]string } func copyRows(ctx context.Context, tx *sql.Tx, spec copySpec) error { colList := strings.Join(spec.cols, ", ") // Сначала вычитываем всё и закрываем курсор, потом вставляем (Exec при // открытых Rows на соединении транзакции невозможен). type rowData struct { oldID, parentID int64 rest []any } var all []rowData //nolint:gosec // имена таблиц/столбцов — константы этого файла rows, err := tx.QueryContext(ctx, fmt.Sprintf( `SELECT id, %s, %s FROM %s ORDER BY id`, spec.parent.col, colList, spec.from)) if err != nil { return fmt.Errorf("ulid migration: read %s: %w", spec.from, err) } for rows.Next() { r := rowData{rest: make([]any, len(spec.cols))} dest := append([]any{&r.oldID, &r.parentID}, scanPtrs(r.rest)...) if err := rows.Scan(dest...); err != nil { _ = rows.Close() return fmt.Errorf("ulid migration: scan %s: %w", spec.from, err) } all = append(all, r) } if err := rows.Err(); err != nil { _ = rows.Close() return fmt.Errorf("ulid migration: iterate %s: %w", spec.from, err) } _ = rows.Close() ph := strings.TrimSuffix(strings.Repeat("?, ", len(spec.cols)), ", ") //nolint:gosec // имена таблиц/столбцов — константы этого файла insert := fmt.Sprintf(`INSERT INTO %s (id, %s, %s) VALUES (?, ?, %s)`, spec.to, spec.parent.col, colList, ph) for _, r := range all { newParent, ok := spec.parent.ids[r.parentID] if !ok { return fmt.Errorf("ulid migration: %s row %d references unknown %s %d", spec.from, r.oldID, spec.parent.col, r.parentID) } args := append([]any{spec.ids[r.oldID], newParent}, r.rest...) if _, err := tx.ExecContext(ctx, insert, args...); err != nil { return fmt.Errorf("ulid migration: insert %s row %d: %w", spec.to, r.oldID, err) } } return nil } // scanPtrs — указатели на элементы среза для rows.Scan (значения любого типа // SQLite едут через any и вставляются обратно как есть). func scanPtrs(vals []any) []any { out := make([]any, len(vals)) for i := range vals { out[i] = &vals[i] } return out } // checkForeignKeys — финальная самопроверка целостности после rebuild. func checkForeignKeys(ctx context.Context, tx *sql.Tx) error { rows, err := tx.QueryContext(ctx, `PRAGMA foreign_key_check`) if err != nil { return fmt.Errorf("ulid migration: foreign_key_check: %w", err) } defer func() { _ = rows.Close() }() if rows.Next() { var table string var rowid, parent, fkid any _ = rows.Scan(&table, &rowid, &parent, &fkid) return fmt.Errorf("ulid migration: foreign key violation in %s after rebuild", table) } return rows.Err() }