hyapp-server/services/user-service/internal/storage/mysql/user_outbox.go

package mysql

import (
	"context"
	"database/sql"
	"strings"
	"time"

	"hyapp/pkg/appcode"
	"hyapp/pkg/xerr"
)

// UserOutboxRecord is a durable user-service fact waiting for MQ fanout.
type UserOutboxRecord struct {
	AppCode       string
	EventID       string
	EventType     string
	AggregateType string
	AggregateID   int64
	PayloadJSON   string
	CreatedAtMS   int64
	RetryCount    int
	LastError     string
	WorkerID      string
	LockUntilMS   int64
}

// ClaimPendingUserOutbox locks pending user facts so one worker can publish them.
func (r *Repository) ClaimPendingUserOutbox(ctx context.Context, workerID string, batchSize int) ([]UserOutboxRecord, error) {
	if r == nil || r.db == nil {
		return nil, xerr.New(xerr.Unavailable, "mysql repository is not configured")
	}
	if batchSize <= 0 {
		batchSize = 100
	}
	workerID = strings.TrimSpace(workerID)
	if workerID == "" {
		workerID = "user-outbox-worker"
	}
	tx, err := r.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer func() { _ = tx.Rollback() }()
	nowMS := time.Now().UTC().UnixMilli()
	lockUntilMS := nowMS + (30 * time.Second).Milliseconds()
	appCode := appcode.FromContext(ctx)
	records := make([]UserOutboxRecord, 0, batchSize)
	claimBranches := []struct {
		query string
		args  []any
	}{
		{
			query: `
		SELECT app_code, event_id, event_type, aggregate_type, aggregate_id, CAST(payload_json AS CHAR), created_at_ms
		FROM user_outbox FORCE INDEX (idx_user_outbox_status_created)
		WHERE app_code = ? AND status = 'pending' AND next_retry_at_ms <= ? AND lock_until_ms = 0
		ORDER BY created_at_ms ASC, event_id ASC
		LIMIT ? FOR UPDATE SKIP LOCKED
	`,
			args: []any{appCode, nowMS},
		},
		{
			query: `
		SELECT app_code, event_id, event_type, aggregate_type, aggregate_id, CAST(payload_json AS CHAR), created_at_ms
		FROM user_outbox FORCE INDEX (idx_user_outbox_status_created)
		WHERE app_code = ? AND status = 'retryable' AND next_retry_at_ms <= ? AND lock_until_ms = 0
		ORDER BY created_at_ms ASC, event_id ASC
		LIMIT ? FOR UPDATE SKIP LOCKED
	`,
			args: []any{appCode, nowMS},
		},
		{
			query: `
		SELECT app_code, event_id, event_type, aggregate_type, aggregate_id, CAST(payload_json AS CHAR), created_at_ms
		FROM user_outbox FORCE INDEX (idx_user_outbox_lock)
		WHERE app_code = ? AND status = 'running' AND lock_until_ms <= ?
		ORDER BY created_at_ms ASC, event_id ASC
		LIMIT ? FOR UPDATE SKIP LOCKED
	`,
			args: []any{appCode, nowMS},
		},
	}
	for _, branch := range claimBranches {
		remaining := batchSize - len(records)
		if remaining <= 0 {
			break
		}
		args := append(append([]any{}, branch.args...), remaining)
		branchRecords, err := queryUserOutboxRecords(ctx, tx, branch.query, args...)
		if err != nil {
			return nil, err
		}
		records = append(records, branchRecords...)
	}
	for _, record := range records {
		if _, err := tx.ExecContext(ctx, `
			UPDATE user_outbox
			SET status = 'running', worker_id = ?, lock_until_ms = ?, updated_at_ms = ?
			WHERE app_code = ? AND event_id = ?
		`, workerID, lockUntilMS, nowMS, appcode.Normalize(record.AppCode), record.EventID); err != nil {
			return nil, err
		}
	}
	if err := tx.Commit(); err != nil {
		return nil, err
	}
	for index := range records {
		records[index].WorkerID = workerID
		records[index].LockUntilMS = lockUntilMS
	}
	return records, nil
}

func queryUserOutboxRecords(ctx context.Context, tx *sql.Tx, query string, args ...any) ([]UserOutboxRecord, error) {
	rows, err := tx.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	records := make([]UserOutboxRecord, 0)
	for rows.Next() {
		var record UserOutboxRecord
		if err := rows.Scan(&record.AppCode, &record.EventID, &record.EventType, &record.AggregateType, &record.AggregateID, &record.PayloadJSON, &record.CreatedAtMS); err != nil {
			return nil, err
		}
		records = append(records, record)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return records, nil
}

// MarkUserOutboxDelivered marks a user fact as published after RocketMQ accepts it.
func (r *Repository) MarkUserOutboxDelivered(ctx context.Context, eventID string) error {
	if r == nil || r.db == nil {
		return xerr.New(xerr.Unavailable, "mysql repository is not configured")
	}
	_, err := r.db.ExecContext(ctx, `
		UPDATE user_outbox
		SET status = 'delivered', worker_id = '', lock_until_ms = 0, last_error = '', updated_at_ms = ?
		WHERE app_code = ? AND event_id = ?
	`, time.Now().UTC().UnixMilli(), appcode.FromContext(ctx), eventID)
	return err
}

// MarkUserOutboxRetryable releases a failed publish with a bounded retry delay.
func (r *Repository) MarkUserOutboxRetryable(ctx context.Context, eventID string, lastErr string, nextRetryAtMS int64) error {
	if r == nil || r.db == nil {
		return xerr.New(xerr.Unavailable, "mysql repository is not configured")
	}
	_, err := r.db.ExecContext(ctx, `
		UPDATE user_outbox
		SET status = 'retryable', worker_id = '', lock_until_ms = 0, retry_count = retry_count + 1,
		    next_retry_at_ms = ?, last_error = ?, updated_at_ms = ?
		WHERE app_code = ? AND event_id = ? AND status = 'running'
	`, nextRetryAtMS, trimUserOutboxError(lastErr), time.Now().UTC().UnixMilli(), appcode.FromContext(ctx), eventID)
	return err
}

func trimUserOutboxError(value string) string {
	value = strings.TrimSpace(value)
	if len(value) > 512 {
		return value[:512]
	}
	return value
}