# SSE 流式传输

## 概述

Server-Sent Events (SSE) 是 PR-Helper 的核心通信机制，用于实时推送克隆进度、PR 生成和代码审查结果。

## 为什么选择 SSE

### 对比 WebSocket

| 特性 | SSE | WebSocket |
|------|-----|-----------|
| 方向 | 单向（服务器→客户端） | 双向 |
| 协议 | HTTP | 独立协议 |
| 实现 | 简单 | 复杂 |
| 重连 | 自动 | 手动 |
| 兼容性 | 好 | 需要升级 |

### 适用场景

- 服务器推送数据到客户端
- 不需要客户端频繁发送数据
- 需要 HTTP 兼容性
- 需要自动重连

## SSE 协议

### 基本格式

```
event: message_type
data: {"key": "value"}

```

- `event:` 事件类型（可选）
- `data:` 事件数据（JSON 字符串）
- 空行分隔事件

### 多行数据

```
event: content
data: {"content": "第一行\n第二行"}
```

## 后端实现

### 设置响应头

```go
c.Header("Content-Type", "text/event-stream")
c.Header("Cache-Control", "no-cache")
c.Header("Connection", "keep-alive")
c.Header("X-Accel-Buffering", "no")
c.Status(http.StatusOK)
```

### 发送事件

```go
sendEvent := func(event string, data interface{}) {
    jsonData, err := json.Marshal(data)
    if err != nil {
        jsonData = []byte(`{"error":"序列化事件数据失败"}`)
    }
    fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", event, jsonData)
    flusher.Flush()
}
```

### 获取 Flusher

```go
flusher, ok := c.Writer.(http.Flusher)
if !ok {
    c.JSON(http.StatusInternalServerError, gin.H{"error": "服务器不支持流式传输"})
    return
}
```

## 前端实现

### SSE 客户端

```javascript
const SSE = {
    async post(url, body, handlers = {}) {
        const controller = new AbortController();

        const run = async () => {
            try {
                const resp = await fetch(url, {
                    method: 'POST',
                    headers: { 'Content-Type': 'application/json' },
                    credentials: 'same-origin',
                    body: JSON.stringify(body),
                    signal: controller.signal,
                });

                if (!resp.ok) {
                    const errText = await resp.text();
                    // 处理错误...
                    return;
                }

                const reader = resp.body.getReader();
                const decoder = new TextDecoder();
                let buffer = '';
                let currentEvent = '';

                while (true) {
                    const { done, value } = await reader.read();
                    if (done) break;

                    buffer += decoder.decode(value, { stream: true });
                    const lines = buffer.split('\n');
                    buffer = lines.pop() || '';

                    for (const line of lines) {
                        if (line.startsWith('event: ')) {
                            currentEvent = line.slice(7).trim();
                        } else if (line.startsWith('data: ')) {
                            const raw = line.slice(6);
                            let data;
                            try {
                                data = JSON.parse(raw);
                            } catch (_) {
                                data = raw;
                            }

                            if (currentEvent && handlers[currentEvent]) {
                                handlers[currentEvent](data);
                            }
                        }
                    }
                }

                if (handlers.done) handlers.done();
            } catch (err) {
                if (err.name === 'AbortError') return;
                if (handlers.error) handlers.error({ message: err.message });
            }
        };

        run();
        return { abort: () => controller.abort() };
    }
};
```

### 使用示例

```javascript
SSE.post('/api/repos/1/review', {
    base: 'main',
    head: 'feature',
    top_n: 20,
    concurrency: 5
}, {
    // 开始事件
    start: (data) => {
        console.log(`开始审查 ${data.total_files} 个文件`);
        updateProgress(0, data.reviewed_files);
    },

    // 文件开始
    file_start: (data) => {
        console.log(`正在审查 ${data.file} (${data.index}/${data.total})`);
        highlightCurrentFile(data.file);
    },

    // 内容流
    content: (data) => {
        appendToOutput(data.content);
    },

    // 审查建议
    suggestion: (data) => {
        showSuggestion(data.file, data.severity, data.content);
    },

    // 文件结束
    file_end: (data) => {
        markFileComplete(data.file);
    },

    // 汇总
    summary: (data) => {
        showSummary(data.score, data.overall, data.findings, data.recommendations);
    },

    // 分析保存
    analysis_saved: (data) => {
        console.log(`分析已保存，ID: ${data.analysis_id}`);
    },

    // 进度
    progress: (data) => {
        if (data.step === 'generating_summary') {
            showSpinner('生成汇总中...');
        }
    },

    // 完成
    done: () => {
        console.log('审查完成');
        hideSpinner();
    },

    // 错误
    error: (data) => {
        console.error('错误:', data.message);
        showError(data.message);
    }
});
```

