dashboard-cdc-worker/internal/storage/repository.go

package storage

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"math"
	"sync"
	"time"

	"github.com/go-mysql-org/go-mysql/mysql"
	driverMysql "github.com/go-sql-driver/mysql"

	"dashboard-cdc-worker/internal/config"
	"dashboard-cdc-worker/internal/dashboard"
	"dashboard-cdc-worker/internal/monitor"
)

const serviceName = "dashboard-cdc-worker"
const dashboardBackfillLockName = "country_dashboard_backfill_lock"

type MySQLRepository struct {
	db            *sql.DB
	cfg           config.Config
	metrics       *monitor.Metrics
	userCache     *ttlCache[int64, dashboard.UserCountry]
	gameNameCache *ttlCache[string, string]
	giftMu        sync.Mutex
	giftHighWater map[string]int64
}

func NewMySQLRepository(db *sql.DB, cfg config.Config, metrics *monitor.Metrics) *MySQLRepository {
	return &MySQLRepository{
		db:            db,
		cfg:           cfg,
		metrics:       metrics,
		userCache:     newTTLCache[int64, dashboard.UserCountry](30*time.Minute, 200000),
		gameNameCache: newTTLCache[string, string](6*time.Hour, 10000),
		giftHighWater: make(map[string]int64),
	}
}

func (r *MySQLRepository) WithTx(ctx context.Context, fn func(*sql.Tx) error) error {
	const maxAttempts = 6
	var lastErr error
	for attempt := 1; attempt <= maxAttempts; attempt++ {
		tx, err := r.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
		if err != nil {
			return err
		}
		err = fn(tx)
		if err != nil {
			_ = tx.Rollback()
		} else {
			err = tx.Commit()
		}
		if err == nil {
			return nil
		}
		lastErr = err
		if !isRetryableMySQLError(err) || attempt == maxAttempts {
			return err
		}
		if r.metrics != nil {
			r.metrics.RecordRetry()
		}
		backoff := time.Duration(math.Pow(2, float64(attempt-1))) * 100 * time.Millisecond
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(backoff):
		}
	}
	return lastErr
}

func (r *MySQLRepository) WithDashboardLock(ctx context.Context, timeout time.Duration, fn func() error) error {
	conn, err := r.db.Conn(ctx)
	if err != nil {
		return err
	}
	defer conn.Close()

	timeoutSeconds := int(math.Ceil(timeout.Seconds()))
	if timeoutSeconds < 0 {
		timeoutSeconds = 0
	}
	var acquired sql.NullInt64
	if err := conn.QueryRowContext(ctx, `SELECT GET_LOCK(?, ?)`, dashboardBackfillLockName, timeoutSeconds).Scan(&acquired); err != nil {
		return err
	}
	if !acquired.Valid || acquired.Int64 != 1 {
		if r.metrics != nil {
			r.metrics.RecordLockWait()
		}
		return dashboard.ErrDashboardBackfillLocked
	}
	defer func() {
		var released sql.NullInt64
		_ = conn.QueryRowContext(context.Background(), `SELECT RELEASE_LOCK(?)`, dashboardBackfillLockName).Scan(&released)
	}()
	return fn()
}

func isRetryableMySQLError(err error) bool {
	var mysqlErr *driverMysql.MySQLError
	if !errors.As(err, &mysqlErr) {
		return false
	}
	return mysqlErr.Number == 1213 || mysqlErr.Number == 1205
}

func (r *MySQLRepository) LoadCheckpoint(ctx context.Context) (mysql.Position, bool, error) {
	var file string
	var pos uint32
	err := r.db.QueryRowContext(ctx, `
SELECT binlog_file, binlog_pos
FROM dashboard_cdc_checkpoint
WHERE service_name = ?
`, serviceName).Scan(&file, &pos)
	if err == sql.ErrNoRows {
		return mysql.Position{}, false, nil
	}
	if err != nil {
		return mysql.Position{}, false, err
	}
	position := mysql.Position{Name: file, Pos: pos}
	if r.metrics != nil {
		r.metrics.SetCheckpointPosition(position)
	}
	return position, true, nil
}

func (r *MySQLRepository) SaveCheckpoint(ctx context.Context, pos mysql.Position) error {
	_, err := r.db.ExecContext(ctx, `
INSERT INTO dashboard_cdc_checkpoint (service_name, binlog_file, binlog_pos, updated_at)
VALUES (?, ?, ?, NOW())
ON DUPLICATE KEY UPDATE
  binlog_file = VALUES(binlog_file),
  binlog_pos = VALUES(binlog_pos),
  updated_at = NOW()
`, serviceName, pos.Name, pos.Pos)
	if err == nil && r.metrics != nil {
		r.metrics.SetCheckpointPosition(pos)
	}
	return err
}

func (r *MySQLRepository) MarkEventApplied(ctx context.Context, tx *sql.Tx, eventKey string, source string) (bool, error) {
	result, err := tx.ExecContext(ctx, `
INSERT IGNORE INTO dashboard_cdc_event_dedup (event_key, source, created_at)
VALUES (?, ?, NOW())
`, eventKey, source)
	if err != nil {
		return false, err
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return false, err
	}
	return affected > 0, nil
}

func execAll(ctx context.Context, db *sql.DB, statements []string) error {
	for _, stmt := range statements {
		if _, err := db.ExecContext(ctx, stmt); err != nil {
			return fmt.Errorf("exec schema failed: %w; sql=%s", err, stmt)
		}
	}
	return nil
}