| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "hash" |
| 11 | "os" |
| 12 | |
| 13 | "reasonix/internal/fileops" |
| 14 | "reasonix/internal/store" |
| 15 | ) |
| 16 | |
| 17 | func restoreEventPagerScan(ctx context.Context, db *sql.DB, sourceKey string, size int64) (eventPagerScanProgress, error) { |
| 18 | p := eventPagerScanProgress{Version: 1, SourceKey: sourceKey} |
| 19 | _, err := db.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS event_locations(position INTEGER PRIMARY KEY,offset INTEGER NOT NULL,length INTEGER NOT NULL,at INTEGER NOT NULL); |
| 20 | CREATE TABLE IF NOT EXISTS event_pending(position INTEGER PRIMARY KEY,offset INTEGER NOT NULL,length INTEGER NOT NULL);`) |
| 21 | if err != nil { |
| 22 | return p, err |
| 23 | } |
| 24 | var raw string |
| 25 | err = db.QueryRowContext(ctx, `SELECT value FROM metadata WHERE key='event_scan_progress'`).Scan(&raw) |
| 26 | if err != nil && !errors.Is(err, sql.ErrNoRows) { |
| 27 | return p, err |
| 28 | } |
| 29 | var saved eventPagerScanProgress |
| 30 | valid := err == nil && json.Unmarshal([]byte(raw), &saved) == nil && saved.Version == 1 && saved.SourceKey == sourceKey && |
| 31 | saved.Offset >= 0 && saved.Offset <= size && saved.Count >= 0 && saved.Pending >= 0 && len(saved.Record.Messages) == 0 && |
| 32 | (!saved.InMessages && saved.Pending == 0 || saved.InMessages && !saved.Done && saved.Pending > 0) |
| 33 | for _, table := range []struct { |
| 34 | name string |
| 35 | rows int |
| 36 | }{{"event_locations", saved.Count}, {"event_pending", saved.Pending}} { |
| 37 | var count, first, last int |
| 38 | var end int64 |
| 39 | err := db.QueryRowContext(ctx, `SELECT COUNT(*),COALESCE(MIN(position),0),COALESCE(MAX(position),-1),COALESCE(MAX(offset+length),0) FROM `+table.name).Scan(&count, &first, &last, &end) |
| 40 | if err != nil { |
| 41 | return p, err |
| 42 | } |
| 43 | valid = valid && count == table.rows && first == 0 && last == count-1 && end <= saved.Offset |
| 44 | } |
| 45 | if valid { |
| 46 | return saved, nil |
| 47 | } |
| 48 | // Without a matching durable parser position, no rows from an interrupted |
| 49 | // build may be recognized as a validated prefix of the source. |
| 50 | _, err = db.ExecContext(ctx, `DELETE FROM event_locations; DELETE FROM event_pending; DELETE FROM entries; DELETE FROM metadata`) |
| 51 | return p, err |
| 52 | } |
| 53 | |
| 54 | func validateEventPagerFile(source string, file *os.File, fingerprint string) error { |
| 55 | key, err := displayImportFileKey(store.SessionEventLog(source), file) |
| 56 | if err != nil { |
| 57 | return err |
| 58 | } |
| 59 | info, err := os.Stat(source) |
| 60 | if err != nil { |
| 61 | return err |
| 62 | } |
| 63 | target, version := fileops.DiskSnapshot(source, info) |
| 64 | if fmt.Sprintf("%s:%s:event:%s:schema1", target.Key, version, key) != fingerprint { |
| 65 | return ErrDisplaySourceChanged |
| 66 | } |
| 67 | return nil |
| 68 | } |
| 69 | |
| 70 | func restoreEventProjection(ctx context.Context, db *sql.DB, idx *SessionDisplayIndex, users *int, digest hash.Hash, count int) error { |
| 71 | var raw string |
| 72 | err := db.QueryRowContext(ctx, `SELECT value FROM metadata WHERE key='event_projection_progress'`).Scan(&raw) |
| 73 | if err != nil && !errors.Is(err, sql.ErrNoRows) { |
| 74 | return err |
| 75 | } |
| 76 | var saved checkpointPagerProgress |
| 77 | valid := err == nil && json.Unmarshal([]byte(raw), &saved) == nil && saved.Header.SchemaVersion == SessionDisplayIndexSchemaVersion && |
| 78 | saved.Header.TranscriptSize == idx.TranscriptSize && saved.Header.Entries == nil && saved.Header.MessageCount >= 0 && |
| 79 | saved.Header.MessageCount <= count && saved.Users >= 0 && saved.Users <= saved.Header.MessageCount && |
| 80 | saved.Header.AuthoredTurns >= 0 && saved.Header.AuthoredTurns <= saved.Header.MessageCount |
| 81 | var rows, first, last int |
| 82 | if err := db.QueryRowContext(ctx, `SELECT COUNT(*),COALESCE(MIN(position),0),COALESCE(MAX(position),-1) FROM entries`).Scan(&rows, &first, &last); err != nil { |
| 83 | return err |
| 84 | } |
| 85 | valid = valid && rows == saved.Header.MessageCount && first == 0 && last == rows-1 |
| 86 | if valid && digest.(encoding.BinaryUnmarshaler).UnmarshalBinary(saved.Hash) == nil { |
| 87 | *idx, *users = saved.Header, saved.Users |
| 88 | return nil |
| 89 | } |
| 90 | digest.Reset() |
| 91 | _, err = db.ExecContext(ctx, `DELETE FROM entries; DELETE FROM metadata WHERE key='event_projection_progress'`) |
| 92 | return err |
| 93 | } |
| 94 | |
| 95 | func saveEventProjection(ctx context.Context, tx *sql.Tx, idx SessionDisplayIndex, users int, digest hash.Hash) error { |
| 96 | state, err := digest.(encoding.BinaryMarshaler).MarshalBinary() |
| 97 | if err != nil { |
| 98 | return err |
| 99 | } |
| 100 | body, err := json.Marshal(checkpointPagerProgress{Header: idx, Users: users, Hash: state}) |
| 101 | if err != nil { |
| 102 | return err |
| 103 | } |
| 104 | _, err = tx.ExecContext(ctx, `INSERT OR REPLACE INTO metadata VALUES('event_projection_progress',?)`, string(body)) |
| 105 | return err |
| 106 | } |
| 107 |