# temporal-streaming-go

**Repository Path**: mefaso/temporal-streaming-go

## Basic Information

- **Project Name**: temporal-streaming-go
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-07-02
- **Last Updated**: 2026-07-02

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# Temporal Streaming Go

**给已有 Temporal 项目加流式输出能力。**

Temporal 管任务执行和重试，Redis Stream 管实时事件，SSE 推给前端。三层各管各的。

---

## 快速理解

```
用户 POST /runs
  → Temporal 启动 workflow
  → workflow 调 activity
  → activity 边跑边往 Redis Stream 写事件（token/日志/进度）
  → 用户 GET /runs/:wfId/streams/:runId（SSE）
  → API 从 Redis Stream 读事件，实时推给前端
  → activity 跑完，写 run_completed 事件，SSE 关闭
```

**关键原则：**
- workflow 不碰 Redis（必须可重放）
- 只有 activity 写 Redis
- token/日志/进度是 fire-and-forget（Redis 挂了不影响 Temporal）
- 完成事件重试 3 次确保写入

---

## 集成到你已有的 Temporal 项目（3 步）

### 第 1 步：引入流式层

#### Go 项目

直接复制 4 个文件：

```
emitter.go       ← 写 Redis Stream
sse.go           ← 读 Redis Stream 推 SSE
models.go        ← 数据结构 + key 规则
repository.go    ← 存 run 状态到 Redis（可选）
```

```bash
go get github.com/redis/go-redis/v9
```

#### 非 Go 项目（Python / Node.js / Java 等）

这个项目的核心是**设计模式**，不是特定语言。任何语言都能用，因为 Redis Stream 和 Temporal 都有多语言 SDK。

你不需要复制 Go 代码。只需要在你自己的语言里实现两段逻辑：

**1. Activity 里写事件（XADD）**

```python
# Python 示例
import redis, json
rdb = redis.from_url("redis://localhost:6379")

def generate_activity(prompt, workflow_id, run_id, attempt):
    key = f"wf:{workflow_id}:run:{run_id}"
    meta = {"workflowId": workflow_id, "runId": run_id, "step": "llm_generate", "attempt": attempt}

    # step_started
    rdb.xadd(key, {"data": json.dumps({"type": "step_started", **meta, "payload": {"prompt": prompt}})}, maxlen=10000, approximate=True)

    # 逐 token
    for i, chunk in enumerate(llm_stream(prompt)):
        rdb.xadd(key, {"data": json.dumps({"type": "token", **meta, "seq": i, "payload": {"text": chunk}})}, maxlen=10000, approximate=True)

    # run_completed
    rdb.xadd(key, {"data": json.dumps({"type": "run_completed", **meta, "payload": {"result": result}})}, maxlen=10000, approximate=True)
```

**2. SSE 端读事件（XREAD BLOCK）**

```python
@app.get("/runs/{workflow_id}/streams/{run_id}")
async def stream(workflow_id: str, run_id: str, request: Request):
    key = f"wf:{workflow_id}:run:{run_id}"
    last_id = request.headers.get("last-event-id", "0")

    async def event_stream():
        nonlocal last_id
        while True:
            resp = rdb.xread({key: last_id}, block=15000, count=100)
            if not resp:
                yield "event: heartbeat\ndata: {}\n\n"
                continue
            for _, messages in resp:
                for msg_id, fields in messages:
                    last_id = msg_id.decode()
                    evt = json.loads(fields[b"data"].decode())
                    yield f"id: {last_id}\nevent: {evt['type']}\ndata: {json.dumps(evt)}\n\n"
                    if evt["type"] in ("run_completed", "run_failed"):
                        return
    return EventSourceResponse(event_stream())
```

**任何语言只需做这两件事：**
1. Activity 里 `XADD` 写事件到 `wf:{workflowId}:run:{runId}`
2. API 里 `XREAD BLOCK` 读事件推 SSE，支持 `Last-Event-ID` 断线重连

