From 8d8fb77df976f8762f7f14f40a142b2c44c1be54 Mon Sep 17 00:00:00 2001 From: zhx Date: Tue, 7 Jul 2026 19:43:39 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E7=9F=AD=E6=88=BF=E5=8F=B7?= =?UTF-8?q?=E5=85=A5=E6=88=BF=E5=92=8C=E9=BA=A6=E5=BF=83=E8=B7=B3=E6=81=A2?= =?UTF-8?q?=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../internal/room/service/helpers.go | 44 +++++++++++++ .../internal/room/service/presence.go | 6 +- .../room/service/presence_short_id_test.go | 62 +++++++++++++++++++ .../internal/room/service/recovery.go | 6 +- .../internal/room/service/recovery_test.go | 43 +++++++++++++ .../internal/room/service/repository.go | 2 + .../internal/storage/mysql/room_meta.go | 24 +++++++ .../internal/storage/mysql/room_meta_test.go | 25 ++++++++ 8 files changed, 209 insertions(+), 3 deletions(-) create mode 100644 services/room-service/internal/room/service/presence_short_id_test.go create mode 100644 services/room-service/internal/room/service/recovery_test.go create mode 100644 services/room-service/internal/storage/mysql/room_meta_test.go diff --git a/services/room-service/internal/room/service/helpers.go b/services/room-service/internal/room/service/helpers.go index 779e8062..67e31e0e 100644 --- a/services/room-service/internal/room/service/helpers.go +++ b/services/room-service/internal/room/service/helpers.go @@ -4,6 +4,7 @@ import ( "context" "log/slog" "slices" + "strings" "time" roomv1 "hyapp.local/api/proto/room/v1" @@ -201,6 +202,49 @@ func baseFromMeta(meta *roomv1.RequestMeta) command.Base { } } +func (s *Service) canonicalRoomCommandMeta(ctx context.Context, meta *roomv1.RequestMeta) (*roomv1.RequestMeta, error) { + if meta == nil { + return nil, nil + } + requestedRoomID := strings.TrimSpace(meta.GetRoomId()) + if requestedRoomID == "" { + return meta, nil + } + + if roomMeta, exists, err := s.repository.GetRoomMeta(ctx, requestedRoomID); err != nil { + return nil, err + } else if exists { + if roomMeta.RoomID == meta.GetRoomId() { + return meta, nil + } + // 精确 room_id 命中时也用数据库里的规范值，避免首尾空白进入 lease key 或 command log。 + return requestMetaWithRoomID(meta, roomMeta.RoomID), nil + } + + roomMeta, exists, err := s.repository.GetRoomMetaByShortID(ctx, requestedRoomID) + if err != nil || !exists { + return meta, err + } + // 客户端可能只知道展示短房号；进入 Room Cell 前必须收敛到内部 room_id，保证 lease、command log、snapshot 分区一致。 + return requestMetaWithRoomID(meta, roomMeta.RoomID), nil +} + +func requestMetaWithRoomID(meta *roomv1.RequestMeta, roomID string) *roomv1.RequestMeta { + if meta == nil { + return nil + } + return &roomv1.RequestMeta{ + RequestId: meta.GetRequestId(), + CommandId: meta.GetCommandId(), + ActorUserId: meta.GetActorUserId(), + RoomId: roomID, + GatewayNodeId: meta.GetGatewayNodeId(), + SessionId: meta.GetSessionId(), + SentAtMs: meta.GetSentAtMs(), + AppCode: meta.GetAppCode(), + } +} + func contextFromMeta(ctx context.Context, meta *roomv1.RequestMeta) context.Context { if meta == nil { return appcode.WithContext(ctx, "") diff --git a/services/room-service/internal/room/service/presence.go b/services/room-service/internal/room/service/presence.go index 60ea3c86..b475ac33 100644 --- a/services/room-service/internal/room/service/presence.go +++ b/services/room-service/internal/room/service/presence.go @@ -24,10 +24,14 @@ const joinRoomResponseModeLite = "lite" // JoinRoom 把用户业务态接入房间。 func (s *Service) JoinRoom(ctx context.Context, req *roomv1.JoinRoomRequest) (*roomv1.JoinRoomResponse, error) { ctx = contextFromMeta(ctx, req.GetMeta()) + meta, err := s.canonicalRoomCommandMeta(ctx, req.GetMeta()) + if err != nil { + return nil, err + } password := strings.TrimSpace(req.GetPassword()) // JoinRoom 只维护 room-service presence，不创建任何本地长连接订阅。 cmd := command.JoinRoom{ - Base: baseFromMeta(req.GetMeta()), + Base: baseFromMeta(meta), Role: req.GetRole(), ActorCountryID: req.GetActorCountryId(), ActorRegionID: req.GetActorRegionId(), diff --git a/services/room-service/internal/room/service/presence_short_id_test.go b/services/room-service/internal/room/service/presence_short_id_test.go new file mode 100644 index 00000000..fd4e3763 --- /dev/null +++ b/services/room-service/internal/room/service/presence_short_id_test.go @@ -0,0 +1,62 @@ +package service_test + +import ( + "context" + "testing" + "time" + + roomv1 "hyapp.local/api/proto/room/v1" + "hyapp/pkg/appcode" + "hyapp/services/room-service/internal/integration" + roomservice "hyapp/services/room-service/internal/room/service" + "hyapp/services/room-service/internal/router" + "hyapp/services/room-service/internal/testutil/mysqltest" +) + +func TestJoinRoomAcceptsRoomShortIDWithoutWritingShortIDCommandLog(t *testing.T) { + ctx := context.Background() + repository := mysqltest.NewRepository(t) + svc := roomservice.New(roomservice.Config{ + NodeID: "node-short-id-join-test", + LeaseTTL: 10 * time.Second, + RankLimit: 20, + SnapshotEveryN: 1, + }, router.NewMemoryDirectory(), repository, followTestWallet{}, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher()) + + roomID := "room-short-id-join" + roomShortID := "167822" + ownerID := int64(6101) + viewerID := int64(6102) + if _, err := svc.CreateRoom(ctx, &roomv1.CreateRoomRequest{ + Meta: roomservice.NewRequestMeta(roomID, ownerID), + SeatCount: 10, + Mode: "voice", + VisibleRegionId: 6101, + RoomName: "Short ID Join", + RoomAvatar: testRoomCoverURL, + RoomShortId: roomShortID, + }); err != nil { + t.Fatalf("create room: %v", err) + } + + joinMeta := roomservice.NewRequestMeta(roomShortID, viewerID) + joinMeta.CommandId = "cmd-join-by-room-short-id" + resp, err := svc.JoinRoom(ctx, &roomv1.JoinRoomRequest{ + Meta: joinMeta, + Role: "audience", + }) + if err != nil { + t.Fatalf("join room by short id: %v", err) + } + if resp.GetRoom().GetRoomId() != roomID || resp.GetRoom().GetRoomShortId() != roomShortID { + t.Fatalf("join response must expose canonical room id and short id, got %+v", resp.GetRoom()) + } + + repoCtx := appcode.WithContext(context.Background(), appcode.Default) + if _, exists, err := repository.GetCommand(repoCtx, roomID, joinMeta.GetCommandId()); err != nil || !exists { + t.Fatalf("join command must be written under canonical room id, exists=%v err=%v", exists, err) + } + if _, exists, err := repository.GetCommand(repoCtx, roomShortID, joinMeta.GetCommandId()); err != nil || exists { + t.Fatalf("short id must not create an independent command log partition, exists=%v err=%v", exists, err) + } +} diff --git a/services/room-service/internal/room/service/recovery.go b/services/room-service/internal/room/service/recovery.go index a8f1419d..02bbf930 100644 --- a/services/room-service/internal/room/service/recovery.go +++ b/services/room-service/internal/room/service/recovery.go @@ -259,10 +259,12 @@ func replay(current *state.RoomState, cmd command.Command, committedAtMS int64) case *command.MicHeartbeat: seat, exists := current.SeatByUser(typed.TargetUserID) if !exists { - return fmt.Errorf("mic_heartbeat replay target is not on seat: %d", typed.TargetUserID) + // 麦上心跳只是当前会话的 liveness 刷新；旧心跳晚于离房/下麦落库时，不能重新占麦，也不能阻断房间恢复。 + return nil } if seat.MicSessionID != typed.MicSessionID || seat.PublishState != state.MicPublishPublishing { - return fmt.Errorf("mic_heartbeat replay target session is not publishing: %d", typed.TargetUserID) + // 会话已被新一轮上麦或 pending 状态替换时，旧心跳不再代表当前麦位事实，恢复时按过期事件跳过。 + return nil } if existing, ok := current.OnlineUsers[typed.TargetUserID]; ok { // 麦上心跳回放要恢复同步刷新的 presence 时间，避免 snapshot 落后时重启后又被 stale worker 清理。 diff --git a/services/room-service/internal/room/service/recovery_test.go b/services/room-service/internal/room/service/recovery_test.go new file mode 100644 index 00000000..7759ac4b --- /dev/null +++ b/services/room-service/internal/room/service/recovery_test.go @@ -0,0 +1,43 @@ +package service + +import ( + "testing" + + "hyapp/services/room-service/internal/room/command" + "hyapp/services/room-service/internal/room/state" +) + +func TestReplayMicHeartbeatSkipsStaleMissingSeat(t *testing.T) { + current := state.NewRoomState("room-replay-stale-heartbeat", 7101, 8, "voice") + current.OnlineUsers[7101] = &state.RoomUserState{UserID: 7101, Role: "owner", JoinedAtMS: 1000, LastSeenAtMS: 1000} + + if err := replay(current, &command.MicHeartbeat{ + Base: command.Base{ActorID: 7101, Room: current.RoomID, SentAtMS: 2000}, + TargetUserID: 7101, + MicSessionID: "mic-old", + }, 2000); err != nil { + t.Fatalf("stale heartbeat without seat must be skipped: %v", err) + } + if current.Version != 0 || current.OnlineUsers[7101].LastSeenAtMS != 1000 { + t.Fatalf("stale heartbeat must not mutate state, version=%d user=%+v", current.Version, current.OnlineUsers[7101]) + } +} + +func TestReplayMicHeartbeatSkipsStaleSession(t *testing.T) { + current := state.NewRoomState("room-replay-stale-session", 7201, 8, "voice") + current.OnlineUsers[7201] = &state.RoomUserState{UserID: 7201, Role: "owner", JoinedAtMS: 1000, LastSeenAtMS: 1000} + current.MicSeats[0].UserID = 7201 + current.MicSeats[0].PublishState = state.MicPublishPending + current.MicSeats[0].MicSessionID = "mic-new" + + if err := replay(current, &command.MicHeartbeat{ + Base: command.Base{ActorID: 7201, Room: current.RoomID, SentAtMS: 2000}, + TargetUserID: 7201, + MicSessionID: "mic-old", + }, 2000); err != nil { + t.Fatalf("stale heartbeat for replaced session must be skipped: %v", err) + } + if current.Version != 0 || current.MicSeats[0].MicHeartbeatAtMS != 0 || current.OnlineUsers[7201].LastSeenAtMS != 1000 { + t.Fatalf("stale heartbeat must not refresh pending/replaced session, version=%d seat=%+v user=%+v", current.Version, current.MicSeats[0], current.OnlineUsers[7201]) + } +} diff --git a/services/room-service/internal/room/service/repository.go b/services/room-service/internal/room/service/repository.go index c75487c9..32307977 100644 --- a/services/room-service/internal/room/service/repository.go +++ b/services/room-service/internal/room/service/repository.go @@ -715,6 +715,8 @@ type RoomMetaStore interface { SaveRoomMeta(ctx context.Context, meta RoomMeta) error // GetRoomMeta 读取房间基础元数据，恢复无内存 Cell 的房间依赖它。 GetRoomMeta(ctx context.Context, roomID string) (RoomMeta, bool, error) + // GetRoomMetaByShortID 用客户端展示短房号解析内部 room_id；命令链路必须先 canonicalize 后再取 lease 和写 command log。 + GetRoomMetaByShortID(ctx context.Context, roomShortID string) (RoomMeta, bool, error) // GetRoomMetaByOwner 查询 owner 已创建的房间；产品规则要求一个用户只能创建一个房间。 GetRoomMetaByOwner(ctx context.Context, ownerUserID int64) (RoomMeta, bool, error) // SaveRoomBackground 保存房间 owner 上传过的背景图 URL；相同房间相同 URL 幂等返回同一条素材。 diff --git a/services/room-service/internal/storage/mysql/room_meta.go b/services/room-service/internal/storage/mysql/room_meta.go index 0cfc175f..f8f2c543 100644 --- a/services/room-service/internal/storage/mysql/room_meta.go +++ b/services/room-service/internal/storage/mysql/room_meta.go @@ -60,6 +60,30 @@ func (r *Repository) GetRoomMeta(ctx context.Context, roomID string) (roomservic return meta, true, nil } +// GetRoomMetaByShortID 通过展示短房号解析内部房间 ID。 +func (r *Repository) GetRoomMetaByShortID(ctx context.Context, roomShortID string) (roomservice.RoomMeta, bool, error) { + // rooms.uk_rooms_short_id 保证同 app 下短房号唯一；这里只做解析，不改变 GetRoomMeta 的精确 room_id 语义。 + row := r.db.QueryRowContext(ctx, + `SELECT app_code, room_id, room_short_id, owner_user_id, seat_count, mode, status, room_password_hash, visible_region_id + FROM rooms + WHERE app_code = ? AND room_short_id = ? + LIMIT 1`, + appcode.FromContext(ctx), + strings.TrimSpace(roomShortID), + ) + + var meta roomservice.RoomMeta + if err := row.Scan(&meta.AppCode, &meta.RoomID, &meta.RoomShortID, &meta.OwnerUserID, &meta.SeatCount, &meta.Mode, &meta.Status, &meta.RoomPasswordHash, &meta.VisibleRegionID); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return roomservice.RoomMeta{}, false, nil + } + + return roomservice.RoomMeta{}, false, err + } + + return meta, true, nil +} + // GetRoomMetaByOwner 查询用户已经创建的房间。 func (r *Repository) GetRoomMetaByOwner(ctx context.Context, ownerUserID int64) (roomservice.RoomMeta, bool, error) { // rooms.owner_user_id 有唯一约束，查询最多返回一条；该读路径用于 CreateRoom 前置失败。 diff --git a/services/room-service/internal/storage/mysql/room_meta_test.go b/services/room-service/internal/storage/mysql/room_meta_test.go new file mode 100644 index 00000000..ead304d4 --- /dev/null +++ b/services/room-service/internal/storage/mysql/room_meta_test.go @@ -0,0 +1,25 @@ +package mysql + +import ( + "context" + "testing" + + "hyapp/pkg/appcode" +) + +func TestGetRoomMetaByShortIDReturnsCanonicalRoomID(t *testing.T) { + ctx := appcode.WithContext(context.Background(), appcode.Default) + repo := newCommandLogTestRepository(t, ctx) + insertCommandLogTestRoom(t, ctx, repo, "room-meta-short") + + meta, exists, err := repo.GetRoomMetaByShortID(ctx, " room-meta-short-short ") + if err != nil { + t.Fatalf("get room meta by short id: %v", err) + } + if !exists { + t.Fatal("room short id should resolve existing room meta") + } + if meta.RoomID != "room-meta-short" || meta.RoomShortID != "room-meta-short-short" { + t.Fatalf("short id must return canonical room meta, got %+v", meta) + } +}