## 事件类型

### PR 生成事件

| 事件 | 数据 | 说明 |
|------|------|------|
| `content` | `{content: string}` | Markdown 片段 |
| `done` | `{content: ""}` | 完成 |

### 代码审查事件

| 事件 | 数据 | 说明 |
|------|------|------|
| `start` | `{total_files, reviewed_files, top_n}` | 开始 |
| `file_start` | `{file, index, total}` | 文件开始 |
| `content` | `{content: string}` | LLM 输出 |
| `suggestion` | `{file, severity, content}` | 建议 |
| `file_end` | `{file}` | 文件结束 |
| `summary` | `{score, overall, findings, recommendations}` | 汇总 |
| `progress` | `{step: string}` | 进度 |
| `error` | `{message: string}` | 错误 |
| `analysis_saved` | `{analysis_id}` | 保存完成 |
| `done` | `{content: ""}` | 完成 |

### 克隆事件

| 事件 | 数据 | 说明 |
|------|------|------|
| `progress` | `{step, current, total}` | 进度 |
| `error` | `{message: string}` | 错误 |
| `done` | `{repo_id, branches, tags, commit_num}` | 完成 |

## 错误处理

### 后端错误

```go
if err != nil {
    sendEvent("error", map[string]interface{}{
        "message": err.Error(),
    })
    return
}
```

### 前端错误

```javascript
error: (data) => {
    // 显示错误消息
    showErrorToast(data.message);

    // 重置 UI
    resetProgressBar();
    hideSpinner();
}
```

### 网络错误

```javascript
catch (err) {
    if (err.name === 'AbortError') {
        // 用户中断，忽略
        return;
    }
    if (handlers.error) {
        handlers.error({ message: err.message });
    }
}
```

## 中断请求

### 前端中断

```javascript
const controller = SSE.post(url, body, handlers);

// 用户点击取消按钮
cancelButton.onclick = () => {
    controller.abort();
};
```

### 后端处理

- 客户端断开连接时，Gin 会检测到
- 服务层应检查 context 取消
- 长时间运行的操作应支持取消

## 性能优化

### 缓冲控制

```
X-Accel-Buffering: no
```

- 禁用 Nginx 缓冲
- 确保事件立即发送

### 批量发送

- 避免频繁发送小事件
- 合并相关数据

### 压缩

- SSE 不支持 gzip 压缩
- 数据量大时考虑压缩正文

## 并发安全

### 线程安全回调

```go
safeCallback := callback
if callback != nil {
    safeCallback = func(event string, data interface{}) {
        mu.Lock()
        defer mu.Unlock()
        callback(event, data)
    }
}
```

- 使用互斥锁保护回调
- 避免并发写入

### Goroutine 管理

```go
var wg sync.WaitGroup

for i, file := range files {
    wg.Add(1)
    go func(idx int, f FileDiff) {
        defer wg.Done()
        // ...
    }(i, file)
}

wg.Wait()
```

- 等待所有 goroutine 完成
- 避免资源泄漏

## 测试

### 手动测试

```bash
curl -N -X POST http://localhost:8080/api/repos/1/review \
  -H "Content-Type: application/json" \
  -d '{"base":"main","head":"feature"}'
```

### 自动化测试

```go
func TestSSEStream(t *testing.T) {
    // 创建测试服务器
    // 发送请求
    // 读取事件流
    // 验证事件序列
}
```

## 监控

### 关键指标

- 连接数
- 事件发送速率
- 错误率
- 响应时间

### 日志

```go
log.Printf("SSE connected: %s", c.ClientIP())
log.Printf("SSE event: %s", event)
log.Printf("SSE disconnected: %s", c.ClientIP())
```

## 安全考虑

### 认证

- 所有 SSE 端点需要认证
- 使用会话 Cookie

### 速率限制

- 限制并发 SSE 连接数
- 限制事件发送频率

### 数据验证

- 验证输入参数
- 防止注入攻击

## 故障排查

### 常见问题

1. **事件不发送**
   - 检查 Flusher 是否可用
   - 确认响应头设置正确
   - 检查 Nginx 缓冲配置

2. **连接断开**
   - 检查超时设置
   - 确认网络稳定
   - 查看错误日志

3. **数据乱码**
   - 确认 JSON 序列化正确
   - 检查字符编码
   - 验证事件格式

### 调试工具

- 浏览器开发者工具 Network 面板
- curl 命令行测试
- Wireshark 抓包