erpnew/docs/sse_refactor.md

SSE 模块重构文档

1. 背景与问题

1.1 当前状态

SSE（Server-Sent Events）模块已被停用（返回 405），原因是出现了致命的数据库连接泄漏问题：

• Django 的 SSE 长连接会无限制地消耗数据库连接（Database Connection）
• 连接长时间不释放，导致数据库连接池耗尽
• 严重影响系统稳定性，导致其他业务无法正常访问数据库

1.2 重构目标

将 SSE 模块迁移到独立的 Golang 进程运行，与 Django 后端解耦：

• Golang 的 goroutine 模型天然适合处理大量长连接
• 不依赖 Django 的数据库连接池
• 更好的资源管理和并发性能
• 系统解耦，SSE 服务可独立扩展和部署

---

2. 当前架构概述

2.1 目录结构

sse/
├── __init__.py
├── admin.py
├── apps.py
├── auth_utils.py          # JWT 认证工具
├── client_example.js      # 前端客户端示例
├── migrations/
├── minimal_sse.py         # 极简实现（尝试解决关闭问题）
├── models.py              # 空模型文件
├── serializers.py         # DRF 序列化器
├── services.py            # 核心服务层（连接管理、消息推送）
├── simple_sse.py          # 简化实现（尝试解决关闭问题）
└── views.py               # API 视图（已禁用）

2.2 当前 API 端点

端点	方法	描述	状态
/sse/	GET	SSE 事件流订阅	❌ 已禁用
/sse/push/	POST	推送测试事件	✅ 可用
/sse/status/	GET	获取连接状态	✅ 可用
/sse/shutdown/	POST	关闭商户连接	✅ 可用

---

3. 核心设计原则

3.1 透明转发架构（关键设计）

Golang SSE 服务应该是"不关心业务内容的转发层"，这样可以避免 Django 和 Golang 之间的消息类型重复定义。

┌─────────────┐     任意 JSON      ┌─────────────┐     原样转发      ┌─────────────┐
│   Django    │ ─────────────────► │  Golang SSE │ ─────────────────► │   前端      │
│  (业务层)   │   merchant_id +    │  (转发层)   │                    │  (消费者)   │
└─────────────┘   payload          └─────────────┘                    └─────────────┘

核心原则：

1. Django 是唯一定义消息类型的地方 - 所有业务事件类型只在 Django 定义
2. Golang 只负责三件事 - 认证、路由（按 merchant_id）、转发
3. Golang 不解析业务内容 - 使用 json.RawMessage 原样转发
4. 新增消息类型 = 只改 Django - Golang 服务一旦写好几乎不需要再改

Golang 内部推送接口：

// 只需要这一个结构体，永远不变
type BroadcastRequest struct {
    MerchantID int64           `json:"merchant_id"`
    Payload    json.RawMessage `json:"payload"`  // 不解析，直接转发
}

func handleBroadcast(w http.ResponseWriter, r *http.Request) {
    var req BroadcastRequest
    json.NewDecoder(r.Body).Decode(&req)
    
    // 直接转发给该商户的所有连接，不解析 payload
    broadcastToMerchant(req.MerchantID, req.Payload)
}

Django 推送封装：

# sse/client.py - Django 调用 Golang SSE 服务的客户端

import requests
from django.conf import settings

SSE_SERVICE_URL = getattr(settings, 'SSE_SERVICE_URL', 'http://localhost:8080')

def push_to_sse(merchant_id: int, payload: dict):
    """向 Golang SSE 服务推送消息（透明转发）"""
    try:
        requests.post(
            f"{SSE_SERVICE_URL}/internal/broadcast",
            json={
                "merchant_id": merchant_id,
                "payload": payload  # 任意 JSON，Golang 不解析
            },
            timeout=5
        )
    except Exception as e:
        logger.warning(f"SSE push failed: {e}")

3.2 多租户隔离

系统是多商户（Multi-tenant）架构，SSE 消息必须按商户隔离：

• 每个用户通过 JWT Token 中的 user.employee.merchant_id 确定所属商户
• 消息只推送给同一商户下的所有连接客户端
• 不同商户之间的消息完全隔离

3.3 认证机制

使用 JWT（JSON Web Token）认证：

Authorization: Bearer <JWT_TOKEN>

认证流程：

1. 从请求头提取 Bearer Token
2. 使用 Django Simple JWT 验证 Token
3. 获取用户信息：user = jwt_authenticator.get_user(validated_token)
4. 获取商户 ID：merchant_id = user.employee.merchant_id
5. 无商户关联的用户拒绝连接

