old/data-pipeline-roadmap.md

Data Pipeline & Streaming Services Upgrade Roadmap

Executive Summary

This document outlines a technical roadmap to upgrade the data pipelines and streaming services of the CLI Proxy API. The current implementation suffers from significant buffering bottlenecks, excessive I/O operations (double-writes), and scalability limits in log retrieval. The proposed upgrades focus on zero-copy streaming, asynchronous processing, and optimized data persistence.

Current Architecture Assessment

1. Proxy & Streaming Service

Current State:

• Buffering: The proxy buffers the entire response body in memory for non-streaming responses (and gzip streams) to perform content rewriting.
• Gzip Handling: internal/api/modules/amp/proxy.go reads the full body into memory to decompress it if Content-Encoding is present, defeating the purpose of streaming for compressed upstream responses.
• Response Rewriting: ResponseRewriter buffers non-streaming bodies entirely. For SSE, it attempts to parse chunks individually, which is brittle if JSON tokens span across network chunk boundaries.

Bottlenecks:

• High memory pressure during large response payloads.
• Increased Time-To-First-Byte (TTFB) due to buffering.
• Potential data corruption in SSE if network chunks split data: lines or JSON fields.

2. Data Pipeline (Logging)

Current State:

• Write Path: FileStreamingLogWriter writes chunks to a temporary file asynchronously. However, Close() triggers a synchronous "assembly" phase that reads the temp file back and writes it to the final log file. This results in 2x Disk I/O (Write Temp -> Read Temp -> Write Final).
• Read Path: LogRepository scans all files in the log directory to build a list or find logs. Reading a specific log involves iterating through lines in memory (logAccumulator).

Bottlenecks:

• Double I/O penalty for every logged request.
• Log retrieval performance degrades linearly (O(N)) with the number of log files.
• Synchronous blocking on file system operations during request finalization.

---

Technical Roadmap

Phase 1: Zero-Buffer Streaming Proxy

Goal: Eliminate memory buffering in the proxy layer to minimize latency and memory footprint.

1.1 Streaming Decompression

• Task: Refactor proxy.go to use a streaming gzip.Reader (or brotli/zstd wrappers) that wraps the http.Response.Body.
• Implementation: Create a DecompressingReadCloser that transparently decompresses as Read() is called, rather than pre-reading the whole body.
• Benefit: Constant memory usage regardless of response size.

1.2 Streaming Response Rewriter

• Task: Rewrite ResponseRewriter to use a streaming JSON parser (e.g., json.Decoder or a token-based replacer) instead of gjson/sjson on full buffers.
• Implementation:
  • Create a TokenReplacingReader that scans the stream for specific keys (model, modelVersion) and replaces values on the fly.
  • Ensure it maintains state across Read() calls to handle tokens split across buffer boundaries.
• Benefit: Zero-latency overhead for model name rewriting; safe for large JSON bodies.

Phase 2: Robust SSE Handling

Goal: Ensure 100% reliability for streaming AI responses (Server-Sent Events).

2.1 Stateful SSE Parser

• Task: Replace the naive line-splitting logic in response_rewriter.go.
• Implementation:
  • Implement a state machine that buffers only incomplete lines.
  • Process full data: {...} lines as they become available.
  • Handle multi-line JSON data correctly.
• Benefit: Prevents corruption when network packets fragment SSE messages.

Phase 3: High-Performance Logging Pipeline

Goal: Decouple logging from request latency and reduce I/O.

3.1 Eliminate Double-Writes

• Task: Redesign the log storage format to allow append-only writing without post-request assembly.
• Implementation:
  • Change log format to a structured line-based JSON (NDJSON) or a format that doesn't require a specific "header-first, body-second" physical layout if possible.
  • Alternatively, keep the temp file approach but use sendfile (via io.Copy optimizations) to merge files efficiently, or just move/rename the temp file to the final location if the order can be adjusted.
• Recommendation: Switch to a directory-per-request or a pure append-only log file where request metadata and body chunks are interleaved but tagged with a Request ID. This allows writing directly to the final destination.

3.2 Async Log Persister

• Task: Move file I/O entirely out of the request context.
• Implementation:
  • A background worker pool receives LogEntry objects (metadata, body chunks) via a buffered channel.
  • Workers handle file opening/writing/closing independently of the HTTP handler.
• Benefit: Zero impact of disk latency on API response times.

Phase 4: Scalable Data Access

Goal: Make log retrieval instant regardless of history size.

4.1 Indexing Strategy

• Task: Stop scanning all files for listing/searching.
• Implementation:
  • Maintain a lightweight index.json or SQLite DB that tracks: RequestID, Timestamp, Path, StatusCode, Filename.
  • Update the index asynchronously when logs are finalized.
• Benefit: O(1) lookup by Request ID; O(log N) lookup by time range.

4.2 Optimized Reader

• Task: Read logs efficiently.
• Implementation:
  • When tailing logs (latest), read the file backwards from the end (using Seek) rather than scanning from the start.
  • Implement pagination for log listing based on the index.

---

Execution Plan

1. Step 1 (Critical): Fix the Proxy buffering. This is the biggest risk for production stability.

   • Refactor proxy.go gzip handling.
   • Refactor ResponseRewriter for streaming JSON.

2. Step 2 (Reliability): Fix SSE parsing in ResponseRewriter.

   • Implement stateful line buffering.

3. Step 3 (Performance): Optimize Log Writing.

   • Refactor RequestLogger to avoid double-write.

4. Step 4 (Scalability): Implement Log Indexing.

   • Add LogIndexService and update LogRepository to use it.