507 lines
18 KiB
Go
507 lines
18 KiB
Go
// Package service 承载 im-service 的领域逻辑：连接注册、房间频道、消息时间线、ACK 和 room bridge。
|
||
package service
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"sync"
|
||
|
||
imv1 "hyapp/api/proto/im/v1"
|
||
roomv1 "hyapp/api/proto/room/v1"
|
||
"hyapp/pkg/clock"
|
||
"hyapp/pkg/idgen"
|
||
"hyapp/pkg/xerr"
|
||
"hyapp/services/im-service/internal/auth"
|
||
"hyapp/services/im-service/internal/session"
|
||
"hyapp/services/im-service/internal/store"
|
||
)
|
||
|
||
// GuardClient 抽象 im-service 对 room-service 守卫查询的依赖。
|
||
type GuardClient interface {
|
||
// CheckSpeakPermission 校验当前用户能否向房间公屏发言，禁言、踢人和房间状态都由 room-service 判定。
|
||
CheckSpeakPermission(ctx context.Context, req *roomv1.CheckSpeakPermissionRequest) (*roomv1.CheckSpeakPermissionResponse, error)
|
||
// VerifyRoomPresence 校验当前用户能否订阅房间频道，防止只连上 WS 但未进房的用户偷收消息。
|
||
VerifyRoomPresence(ctx context.Context, req *roomv1.VerifyRoomPresenceRequest) (*roomv1.VerifyRoomPresenceResponse, error)
|
||
}
|
||
|
||
// Config 收敛 im-service 领域层运行参数。
|
||
type Config struct {
|
||
// NodeID 写入 session，未来多节点路由可以基于它定位真实连接所在实例。
|
||
NodeID string
|
||
// RoomHistoryLimit 控制房间最近历史窗口，房间消息不做逐用户 ACK，靠窗口补拉恢复。
|
||
RoomHistoryLimit int
|
||
// Clock 允许测试注入稳定时间，生产默认使用系统时间。
|
||
Clock clock.Clock
|
||
}
|
||
|
||
// OutgoingFrame 是 WS 和测试共用的标准下行结构。
|
||
type OutgoingFrame struct {
|
||
// Type 是客户端帧类型，例如 message、history、force_resync。
|
||
Type string `json:"type"`
|
||
// RoomID 标识房间相关下行帧所属房间。
|
||
RoomID string `json:"room_id,omitempty"`
|
||
// Reason 承载 force_resync 等控制帧的人类可读原因。
|
||
Reason string `json:"reason,omitempty"`
|
||
// Error 保留给 WS 错误帧；当前 handler 遇错直接断开，暂未下发错误帧。
|
||
Error string `json:"error,omitempty"`
|
||
// Message 承载单条消息投递。
|
||
Message *store.Message `json:"message,omitempty"`
|
||
// History 承载 pull_history 的房间最近窗口返回。
|
||
History []store.Message `json:"history,omitempty"`
|
||
}
|
||
|
||
// Service 承载 im-service 的消息、连接和投递逻辑。
|
||
type Service struct {
|
||
// nodeID 标识本实例，写入 session 用于在线状态定位。
|
||
nodeID string
|
||
// roomHistoryLimit 是房间消息最近窗口上限。
|
||
roomHistoryLimit int
|
||
// clock 为消息创建时间和 session 时间提供统一来源。
|
||
clock clock.Clock
|
||
// verifier 校验 WS auth 帧中的 JWT。
|
||
verifier *auth.HMACVerifier
|
||
// guards 是 room-service 守卫查询接口，im-service 不复制房间业务规则。
|
||
guards GuardClient
|
||
// sessions 保存真实 WS 连接和房间频道订阅。
|
||
sessions *session.Manager
|
||
// store 保存当前 v1 的内存消息历史、room_seq 和 ACK 待确认集合。
|
||
store *store.MemoryStore
|
||
|
||
// sinksMu 保护 conn_id -> 下行 channel 映射，防止并发断线和投递竞争。
|
||
sinksMu sync.Mutex
|
||
// sinks 是真实 WS 写协程消费的下行队列。
|
||
sinks map[string]chan OutgoingFrame
|
||
|
||
// dedupeMu 保护 roomEvents 和 notices 两类幂等表。
|
||
dedupeMu sync.Mutex
|
||
// roomEvents 保存 room-service event_id 到 im message identity 的映射，确保补投不会重复入时间线。
|
||
roomEvents map[string]struct {
|
||
roomSeq int64
|
||
messageID string
|
||
}
|
||
// notices 保存 notice_id 到 message_id 的映射，确保系统通知补偿调用可幂等返回。
|
||
notices map[string]string
|
||
}
|
||
|
||
// New 初始化 im-service 领域服务。
|
||
func New(cfg Config, verifier *auth.HMACVerifier, guards GuardClient) *Service {
|
||
usedClock := cfg.Clock
|
||
if usedClock == nil {
|
||
// 未注入时使用系统时钟，测试可传入固定时钟避免时间相关断言抖动。
|
||
usedClock = clock.System{}
|
||
}
|
||
|
||
limit := cfg.RoomHistoryLimit
|
||
if limit <= 0 {
|
||
// 房间最近历史默认 200 条，匹配 v1 方案中的重进房补拉窗口。
|
||
limit = 200
|
||
}
|
||
|
||
return &Service{
|
||
nodeID: cfg.NodeID,
|
||
roomHistoryLimit: limit,
|
||
clock: usedClock,
|
||
verifier: verifier,
|
||
guards: guards,
|
||
sessions: session.NewManager(),
|
||
store: store.NewMemoryStore(),
|
||
sinks: make(map[string]chan OutgoingFrame),
|
||
roomEvents: make(map[string]struct {
|
||
roomSeq int64
|
||
messageID string
|
||
}),
|
||
notices: make(map[string]string),
|
||
}
|
||
}
|
||
|
||
// HealthCheck 覆盖 im-service 内存态核心依赖；当前 session manager 没有后台 loop。
|
||
func (s *Service) HealthCheck() error {
|
||
if s == nil {
|
||
// nil service 说明 App 装配失败，live 必须失败而不是让空指针在请求路径上暴露。
|
||
return errors.New("im service is not initialized")
|
||
}
|
||
if s.sessions == nil {
|
||
// session manager 缺失会导致所有连接和投递路径不可用。
|
||
return errors.New("session manager is not initialized")
|
||
}
|
||
if s.store == nil {
|
||
// message store 缺失会导致 room_seq、历史和 ACK 状态不可用。
|
||
return errors.New("message store is not initialized")
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// Connect 校验 JWT，并注册一条真实在线连接。
|
||
func (s *Service) Connect(token string) (string, <-chan OutgoingFrame, error) {
|
||
claims, err := s.verifier.Verify(token)
|
||
if err != nil {
|
||
// 鉴权失败不创建 session，也不暴露部分连接态给后续帧复用。
|
||
return "", nil, err
|
||
}
|
||
|
||
now := s.clock.Now().UnixMilli()
|
||
connID := idgen.New("conn")
|
||
// 下行队列有限缓冲，避免业务 goroutine 被单个慢客户端长期阻塞。
|
||
out := make(chan OutgoingFrame, 32)
|
||
|
||
// session 只保存连接态和订阅态，不保存用户完整资料。
|
||
s.sessions.AddSession(&session.Session{
|
||
ConnID: connID,
|
||
UserID: claims.UserID,
|
||
DeviceID: claims.DeviceID,
|
||
NodeID: s.nodeID,
|
||
ConnectedAtMS: now,
|
||
LastSeenAtMS: now,
|
||
SubscribedRooms: make(map[string]bool),
|
||
})
|
||
|
||
s.sinksMu.Lock()
|
||
// sink 在 session 之后注册，保证投递路径能先通过 session 索引找到连接归属。
|
||
s.sinks[connID] = out
|
||
s.sinksMu.Unlock()
|
||
|
||
return connID, out, nil
|
||
}
|
||
|
||
// Disconnect 移除在线连接及其消息下行通道。
|
||
func (s *Service) Disconnect(connID string) {
|
||
// 先移除 session 索引，后续广播不会再把该连接列为候选接收者。
|
||
s.sessions.RemoveSession(connID)
|
||
|
||
s.sinksMu.Lock()
|
||
defer s.sinksMu.Unlock()
|
||
|
||
if sink, exists := s.sinks[connID]; exists {
|
||
// 关闭 channel 会结束 WS 写协程；重复断开由 exists 判断保持幂等。
|
||
close(sink)
|
||
delete(s.sinks, connID)
|
||
}
|
||
}
|
||
|
||
// ConnectionUserID 返回连接归属用户，供传输层在下线日志里记录真实连接身份。
|
||
func (s *Service) ConnectionUserID(connID string) (int64, bool) {
|
||
current, exists := s.sessions.GetSession(connID)
|
||
if !exists {
|
||
// handler 可能在鉴权失败或并发断线后查询，此时连接身份不可用。
|
||
return 0, false
|
||
}
|
||
|
||
return current.UserID, true
|
||
}
|
||
|
||
// SubscribeRoom 在业务上确认用户仍在房间后，把连接加入房间频道。
|
||
func (s *Service) SubscribeRoom(ctx context.Context, connID string, roomID string) error {
|
||
current, exists := s.sessions.GetSession(connID)
|
||
if !exists {
|
||
// 未鉴权或已断开的连接不能订阅任何房间频道。
|
||
return xerr.New(xerr.NotFound, "session not found")
|
||
}
|
||
|
||
// im-service 必须同步询问 room-service，避免被踢用户仅凭旧 WS 连接继续收房间消息。
|
||
guard, err := s.guards.VerifyRoomPresence(ctx, &roomv1.VerifyRoomPresenceRequest{
|
||
RoomId: roomID,
|
||
UserId: current.UserID,
|
||
SessionId: connID,
|
||
})
|
||
if err != nil {
|
||
// room-service 守卫查询失败时不本地放行，宁可让客户端重试。
|
||
return err
|
||
}
|
||
|
||
if !guard.GetPresent() {
|
||
// 业务拒绝直接返回权限错误，原因由 room-service 统一给出。
|
||
return xerr.New(xerr.PermissionDenied, guard.GetReason())
|
||
}
|
||
|
||
s.sessions.SubscribeRoom(connID, roomID)
|
||
|
||
return nil
|
||
}
|
||
|
||
// UnsubscribeRoom 把连接从房间频道移出。
|
||
func (s *Service) UnsubscribeRoom(connID string, roomID string) {
|
||
s.sessions.UnsubscribeRoom(connID, roomID)
|
||
}
|
||
|
||
// PullRoomHistory 返回房间最近 N 条历史消息。
|
||
func (s *Service) PullRoomHistory(connID string, roomID string, limit int) ([]store.Message, error) {
|
||
if _, exists := s.sessions.GetSession(connID); !exists {
|
||
// 历史拉取仍要求有效 WS session，避免匿名扫描房间最近消息。
|
||
return nil, xerr.New(xerr.NotFound, "session not found")
|
||
}
|
||
|
||
return s.store.RoomHistory(roomID, limit), nil
|
||
}
|
||
|
||
// SendRoomChat 先同步问 room-service，再分配 room_seq、入历史并广播。
|
||
func (s *Service) SendRoomChat(ctx context.Context, connID string, roomID string, content string) (store.Message, error) {
|
||
current, exists := s.sessions.GetSession(connID)
|
||
if !exists {
|
||
// 未连接用户没有发言身份，不能绕过 WS auth 直接发公屏。
|
||
return store.Message{}, xerr.New(xerr.NotFound, "session not found")
|
||
}
|
||
|
||
// 发言权限始终由 room-service 决定，im-service 只做连接层和投递层闭环。
|
||
guard, err := s.guards.CheckSpeakPermission(ctx, &roomv1.CheckSpeakPermissionRequest{
|
||
RoomId: roomID,
|
||
UserId: current.UserID,
|
||
})
|
||
if err != nil {
|
||
// 守卫 RPC 失败时不能降级为允许发言，否则会绕过禁言和房间关闭聊天状态。
|
||
return store.Message{}, err
|
||
}
|
||
|
||
if !guard.GetAllowed() {
|
||
// 拒绝原因透传给上层，调用方可转换成 WS 错误或直接断开。
|
||
return store.Message{}, xerr.New(xerr.PermissionDenied, guard.GetReason())
|
||
}
|
||
|
||
// room_seq 由 im-service 分配，确保房间系统消息和公屏消息进入同一时间线。
|
||
message := store.Message{
|
||
MessageID: idgen.New("msg"),
|
||
MessageType: "room_chat_message",
|
||
FromUserID: current.UserID,
|
||
RoomID: roomID,
|
||
RoomSeq: s.store.NextRoomSeq(roomID),
|
||
Content: content,
|
||
CreatedAtMS: s.clock.Now().UnixMilli(),
|
||
}
|
||
|
||
// 先入历史再广播，客户端收到消息后重拉历史一定能看到同一条记录。
|
||
s.store.AppendRoomMessage(roomID, message, s.roomHistoryLimit)
|
||
s.broadcastRoom(roomID, messageFrame(message))
|
||
|
||
return message, nil
|
||
}
|
||
|
||
// SendPeerMessage 投递点对点消息，并给接收方建立 ACK 待确认项。
|
||
func (s *Service) SendPeerMessage(connID string, toUserID int64, content string) (store.Message, error) {
|
||
current, exists := s.sessions.GetSession(connID)
|
||
if !exists {
|
||
// 单聊发送也必须基于已鉴权连接，FromUserID 不接受客户端自报。
|
||
return store.Message{}, xerr.New(xerr.NotFound, "session not found")
|
||
}
|
||
|
||
// v1 单聊只做基础在线投递和 ACK，不做复杂会话同步与已读未读模型。
|
||
message := store.Message{
|
||
MessageID: idgen.New("msg"),
|
||
MessageType: "peer_message",
|
||
FromUserID: current.UserID,
|
||
ToUserID: toUserID,
|
||
Content: content,
|
||
RequiresACK: true,
|
||
CreatedAtMS: s.clock.Now().UnixMilli(),
|
||
}
|
||
|
||
// 单聊需要持久化基础历史，并为接收方登记 ACK 待确认项。
|
||
s.store.SavePeerMessage(current.UserID, toUserID, message)
|
||
s.store.AddPendingACK(toUserID, message)
|
||
|
||
// 同时推给发送方和接收方在线连接，发送方可用同一消息 ID 更新本地状态。
|
||
s.pushToUser(current.UserID, messageFrame(message))
|
||
s.pushToUser(toUserID, messageFrame(message))
|
||
|
||
return message, nil
|
||
}
|
||
|
||
// Ack 标记单聊或系统通知已经被目标用户确认。
|
||
func (s *Service) Ack(connID string, messageID string) error {
|
||
current, exists := s.sessions.GetSession(connID)
|
||
if !exists {
|
||
// ACK 必须绑定当前连接对应用户，不能由客户端指定 user_id。
|
||
return xerr.New(xerr.NotFound, "session not found")
|
||
}
|
||
|
||
if !s.store.Ack(current.UserID, messageID) {
|
||
// 只允许确认属于当前用户的 pending ACK，防止跨用户清理待确认项。
|
||
return xerr.New(xerr.NotFound, "ack target not found")
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// PublishRoomEvent 把 room-service 发来的房间系统事件写入统一房间时间线，并立即广播。
|
||
func (s *Service) PublishRoomEvent(_ context.Context, req *imv1.PublishRoomEventRequest) (*imv1.PublishRoomEventResponse, error) {
|
||
event := req.GetEvent()
|
||
if event == nil || event.GetRoomId() == "" || event.GetEventId() == "" {
|
||
// room_id 和 event_id 是写入房间时间线及幂等表的硬依赖。
|
||
return nil, xerr.New(xerr.InvalidArgument, "room event is incomplete")
|
||
}
|
||
|
||
s.dedupeMu.Lock()
|
||
cached, exists := s.roomEvents[event.GetEventId()]
|
||
s.dedupeMu.Unlock()
|
||
if exists {
|
||
// Outbox 补投或同步重试命中相同 event_id 时，返回首次写入的 room_seq 和 message_id。
|
||
return &imv1.PublishRoomEventResponse{
|
||
Accepted: true,
|
||
RoomSeq: cached.roomSeq,
|
||
MessageId: cached.messageID,
|
||
}, nil
|
||
}
|
||
|
||
// room-service 只发送业务事件，im-service 在这里转换成客户端可消费的房间系统消息。
|
||
message := store.Message{
|
||
MessageID: idgen.New("msg"),
|
||
MessageType: "room_system_message",
|
||
FromUserID: event.GetActorUserId(),
|
||
ToUserID: event.GetTargetUserId(),
|
||
RoomID: event.GetRoomId(),
|
||
RoomSeq: s.store.NextRoomSeq(event.GetRoomId()),
|
||
EventID: event.GetEventId(),
|
||
Content: event.GetText(),
|
||
Attributes: cloneAttributes(event.GetAttributes()),
|
||
CreatedAtMS: s.clock.Now().UnixMilli(),
|
||
}
|
||
|
||
// 先写最近历史再广播，保证客户端收到事件后拉历史能补到同一 room_seq。
|
||
s.store.AppendRoomMessage(event.GetRoomId(), message, s.roomHistoryLimit)
|
||
s.broadcastRoom(event.GetRoomId(), messageFrame(message))
|
||
|
||
s.dedupeMu.Lock()
|
||
// 幂等表在广播后记录；当前内存实现重启会丢，后续持久化 store 应保持同一语义。
|
||
s.roomEvents[event.GetEventId()] = struct {
|
||
roomSeq int64
|
||
messageID string
|
||
}{
|
||
roomSeq: message.RoomSeq,
|
||
messageID: message.MessageID,
|
||
}
|
||
s.dedupeMu.Unlock()
|
||
|
||
return &imv1.PublishRoomEventResponse{
|
||
Accepted: true,
|
||
RoomSeq: message.RoomSeq,
|
||
MessageId: message.MessageID,
|
||
}, nil
|
||
}
|
||
|
||
// PushSystemNotice 给指定用户投递系统通知，并要求接收方 ACK。
|
||
func (s *Service) PushSystemNotice(_ context.Context, req *imv1.PushSystemNoticeRequest) (*imv1.PushSystemNoticeResponse, error) {
|
||
s.dedupeMu.Lock()
|
||
if messageID, exists := s.notices[req.GetNoticeId()]; exists && req.GetNoticeId() != "" {
|
||
s.dedupeMu.Unlock()
|
||
// notice_id 重复说明调用方正在补偿或重试，返回原 message_id 避免重复通知用户。
|
||
return &imv1.PushSystemNoticeResponse{
|
||
Accepted: true,
|
||
MessageId: messageID,
|
||
}, nil
|
||
}
|
||
s.dedupeMu.Unlock()
|
||
|
||
// 系统通知和单聊一样需要 ACK，区别是没有 FromUserID，且按目标用户保存。
|
||
message := store.Message{
|
||
MessageID: idgen.New("msg"),
|
||
MessageType: "system_notice",
|
||
ToUserID: req.GetUserId(),
|
||
Content: req.GetBody(),
|
||
Attributes: map[string]string{
|
||
"title": req.GetTitle(),
|
||
},
|
||
RequiresACK: true,
|
||
CreatedAtMS: s.clock.Now().UnixMilli(),
|
||
}
|
||
|
||
for key, value := range req.GetAttributes() {
|
||
// 调用方属性覆盖 title 以外的扩展字段，客户端按约定解释。
|
||
message.Attributes[key] = value
|
||
}
|
||
|
||
// 先保存和登记 ACK，再进行在线推送；离线补投可以基于待确认集合扩展。
|
||
s.store.SaveSystemNotice(req.GetUserId(), message)
|
||
s.store.AddPendingACK(req.GetUserId(), message)
|
||
s.pushToUser(req.GetUserId(), messageFrame(message))
|
||
|
||
if req.GetNoticeId() != "" {
|
||
s.dedupeMu.Lock()
|
||
// 只有调用方提供 notice_id 时才建立幂等关系，普通通知允许多次发送。
|
||
s.notices[req.GetNoticeId()] = message.MessageID
|
||
s.dedupeMu.Unlock()
|
||
}
|
||
|
||
return &imv1.PushSystemNoticeResponse{
|
||
Accepted: true,
|
||
MessageId: message.MessageID,
|
||
}, nil
|
||
}
|
||
|
||
// ForceRoomResync 主动通知某个用户重新拉房间状态。
|
||
func (s *Service) ForceRoomResync(_ context.Context, req *imv1.ForceRoomResyncRequest) (*imv1.ForceRoomResyncResponse, error) {
|
||
// 该控制帧不修改 im-service 状态，只通知在线客户端重新走 room/gateway 查询或重新订阅。
|
||
s.pushToUser(req.GetUserId(), OutgoingFrame{
|
||
Type: "force_resync",
|
||
RoomID: req.GetRoomId(),
|
||
Reason: req.GetReason(),
|
||
})
|
||
|
||
return &imv1.ForceRoomResyncResponse{Accepted: true}, nil
|
||
}
|
||
|
||
// MarshalFrame 统一把下行帧编码成 JSON，供 WS 写回客户端。
|
||
func MarshalFrame(frame OutgoingFrame) ([]byte, error) {
|
||
// JSON 编码集中在 service 层，WS handler 不需要理解具体下行结构。
|
||
return json.Marshal(frame)
|
||
}
|
||
|
||
// PendingACK 返回指定用户当前待确认消息数量。
|
||
func (s *Service) PendingACK(userID int64) int {
|
||
return s.store.PendingACK(userID)
|
||
}
|
||
|
||
func messageFrame(message store.Message) OutgoingFrame {
|
||
// 所有消息类型都通过统一 message 帧下发，客户端根据 MessageType 分流展示。
|
||
return OutgoingFrame{
|
||
Type: "message",
|
||
RoomID: message.RoomID,
|
||
Message: &message,
|
||
}
|
||
}
|
||
|
||
func (s *Service) broadcastRoom(roomID string, frame OutgoingFrame) {
|
||
// ListByRoom 返回连接快照；投递阶段不持有 session manager 锁，避免慢客户端阻塞索引读写。
|
||
for _, connID := range s.sessions.ListByRoom(roomID) {
|
||
s.sendFrame(connID, frame)
|
||
}
|
||
}
|
||
|
||
func (s *Service) pushToUser(userID int64, frame OutgoingFrame) {
|
||
// 同一用户可能多端在线，系统通知和单聊都推给当前所有连接。
|
||
for _, connID := range s.sessions.ListByUser(userID) {
|
||
s.sendFrame(connID, frame)
|
||
}
|
||
}
|
||
|
||
func (s *Service) sendFrame(connID string, frame OutgoingFrame) {
|
||
s.sinksMu.Lock()
|
||
sink, exists := s.sinks[connID]
|
||
s.sinksMu.Unlock()
|
||
|
||
if !exists {
|
||
// session 索引和 sink 移除不是同一把锁，断线竞态下缺失 sink 按投递失败处理。
|
||
return
|
||
}
|
||
|
||
select {
|
||
case sink <- frame:
|
||
default:
|
||
// 下行队列满说明客户端消费过慢；当前 v1 丢弃该帧，后续靠历史窗口或 ACK 补偿。
|
||
}
|
||
}
|
||
|
||
func cloneAttributes(values map[string]string) map[string]string {
|
||
if len(values) == 0 {
|
||
// nil attributes 会被 JSON omitempty 省略，减少房间广播帧体积。
|
||
return nil
|
||
}
|
||
|
||
// 拷贝 attributes，避免调用方复用 proto map 后影响已经写入历史的消息。
|
||
cloned := make(map[string]string, len(values))
|
||
for key, value := range values {
|
||
cloned[key] = value
|
||
}
|
||
|
||
return cloned
|
||
}
|