package cdc

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

	"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"
)

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
}

func newHandler(cfg config.Config, store dashboard.Store, projector *dashboard.Projector, logger *slog.Logger) *handler {
	return &handler{
		cfg:       cfg,
		store:     store,
		projector: projector,
		mapper:    dashboard.NewMapper(store, cfg.Dashboard.SysOrigins, cfg.Dashboard.UnknownCountryCode),
		logger:    logger,
	}
}

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 {
				return err
			}
		}
	case canal.DeleteAction:
		for index, row := range e.Rows {
			if err := h.applyRows(ctx, e, index, row, nil); err != nil {
				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 {
				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 !force && h.cfg.CDC.FlushInterval > 0 {
		// canal already throttles calls; keep this hook simple and always persist when invoked.
	}
	return h.store.SaveCheckpoint(context.Background(), pos)
}

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...)
	}

	eventKey := fmt.Sprintf("%s:%d:%s:%s:%d", h.currentFile(), e.Header.LogPos, source, e.Action, rowIndex)
	if user != nil {
		return h.projector.ApplyUser(ctx, eventKey, source, *user, contributions)
	}
	return h.projector.Apply(ctx, eventKey, source, contributions)
}