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.