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 err := forEachStrictJSONLLineContext(ctx, r, func(line []byte, lineNo int) error {
33 row, err := decodeSessionLogRow(line)
34 if err != nil {
35 return fmt.Errorf("line %d: %w", lineNo, err)
36 }
37 if !sawMeta && row.typeName != "meta" {
38 return fmt.Errorf("line %d: first row is not meta", lineNo)
39 }
40 if row.typeName == "meta" {
41 sawMeta = true
42 }
43 return fn(row)
44 })
45 if err != nil {
46 return err
47 }
48 if !sawMeta {
49 return fmt.Errorf("file is empty")
50 }
51 return nil
52}
53
54func decodeSessionLogRow(line []byte) (sessionLogRow, error) {
55 var wire struct {
56 Type string `json:"type"`
57 Meta *SessionMeta `json:"meta"`
58 Message json.RawMessage `json:"message"`
59 Messages json.RawMessage `json:"messages"`
60 Cumulative *provider.Usage `json:"cumulative"`
61 Title *string `json:"title"`
62 Extension string `json:"extension"`
63 State json.RawMessage `json:"state"`
64 }
65 if err := json.Unmarshal(line, &wire); err != nil {
66 return sessionLogRow{}, fmt.Errorf("invalid JSON: %w", err)
67 }
68 if wire.Type == "" {
69 return sessionLogRow{}, fmt.Errorf("missing row type")
70 }
71
72 row := sessionLogRow{
73 typeName: wire.Type,
74 cumulative: wire.Cumulative,
75 title: wire.Title,
76 extension: wire.Extension,
77 state: wire.State,
78 }
79 switch wire.Type {
80 case "meta":
81 if wire.Meta == nil || wire.Meta.ID == "" {
82 return sessionLogRow{}, fmt.Errorf("meta row has no session id")
83 }
84 row.meta = wire.Meta
85 case "message":
86 if len(wire.Message) == 0 || bytes.Equal(bytes.TrimSpace(wire.Message), []byte("null")) {
87 return sessionLogRow{}, fmt.Errorf("invalid message row: message row has no message")
88 }
89 message, err := hydrateMessageObject(wire.Message)
90 if err != nil {
91 return sessionLogRow{}, fmt.Errorf("invalid message row: %w", err)
92 }
93 row.message = &message
94 case "compaction":
95 messages, err := hydrateCompactionMessages(wire.Messages)
96 if err != nil {
97 return sessionLogRow{}, fmt.Errorf("invalid compaction row: %w", err)
98 }
99 row.messages = messages
100 case "usage":
101 if wire.Cumulative == nil {
102 return sessionLogRow{}, fmt.Errorf("usage row has no cumulative usage")
103 }
104 case "rename":
105 if wire.Title == nil {
106 return sessionLogRow{}, fmt.Errorf("rename row has no title")
107 }
108 case "extension_state":
109 // Invalid or oversized extension snapshots are ignored by reducers for
110 // compatibility with existing sessions; transcript rows stay strict.
111 default:
112 return sessionLogRow{}, fmt.Errorf("unknown row type %q", wire.Type)
113 }
114 return row, nil
115}
116
117func hydrateCompactionMessages(raw json.RawMessage) ([]provider.Message, error) {
118 if len(raw) == 0 || bytes.Equal(bytes.TrimSpace(raw), []byte("null")) {
119 // Older writers omitted the field for an empty compaction and some
120 // wrote it as null. Both represent a valid empty replacement.
121 return []provider.Message{}, nil
122 }
123 var rawMessages []json.RawMessage
124 if err := json.Unmarshal(raw, &rawMessages); err != nil {
125 return nil, fmt.Errorf("invalid messages: %w", err)
126 }
127 messages := make([]provider.Message, 0, len(rawMessages))
128 for idx, rawMessage := range rawMessages {
129 message, err := hydrateMessageObject(rawMessage)
130 if err != nil {
131 return nil, fmt.Errorf("message %d: %w", idx, err)
132 }
133 messages = append(messages, message)
134 }
135 return messages, nil
136}