dashboard-cdc-worker/internal/cdc/handler.go

package cdc

import (
	"context"
	"fmt"
	"log/slog"
	"sync"
	"time"

	"github.com/go-mysql-org/go-mysql/canal"
	"github.com/go-mysql-org/go-mysql/mysql"
	"github.com/go-mysql-org/go-mysql/replication"

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

type handler struct {
	canal.DummyEventHandler
	cfg       config.Config
	store     dashboard.Store
	projector *dashboard.Projector
	mapper    *dashboard.Mapper
	logger    *slog.Logger
	mu        sync.RWMutex
	file      string
	batch     []dashboard.Event
	lastFlush time.Time
	metrics   *monitor.Metrics
}

func newHandler(cfg config.Config, store dashboard.Store, projector *dashboard.Projector,
	logger *slog.Logger, metrics *monitor.Metrics) *handler {
	return &handler{
		cfg:       cfg,
		store:     store,
		projector: projector,
		mapper: dashboard.NewMapper(store, cfg.Dashboard.SysOrigins,
			cfg.Dashboard.UnknownCountryCode, cfg.User.CreateTimeDisplayOffset,
			cfg.Dashboard.SourceTimeOffset),
		logger:    logger,
		batch:     make([]dashboard.Event, 0, cfg.CDC.BatchSize),
		lastFlush: time.Now(),
		metrics:   metrics,
	}
}

func (h *handler) OnRotate(_ *replication.EventHeader, rotateEvent *replication.RotateEvent) error {
	h.setCurrentFile(string(rotateEvent.NextLogName))
	return nil
}

func (h *handler) OnRow(e *canal.RowsEvent) error {
	ctx := context.Background()
	source := e.Table.Schema + "." + e.Table.Name
	switch e.Action {
	case canal.InsertAction:
		for index, row := range e.Rows {
			if err := h.applyRows(ctx, e, index, nil, row); err != nil {
				h.recordError(err)
				return err
			}
		}
	case canal.DeleteAction:
		for index, row := range e.Rows {
			if err := h.applyRows(ctx, e, index, row, nil); err != nil {
				h.recordError(err)
				return err
			}
		}
	case canal.UpdateAction:
		for index := 0; index+1 < len(e.Rows); index += 2 {
			if err := h.applyRows(ctx, e, index/2, e.Rows[index], e.Rows[index+1]); err != nil {
				h.recordError(err)
				return err
			}
		}
	default:
		h.logger.Debug("ignore unsupported row action", "source", source, "action", e.Action)
	}
	return nil
}

func (h *handler) OnPosSynced(_ *replication.EventHeader, pos mysql.Position, _ mysql.GTIDSet, force bool) error {
	if h.cfg.CDC.DryRun {
		return nil
	}
	if len(h.batch) > 0 {
		if force || h.shouldFlush() {
			if err := h.flushBatch(context.Background()); err != nil {
				h.recordError(err)
				return err
			}
		} else {
			return nil
		}
	}
	if err := h.store.SaveCheckpoint(context.Background(), pos); err != nil {
		h.recordError(err)
		return err
	}
	if h.metrics != nil {
		h.metrics.SetLastSyncedPosition(pos)
	}
	return nil
}

func (h *handler) String() string {
	return "dashboard-cdc-worker"
}

func (h *handler) setCurrentFile(file string) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.file = file
}

func (h *handler) currentFile() string {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.file
}

func (h *handler) applyRows(ctx context.Context, e *canal.RowsEvent, rowIndex int, oldRaw []interface{}, newRaw []interface{}) error {
	var contributions []dashboard.Contribution
	var user *dashboard.UserCountry
	source := e.Table.Schema + "." + e.Table.Name

	if oldRaw != nil {
		oldRow := rowMap(e, oldRaw)
		items, err := h.mapper.Map(ctx, e.Table.Schema, e.Table.Name, canal.DeleteAction, oldRow)
		if err != nil {
			return err
		}
		contributions = append(contributions, items...)
	}
	if newRaw != nil {
		newRow := rowMap(e, newRaw)
		if e.Table.Name == "user_base_info" {
			u := dashboard.UserCountryFromRow(newRow, h.cfg.Dashboard.UnknownCountryCode)
			user = &u
		}
		items, err := h.mapper.Map(ctx, e.Table.Schema, e.Table.Name, canal.InsertAction, newRow)
		if err != nil {
			return err
		}
		contributions = append(contributions, items...)
	}

	if user == nil && len(contributions) == 0 {
		return nil
	}
	eventKey := fmt.Sprintf("%s:%d:%s:%s:%d", h.currentFile(), e.Header.LogPos, source, e.Action, rowIndex)
	if key := stableRechargeEventKey(source, contributions); key != "" {
		eventKey = key
	} else if key := stableContributionEventKey(contributions); key != "" {
		eventKey = key
	}
	event := dashboard.Event{
		EventKey:      eventKey,
		Source:        source,
		User:          user,
		Contributions: contributions,
	}
	h.batch = append(h.batch, event)
	if h.shouldFlush() {
		return h.flushBatch(ctx)
	}
	return nil
}

func stableContributionEventKey(contributions []dashboard.Contribution) string {
	for _, contribution := range contributions {
		if contribution.DeltaSign < 0 || contribution.EventKey == "" {
			continue
		}
		return contribution.EventKey
	}
	return ""
}

func stableRechargeEventKey(source string, contributions []dashboard.Contribution) string {
	if source != "likei.order_purchase_history" && source != "likei.order_user_purchase_pay" {
		return ""
	}
	for _, contribution := range contributions {
		if contribution.DeltaSign < 0 || contribution.SourcePK == "" || contribution.RechargeDetail == nil {
			continue
		}
		return fmt.Sprintf("reconcile:%s:%s:complete", source, contribution.SourcePK)
	}
	return ""
}

func (h *handler) shouldFlush() bool {
	if len(h.batch) == 0 {
		return false
	}
	if len(h.batch) >= h.cfg.CDC.BatchSize {
		return true
	}
	return h.cfg.CDC.FlushInterval > 0 && time.Since(h.lastFlush) >= h.cfg.CDC.FlushInterval
}

func (h *handler) flushBatch(ctx context.Context) error {
	if len(h.batch) == 0 {
		h.lastFlush = time.Now()
		return nil
	}
	events := make([]dashboard.Event, len(h.batch))
	copy(events, h.batch)
	start := time.Now()
	if err := h.projector.ApplyEvents(ctx, events); err != nil {
		return err
	}
	if h.metrics != nil {
		h.metrics.RecordBatch(len(events), time.Since(start))
	}
	h.logger.Info("dashboard cdc batch flushed", "events", len(events))
	h.batch = h.batch[:0]
	h.lastFlush = time.Now()
	return nil
}

func (h *handler) recordError(err error) {
	if h.metrics != nil {
		h.metrics.RecordError(err)
	}
}