mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-09-10 18:45:53 +00:00
fix(recovery): migrate legacy point schema before indexing
This commit is contained in:
rcourtman
@@ -223,6 +223,12 @@ those shared store reads, including recovery-backed reporting, alert rollups,
|
||||
and tenant-scoped AI recovery-point adapters. A malformed metadata blob may
|
||||
degrade row-local enrichment, but it must not take down adjacent readers that
|
||||
consume the same canonical recovery store.
|
||||
That same recovery-store migration boundary must keep legacy schema upgrades in
|
||||
dependency order. When a persisted `recovery.db` predates columns such as
|
||||
`item_type`, the store must add the migrated columns before creating indexes or
|
||||
running query paths that reference them, so opening a legacy store backfills
|
||||
cleanly instead of returning `500` from `/api/recovery/points` or
|
||||
`/api/recovery/rollups` during schema initialization.
|
||||
That same hook-boundary normalization also owns the runtime recovery display
|
||||
model. Canonical recovery points and rollups must expose `display.itemLabel`
|
||||
and `display.itemType` to recovery consumers, while legacy transport fields
|
||||
|
||||
@@ -286,6 +286,135 @@ func TestRecoveryPointsEndpointToleratesMalformedPersistedMetadata(t *testing.T)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecoveryPointsEndpointMigratesLegacySchemaMissingItemType(t *testing.T) {
|
||||
srv := newIntegrationServerWithoutMock(t, nil)
|
||||
|
||||
dbPath := seedLegacyRecoveryPointWithoutItemTypeForIntegration(t, srv.config, recovery.RecoveryPoint{
|
||||
ID: "legacy-point-no-item-type",
|
||||
Provider: recovery.ProviderKubernetes,
|
||||
Kind: recovery.KindSnapshot,
|
||||
Mode: recovery.ModeSnapshot,
|
||||
Outcome: recovery.OutcomeSuccess,
|
||||
StartedAt: timePtrIntegration(time.Date(2026, 2, 24, 12, 0, 0, 0, time.UTC)),
|
||||
CompletedAt: timePtrIntegration(time.Date(2026, 2, 24, 12, 0, 0, 0, time.UTC)),
|
||||
SubjectRef: &recovery.ExternalRef{
|
||||
Type: "k8s-pvc",
|
||||
Namespace: "default",
|
||||
Name: "data",
|
||||
},
|
||||
RepositoryRef: &recovery.ExternalRef{
|
||||
Type: "velero-backup-storage-location",
|
||||
Name: "repo-a",
|
||||
},
|
||||
Details: map[string]any{
|
||||
"foo": "bar",
|
||||
},
|
||||
})
|
||||
|
||||
res, err := http.Get(srv.server.URL + "/api/recovery/points?limit=500")
|
||||
if err != nil {
|
||||
t.Fatalf("points request failed: %v", err)
|
||||
}
|
||||
defer res.Body.Close()
|
||||
|
||||
if res.StatusCode != http.StatusOK {
|
||||
body, _ := io.ReadAll(res.Body)
|
||||
t.Fatalf("unexpected status: got %d want %d, body=%s", res.StatusCode, http.StatusOK, string(body))
|
||||
}
|
||||
|
||||
assertRecoveryIntegrationColumnExists(t, dbPath, "item_type")
|
||||
|
||||
var payload struct {
|
||||
Data []struct {
|
||||
ID string `json:"id"`
|
||||
Display struct {
|
||||
SubjectLabel string `json:"subjectLabel"`
|
||||
ItemType string `json:"itemType"`
|
||||
} `json:"display"`
|
||||
} `json:"data"`
|
||||
}
|
||||
body, err := io.ReadAll(res.Body)
|
||||
if err != nil {
|
||||
t.Fatalf("read points response: %v", err)
|
||||
}
|
||||
if err := json.Unmarshal(body, &payload); err != nil {
|
||||
t.Fatalf("unmarshal points response: %v", err)
|
||||
}
|
||||
if len(payload.Data) != 1 {
|
||||
t.Fatalf("expected exactly 1 point, got %d body=%s", len(payload.Data), string(body))
|
||||
}
|
||||
if payload.Data[0].ID != "legacy-point-no-item-type" {
|
||||
t.Fatalf("point id = %q, want %q", payload.Data[0].ID, "legacy-point-no-item-type")
|
||||
}
|
||||
if payload.Data[0].Display.SubjectLabel != "default/data" {
|
||||
t.Fatalf("display.subjectLabel = %q, want %q", payload.Data[0].Display.SubjectLabel, "default/data")
|
||||
}
|
||||
if payload.Data[0].Display.ItemType != "pvc" {
|
||||
t.Fatalf("display.itemType = %q, want %q", payload.Data[0].Display.ItemType, "pvc")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecoveryRollupsEndpointMigratesLegacySchemaMissingItemType(t *testing.T) {
|
||||
srv := newIntegrationServerWithoutMock(t, nil)
|
||||
|
||||
dbPath := seedLegacyRecoveryPointWithoutItemTypeForIntegration(t, srv.config, recovery.RecoveryPoint{
|
||||
ID: "legacy-rollup-no-item-type",
|
||||
Provider: recovery.ProviderTrueNAS,
|
||||
Kind: recovery.KindSnapshot,
|
||||
Mode: recovery.ModeSnapshot,
|
||||
Outcome: recovery.OutcomeSuccess,
|
||||
StartedAt: timePtrIntegration(time.Date(2026, 2, 25, 12, 0, 0, 0, time.UTC)),
|
||||
CompletedAt: timePtrIntegration(time.Date(2026, 2, 25, 12, 0, 0, 0, time.UTC)),
|
||||
SubjectRef: &recovery.ExternalRef{
|
||||
Type: "truenas-dataset",
|
||||
Name: "tank/apps",
|
||||
ID: "tank/apps",
|
||||
},
|
||||
})
|
||||
|
||||
res, err := http.Get(srv.server.URL + "/api/recovery/rollups?limit=500")
|
||||
if err != nil {
|
||||
t.Fatalf("rollups request failed: %v", err)
|
||||
}
|
||||
defer res.Body.Close()
|
||||
|
||||
if res.StatusCode != http.StatusOK {
|
||||
body, _ := io.ReadAll(res.Body)
|
||||
t.Fatalf("unexpected status: got %d want %d, body=%s", res.StatusCode, http.StatusOK, string(body))
|
||||
}
|
||||
|
||||
assertRecoveryIntegrationColumnExists(t, dbPath, "item_type")
|
||||
|
||||
var payload struct {
|
||||
Data []struct {
|
||||
RollupID string `json:"rollupId"`
|
||||
Display struct {
|
||||
SubjectLabel string `json:"subjectLabel"`
|
||||
ItemType string `json:"itemType"`
|
||||
} `json:"display"`
|
||||
} `json:"data"`
|
||||
}
|
||||
body, err := io.ReadAll(res.Body)
|
||||
if err != nil {
|
||||
t.Fatalf("read rollups response: %v", err)
|
||||
}
|
||||
if err := json.Unmarshal(body, &payload); err != nil {
|
||||
t.Fatalf("unmarshal rollups response: %v", err)
|
||||
}
|
||||
if len(payload.Data) != 1 {
|
||||
t.Fatalf("expected exactly 1 rollup, got %d body=%s", len(payload.Data), string(body))
|
||||
}
|
||||
if payload.Data[0].RollupID == "" {
|
||||
t.Fatalf("expected non-empty rollup id body=%s", string(body))
|
||||
}
|
||||
if payload.Data[0].Display.SubjectLabel != "tank/apps" {
|
||||
t.Fatalf("display.subjectLabel = %q, want %q", payload.Data[0].Display.SubjectLabel, "tank/apps")
|
||||
}
|
||||
if payload.Data[0].Display.ItemType != "dataset" {
|
||||
t.Fatalf("display.itemType = %q, want %q", payload.Data[0].Display.ItemType, "dataset")
|
||||
}
|
||||
}
|
||||
|
||||
func seedRecoveryPointForIntegration(t *testing.T, cfg *config.Config, point recovery.RecoveryPoint) string {
|
||||
t.Helper()
|
||||
|
||||
@@ -302,6 +431,246 @@ func seedRecoveryPointForIntegration(t *testing.T, cfg *config.Config, point rec
|
||||
return filepath.Join(cfg.DataPath, "recovery", "recovery.db")
|
||||
}
|
||||
|
||||
func seedLegacyRecoveryPointWithoutItemTypeForIntegration(t *testing.T, cfg *config.Config, point recovery.RecoveryPoint) string {
|
||||
t.Helper()
|
||||
|
||||
dbPath := filepath.Join(cfg.DataPath, "recovery", "recovery.db")
|
||||
if err := os.MkdirAll(filepath.Dir(dbPath), 0o700); err != nil {
|
||||
t.Fatalf("MkdirAll(recovery dir): %v", err)
|
||||
}
|
||||
|
||||
db, err := sql.Open("sqlite", dbPath)
|
||||
if err != nil {
|
||||
t.Fatalf("sql.Open(%q): %v", dbPath, err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
schema := `
|
||||
CREATE TABLE recovery_points (
|
||||
id TEXT PRIMARY KEY,
|
||||
provider TEXT NOT NULL,
|
||||
kind TEXT NOT NULL,
|
||||
mode TEXT NOT NULL,
|
||||
outcome TEXT NOT NULL,
|
||||
started_at_ms INTEGER,
|
||||
completed_at_ms INTEGER,
|
||||
size_bytes INTEGER,
|
||||
verified INTEGER,
|
||||
encrypted INTEGER,
|
||||
immutable INTEGER,
|
||||
subject_key TEXT,
|
||||
repository_key TEXT,
|
||||
subject_resource_id TEXT,
|
||||
repository_resource_id TEXT,
|
||||
subject_ref_json TEXT,
|
||||
repository_ref_json TEXT,
|
||||
details_json TEXT,
|
||||
subject_label TEXT,
|
||||
subject_type TEXT,
|
||||
is_workload INTEGER,
|
||||
cluster_label TEXT,
|
||||
node_host_label TEXT,
|
||||
namespace_label TEXT,
|
||||
entity_id_label TEXT,
|
||||
repository_label TEXT,
|
||||
details_summary TEXT,
|
||||
created_at_ms INTEGER NOT NULL,
|
||||
updated_at_ms INTEGER NOT NULL
|
||||
);
|
||||
|
||||
CREATE INDEX idx_recovery_points_completed
|
||||
ON recovery_points(completed_at_ms);
|
||||
|
||||
CREATE INDEX idx_recovery_points_provider_completed
|
||||
ON recovery_points(provider, completed_at_ms);
|
||||
|
||||
CREATE INDEX idx_recovery_points_subject_completed
|
||||
ON recovery_points(subject_resource_id, completed_at_ms);
|
||||
|
||||
CREATE INDEX idx_recovery_points_subject_key_completed
|
||||
ON recovery_points(subject_key, completed_at_ms);
|
||||
|
||||
CREATE INDEX idx_recovery_points_cluster_completed
|
||||
ON recovery_points(cluster_label, completed_at_ms);
|
||||
|
||||
CREATE INDEX idx_recovery_points_node_completed
|
||||
ON recovery_points(node_host_label, completed_at_ms);
|
||||
|
||||
CREATE INDEX idx_recovery_points_namespace_completed
|
||||
ON recovery_points(namespace_label, completed_at_ms);
|
||||
`
|
||||
if _, err := db.Exec(schema); err != nil {
|
||||
t.Fatalf("create legacy recovery schema: %v", err)
|
||||
}
|
||||
|
||||
var (
|
||||
subjectRefJSON sql.NullString
|
||||
repositoryRefJSON sql.NullString
|
||||
detailsJSON sql.NullString
|
||||
sizeBytes sql.NullInt64
|
||||
verified sql.NullInt64
|
||||
encrypted sql.NullInt64
|
||||
immutable sql.NullInt64
|
||||
subjectRID sql.NullString
|
||||
repositoryRID sql.NullString
|
||||
startedAtMs sql.NullInt64
|
||||
completedAtMs sql.NullInt64
|
||||
)
|
||||
|
||||
if point.SubjectRef != nil {
|
||||
data, err := json.Marshal(point.SubjectRef)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal subject ref: %v", err)
|
||||
}
|
||||
subjectRefJSON = sql.NullString{String: string(data), Valid: true}
|
||||
}
|
||||
if point.RepositoryRef != nil {
|
||||
data, err := json.Marshal(point.RepositoryRef)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal repository ref: %v", err)
|
||||
}
|
||||
repositoryRefJSON = sql.NullString{String: string(data), Valid: true}
|
||||
}
|
||||
if len(point.Details) > 0 {
|
||||
data, err := json.Marshal(point.Details)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal details: %v", err)
|
||||
}
|
||||
detailsJSON = sql.NullString{String: string(data), Valid: true}
|
||||
}
|
||||
if point.SizeBytes != nil {
|
||||
sizeBytes = sql.NullInt64{Int64: *point.SizeBytes, Valid: true}
|
||||
}
|
||||
if point.Verified != nil {
|
||||
if *point.Verified {
|
||||
verified = sql.NullInt64{Int64: 1, Valid: true}
|
||||
} else {
|
||||
verified = sql.NullInt64{Int64: 0, Valid: true}
|
||||
}
|
||||
}
|
||||
if point.Encrypted != nil {
|
||||
if *point.Encrypted {
|
||||
encrypted = sql.NullInt64{Int64: 1, Valid: true}
|
||||
} else {
|
||||
encrypted = sql.NullInt64{Int64: 0, Valid: true}
|
||||
}
|
||||
}
|
||||
if point.Immutable != nil {
|
||||
if *point.Immutable {
|
||||
immutable = sql.NullInt64{Int64: 1, Valid: true}
|
||||
} else {
|
||||
immutable = sql.NullInt64{Int64: 0, Valid: true}
|
||||
}
|
||||
}
|
||||
if point.SubjectResourceID != "" {
|
||||
subjectRID = sql.NullString{String: point.SubjectResourceID, Valid: true}
|
||||
}
|
||||
if point.RepositoryResourceID != "" {
|
||||
repositoryRID = sql.NullString{String: point.RepositoryResourceID, Valid: true}
|
||||
}
|
||||
if point.StartedAt != nil {
|
||||
startedAtMs = sql.NullInt64{Int64: point.StartedAt.UTC().UnixMilli(), Valid: true}
|
||||
}
|
||||
if point.CompletedAt != nil {
|
||||
completedAtMs = sql.NullInt64{Int64: point.CompletedAt.UTC().UnixMilli(), Valid: true}
|
||||
}
|
||||
|
||||
idx := recovery.DeriveIndex(point)
|
||||
isWorkload := 0
|
||||
if idx.IsWorkload {
|
||||
isWorkload = 1
|
||||
}
|
||||
|
||||
createdAtMs := time.Date(2026, 2, 26, 12, 0, 0, 0, time.UTC).UnixMilli()
|
||||
updatedAtMs := createdAtMs
|
||||
|
||||
if _, err := db.ExecContext(context.Background(), `
|
||||
INSERT INTO recovery_points (
|
||||
id, provider, kind, mode, outcome,
|
||||
started_at_ms, completed_at_ms, size_bytes,
|
||||
verified, encrypted, immutable,
|
||||
subject_key, repository_key,
|
||||
subject_resource_id, repository_resource_id,
|
||||
subject_ref_json, repository_ref_json, details_json,
|
||||
subject_label, subject_type, is_workload,
|
||||
cluster_label, node_host_label, namespace_label, entity_id_label,
|
||||
repository_label, details_summary,
|
||||
created_at_ms, updated_at_ms
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
`,
|
||||
point.ID,
|
||||
string(point.Provider),
|
||||
string(point.Kind),
|
||||
string(point.Mode),
|
||||
string(point.Outcome),
|
||||
startedAtMs,
|
||||
completedAtMs,
|
||||
sizeBytes,
|
||||
verified,
|
||||
encrypted,
|
||||
immutable,
|
||||
recovery.SubjectKeyForPoint(point),
|
||||
nil,
|
||||
subjectRID,
|
||||
repositoryRID,
|
||||
subjectRefJSON,
|
||||
repositoryRefJSON,
|
||||
detailsJSON,
|
||||
idx.SubjectLabel,
|
||||
idx.SubjectType,
|
||||
isWorkload,
|
||||
idx.ClusterLabel,
|
||||
idx.NodeHostLabel,
|
||||
idx.NamespaceLabel,
|
||||
idx.EntityIDLabel,
|
||||
idx.RepositoryLabel,
|
||||
idx.DetailsSummary,
|
||||
createdAtMs,
|
||||
updatedAtMs,
|
||||
); err != nil {
|
||||
t.Fatalf("insert legacy recovery point: %v", err)
|
||||
}
|
||||
|
||||
return dbPath
|
||||
}
|
||||
|
||||
func assertRecoveryIntegrationColumnExists(t *testing.T, dbPath string, column string) {
|
||||
t.Helper()
|
||||
|
||||
db, err := sql.Open("sqlite", dbPath)
|
||||
if err != nil {
|
||||
t.Fatalf("sql.Open(%q): %v", dbPath, err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
rows, err := db.QueryContext(context.Background(), `PRAGMA table_info(recovery_points)`)
|
||||
if err != nil {
|
||||
t.Fatalf("PRAGMA table_info(recovery_points): %v", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
for rows.Next() {
|
||||
var (
|
||||
cid int
|
||||
name string
|
||||
colType string
|
||||
notNull int
|
||||
defaultVal sql.NullString
|
||||
primaryKey int
|
||||
)
|
||||
if err := rows.Scan(&cid, &name, &colType, ¬Null, &defaultVal, &primaryKey); err != nil {
|
||||
t.Fatalf("scan table_info row: %v", err)
|
||||
}
|
||||
if name == column {
|
||||
return
|
||||
}
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
t.Fatalf("table_info rows err: %v", err)
|
||||
}
|
||||
t.Fatalf("expected recovery_points column %q to exist after migration", column)
|
||||
}
|
||||
|
||||
func corruptRecoveryIntegrationRowJSON(t *testing.T, dbPath string, rowID string, corruptSubject bool, corruptRepository bool, corruptDetails bool) {
|
||||
t.Helper()
|
||||
|
||||
|
||||
@@ -223,9 +223,6 @@ func (s *Store) initSchema() error {
|
||||
CREATE INDEX IF NOT EXISTS idx_recovery_points_provider_completed
|
||||
ON recovery_points(provider, completed_at_ms);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_recovery_points_item_type_completed
|
||||
ON recovery_points(item_type, completed_at_ms);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_recovery_points_subject_completed
|
||||
ON recovery_points(subject_resource_id, completed_at_ms);
|
||||
|
||||
@@ -268,6 +265,9 @@ func (s *Store) initSchema() error {
|
||||
// Indexes on migrated columns — must run AFTER ensureColumn so pre-existing
|
||||
// databases that lack these columns don't crash with "no such column".
|
||||
postMigrationIndexes := `
|
||||
CREATE INDEX IF NOT EXISTS idx_recovery_points_item_type_completed
|
||||
ON recovery_points(item_type, completed_at_ms);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_recovery_points_cluster_completed
|
||||
ON recovery_points(cluster_label, completed_at_ms);
|
||||
|
||||
|
||||
@@ -2,6 +2,8 @@ package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -139,3 +141,298 @@ func TestStore_ListPoints_IgnoresMalformedJSONFields(t *testing.T) {
|
||||
t.Fatalf("Display = %#v, want preserved normalized item type pvc", got[0].Display)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStore_OpenMigratesLegacySchemaMissingItemType(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
dir := t.TempDir()
|
||||
dbPath := filepath.Join(dir, "recovery.db")
|
||||
|
||||
now := time.Date(2026, 2, 24, 12, 0, 0, 0, time.UTC)
|
||||
legacyPoint := recovery.RecoveryPoint{
|
||||
ID: "legacy-point-no-item-type",
|
||||
Provider: recovery.ProviderKubernetes,
|
||||
Kind: recovery.KindSnapshot,
|
||||
Mode: recovery.ModeSnapshot,
|
||||
Outcome: recovery.OutcomeSuccess,
|
||||
StartedAt: &now,
|
||||
CompletedAt: &now,
|
||||
SubjectRef: &recovery.ExternalRef{
|
||||
Type: "k8s-pvc",
|
||||
Namespace: "default",
|
||||
Name: "data",
|
||||
},
|
||||
RepositoryRef: &recovery.ExternalRef{
|
||||
Type: "velero-backup-storage-location",
|
||||
Name: "repo-a",
|
||||
},
|
||||
Details: map[string]any{
|
||||
"foo": "bar",
|
||||
},
|
||||
}
|
||||
|
||||
createLegacyRecoveryDBWithoutItemType(t, dbPath, legacyPoint)
|
||||
|
||||
store, err := Open(dbPath)
|
||||
if err != nil {
|
||||
t.Fatalf("Open() error = %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = store.Close() })
|
||||
|
||||
assertRecoveryColumnExists(t, dbPath, "item_type")
|
||||
|
||||
points, total, err := store.ListPoints(context.Background(), recovery.ListPointsOptions{Page: 1, Limit: 50})
|
||||
if err != nil {
|
||||
t.Fatalf("ListPoints() error = %v", err)
|
||||
}
|
||||
if total != 1 || len(points) != 1 {
|
||||
t.Fatalf("ListPoints() total=%d len=%d, want 1/1", total, len(points))
|
||||
}
|
||||
if points[0].Display == nil || points[0].Display.ItemType != "pvc" {
|
||||
t.Fatalf("ListPoints() display = %#v, want item type pvc", points[0].Display)
|
||||
}
|
||||
|
||||
rollups, rollupTotal, err := store.ListRollups(context.Background(), recovery.ListPointsOptions{Page: 1, Limit: 50})
|
||||
if err != nil {
|
||||
t.Fatalf("ListRollups() error = %v", err)
|
||||
}
|
||||
if rollupTotal != 1 || len(rollups) != 1 {
|
||||
t.Fatalf("ListRollups() total=%d len=%d, want 1/1", rollupTotal, len(rollups))
|
||||
}
|
||||
if rollups[0].Display == nil || rollups[0].Display.ItemType != "pvc" {
|
||||
t.Fatalf("ListRollups() display = %#v, want item type pvc", rollups[0].Display)
|
||||
}
|
||||
}
|
||||
|
||||
func createLegacyRecoveryDBWithoutItemType(t *testing.T, dbPath string, point recovery.RecoveryPoint) {
|
||||
t.Helper()
|
||||
|
||||
db, err := sql.Open("sqlite", dbPath)
|
||||
if err != nil {
|
||||
t.Fatalf("sql.Open(%q): %v", dbPath, err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
schema := `
|
||||
CREATE TABLE recovery_points (
|
||||
id TEXT PRIMARY KEY,
|
||||
provider TEXT NOT NULL,
|
||||
kind TEXT NOT NULL,
|
||||
mode TEXT NOT NULL,
|
||||
outcome TEXT NOT NULL,
|
||||
started_at_ms INTEGER,
|
||||
completed_at_ms INTEGER,
|
||||
size_bytes INTEGER,
|
||||
verified INTEGER,
|
||||
encrypted INTEGER,
|
||||
immutable INTEGER,
|
||||
subject_key TEXT,
|
||||
repository_key TEXT,
|
||||
subject_resource_id TEXT,
|
||||
repository_resource_id TEXT,
|
||||
subject_ref_json TEXT,
|
||||
repository_ref_json TEXT,
|
||||
details_json TEXT,
|
||||
subject_label TEXT,
|
||||
subject_type TEXT,
|
||||
is_workload INTEGER,
|
||||
cluster_label TEXT,
|
||||
node_host_label TEXT,
|
||||
namespace_label TEXT,
|
||||
entity_id_label TEXT,
|
||||
repository_label TEXT,
|
||||
details_summary TEXT,
|
||||
created_at_ms INTEGER NOT NULL,
|
||||
updated_at_ms INTEGER NOT NULL
|
||||
);
|
||||
|
||||
CREATE INDEX idx_recovery_points_completed
|
||||
ON recovery_points(completed_at_ms);
|
||||
|
||||
CREATE INDEX idx_recovery_points_provider_completed
|
||||
ON recovery_points(provider, completed_at_ms);
|
||||
|
||||
CREATE INDEX idx_recovery_points_subject_completed
|
||||
ON recovery_points(subject_resource_id, completed_at_ms);
|
||||
|
||||
CREATE INDEX idx_recovery_points_subject_key_completed
|
||||
ON recovery_points(subject_key, completed_at_ms);
|
||||
|
||||
CREATE INDEX idx_recovery_points_cluster_completed
|
||||
ON recovery_points(cluster_label, completed_at_ms);
|
||||
|
||||
CREATE INDEX idx_recovery_points_node_completed
|
||||
ON recovery_points(node_host_label, completed_at_ms);
|
||||
|
||||
CREATE INDEX idx_recovery_points_namespace_completed
|
||||
ON recovery_points(namespace_label, completed_at_ms);
|
||||
`
|
||||
if _, err := db.Exec(schema); err != nil {
|
||||
t.Fatalf("create legacy recovery schema: %v", err)
|
||||
}
|
||||
|
||||
var (
|
||||
subjectRefJSON sql.NullString
|
||||
repositoryRefJSON sql.NullString
|
||||
detailsJSON sql.NullString
|
||||
sizeBytes sql.NullInt64
|
||||
verified sql.NullInt64
|
||||
encrypted sql.NullInt64
|
||||
immutable sql.NullInt64
|
||||
subjectRID sql.NullString
|
||||
repositoryRID sql.NullString
|
||||
startedAtMs sql.NullInt64
|
||||
completedAtMs sql.NullInt64
|
||||
)
|
||||
|
||||
if point.SubjectRef != nil {
|
||||
data, err := json.Marshal(point.SubjectRef)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal subject ref: %v", err)
|
||||
}
|
||||
subjectRefJSON = sql.NullString{String: string(data), Valid: true}
|
||||
}
|
||||
if point.RepositoryRef != nil {
|
||||
data, err := json.Marshal(point.RepositoryRef)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal repository ref: %v", err)
|
||||
}
|
||||
repositoryRefJSON = sql.NullString{String: string(data), Valid: true}
|
||||
}
|
||||
if len(point.Details) > 0 {
|
||||
data, err := json.Marshal(point.Details)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal details: %v", err)
|
||||
}
|
||||
detailsJSON = sql.NullString{String: string(data), Valid: true}
|
||||
}
|
||||
if point.SizeBytes != nil {
|
||||
sizeBytes = sql.NullInt64{Int64: *point.SizeBytes, Valid: true}
|
||||
}
|
||||
if point.Verified != nil {
|
||||
if *point.Verified {
|
||||
verified = sql.NullInt64{Int64: 1, Valid: true}
|
||||
} else {
|
||||
verified = sql.NullInt64{Int64: 0, Valid: true}
|
||||
}
|
||||
}
|
||||
if point.Encrypted != nil {
|
||||
if *point.Encrypted {
|
||||
encrypted = sql.NullInt64{Int64: 1, Valid: true}
|
||||
} else {
|
||||
encrypted = sql.NullInt64{Int64: 0, Valid: true}
|
||||
}
|
||||
}
|
||||
if point.Immutable != nil {
|
||||
if *point.Immutable {
|
||||
immutable = sql.NullInt64{Int64: 1, Valid: true}
|
||||
} else {
|
||||
immutable = sql.NullInt64{Int64: 0, Valid: true}
|
||||
}
|
||||
}
|
||||
if point.SubjectResourceID != "" {
|
||||
subjectRID = sql.NullString{String: point.SubjectResourceID, Valid: true}
|
||||
}
|
||||
if point.RepositoryResourceID != "" {
|
||||
repositoryRID = sql.NullString{String: point.RepositoryResourceID, Valid: true}
|
||||
}
|
||||
if point.StartedAt != nil {
|
||||
startedAtMs = sql.NullInt64{Int64: point.StartedAt.UTC().UnixMilli(), Valid: true}
|
||||
}
|
||||
if point.CompletedAt != nil {
|
||||
completedAtMs = sql.NullInt64{Int64: point.CompletedAt.UTC().UnixMilli(), Valid: true}
|
||||
}
|
||||
|
||||
idx := recovery.DeriveIndex(point)
|
||||
isWorkload := 0
|
||||
if idx.IsWorkload {
|
||||
isWorkload = 1
|
||||
}
|
||||
|
||||
createdAtMs := time.Date(2026, 2, 26, 12, 0, 0, 0, time.UTC).UnixMilli()
|
||||
updatedAtMs := createdAtMs
|
||||
|
||||
if _, err := db.ExecContext(context.Background(), `
|
||||
INSERT INTO recovery_points (
|
||||
id, provider, kind, mode, outcome,
|
||||
started_at_ms, completed_at_ms, size_bytes,
|
||||
verified, encrypted, immutable,
|
||||
subject_key, repository_key,
|
||||
subject_resource_id, repository_resource_id,
|
||||
subject_ref_json, repository_ref_json, details_json,
|
||||
subject_label, subject_type, is_workload,
|
||||
cluster_label, node_host_label, namespace_label, entity_id_label,
|
||||
repository_label, details_summary,
|
||||
created_at_ms, updated_at_ms
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
`,
|
||||
point.ID,
|
||||
string(point.Provider),
|
||||
string(point.Kind),
|
||||
string(point.Mode),
|
||||
string(point.Outcome),
|
||||
startedAtMs,
|
||||
completedAtMs,
|
||||
sizeBytes,
|
||||
verified,
|
||||
encrypted,
|
||||
immutable,
|
||||
recovery.SubjectKeyForPoint(point),
|
||||
nil,
|
||||
subjectRID,
|
||||
repositoryRID,
|
||||
subjectRefJSON,
|
||||
repositoryRefJSON,
|
||||
detailsJSON,
|
||||
idx.SubjectLabel,
|
||||
idx.SubjectType,
|
||||
isWorkload,
|
||||
idx.ClusterLabel,
|
||||
idx.NodeHostLabel,
|
||||
idx.NamespaceLabel,
|
||||
idx.EntityIDLabel,
|
||||
idx.RepositoryLabel,
|
||||
idx.DetailsSummary,
|
||||
createdAtMs,
|
||||
updatedAtMs,
|
||||
); err != nil {
|
||||
t.Fatalf("insert legacy recovery point: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func assertRecoveryColumnExists(t *testing.T, dbPath string, column string) {
|
||||
t.Helper()
|
||||
|
||||
db, err := sql.Open("sqlite", dbPath)
|
||||
if err != nil {
|
||||
t.Fatalf("sql.Open(%q): %v", dbPath, err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
rows, err := db.QueryContext(context.Background(), `PRAGMA table_info(recovery_points)`)
|
||||
if err != nil {
|
||||
t.Fatalf("PRAGMA table_info(recovery_points): %v", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
for rows.Next() {
|
||||
var (
|
||||
cid int
|
||||
name string
|
||||
colType string
|
||||
notNull int
|
||||
defaultVal sql.NullString
|
||||
primaryKey int
|
||||
)
|
||||
if err := rows.Scan(&cid, &name, &colType, ¬Null, &defaultVal, &primaryKey); err != nil {
|
||||
t.Fatalf("scan table_info row: %v", err)
|
||||
}
|
||||
if name == column {
|
||||
return
|
||||
}
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
t.Fatalf("table_info rows err: %v", err)
|
||||
}
|
||||
t.Fatalf("expected recovery_points column %q to exist after migration", column)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user