package dashboard

import (
	"context"
	"database/sql"
	"errors"
	"log/slog"
	"time"

	"github.com/redis/go-redis/v9"

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

type Projector struct {
	cfg         config.Config
	store       Store
	redisClient *redis.Client
	storageZone *time.Location
	metrics     *monitor.Metrics
}

func NewProjector(cfg config.Config, store Store, redisClient *redis.Client, metrics *monitor.Metrics) *Projector {
	storageZone := mustLocation(cfg.Dashboard.StorageTimezone)
	return &Projector{
		cfg:         cfg,
		store:       store,
		redisClient: redisClient,
		storageZone: storageZone,
		metrics:     metrics,
	}
}

func (p *Projector) Apply(ctx context.Context, eventKey string, source string, contributions []Contribution) error {
	return p.apply(ctx, eventKey, source, nil, contributions)
}

func (p *Projector) ApplyUser(ctx context.Context, eventKey string, source string, user UserCountry, contributions []Contribution) error {
	return p.apply(ctx, eventKey, source, &user, contributions)
}

func (p *Projector) apply(ctx context.Context, eventKey string, source string, user *UserCountry, contributions []Contribution) error {
	event := Event{EventKey: eventKey, Source: source, User: user, Contributions: contributions}
	return p.ApplyEvents(ctx, []Event{event})
}

func (p *Projector) ApplyEvents(ctx context.Context, events []Event) error {
	if len(events) == 0 {
		return nil
	}
	events = compactEvents(events)
	if len(events) == 0 {
		return nil
	}
	if p.cfg.CDC.DryRun {
		slog.Info("dashboard cdc dry run batch", "events", len(events))
		return nil
	}

	var appliedContributions []Contribution
	err := p.withDashboardLock(ctx, func() error {
		return p.store.WithTx(ctx, func(tx *sql.Tx) error {
			batchContributions := make([]Contribution, 0, len(events))
			for _, event := range events {
				applied, err := p.store.MarkEventApplied(ctx, tx, event.EventKey, event.Source)
				if err != nil {
					return err
				}
				if !applied {
					continue
				}
				if event.User != nil {
					if err := p.store.UpsertUserCountry(ctx, tx, *event.User); err != nil {
						return err
					}
				}
				for _, contribution := range event.Contributions {
					if contribution.Empty() {
						continue
					}
					if contribution.EventKey == "" {
						contribution.EventKey = event.EventKey
					}
					if contribution.DeltaSign == 0 {
						contribution.DeltaSign = 1
					}
					batchContributions = append(batchContributions, contribution)
				}
			}
			if err := p.store.ApplyContributionBatch(ctx, tx, batchContributions,
				p.cfg.Dashboard.StorageTimezone, p.cfg.Dashboard.StatTimezones); err != nil {
				return err
			}
			appliedContributions = batchContributions
			return nil
		})
	})
	if err != nil {
		return err
	}
	return p.writeRealtimeRedis(ctx, appliedContributions)
}

func (p *Projector) withDashboardLock(ctx context.Context, fn func() error) error {
	for {
		err := p.store.WithDashboardLock(ctx, p.cfg.CDC.BackfillLockWait, fn)
		if !errors.Is(err, ErrDashboardBackfillLocked) {
			return err
		}
		if p.metrics != nil {
			p.metrics.RecordLockWait()
		}
		slog.Warn("dashboard cdc paused because backfill lock is held")
		timer := time.NewTimer(p.cfg.CDC.FlushInterval)
		select {
		case <-ctx.Done():
			timer.Stop()
			return ctx.Err()
		case <-timer.C:
		}
	}
}

func compactEvents(events []Event) []Event {
	out := make([]Event, 0, len(events))
	for _, event := range events {
		if event.User == nil && len(event.Contributions) == 0 {
			continue
		}
		out = append(out, event)
	}
	return out
}

func mustLocation(name string) *time.Location {
	location, err := time.LoadLocation(name)
	if err != nil {
		slog.Warn("invalid timezone, fallback to UTC", "timezone", name, "error", err)
		return time.UTC
	}
	return location
}