### 第 2 步：在你的 activity 里加 emitter（Go 项目）

```go
// 你原来的 activity
func (a *MyActivities) DoWork(ctx context.Context, input MyInput) (MyResult, error) {
    // 1. 构造元数据
    meta := StreamMeta{
        WorkflowID: input.WorkflowID,
        RunID:      input.RunID,
        Step:       "do_work",
    }

    // 2. 通知前端开始了
    a.Emitter.StepStarted(ctx, meta, map[string]any{"input": input})

    // 3. 你的业务逻辑 —— 边跑边发事件
    stream := callLLMStream(input.Prompt)
    var seq int64
    for chunk := range stream {
        a.Emitter.Token(ctx, meta, seq, chunk.Text)  // 前端立刻看到
        seq++
    }

    // 4. 通知前端完成了
    result := finalize(stream)
    a.Emitter.Complete(ctx, meta, result)

    return result, nil
}
```

**workflow 不用改。** 只有 activity 加几行代码。

### 第 3 步：加一个 SSE 路由

```go
r.GET("/runs/:workflowId/streams/:runId", func(c *gin.Context) {
    StreamSSE(c, redisClient, c.Param("workflowId"), c.Param("runId"))
})
```

---

## 事件类型

| 方法 | 事件 type | 用途 | 失败行为 |
|------|-----------|------|----------|
| `StepStarted(ctx, meta, info)` | `step_started` | 步骤开始 | 记日志，不阻断 |
| `Token(ctx, meta, seq, text)` | `token` | LLM 逐 token 输出 | 记日志，不阻断 |
| `Log(ctx, meta, level, msg, fields)` | `log` | 日志 | 记日志，不阻断 |
| `Progress(ctx, meta, current, total)` | `progress` | 进度条 | 记日志，不阻断 |
| `Complete(ctx, meta, result)` | `run_completed` | 完成（终态） | 重试 3 次 |
| `Fail(ctx, meta, err)` | `run_failed` | 失败（终态） | 重试 3 次 |

---

## SSE 协议细节

### 响应头
```
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive
X-Accel-Buffering: no
```

### 事件格式
```
id: 1234567890-0
event: token
data: {"seq":0,"step":"llm_generate","attempt":1,"ts":"...","payload":{"text":"你"}}

```

- `id` 是 Redis Stream ID，浏览器自动用于 `Last-Event-ID`
- `event` 是事件类型
- 收到 `run_completed` 或 `run_failed` 后 SSE 关闭

### 断线重连
浏览器 `EventSource` 断线后自动重连，自动带 `Last-Event-ID` header。服务器从该 ID 续读 Redis Stream，不丢消息。

```js
const es = new EventSource('/runs/wf-123/streams/run-456')
es.addEventListener('token', (e) => {
    const data = JSON.parse(e.data)
    console.log(data.payload.text)  // 逐 token
})
es.addEventListener('run_completed', (e) => {
    es.close()
})
```

---

## Activity 重试与去重

Temporal activity 重试时会重复执行。每个事件带 `attempt` 字段（第几次重试）。

前端去重逻辑：
```js
const seen = new Set()
es.addEventListener('token', (e) => {
    const data = JSON.parse(e.data)
    const key = `${data.attempt}:${data.seq}`
    if (seen.has(key)) return  // 重复，跳过
    seen.add(key)
    appendText(data.payload.text)
})
```

---

## Demo

这个项目本身包含一个完整可运行的 demo：

```bash
# 启动 Redis 和 Temporal
docker run -d --name redis -p 6379:6379 redis:7
docker run -d --name temporal -p 7233:7233 temporalio/auto-setup:1.23

# 启动项目
cd temporal-streaming-go
go run .

# 测试
curl -X POST http://localhost:8080/runs -H 'Content-Type: application/json' -d '{"prompt":"你好"}'
# 返回 streamUrl，然后：
curl -N http://localhost:8080/runs/{workflowId}/streams/{runId}
```