重要：Golang 服务需要能够验证 Django 签发的 JWT Token，需要共享相同的 SECRET_KEY 或使用 RS256 算法。

---

4. 消息规范

4.1 消息格式

所有 SSE 消息均为 JSON 格式，通过 data: 前缀发送：

data: {"type": "xxx", ...}\n\n

4.2 系统消息类型

连接成功消息

当客户端成功建立 SSE 连接后，立即发送：

{
  "type": "connected",
  "message": "SSE connection established",
  "merchant_id": 1
}

心跳消息

保持连接活跃，使用 SSE 注释格式：

: heartbeat\n\n

或带时间戳的格式：

: heartbeat 1704067200.123\n\n

心跳间隔建议：30 秒

服务器关闭消息

服务器主动关闭连接时发送：

{
  "type": "server_shutdown",
  "message": "Server shutting down your connections, please reconnect later"
}

4.3 业务消息类型

业务消息统一格式：

{
  "mode": "simple_message",
  "type": "<event_type>",
  "message": "<human_readable_message>",
  "object_id": <related_object_id>,
  "merchant_id": <merchant_id>
}

字段说明：

• mode: 固定为 "simple_message"，标识这是业务消息
• type: 事件类型，用于前端区分处理逻辑
• message: 人类可读的消息内容（中文）
• object_id: 关联业务对象的 ID（整数或字符串）
• merchant_id: 商户 ID，便于客户端二次验证

已定义的业务事件类型

事件类型	说明	触发场景
order_paid	订单已支付	订单支付完成
stock_change	库存变动	库存入库/出库完成

注意：未来可能增加更多事件类型，Golang 服务应设计为可扩展的。

---

5. 服务接口设计

5.1 SSE 连接端点

端点: GET /sse/

请求头:

Authorization: Bearer <JWT_TOKEN>
Accept: text/event-stream
Cache-Control: no-cache

响应头:

Content-Type: text/event-stream
Cache-Control: no-cache
X-Accel-Buffering: no
Connection: keep-alive

CORS 支持:

Access-Control-Allow-Origin: <request_origin>
Access-Control-Allow-Credentials: true
Access-Control-Allow-Methods: GET, OPTIONS
Access-Control-Allow-Headers: authorization, content-type, cache-control, accept

5.2 消息推送接口（透明转发）

Django 后端通过 HTTP API 将消息推送给 Golang SSE 服务。

关键设计：Golang 不解析 payload 内容，只负责路由和转发。

端点: POST /internal/broadcast

POST /internal/broadcast
Content-Type: application/json

{
  "merchant_id": 1,
  "payload": {
    "action": "invalidate",
    "resources": ["stock-records"],
    "message": "库存已更新",
    "toast": true
  }
}

Golang 处理逻辑：

type BroadcastRequest struct {
    MerchantID int64           `json:"merchant_id"`
    Payload    json.RawMessage `json:"payload"`  // 原样转发，不解析
}

func handleBroadcast(w http.ResponseWriter, r *http.Request) {
    var req BroadcastRequest
    json.NewDecoder(r.Body).Decode(&req)
    
    // 直接转发，不解析 payload 内容
    broadcastToMerchant(req.MerchantID, req.Payload)
    
    w.WriteHeader(http.StatusOK)
}

备选方案：Redis Pub/Sub

如果需要更高的可靠性或解耦：

• Django 发布消息到 Redis Channel：sse:merchant:{merchant_id}
• Golang 服务订阅对应 Channel 并转发给客户端
• 优点：异步、解耦、可靠性更高
• 缺点：增加 Redis 依赖和复杂度

5.3 状态查询接口

端点: GET /sse/status/

响应:

{
  "status": "running",
  "total_clients": 10,
  "merchant_clients": 3,
  "merchant_id": 1,
  "message": "SSE server is running with 10 total connections, 3 for your merchant"
}

5.4 连接关闭接口

端点: POST /sse/shutdown/

关闭指定商户的所有 SSE 连接。

响应:

{
  "status": "ok",
  "message": "Your merchant's SSE connections have been closed",
  "merchant_id": 1,
  "clients": 3
}

---

6. 连接管理

6.1 连接存储结构

按商户组织连接：

// 伪代码
type ConnectionManager struct {
    // merchant_id -> []Connection
    connections map[int64][]*Connection
    mutex       sync.RWMutex
}

6.2 连接生命周期

1. 建立连接

   • 验证 JWT Token
   • 提取 merchant_id
   • 注册到连接管理器
   • 发送 connected 消息

