| 1 | // Package topicstate persists authoritative Desktop topic metadata in SQLite. |
| 2 | // Unlike projection databases, a Store owns user-visible state and must never |
| 3 | // silently fall back to an empty in-memory database. |
| 4 | package topicstate |
| 5 | |
| 6 | import ( |
| 7 | "bytes" |
| 8 | "context" |
| 9 | "database/sql" |
| 10 | "encoding/json" |
| 11 | "errors" |
| 12 | "fmt" |
| 13 | "net/url" |
| 14 | "os" |
| 15 | "path/filepath" |
| 16 | "sort" |
| 17 | "strings" |
| 18 | "time" |
| 19 | |
| 20 | "reasonix/internal/sqliteuri" |
| 21 | |
| 22 | moderncsqlite "modernc.org/sqlite" |
| 23 | sqlite3 "modernc.org/sqlite/lib" |
| 24 | ) |
| 25 | |
| 26 | const schemaVersion = 1 |
| 27 | |
| 28 | // FutureSchemaError means a newer Reasonix version owns this database. Callers |
| 29 | // must leave the file untouched and may only use an explicitly compatible |
| 30 | // legacy read path. |
| 31 | type FutureSchemaError struct { |
| 32 | Found int |
| 33 | Supported int |
| 34 | } |
| 35 | |
| 36 | func (e *FutureSchemaError) Error() string { |
| 37 | return fmt.Sprintf("topic state schema %d is newer than supported %d", e.Found, e.Supported) |
| 38 | } |
| 39 | |
| 40 | // Record is the complete durable metadata for one Desktop topic. AutoMeta is |
| 41 | // retained as raw JSON so a read/modify/write by this version preserves fields |
| 42 | // introduced by a future compatible writer. |
| 43 | type Record struct { |
| 44 | TopicID string |
| 45 | Title string |
| 46 | TitleSource string |
| 47 | CreatedAtMS int64 |
| 48 | AutoMeta json.RawMessage |
| 49 | RowRevision int64 |
| 50 | UpdatedAtMS int64 |
| 51 | } |
| 52 | |
| 53 | // State tracks the authoritative revision and the legacy JSON mirror outbox. |
| 54 | type State struct { |
| 55 | Revision int64 |
| 56 | LegacyBridge bool |
| 57 | LegacyExportedRevision int64 |
| 58 | LegacyPendingRevision int64 |
| 59 | LegacyTitlesDigest string |
| 60 | LegacySourcesDigest string |
| 61 | LegacyCreatedAtsDigest string |
| 62 | LegacyAutoMetaDigest string |
| 63 | } |
| 64 | |
| 65 | // Snapshot is one transactionally consistent view of all topic metadata. |
| 66 | type Snapshot struct { |
| 67 | Records map[string]Record |
| 68 | State State |
| 69 | } |
| 70 | |
| 71 | // Store is a single-scope SQLite topic database. |
| 72 | type Store struct { |
| 73 | path string |
| 74 | db *sql.DB |
| 75 | now func() time.Time |
| 76 | } |
| 77 | |
| 78 | // Open opens or creates a durable topic database at path. |
| 79 | func Open(ctx context.Context, path string) (*Store, error) { |
| 80 | path = strings.TrimSpace(path) |
| 81 | if path == "" { |
| 82 | return nil, errors.New("topic state path is empty") |
| 83 | } |
| 84 | if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { |
| 85 | return nil, fmt.Errorf("create topic state directory: %w", err) |
| 86 | } |
| 87 | _ = os.Chmod(filepath.Dir(path), 0o700) |
| 88 | |
| 89 | dsn, err := sqliteuri.Disk(path, url.Values{ |
| 90 | "_pragma": {"busy_timeout(2000)", "foreign_keys(1)"}, |
| 91 | }) |
| 92 | if err != nil { |
| 93 | return nil, err |
| 94 | } |
| 95 | db, err := sql.Open("sqlite", dsn) |
| 96 | if err != nil { |
| 97 | return nil, err |
| 98 | } |
| 99 | db.SetMaxOpenConns(1) |
| 100 | db.SetMaxIdleConns(1) |
| 101 | fail := func(err error) (*Store, error) { |
| 102 | _ = db.Close() |
| 103 | return nil, err |
| 104 | } |
| 105 | if err := db.PingContext(ctx); err != nil { |
| 106 | return fail(err) |
| 107 | } |
| 108 | for _, pragma := range []string{`PRAGMA foreign_keys=ON`, `PRAGMA busy_timeout=2000`} { |
| 109 | if _, err := db.ExecContext(ctx, pragma); err != nil { |
| 110 | return fail(err) |
| 111 | } |
| 112 | } |
| 113 | var integrity string |
| 114 | if err := db.QueryRowContext(ctx, `PRAGMA quick_check`).Scan(&integrity); err != nil { |
| 115 | return fail(err) |
| 116 | } |
| 117 | if integrity != "ok" { |
| 118 | return fail(fmt.Errorf("topic state quick check: %s", integrity)) |
| 119 | } |
| 120 | if err := applyMigrations(ctx, db, time.Now); err != nil { |
| 121 | return fail(err) |
| 122 | } |
| 123 | if _, err := db.ExecContext(ctx, `PRAGMA journal_mode=WAL`); err != nil { |
| 124 | return fail(err) |
| 125 | } |
| 126 | for _, pragma := range []string{`PRAGMA synchronous=FULL`, `PRAGMA secure_delete=ON`} { |
| 127 | if _, err := db.ExecContext(ctx, pragma); err != nil { |
| 128 | return fail(err) |
| 129 | } |
| 130 | } |
| 131 | _ = os.Chmod(path, 0o600) |
| 132 | _ = os.Chmod(path+"-wal", 0o600) |
| 133 | _ = os.Chmod(path+"-shm", 0o600) |
| 134 | return &Store{path: path, db: db, now: time.Now}, nil |
| 135 | } |
| 136 | |
| 137 | func applyMigrations(ctx context.Context, db *sql.DB, now func() time.Time) error { |
| 138 | if _, err := db.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS schema_migrations ( |
| 139 | version INTEGER PRIMARY KEY, |
| 140 | applied_at INTEGER NOT NULL |
| 141 | )`); err != nil { |
| 142 | return err |
| 143 | } |
| 144 | var current int |
| 145 | if err := db.QueryRowContext(ctx, `SELECT COALESCE(MAX(version), 0) FROM schema_migrations`).Scan(¤t); err != nil { |
| 146 | return err |
| 147 | } |
| 148 | if current > schemaVersion { |
| 149 | return &FutureSchemaError{Found: current, Supported: schemaVersion} |
| 150 | } |
| 151 | if current == schemaVersion { |
| 152 | return nil |
| 153 | } |
| 154 | tx, err := db.BeginTx(ctx, nil) |
| 155 | if err != nil { |
| 156 | return err |
| 157 | } |
| 158 | defer func() { _ = tx.Rollback() }() |
| 159 | if _, err := tx.ExecContext(ctx, `CREATE TABLE topics ( |
| 160 | topic_id TEXT PRIMARY KEY, |
| 161 | title TEXT NOT NULL DEFAULT '', |
| 162 | title_source TEXT NOT NULL DEFAULT '', |
| 163 | created_at_ms INTEGER NOT NULL DEFAULT 0, |
| 164 | auto_meta_json TEXT NOT NULL DEFAULT '', |
| 165 | row_revision INTEGER NOT NULL DEFAULT 0, |
| 166 | updated_at_ms INTEGER NOT NULL DEFAULT 0 |
| 167 | )`); err != nil { |
| 168 | return fmt.Errorf("create topics table: %w", err) |
| 169 | } |
| 170 | if _, err := tx.ExecContext(ctx, `CREATE TABLE store_state ( |
| 171 | id INTEGER PRIMARY KEY CHECK (id = 1), |
| 172 | revision INTEGER NOT NULL DEFAULT 0, |
| 173 | legacy_bridge INTEGER NOT NULL DEFAULT 0, |
| 174 | legacy_exported_revision INTEGER NOT NULL DEFAULT 0, |
| 175 | legacy_pending_revision INTEGER NOT NULL DEFAULT 0, |
| 176 | legacy_titles_digest TEXT NOT NULL DEFAULT '', |
| 177 | legacy_sources_digest TEXT NOT NULL DEFAULT '', |
| 178 | legacy_created_ats_digest TEXT NOT NULL DEFAULT '', |
| 179 | legacy_auto_meta_digest TEXT NOT NULL DEFAULT '' |
| 180 | )`); err != nil { |
| 181 | return fmt.Errorf("create store state table: %w", err) |
| 182 | } |
| 183 | if _, err := tx.ExecContext(ctx, `INSERT INTO store_state(id) VALUES(1)`); err != nil { |
| 184 | return fmt.Errorf("initialize store state: %w", err) |
| 185 | } |
| 186 | if _, err := tx.ExecContext(ctx, `INSERT INTO schema_migrations(version, applied_at) VALUES(?, ?)`, schemaVersion, now().UnixMilli()); err != nil { |
| 187 | return err |
| 188 | } |
| 189 | return tx.Commit() |
| 190 | } |
| 191 | |
| 192 | // Close releases the underlying SQLite handle. |
| 193 | func (s *Store) Close() error { |
| 194 | if s == nil || s.db == nil { |
| 195 | return nil |
| 196 | } |
| 197 | return s.db.Close() |
| 198 | } |
| 199 | |
| 200 | // Path returns the database path for diagnostics and lock coordination. |
| 201 | func (s *Store) Path() string { |
| 202 | if s == nil { |
| 203 | return "" |
| 204 | } |
| 205 | return s.path |
| 206 | } |
| 207 | |
| 208 | // Snapshot returns all records and bridge state from one read transaction. |
| 209 | func (s *Store) Snapshot(ctx context.Context) (Snapshot, error) { |
| 210 | tx, err := s.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true}) |
| 211 | if err != nil { |
| 212 | return Snapshot{}, err |
| 213 | } |
| 214 | defer func() { _ = tx.Rollback() }() |
| 215 | state, err := readState(ctx, tx) |
| 216 | if err != nil { |
| 217 | return Snapshot{}, err |
| 218 | } |
| 219 | records, err := readRecords(ctx, tx) |
| 220 | if err != nil { |
| 221 | return Snapshot{}, err |
| 222 | } |
| 223 | if err := tx.Commit(); err != nil { |
| 224 | return Snapshot{}, err |
| 225 | } |
| 226 | return Snapshot{Records: records, State: state}, nil |
| 227 | } |
| 228 | |
| 229 | // SetLegacyBridge enables the compatibility outbox permanently. Enabling it |
| 230 | // marks the current revision pending so the caller must publish a full mirror. |
| 231 | func (s *Store) SetLegacyBridge(ctx context.Context) (State, error) { |
| 232 | tx, err := s.db.BeginTx(ctx, nil) |
| 233 | if err != nil { |
| 234 | return State{}, err |
| 235 | } |
| 236 | defer func() { _ = tx.Rollback() }() |
| 237 | state, err := readState(ctx, tx) |
| 238 | if err != nil { |
| 239 | return State{}, err |
| 240 | } |
| 241 | if !state.LegacyBridge { |
| 242 | state.LegacyBridge = true |
| 243 | state.LegacyPendingRevision = state.Revision |
| 244 | if _, err := tx.ExecContext(ctx, `UPDATE store_state SET legacy_bridge=1, legacy_pending_revision=? WHERE id=1`, state.LegacyPendingRevision); err != nil { |
| 245 | return State{}, err |
| 246 | } |
| 247 | } |
| 248 | if err := tx.Commit(); err != nil { |
| 249 | return State{}, err |
| 250 | } |
| 251 | return state, nil |
| 252 | } |
| 253 | |
| 254 | // MarkLegacyExported acknowledges a complete legacy mirror. It only clears the |
| 255 | // outbox when expectedRevision is still the current authoritative revision. |
| 256 | func (s *Store) MarkLegacyExported(ctx context.Context, expectedRevision int64, digests [4]string) (State, error) { |
| 257 | tx, err := s.db.BeginTx(ctx, nil) |
| 258 | if err != nil { |
| 259 | return State{}, err |
| 260 | } |
| 261 | defer func() { _ = tx.Rollback() }() |
| 262 | state, err := readState(ctx, tx) |
| 263 | if err != nil { |
| 264 | return State{}, err |
| 265 | } |
| 266 | if state.Revision == expectedRevision { |
| 267 | state.LegacyExportedRevision = expectedRevision |
| 268 | state.LegacyPendingRevision = 0 |
| 269 | state.LegacyTitlesDigest = digests[0] |
| 270 | state.LegacySourcesDigest = digests[1] |
| 271 | state.LegacyCreatedAtsDigest = digests[2] |
| 272 | state.LegacyAutoMetaDigest = digests[3] |
| 273 | if _, err := tx.ExecContext(ctx, `UPDATE store_state SET |
| 274 | legacy_exported_revision=?, legacy_pending_revision=0, |
| 275 | legacy_titles_digest=?, legacy_sources_digest=?, |
| 276 | legacy_created_ats_digest=?, legacy_auto_meta_digest=? |
| 277 | WHERE id=1`, expectedRevision, digests[0], digests[1], digests[2], digests[3]); err != nil { |
| 278 | return State{}, err |
| 279 | } |
| 280 | } |
| 281 | if err := tx.Commit(); err != nil { |
| 282 | return State{}, err |
| 283 | } |
| 284 | return state, nil |
| 285 | } |
| 286 | |
| 287 | // Update atomically changes one topic. Empty records are removed. |
| 288 | func (s *Store) Update(ctx context.Context, topicID string, mutate func(*Record)) (State, error) { |
| 289 | topicID = strings.TrimSpace(topicID) |
| 290 | if topicID == "" { |
| 291 | return State{}, errors.New("topic id is empty") |
| 292 | } |
| 293 | return s.mutateOne(ctx, topicID, mutate) |
| 294 | } |
| 295 | |
| 296 | // Delete removes all metadata for one topic in a single transaction. |
| 297 | func (s *Store) Delete(ctx context.Context, topicID string) (State, error) { |
| 298 | topicID = strings.TrimSpace(topicID) |
| 299 | if topicID == "" { |
| 300 | return s.currentState(ctx) |
| 301 | } |
| 302 | return s.mutateOne(ctx, topicID, func(record *Record) { |
| 303 | *record = Record{TopicID: topicID} |
| 304 | }) |
| 305 | } |
| 306 | |
| 307 | func (s *Store) mutateOne(ctx context.Context, topicID string, mutate func(*Record)) (State, error) { |
| 308 | tx, err := s.db.BeginTx(ctx, nil) |
| 309 | if err != nil { |
| 310 | return State{}, err |
| 311 | } |
| 312 | defer func() { _ = tx.Rollback() }() |
| 313 | state, err := readState(ctx, tx) |
| 314 | if err != nil { |
| 315 | return State{}, err |
| 316 | } |
| 317 | record := Record{TopicID: topicID} |
| 318 | var autoMeta string |
| 319 | err = tx.QueryRowContext(ctx, `SELECT title, title_source, created_at_ms, |
| 320 | auto_meta_json, row_revision, updated_at_ms FROM topics WHERE topic_id=?`, topicID).Scan( |
| 321 | &record.Title, &record.TitleSource, &record.CreatedAtMS, &autoMeta, |
| 322 | &record.RowRevision, &record.UpdatedAtMS, |
| 323 | ) |
| 324 | if err != nil && !errors.Is(err, sql.ErrNoRows) { |
| 325 | return State{}, err |
| 326 | } |
| 327 | if autoMeta != "" { |
| 328 | record.AutoMeta = json.RawMessage(autoMeta) |
| 329 | } |
| 330 | before := cloneRecord(record) |
| 331 | mutate(&record) |
| 332 | record.TopicID = topicID |
| 333 | normalizeRecord(&record) |
| 334 | if recordEqual(before, record) { |
| 335 | if err := tx.Commit(); err != nil { |
| 336 | return State{}, err |
| 337 | } |
| 338 | return state, nil |
| 339 | } |
| 340 | state.Revision++ |
| 341 | if recordEmpty(record) { |
| 342 | if _, err := tx.ExecContext(ctx, `DELETE FROM topics WHERE topic_id=?`, topicID); err != nil { |
| 343 | return State{}, err |
| 344 | } |
| 345 | } else { |
| 346 | record.RowRevision = state.Revision |
| 347 | record.UpdatedAtMS = s.now().UnixMilli() |
| 348 | if _, err := tx.ExecContext(ctx, `INSERT INTO topics( |
| 349 | topic_id, title, title_source, created_at_ms, auto_meta_json, |
| 350 | row_revision, updated_at_ms) VALUES(?, ?, ?, ?, ?, ?, ?) |
| 351 | ON CONFLICT(topic_id) DO UPDATE SET |
| 352 | title=excluded.title, title_source=excluded.title_source, |
| 353 | created_at_ms=excluded.created_at_ms, auto_meta_json=excluded.auto_meta_json, |
| 354 | row_revision=excluded.row_revision, updated_at_ms=excluded.updated_at_ms`, |
| 355 | record.TopicID, record.Title, record.TitleSource, record.CreatedAtMS, |
| 356 | string(record.AutoMeta), record.RowRevision, record.UpdatedAtMS); err != nil { |
| 357 | return State{}, err |
| 358 | } |
| 359 | } |
| 360 | if state.LegacyBridge { |
| 361 | state.LegacyPendingRevision = state.Revision |
| 362 | } |
| 363 | if _, err := tx.ExecContext(ctx, `UPDATE store_state SET revision=?, legacy_pending_revision=? WHERE id=1`, state.Revision, state.LegacyPendingRevision); err != nil { |
| 364 | return State{}, err |
| 365 | } |
| 366 | if err := tx.Commit(); err != nil { |
| 367 | return State{}, err |
| 368 | } |
| 369 | return state, nil |
| 370 | } |
| 371 | |
| 372 | // ReplaceAll transactionally replaces the complete record set. It is intended |
| 373 | // for first migration and deterministic legacy reconciliation. |
| 374 | func (s *Store) ReplaceAll(ctx context.Context, records map[string]Record) (State, error) { |
| 375 | return s.mutate(ctx, func(current map[string]Record) { |
| 376 | clear(current) |
| 377 | for id, record := range records { |
| 378 | record.TopicID = strings.TrimSpace(id) |
| 379 | normalizeRecord(&record) |
| 380 | if record.TopicID != "" && !recordEmpty(record) { |
| 381 | current[record.TopicID] = cloneRecord(record) |
| 382 | } |
| 383 | } |
| 384 | }) |
| 385 | } |
| 386 | |
| 387 | // ReplaceTitles replaces only the title field while preserving every other |
| 388 | // field, including unknown auto-title metadata. |
| 389 | func (s *Store) ReplaceTitles(ctx context.Context, values map[string]string) (State, error) { |
| 390 | return s.replaceField(ctx, func(records map[string]Record) { |
| 391 | for id, record := range records { |
| 392 | record.Title = "" |
| 393 | records[id] = record |
| 394 | } |
| 395 | for id, value := range values { |
| 396 | record := records[id] |
| 397 | record.TopicID, record.Title = id, value |
| 398 | records[id] = record |
| 399 | } |
| 400 | }) |
| 401 | } |
| 402 | |
| 403 | // ReplaceSources replaces only the title source field. |
| 404 | func (s *Store) ReplaceSources(ctx context.Context, values map[string]string) (State, error) { |
| 405 | return s.replaceField(ctx, func(records map[string]Record) { |
| 406 | for id, record := range records { |
| 407 | record.TitleSource = "" |
| 408 | records[id] = record |
| 409 | } |
| 410 | for id, value := range values { |
| 411 | record := records[id] |
| 412 | record.TopicID, record.TitleSource = id, value |
| 413 | records[id] = record |
| 414 | } |
| 415 | }) |
| 416 | } |
| 417 | |
| 418 | // MergeMissingTitleIndex fills title-index gaps from a scan-built snapshot |
| 419 | // without replacing values written after that scan began. Background session |
| 420 | // repair uses this compare-at-commit behavior so a concurrent manual rename |
| 421 | // cannot be reverted by stale whole-map data. |
| 422 | func (s *Store) MergeMissingTitleIndex(ctx context.Context, titles, sources map[string]string, deleted map[string]bool) (State, error) { |
| 423 | return s.replaceField(ctx, func(records map[string]Record) { |
| 424 | for id := range deleted { |
| 425 | delete(records, id) |
| 426 | } |
| 427 | for id, value := range titles { |
| 428 | if deleted[id] { |
| 429 | continue |
| 430 | } |
| 431 | record := records[id] |
| 432 | if strings.TrimSpace(record.Title) != "" { |
| 433 | continue |
| 434 | } |
| 435 | record.TopicID, record.Title = id, value |
| 436 | records[id] = record |
| 437 | } |
| 438 | for id, value := range sources { |
| 439 | if deleted[id] { |
| 440 | continue |
| 441 | } |
| 442 | record := records[id] |
| 443 | if strings.TrimSpace(record.TitleSource) != "" { |
| 444 | continue |
| 445 | } |
| 446 | record.TopicID, record.TitleSource = id, value |
| 447 | records[id] = record |
| 448 | } |
| 449 | }) |
| 450 | } |
| 451 | |
| 452 | // ReplaceCreatedAts replaces only topic creation timestamps. |
| 453 | func (s *Store) ReplaceCreatedAts(ctx context.Context, values map[string]int64) (State, error) { |
| 454 | return s.replaceField(ctx, func(records map[string]Record) { |
| 455 | for id, record := range records { |
| 456 | record.CreatedAtMS = 0 |
| 457 | records[id] = record |
| 458 | } |
| 459 | for id, value := range values { |
| 460 | record := records[id] |
| 461 | record.TopicID, record.CreatedAtMS = id, value |
| 462 | records[id] = record |
| 463 | } |
| 464 | }) |
| 465 | } |
| 466 | |
| 467 | // ReplaceAutoMeta replaces only raw automatic-title metadata. |
| 468 | func (s *Store) ReplaceAutoMeta(ctx context.Context, values map[string]json.RawMessage) (State, error) { |
| 469 | return s.replaceField(ctx, func(records map[string]Record) { |
| 470 | for id, record := range records { |
| 471 | record.AutoMeta = nil |
| 472 | records[id] = record |
| 473 | } |
| 474 | for id, value := range values { |
| 475 | record := records[id] |
| 476 | record.TopicID, record.AutoMeta = id, append(json.RawMessage(nil), value...) |
| 477 | records[id] = record |
| 478 | } |
| 479 | }) |
| 480 | } |
| 481 | |
| 482 | func (s *Store) replaceField(ctx context.Context, mutate func(map[string]Record)) (State, error) { |
| 483 | return s.mutate(ctx, func(records map[string]Record) { |
| 484 | mutate(records) |
| 485 | for id, record := range records { |
| 486 | normalizeRecord(&record) |
| 487 | if recordEmpty(record) { |
| 488 | delete(records, id) |
| 489 | } else { |
| 490 | records[id] = record |
| 491 | } |
| 492 | } |
| 493 | }) |
| 494 | } |
| 495 | |
| 496 | func (s *Store) mutate(ctx context.Context, mutate func(map[string]Record)) (State, error) { |
| 497 | tx, err := s.db.BeginTx(ctx, nil) |
| 498 | if err != nil { |
| 499 | return State{}, err |
| 500 | } |
| 501 | defer func() { _ = tx.Rollback() }() |
| 502 | state, err := readState(ctx, tx) |
| 503 | if err != nil { |
| 504 | return State{}, err |
| 505 | } |
| 506 | records, err := readRecords(ctx, tx) |
| 507 | if err != nil { |
| 508 | return State{}, err |
| 509 | } |
| 510 | before := cloneRecords(records) |
| 511 | mutate(records) |
| 512 | if recordsEqual(before, records) { |
| 513 | if err := tx.Commit(); err != nil { |
| 514 | return State{}, err |
| 515 | } |
| 516 | return state, nil |
| 517 | } |
| 518 | nowMS := s.now().UnixMilli() |
| 519 | state.Revision++ |
| 520 | for id, record := range records { |
| 521 | if previous, ok := before[id]; !ok || !recordEqual(previous, record) { |
| 522 | record.RowRevision = state.Revision |
| 523 | record.UpdatedAtMS = nowMS |
| 524 | records[id] = record |
| 525 | } |
| 526 | } |
| 527 | if err := writeRecords(ctx, tx, records); err != nil { |
| 528 | return State{}, err |
| 529 | } |
| 530 | if state.LegacyBridge { |
| 531 | state.LegacyPendingRevision = state.Revision |
| 532 | } |
| 533 | if _, err := tx.ExecContext(ctx, `UPDATE store_state SET revision=?, legacy_pending_revision=? WHERE id=1`, state.Revision, state.LegacyPendingRevision); err != nil { |
| 534 | return State{}, err |
| 535 | } |
| 536 | if err := tx.Commit(); err != nil { |
| 537 | return State{}, err |
| 538 | } |
| 539 | return state, nil |
| 540 | } |
| 541 | |
| 542 | func (s *Store) currentState(ctx context.Context) (State, error) { |
| 543 | return readState(ctx, s.db) |
| 544 | } |
| 545 | |
| 546 | type rowQuerier interface { |
| 547 | QueryRowContext(context.Context, string, ...any) *sql.Row |
| 548 | } |
| 549 | |
| 550 | func readState(ctx context.Context, q rowQuerier) (State, error) { |
| 551 | var state State |
| 552 | var bridge int |
| 553 | err := q.QueryRowContext(ctx, `SELECT revision, legacy_bridge, |
| 554 | legacy_exported_revision, legacy_pending_revision, |
| 555 | legacy_titles_digest, legacy_sources_digest, |
| 556 | legacy_created_ats_digest, legacy_auto_meta_digest |
| 557 | FROM store_state WHERE id=1`).Scan( |
| 558 | &state.Revision, &bridge, &state.LegacyExportedRevision, |
| 559 | &state.LegacyPendingRevision, &state.LegacyTitlesDigest, |
| 560 | &state.LegacySourcesDigest, &state.LegacyCreatedAtsDigest, |
| 561 | &state.LegacyAutoMetaDigest, |
| 562 | ) |
| 563 | state.LegacyBridge = bridge != 0 |
| 564 | return state, err |
| 565 | } |
| 566 | |
| 567 | type rowsQuerier interface { |
| 568 | QueryContext(context.Context, string, ...any) (*sql.Rows, error) |
| 569 | } |
| 570 | |
| 571 | func readRecords(ctx context.Context, q rowsQuerier) (map[string]Record, error) { |
| 572 | rows, err := q.QueryContext(ctx, `SELECT topic_id, title, title_source, |
| 573 | created_at_ms, auto_meta_json, row_revision, updated_at_ms FROM topics`) |
| 574 | if err != nil { |
| 575 | return nil, err |
| 576 | } |
| 577 | defer rows.Close() |
| 578 | records := map[string]Record{} |
| 579 | for rows.Next() { |
| 580 | var record Record |
| 581 | var autoMeta string |
| 582 | if err := rows.Scan(&record.TopicID, &record.Title, &record.TitleSource, |
| 583 | &record.CreatedAtMS, &autoMeta, &record.RowRevision, &record.UpdatedAtMS); err != nil { |
| 584 | return nil, err |
| 585 | } |
| 586 | if autoMeta != "" { |
| 587 | record.AutoMeta = json.RawMessage(autoMeta) |
| 588 | } |
| 589 | records[record.TopicID] = record |
| 590 | } |
| 591 | return records, rows.Err() |
| 592 | } |
| 593 | |
| 594 | func writeRecords(ctx context.Context, tx *sql.Tx, records map[string]Record) error { |
| 595 | if _, err := tx.ExecContext(ctx, `DELETE FROM topics`); err != nil { |
| 596 | return err |
| 597 | } |
| 598 | stmt, err := tx.PrepareContext(ctx, `INSERT INTO topics( |
| 599 | topic_id, title, title_source, created_at_ms, auto_meta_json, |
| 600 | row_revision, updated_at_ms) VALUES(?, ?, ?, ?, ?, ?, ?)`) |
| 601 | if err != nil { |
| 602 | return err |
| 603 | } |
| 604 | defer stmt.Close() |
| 605 | ids := make([]string, 0, len(records)) |
| 606 | for id := range records { |
| 607 | ids = append(ids, id) |
| 608 | } |
| 609 | sort.Strings(ids) |
| 610 | for _, id := range ids { |
| 611 | record := records[id] |
| 612 | normalizeRecord(&record) |
| 613 | if recordEmpty(record) { |
| 614 | continue |
| 615 | } |
| 616 | if _, err := stmt.ExecContext(ctx, record.TopicID, record.Title, |
| 617 | record.TitleSource, record.CreatedAtMS, string(record.AutoMeta), |
| 618 | record.RowRevision, record.UpdatedAtMS); err != nil { |
| 619 | return err |
| 620 | } |
| 621 | } |
| 622 | return nil |
| 623 | } |
| 624 | |
| 625 | func normalizeRecord(record *Record) { |
| 626 | record.TopicID = strings.TrimSpace(record.TopicID) |
| 627 | record.Title = strings.TrimSpace(record.Title) |
| 628 | record.TitleSource = strings.TrimSpace(record.TitleSource) |
| 629 | if record.CreatedAtMS < 0 { |
| 630 | record.CreatedAtMS = 0 |
| 631 | } |
| 632 | if len(bytes.TrimSpace(record.AutoMeta)) == 0 || bytes.Equal(bytes.TrimSpace(record.AutoMeta), []byte("null")) || bytes.Equal(bytes.TrimSpace(record.AutoMeta), []byte("{}")) { |
| 633 | record.AutoMeta = nil |
| 634 | } else { |
| 635 | record.AutoMeta = append(json.RawMessage(nil), bytes.TrimSpace(record.AutoMeta)...) |
| 636 | } |
| 637 | } |
| 638 | |
| 639 | func recordEmpty(record Record) bool { |
| 640 | return record.Title == "" && record.TitleSource == "" && record.CreatedAtMS <= 0 && len(record.AutoMeta) == 0 |
| 641 | } |
| 642 | |
| 643 | func cloneRecord(record Record) Record { |
| 644 | record.AutoMeta = append(json.RawMessage(nil), record.AutoMeta...) |
| 645 | return record |
| 646 | } |
| 647 | |
| 648 | func cloneRecords(records map[string]Record) map[string]Record { |
| 649 | clone := make(map[string]Record, len(records)) |
| 650 | for id, record := range records { |
| 651 | clone[id] = cloneRecord(record) |
| 652 | } |
| 653 | return clone |
| 654 | } |
| 655 | |
| 656 | func recordsEqual(a, b map[string]Record) bool { |
| 657 | if len(a) != len(b) { |
| 658 | return false |
| 659 | } |
| 660 | for id, record := range a { |
| 661 | if other, ok := b[id]; !ok || !recordEqual(record, other) { |
| 662 | return false |
| 663 | } |
| 664 | } |
| 665 | return true |
| 666 | } |
| 667 | |
| 668 | func recordEqual(a, b Record) bool { |
| 669 | return a.TopicID == b.TopicID && a.Title == b.Title && |
| 670 | a.TitleSource == b.TitleSource && a.CreatedAtMS == b.CreatedAtMS && |
| 671 | bytes.Equal(a.AutoMeta, b.AutoMeta) |
| 672 | } |
| 673 | |
| 674 | // IsCorruptionError reports only integrity-level failures. Busy, permissions, |
| 675 | // IO errors, and future schemas must never cause an authoritative DB rename. |
| 676 | func IsCorruptionError(err error) bool { |
| 677 | if err == nil { |
| 678 | return false |
| 679 | } |
| 680 | var future *FutureSchemaError |
| 681 | if errors.As(err, &future) { |
| 682 | return false |
| 683 | } |
| 684 | var sqliteErr *moderncsqlite.Error |
| 685 | if errors.As(err, &sqliteErr) { |
| 686 | switch sqliteErr.Code() & 0xff { |
| 687 | case sqlite3.SQLITE_CORRUPT, sqlite3.SQLITE_NOTADB: |
| 688 | return true |
| 689 | } |
| 690 | } |
| 691 | message := strings.ToLower(err.Error()) |
| 692 | return strings.Contains(message, "quick check") || |
| 693 | strings.Contains(message, "malformed") || |
| 694 | strings.Contains(message, "file is not a database") || |
| 695 | strings.Contains(message, "not a database") |
| 696 | } |
| 697 | |
| 698 | // Quarantine preserves a corrupt database and its WAL sidecars for diagnosis. |
| 699 | func Quarantine(path string, now time.Time) (string, error) { |
| 700 | if strings.TrimSpace(path) == "" { |
| 701 | return "", errors.New("topic state path is empty") |
| 702 | } |
| 703 | quarantined := fmt.Sprintf("%s.corrupt-%d", path, now.UnixMilli()) |
| 704 | if err := os.Rename(path, quarantined); err != nil { |
| 705 | return "", err |
| 706 | } |
| 707 | moved := []string{} |
| 708 | for _, suffix := range []string{"-wal", "-shm"} { |
| 709 | if err := os.Rename(path+suffix, quarantined+suffix); err == nil { |
| 710 | moved = append(moved, suffix) |
| 711 | } else if !errors.Is(err, os.ErrNotExist) { |
| 712 | for _, movedSuffix := range moved { |
| 713 | _ = os.Rename(quarantined+movedSuffix, path+movedSuffix) |
| 714 | } |
| 715 | _ = os.Rename(quarantined, path) |
| 716 | return "", err |
| 717 | } |
| 718 | } |
| 719 | return quarantined, nil |
| 720 | } |
| 721 |