| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/sha256" |
| 6 | "database/sql" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "strings" |
| 12 | "unicode/utf8" |
| 13 | |
| 14 | "reasonix/internal/projectiondb" |
| 15 | "reasonix/internal/provider" |
| 16 | "reasonix/internal/sessioncontent" |
| 17 | ) |
| 18 | |
| 19 | const searchIndexVersion = 4 |
| 20 | |
| 21 | var searchMigrations = []projectiondb.Migration{{Version: 1, Apply: func(ctx context.Context, tx *sql.Tx) error { |
| 22 | for _, statement := range []string{ |
| 23 | `CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)`, |
| 24 | `CREATE TABLE documents (rowid INTEGER PRIMARY KEY, message_id TEXT NOT NULL, version INTEGER NOT NULL, position INTEGER NOT NULL, event_sequence INTEGER NOT NULL, valid_to INTEGER NOT NULL DEFAULT 0, role TEXT NOT NULL, preview TEXT NOT NULL, text TEXT NOT NULL, current INTEGER NOT NULL, UNIQUE(message_id,version))`, |
| 25 | `CREATE INDEX documents_snapshot_position ON documents(position DESC,event_sequence,valid_to)`, |
| 26 | `CREATE INDEX documents_current_id ON documents(message_id) WHERE current=1`, |
| 27 | `CREATE VIRTUAL TABLE documents_fts USING fts5(text, content='documents', content_rowid='rowid', tokenize='trigram')`, |
| 28 | `CREATE TRIGGER documents_ai AFTER INSERT ON documents BEGIN INSERT INTO documents_fts(rowid,text) VALUES (new.rowid,new.text); END`, |
| 29 | `CREATE TRIGGER documents_ad AFTER DELETE ON documents BEGIN INSERT INTO documents_fts(documents_fts,rowid,text) VALUES('delete',old.rowid,old.text); END`, |
| 30 | `CREATE TRIGGER documents_au AFTER UPDATE OF text ON documents BEGIN INSERT INTO documents_fts(documents_fts,rowid,text) VALUES('delete',old.rowid,old.text); INSERT INTO documents_fts(rowid,text) VALUES(new.rowid,new.text); END`, |
| 31 | } { |
| 32 | if _, err := tx.ExecContext(ctx, statement); err != nil { |
| 33 | return err |
| 34 | } |
| 35 | } |
| 36 | return nil |
| 37 | }}} |
| 38 | |
| 39 | type searchBuildState struct { |
| 40 | positions map[string]int64 |
| 41 | versions map[string]int |
| 42 | nextPosition int64 |
| 43 | } |
| 44 | |
| 45 | type searchPreparation struct { |
| 46 | done chan struct{} |
| 47 | err error |
| 48 | } |
| 49 | |
| 50 | func searchIndexPath(root, sessionID string) string { |
| 51 | return filepath.Join(root, ".query-cache", filepath.Base(sessionID), "search-v1.sqlite") |
| 52 | } |
| 53 | |
| 54 | func (q *Query) SearchHistory(ctx context.Context, ref SessionRef, textQuery, cursor string, limit int) (SearchHistoryPage, error) { |
| 55 | if q == nil { |
| 56 | return SearchHistoryPage{}, errors.New("session: nil query") |
| 57 | } |
| 58 | if err := ref.validate(q.hostID); err != nil { |
| 59 | return SearchHistoryPage{}, err |
| 60 | } |
| 61 | textQuery = strings.TrimSpace(textQuery) |
| 62 | if textQuery == "" { |
| 63 | return SearchHistoryPage{}, errors.New("session: history search query is required") |
| 64 | } |
| 65 | if limit <= 0 { |
| 66 | limit = 50 |
| 67 | } |
| 68 | limit = min(limit, 200) |
| 69 | filesystem, ok := q.persistence.(*FilesystemPersistence) |
| 70 | if !ok { |
| 71 | return SearchHistoryPage{}, errors.New("session: history search requires filesystem persistence") |
| 72 | } |
| 73 | path := searchIndexPath(filesystem.Root, ref.SessionID) |
| 74 | if _, err := os.Stat(path); errors.Is(err, os.ErrNotExist) { |
| 75 | preparation := q.prepareSearchIndex(filesystem, ref.SessionID, path, ctx) |
| 76 | select { |
| 77 | case <-preparation.done: |
| 78 | if preparation.err != nil { |
| 79 | return SearchHistoryPage{Status: "failed"}, preparation.err |
| 80 | } |
| 81 | default: |
| 82 | return SearchHistoryPage{Status: "preparing"}, nil |
| 83 | } |
| 84 | } |
| 85 | lock := q.projectionLock("search", ref.SessionID) |
| 86 | lock.Lock() |
| 87 | err := ensureSearchIndex(ctx, filesystem, ref.SessionID, path) |
| 88 | lock.Unlock() |
| 89 | if err != nil { |
| 90 | return SearchHistoryPage{}, err |
| 91 | } |
| 92 | handle, err := projectiondb.Open(ctx, projectiondb.OpenOptions{Path: path, Migrations: searchMigrations, RequireDisk: true, MaxOpenConns: 1}) |
| 93 | if err != nil { |
| 94 | return SearchHistoryPage{}, err |
| 95 | } |
| 96 | defer handle.DB.Close() |
| 97 | metadata, err := readSearchMetadata(ctx, handle.DB) |
| 98 | if err != nil { |
| 99 | return SearchHistoryPage{}, err |
| 100 | } |
| 101 | snapshot := metadata.durableSequence |
| 102 | before := int64(^uint64(0) >> 1) |
| 103 | digest := fmt.Sprintf("%x", sha256.Sum256([]byte(textQuery))) |
| 104 | if cursor != "" { |
| 105 | parsed, err := decodeSearchHistoryCursor(cursor) |
| 106 | if err != nil { |
| 107 | return SearchHistoryPage{}, err |
| 108 | } |
| 109 | if parsed.SessionID != ref.SessionID || parsed.StorageRevision != StorageRevision || parsed.Projection != searchIndexVersion || parsed.QueryDigest != digest || parsed.SnapshotSequence > snapshot || parsed.BeforePosition <= 0 || parsed.Generation != metadata.generation { |
| 110 | return SearchHistoryPage{Status: "stale_cursor", CoverageSequence: metadata.durableSequence}, nil |
| 111 | } |
| 112 | snapshot, before = parsed.SnapshotSequence, parsed.BeforePosition |
| 113 | } |
| 114 | var rows *sql.Rows |
| 115 | if utf8.RuneCountInString(textQuery) >= 3 { |
| 116 | match := `"` + strings.ReplaceAll(textQuery, `"`, `""`) + `"` |
| 117 | rows, err = handle.DB.QueryContext(ctx, `SELECT d.message_id,d.position,d.role,d.preview,d.event_sequence FROM documents_fts JOIN documents d ON d.rowid=documents_fts.rowid WHERE documents_fts MATCH ? AND instr(d.text,?)>0 AND d.position<? AND d.event_sequence<=? AND (d.valid_to=0 OR d.valid_to>?) ORDER BY d.position DESC LIMIT ?`, match, textQuery, before, snapshot, snapshot, limit+1) |
| 118 | } else { |
| 119 | rows, err = handle.DB.QueryContext(ctx, `SELECT message_id,position,role,preview,event_sequence FROM documents WHERE instr(text,?)>0 AND position<? AND event_sequence<=? AND (valid_to=0 OR valid_to>?) ORDER BY position DESC LIMIT ?`, textQuery, before, snapshot, snapshot, limit+1) |
| 120 | } |
| 121 | if err != nil { |
| 122 | return SearchHistoryPage{}, err |
| 123 | } |
| 124 | defer rows.Close() |
| 125 | page := SearchHistoryPage{Hits: []SearchHistoryHit{}, SnapshotSequence: snapshot, CoverageSequence: metadata.durableSequence, Status: "ready"} |
| 126 | for rows.Next() { |
| 127 | var hit SearchHistoryHit |
| 128 | if err := rows.Scan(&hit.MessageID, &hit.Position, &hit.Role, &hit.Preview, &hit.EventSequence); err != nil { |
| 129 | return SearchHistoryPage{}, err |
| 130 | } |
| 131 | if len(page.Hits) == limit { |
| 132 | page.HasMore = true |
| 133 | break |
| 134 | } |
| 135 | page.Hits = append(page.Hits, hit) |
| 136 | } |
| 137 | if err := rows.Err(); err != nil { |
| 138 | return SearchHistoryPage{}, err |
| 139 | } |
| 140 | if page.HasMore && len(page.Hits) > 0 { |
| 141 | page.NextCursor, err = encodeSearchHistoryCursor(searchHistoryCursor{SessionID: ref.SessionID, StorageRevision: StorageRevision, SnapshotSequence: snapshot, BeforePosition: page.Hits[len(page.Hits)-1].Position, Projection: searchIndexVersion, QueryDigest: digest, Generation: metadata.generation}) |
| 142 | if err != nil { |
| 143 | return SearchHistoryPage{}, err |
| 144 | } |
| 145 | } |
| 146 | return page, nil |
| 147 | } |
| 148 | |
| 149 | func (q *Query) prepareSearchIndex(filesystem *FilesystemPersistence, sessionID, path string, callers ...context.Context) *searchPreparation { |
| 150 | q.searchMu.Lock() |
| 151 | if current := q.searchBuilds[sessionID]; current != nil { |
| 152 | select { |
| 153 | case <-current.done: |
| 154 | if !errors.Is(current.err, context.Canceled) { |
| 155 | q.searchMu.Unlock() |
| 156 | return current |
| 157 | } |
| 158 | default: |
| 159 | q.searchMu.Unlock() |
| 160 | return current |
| 161 | } |
| 162 | } |
| 163 | preparation := &searchPreparation{done: make(chan struct{})} |
| 164 | q.searchBuilds[sessionID] = preparation |
| 165 | q.searchMu.Unlock() |
| 166 | q.rebuildMu.Lock() |
| 167 | if q.closed { |
| 168 | q.rebuildMu.Unlock() |
| 169 | preparation.err = context.Canceled |
| 170 | close(preparation.done) |
| 171 | return preparation |
| 172 | } |
| 173 | q.rebuildWG.Add(1) |
| 174 | q.rebuildMu.Unlock() |
| 175 | ctx := q.historyReadContext(sessionID, callers...) |
| 176 | go func() { |
| 177 | defer q.rebuildWG.Done() |
| 178 | if err := q.slots.acquire(ctx, rebuildPrioritySearch); err != nil { |
| 179 | preparation.err = err |
| 180 | close(preparation.done) |
| 181 | return |
| 182 | } |
| 183 | defer q.slots.release() |
| 184 | release, err := q.acquireHistoryPreparation(ctx) |
| 185 | if err != nil { |
| 186 | preparation.err = err |
| 187 | close(preparation.done) |
| 188 | return |
| 189 | } |
| 190 | defer release() |
| 191 | lock := q.projectionLock("search", sessionID) |
| 192 | lock.Lock() |
| 193 | preparation.err = ensureSearchIndex(ctx, filesystem, sessionID, path) |
| 194 | lock.Unlock() |
| 195 | close(preparation.done) |
| 196 | }() |
| 197 | return preparation |
| 198 | } |
| 199 | |
| 200 | type searchMetadata struct { |
| 201 | sessionID string |
| 202 | logSize int64 |
| 203 | storageRevision int |
| 204 | projection int |
| 205 | durableSequence uint64 |
| 206 | generation string |
| 207 | } |
| 208 | |
| 209 | func readSearchMetadata(ctx context.Context, db *sql.DB) (searchMetadata, error) { |
| 210 | values := map[string]string{} |
| 211 | rows, err := db.QueryContext(ctx, `SELECT key,value FROM metadata`) |
| 212 | if err != nil { |
| 213 | return searchMetadata{}, err |
| 214 | } |
| 215 | defer rows.Close() |
| 216 | for rows.Next() { |
| 217 | var key, value string |
| 218 | if err := rows.Scan(&key, &value); err != nil { |
| 219 | return searchMetadata{}, err |
| 220 | } |
| 221 | values[key] = value |
| 222 | } |
| 223 | metadata := searchMetadata{sessionID: values["session_id"], generation: values["generation"]} |
| 224 | if _, err := fmt.Sscan(values["log_size"], &metadata.logSize); err != nil { |
| 225 | return searchMetadata{}, err |
| 226 | } |
| 227 | if _, err := fmt.Sscan(values["storage_revision"], &metadata.storageRevision); err != nil { |
| 228 | return searchMetadata{}, err |
| 229 | } |
| 230 | if _, err := fmt.Sscan(values["projection_version"], &metadata.projection); err != nil { |
| 231 | return searchMetadata{}, err |
| 232 | } |
| 233 | if _, err := fmt.Sscan(values["durable_sequence"], &metadata.durableSequence); err != nil { |
| 234 | return searchMetadata{}, err |
| 235 | } |
| 236 | return metadata, rows.Err() |
| 237 | } |
| 238 | |
| 239 | func ensureSearchIndex(ctx context.Context, persistence *FilesystemPersistence, sessionID, path string) error { |
| 240 | dir := filepath.Join(persistence.Root, sessionID) |
| 241 | revision, err := revisionOfLog(dir) |
| 242 | if err != nil { |
| 243 | return err |
| 244 | } |
| 245 | if handle, err := projectiondb.Open(ctx, projectiondb.OpenOptions{Path: path, Migrations: searchMigrations, RequireDisk: true, MaxOpenConns: 1}); err == nil { |
| 246 | metadata, metaErr := readSearchMetadata(ctx, handle.DB) |
| 247 | _ = handle.DB.Close() |
| 248 | if metaErr == nil && metadata.sessionID == sessionID && metadata.storageRevision == StorageRevision && metadata.projection == searchIndexVersion { |
| 249 | if metadata.logSize == revision.Size { |
| 250 | return nil |
| 251 | } |
| 252 | if metadata.logSize >= 0 && metadata.logSize < revision.Size { |
| 253 | if err := incrementSearchIndex(ctx, dir, path, revision, metadata); err == nil { |
| 254 | return nil |
| 255 | } |
| 256 | } |
| 257 | } |
| 258 | } |
| 259 | return rebuildSearchIndex(ctx, dir, path, sessionID, revision) |
| 260 | } |
| 261 | |
| 262 | func rebuildSearchIndex(ctx context.Context, dir, path, sessionID string, revision logRevision) error { |
| 263 | return projectiondb.Rebuild(ctx, projectiondb.OpenOptions{Path: path, Migrations: searchMigrations, RequireDisk: true, MaxOpenConns: 1, QuickCheck: true}, func(ctx context.Context, db *sql.DB) error { |
| 264 | if err := configureHistoryRebuild(ctx, db); err != nil { |
| 265 | return err |
| 266 | } |
| 267 | metadata := searchMetadata{sessionID: sessionID, storageRevision: StorageRevision, projection: searchIndexVersion, generation: randomID()} |
| 268 | return populateSearchIndex(ctx, dir, db, 0, 1, revision, metadata, searchBuildState{positions: map[string]int64{}, versions: map[string]int{}}) |
| 269 | }) |
| 270 | } |
| 271 | |
| 272 | func incrementSearchIndex(ctx context.Context, dir, path string, revision logRevision, metadata searchMetadata) error { |
| 273 | handle, err := projectiondb.Open(ctx, projectiondb.OpenOptions{Path: path, Migrations: searchMigrations, RequireDisk: true, MaxOpenConns: 1}) |
| 274 | if err != nil { |
| 275 | return err |
| 276 | } |
| 277 | defer handle.DB.Close() |
| 278 | state := searchBuildState{positions: map[string]int64{}, versions: map[string]int{}} |
| 279 | rows, err := handle.DB.QueryContext(ctx, `SELECT message_id,position,version FROM documents WHERE current=1`) |
| 280 | if err != nil { |
| 281 | return err |
| 282 | } |
| 283 | for rows.Next() { |
| 284 | var id string |
| 285 | var position int64 |
| 286 | var version int |
| 287 | if err := rows.Scan(&id, &position, &version); err != nil { |
| 288 | _ = rows.Close() |
| 289 | return err |
| 290 | } |
| 291 | state.positions[id], state.versions[id] = position, version |
| 292 | state.nextPosition = max(state.nextPosition, position) |
| 293 | } |
| 294 | if err := errors.Join(rows.Err(), rows.Close()); err != nil { |
| 295 | return err |
| 296 | } |
| 297 | versionRows, err := handle.DB.QueryContext(ctx, `SELECT message_id,MAX(version) FROM documents GROUP BY message_id`) |
| 298 | if err != nil { |
| 299 | return err |
| 300 | } |
| 301 | for versionRows.Next() { |
| 302 | var id string |
| 303 | var version int |
| 304 | if err := versionRows.Scan(&id, &version); err != nil { |
| 305 | _ = versionRows.Close() |
| 306 | return err |
| 307 | } |
| 308 | state.versions[id] = version |
| 309 | } |
| 310 | if err := errors.Join(versionRows.Err(), versionRows.Close()); err != nil { |
| 311 | return err |
| 312 | } |
| 313 | return populateSearchIndex(ctx, dir, handle.DB, metadata.logSize, metadata.durableSequence+1, revision, metadata, state) |
| 314 | } |
| 315 | |
| 316 | func populateSearchIndex(ctx context.Context, dir string, db *sql.DB, startOffset int64, nextSequence uint64, revision logRevision, metadata searchMetadata, state searchBuildState) error { |
| 317 | manifest, err := readManifest(filepath.Join(dir, "manifest.json")) |
| 318 | if err != nil { |
| 319 | return err |
| 320 | } |
| 321 | log, err := os.Open(logPathForManifest(dir, manifest)) |
| 322 | if err != nil { |
| 323 | return err |
| 324 | } |
| 325 | defer log.Close() |
| 326 | tx, err := db.BeginTx(ctx, nil) |
| 327 | if err != nil { |
| 328 | return err |
| 329 | } |
| 330 | defer func() { _ = tx.Rollback() }() |
| 331 | content := contentStoreForSessionDir(dir) |
| 332 | var buildErr error |
| 333 | progress, err := scanHistoryLog(ctx, log, startOffset, nextSequence, revision.Size, content, func(commit Commit) bool { |
| 334 | for _, event := range commit.Events { |
| 335 | if err := indexSearchEvent(ctx, tx, content, &state, event); err != nil { |
| 336 | buildErr = err |
| 337 | return false |
| 338 | } |
| 339 | } |
| 340 | return true |
| 341 | }) |
| 342 | if err != nil || buildErr != nil { |
| 343 | return errors.Join(err, buildErr) |
| 344 | } |
| 345 | values := map[string]string{ |
| 346 | "session_id": metadata.sessionID, "log_size": fmt.Sprint(progress.end), |
| 347 | "log_mtime_ns": fmt.Sprint(progress.modTimeNS), "storage_revision": fmt.Sprint(StorageRevision), |
| 348 | "projection_version": fmt.Sprint(searchIndexVersion), "durable_sequence": fmt.Sprint(progress.sequence), |
| 349 | "generation": metadata.generation, |
| 350 | } |
| 351 | for key, value := range values { |
| 352 | if _, err := tx.ExecContext(ctx, `INSERT INTO metadata(key,value) VALUES(?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, key, value); err != nil { |
| 353 | return err |
| 354 | } |
| 355 | } |
| 356 | return tx.Commit() |
| 357 | } |
| 358 | |
| 359 | func indexSearchEvent(ctx context.Context, tx *sql.Tx, content *sessioncontent.Store, state *searchBuildState, event Event) error { |
| 360 | if event.Kind != "message/complete" && event.Kind != "message/upsert" && event.Kind != "message/retract" && event.Kind != "history/replace" && event.Kind != "legacy/import" { |
| 361 | return nil |
| 362 | } |
| 363 | payload := event.Payload |
| 364 | if event.PayloadRef != nil { |
| 365 | var err error |
| 366 | payload, err = resolveContentPayload(ctx, content, *event.PayloadRef) |
| 367 | if err != nil { |
| 368 | return err |
| 369 | } |
| 370 | } |
| 371 | switch event.Kind { |
| 372 | case "message/retract": |
| 373 | ids, err := retractedMessageIDs(event, payload) |
| 374 | if err != nil { |
| 375 | return err |
| 376 | } |
| 377 | for _, id := range ids { |
| 378 | if _, err := tx.ExecContext(ctx, `UPDATE documents SET current=0,valid_to=? WHERE message_id=? AND current=1`, event.Sequence, id); err != nil { |
| 379 | return err |
| 380 | } |
| 381 | delete(state.positions, id) |
| 382 | } |
| 383 | return nil |
| 384 | case "message/complete", "message/upsert": |
| 385 | var body struct { |
| 386 | Message *provider.Message `json:"message"` |
| 387 | } |
| 388 | if err := strictPayload(payload, &body); err != nil || body.Message == nil { |
| 389 | return damagedPayload(event, err) |
| 390 | } |
| 391 | return indexSearchMessage(ctx, tx, state, *body.Message, event.Sequence, event.Kind == "message/upsert") |
| 392 | case "history/replace", "legacy/import": |
| 393 | messages, err := replacementEventMessages(event, payload) |
| 394 | if err != nil { |
| 395 | return err |
| 396 | } |
| 397 | if _, err := tx.ExecContext(ctx, `UPDATE documents SET current=0,valid_to=? WHERE current=1`, event.Sequence); err != nil { |
| 398 | return err |
| 399 | } |
| 400 | state.positions = map[string]int64{} |
| 401 | state.nextPosition = 0 |
| 402 | for _, message := range messages { |
| 403 | if err := indexSearchMessage(ctx, tx, state, message, event.Sequence, false); err != nil { |
| 404 | return err |
| 405 | } |
| 406 | } |
| 407 | } |
| 408 | return nil |
| 409 | } |
| 410 | |
| 411 | func indexSearchMessage(ctx context.Context, tx *sql.Tx, state *searchBuildState, message provider.Message, sequence uint64, upsert bool) error { |
| 412 | id := strings.TrimSpace(message.ID) |
| 413 | if id == "" { |
| 414 | return errors.New("session: search document has no stable message id") |
| 415 | } |
| 416 | position, exists := state.positions[id] |
| 417 | if !exists { |
| 418 | state.nextPosition++ |
| 419 | position = state.nextPosition |
| 420 | state.positions[id] = position |
| 421 | } else if !upsert { |
| 422 | // A repeated message/complete keeps the id's first message. |
| 423 | return nil |
| 424 | } |
| 425 | if exists { |
| 426 | if _, err := tx.ExecContext(ctx, `UPDATE documents SET current=0,valid_to=? WHERE message_id=? AND current=1`, sequence, id); err != nil { |
| 427 | return err |
| 428 | } |
| 429 | } |
| 430 | version := state.versions[id] + 1 |
| 431 | state.versions[id] = version |
| 432 | _, err := tx.ExecContext(ctx, `INSERT INTO documents(message_id,version,position,event_sequence,valid_to,role,preview,text,current) VALUES(?,?,?,?,0,?,?,?,1)`, id, version, position, sequence, string(message.Role), messagePreview(message), messageSearchText(message)) |
| 433 | return err |
| 434 | } |
| 435 |