2. 维护连接

   • 定期发送心跳（30 秒）
   • 监听消息队列，转发给客户端
   • 处理客户端断开

3. 关闭连接

   • 发送 server_shutdown 消息
   • 从连接管理器移除
   • 释放资源

6.3 异常处理

• 客户端断开：自动检测并清理
• 消息队列满：丢弃旧消息或断开慢客户端
• 认证失败：返回 403 Forbidden
• 服务关闭：优雅地关闭所有连接

---

7. 前端客户端参考

7.1 使用 Fetch API（推荐）

function createSSEConnection(url, token, onMessage, onError, onClose) {
    const controller = new AbortController();
    
    fetch(url, {
        method: 'GET',
        headers: {
            'Authorization': `Bearer ${token}`,
            'Accept': 'text/event-stream',
            'Cache-Control': 'no-cache',
        },
        signal: controller.signal
    })
    .then(response => {
        if (!response.ok) throw new Error(`HTTP ${response.status}`);
        
        const reader = response.body.getReader();
        const decoder = new TextDecoder();
        let buffer = '';
        
        function processBuffer() {
            const lines = buffer.split('\n');
            buffer = lines.pop();
            
            for (const line of lines) {
                if (line.startsWith('data: ')) {
                    const data = line.substring(6);
                    try {
                        onMessage(JSON.parse(data));
                    } catch (e) {
                        console.error('Parse error:', e);
                    }
                }
            }
        }
        
        function read() {
            return reader.read().then(({ done, value }) => {
                if (done) { onClose?.(); return; }
                buffer += decoder.decode(value, { stream: true });
                processBuffer();
                return read();
            });
        }
        
        return read();
    })
    .catch(onError);
    
    return { close: () => controller.abort() };
}

7.2 事件处理示例

const sse = createSSEConnection('/sse/', jwtToken, 
    (event) => {
        switch (event.type) {
            case 'connected':
                console.log(`连接成功，商户: ${event.merchant_id}`);
                break;
            case 'stock_change':
                // 刷新库存列表
                refreshStockList();
                break;
            case 'order_paid':
                // 显示订单支付通知
                showNotification(event.message);
                break;
            case 'server_shutdown':
                // 准备重连
                scheduleReconnect();
                break;
        }
    },
    (error) => {
        console.error('SSE Error:', error);
        scheduleReconnect();
    },
    () => console.log('SSE Closed')
);

---

8. 前端消息处理简化方案

8.1 问题分析

如果每种业务事件都需要前端单独处理，会造成：

• 前端需要维护一份事件类型清单
• 新增事件类型需要同时改 Django 和前端
• switch-case 代码越来越长

8.2 混合消息模式（推荐）

实际业务中存在两种不同的通知需求，需要用不同的模式处理：

模式	用途	典型场景
资源失效	数据刷新，无特定对象	库存变动后刷新列表
对象事件	指向特定对象的通知	员工被指派处理特定任务

8.3 统一消息格式

interface SSEMessage {
  // 必填：动作类型
  action: 'invalidate' | 'notify' | 'connected' | 'server_shutdown';
  
  // 资源失效相关（可选）
  resources?: string[];           // 需要刷新的资源列表
  
  // 对象事件相关（可选）
  object_type?: string;           // 对象类型，如 'printing-job'
  object_id?: number | string;    // 具体对象ID
  
  // 用户定向（可选）
  target_user_ids?: number[];     // 目标用户ID列表，为空表示广播给所有人
  
  // 通知展示（可选）
  message?: string;               // 用户可见的提示消息
  toast?: boolean;                // 是否显示 Toast 通知
  url?: string;                   // 点击通知后跳转的URL
}

8.4 消息示例

模式1：纯资源刷新（不需要知道具体对象）

{
  "action": "invalidate",
  "resources": ["stock-records", "inventory"],
  "message": "库存已更新",
  "toast": true
}

模式2：对象事件（需要指向特定对象）

{
  "action": "notify",
  "object_type": "printing-job",
  "object_id": 123,
  "target_user_ids": [5],
  "message": "您有新的印染任务 #123",
  "toast": true,
  "url": "/printing-jobs/123"
}

模式3：混合（既刷新数据，又指向特定对象）

{
  "action": "notify",
  "object_type": "printing-job",
  "object_id": 123,
  "resources": ["printing-jobs"],
  "message": "印染任务 #123 已更新",
  "toast": true
}

8.5 目标用户过滤策略

问题：当消息需要通知特定用户时，过滤应该在哪里做？

