1791 lines
60 KiB
Go
1791 lines
60 KiB
Go
// Package mysql 实现 room-service 的 MySQL 持久化边界。
|
||
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"errors"
|
||
"sort"
|
||
"strings"
|
||
"time"
|
||
|
||
mysqlerr "github.com/go-sql-driver/mysql"
|
||
"google.golang.org/protobuf/proto"
|
||
roomeventsv1 "hyapp.local/api/proto/events/room/v1"
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/xerr"
|
||
"hyapp/services/room-service/internal/room/outbox"
|
||
roomservice "hyapp/services/room-service/internal/room/service"
|
||
)
|
||
|
||
// Repository 把 room-service 的权威状态持久化到 MySQL。
|
||
type Repository struct {
|
||
// db 是标准库连接池，所有 repository 方法都通过它访问 MySQL。
|
||
db *sql.DB
|
||
}
|
||
|
||
// Open 建立 MySQL 连接池，并验证数据库可用。
|
||
func Open(ctx context.Context, dsn string, maxOpenConns int, maxIdleConns int) (*Repository, error) {
|
||
// sql.Open 只校验 driver 和 DSN 形态，真实连通性由后续 PingContext 验证。
|
||
db, err := sql.Open("mysql", dsn)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
if maxOpenConns > 0 {
|
||
// 只在配置显式给出正数时覆盖标准库默认值。
|
||
db.SetMaxOpenConns(maxOpenConns)
|
||
}
|
||
|
||
if maxIdleConns > 0 {
|
||
// 空闲连接数需要小于等于最大连接数，由配置层负责合理设置。
|
||
db.SetMaxIdleConns(maxIdleConns)
|
||
}
|
||
|
||
// 长连接定期轮换，避免云数据库或网络层回收导致长期坏连接驻留。
|
||
db.SetConnMaxLifetime(30 * time.Minute)
|
||
|
||
if err := db.PingContext(ctx); err != nil {
|
||
// 启动探测失败时必须关闭连接池，避免返回半可用 repository。
|
||
_ = db.Close()
|
||
return nil, err
|
||
}
|
||
|
||
return &Repository{db: db}, nil
|
||
}
|
||
|
||
// Close 关闭底层 MySQL 连接池。
|
||
func (r *Repository) Close() error {
|
||
// App.Close 调用该方法释放所有空闲连接，并阻止后续新查询。
|
||
return r.db.Close()
|
||
}
|
||
|
||
// Ping 验证 MySQL 连接池当前仍能在调用方 deadline 内拿到可用连接。
|
||
func (r *Repository) Ping(ctx context.Context) error {
|
||
// ready 探针通过该方法确认 command log/snapshot/outbox 的持久化来源可用。
|
||
return r.db.PingContext(ctx)
|
||
}
|
||
|
||
// Migrate 创建 room-service 首版需要的 MySQL 表。
|
||
func (r *Repository) Migrate(ctx context.Context) error {
|
||
// 这里保持与 deploy/mysql/initdb 的表结构一致，服务本地启动可自动建表。
|
||
statements := []string{
|
||
`CREATE TABLE IF NOT EXISTS rooms (
|
||
app_code VARCHAR(32) NOT NULL,
|
||
room_id VARCHAR(64) NOT NULL,
|
||
room_short_id VARCHAR(32) NOT NULL DEFAULT '',
|
||
owner_user_id BIGINT NOT NULL,
|
||
seat_count INT NOT NULL,
|
||
mode VARCHAR(64) NOT NULL,
|
||
status VARCHAR(32) NOT NULL,
|
||
room_password_hash VARCHAR(128) NOT NULL DEFAULT '',
|
||
close_reason VARCHAR(512) NOT NULL DEFAULT '',
|
||
closed_by_admin_id BIGINT UNSIGNED NOT NULL DEFAULT 0,
|
||
closed_by_admin_name VARCHAR(128) NOT NULL DEFAULT '',
|
||
closed_at_ms BIGINT NOT NULL DEFAULT 0,
|
||
visible_region_id BIGINT NOT NULL DEFAULT 0,
|
||
created_at_ms BIGINT NOT NULL,
|
||
updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, room_id),
|
||
UNIQUE KEY uk_rooms_short_id (app_code, room_short_id),
|
||
UNIQUE KEY uk_rooms_owner_user (app_code, owner_user_id),
|
||
KEY idx_rooms_region_status (app_code, visible_region_id, status)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS room_list_entries (
|
||
app_code VARCHAR(32) NOT NULL,
|
||
room_id VARCHAR(64) NOT NULL,
|
||
room_short_id VARCHAR(32) NOT NULL DEFAULT '',
|
||
visible_region_id BIGINT NOT NULL DEFAULT 0,
|
||
owner_user_id BIGINT NOT NULL,
|
||
title VARCHAR(128) NOT NULL DEFAULT '',
|
||
cover_url VARCHAR(512) NOT NULL DEFAULT '',
|
||
mode VARCHAR(64) NOT NULL,
|
||
status VARCHAR(32) NOT NULL,
|
||
locked BOOLEAN NOT NULL DEFAULT FALSE,
|
||
close_reason VARCHAR(512) NOT NULL DEFAULT '',
|
||
closed_by_admin_id BIGINT UNSIGNED NOT NULL DEFAULT 0,
|
||
closed_by_admin_name VARCHAR(128) NOT NULL DEFAULT '',
|
||
closed_at_ms BIGINT NOT NULL DEFAULT 0,
|
||
heat BIGINT NOT NULL DEFAULT 0,
|
||
online_count INT NOT NULL DEFAULT 0,
|
||
seat_count INT NOT NULL DEFAULT 0,
|
||
occupied_seat_count INT NOT NULL DEFAULT 0,
|
||
sort_score BIGINT NOT NULL DEFAULT 0,
|
||
created_at_ms BIGINT NOT NULL,
|
||
updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, room_id),
|
||
KEY idx_room_list_short_id (app_code, room_short_id),
|
||
KEY idx_room_list_region_hot (app_code, visible_region_id, status, sort_score DESC, room_id),
|
||
KEY idx_room_list_region_new (app_code, visible_region_id, status, created_at_ms DESC, room_id),
|
||
KEY idx_room_list_owner (app_code, owner_user_id, updated_at_ms)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS room_user_presence (
|
||
app_code VARCHAR(32) NOT NULL,
|
||
user_id BIGINT NOT NULL,
|
||
room_id VARCHAR(64) NOT NULL,
|
||
role VARCHAR(32) NOT NULL,
|
||
publish_state VARCHAR(32) NOT NULL DEFAULT '',
|
||
mic_session_id VARCHAR(128) NOT NULL DEFAULT '',
|
||
room_version BIGINT NOT NULL,
|
||
status VARCHAR(32) NOT NULL,
|
||
last_seen_at_ms BIGINT NOT NULL DEFAULT 0,
|
||
updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, user_id),
|
||
KEY idx_room_user_presence_room (app_code, room_id, status),
|
||
KEY idx_room_user_presence_updated (app_code, status, updated_at_ms)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS room_snapshots (
|
||
app_code VARCHAR(32) NOT NULL,
|
||
room_id VARCHAR(64) NOT NULL,
|
||
room_version BIGINT NOT NULL,
|
||
payload LONGBLOB NOT NULL,
|
||
created_at_ms BIGINT NOT NULL,
|
||
updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, room_id),
|
||
INDEX idx_room_snapshots_version (app_code, room_id, room_version)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS room_command_log (
|
||
app_code VARCHAR(32) NOT NULL,
|
||
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
|
||
room_id VARCHAR(64) NOT NULL,
|
||
room_version BIGINT NOT NULL,
|
||
command_id VARCHAR(128) NOT NULL,
|
||
command_type VARCHAR(64) NOT NULL,
|
||
owner_node_id VARCHAR(128) NOT NULL DEFAULT '',
|
||
lease_token VARCHAR(128) NOT NULL DEFAULT '',
|
||
payload LONGBLOB NOT NULL,
|
||
replayable BOOLEAN NOT NULL,
|
||
created_at_ms BIGINT NOT NULL,
|
||
UNIQUE KEY uk_room_command (app_code, room_id, command_id),
|
||
KEY idx_room_command_replay (app_code, room_id, room_version)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS room_outbox (
|
||
app_code VARCHAR(32) NOT NULL,
|
||
event_id VARCHAR(128) NOT NULL,
|
||
event_type VARCHAR(64) NOT NULL,
|
||
room_id VARCHAR(64) NOT NULL,
|
||
status VARCHAR(32) NOT NULL,
|
||
worker_id VARCHAR(128) NOT NULL DEFAULT '',
|
||
lock_until_ms BIGINT NULL,
|
||
envelope LONGBLOB NOT NULL,
|
||
created_at_ms BIGINT NOT NULL,
|
||
retry_count INT NOT NULL DEFAULT 0,
|
||
next_retry_at_ms BIGINT NOT NULL DEFAULT 0,
|
||
last_error TEXT NULL,
|
||
updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, event_id),
|
||
KEY idx_room_outbox_pending (app_code, status, next_retry_at_ms, created_at_ms),
|
||
KEY idx_room_outbox_claim (app_code, status, lock_until_ms, created_at_ms)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS room_user_feed_entries (
|
||
app_code VARCHAR(32) NOT NULL,
|
||
user_id BIGINT NOT NULL,
|
||
feed_type VARCHAR(32) NOT NULL,
|
||
room_id VARCHAR(64) NOT NULL,
|
||
created_at_ms BIGINT NOT NULL,
|
||
updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, user_id, feed_type, room_id),
|
||
KEY idx_room_user_feed_list (app_code, user_id, feed_type, updated_at_ms DESC, room_id)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS room_follows (
|
||
app_code VARCHAR(32) NOT NULL,
|
||
user_id BIGINT NOT NULL,
|
||
room_id VARCHAR(64) NOT NULL,
|
||
status VARCHAR(32) NOT NULL,
|
||
followed_at_ms BIGINT NOT NULL,
|
||
updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, user_id, room_id),
|
||
KEY idx_room_follows_user_list (app_code, user_id, status, followed_at_ms DESC, room_id),
|
||
KEY idx_room_follows_room (app_code, room_id, status)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS room_region_pins (
|
||
id BIGINT NOT NULL AUTO_INCREMENT,
|
||
app_code VARCHAR(32) NOT NULL,
|
||
visible_region_id BIGINT NOT NULL,
|
||
room_id VARCHAR(64) NOT NULL,
|
||
weight BIGINT NOT NULL DEFAULT 0,
|
||
status VARCHAR(32) NOT NULL,
|
||
pinned_at_ms BIGINT NOT NULL,
|
||
expires_at_ms BIGINT NOT NULL,
|
||
cancelled_at_ms BIGINT NOT NULL DEFAULT 0,
|
||
created_by_admin_id BIGINT UNSIGNED NOT NULL DEFAULT 0,
|
||
cancelled_by_admin_id BIGINT UNSIGNED NOT NULL DEFAULT 0,
|
||
created_at_ms BIGINT NOT NULL,
|
||
updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (id),
|
||
UNIQUE KEY uk_room_region_pins_room (app_code, visible_region_id, room_id),
|
||
KEY idx_room_region_pins_active (app_code, visible_region_id, status, expires_at_ms, weight DESC, room_id),
|
||
KEY idx_room_region_pins_room (app_code, room_id, status)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS room_seat_configs (
|
||
app_code VARCHAR(32) NOT NULL,
|
||
allowed_seat_counts VARCHAR(128) NOT NULL,
|
||
default_seat_count INT NOT NULL,
|
||
created_at_ms BIGINT NOT NULL,
|
||
updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`INSERT IGNORE INTO room_seat_configs (app_code, allowed_seat_counts, default_seat_count, created_at_ms, updated_at_ms)
|
||
VALUES ('lalu', '[10,15,20,25,30]', 15, 0, 0)`,
|
||
`ALTER TABLE rooms MODIFY COLUMN app_code VARCHAR(32) NOT NULL`,
|
||
`ALTER TABLE room_list_entries MODIFY COLUMN app_code VARCHAR(32) NOT NULL`,
|
||
`ALTER TABLE room_user_presence MODIFY COLUMN app_code VARCHAR(32) NOT NULL`,
|
||
`ALTER TABLE room_snapshots MODIFY COLUMN app_code VARCHAR(32) NOT NULL`,
|
||
`ALTER TABLE room_command_log MODIFY COLUMN app_code VARCHAR(32) NOT NULL`,
|
||
`ALTER TABLE room_outbox MODIFY COLUMN app_code VARCHAR(32) NOT NULL`,
|
||
`ALTER TABLE room_user_feed_entries MODIFY COLUMN app_code VARCHAR(32) NOT NULL`,
|
||
`ALTER TABLE room_follows MODIFY COLUMN app_code VARCHAR(32) NOT NULL`,
|
||
`ALTER TABLE room_region_pins MODIFY COLUMN app_code VARCHAR(32) NOT NULL`,
|
||
`ALTER TABLE room_seat_configs MODIFY COLUMN app_code VARCHAR(32) NOT NULL`,
|
||
}
|
||
|
||
for _, statement := range statements {
|
||
// 每条 DDL 独立执行，失败时返回具体 SQL 错误给启动流程。
|
||
if _, err := r.db.ExecContext(ctx, statement); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
if err := r.ensureRoomListLockSchema(ctx); err != nil {
|
||
return err
|
||
}
|
||
if err := r.ensureRoomPasswordSchema(ctx); err != nil {
|
||
return err
|
||
}
|
||
if err := r.ensureRoomAdminCloseSchema(ctx); err != nil {
|
||
return err
|
||
}
|
||
|
||
return r.ensureOutboxRetrySchema(ctx)
|
||
}
|
||
|
||
func (r *Repository) ensureRoomPasswordSchema(ctx context.Context) error {
|
||
hasColumn, err := r.columnExists(ctx, "rooms", "room_password_hash")
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !hasColumn {
|
||
// rooms 是最新快照恢复密码哈希的持久来源；列表只使用 locked 布尔投影。
|
||
if _, err := r.db.ExecContext(ctx, `ALTER TABLE rooms ADD COLUMN room_password_hash VARCHAR(128) NOT NULL DEFAULT '' AFTER status`); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *Repository) ensureRoomListLockSchema(ctx context.Context) error {
|
||
hasColumn, err := r.columnExists(ctx, "room_list_entries", "locked")
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !hasColumn {
|
||
// locked 是列表/搜索需要的纯投影字段，历史行默认 false，后续房间状态变更会刷新为真实值。
|
||
if _, err := r.db.ExecContext(ctx, `ALTER TABLE room_list_entries ADD COLUMN locked BOOLEAN NOT NULL DEFAULT FALSE AFTER status`); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *Repository) ensureRoomAdminCloseSchema(ctx context.Context) error {
|
||
columns := []struct {
|
||
table string
|
||
column string
|
||
statement string
|
||
}{
|
||
{"rooms", "close_reason", `ALTER TABLE rooms ADD COLUMN close_reason VARCHAR(512) NOT NULL DEFAULT '' AFTER room_password_hash`},
|
||
{"rooms", "closed_by_admin_id", `ALTER TABLE rooms ADD COLUMN closed_by_admin_id BIGINT UNSIGNED NOT NULL DEFAULT 0 AFTER close_reason`},
|
||
{"rooms", "closed_by_admin_name", `ALTER TABLE rooms ADD COLUMN closed_by_admin_name VARCHAR(128) NOT NULL DEFAULT '' AFTER closed_by_admin_id`},
|
||
{"rooms", "closed_at_ms", `ALTER TABLE rooms ADD COLUMN closed_at_ms BIGINT NOT NULL DEFAULT 0 AFTER closed_by_admin_name`},
|
||
{"room_list_entries", "close_reason", `ALTER TABLE room_list_entries ADD COLUMN close_reason VARCHAR(512) NOT NULL DEFAULT '' AFTER locked`},
|
||
{"room_list_entries", "closed_by_admin_id", `ALTER TABLE room_list_entries ADD COLUMN closed_by_admin_id BIGINT UNSIGNED NOT NULL DEFAULT 0 AFTER close_reason`},
|
||
{"room_list_entries", "closed_by_admin_name", `ALTER TABLE room_list_entries ADD COLUMN closed_by_admin_name VARCHAR(128) NOT NULL DEFAULT '' AFTER closed_by_admin_id`},
|
||
{"room_list_entries", "closed_at_ms", `ALTER TABLE room_list_entries ADD COLUMN closed_at_ms BIGINT NOT NULL DEFAULT 0 AFTER closed_by_admin_name`},
|
||
}
|
||
for _, item := range columns {
|
||
hasColumn, err := r.columnExists(ctx, item.table, item.column)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !hasColumn {
|
||
if _, err := r.db.ExecContext(ctx, item.statement); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *Repository) ensureOutboxRetrySchema(ctx context.Context) error {
|
||
hasColumn, err := r.columnExists(ctx, "room_outbox", "next_retry_at_ms")
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !hasColumn {
|
||
if _, err := r.db.ExecContext(ctx, `ALTER TABLE room_outbox ADD COLUMN next_retry_at_ms BIGINT NOT NULL DEFAULT 0 AFTER retry_count`); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
hasIndex, err := r.indexExists(ctx, "room_outbox", "idx_room_outbox_retry")
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !hasIndex {
|
||
if _, err := r.db.ExecContext(ctx, `ALTER TABLE room_outbox ADD INDEX idx_room_outbox_retry (app_code, status, next_retry_at_ms, created_at_ms)`); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *Repository) columnExists(ctx context.Context, tableName string, columnName string) (bool, error) {
|
||
var exists int
|
||
err := r.db.QueryRowContext(ctx, `
|
||
SELECT COUNT(*)
|
||
FROM information_schema.COLUMNS
|
||
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?
|
||
`, tableName, columnName).Scan(&exists)
|
||
return exists > 0, err
|
||
}
|
||
|
||
func (r *Repository) indexExists(ctx context.Context, tableName string, indexName string) (bool, error) {
|
||
var exists int
|
||
err := r.db.QueryRowContext(ctx, `
|
||
SELECT COUNT(*)
|
||
FROM information_schema.STATISTICS
|
||
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND INDEX_NAME = ?
|
||
`, tableName, indexName).Scan(&exists)
|
||
return exists > 0, err
|
||
}
|
||
|
||
// SaveRoomMeta 插入房间元数据。
|
||
func (r *Repository) SaveRoomMeta(ctx context.Context, meta roomservice.RoomMeta) error {
|
||
// rooms 表只保存恢复初始状态需要的元数据，实时态以 snapshot/command log 为准。
|
||
appCode := normalizedRecordAppCode(ctx, meta.AppCode)
|
||
nowMS := time.Now().UTC().UnixMilli()
|
||
_, err := r.db.ExecContext(ctx,
|
||
`INSERT INTO rooms (app_code, room_id, room_short_id, owner_user_id, seat_count, mode, status, room_password_hash, visible_region_id, created_at_ms, updated_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||
appCode,
|
||
meta.RoomID,
|
||
meta.RoomShortID,
|
||
meta.OwnerUserID,
|
||
meta.SeatCount,
|
||
meta.Mode,
|
||
meta.Status,
|
||
meta.RoomPasswordHash,
|
||
meta.VisibleRegionID,
|
||
nowMS,
|
||
nowMS,
|
||
)
|
||
|
||
return mapRoomMetaDuplicateError(err)
|
||
}
|
||
|
||
// GetRoomMeta 查询房间元数据。
|
||
func (r *Repository) GetRoomMeta(ctx context.Context, roomID string) (roomservice.RoomMeta, bool, error) {
|
||
// meta 是无内存 Cell 恢复的入口；不存在时上层返回 room not found。
|
||
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_id = ?`,
|
||
appcode.FromContext(ctx),
|
||
roomID,
|
||
)
|
||
|
||
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) {
|
||
// 未创建房间不是数据库错误，用 exists=false 表达。
|
||
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 前置失败。
|
||
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 owner_user_id = ?
|
||
LIMIT 1`,
|
||
appcode.FromContext(ctx),
|
||
ownerUserID,
|
||
)
|
||
|
||
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
|
||
}
|
||
|
||
// GetCommand 读取已提交命令，用于幂等重试和 command_id 冲突判定。
|
||
func (r *Repository) GetCommand(ctx context.Context, roomID string, commandID string) (roomservice.CommandRecord, bool, error) {
|
||
// 幂等检查只看 command log，只有成功提交过的命令才算 seen。
|
||
row := r.db.QueryRowContext(ctx,
|
||
`SELECT app_code, room_id, room_version, command_id, command_type, owner_node_id, lease_token, payload, replayable, created_at_ms
|
||
FROM room_command_log
|
||
WHERE app_code = ? AND room_id = ? AND command_id = ?
|
||
LIMIT 1`,
|
||
appcode.FromContext(ctx),
|
||
roomID,
|
||
commandID,
|
||
)
|
||
|
||
var record roomservice.CommandRecord
|
||
if err := row.Scan(&record.AppCode, &record.RoomID, &record.RoomVersion, &record.CommandID, &record.CommandType, &record.OwnerNodeID, &record.LeaseToken, &record.Payload, &record.Replayable, &record.CreatedAtMS); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
// 没有命令日志表示本命令尚未成功提交。
|
||
return roomservice.CommandRecord{}, false, nil
|
||
}
|
||
|
||
return roomservice.CommandRecord{}, false, err
|
||
}
|
||
|
||
return record, true, nil
|
||
}
|
||
|
||
// SaveCommand 追加一条成功命令日志。
|
||
func (r *Repository) SaveCommand(ctx context.Context, record roomservice.CommandRecord) error {
|
||
// ON DUPLICATE KEY 保证重复命令写入不会破坏已提交记录。
|
||
appCode := normalizedRecordAppCode(ctx, record.AppCode)
|
||
_, err := r.db.ExecContext(ctx,
|
||
`INSERT INTO room_command_log (app_code, room_id, room_version, command_id, command_type, owner_node_id, lease_token, payload, replayable, created_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE command_id = command_id`,
|
||
appCode,
|
||
record.RoomID,
|
||
record.RoomVersion,
|
||
record.CommandID,
|
||
record.CommandType,
|
||
record.OwnerNodeID,
|
||
record.LeaseToken,
|
||
record.Payload,
|
||
record.Replayable,
|
||
record.CreatedAtMS,
|
||
)
|
||
|
||
return err
|
||
}
|
||
|
||
// SaveMutation 在同一事务中提交命令日志、outbox 和可选房间状态。
|
||
func (r *Repository) SaveMutation(ctx context.Context, commit roomservice.MutationCommit) error {
|
||
appCode := normalizedRecordAppCode(ctx, commit.Command.AppCode)
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
if _, err := tx.ExecContext(ctx,
|
||
`INSERT INTO room_command_log (app_code, room_id, room_version, command_id, command_type, owner_node_id, lease_token, payload, replayable, created_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE command_id = command_id`,
|
||
appCode,
|
||
commit.Command.RoomID,
|
||
commit.Command.RoomVersion,
|
||
commit.Command.CommandID,
|
||
commit.Command.CommandType,
|
||
commit.Command.OwnerNodeID,
|
||
commit.Command.LeaseToken,
|
||
commit.Command.Payload,
|
||
commit.Command.Replayable,
|
||
commit.Command.CreatedAtMS,
|
||
); err != nil {
|
||
_ = tx.Rollback()
|
||
return err
|
||
}
|
||
|
||
for _, record := range commit.OutboxRecords {
|
||
record.AppCode = normalizedRecordAppCode(ctx, record.AppCode)
|
||
if record.Envelope != nil {
|
||
record.Envelope.AppCode = record.AppCode
|
||
}
|
||
envelopeBytes, err := proto.Marshal(record.Envelope)
|
||
if err != nil {
|
||
_ = tx.Rollback()
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx,
|
||
`INSERT INTO room_outbox (app_code, event_id, event_type, room_id, status, envelope, created_at_ms, retry_count, next_retry_at_ms, last_error, updated_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE event_id = event_id`,
|
||
record.AppCode,
|
||
record.EventID,
|
||
record.EventType,
|
||
record.RoomID,
|
||
record.Status,
|
||
envelopeBytes,
|
||
record.CreatedAtMS,
|
||
record.RetryCount,
|
||
record.NextRetryAtMS,
|
||
nullableString(record.LastError),
|
||
record.CreatedAtMS,
|
||
); err != nil {
|
||
_ = tx.Rollback()
|
||
return err
|
||
}
|
||
}
|
||
|
||
if strings.TrimSpace(commit.RoomStatus) != "" {
|
||
if _, err := tx.ExecContext(ctx,
|
||
`UPDATE rooms
|
||
SET status = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND room_id = ?`,
|
||
commit.RoomStatus,
|
||
commit.Command.CreatedAtMS,
|
||
appCode,
|
||
commit.Command.RoomID,
|
||
); err != nil {
|
||
_ = tx.Rollback()
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx,
|
||
`UPDATE room_list_entries
|
||
SET status = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND room_id = ?`,
|
||
commit.RoomStatus,
|
||
commit.Command.CreatedAtMS,
|
||
appCode,
|
||
commit.Command.RoomID,
|
||
); err != nil {
|
||
_ = tx.Rollback()
|
||
return err
|
||
}
|
||
}
|
||
if commit.RoomSeatCount != nil {
|
||
if _, err := tx.ExecContext(ctx,
|
||
`UPDATE rooms
|
||
SET seat_count = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND room_id = ?`,
|
||
*commit.RoomSeatCount,
|
||
commit.Command.CreatedAtMS,
|
||
appCode,
|
||
commit.Command.RoomID,
|
||
); err != nil {
|
||
_ = tx.Rollback()
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx,
|
||
`UPDATE room_list_entries
|
||
SET seat_count = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND room_id = ?`,
|
||
*commit.RoomSeatCount,
|
||
commit.Command.CreatedAtMS,
|
||
appCode,
|
||
commit.Command.RoomID,
|
||
); err != nil {
|
||
_ = tx.Rollback()
|
||
return err
|
||
}
|
||
}
|
||
if commit.RoomPasswordHash != nil {
|
||
locked := strings.TrimSpace(*commit.RoomPasswordHash) != ""
|
||
if _, err := tx.ExecContext(ctx,
|
||
`UPDATE rooms
|
||
SET room_password_hash = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND room_id = ?`,
|
||
*commit.RoomPasswordHash,
|
||
commit.Command.CreatedAtMS,
|
||
appCode,
|
||
commit.Command.RoomID,
|
||
); err != nil {
|
||
_ = tx.Rollback()
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx,
|
||
`UPDATE room_list_entries
|
||
SET locked = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND room_id = ?`,
|
||
locked,
|
||
commit.Command.CreatedAtMS,
|
||
appCode,
|
||
commit.Command.RoomID,
|
||
); err != nil {
|
||
_ = tx.Rollback()
|
||
return err
|
||
}
|
||
}
|
||
|
||
return tx.Commit()
|
||
}
|
||
|
||
// GetRoomSeatConfig 读取当前 App 的房间座位数配置。
|
||
func (r *Repository) GetRoomSeatConfig(ctx context.Context) (roomservice.RoomSeatConfig, bool, error) {
|
||
row := r.db.QueryRowContext(ctx,
|
||
`SELECT app_code, allowed_seat_counts, default_seat_count, updated_at_ms
|
||
FROM room_seat_configs
|
||
WHERE app_code = ?`,
|
||
appcode.FromContext(ctx),
|
||
)
|
||
|
||
var config roomservice.RoomSeatConfig
|
||
var rawAllowed string
|
||
if err := row.Scan(&config.AppCode, &rawAllowed, &config.DefaultSeatCount, &config.UpdatedAtMS); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return roomservice.RoomSeatConfig{}, false, nil
|
||
}
|
||
return roomservice.RoomSeatConfig{}, false, err
|
||
}
|
||
if err := json.Unmarshal([]byte(rawAllowed), &config.AllowedSeatCounts); err != nil {
|
||
return roomservice.RoomSeatConfig{}, false, err
|
||
}
|
||
|
||
return config, true, nil
|
||
}
|
||
|
||
// UpsertRoomSeatConfig 写入当前 App 的房间座位数配置。
|
||
func (r *Repository) UpsertRoomSeatConfig(ctx context.Context, config roomservice.RoomSeatConfig) error {
|
||
appCode := normalizedRecordAppCode(ctx, config.AppCode)
|
||
rawAllowed, err := json.Marshal(config.AllowedSeatCounts)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
nowMS := config.UpdatedAtMS
|
||
if nowMS <= 0 {
|
||
nowMS = time.Now().UTC().UnixMilli()
|
||
}
|
||
_, err = r.db.ExecContext(ctx,
|
||
`INSERT INTO room_seat_configs (app_code, allowed_seat_counts, default_seat_count, created_at_ms, updated_at_ms)
|
||
VALUES (?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
allowed_seat_counts = VALUES(allowed_seat_counts),
|
||
default_seat_count = VALUES(default_seat_count),
|
||
updated_at_ms = VALUES(updated_at_ms)`,
|
||
appCode,
|
||
string(rawAllowed),
|
||
config.DefaultSeatCount,
|
||
nowMS,
|
||
nowMS,
|
||
)
|
||
return err
|
||
}
|
||
|
||
// ListCommandsAfter 返回快照版本之后的可回放命令日志。
|
||
func (r *Repository) ListCommandsAfter(ctx context.Context, roomID string, roomVersion int64) ([]roomservice.CommandRecord, error) {
|
||
// 恢复必须按 room_version 再按自增 id 排序，确保同版本异常数据也有稳定顺序。
|
||
rows, err := r.db.QueryContext(ctx,
|
||
`SELECT app_code, room_id, room_version, command_id, command_type, owner_node_id, lease_token, payload, replayable, created_at_ms
|
||
FROM room_command_log
|
||
WHERE app_code = ? AND room_id = ? AND room_version > ?
|
||
ORDER BY room_version ASC, id ASC`,
|
||
appcode.FromContext(ctx),
|
||
roomID,
|
||
roomVersion,
|
||
)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
records := make([]roomservice.CommandRecord, 0)
|
||
for rows.Next() {
|
||
// payload 保持原始字节，领域层根据 command_type 反序列化。
|
||
var record roomservice.CommandRecord
|
||
if err := rows.Scan(&record.AppCode, &record.RoomID, &record.RoomVersion, &record.CommandID, &record.CommandType, &record.OwnerNodeID, &record.LeaseToken, &record.Payload, &record.Replayable, &record.CreatedAtMS); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
records = append(records, record)
|
||
}
|
||
|
||
if err := rows.Err(); err != nil {
|
||
// rows.Err 捕获迭代过程中发生的底层连接或扫描错误。
|
||
return nil, err
|
||
}
|
||
|
||
return records, nil
|
||
}
|
||
|
||
// SaveSnapshot 保存房间最新快照。
|
||
func (r *Repository) SaveSnapshot(ctx context.Context, snapshot roomservice.SnapshotRecord) error {
|
||
// 快照按 room_id 保留最新一条，版本更旧的写入不会覆盖较新的恢复来源。
|
||
appCode := normalizedRecordAppCode(ctx, snapshot.AppCode)
|
||
_, err := r.db.ExecContext(ctx,
|
||
`INSERT INTO room_snapshots (app_code, room_id, room_version, payload, created_at_ms, updated_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
room_version = IF(VALUES(room_version) >= room_version, VALUES(room_version), room_version),
|
||
payload = IF(VALUES(room_version) >= room_version, VALUES(payload), payload),
|
||
created_at_ms = IF(VALUES(room_version) >= room_version, VALUES(created_at_ms), created_at_ms),
|
||
updated_at_ms = IF(VALUES(room_version) >= room_version, VALUES(updated_at_ms), updated_at_ms)`,
|
||
appCode,
|
||
snapshot.RoomID,
|
||
snapshot.RoomVersion,
|
||
snapshot.Payload,
|
||
snapshot.CreatedAtMS,
|
||
snapshot.CreatedAtMS,
|
||
)
|
||
|
||
return err
|
||
}
|
||
|
||
// GetLatestSnapshot 读取房间最新快照。
|
||
func (r *Repository) GetLatestSnapshot(ctx context.Context, roomID string) (roomservice.SnapshotRecord, bool, error) {
|
||
// room_snapshots 主键是 room_id，因此查询结果最多一条。
|
||
row := r.db.QueryRowContext(ctx,
|
||
`SELECT app_code, room_id, room_version, payload, created_at_ms
|
||
FROM room_snapshots
|
||
WHERE app_code = ? AND room_id = ?`,
|
||
appcode.FromContext(ctx),
|
||
roomID,
|
||
)
|
||
|
||
var snapshot roomservice.SnapshotRecord
|
||
if err := row.Scan(&snapshot.AppCode, &snapshot.RoomID, &snapshot.RoomVersion, &snapshot.Payload, &snapshot.CreatedAtMS); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
// 没有快照时上层会从 RoomMeta 和 command log 从头恢复。
|
||
return roomservice.SnapshotRecord{}, false, nil
|
||
}
|
||
|
||
return roomservice.SnapshotRecord{}, false, err
|
||
}
|
||
|
||
return snapshot, true, nil
|
||
}
|
||
|
||
// SaveOutbox 追加待投递 outbox 事件。
|
||
func (r *Repository) SaveOutbox(ctx context.Context, records []outbox.Record) error {
|
||
if len(records) == 0 {
|
||
// 没有事件时保持无副作用，方便领域层统一调用。
|
||
return nil
|
||
}
|
||
|
||
// 多条事件用一个事务写入，保证同一命令产生的事件不会部分入库。
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
for _, record := range records {
|
||
record.AppCode = normalizedRecordAppCode(ctx, record.AppCode)
|
||
if record.Envelope != nil {
|
||
record.Envelope.AppCode = record.AppCode
|
||
}
|
||
// envelope 是 protobuf 信封，存储层不解析具体事件 body。
|
||
envelopeBytes, err := proto.Marshal(record.Envelope)
|
||
if err != nil {
|
||
_ = tx.Rollback()
|
||
return err
|
||
}
|
||
|
||
if _, err := tx.ExecContext(ctx,
|
||
`INSERT INTO room_outbox (app_code, event_id, event_type, room_id, status, envelope, created_at_ms, retry_count, next_retry_at_ms, last_error, updated_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE event_id = event_id`,
|
||
record.AppCode,
|
||
record.EventID,
|
||
record.EventType,
|
||
record.RoomID,
|
||
record.Status,
|
||
envelopeBytes,
|
||
record.CreatedAtMS,
|
||
record.RetryCount,
|
||
record.NextRetryAtMS,
|
||
nullableString(record.LastError),
|
||
record.CreatedAtMS,
|
||
); err != nil {
|
||
// 任一事件失败就回滚整批，避免 GiftSent/HeatChanged/RankChanged 部分缺失。
|
||
_ = tx.Rollback()
|
||
return err
|
||
}
|
||
}
|
||
|
||
// commit 成功后事件才对 outbox worker 可见。
|
||
return tx.Commit()
|
||
}
|
||
|
||
// ListPendingOutbox 返回待投递事件。
|
||
func (r *Repository) ListPendingOutbox(ctx context.Context, limit int) ([]outbox.Record, error) {
|
||
if limit <= 0 {
|
||
// 默认批量上限避免调用方传 0 导致全表扫描。
|
||
limit = 100
|
||
}
|
||
|
||
// 按创建时间顺序扫描 pending，尽量保持事件投递顺序。
|
||
nowMS := time.Now().UTC().UnixMilli()
|
||
rows, err := r.db.QueryContext(ctx,
|
||
`SELECT app_code, event_id, event_type, room_id, status, worker_id, lock_until_ms, envelope, created_at_ms, retry_count, next_retry_at_ms, COALESCE(last_error, '')
|
||
FROM room_outbox
|
||
WHERE app_code = ? AND status IN (?, ?)
|
||
AND (status = ? OR next_retry_at_ms <= ?)
|
||
ORDER BY created_at_ms ASC
|
||
LIMIT ?`,
|
||
appcode.FromContext(ctx),
|
||
outbox.StatusPending,
|
||
outbox.StatusRetryable,
|
||
outbox.StatusPending,
|
||
nowMS,
|
||
limit,
|
||
)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
records := make([]outbox.Record, 0, limit)
|
||
for rows.Next() {
|
||
// 每条 outbox 记录恢复 envelope，发布器只消费 protobuf 信封。
|
||
var envelopeBytes []byte
|
||
var record outbox.Record
|
||
var lockUntil sql.Null[int64]
|
||
if err := rows.Scan(&record.AppCode, &record.EventID, &record.EventType, &record.RoomID, &record.Status, &record.WorkerID, &lockUntil, &envelopeBytes, &record.CreatedAtMS, &record.RetryCount, &record.NextRetryAtMS, &record.LastError); err != nil {
|
||
return nil, err
|
||
}
|
||
if lockUntil.Valid {
|
||
record.LockUntilMS = lockUntil.V
|
||
}
|
||
|
||
var envelope roomeventsv1.EventEnvelope
|
||
if err := proto.Unmarshal(envelopeBytes, &envelope); err != nil {
|
||
// envelope 损坏说明 outbox 数据不可投递，需要暴露错误而不是跳过。
|
||
return nil, err
|
||
}
|
||
|
||
record.Envelope = &envelope
|
||
record.Envelope.AppCode = normalizedRecordAppCode(ctx, record.AppCode)
|
||
records = append(records, record)
|
||
}
|
||
|
||
if err := rows.Err(); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
return records, nil
|
||
}
|
||
|
||
// ClaimPendingOutbox 抢占一批待投递事件，避免多 worker 同时投递同一批。
|
||
func (r *Repository) ClaimPendingOutbox(ctx context.Context, workerID string, limit int, lockUntilMS int64) ([]outbox.Record, error) {
|
||
if limit <= 0 {
|
||
limit = 100
|
||
}
|
||
if strings.TrimSpace(workerID) == "" {
|
||
workerID = "room-outbox-worker"
|
||
}
|
||
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
nowMS := time.Now().UTC().UnixMilli()
|
||
rows, err := tx.QueryContext(ctx,
|
||
`SELECT app_code, event_id, event_type, room_id, status, worker_id, lock_until_ms, envelope, created_at_ms, retry_count, next_retry_at_ms, COALESCE(last_error, '')
|
||
FROM room_outbox
|
||
WHERE app_code = ?
|
||
AND (
|
||
status = ?
|
||
OR (status = ? AND next_retry_at_ms <= ?)
|
||
OR (status = ? AND lock_until_ms IS NOT NULL AND lock_until_ms <= ?)
|
||
)
|
||
ORDER BY created_at_ms ASC
|
||
LIMIT ?
|
||
FOR UPDATE SKIP LOCKED`,
|
||
appcode.FromContext(ctx),
|
||
outbox.StatusPending,
|
||
outbox.StatusRetryable,
|
||
nowMS,
|
||
outbox.StatusDelivering,
|
||
nowMS,
|
||
limit,
|
||
)
|
||
if err != nil {
|
||
_ = tx.Rollback()
|
||
return nil, err
|
||
}
|
||
|
||
records := make([]outbox.Record, 0, limit)
|
||
eventIDs := make([]string, 0, limit)
|
||
for rows.Next() {
|
||
var envelopeBytes []byte
|
||
var record outbox.Record
|
||
var lockUntilValue sql.Null[int64]
|
||
if err := rows.Scan(&record.AppCode, &record.EventID, &record.EventType, &record.RoomID, &record.Status, &record.WorkerID, &lockUntilValue, &envelopeBytes, &record.CreatedAtMS, &record.RetryCount, &record.NextRetryAtMS, &record.LastError); err != nil {
|
||
_ = rows.Close()
|
||
_ = tx.Rollback()
|
||
return nil, err
|
||
}
|
||
if lockUntilValue.Valid {
|
||
record.LockUntilMS = lockUntilValue.V
|
||
}
|
||
|
||
var envelope roomeventsv1.EventEnvelope
|
||
if err := proto.Unmarshal(envelopeBytes, &envelope); err != nil {
|
||
_ = rows.Close()
|
||
_ = tx.Rollback()
|
||
return nil, err
|
||
}
|
||
|
||
record.Envelope = &envelope
|
||
record.Envelope.AppCode = normalizedRecordAppCode(ctx, record.AppCode)
|
||
records = append(records, record)
|
||
eventIDs = append(eventIDs, record.EventID)
|
||
}
|
||
if err := rows.Close(); err != nil {
|
||
_ = tx.Rollback()
|
||
return nil, err
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
_ = tx.Rollback()
|
||
return nil, err
|
||
}
|
||
|
||
for _, eventID := range eventIDs {
|
||
if _, err := tx.ExecContext(ctx,
|
||
`UPDATE room_outbox
|
||
SET status = ?, worker_id = ?, lock_until_ms = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND event_id = ?`,
|
||
outbox.StatusDelivering,
|
||
workerID,
|
||
lockUntilMS,
|
||
nowMS,
|
||
appcode.FromContext(ctx),
|
||
eventID,
|
||
); err != nil {
|
||
_ = tx.Rollback()
|
||
return nil, err
|
||
}
|
||
}
|
||
|
||
if err := tx.Commit(); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
for index := range records {
|
||
records[index].Status = outbox.StatusDelivering
|
||
records[index].WorkerID = workerID
|
||
records[index].LockUntilMS = lockUntilMS
|
||
}
|
||
|
||
return records, nil
|
||
}
|
||
|
||
// MarkOutboxDelivered 标记事件投递成功。
|
||
func (r *Repository) MarkOutboxDelivered(ctx context.Context, eventID string) error {
|
||
// 成功后清空 last_error，后续扫描不再返回该事件。
|
||
_, err := r.db.ExecContext(ctx,
|
||
`UPDATE room_outbox
|
||
SET status = ?, worker_id = '', lock_until_ms = NULL, last_error = NULL, updated_at_ms = ?
|
||
WHERE app_code = ? AND event_id = ?`,
|
||
outbox.StatusDelivered,
|
||
time.Now().UTC().UnixMilli(),
|
||
appcode.FromContext(ctx),
|
||
eventID,
|
||
)
|
||
|
||
return err
|
||
}
|
||
|
||
// MarkOutboxRetryable 标记事件投递失败并写入下一次重试时间。
|
||
func (r *Repository) MarkOutboxRetryable(ctx context.Context, eventID string, lastErr string, nextRetryAtMS int64) error {
|
||
// 失败转入 retryable，worker 只有到 next_retry_at_ms 后才会重新抢占。
|
||
_, err := r.db.ExecContext(ctx,
|
||
`UPDATE room_outbox
|
||
SET status = ?, worker_id = '', lock_until_ms = NULL, retry_count = retry_count + 1, next_retry_at_ms = ?, last_error = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND event_id = ?`,
|
||
outbox.StatusRetryable,
|
||
nextRetryAtMS,
|
||
lastErr,
|
||
time.Now().UTC().UnixMilli(),
|
||
appcode.FromContext(ctx),
|
||
eventID,
|
||
)
|
||
|
||
return err
|
||
}
|
||
|
||
// MarkOutboxDead 把超过最大重试次数的事件转入 failed。
|
||
func (r *Repository) MarkOutboxDead(ctx context.Context, eventID string, lastErr string) error {
|
||
_, err := r.db.ExecContext(ctx,
|
||
`UPDATE room_outbox
|
||
SET status = ?, worker_id = '', lock_until_ms = NULL, last_error = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND event_id = ?`,
|
||
outbox.StatusFailed,
|
||
lastErr,
|
||
time.Now().UTC().UnixMilli(),
|
||
appcode.FromContext(ctx),
|
||
eventID,
|
||
)
|
||
|
||
return err
|
||
}
|
||
|
||
// UpsertRoomListEntry 写入房间列表读模型。
|
||
func (r *Repository) UpsertRoomListEntry(ctx context.Context, entry roomservice.RoomListEntry) error {
|
||
// created_at_ms 首次插入后保持不变，避免 Join/Leave/SendGift 等更新把 new tab 顺序洗掉。
|
||
appCode := normalizedRecordAppCode(ctx, entry.AppCode)
|
||
_, err := r.db.ExecContext(ctx,
|
||
`INSERT INTO room_list_entries (
|
||
app_code, room_id, room_short_id, visible_region_id, owner_user_id, title, cover_url, mode, status, locked,
|
||
heat, online_count, seat_count, occupied_seat_count, sort_score, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
room_short_id = VALUES(room_short_id),
|
||
visible_region_id = VALUES(visible_region_id),
|
||
owner_user_id = VALUES(owner_user_id),
|
||
title = VALUES(title),
|
||
cover_url = VALUES(cover_url),
|
||
mode = VALUES(mode),
|
||
status = VALUES(status),
|
||
locked = VALUES(locked),
|
||
heat = VALUES(heat),
|
||
online_count = VALUES(online_count),
|
||
seat_count = VALUES(seat_count),
|
||
occupied_seat_count = VALUES(occupied_seat_count),
|
||
sort_score = VALUES(sort_score),
|
||
updated_at_ms = VALUES(updated_at_ms)`,
|
||
appCode,
|
||
entry.RoomID,
|
||
entry.RoomShortID,
|
||
entry.VisibleRegionID,
|
||
entry.OwnerUserID,
|
||
entry.Title,
|
||
entry.CoverURL,
|
||
entry.Mode,
|
||
entry.Status,
|
||
entry.Locked,
|
||
entry.Heat,
|
||
entry.OnlineCount,
|
||
entry.SeatCount,
|
||
entry.OccupiedSeatCount,
|
||
entry.SortScore,
|
||
entry.CreatedAtMS,
|
||
entry.UpdatedAtMS,
|
||
)
|
||
|
||
return err
|
||
}
|
||
|
||
// ListRoomListEntries 按区域读取房间列表读模型。
|
||
func (r *Repository) ListRoomListEntries(ctx context.Context, query roomservice.RoomListQuery) ([]roomservice.RoomListEntry, error) {
|
||
if query.Limit <= 0 {
|
||
query.Limit = 20
|
||
}
|
||
query.AppCode = normalizedRecordAppCode(ctx, query.AppCode)
|
||
if query.NowMS <= 0 {
|
||
query.NowMS = time.Now().UTC().UnixMilli()
|
||
}
|
||
|
||
sqlText, args := buildRoomListQuerySQL(query)
|
||
rows, err := r.db.QueryContext(ctx, sqlText, args...)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
entries := make([]roomservice.RoomListEntry, 0, query.Limit)
|
||
for rows.Next() {
|
||
var entry roomservice.RoomListEntry
|
||
var pinned int64
|
||
if err := rows.Scan(
|
||
&entry.AppCode,
|
||
&entry.RoomID,
|
||
&entry.RoomShortID,
|
||
&entry.VisibleRegionID,
|
||
&entry.OwnerUserID,
|
||
&entry.Title,
|
||
&entry.CoverURL,
|
||
&entry.Mode,
|
||
&entry.Status,
|
||
&entry.Locked,
|
||
&entry.Heat,
|
||
&entry.OnlineCount,
|
||
&entry.SeatCount,
|
||
&entry.OccupiedSeatCount,
|
||
&entry.SortScore,
|
||
&entry.CreatedAtMS,
|
||
&entry.UpdatedAtMS,
|
||
&pinned,
|
||
&entry.PinWeight,
|
||
&entry.PinnedUntilMS,
|
||
); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
entry.IsPinned = pinned > 0
|
||
entries = append(entries, entry)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
return entries, nil
|
||
}
|
||
|
||
// UpsertRoomUserFeedEntry 写入用户房间关系流索引。
|
||
func (r *Repository) UpsertRoomUserFeedEntry(ctx context.Context, entry roomservice.RoomUserFeedEntry) error {
|
||
// feed 只保存用户到房间的索引；房间卡片字段仍以 room_list_entries 为准，避免复制不一致。
|
||
appCode := normalizedRecordAppCode(ctx, entry.AppCode)
|
||
_, err := r.db.ExecContext(ctx,
|
||
`INSERT INTO room_user_feed_entries (app_code, user_id, feed_type, room_id, created_at_ms, updated_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE updated_at_ms = VALUES(updated_at_ms)`,
|
||
appCode,
|
||
entry.UserID,
|
||
entry.FeedType,
|
||
entry.RoomID,
|
||
entry.CreatedAtMS,
|
||
entry.UpdatedAtMS,
|
||
)
|
||
|
||
return err
|
||
}
|
||
|
||
// ListRoomUserFeedEntries 按用户关系流读取房间卡片。
|
||
func (r *Repository) ListRoomUserFeedEntries(ctx context.Context, query roomservice.RoomUserFeedQuery) ([]roomservice.RoomListEntry, error) {
|
||
if query.Limit <= 0 {
|
||
query.Limit = 20
|
||
}
|
||
query.AppCode = normalizedRecordAppCode(ctx, query.AppCode)
|
||
|
||
sqlText, args := buildRoomUserFeedQuerySQL(query)
|
||
rows, err := r.db.QueryContext(ctx, sqlText, args...)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
entries := make([]roomservice.RoomListEntry, 0, query.Limit)
|
||
for rows.Next() {
|
||
var entry roomservice.RoomListEntry
|
||
if err := rows.Scan(
|
||
&entry.AppCode,
|
||
&entry.RoomID,
|
||
&entry.RoomShortID,
|
||
&entry.VisibleRegionID,
|
||
&entry.OwnerUserID,
|
||
&entry.Title,
|
||
&entry.CoverURL,
|
||
&entry.Mode,
|
||
&entry.Status,
|
||
&entry.Locked,
|
||
&entry.Heat,
|
||
&entry.OnlineCount,
|
||
&entry.SeatCount,
|
||
&entry.OccupiedSeatCount,
|
||
&entry.SortScore,
|
||
&entry.CreatedAtMS,
|
||
&entry.UpdatedAtMS,
|
||
); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
entries = append(entries, entry)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
return entries, nil
|
||
}
|
||
|
||
// FollowRoom 建立或恢复用户对房间的关注关系。
|
||
func (r *Repository) FollowRoom(ctx context.Context, userID int64, roomID string, nowMS int64) (roomservice.RoomFollowRecord, error) {
|
||
appCode := appcode.FromContext(ctx)
|
||
if nowMS <= 0 {
|
||
nowMS = time.Now().UTC().UnixMilli()
|
||
}
|
||
if _, err := r.db.ExecContext(ctx,
|
||
`INSERT INTO room_follows (app_code, user_id, room_id, status, followed_at_ms, updated_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
followed_at_ms = IF(status = ?, followed_at_ms, VALUES(followed_at_ms)),
|
||
status = VALUES(status),
|
||
updated_at_ms = VALUES(updated_at_ms)`,
|
||
appCode,
|
||
userID,
|
||
roomID,
|
||
roomFollowStatusActiveSQL,
|
||
nowMS,
|
||
nowMS,
|
||
roomFollowStatusActiveSQL,
|
||
); err != nil {
|
||
return roomservice.RoomFollowRecord{}, err
|
||
}
|
||
|
||
record, _, err := r.getRoomFollowRecord(ctx, userID, roomID)
|
||
return record, err
|
||
}
|
||
|
||
// UnfollowRoom 取消用户对房间的关注关系；没有现存关系时不写入新行。
|
||
func (r *Repository) UnfollowRoom(ctx context.Context, userID int64, roomID string, nowMS int64) error {
|
||
if nowMS <= 0 {
|
||
nowMS = time.Now().UTC().UnixMilli()
|
||
}
|
||
_, err := r.db.ExecContext(ctx,
|
||
`UPDATE room_follows
|
||
SET status = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND user_id = ? AND room_id = ? AND status = ?`,
|
||
roomFollowStatusCancelledSQL,
|
||
nowMS,
|
||
appcode.FromContext(ctx),
|
||
userID,
|
||
roomID,
|
||
roomFollowStatusActiveSQL,
|
||
)
|
||
return err
|
||
}
|
||
|
||
// IsRoomFollowed 查询用户是否正在关注指定房间。
|
||
func (r *Repository) IsRoomFollowed(ctx context.Context, userID int64, roomID string) (bool, error) {
|
||
record, exists, err := r.getRoomFollowRecord(ctx, userID, roomID)
|
||
if err != nil || !exists {
|
||
return false, err
|
||
}
|
||
return record.Status == roomFollowStatusActiveSQL, nil
|
||
}
|
||
|
||
func (r *Repository) getRoomFollowRecord(ctx context.Context, userID int64, roomID string) (roomservice.RoomFollowRecord, bool, error) {
|
||
row := r.db.QueryRowContext(ctx,
|
||
`SELECT app_code, user_id, room_id, status, followed_at_ms, updated_at_ms
|
||
FROM room_follows
|
||
WHERE app_code = ? AND user_id = ? AND room_id = ?
|
||
LIMIT 1`,
|
||
appcode.FromContext(ctx),
|
||
userID,
|
||
roomID,
|
||
)
|
||
|
||
var record roomservice.RoomFollowRecord
|
||
if err := row.Scan(&record.AppCode, &record.UserID, &record.RoomID, &record.Status, &record.FollowedAtMS, &record.UpdatedAtMS); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return roomservice.RoomFollowRecord{}, false, nil
|
||
}
|
||
return roomservice.RoomFollowRecord{}, false, err
|
||
}
|
||
return record, true, nil
|
||
}
|
||
|
||
// ListRoomFollowEntries 读取当前用户关注的 active 房间卡片。
|
||
func (r *Repository) ListRoomFollowEntries(ctx context.Context, query roomservice.RoomFollowQuery) ([]roomservice.RoomListEntry, error) {
|
||
if query.Limit <= 0 {
|
||
query.Limit = 20
|
||
}
|
||
query.AppCode = normalizedRecordAppCode(ctx, query.AppCode)
|
||
|
||
sqlText, args := buildRoomFollowQuerySQL(query)
|
||
rows, err := r.db.QueryContext(ctx, sqlText, args...)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
entries := make([]roomservice.RoomListEntry, 0, query.Limit)
|
||
for rows.Next() {
|
||
var entry roomservice.RoomListEntry
|
||
if err := rows.Scan(
|
||
&entry.AppCode,
|
||
&entry.RoomID,
|
||
&entry.RoomShortID,
|
||
&entry.VisibleRegionID,
|
||
&entry.OwnerUserID,
|
||
&entry.Title,
|
||
&entry.CoverURL,
|
||
&entry.Mode,
|
||
&entry.Status,
|
||
&entry.Locked,
|
||
&entry.Heat,
|
||
&entry.OnlineCount,
|
||
&entry.SeatCount,
|
||
&entry.OccupiedSeatCount,
|
||
&entry.SortScore,
|
||
&entry.CreatedAtMS,
|
||
&entry.UpdatedAtMS,
|
||
); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
entries = append(entries, entry)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
return entries, nil
|
||
}
|
||
|
||
// ListRoomRelatedFeedEntries 根据 user-service 给出的好友/关注用户查询 active 房间卡片。
|
||
func (r *Repository) ListRoomRelatedFeedEntries(ctx context.Context, query roomservice.RoomRelatedFeedQuery) ([]roomservice.RoomListEntry, error) {
|
||
if query.Limit <= 0 {
|
||
query.Limit = 20
|
||
}
|
||
query.AppCode = normalizedRecordAppCode(ctx, query.AppCode)
|
||
relations := normalizeRoomFeedRelations(query.RelatedUsers)
|
||
if len(relations) == 0 {
|
||
return []roomservice.RoomListEntry{}, nil
|
||
}
|
||
|
||
sqlText, args := buildRoomRelatedFeedQuerySQL(query, relations)
|
||
rows, err := r.db.QueryContext(ctx, sqlText, args...)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
entries := make([]roomservice.RoomListEntry, 0, query.Limit)
|
||
for rows.Next() {
|
||
var entry roomservice.RoomListEntry
|
||
if err := rows.Scan(
|
||
&entry.AppCode,
|
||
&entry.RoomID,
|
||
&entry.RoomShortID,
|
||
&entry.VisibleRegionID,
|
||
&entry.OwnerUserID,
|
||
&entry.Title,
|
||
&entry.CoverURL,
|
||
&entry.Mode,
|
||
&entry.Status,
|
||
&entry.Locked,
|
||
&entry.Heat,
|
||
&entry.OnlineCount,
|
||
&entry.SeatCount,
|
||
&entry.OccupiedSeatCount,
|
||
&entry.SortScore,
|
||
&entry.CreatedAtMS,
|
||
&entry.UpdatedAtMS,
|
||
); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
entry.FeedSubjectUserID, entry.UpdatedAtMS = relatedFeedSortKey(entry, relations)
|
||
if entry.FeedSubjectUserID == 0 || !roomRelatedFeedAfterCursor(entry, query.CursorUpdatedAtMS, query.CursorSubjectUserID, query.CursorRoomID) {
|
||
continue
|
||
}
|
||
entries = append(entries, entry)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
sort.SliceStable(entries, func(i, j int) bool {
|
||
if entries[i].UpdatedAtMS != entries[j].UpdatedAtMS {
|
||
return entries[i].UpdatedAtMS > entries[j].UpdatedAtMS
|
||
}
|
||
if entries[i].FeedSubjectUserID != entries[j].FeedSubjectUserID {
|
||
return entries[i].FeedSubjectUserID > entries[j].FeedSubjectUserID
|
||
}
|
||
return entries[i].RoomID < entries[j].RoomID
|
||
})
|
||
if len(entries) > query.Limit {
|
||
entries = entries[:query.Limit]
|
||
}
|
||
return entries, nil
|
||
}
|
||
|
||
const roomListSelectColumns = `
|
||
SELECT app_code, room_id, room_short_id, visible_region_id, owner_user_id, title, cover_url, mode, status, locked,
|
||
heat, online_count, seat_count, occupied_seat_count, sort_score, created_at_ms, updated_at_ms
|
||
FROM room_list_entries`
|
||
|
||
const roomUserFeedSelectColumns = `
|
||
SELECT r.app_code, r.room_id, r.room_short_id, r.visible_region_id, r.owner_user_id, r.title, r.cover_url, r.mode, r.status, r.locked,
|
||
r.heat, r.online_count, r.seat_count, r.occupied_seat_count, r.sort_score, r.created_at_ms, f.updated_at_ms
|
||
FROM room_user_feed_entries f
|
||
JOIN room_list_entries r ON r.app_code = f.app_code AND r.room_id = f.room_id`
|
||
|
||
const roomFollowSelectColumns = `
|
||
SELECT r.app_code, r.room_id, r.room_short_id, r.visible_region_id, r.owner_user_id, r.title, r.cover_url, r.mode, r.status, r.locked,
|
||
r.heat, r.online_count, r.seat_count, r.occupied_seat_count, r.sort_score, r.created_at_ms, f.followed_at_ms
|
||
FROM room_follows f
|
||
JOIN room_list_entries r ON r.app_code = f.app_code AND r.room_id = f.room_id`
|
||
|
||
func buildRoomListQuerySQL(query roomservice.RoomListQuery) (string, []any) {
|
||
pinnedExpr := "CASE WHEN p.room_id IS NULL THEN 0 ELSE 1 END"
|
||
pinWeightExpr := "COALESCE(p.weight, 0)"
|
||
pinExpiresExpr := "COALESCE(p.expires_at_ms, 0)"
|
||
selectSQL := `
|
||
SELECT r.app_code, r.room_id, r.room_short_id, r.visible_region_id, r.owner_user_id, r.title, r.cover_url, r.mode, r.status, r.locked,
|
||
r.heat, r.online_count, r.seat_count, r.occupied_seat_count, r.sort_score, r.created_at_ms, r.updated_at_ms,
|
||
` + pinnedExpr + ` AS is_pinned, ` + pinWeightExpr + ` AS pin_weight, ` + pinExpiresExpr + ` AS pinned_until_ms
|
||
FROM room_list_entries r
|
||
LEFT JOIN room_region_pins p
|
||
ON p.app_code = r.app_code
|
||
AND p.visible_region_id = r.visible_region_id
|
||
AND p.room_id = r.room_id
|
||
AND p.status = ?
|
||
AND p.expires_at_ms > ?`
|
||
where := []string{"r.app_code = ?", "r.visible_region_id = ?", "r.status = ?"}
|
||
args := []any{"active", query.NowMS, appcode.Normalize(query.AppCode), query.VisibleRegionID, "active"}
|
||
if query.OwnerUserID > 0 {
|
||
where = append(where, "r.owner_user_id = ?")
|
||
args = append(args, query.OwnerUserID)
|
||
}
|
||
if strings.TrimSpace(query.Query) != "" {
|
||
where = append(where, "(r.room_id LIKE ? ESCAPE '\\\\' OR r.room_short_id LIKE ? ESCAPE '\\\\' OR r.title LIKE ? ESCAPE '\\\\')")
|
||
like := "%" + escapeRoomListLike(query.Query) + "%"
|
||
args = append(args, like, like, like)
|
||
}
|
||
if query.Tab == "new" {
|
||
if query.CursorRoomID != "" {
|
||
where = append(where, roomListAfterCursorSQL(pinnedExpr, pinWeightExpr, pinExpiresExpr, "r.created_at_ms"))
|
||
args = append(args, roomListAfterCursorArgs(query.CursorPinned, query.CursorPinWeight, query.CursorPinnedUntilMS, query.CursorCreatedAtMS, query.CursorRoomID)...)
|
||
}
|
||
args = append(args, query.Limit)
|
||
return selectSQL + "\n\tWHERE " + strings.Join(where, " AND ") + "\n\tORDER BY " + pinnedExpr + " DESC, " + pinWeightExpr + " DESC, " + pinExpiresExpr + " DESC, r.created_at_ms DESC, r.room_id ASC\n\tLIMIT ?", args
|
||
}
|
||
|
||
if query.CursorRoomID != "" {
|
||
where = append(where, roomListAfterCursorSQL(pinnedExpr, pinWeightExpr, pinExpiresExpr, "r.sort_score"))
|
||
args = append(args, roomListAfterCursorArgs(query.CursorPinned, query.CursorPinWeight, query.CursorPinnedUntilMS, query.CursorSortScore, query.CursorRoomID)...)
|
||
}
|
||
args = append(args, query.Limit)
|
||
return selectSQL + "\n\tWHERE " + strings.Join(where, " AND ") + "\n\tORDER BY " + pinnedExpr + " DESC, " + pinWeightExpr + " DESC, " + pinExpiresExpr + " DESC, r.sort_score DESC, r.room_id ASC\n\tLIMIT ?", args
|
||
}
|
||
|
||
func roomListAfterCursorSQL(pinnedExpr string, pinWeightExpr string, pinExpiresExpr string, sortExpr string) string {
|
||
// 置顶分区、权重、过期时间和原有排序键共同组成游标，避免区域置顶跨页时重复或跳过普通房间。
|
||
return "(" +
|
||
pinnedExpr + " < ? OR (" +
|
||
pinnedExpr + " = ? AND " + pinWeightExpr + " < ?) OR (" +
|
||
pinnedExpr + " = ? AND " + pinWeightExpr + " = ? AND " + pinExpiresExpr + " < ?) OR (" +
|
||
pinnedExpr + " = ? AND " + pinWeightExpr + " = ? AND " + pinExpiresExpr + " = ? AND " + sortExpr + " < ?) OR (" +
|
||
pinnedExpr + " = ? AND " + pinWeightExpr + " = ? AND " + pinExpiresExpr + " = ? AND " + sortExpr + " = ? AND r.room_id > ?))"
|
||
}
|
||
|
||
func roomListAfterCursorArgs(pinned bool, pinWeight int64, pinnedUntilMS int64, sortValue int64, roomID string) []any {
|
||
pinnedValue := 0
|
||
if pinned {
|
||
pinnedValue = 1
|
||
}
|
||
return []any{
|
||
pinnedValue,
|
||
pinnedValue, pinWeight,
|
||
pinnedValue, pinWeight, pinnedUntilMS,
|
||
pinnedValue, pinWeight, pinnedUntilMS, sortValue,
|
||
pinnedValue, pinWeight, pinnedUntilMS, sortValue, roomID,
|
||
}
|
||
}
|
||
|
||
func buildRoomUserFeedQuerySQL(query roomservice.RoomUserFeedQuery) (string, []any) {
|
||
where := []string{"f.app_code = ?", "f.user_id = ?", "f.feed_type = ?", "r.visible_region_id = ?", "r.status = ?"}
|
||
args := []any{appcode.Normalize(query.AppCode), query.UserID, query.FeedType, query.VisibleRegionID, "active"}
|
||
if strings.TrimSpace(query.Query) != "" {
|
||
where = append(where, "(r.room_id LIKE ? ESCAPE '\\\\' OR r.room_short_id LIKE ? ESCAPE '\\\\' OR r.title LIKE ? ESCAPE '\\\\')")
|
||
like := "%" + escapeRoomListLike(query.Query) + "%"
|
||
args = append(args, like, like, like)
|
||
}
|
||
if query.CursorRoomID != "" {
|
||
where = append(where, "(f.updated_at_ms < ? OR (f.updated_at_ms = ? AND f.room_id > ?))")
|
||
args = append(args, query.CursorUpdatedAtMS, query.CursorUpdatedAtMS, query.CursorRoomID)
|
||
}
|
||
args = append(args, query.Limit)
|
||
return roomUserFeedSelectColumns + "\n\tWHERE " + strings.Join(where, " AND ") + "\n\tORDER BY f.updated_at_ms DESC, f.room_id ASC\n\tLIMIT ?", args
|
||
}
|
||
|
||
func buildRoomFollowQuerySQL(query roomservice.RoomFollowQuery) (string, []any) {
|
||
where := []string{"f.app_code = ?", "f.user_id = ?", "f.status = ?", "r.visible_region_id = ?", "r.status = ?"}
|
||
args := []any{appcode.Normalize(query.AppCode), query.UserID, roomFollowStatusActiveSQL, query.VisibleRegionID, "active"}
|
||
if strings.TrimSpace(query.Query) != "" {
|
||
where = append(where, "(r.room_id LIKE ? ESCAPE '\\\\' OR r.room_short_id LIKE ? ESCAPE '\\\\' OR r.title LIKE ? ESCAPE '\\\\')")
|
||
like := "%" + escapeRoomListLike(query.Query) + "%"
|
||
args = append(args, like, like, like)
|
||
}
|
||
if query.CursorRoomID != "" {
|
||
where = append(where, "(f.followed_at_ms < ? OR (f.followed_at_ms = ? AND f.room_id > ?))")
|
||
args = append(args, query.CursorFollowedAtMS, query.CursorFollowedAtMS, query.CursorRoomID)
|
||
}
|
||
args = append(args, query.Limit)
|
||
return roomFollowSelectColumns + "\n\tWHERE " + strings.Join(where, " AND ") + "\n\tORDER BY f.followed_at_ms DESC, f.room_id ASC\n\tLIMIT ?", args
|
||
}
|
||
|
||
func buildRoomRelatedFeedQuerySQL(query roomservice.RoomRelatedFeedQuery, relations map[int64]int64) (string, []any) {
|
||
where := []string{"app_code = ?", "visible_region_id = ?", "status = ?"}
|
||
args := []any{appcode.Normalize(query.AppCode), query.VisibleRegionID, "active"}
|
||
|
||
placeholders := make([]string, 0, len(relations))
|
||
relatedIDs := make([]any, 0, len(relations))
|
||
for userID := range relations {
|
||
placeholders = append(placeholders, "?")
|
||
relatedIDs = append(relatedIDs, userID)
|
||
}
|
||
inClause := strings.Join(placeholders, ",")
|
||
where = append(where, "owner_user_id IN ("+inClause+")")
|
||
args = append(args, relatedIDs...)
|
||
|
||
if strings.TrimSpace(query.Query) != "" {
|
||
where = append(where, "(room_id LIKE ? ESCAPE '\\\\' OR room_short_id LIKE ? ESCAPE '\\\\' OR title LIKE ? ESCAPE '\\\\')")
|
||
like := "%" + escapeRoomListLike(query.Query) + "%"
|
||
args = append(args, like, like, like)
|
||
}
|
||
|
||
return roomListSelectColumns + "\n\tWHERE " + strings.Join(where, " AND "), args
|
||
}
|
||
|
||
func normalizeRoomFeedRelations(items []roomservice.RoomFeedRelatedUser) map[int64]int64 {
|
||
relations := make(map[int64]int64, len(items))
|
||
for _, item := range items {
|
||
if item.UserID <= 0 || item.RelationUpdatedAtMS <= 0 {
|
||
continue
|
||
}
|
||
if item.RelationUpdatedAtMS > relations[item.UserID] {
|
||
relations[item.UserID] = item.RelationUpdatedAtMS
|
||
}
|
||
}
|
||
return relations
|
||
}
|
||
|
||
func relatedFeedSortKey(entry roomservice.RoomListEntry, relations map[int64]int64) (int64, int64) {
|
||
bestUserID, bestUpdatedAtMS := int64(0), int64(0)
|
||
choose := func(userID int64) {
|
||
updatedAtMS := relations[userID]
|
||
if updatedAtMS == 0 {
|
||
return
|
||
}
|
||
if updatedAtMS > bestUpdatedAtMS || (updatedAtMS == bestUpdatedAtMS && userID > bestUserID) {
|
||
bestUserID = userID
|
||
bestUpdatedAtMS = updatedAtMS
|
||
}
|
||
}
|
||
choose(entry.OwnerUserID)
|
||
return bestUserID, bestUpdatedAtMS
|
||
}
|
||
|
||
func roomRelatedFeedAfterCursor(entry roomservice.RoomListEntry, cursorUpdatedAtMS int64, cursorSubjectUserID int64, cursorRoomID string) bool {
|
||
if cursorUpdatedAtMS <= 0 || cursorSubjectUserID <= 0 || cursorRoomID == "" {
|
||
return true
|
||
}
|
||
if entry.UpdatedAtMS != cursorUpdatedAtMS {
|
||
return entry.UpdatedAtMS < cursorUpdatedAtMS
|
||
}
|
||
if entry.FeedSubjectUserID != cursorSubjectUserID {
|
||
return entry.FeedSubjectUserID < cursorSubjectUserID
|
||
}
|
||
return entry.RoomID > cursorRoomID
|
||
}
|
||
|
||
// ProjectRoomPresence 用最新 RoomSnapshot 投影用户当前房间读模型。
|
||
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, 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),
|
||
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.LastSeenAtMS,
|
||
snapshot.UpdatedAtMS,
|
||
); err != nil {
|
||
_ = tx.Rollback()
|
||
return err
|
||
}
|
||
}
|
||
|
||
return tx.Commit()
|
||
}
|
||
|
||
// 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, 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.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
|
||
}
|
||
|
||
rows, err := r.db.QueryContext(ctx,
|
||
`SELECT app_code, user_id, room_id, role, publish_state, mic_session_id, room_version, status, last_seen_at_ms, updated_at_ms
|
||
FROM room_user_presence
|
||
WHERE app_code = ? AND room_id = ? AND status = ?
|
||
ORDER BY last_seen_at_ms DESC, user_id ASC
|
||
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.LastSeenAtMS,
|
||
&presence.UpdatedAtMS,
|
||
); 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
|
||
}
|
||
|
||
const (
|
||
roomPresenceStatusActiveSQL = "active"
|
||
roomPresenceStatusLeftSQL = "left"
|
||
roomFollowStatusActiveSQL = "active"
|
||
roomFollowStatusCancelledSQL = "cancelled"
|
||
)
|
||
|
||
func escapeRoomListLike(value string) string {
|
||
var builder strings.Builder
|
||
for _, r := range value {
|
||
if r == '\\' || r == '%' || r == '_' {
|
||
builder.WriteRune('\\')
|
||
}
|
||
builder.WriteRune(r)
|
||
}
|
||
return builder.String()
|
||
}
|
||
|
||
func nullableString(value string) sql.Null[string] {
|
||
// 空错误写 NULL，便于 SQL 查询区分没有错误和空字符串。
|
||
return sql.Null[string]{
|
||
V: value,
|
||
Valid: value != "",
|
||
}
|
||
}
|
||
|
||
func normalizedRecordAppCode(ctx context.Context, value string) string {
|
||
value = strings.TrimSpace(value)
|
||
if value == "" {
|
||
return appcode.FromContext(ctx)
|
||
}
|
||
|
||
return appcode.Normalize(value)
|
||
}
|
||
|
||
func mapRoomMetaDuplicateError(err error) error {
|
||
if err == nil {
|
||
return nil
|
||
}
|
||
|
||
var mysqlErr *mysqlerr.MySQLError
|
||
if !errors.As(err, &mysqlErr) || mysqlErr.Number != 1062 {
|
||
return err
|
||
}
|
||
|
||
message := mysqlErr.Message
|
||
if strings.Contains(message, "uk_rooms_owner_user") || strings.Contains(message, "owner_user_id") {
|
||
return xerr.New(xerr.Conflict, "owner already has room")
|
||
}
|
||
|
||
return xerr.New(xerr.Conflict, "room already exists")
|
||
}
|