如何在Lambda场景下通过graph-gophers基于连接ID推送订阅消息？

1. 重构订阅解析器逻辑 #

Lambda的短生命周期决定了没法维持长连接监听channel，所以要彻底调整graph-gophers的订阅解析器逻辑：

• 解析客户端的订阅请求，提取订阅的事件主题（比如newComment）、用户上下文等信息
• 把WebSocket的连接ID、订阅主题、用户标识等数据存入数据库（比如DynamoDB），完成订阅注册
• 返回一个立即关闭的空channel给graph-gophers（满足库的接口要求），同时给客户端返回订阅成功的确认消息

示例代码片段：

func commentSubscriptionResolver(ctx context.Context, args map[string]interface{}) (<-chan interface{}, error) {
    // 从API Gateway请求上下文里取出连接ID
    connID := ctx.Value("connectionID").(string)
    topic := args["postID"].(string) // 比如按文章ID订阅评论

    // 将订阅关系写入数据库
    err := db.SaveSubscription(connID, topic, ctx.Value("userID").(string))
    if err != nil {
        return nil, err
    }

    // 返回关闭的channel，避免graph-gophers阻塞
    ch := make(chan interface{})
    close(ch)
    return ch, nil
}

2. 实现异步推送触发流程 #

当业务事件发生时（比如新评论创建），通过以下步骤触发推送：

• 用单独的Lambda处理事件触发（可以通过EventBridge、SQS或者业务逻辑直接调用）
• 从数据库中查询所有订阅了对应主题的有效连接ID
• 调用AWS SDK的PostToConnection API，给每个连接ID推送符合GraphQL WebSocket规范的消息

示例推送代码：

import (
    "encoding/json"
    "github.com/aws/aws-sdk-go/aws"
    "github.com/aws/aws-sdk-go/aws/session"
    "github.com/aws/aws-sdk-go/service/apigatewaymanagementapi"
)

func pushNewComment(topic string, comment Comment) error {
    // 1. 查询订阅该主题的所有连接ID
    connIDs, err := db.GetSubscribedConnIDs(topic)
    if err != nil {
        return err
    }

    // 2. 初始化API Gateway管理客户端
    sess := session.Must(session.NewSession(&aws.Config{Region: aws.String("us-east-1")}))
    apiClient := apigatewaymanagementapi.New(sess, &aws.Config{
        Endpoint: aws.String("https://your-api-id.execute-api.us-east-1.amazonaws.com/prod"),
    })

    // 3. 构造GraphQL订阅消息格式
    payload := map[string]interface{}{
        "newComment": comment,
    }
    msg := map[string]interface{}{
        "type":    "data",
        "id":      "sub-123", // 可使用客户端订阅时传入的ID
        "payload": payload,
    }
    msgBytes, _ := json.Marshal(msg)

    // 4. 逐个推送并清理无效连接
    for _, connID := range connIDs {
        _, err := apiClient.PostToConnection(&apigatewaymanagementapi.PostToConnectionInput{
            ConnectionId: aws.String(connID),
            Data:         msgBytes,
        })
        if err != nil {
            // 若返回GoneException，说明连接已断开，从数据库删除该ID
            if err.(*apigatewaymanagementapi.GoneException) != nil {
                db.DeleteSubscription(connID)
            }
        }
    }
    return nil
}

3. 处理连接生命周期的清理 #

WebSocket连接断开时，API Gateway会触发$disconnect事件到Lambda，在这个处理函数中：

• 从请求上下文取出连接ID
• 从数据库中删除该连接ID对应的所有订阅关系

4. 关键配置与注意事项 #

• 确保API Gateway WebSocket配置中，$connect、$disconnect和自定义GraphQL路由都绑定了对应的Lambda处理函数
• 在$connect阶段完成用户认证，将用户ID与连接ID绑定存储，后续推送可实现精准的用户级过滤
• 推送时要批量处理连接ID，避免Lambda超时，可结合SQS做异步批量推送优化

内容的提问来源于stack exchange，提问作者Joey Yi Zhao