方案	描述	优缺点
A. Django 多次推送	Django 为每个目标用户单独调用推送	复杂，需要跟踪用户连接状态
B. 前端过滤（推荐）	消息带 target_user_ids，前端自己过滤	✅ 简单，保持透明转发
C. Golang 过滤	Golang 解析 target_user_ids 只推给特定连接	打破透明转发原则

选择方案 B 的理由：

1. 保持透明转发 - Golang 无需改动
2. 前端过滤逻辑极简 - 一行代码
3. 带宽浪费可忽略 - 同商户内并发用户数有限

8.6 前端通用处理器

// hooks/useSSE.ts
import { useQueryClient } from '@tanstack/react-query';
import { message, Modal } from 'antd';
import { useNavigate } from 'react-router-dom';
import { useCurrentUser } from './useCurrentUser';

interface SSEMessage {
  action: 'invalidate' | 'notify' | 'connected' | 'server_shutdown';
  resources?: string[];
  object_type?: string;
  object_id?: number | string;
  target_user_ids?: number[];
  message?: string;
  toast?: boolean;
  url?: string;
}

export function useSSEHandler() {
  const queryClient = useQueryClient();
  const navigate = useNavigate();
  const { user } = useCurrentUser();

  const handleMessage = (event: SSEMessage) => {
    // ========== 用户过滤 ==========
    // 如果指定了目标用户，且当前用户不在列表中，则忽略
    if (event.target_user_ids && event.target_user_ids.length > 0) {
      if (!event.target_user_ids.includes(user.id)) {
        return;  // 不是发给我的，忽略
      }
    }

    // ========== 系统消息 ==========
    if (event.action === 'connected') {
      console.log('SSE 连接成功');
      return;
    }
    
    if (event.action === 'server_shutdown') {
      scheduleReconnect();
      return;
    }

    // ========== 资源失效处理 ==========
    if (event.resources && event.resources.length > 0) {
      for (const resource of event.resources) {
        queryClient.invalidateQueries({ queryKey: [resource] });
      }
    }

    // ========== 通知展示 ==========
    if (event.toast && event.message) {
      if (event.url) {
        // 可点击的通知
        message.info(
          <span onClick={() => navigate(event.url!)} style={{ cursor: 'pointer' }}>
            {event.message}
          </span>
        );
      } else {
        message.info(event.message);
      }
    }
  };

  return { handleMessage };
}

8.7 Django 侧封装

# sse/events.py - 统一的事件发送接口

from sse.client import push_to_sse

def notify_resource_changed(
    merchant_id: int,
    resources: list[str],
    message: str = None,
    toast: bool = False
):
    """
    通知前端资源已变化，需要刷新（纯资源失效模式）
    """
    push_to_sse(merchant_id, {
        "action": "invalidate",
        "resources": resources,
        "message": message,
        "toast": toast
    })


def notify_object_event(
    merchant_id: int,
    object_type: str,
    object_id: int,
    message: str,
    target_user_ids: list[int] = None,
    resources: list[str] = None,
    url: str = None,
    toast: bool = True
):
    """
    通知特定对象的事件（对象事件模式）
    
    Args:
        merchant_id: 商户ID
        object_type: 对象类型，如 'printing-job', 'plate-order'
        object_id: 对象ID
        message: 用户可见的消息
        target_user_ids: 目标用户ID列表，为空表示通知所有人
        resources: 可选，同时触发哪些资源刷新
        url: 可选，点击通知后跳转的URL
        toast: 是否显示 Toast
    """
    payload = {
        "action": "notify",
        "object_type": object_type,
        "object_id": object_id,
        "message": message,
        "toast": toast
    }
    
    if target_user_ids:
        payload["target_user_ids"] = target_user_ids
    if resources:
        payload["resources"] = resources
    if url:
        payload["url"] = url
    
    push_to_sse(merchant_id, payload)


# ========== 业务代码调用示例 ==========

# 示例1：库存变动（纯资源刷新，通知所有人）
def on_stock_change(merchant_id, record_id):
    notify_resource_changed(
        merchant_id=merchant_id,
        resources=["stock-records", "inventory"], # 资源为空意味着是一则普通的消息通知
        message="库存已更新",
        toast=True
    )

# 示例2：任务指派（对象事件，只通知被指派的人）
def on_job_assigned(merchant_id, job_id, assigned_user_id):
    notify_object_event(
        merchant_id=merchant_id,
        object_type="printing-job",
        object_id=job_id,
        message=f"您有新的印染任务 #{job_id}",
        target_user_ids=[assigned_user_id], # 可能为空意味着需要全员接收该消息
        resources=["printing-jobs"],  # 同时刷新任务列表
        url=f"/printing-jobs/{job_id}"
    )

