| 1 | package topicstate |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "net/url" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "runtime" |
| 12 | "sync" |
| 13 | "testing" |
| 14 | "time" |
| 15 | |
| 16 | "reasonix/internal/sqliteuri" |
| 17 | |
| 18 | _ "modernc.org/sqlite" |
| 19 | ) |
| 20 | |
| 21 | func TestStorePersistsAtomicTopicRecord(t *testing.T) { |
| 22 | ctx := context.Background() |
| 23 | path := filepath.Join(t.TempDir(), "topic-state-v1.sqlite") |
| 24 | store, err := Open(ctx, path) |
| 25 | if err != nil { |
| 26 | t.Fatal(err) |
| 27 | } |
| 28 | state, err := store.Update(ctx, "topic-1", func(record *Record) { |
| 29 | record.Title = "First title" |
| 30 | record.TitleSource = "manual" |
| 31 | record.CreatedAtMS = 123 |
| 32 | record.AutoMeta = json.RawMessage(`{"stage":2,"future":"kept"}`) |
| 33 | }) |
| 34 | if err != nil { |
| 35 | t.Fatal(err) |
| 36 | } |
| 37 | if state.Revision != 1 || state.LegacyPendingRevision != 0 { |
| 38 | t.Fatalf("state = %+v", state) |
| 39 | } |
| 40 | if err := store.Close(); err != nil { |
| 41 | t.Fatal(err) |
| 42 | } |
| 43 | |
| 44 | store, err = Open(ctx, path) |
| 45 | if err != nil { |
| 46 | t.Fatal(err) |
| 47 | } |
| 48 | defer store.Close() |
| 49 | snapshot, err := store.Snapshot(ctx) |
| 50 | if err != nil { |
| 51 | t.Fatal(err) |
| 52 | } |
| 53 | record := snapshot.Records["topic-1"] |
| 54 | if record.Title != "First title" || record.TitleSource != "manual" || record.CreatedAtMS != 123 { |
| 55 | t.Fatalf("record = %+v", record) |
| 56 | } |
| 57 | var meta map[string]any |
| 58 | if err := json.Unmarshal(record.AutoMeta, &meta); err != nil { |
| 59 | t.Fatal(err) |
| 60 | } |
| 61 | if meta["future"] != "kept" { |
| 62 | t.Fatalf("auto meta = %s", record.AutoMeta) |
| 63 | } |
| 64 | if info, err := os.Stat(path); err != nil { |
| 65 | t.Fatal(err) |
| 66 | } else if runtime.GOOS != "windows" && info.Mode().Perm() != 0o600 { |
| 67 | t.Fatalf("database mode = %o, want 600", info.Mode().Perm()) |
| 68 | } |
| 69 | } |
| 70 | |
| 71 | func TestStoreLegacyOutboxTracksCommittedRevision(t *testing.T) { |
| 72 | ctx := context.Background() |
| 73 | store, err := Open(ctx, filepath.Join(t.TempDir(), "topic-state-v1.sqlite")) |
| 74 | if err != nil { |
| 75 | t.Fatal(err) |
| 76 | } |
| 77 | defer store.Close() |
| 78 | |
| 79 | state, err := store.SetLegacyBridge(ctx) |
| 80 | if err != nil { |
| 81 | t.Fatal(err) |
| 82 | } |
| 83 | if !state.LegacyBridge { |
| 84 | t.Fatal("legacy bridge was not enabled") |
| 85 | } |
| 86 | state, err = store.Update(ctx, "topic-1", func(record *Record) { record.Title = "renamed" }) |
| 87 | if err != nil { |
| 88 | t.Fatal(err) |
| 89 | } |
| 90 | if state.LegacyPendingRevision != state.Revision || state.Revision == 0 { |
| 91 | t.Fatalf("state = %+v", state) |
| 92 | } |
| 93 | digests := [4]string{"titles", "sources", "created", "auto"} |
| 94 | state, err = store.MarkLegacyExported(ctx, state.Revision, digests) |
| 95 | if err != nil { |
| 96 | t.Fatal(err) |
| 97 | } |
| 98 | if state.LegacyPendingRevision != 0 || state.LegacyExportedRevision != state.Revision { |
| 99 | t.Fatalf("state = %+v", state) |
| 100 | } |
| 101 | if state.LegacyTitlesDigest != "titles" || state.LegacyAutoMetaDigest != "auto" { |
| 102 | t.Fatalf("digests not retained: %+v", state) |
| 103 | } |
| 104 | } |
| 105 | |
| 106 | func TestReplaceFieldPreservesUnknownAutoMetadata(t *testing.T) { |
| 107 | ctx := context.Background() |
| 108 | store, err := Open(ctx, filepath.Join(t.TempDir(), "topic-state-v1.sqlite")) |
| 109 | if err != nil { |
| 110 | t.Fatal(err) |
| 111 | } |
| 112 | defer store.Close() |
| 113 | _, err = store.Update(ctx, "topic-1", func(record *Record) { |
| 114 | record.Title = "old" |
| 115 | record.AutoMeta = json.RawMessage(`{"stage":1,"future":{"value":7}}`) |
| 116 | }) |
| 117 | if err != nil { |
| 118 | t.Fatal(err) |
| 119 | } |
| 120 | if _, err := store.ReplaceTitles(ctx, map[string]string{"topic-1": "new"}); err != nil { |
| 121 | t.Fatal(err) |
| 122 | } |
| 123 | snapshot, err := store.Snapshot(ctx) |
| 124 | if err != nil { |
| 125 | t.Fatal(err) |
| 126 | } |
| 127 | if got := string(snapshot.Records["topic-1"].AutoMeta); got != `{"stage":1,"future":{"value":7}}` { |
| 128 | t.Fatalf("auto metadata changed: %s", got) |
| 129 | } |
| 130 | } |
| 131 | |
| 132 | func TestMergeMissingTitleIndexDoesNotOverwriteNewerRename(t *testing.T) { |
| 133 | ctx := context.Background() |
| 134 | store, err := Open(ctx, filepath.Join(t.TempDir(), "topic-state-v1.sqlite")) |
| 135 | if err != nil { |
| 136 | t.Fatal(err) |
| 137 | } |
| 138 | defer store.Close() |
| 139 | |
| 140 | if _, err := store.Update(ctx, "topic-1", func(record *Record) { |
| 141 | record.Title = "New manual title" |
| 142 | record.TitleSource = "manual" |
| 143 | }); err != nil { |
| 144 | t.Fatal(err) |
| 145 | } |
| 146 | before, err := store.Snapshot(ctx) |
| 147 | if err != nil { |
| 148 | t.Fatal(err) |
| 149 | } |
| 150 | if _, err := store.MergeMissingTitleIndex(ctx, |
| 151 | map[string]string{"topic-1": "Stale repaired title", "topic-2": "Recovered title"}, |
| 152 | map[string]string{"topic-1": "auto", "topic-2": "manual"}, nil); err != nil { |
| 153 | t.Fatal(err) |
| 154 | } |
| 155 | after, err := store.Snapshot(ctx) |
| 156 | if err != nil { |
| 157 | t.Fatal(err) |
| 158 | } |
| 159 | if got := after.Records["topic-1"]; got.Title != "New manual title" || got.TitleSource != "manual" { |
| 160 | t.Fatalf("newer rename was overwritten: %+v", got) |
| 161 | } |
| 162 | if got := after.Records["topic-2"]; got.Title != "Recovered title" || got.TitleSource != "manual" { |
| 163 | t.Fatalf("missing topic was not repaired: %+v", got) |
| 164 | } |
| 165 | if after.State.Revision != before.State.Revision+1 { |
| 166 | t.Fatalf("revision = %d, want %d", after.State.Revision, before.State.Revision+1) |
| 167 | } |
| 168 | } |
| 169 | |
| 170 | func TestStoreRejectsFutureSchemaWithoutChangingFile(t *testing.T) { |
| 171 | ctx := context.Background() |
| 172 | path := filepath.Join(t.TempDir(), "topic-state-v1.sqlite") |
| 173 | dsn, err := sqliteuri.Disk(path, url.Values{"_pragma": {"busy_timeout(2000)", "foreign_keys(1)"}}) |
| 174 | if err != nil { |
| 175 | t.Fatal(err) |
| 176 | } |
| 177 | db, err := sql.Open("sqlite", dsn) |
| 178 | if err != nil { |
| 179 | t.Fatal(err) |
| 180 | } |
| 181 | if _, err := db.Exec(`CREATE TABLE schema_migrations(version INTEGER PRIMARY KEY, applied_at INTEGER NOT NULL)`); err != nil { |
| 182 | t.Fatal(err) |
| 183 | } |
| 184 | if _, err := db.Exec(`INSERT INTO schema_migrations(version, applied_at) VALUES(2, 0)`); err != nil { |
| 185 | t.Fatal(err) |
| 186 | } |
| 187 | if err := db.Close(); err != nil { |
| 188 | t.Fatal(err) |
| 189 | } |
| 190 | before, err := os.ReadFile(path) |
| 191 | if err != nil { |
| 192 | t.Fatal(err) |
| 193 | } |
| 194 | _, err = Open(ctx, path) |
| 195 | var future *FutureSchemaError |
| 196 | if !errors.As(err, &future) { |
| 197 | t.Fatalf("Open error = %v, want FutureSchemaError", err) |
| 198 | } |
| 199 | after, err := os.ReadFile(path) |
| 200 | if err != nil { |
| 201 | t.Fatal(err) |
| 202 | } |
| 203 | if string(before) != string(after) { |
| 204 | t.Fatal("future-schema database was modified") |
| 205 | } |
| 206 | } |
| 207 | |
| 208 | func TestStoreConcurrentUpdatesAreSerialized(t *testing.T) { |
| 209 | ctx := context.Background() |
| 210 | store, err := Open(ctx, filepath.Join(t.TempDir(), "topic-state-v1.sqlite")) |
| 211 | if err != nil { |
| 212 | t.Fatal(err) |
| 213 | } |
| 214 | defer store.Close() |
| 215 | const writers = 12 |
| 216 | var wg sync.WaitGroup |
| 217 | errs := make(chan error, writers) |
| 218 | for i := range writers { |
| 219 | wg.Add(1) |
| 220 | go func(i int) { |
| 221 | defer wg.Done() |
| 222 | _, err := store.Update(ctx, "topic-"+time.Unix(int64(i), 0).UTC().Format("150405"), func(record *Record) { |
| 223 | record.Title = "title" |
| 224 | }) |
| 225 | errs <- err |
| 226 | }(i) |
| 227 | } |
| 228 | wg.Wait() |
| 229 | close(errs) |
| 230 | for err := range errs { |
| 231 | if err != nil { |
| 232 | t.Fatal(err) |
| 233 | } |
| 234 | } |
| 235 | snapshot, err := store.Snapshot(ctx) |
| 236 | if err != nil { |
| 237 | t.Fatal(err) |
| 238 | } |
| 239 | if len(snapshot.Records) != writers || snapshot.State.Revision != writers { |
| 240 | t.Fatalf("records=%d revision=%d", len(snapshot.Records), snapshot.State.Revision) |
| 241 | } |
| 242 | } |
| 243 |