1package core
2
3import (
4 "bytes"
5 "context"
6 "encoding/json"
7 "fmt"
8 "io"
9
10 "github.com/bnema/zut/packages/provider"
11)
12
13// sessionLogRow is the validated, hydrated representation shared by complete
14// session readers. It keeps the JSONL format private while giving each reader
15// one typed stream to reduce into its own projection.
16type sessionLogRow struct {
17 typeName string
18 meta *SessionMeta
19 message *provider.Message
20 messages []provider.Message
21 cumulative *provider.Usage
22 title *string
23 extension string
24 state json.RawMessage
25}
26
27// forEachSessionLogRowContext owns the strict persisted-row contract: JSONL
28// framing, the leading meta row, known row types, required fields, and message
29// content hydration. Callers only implement projection-specific reduction.
30func forEachSessionLogRowContext(ctx context.Context, r io.Reader, fn func(sessionLogRow) error) error {
31 sawMeta := false
32 sessionID := ""
33 err := forEachStrictJSONLLineContext(ctx, r, func(line []byte, lineNo int) error {
34 row, err := decodeSessionLogRow(line)
35 if err != nil {
36 return fmt.Errorf("line %d: %w", lineNo, err)
37 }
38 if !sawMeta && row.typeName != "meta" {
39 return fmt.Errorf("line %d: first row is not meta", lineNo)
40 }
41 if row.typeName == "meta" {
42 if sessionID != "" && row.meta.ID != sessionID {
43 return fmt.Errorf("line %d: session id changed from %q to %q", lineNo, sessionID, row.meta.ID)
44 }
45 sessionID = row.meta.ID
46 sawMeta = true
47 }
48 return fn(row)
49 })
50 if err != nil {
51 return err
52 }
53 if !sawMeta {
54 return fmt.Errorf("file is empty")
55 }
56 return nil
57}
58
59func decodeSessionLogRow(line []byte) (sessionLogRow, error) {
60 var wire struct {
61 Type string `json:"type"`
62 Meta *SessionMeta `json:"meta"`
63 Message json.RawMessage `json:"message"`
64 Messages json.RawMessage `json:"messages"`
65 Cumulative *provider.Usage `json:"cumulative"`
66 Title *string `json:"title"`
67 Extension string `json:"extension"`
68 State json.RawMessage `json:"state"`
69 }
70 if err := json.Unmarshal(line, &wire); err != nil {
71 return sessionLogRow{}, fmt.Errorf("invalid JSON: %w", err)
72 }
73 if wire.Type == "" {
74 return sessionLogRow{}, fmt.Errorf("missing row type")
75 }
76
77 row := sessionLogRow{
78 typeName: wire.Type,
79 cumulative: wire.Cumulative,
80 title: wire.Title,
81 extension: wire.Extension,
82 state: wire.State,
83 }
84 switch wire.Type {
85 case "meta":
86 if wire.Meta == nil || wire.Meta.ID == "" {
87 return sessionLogRow{}, fmt.Errorf("meta row has no session id")
88 }
89 row.meta = wire.Meta
90 case "message":
91 if len(wire.Message) == 0 || bytes.Equal(bytes.TrimSpace(wire.Message), []byte("null")) {
92 return sessionLogRow{}, fmt.Errorf("invalid message row: message row has no message")
93 }
94 message, err := hydrateMessageObject(wire.Message)
95 if err != nil {
96 return sessionLogRow{}, fmt.Errorf("invalid message row: %w", err)
97 }
98 row.message = &message
99 case "compaction":
100 messages, err := hydrateCompactionMessages(wire.Messages)
101 if err != nil {
102 return sessionLogRow{}, fmt.Errorf("invalid compaction row: %w", err)
103 }
104 row.messages = messages
105 case "usage":
106 if wire.Cumulative == nil {
107 return sessionLogRow{}, fmt.Errorf("usage row has no cumulative usage")
108 }
109 case "rename":
110 if wire.Title == nil {
111 return sessionLogRow{}, fmt.Errorf("rename row has no title")
112 }
113 case "extension_state":
114 // Invalid or oversized extension snapshots are ignored by reducers for
115 // compatibility with existing sessions; transcript rows stay strict.
116 default:
117 return sessionLogRow{}, fmt.Errorf("unknown row type %q", wire.Type)
118 }
119 return row, nil
120}
121
122func hydrateCompactionMessages(raw json.RawMessage) ([]provider.Message, error) {
123 if len(raw) == 0 || bytes.Equal(bytes.TrimSpace(raw), []byte("null")) {
124 // Older writers omitted the field for an empty compaction and some
125 // wrote it as null. Both represent a valid empty replacement.
126 return []provider.Message{}, nil
127 }
128 var rawMessages []json.RawMessage
129 if err := json.Unmarshal(raw, &rawMessages); err != nil {
130 return nil, fmt.Errorf("invalid messages: %w", err)
131 }
132 messages := make([]provider.Message, 0, len(rawMessages))
133 for idx, rawMessage := range rawMessages {
134 message, err := hydrateMessageObject(rawMessage)
135 if err != nil {
136 return nil, fmt.Errorf("message %d: %w", idx, err)
137 }
138 messages = append(messages, message)
139 }
140 return messages, nil
141}