479 lines
16 KiB
Go
479 lines
16 KiB
Go
// Package mysql 实现 room-service 的 MySQL 持久化边界。
|
||
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"strings"
|
||
"time"
|
||
|
||
"google.golang.org/protobuf/proto"
|
||
roomeventsv1 "hyapp.local/api/proto/events/room/v1"
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/services/room-service/internal/room/outbox"
|
||
)
|
||
|
||
// SaveOutbox 追加待投递 outbox 事件。
|
||
func (r *Repository) SaveOutbox(ctx context.Context, records []outbox.Record) error {
|
||
return r.saveOutboxInTable(ctx, roomOutboxTable, records)
|
||
}
|
||
|
||
// SaveRobotOutbox 追加机器人房间展示 outbox 事件。
|
||
func (r *Repository) SaveRobotOutbox(ctx context.Context, records []outbox.Record) error {
|
||
return r.saveOutboxInTable(ctx, roomRobotOutboxTable, records)
|
||
}
|
||
|
||
func (r *Repository) saveOutboxInTable(ctx context.Context, table string, records []outbox.Record) error {
|
||
table = roomOutboxTableName(table)
|
||
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 `+table+` (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) {
|
||
return r.listPendingOutboxFromTable(ctx, roomOutboxTable, limit)
|
||
}
|
||
|
||
// ListPendingRobotOutbox 返回机器人房间待投递展示事件。
|
||
func (r *Repository) ListPendingRobotOutbox(ctx context.Context, limit int) ([]outbox.Record, error) {
|
||
return r.listPendingOutboxFromTable(ctx, roomRobotOutboxTable, limit)
|
||
}
|
||
|
||
func (r *Repository) listPendingOutboxFromTable(ctx context.Context, table string, limit int) ([]outbox.Record, error) {
|
||
if limit <= 0 {
|
||
// 默认批量上限避免调用方传 0 导致全表扫描。
|
||
limit = 100
|
||
}
|
||
table = roomOutboxTableName(table)
|
||
pendingIndex, _ := roomOutboxIndexes(table)
|
||
|
||
// 按创建时间顺序扫描 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 `+table+` FORCE INDEX (`+pendingIndex+`)
|
||
WHERE app_code = ? AND status IN (?, ?)
|
||
AND next_retry_at_ms <= ?
|
||
ORDER BY created_at_ms ASC
|
||
LIMIT ?`,
|
||
appcode.FromContext(ctx),
|
||
outbox.StatusPending,
|
||
outbox.StatusRetryable,
|
||
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) {
|
||
return r.claimPendingOutboxFromTable(ctx, roomOutboxTable, workerID, limit, lockUntilMS)
|
||
}
|
||
|
||
// ClaimPendingRobotOutbox 抢占机器人房间展示事件，避免机器人流量占用主 room_outbox worker。
|
||
func (r *Repository) ClaimPendingRobotOutbox(ctx context.Context, workerID string, limit int, lockUntilMS int64) ([]outbox.Record, error) {
|
||
return r.claimPendingOutboxFromTable(ctx, roomRobotOutboxTable, workerID, limit, lockUntilMS)
|
||
}
|
||
|
||
func (r *Repository) claimPendingOutboxFromTable(ctx context.Context, table string, workerID string, limit int, lockUntilMS int64) ([]outbox.Record, error) {
|
||
if limit <= 0 {
|
||
limit = 100
|
||
}
|
||
if strings.TrimSpace(workerID) == "" {
|
||
workerID = "room-outbox-worker"
|
||
}
|
||
table = roomOutboxTableName(table)
|
||
pendingIndex, claimIndex := roomOutboxIndexes(table)
|
||
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
nowMS := time.Now().UTC().UnixMilli()
|
||
appCode := appcode.FromContext(ctx)
|
||
records := make([]outbox.Record, 0, limit)
|
||
// pending 新事件的 next_retry_at_ms 固定为 0，真正决定可见性的是未被抢占的 lock_until_ms；
|
||
// 因此 pending 分支必须走 claim 索引按 created_at_ms 直接取最早记录，避免在机器人 outbox 堆积时用
|
||
// next_retry_at_ms range 扫描数百万历史行再 filesort。retryable 才按 next_retry_at_ms 到期时间排序。
|
||
claimBranches := []struct {
|
||
query string
|
||
args []any
|
||
}{
|
||
{
|
||
query: `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 ` + table + ` FORCE INDEX (` + claimIndex + `)
|
||
WHERE app_code = ? AND status = ? AND lock_until_ms IS NULL AND next_retry_at_ms <= ?
|
||
ORDER BY created_at_ms ASC, event_id ASC
|
||
LIMIT ? FOR UPDATE SKIP LOCKED`,
|
||
args: []any{appCode, outbox.StatusPending, nowMS},
|
||
},
|
||
{
|
||
query: `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 ` + table + ` FORCE INDEX (` + pendingIndex + `)
|
||
WHERE app_code = ? AND status = ? AND next_retry_at_ms <= ? AND lock_until_ms IS NULL
|
||
ORDER BY next_retry_at_ms ASC, created_at_ms ASC, event_id ASC
|
||
LIMIT ? FOR UPDATE SKIP LOCKED`,
|
||
args: []any{appCode, outbox.StatusRetryable, nowMS},
|
||
},
|
||
{
|
||
query: `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 ` + table + ` FORCE INDEX (` + claimIndex + `)
|
||
WHERE app_code = ? AND status = ? AND lock_until_ms <= ?
|
||
ORDER BY created_at_ms ASC, event_id ASC
|
||
LIMIT ? FOR UPDATE SKIP LOCKED`,
|
||
args: []any{appCode, outbox.StatusDelivering, nowMS},
|
||
},
|
||
}
|
||
for _, branch := range claimBranches {
|
||
remaining := limit - len(records)
|
||
if remaining <= 0 {
|
||
break
|
||
}
|
||
args := append(append([]any{}, branch.args...), remaining)
|
||
branchRecords, err := r.queryRoomOutboxRecords(ctx, tx, branch.query, args...)
|
||
if err != nil {
|
||
_ = tx.Rollback()
|
||
return nil, err
|
||
}
|
||
records = append(records, branchRecords...)
|
||
}
|
||
|
||
for _, record := range records {
|
||
if _, err := tx.ExecContext(ctx,
|
||
`UPDATE `+table+`
|
||
SET status = ?, worker_id = ?, lock_until_ms = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND event_id = ?`,
|
||
outbox.StatusDelivering,
|
||
workerID,
|
||
lockUntilMS,
|
||
nowMS,
|
||
appCode,
|
||
record.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
|
||
}
|
||
|
||
func (r *Repository) queryRoomOutboxRecords(ctx context.Context, tx *sql.Tx, query string, args ...any) ([]outbox.Record, error) {
|
||
rows, err := tx.QueryContext(ctx, query, args...)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
records := make([]outbox.Record, 0)
|
||
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 {
|
||
return nil, err
|
||
}
|
||
if lockUntilValue.Valid {
|
||
record.LockUntilMS = lockUntilValue.V
|
||
}
|
||
|
||
var envelope roomeventsv1.EventEnvelope
|
||
if err := proto.Unmarshal(envelopeBytes, &envelope); err != nil {
|
||
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
|
||
}
|
||
|
||
// MarkOutboxDelivered 标记事件投递成功。
|
||
func (r *Repository) MarkOutboxDelivered(ctx context.Context, eventID string) error {
|
||
return r.markOutboxDeliveredInTable(ctx, roomOutboxTable, eventID)
|
||
}
|
||
|
||
// MarkRobotOutboxDelivered 标记机器人展示事件已经投递成功。
|
||
func (r *Repository) MarkRobotOutboxDelivered(ctx context.Context, eventID string) error {
|
||
return r.markOutboxDeliveredInTable(ctx, roomRobotOutboxTable, eventID)
|
||
}
|
||
|
||
func (r *Repository) markOutboxDeliveredInTable(ctx context.Context, table string, eventID string) error {
|
||
table = roomOutboxTableName(table)
|
||
// 成功后清空 last_error，后续扫描不再返回该事件。
|
||
_, err := r.db.ExecContext(ctx,
|
||
`UPDATE `+table+`
|
||
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 {
|
||
return r.markOutboxRetryableInTable(ctx, roomOutboxTable, eventID, lastErr, nextRetryAtMS)
|
||
}
|
||
|
||
// MarkRobotOutboxRetryable 记录机器人展示事件投递失败，并写入独立重试退避。
|
||
func (r *Repository) MarkRobotOutboxRetryable(ctx context.Context, eventID string, lastErr string, nextRetryAtMS int64) error {
|
||
return r.markOutboxRetryableInTable(ctx, roomRobotOutboxTable, eventID, lastErr, nextRetryAtMS)
|
||
}
|
||
|
||
func (r *Repository) markOutboxRetryableInTable(ctx context.Context, table string, eventID string, lastErr string, nextRetryAtMS int64) error {
|
||
table = roomOutboxTableName(table)
|
||
// 失败转入 retryable，worker 只有到 next_retry_at_ms 后才会重新抢占。
|
||
_, err := r.db.ExecContext(ctx,
|
||
`UPDATE `+table+`
|
||
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 {
|
||
return r.markOutboxDeadInTable(ctx, roomOutboxTable, eventID, lastErr)
|
||
}
|
||
|
||
// MarkRobotOutboxDead 把超过最大重试次数的机器人展示事件转入 failed。
|
||
func (r *Repository) MarkRobotOutboxDead(ctx context.Context, eventID string, lastErr string) error {
|
||
return r.markOutboxDeadInTable(ctx, roomRobotOutboxTable, eventID, lastErr)
|
||
}
|
||
|
||
func (r *Repository) markOutboxDeadInTable(ctx context.Context, table string, eventID string, lastErr string) error {
|
||
table = roomOutboxTableName(table)
|
||
_, err := r.db.ExecContext(ctx,
|
||
`UPDATE `+table+`
|
||
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
|
||
}
|
||
|
||
// DeleteRobotOutbox 直接删除一条机器人展示事件。
|
||
func (r *Repository) DeleteRobotOutbox(ctx context.Context, eventID string) error {
|
||
// robot outbox 只承载 App 展示补偿；超过业务展示窗口后删除比标记 failed 更能降低后续扫描成本。
|
||
_, err := r.db.ExecContext(ctx,
|
||
`DELETE FROM `+roomRobotOutboxTable+`
|
||
WHERE app_code = ? AND event_id = ?`,
|
||
appcode.FromContext(ctx),
|
||
eventID,
|
||
)
|
||
return err
|
||
}
|
||
|
||
// CleanupRobotOutbox 批量删除机器人展示 outbox 的过期记录，不做归档。
|
||
func (r *Repository) CleanupRobotOutbox(ctx context.Context, activeBeforeMS int64, terminalBeforeMS int64, limit int) (int64, int64, error) {
|
||
if limit <= 0 {
|
||
// 清理必须有批量上限，避免误配置导致一次 DELETE 长时间持锁。
|
||
limit = 50000
|
||
}
|
||
|
||
appCode := appcode.FromContext(ctx)
|
||
nowMS := time.Now().UTC().UnixMilli()
|
||
var activeDeleted int64
|
||
var terminalDeleted int64
|
||
if activeBeforeMS > 0 {
|
||
// pending/retryable 没有正在处理的 worker，超过展示窗口后可以直接删除。
|
||
deleted, err := r.deleteRobotOutboxBatch(ctx, `
|
||
DELETE FROM `+roomRobotOutboxTable+`
|
||
WHERE app_code = ?
|
||
AND status IN (?, ?)
|
||
AND created_at_ms < ?
|
||
ORDER BY created_at_ms ASC, event_id ASC
|
||
LIMIT ?`,
|
||
appCode,
|
||
outbox.StatusPending,
|
||
outbox.StatusRetryable,
|
||
activeBeforeMS,
|
||
limit,
|
||
)
|
||
if err != nil {
|
||
return activeDeleted, terminalDeleted, err
|
||
}
|
||
activeDeleted += deleted
|
||
|
||
// delivering 可能正被其他 worker 处理，只删除锁已过期或历史异常无锁的旧记录。
|
||
deleted, err = r.deleteRobotOutboxBatch(ctx, `
|
||
DELETE FROM `+roomRobotOutboxTable+`
|
||
WHERE app_code = ?
|
||
AND status = ?
|
||
AND created_at_ms < ?
|
||
AND (lock_until_ms IS NULL OR lock_until_ms <= ?)
|
||
ORDER BY created_at_ms ASC, event_id ASC
|
||
LIMIT ?`,
|
||
appCode,
|
||
outbox.StatusDelivering,
|
||
activeBeforeMS,
|
||
nowMS,
|
||
limit,
|
||
)
|
||
if err != nil {
|
||
return activeDeleted, terminalDeleted, err
|
||
}
|
||
activeDeleted += deleted
|
||
}
|
||
|
||
if terminalBeforeMS > 0 {
|
||
// delivered/failed 对机器人展示没有审计价值，保留短窗口方便排障后即可删除。
|
||
deleted, err := r.deleteRobotOutboxBatch(ctx, `
|
||
DELETE FROM `+roomRobotOutboxTable+`
|
||
WHERE app_code = ?
|
||
AND status IN (?, ?)
|
||
AND updated_at_ms < ?
|
||
ORDER BY updated_at_ms ASC, event_id ASC
|
||
LIMIT ?`,
|
||
appCode,
|
||
outbox.StatusDelivered,
|
||
outbox.StatusFailed,
|
||
terminalBeforeMS,
|
||
limit,
|
||
)
|
||
if err != nil {
|
||
return activeDeleted, terminalDeleted, err
|
||
}
|
||
terminalDeleted += deleted
|
||
}
|
||
|
||
return activeDeleted, terminalDeleted, nil
|
||
}
|
||
|
||
func (r *Repository) deleteRobotOutboxBatch(ctx context.Context, query string, args ...any) (int64, error) {
|
||
result, err := r.db.ExecContext(ctx, query, args...)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
deleted, err := result.RowsAffected()
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
return deleted, nil
|
||
}
|
||
|
||
func roomOutboxTableName(table string) string {
|
||
switch table {
|
||
case roomRobotOutboxTable:
|
||
return roomRobotOutboxTable
|
||
default:
|
||
return roomOutboxTable
|
||
}
|
||
}
|
||
|
||
func roomOutboxIndexes(table string) (pending string, claim string) {
|
||
if roomOutboxTableName(table) == roomRobotOutboxTable {
|
||
return "idx_room_robot_outbox_pending", "idx_room_robot_outbox_claim"
|
||
}
|
||
return "idx_room_outbox_pending", "idx_room_outbox_claim"
|
||
}
|