设计简介¶

// Channel 协议适配器接口
type Channel interface {
    // 基本信息
    Name() string
    Type() ChannelType
    Protocol() ProtocolType

    // 生命周期
    Start(ctx context.Context, handler MessageHandler) error
    Stop(ctx context.Context) error
    IsRunning() bool

    // 消息发送（出站）
    SendMessage(ctx context.Context, msg *Message) error

    // 状态查询
    GetStatus() ChannelStatus
}

// MessageHandler 消息处理回调
type MessageHandler func(ctx context.Context, msg *Message) error

// Gateway WebSocket 网关接口
type Gateway interface {
    // 生命周期
    Start(ctx context.Context) error
    Stop(ctx context.Context) error

    // 消息处理（入站）
    HandleChannelMessage(ctx context.Context, msg *Message) error

    // 客户端管理
    RegisterClientHandler(handler ClientHandler)
    Broadcast(ctx context.Context, msg *Message) error
    SendToClient(ctx context.Context, clientID string, msg *Message) error

    // 状态查询
    GetClientCount() int
    IsRunning() bool
}

// ClientHandler 客户端消息处理器
type ClientHandler func(ctx context.Context, clientID string, msg *Message) (*Message, error)

// Message 标准消息格式
type Message struct {
    ID        string                 `json:"id"`
    ChannelID string                 `json:"channel_id"`      // 来源渠道
    Direction MessageDirection       `json:"direction"`       // inbound/outbound
    From      UserInfo               `json:"from"`
    To        UserInfo               `json:"to"`
    Content   string                 `json:"content"`
    Type      MessageType            `json:"type"`            // text/image/file
    Metadata  map[string]interface{} `json:"metadata"`
    Timestamp time.Time              `json:"timestamp"`
}

// UserInfo 用户信息
type UserInfo struct {
    ID       string `json:"id"`
    Name     string `json:"name"`
    Avatar   string `json:"avatar"`
    Platform string `json:"platform"` // wechat/dingtalk 等
}

// ChannelConfig 渠道配置
type ChannelConfig struct {
    Name    string                 `json:"name"`
    Type    ChannelType            `json:"type"`
    Enabled bool                   `json:"enabled"`
    Webhook WebhookConfig          `json:"webhook"`
    Secrets map[string]string      `json:"secrets"`         // 敏感配置
    Options map[string]interface{} `json:"options"`
}

// WebhookConfig Webhook 配置
type WebhookConfig struct {
    Path   string `json:"path"`    // 如 "/wechat"
    Method string `json:"method"`  // 默认 POST
}

type Gateway struct {
    httpServer *http.Server    // 统一 HTTP 服务器
    channels   map[string]Channel
    mux        *http.ServeMux
}

// 注册 Channel Webhook 路由
func (g *Gateway) registerChannelRoutes() {
    for _, ch := range g.channels {
        path := ch.GetWebhookPath()
        g.mux.HandleFunc(path, g.createWebhookHandler(ch))
    }
}

// WeChatChannel 微信渠道实现
type WeChatChannel struct {
    name      string
    config    ChannelConfig
    handler   MessageHandler
    isRunning bool
}

func (c *WeChatChannel) Start(ctx context.Context, handler MessageHandler) error {
    c.handler = handler
    c.isRunning = true
    return nil
}

func (c *WeChatChannel) HandleWebhook(w http.ResponseWriter, r *http.Request) {
    // 1. 验证签名
    if !c.verifySignature(r) {
        http.Error(w, "Invalid signature", http.StatusUnauthorized)
        return
    }

    // 2. 解析消息
    msg, err := c.parseMessage(r)
    if err != nil {
        http.Error(w, "Invalid message", http.StatusBadRequest)
        return
    }

    // 3. 调用 handler 转发给 Gateway
    if err := c.handler(r.Context(), msg); err != nil {
        http.Error(w, "Processing failed", http.StatusInternalServerError)
        return
    }

    // 4. 返回成功响应
    w.WriteHeader(http.StatusOK)
}

func (g *Gateway) HandleChannelMessage(ctx context.Context, msg *Message) error {
    // 1. 中间件处理
    if err := g.middlewareChain.Execute(ctx, msg); err != nil {
        return fmt.Errorf("middleware failed: %w", err)
    }

    // 2. 业务处理（如果注册了 handler）
    if g.clientHandler != nil {
        response, err := g.clientHandler(ctx, "", msg)
        if err != nil {
            return fmt.Errorf("business logic failed: %w", err)
        }

        // 3. 推送到 WebSocket 客户端
        if response != nil {
            return g.sessionMgr.Broadcast(ctx, response)
        }
    }

    // 4. 无业务 handler 时直接广播
    return g.sessionMgr.Broadcast(ctx, msg)
}

func (g *Gateway) HandleClientMessage(ctx context.Context, clientID string, msg *Message) error {
    // 1. 中间件处理
    if err := g.middlewareChain.Execute(ctx, msg); err != nil {
        return err
    }

    // 2. 业务处理
    if g.clientHandler != nil {
        response, err := g.clientHandler(ctx, clientID, msg)
        if err != nil {
            return err
        }

        // 3. 发送到指定渠道
        if response != nil && response.ChannelID != "" {
            channel, ok := g.channels[response.ChannelID]
            if !ok {
                return fmt.Errorf("channel not found: %s", response.ChannelID)
            }
            return channel.SendMessage(ctx, response)
        }
    }

    return nil
}

