| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/sha256" |
| 6 | "encoding/json" |
| 7 | "os" |
| 8 | "time" |
| 9 | |
| 10 | "reasonix/internal/provider" |
| 11 | |
| 12 | fileencoding "reasonix/internal/fileutil/encoding" |
| 13 | "reasonix/internal/store" |
| 14 | ) |
| 15 | |
| 16 | // SessionHeadIndex is the schema-2 shape of <id>.event-index.json: a snapshot |
| 17 | // of the heads so listings and the catalog never replay a log. It is trusted |
| 18 | // only while LogSize and LogGeneration still match the log. |
| 19 | type SessionHeadIndex struct { |
| 20 | SchemaVersion int `json:"schema_version"` |
| 21 | LogSize int64 `json:"log_size"` |
| 22 | LogGeneration int64 `json:"log_generation"` |
| 23 | SelectedHead string `json:"selected_head"` |
| 24 | MessageCount int `json:"message_count"` |
| 25 | ContentDigest string `json:"content_digest"` |
| 26 | WriterID string `json:"writer_id"` |
| 27 | UpdatedAt time.Time `json:"updated_at"` |
| 28 | Heads []SessionHead `json:"heads"` |
| 29 | } |
| 30 | |
| 31 | // Current reports whether the index still describes the log on disk. |
| 32 | func (idx *SessionHeadIndex) Current(sessionPath string) bool { |
| 33 | if idx == nil { |
| 34 | return false |
| 35 | } |
| 36 | info, err := os.Stat(store.SessionEventLog(sessionPath)) |
| 37 | if err != nil || info.IsDir() || info.Size() != idx.LogSize { |
| 38 | return false |
| 39 | } |
| 40 | header, ok, err := readSessionDAGHeader(sessionPath) |
| 41 | return err == nil && ok && header.generation == idx.LogGeneration |
| 42 | } |
| 43 | |
| 44 | // ReadSessionHeadIndex loads the schema-2 index. A missing file or a schema-1 |
| 45 | // index yields nil, nil so callers fall back to replay. |
| 46 | func ReadSessionHeadIndex(sessionPath string) (*SessionHeadIndex, error) { |
| 47 | path := store.SessionEventIndex(sessionPath) |
| 48 | if path == "" { |
| 49 | return nil, nil |
| 50 | } |
| 51 | b, err := fileencoding.ReadFileUTF8(path) |
| 52 | if err != nil { |
| 53 | if os.IsNotExist(err) { |
| 54 | return nil, nil |
| 55 | } |
| 56 | return nil, err |
| 57 | } |
| 58 | var idx SessionHeadIndex |
| 59 | if err := json.Unmarshal(b, &idx); err != nil { |
| 60 | return nil, err |
| 61 | } |
| 62 | if idx.SchemaVersion != sessionDAGSchemaVersion { |
| 63 | return nil, nil |
| 64 | } |
| 65 | if idx.Heads == nil { |
| 66 | idx.Heads = []SessionHead{} |
| 67 | } |
| 68 | return &idx, nil |
| 69 | } |
| 70 | |
| 71 | func writeSessionDAGIndex(ctx context.Context, sessionPath string, st *sessionDAGState) error { |
| 72 | indexPath := store.SessionEventIndex(sessionPath) |
| 73 | if indexPath == "" || st == nil { |
| 74 | return nil |
| 75 | } |
| 76 | if err := ctx.Err(); err != nil { |
| 77 | return err |
| 78 | } |
| 79 | selected := st.selectedHead() |
| 80 | msgs, _ := st.materialize(selected) |
| 81 | digest, err := digestSessionMessages(msgs) |
| 82 | if err != nil { |
| 83 | return err |
| 84 | } |
| 85 | idx := SessionHeadIndex{ |
| 86 | SchemaVersion: sessionDAGSchemaVersion, |
| 87 | LogSize: st.size, |
| 88 | LogGeneration: st.generation, |
| 89 | SelectedHead: selected, |
| 90 | MessageCount: len(msgs), |
| 91 | ContentDigest: digestString(digest), |
| 92 | WriterID: SessionWriterID(), |
| 93 | UpdatedAt: time.Now().UTC(), |
| 94 | Heads: st.headList(), |
| 95 | } |
| 96 | b, err := marshalJSONIndentContext(ctx, idx) |
| 97 | if err != nil { |
| 98 | return err |
| 99 | } |
| 100 | b = append(b, '\n') |
| 101 | return atomicWriteFileContext(ctx, indexPath, ".session-event-index.*.tmp", "event-index", b, 0o600, false) |
| 102 | } |
| 103 | |
| 104 | // refreshSessionEventIndexContext rewrites the event index in the schema the |
| 105 | // log actually uses: a schema-2 log gets its head index replayed, a schema-1 |
| 106 | // log the listing index. A listing repair must never downgrade a head index. |
| 107 | func refreshSessionEventIndexContext(ctx context.Context, sessionPath string, msgs []provider.Message, digest [sha256.Size]byte, revision int64) error { |
| 108 | probe, err := probeSessionEventLog(sessionPath) |
| 109 | if err != nil { |
| 110 | return err |
| 111 | } |
| 112 | if !probe.dag { |
| 113 | return writeSessionEventIndexContext(ctx, sessionPath, msgs, digest, revision) |
| 114 | } |
| 115 | st, err := replaySessionDAG(ctx, store.SessionEventLog(sessionPath), defaultSessionReplayLimits) |
| 116 | if err != nil { |
| 117 | return err |
| 118 | } |
| 119 | return writeSessionDAGIndex(ctx, sessionPath, st) |
| 120 | } |
| 121 |