253 lines
7.5 KiB
Go
253 lines
7.5 KiB
Go
// Package mysql 实现 room-service 的 MySQL 持久化边界。
|
||
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"strings"
|
||
|
||
"hyapp/pkg/appcode"
|
||
roomservice "hyapp/services/room-service/internal/room/service"
|
||
)
|
||
|
||
func (r *Repository) ProjectRoomPresence(ctx context.Context, snapshot roomservice.RoomPresenceSnapshot) error {
|
||
appCode := normalizedRecordAppCode(ctx, snapshot.AppCode)
|
||
if strings.TrimSpace(snapshot.RoomID) == "" {
|
||
return nil
|
||
}
|
||
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 先把该房间上一版 active presence 全部置为 left，再把本次快照里的在线用户 upsert 回 active。
|
||
// 这样 Join/Leave/Kick/Close 的投影逻辑统一，避免按命令类型维护多份分支。
|
||
if _, err := tx.ExecContext(ctx,
|
||
`UPDATE room_user_presence
|
||
SET status = ?, publish_state = '', mic_session_id = '', room_version = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND room_id = ? AND status = ?`,
|
||
roomPresenceStatusLeftSQL,
|
||
snapshot.RoomVersion,
|
||
snapshot.UpdatedAtMS,
|
||
appCode,
|
||
snapshot.RoomID,
|
||
roomPresenceStatusActiveSQL,
|
||
); err != nil {
|
||
_ = tx.Rollback()
|
||
return err
|
||
}
|
||
|
||
for _, user := range snapshot.Users {
|
||
if user.UserID <= 0 {
|
||
continue
|
||
}
|
||
if _, err := tx.ExecContext(ctx,
|
||
`INSERT INTO room_user_presence (
|
||
app_code, user_id, room_id, role, publish_state, mic_session_id, room_version, status, joined_at_ms, last_seen_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
room_id = VALUES(room_id),
|
||
role = VALUES(role),
|
||
publish_state = VALUES(publish_state),
|
||
mic_session_id = VALUES(mic_session_id),
|
||
room_version = VALUES(room_version),
|
||
status = VALUES(status),
|
||
joined_at_ms = VALUES(joined_at_ms),
|
||
last_seen_at_ms = VALUES(last_seen_at_ms),
|
||
updated_at_ms = VALUES(updated_at_ms)`,
|
||
appCode,
|
||
user.UserID,
|
||
snapshot.RoomID,
|
||
user.Role,
|
||
user.PublishState,
|
||
user.MicSessionID,
|
||
snapshot.RoomVersion,
|
||
roomPresenceStatusActiveSQL,
|
||
user.JoinedAtMS,
|
||
user.LastSeenAtMS,
|
||
snapshot.UpdatedAtMS,
|
||
); err != nil {
|
||
_ = tx.Rollback()
|
||
return err
|
||
}
|
||
}
|
||
|
||
return tx.Commit()
|
||
}
|
||
|
||
// UpsertRoomUserPresence 只写入单个 active 用户，避免 JoinRoom 为整房间在线用户产生 O(N) 写放大。
|
||
func (r *Repository) UpsertRoomUserPresence(ctx context.Context, presence roomservice.RoomPresence) error {
|
||
appCode := normalizedRecordAppCode(ctx, presence.AppCode)
|
||
if presence.UserID <= 0 || strings.TrimSpace(presence.RoomID) == "" {
|
||
return nil
|
||
}
|
||
_, err := r.db.ExecContext(ctx,
|
||
`INSERT INTO room_user_presence (
|
||
app_code, user_id, room_id, role, publish_state, mic_session_id, room_version, status, joined_at_ms, last_seen_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
room_id = VALUES(room_id),
|
||
role = VALUES(role),
|
||
publish_state = VALUES(publish_state),
|
||
mic_session_id = VALUES(mic_session_id),
|
||
room_version = VALUES(room_version),
|
||
status = VALUES(status),
|
||
joined_at_ms = VALUES(joined_at_ms),
|
||
last_seen_at_ms = VALUES(last_seen_at_ms),
|
||
updated_at_ms = VALUES(updated_at_ms)`,
|
||
appCode,
|
||
presence.UserID,
|
||
presence.RoomID,
|
||
presence.Role,
|
||
presence.PublishState,
|
||
presence.MicSessionID,
|
||
presence.RoomVersion,
|
||
roomPresenceStatusActiveSQL,
|
||
presence.JoinedAtMS,
|
||
presence.LastSeenAtMS,
|
||
presence.UpdatedAtMS,
|
||
)
|
||
return err
|
||
}
|
||
|
||
// MarkRoomUserPresenceLeft 只关闭指定用户在指定房间的 active presence，避免旧离房投影误伤新房间。
|
||
func (r *Repository) MarkRoomUserPresenceLeft(ctx context.Context, presence roomservice.RoomPresence) error {
|
||
appCode := normalizedRecordAppCode(ctx, presence.AppCode)
|
||
if presence.UserID <= 0 || strings.TrimSpace(presence.RoomID) == "" {
|
||
return nil
|
||
}
|
||
_, err := r.db.ExecContext(ctx,
|
||
`UPDATE room_user_presence
|
||
SET status = ?, publish_state = '', mic_session_id = '', room_version = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND user_id = ? AND room_id = ? AND status = ?`,
|
||
roomPresenceStatusLeftSQL,
|
||
presence.RoomVersion,
|
||
presence.UpdatedAtMS,
|
||
appCode,
|
||
presence.UserID,
|
||
presence.RoomID,
|
||
roomPresenceStatusActiveSQL,
|
||
)
|
||
return err
|
||
}
|
||
|
||
// GetCurrentRoomPresence 查询用户当前 active 房间，不扫描发现页列表。
|
||
func (r *Repository) GetCurrentRoomPresence(ctx context.Context, userID int64) (roomservice.RoomPresence, bool, error) {
|
||
row := r.db.QueryRowContext(ctx,
|
||
`SELECT app_code, user_id, room_id, role, publish_state, mic_session_id, room_version, status, joined_at_ms, last_seen_at_ms, updated_at_ms
|
||
FROM room_user_presence
|
||
WHERE app_code = ? AND user_id = ? AND status = ?
|
||
LIMIT 1`,
|
||
appcode.FromContext(ctx),
|
||
userID,
|
||
roomPresenceStatusActiveSQL,
|
||
)
|
||
|
||
var presence roomservice.RoomPresence
|
||
if err := row.Scan(
|
||
&presence.AppCode,
|
||
&presence.UserID,
|
||
&presence.RoomID,
|
||
&presence.Role,
|
||
&presence.PublishState,
|
||
&presence.MicSessionID,
|
||
&presence.RoomVersion,
|
||
&presence.Status,
|
||
&presence.JoinedAtMS,
|
||
&presence.LastSeenAtMS,
|
||
&presence.UpdatedAtMS,
|
||
); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return roomservice.RoomPresence{}, false, nil
|
||
}
|
||
return roomservice.RoomPresence{}, false, err
|
||
}
|
||
|
||
return presence, true, nil
|
||
}
|
||
|
||
// ListRoomOnlineUsers 分页查询某个房间当前 active presence。
|
||
func (r *Repository) ListRoomOnlineUsers(ctx context.Context, query roomservice.RoomOnlineUserQuery) (roomservice.RoomOnlineUserPage, error) {
|
||
page := query.Page
|
||
if page <= 0 {
|
||
page = 1
|
||
}
|
||
pageSize := query.PageSize
|
||
if pageSize <= 0 {
|
||
pageSize = 50
|
||
}
|
||
offset := (page - 1) * pageSize
|
||
appCode := normalizedRecordAppCode(ctx, query.AppCode)
|
||
|
||
var total int64
|
||
if err := r.db.QueryRowContext(ctx,
|
||
`SELECT COUNT(*)
|
||
FROM room_user_presence
|
||
WHERE app_code = ? AND room_id = ? AND status = ?`,
|
||
appCode,
|
||
query.RoomID,
|
||
roomPresenceStatusActiveSQL,
|
||
).Scan(&total); err != nil {
|
||
return roomservice.RoomOnlineUserPage{}, err
|
||
}
|
||
|
||
orderBy := "p.last_seen_at_ms DESC, p.user_id ASC"
|
||
if strings.EqualFold(strings.TrimSpace(query.Sort), "gift_value") {
|
||
orderBy = "COALESCE(g.gift_value, 0) DESC, p.last_seen_at_ms DESC, p.user_id ASC"
|
||
}
|
||
|
||
rows, err := r.db.QueryContext(ctx,
|
||
`SELECT p.app_code, p.user_id, p.room_id, p.role, p.publish_state, p.mic_session_id, p.room_version, p.status,
|
||
p.joined_at_ms, p.last_seen_at_ms, p.updated_at_ms, COALESCE(g.gift_value, 0)
|
||
FROM room_user_presence p
|
||
LEFT JOIN room_user_gift_stats g
|
||
ON g.app_code = p.app_code AND g.room_id = p.room_id AND g.user_id = p.user_id
|
||
WHERE p.app_code = ? AND p.room_id = ? AND p.status = ?
|
||
ORDER BY `+orderBy+`
|
||
LIMIT ? OFFSET ?`,
|
||
appCode,
|
||
query.RoomID,
|
||
roomPresenceStatusActiveSQL,
|
||
pageSize,
|
||
offset,
|
||
)
|
||
if err != nil {
|
||
return roomservice.RoomOnlineUserPage{}, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
users := make([]roomservice.RoomPresence, 0, pageSize)
|
||
for rows.Next() {
|
||
var presence roomservice.RoomPresence
|
||
if err := rows.Scan(
|
||
&presence.AppCode,
|
||
&presence.UserID,
|
||
&presence.RoomID,
|
||
&presence.Role,
|
||
&presence.PublishState,
|
||
&presence.MicSessionID,
|
||
&presence.RoomVersion,
|
||
&presence.Status,
|
||
&presence.JoinedAtMS,
|
||
&presence.LastSeenAtMS,
|
||
&presence.UpdatedAtMS,
|
||
&presence.GiftValue,
|
||
); err != nil {
|
||
return roomservice.RoomOnlineUserPage{}, err
|
||
}
|
||
users = append(users, presence)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return roomservice.RoomOnlineUserPage{}, err
|
||
}
|
||
|
||
return roomservice.RoomOnlineUserPage{
|
||
Users: users,
|
||
Total: total,
|
||
Page: page,
|
||
PageSize: pageSize,
|
||
}, nil
|
||
}
|