---

## API

### POST /runs — 启动 workflow
```bash
curl -X POST http://localhost:8080/runs \
  -H 'Content-Type: application/json' \
  -d '{"prompt":"你好"}'
```
```json
{
  "workflowId": "wf-1234567890",
  "runId": "abc-def-ghi",
  "streamUrl": "/runs/wf-1234567890/streams/abc-def-ghi",
  "statusUrl": "/runs/wf-1234567890/status"
}
```

### GET /runs/:workflowId/streams/:runId — SSE 实时流
```bash
curl -N http://localhost:8080/runs/wf-123/streams/abc-456
```

### GET /runs/:workflowId/status — 查状态
```bash
curl http://localhost:8080/runs/wf-123/status?runId=abc-456
```
```json
{
  "workflowId": "wf-123",
  "runId": "abc-456",
  "state": "completed",
  "result": {"text": "回复: 你好"},
  "updatedAt": "2026-07-02T12:00:00Z"
}
```

### GET /healthz — 健康检查
```bash
curl http://localhost:8080/healthz
```

---

## 环境变量

| 变量 | 默认值 | 说明 |
|------|--------|------|
| `HTTP_ADDR` | `:8080` | HTTP 监听地址 |
| `REDIS_ADDR` | `127.0.0.1:6379` | Redis 地址 |
| `TEMPORAL_HOSTPORT` | `127.0.0.1:7233` | Temporal 地址 |
| `TEMPORAL_NAMESPACE` | `default` | Temporal namespace |
| `TEMPORAL_TASK_QUEUE` | `streaming-task-queue` | Task queue 名 |
| `RUN_TTL_HOURS` | `24` | Stream/Status 保留时间 |
| `STREAM_MAX_LEN` | `10000` | Stream 最大条数 |

---

## 文件说明

| 文件 | 角色 | 复制到你项目？ |
|------|------|:-:|
| `emitter.go` | Redis Stream 写入，fire-and-forget + 终态重试 | ✅ |
| `sse.go` | Redis Stream 读取 + SSE 推送 + 断线重连 | ✅ |
| `models.go` | 数据结构 + StreamKey/RunStatusKey | ✅ |
| `repository.go` | Redis 存 run 状态 | 可选 |
| `workflow.go` | Demo workflow | ❌ |
| `activity.go` | Demo activity（展示 emitter 用法） | ❌ |
| `handlers.go` | Demo HTTP handler | ❌ |
| `main.go` | Demo 入口（优雅关停 + CORS + 健康检查） | ❌ |
| `config.go` | Demo 配置 | ❌ |
| `emitter_test.go` | 单元测试 | 参考 |

---

## 生产部署注意

1. **workflowID 用 UUID** — demo 用 UnixNano，高并发可能碰撞。生产用 `github.com/google/uuid`
2. **加鉴权** — demo 没鉴权，生产加 JWT/API Key middleware
3. **加限流** — POST /runs 加 rate limiter
4. **Redis 持久化** — 生产 Redis 开 AOF 或 RDB
5. **Nginx 代理** — 如果用 Nginx，加 `proxy_buffering off` 和 `proxy_read_timeout 3600s`
6. **监控** — 加 Prometheus metrics（QPS、SSE 连接数、Redis 延迟）

---

## 设计决策

| 决策 | 原因 |
|------|------|
| Redis Stream（不是 Pub/Sub） | Stream 保留历史，支持断线重连 |
| fire-and-forget for token/log | Redis 不应该阻断 Temporal |
| 终态事件重试 3 次 | 确保 SSE 能收到完成/失败信号 |
| 事件带 attempt | activity 重试时前端能去重 |
| workflow 不碰 Redis | workflow 必须可重放，不能有副作用 |
| Stream 自动过期 | 避免无限增长，TTL 可配 |