type SessionManager struct {
    sessions    map[string]*Session
    mu          sync.RWMutex
    upgrader    websocket.Upgrader
}

type Session struct {
    ID         string
    Conn       *websocket.Conn
    Connected  time.Time
    LastActive time.Time
    mu         sync.Mutex
}

func (sm *SessionManager) Broadcast(ctx context.Context, msg *Message) error {
    sm.mu.RLock()
    defer sm.mu.RUnlock()

    data, err := json.Marshal(msg)
    if err != nil {
        return err
    }

    for _, session := range sm.sessions {
        go func(s *Session) {
            s.mu.Lock()
            defer s.mu.Unlock()
            s.Conn.WriteMessage(websocket.TextMessage, data)
        }(session)
    }

    return nil
}

// Middleware 中间件接口
type Middleware interface {
    Name() string
    Handle(ctx context.Context, msg *Message, next Handler) error
}

type Handler func(ctx context.Context, msg *Message) error

// MiddlewareChain 中间件链
type MiddlewareChain struct {
    middlewares []Middleware
}

func (mc *MiddlewareChain) Execute(ctx context.Context, msg *Message) error {
    var index int

    var next Handler
    next = func(ctx context.Context, msg *Message) error {
        if index >= len(mc.middlewares) {
            return nil
        }
        mw := mc.middlewares[index]
        index++
        return mw.Handle(ctx, msg, next)
    }

    return next(ctx, msg)
}

type LoggingMiddleware struct {
    logger *slog.Logger
}

func (m *LoggingMiddleware) Handle(ctx context.Context, msg *Message, next Handler) error {
    start := time.Now()

    m.logger.Info("message received",
        "channel", msg.ChannelID,
        "from", msg.From.ID,
        "type", msg.Type,
    )

    err := next(ctx, msg)

    m.logger.Info("message processed",
        "duration", time.Since(start),
        "error", err,
    )

    return err
}

type TokenMiddleware struct {
    token string
}

func (m *TokenMiddleware) Handle(ctx context.Context, msg *Message, next Handler) error {
    // 从 metadata 中验证 token
    if m.token != "" {
        msgToken, ok := msg.Metadata["token"].(string)
        if !ok || msgToken != m.token {
            return fmt.Errorf("invalid token")
        }
    }
    return next(ctx, msg)
}

// 简单的指数退避重试
func retryWithBackoff(ctx context.Context, fn func() error, maxRetries int) error {
    var err error
    for i := 0; i < maxRetries; i++ {
        if err = fn(); err == nil {
            return nil
        }

        // 指数退避：1s, 2s, 4s
        delay := time.Duration(1<<i) * time.Second
        select {
        case <-time.After(delay):
            continue
        case <-ctx.Done():
            return ctx.Err()
        }
    }
    return err
}

// 升级配置
upgrader := websocket.Upgrader{
    CheckOrigin: func(r *http.Request) bool {
        // 生产环境应限制 Origin
        return true
    },
    ReadBufferSize:  1024,
    WriteBufferSize: 1024,
}

type Metrics struct {
    MessagesReceived  int64     // 接收消息数
    MessagesSent      int64     // 发送消息数
    ActiveClients     int       // 活跃客户端数
    ChannelErrors     map[string]int64  // 各渠道错误数
    ProcessingTime    time.Duration     // 平均处理时间
}

// 结构化日志
logger.Info("message processed",
    slog.String("channel", msg.ChannelID),
    slog.String("msg_id", msg.ID),
    slog.Duration("duration", duration),
    slog.String("error", errStr),
)

package main

import (
    "context"
    "log"
    "gort/pkg/gateway"
)

func main() {
    // 1. 创建 Gateway
    gw, err := gateway.New(gateway.Config{
        WebSocketAddr: ":9000",
        HTTPAddr:      ":8080",
    })
    if err != nil {
        log.Fatal(err)
    }

    // 2. 注册渠道
    wechatCh := channels.NewWeChat(channels.WeChatConfig{
        AppID:     os.Getenv("WECHAT_APP_ID"),
        AppSecret: os.Getenv("WECHAT_APP_SECRET"),
        Token:     os.Getenv("WECHAT_TOKEN"),
        WebhookPath: "/webhook/wechat",
    })
    gw.RegisterChannel(wechatCh)

    // 3. 设置业务处理器
    gw.RegisterClientHandler(func(ctx context.Context, clientID string, msg *gateway.Message) (*gateway.Message, error) {
        // 业务逻辑处理
        log.Printf("Received: %s", msg.Content)

        // 返回响应
        return &gateway.Message{
            ChannelID: msg.ChannelID,
            Content:   "Received: " + msg.Content,
            Type:      gateway.MessageTypeText,
        }, nil
    })

    // 4. 启动
    ctx := context.Background()
    if err := gw.Start(ctx); err != nil {
        log.Fatal(err)
    }

    // 5. 优雅关闭
    <-ctx.Done()
    gw.Stop(context.Background())
}