internal/auditlog/stream_wrapper.go (1)

101-109: ⚠️ Potential issue | 🟠 Major

SSE framing/parser is too strict and can drop events while growing memory.

Line 103 only matches \n\n, and Line 125 only accepts data: (with space). Valid CRLF-framed streams or data: lines can be missed, leaving data stuck in pending and causing unbounded growth.

🔧 Proposed hardening patch

+const maxPendingSSEBytes = 1 << 20 // 1 MiB safety cap
+
 func (w *StreamLogWrapper) processSSEData(data []byte) {
 	// Prepend any pending data from previous read
 	if len(w.pending) > 0 {
 		data = append(w.pending, data...)
 		w.pending = nil
 	}
+
+	// Accept both LF and CRLF framing.
+	data = bytes.ReplaceAll(data, []byte("\r\n"), []byte("\n"))
 
 	// Split on double newline (SSE event separator)
 	for {
 		idx := bytes.Index(data, []byte("\n\n"))
 		if idx == -1 {
 			// No complete event, save as pending
 			if len(data) > 0 {
-				w.pending = make([]byte, len(data))
-				copy(w.pending, data)
+				if len(data) > maxPendingSSEBytes {
+					data = data[len(data)-maxPendingSSEBytes:]
+				}
+				w.pending = append(w.pending[:0], data...)
 			}
 			return
 		}
@@
 func (w *StreamLogWrapper) processSSEEvent(event []byte) {
 	// Find the data line
 	lines := bytes.Split(event, []byte("\n"))
 	for _, line := range lines {
-		if bytes.HasPrefix(line, []byte("data: ")) {
-			jsonData := bytes.TrimPrefix(line, []byte("data: "))
+		if bytes.HasPrefix(line, []byte("data:")) {
+			jsonData := bytes.TrimPrefix(line, []byte("data:"))
+			jsonData = bytes.TrimPrefix(jsonData, []byte(" "))
 			// Skip [DONE] marker
 			if bytes.Equal(jsonData, []byte("[DONE]")) {
 				continue
 			}
 			w.parseEventJSON(jsonData)
 		}
 	}
 }

Also applies to: 125-126

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@internal/auditlog/stream_wrapper.go` around lines 101 - 109, The SSE framing
is too strict: update the loop that searches for the event separator to accept
both "\n\n" and CRLF "\r\n\r\n" (e.g., check for either sequence or normalize
CRLF to LF before splitting) and when parsing "data" lines accept both "data:"
and "data: " (allow optional space and trim any trailing '\r'); update the logic
that saves incomplete bytes into w.pending to reuse the existing slice capacity
instead of always allocating a new slice (for example use w.pending =
append(w.pending[:0], data...) or copy into a retained buffer) so partial events
are not dropped and pending cannot grow unbounded; refer to the streaming loop
and the w.pending field and the code that handles data line parsing to make
these changes.

internal/usage/store_postgresql.go (1)

122-128: ⚠️ Potential issue | 🟠 Major

Only start a transaction once the batch spans multiple INSERTs.

After this refactor, anything up to usageInsertMaxRowsPerQuery rows is still a single multi-row INSERT, so batches of 10-4369 entries now pay BEGIN/COMMIT overhead without gaining atomicity. A single PostgreSQL statement is already atomic.

⚙️ Suggested threshold change

- // For larger batches, use a transaction to ensure atomicity
- // For smaller batches, use individual inserts without transaction overhead
- if len(entries) < 10 {
+ // A single multi-row INSERT is already atomic. Only start a transaction
+ // when the batch must be split across multiple statements.
+ if len(entries) <= usageInsertMaxRowsPerQuery {
 		return s.writeBatchSmall(ctx, entries)
 	}

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@internal/usage/store_postgresql.go` around lines 122 - 128, The batch-size
check currently switches to a transaction for any batch with len(entries) >= 10;
change it to only start a transaction when the batch cannot fit in a single
multi-row INSERT by comparing against usageInsertMaxRowsPerQuery instead: call
writeBatchSmall(ctx, entries) when len(entries) <= usageInsertMaxRowsPerQuery
(single-statement atomic insert) and call writeBatchLarge(ctx, entries)
otherwise (requires BEGIN/COMMIT). Update the conditional and adjust the comment
to reflect that single-statement inserts are atomic up to
usageInsertMaxRowsPerQuery; keep the existing writeBatchSmall and
writeBatchLarge call sites unchanged.

internal/auditlog/store_postgresql.go (1)

127-133: ⚠️ Potential issue | 🟠 Major

Only start a transaction once the batch spans multiple INSERTs.

After this refactor, anything up to auditLogInsertMaxRowsPerQuery rows is still a single multi-row INSERT, so batches of 10-4369 entries now pay BEGIN/COMMIT overhead without gaining atomicity. A single PostgreSQL statement is already atomic.

⚙️ Suggested threshold change

- // For larger batches, use a transaction to ensure atomicity
- // For smaller batches, use individual inserts without transaction overhead
- if len(entries) < 10 {
+ // A single multi-row INSERT is already atomic. Only start a transaction
+ // when the batch must be split across multiple statements.
+ if len(entries) <= auditLogInsertMaxRowsPerQuery {
 		return s.writeBatchSmall(ctx, entries)
 	}

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@internal/auditlog/store_postgresql.go` around lines 127 - 133, The decision
to start a transaction should be based on whether the batch requires multiple
INSERT statements, not a fixed small-number threshold. Compute insertsNeeded =
ceil(len(entries) / auditLogInsertMaxRowsPerQuery) and only call the
transactional path (s.writeBatchLarge) when insertsNeeded > 1; otherwise call
the non-transactional single-statement path (s.writeBatchSmall). Update the
branching logic where the current len(entries) < 10 check lives so it uses
auditLogInsertMaxRowsPerQuery and the computed insertsNeeded instead.