300 lines
9.6 KiB
Go
300 lines
9.6 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 追加待投递 room_outbox 事件；机器人展示事件已经改为提交后 best-effort 直发 IM，不再进入 MySQL。
|
||
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 FORCE INDEX (idx_room_outbox_pending)
|
||
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) {
|
||
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()
|
||
appCode := appcode.FromContext(ctx)
|
||
records := make([]outbox.Record, 0, limit)
|
||
// pending 新事件的 next_retry_at_ms 固定为 0，真正决定可见性的是未被抢占的 lock_until_ms；
|
||
// 因此 pending 分支必须走 claim 索引按 created_at_ms 直接取最早记录，避免堆积时 range 扫描历史行。
|
||
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 room_outbox FORCE INDEX (idx_room_outbox_claim)
|
||
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 room_outbox FORCE INDEX (idx_room_outbox_pending)
|
||
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 room_outbox FORCE INDEX (idx_room_outbox_claim)
|
||
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 room_outbox
|
||
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 {
|
||
// 成功后清空 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
|
||
}
|