| 1 | // Package sessionui owns recoverable Desktop editor state, never chat history. |
| 2 | package sessionui |
| 3 | |
| 4 | import ( |
| 5 | "context" |
| 6 | "database/sql" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "os" |
| 11 | "path/filepath" |
| 12 | "strconv" |
| 13 | "sync" |
| 14 | |
| 15 | "reasonix/internal/sqliteuri" |
| 16 | |
| 17 | _ "modernc.org/sqlite" |
| 18 | ) |
| 19 | |
| 20 | var ErrConflict = errors.New("session UI revision conflict") |
| 21 | var ErrFutureVersion = errors.New("unsupported session UI schema") |
| 22 | |
| 23 | type Record struct { |
| 24 | Key string `json:"key"` |
| 25 | Revision string `json:"revision"` |
| 26 | Payload json.RawMessage `json:"payload"` |
| 27 | } |
| 28 | |
| 29 | type Store struct { |
| 30 | mu sync.Mutex |
| 31 | path string |
| 32 | db *sql.DB |
| 33 | closed bool |
| 34 | } |
| 35 | |
| 36 | func New(path string) *Store { return &Store{path: path} } |
| 37 | |
| 38 | func (s *Store) Path() string { return s.path } |
| 39 | |
| 40 | // PurgeComposer runs only after canonical deletion has crossed its tombstone. |
| 41 | func (s *Store) PurgeComposer(ctx context.Context, key string) error { |
| 42 | s.mu.Lock() |
| 43 | defer s.mu.Unlock() |
| 44 | if _, err := os.Stat(s.path); errors.Is(err, os.ErrNotExist) { |
| 45 | return nil |
| 46 | } |
| 47 | if err := s.open(); err != nil { |
| 48 | return err |
| 49 | } |
| 50 | tx, err := s.db.BeginTx(ctx, nil) |
| 51 | if err != nil { |
| 52 | return err |
| 53 | } |
| 54 | defer func() { _ = tx.Rollback() }() |
| 55 | if _, err = tx.ExecContext(ctx, `DELETE FROM records WHERE kind='composer' AND key=?`, key); err != nil { |
| 56 | return err |
| 57 | } |
| 58 | if _, err = tx.ExecContext(ctx, `DELETE FROM conflicts WHERE kind='composer' AND key=?`, key); err != nil { |
| 59 | return err |
| 60 | } |
| 61 | if _, err = tx.ExecContext(ctx, `DELETE FROM records WHERE kind='submission' AND substr(key,1,?)=?`, len(key)+1, key+"/"); err != nil { |
| 62 | return err |
| 63 | } |
| 64 | return tx.Commit() |
| 65 | } |
| 66 | |
| 67 | func (s *Store) open() error { |
| 68 | if s.closed { |
| 69 | return errors.New("session UI store is closed") |
| 70 | } |
| 71 | if s.db != nil { |
| 72 | return nil |
| 73 | } |
| 74 | if err := os.MkdirAll(filepath.Dir(s.path), 0700); err != nil { |
| 75 | return err |
| 76 | } |
| 77 | dsn, err := sqliteuri.Disk(s.path, nil) |
| 78 | if err != nil { |
| 79 | return err |
| 80 | } |
| 81 | db, err := sql.Open("sqlite", dsn) |
| 82 | if err != nil { |
| 83 | return err |
| 84 | } |
| 85 | db.SetMaxOpenConns(1) |
| 86 | ok := false |
| 87 | defer func() { |
| 88 | if !ok { |
| 89 | _ = db.Close() |
| 90 | } |
| 91 | }() |
| 92 | var version int |
| 93 | if err = db.QueryRow("PRAGMA user_version").Scan(&version); err != nil { |
| 94 | return err |
| 95 | } |
| 96 | if version < 0 || version > 1 { |
| 97 | return ErrFutureVersion |
| 98 | } |
| 99 | if version == 0 { |
| 100 | var tables int |
| 101 | if err = db.QueryRow(`SELECT count(*) FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'`).Scan(&tables); err != nil { |
| 102 | return err |
| 103 | } |
| 104 | if tables != 0 { |
| 105 | if err = db.QueryRow("PRAGMA user_version").Scan(&version); err != nil { |
| 106 | return err |
| 107 | } |
| 108 | if version != 1 { |
| 109 | return ErrFutureVersion |
| 110 | } |
| 111 | } |
| 112 | } |
| 113 | for _, statement := range []string{"PRAGMA busy_timeout = 5000", "PRAGMA journal_mode = WAL"} { |
| 114 | if _, err = db.Exec(statement); err != nil { |
| 115 | return err |
| 116 | } |
| 117 | } |
| 118 | if version == 0 { |
| 119 | tx, err := db.Begin() |
| 120 | if err != nil { |
| 121 | return err |
| 122 | } |
| 123 | defer func() { _ = tx.Rollback() }() |
| 124 | for _, statement := range []string{ |
| 125 | `CREATE TABLE IF NOT EXISTS records (kind TEXT NOT NULL, key TEXT NOT NULL, revision INTEGER NOT NULL, payload BLOB NOT NULL, PRIMARY KEY(kind,key))`, |
| 126 | `CREATE TABLE IF NOT EXISTS conflicts (id INTEGER PRIMARY KEY, kind TEXT NOT NULL, key TEXT NOT NULL, expected INTEGER NOT NULL, actual INTEGER NOT NULL, payload BLOB NOT NULL)`, |
| 127 | `PRAGMA user_version = 1`, |
| 128 | } { |
| 129 | if _, err = tx.Exec(statement); err != nil { |
| 130 | return err |
| 131 | } |
| 132 | } |
| 133 | if err = tx.Commit(); err != nil { |
| 134 | return err |
| 135 | } |
| 136 | } |
| 137 | s.db, ok = db, true |
| 138 | return nil |
| 139 | } |
| 140 | |
| 141 | func (s *Store) Close() error { |
| 142 | s.mu.Lock() |
| 143 | defer s.mu.Unlock() |
| 144 | s.closed = true |
| 145 | if s.db == nil { |
| 146 | return nil |
| 147 | } |
| 148 | err := s.db.Close() |
| 149 | s.db = nil |
| 150 | return err |
| 151 | } |
| 152 | |
| 153 | func scan(row interface{ Scan(...any) error }, key string) (Record, error) { |
| 154 | r := Record{Key: key, Revision: "0", Payload: json.RawMessage(`{}`)} |
| 155 | var revision int64 |
| 156 | err := row.Scan(&revision, &r.Payload) |
| 157 | if errors.Is(err, sql.ErrNoRows) { |
| 158 | return r, nil |
| 159 | } |
| 160 | if err != nil { |
| 161 | return r, err |
| 162 | } |
| 163 | r.Revision = strconv.FormatInt(revision, 10) |
| 164 | return r, nil |
| 165 | } |
| 166 | |
| 167 | func (s *Store) Get(ctx context.Context, kind, key string) (Record, error) { |
| 168 | s.mu.Lock() |
| 169 | defer s.mu.Unlock() |
| 170 | if err := s.open(); err != nil { |
| 171 | return Record{}, err |
| 172 | } |
| 173 | return scan(s.db.QueryRowContext(ctx, `SELECT revision,payload FROM records WHERE kind=? AND key=?`, kind, key), key) |
| 174 | } |
| 175 | |
| 176 | // Save uses SQL CAS across processes. A losing editor's complete input is kept |
| 177 | // before returning the winning record; force-overwrite is intentionally absent. |
| 178 | func (s *Store) Save(ctx context.Context, kind, key, expected string, payload json.RawMessage, submissions ...Record) (Record, error) { |
| 179 | if !json.Valid(payload) || key == "" || kind == "" { |
| 180 | return Record{}, errors.New("invalid session UI record") |
| 181 | } |
| 182 | rev, err := strconv.ParseInt(expected, 10, 64) |
| 183 | if err != nil || rev < 0 || rev == int64(^uint64(0)>>1) { |
| 184 | return Record{}, errors.New("invalid session UI revision") |
| 185 | } |
| 186 | s.mu.Lock() |
| 187 | defer s.mu.Unlock() |
| 188 | if err = s.open(); err != nil { |
| 189 | return Record{}, err |
| 190 | } |
| 191 | tx, err := s.db.BeginTx(ctx, nil) |
| 192 | if err != nil { |
| 193 | return Record{}, err |
| 194 | } |
| 195 | defer func() { _ = tx.Rollback() }() |
| 196 | // Write first: acquiring SQLite's writer before reading avoids upgrading a |
| 197 | // stale deferred read transaction when another process saves concurrently. |
| 198 | var result sql.Result |
| 199 | if rev == 0 { |
| 200 | result, err = tx.ExecContext(ctx, `INSERT INTO records(kind,key,revision,payload) VALUES(?,?,1,?) ON CONFLICT(kind,key) DO NOTHING`, kind, key, []byte(payload)) |
| 201 | } else { |
| 202 | result, err = tx.ExecContext(ctx, `UPDATE records SET revision=revision+1,payload=? WHERE kind=? AND key=? AND revision=?`, []byte(payload), kind, key, rev) |
| 203 | } |
| 204 | if err != nil { |
| 205 | return Record{}, err |
| 206 | } |
| 207 | n, err := result.RowsAffected() |
| 208 | if err != nil { |
| 209 | return Record{}, err |
| 210 | } |
| 211 | r, err := scan(tx.QueryRowContext(ctx, `SELECT revision,payload FROM records WHERE kind=? AND key=?`, kind, key), key) |
| 212 | if err != nil { |
| 213 | return r, err |
| 214 | } |
| 215 | // Formal composer input has one saved value. A stale writer receives the |
| 216 | // current revision and may rebase; it does not create alternate input copies. |
| 217 | if n == 0 && kind != "composer" { |
| 218 | if _, err = tx.ExecContext(ctx, `INSERT INTO conflicts(kind,key,expected,actual,payload) VALUES(?,?,?,?,?)`, kind, key, rev, r.Revision, []byte(payload)); err != nil { |
| 219 | return r, err |
| 220 | } |
| 221 | } |
| 222 | if n != 0 { |
| 223 | for _, submission := range submissions { |
| 224 | if !json.Valid(submission.Payload) { |
| 225 | return r, errors.New("invalid submission record") |
| 226 | } |
| 227 | if _, err = tx.ExecContext(ctx, `INSERT INTO records(kind,key,revision,payload) VALUES('submission',?,1,?) ON CONFLICT(kind,key) DO UPDATE SET revision=revision+1,payload=excluded.payload`, submission.Key, []byte(submission.Payload)); err != nil { |
| 228 | return r, err |
| 229 | } |
| 230 | } |
| 231 | } |
| 232 | if err = tx.Commit(); err != nil { |
| 233 | return r, err |
| 234 | } |
| 235 | if n == 0 { |
| 236 | return r, ErrConflict |
| 237 | } |
| 238 | return r, nil |
| 239 | } |
| 240 | |
| 241 | func (s *Store) List(ctx context.Context, kind string) ([]Record, error) { |
| 242 | s.mu.Lock() |
| 243 | defer s.mu.Unlock() |
| 244 | if err := s.open(); err != nil { |
| 245 | return nil, err |
| 246 | } |
| 247 | rows, err := s.db.QueryContext(ctx, `SELECT key,revision,payload FROM records WHERE kind=? ORDER BY key`, kind) |
| 248 | if err != nil { |
| 249 | return nil, err |
| 250 | } |
| 251 | defer rows.Close() |
| 252 | out := []Record{} |
| 253 | for rows.Next() { |
| 254 | var r Record |
| 255 | var rev int64 |
| 256 | if err := rows.Scan(&r.Key, &rev, &r.Payload); err != nil { |
| 257 | return nil, err |
| 258 | } |
| 259 | r.Revision = fmt.Sprint(rev) |
| 260 | out = append(out, r) |
| 261 | } |
| 262 | return out, rows.Err() |
| 263 | } |
| 264 |