# 示例3：订单状态变更（通知相关人员）
def on_order_status_changed(merchant_id, order_id, related_user_ids):
    notify_object_event(
        merchant_id=merchant_id,
        object_type="printing-order",
        object_id=order_id,
        message=f"订单 #{order_id} 状态已更新",
        target_user_ids=related_user_ids,
        resources=["printing-orders"]
    )

8.8 资源命名约定

建议 resources 和 object_type 使用与 API 路径一致的命名：

名称	对应 API	React Query Key
printing-orders	/api/v1/printing-orders/	['printing-orders']
printing-jobs	/api/v1/printing-jobs/	['printing-jobs']
plate-orders	/api/v1/plate-orders/	['plate-orders']
stock-records	/api/v1/stock-change-records/	['stock-records']
inventory	/api/v1/inventory/	['inventory']

8.9 方案优势总结

维度	效果
新增消息类型	只改 Django，Golang 和前端通用处理器不变
目标用户过滤	前端一行代码，保持透明转发
对象导航	支持 object_type + object_id + url
灵活性	纯刷新、对象事件、混合模式都支持

---

9. 部署架构

9.1 推荐架构

                    ┌─────────────────┐
                    │   Nginx/CDN     │
                    └────────┬────────┘
                             │
            ┌────────────────┼────────────────┐
            │                │                │
            ▼                ▼                ▼
    ┌───────────────┐ ┌───────────────┐ ┌───────────────┐
    │  Django API   │ │  Golang SSE   │ │  Static Files │
    │   (Gunicorn)  │ │   Service     │ │    (CDN)      │
    └───────┬───────┘ └───────┬───────┘ └───────────────┘
            │                 │
            │    HTTP/Redis   │
            └────────┬────────┘
                     │
            ┌────────▼────────┐
            │  PostgreSQL/    │
            │  Redis          │
            └─────────────────┘

9.2 Nginx 配置要点

location /sse/ {
    proxy_pass http://golang-sse-service:8080;
    proxy_http_version 1.1;
    proxy_set_header Connection '';
    proxy_buffering off;
    proxy_cache off;
    
    # SSE 超时配置
    proxy_read_timeout 3600s;
    proxy_send_timeout 3600s;
    
    # 禁用缓冲
    add_header X-Accel-Buffering no;
}

9.3 Docker Compose 示例

services:
  sse:
    image: flower-sse:latest
    ports:
      - "8080:8080"
    environment:
      - JWT_SECRET=${SECRET_KEY}
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - redis

---

10. 安全考虑

1. JWT 验证：必须验证 Token 有效性和过期时间
2. 商户隔离：严格检查 merchant_id，防止跨商户数据泄漏
3. 内部 API 保护：消息推送 API 只允许内网访问
4. 连接限制：限制单用户/单商户的最大连接数
5. 速率限制：防止连接风暴

---

11. 监控指标

建议监控以下指标：

• 总连接数
• 各商户连接数
• 消息推送延迟
• 消息推送成功率
• 连接存活时长分布
• 心跳超时断开数

---

12. 迁移计划

Phase 1: 开发与测试

1. 开发 Golang SSE 服务
2. 实现 JWT 认证（与 Django 共享密钥）
3. 实现消息推送接口
4. 单元测试和压力测试

Phase 2: 并行运行

1. 部署 Golang SSE 服务
2. Django 同时向新旧 SSE 推送消息
3. 前端切换到新 SSE 端点
4. 监控对比

Phase 3: 完全迁移

1. 废弃 Django SSE 模块
2. 移除相关代码
3. 更新文档

---

13. 附录：当前代码参考

13.1 核心服务函数签名

# sse/services.py

def push_connection(merchant_id, conn_queue):
    """将新连接添加到指定商户的连接集合"""

def remove_connection(merchant_id, conn_queue):
    """从指定商户移除连接"""

def push_sse_event_to_merchant(merchant_id, event_data: dict):
    """向指定商户的所有客户端广播事件"""

def push_simple_message_with_object_id_to_merchant(
    merchant_id, event_type: str, message: str, object_id
):
    """向指定商户广播简单业务消息"""

def shutdown_merchant_connections(merchant_id):
    """关闭指定商户的所有连接"""

13.2 Django 中的调用示例

# stock/services.py
from sse.services import push_simple_message_with_object_id

# 库存变动后推送通知
push_simple_message_with_object_id(
    event_type='stock_change',
    message=f'产品ID {detail.product_id} 位于 {warehouse_id} 的库存已更新',
    object_id=inventory_record.id
)

---