package clusterevent

import (
	"context"
	"encoding/json"
	"fmt"
	"math/rand"
	"runtime/debug"
	"time"

	"github.com/google/uuid"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/samber/lo"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"

	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
)

var (
	consumeInterval = time.Duration(rand.Int63n(1000)) * time.Millisecond

	MetricsEventDelayDuration = prometheus.NewGaugeVec(prometheus.GaugeOpts{
		Subsystem: "event",
		Name:      "delay_duration_seconds",
		Help:      "Event publish delay in seconds",
	}, []string{"event_topic", "app"})

	MetricsEventPendingTotal = prometheus.NewGaugeVec(prometheus.GaugeOpts{
		Subsystem: "event",
		Name:      "pending",
		Help:      "Total events that have not yet been published",
	}, []string{"event_topic", "app"})

	topics = []Topic{TopicClusterEvent, TopicDisplayEvent, TopicMaintenanceWindowEvent}
)

func init() {
	prometheus.MustRegister(
		MetricsEventDelayDuration,
		MetricsEventPendingTotal,
	)
}

// WALCommitter is the interface for committing events to the db wal table.
//
//go:generate go run go.uber.org/mock/mockgen@latest -source=$GOFILE -package=mock_$GOPACKAGE -destination=./mock_$GOPACKAGE/mock_$GOFILE
type WALCommitter interface {
	Write(ctx context.Context, topic Topic, event Event) bizerr.BizError
	BatchWrite(ctx context.Context, topic Topic, events []Event) bizerr.BizError
}

// WALWatcher is the interface for watching events from the db wal table, and handle the new events.
//
//go:generate go run go.uber.org/mock/mockgen@latest -source=$GOFILE -package=mock_$GOPACKAGE -destination=./mock_$GOPACKAGE/mock_$GOFILE
type WALWatcher interface {
	Start()
}

func NewWALCommitter(dbProvider db.TransProvider) WALCommitter {
	dbWALCommitter := &dbWALCommitter{dbProvider: dbProvider}
	go dbWALCommitter.cleanSentEvent(context.Background())
	return dbWALCommitter
}

func NewWALWatcher(dbProvider db.TransProvider, handler Publisher, appName string) WALWatcher {
	dbWALWatcher := &dbWALWatcher{
		dbProvider: dbProvider,
		handler:    handler,
		appName:    appName,
	}

	return dbWALWatcher
}

type dbWALCommitter struct {
	dbProvider db.TransProvider
}

func (c *dbWALCommitter) Write(ctx context.Context, topic Topic, e Event) bizerr.BizError {
	return c.write(ctx, topic, e)
}

func (c *dbWALCommitter) BatchWrite(ctx context.Context, topic Topic, es []Event) bizerr.BizError {
	return c.write(ctx, topic, es...)
}

func (c *dbWALCommitter) write(ctx context.Context, topic Topic, events ...Event) bizerr.BizError {
	wdb := c.dbProvider.WriteDB(ctx)

	if len(events) == 0 {
		return nil
	}

	evs := make([]*EventModel, 0, len(events))
	now := time.Now()
	for _, each := range events {
		uuidStr := uuid.New().String()
		if each.GetEventCreatedAtMs() == 0 {
			each.SetEventCreatedAtMs(now.UnixMilli())
		}
		payload, err := each.MarshalJsonPayload()
		if err != nil {
			return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeJsonMarshal, "marshal event, ev=%s", each)
		}
		payloadStr := payload
		tilog.Info(
			ctx,
			"event wal: try to write event",
			zap.String("uuid", uuidStr),
			zap.String("topic", topic.String()),
			zap.String("payload", payloadStr),
		)

		eventModel := &EventModel{
			Topic:     topic.String(),
			UUID:      uuidStr,
			Payload:   payloadStr,
			Timestamp: now,
			Status:    StatusCreated,
		}

		evs = append(evs, eventModel)
	}
	if err := wdb.Save(evs).Error; err != nil {
		return bizerr.WrapGormError(err, "save events")
	}
	return nil
}

func (c *dbWALCommitter) cleanSentEvent(ctx context.Context) {
	tilog.Info(ctx, "event wal: start clean sent event cronjob")
	ticker := time.NewTicker(30 * time.Minute)
	defer ticker.Stop()

	for range ticker.C {
		tilog.Info(ctx, "event wal: start to clean sent event")
		wdb := c.dbProvider.WriteDB(ctx)
		err := wdb.Model(&EventModel{}).
			Where("status = ? AND created_at < ?", StatusSent, time.Now().Add(-time.Hour*24*2)).
			Unscoped().
			Delete(&EventModel{}).
			Error
		if err != nil {
			tilog.Error(ctx, "event wal: clean sent event failed", zap.Error(err))
		} else {
			tilog.Info(ctx, "event wal: clean sent event successful")
		}
	}
}

type dbWALWatcher struct {
	dbProvider db.TransProvider
	appName    string

	handler Publisher
}

func (w *dbWALWatcher) Start() {
	go func() {
		defer func() {
			if err := recover(); err != nil {
				tilog.PlainError(
					"event wal: watcher panicked",
					zap.Any("err", err),
					zap.String("stack", string(debug.Stack())),
				)
			}
		}()

		ctx := context.Background()

		for {
			for _, topic := range topics {
				w.handleEventsFromWAL(ctx, topic)
				time.Sleep(consumeInterval)
			}
		}
	}()
}

func (w *dbWALWatcher) handleEventsFromWAL(ctx context.Context, topic Topic) {
	err := w.dbProvider.WriteDB(ctx).Transaction(func(tx *gorm.DB) error {
		var eventModels []EventModel

		// Load all pending events from the WAL table.
		// And lock all rows.
		if err := tx.
			Clauses(clause.Locking{
				Strength: "UPDATE",
				Options:  "SKIP LOCKED",
			}).
			Model(&EventModel{}).
			Where("topic=? AND status=?", topic.String(), StatusCreated).Order("`timestamp` ASC").Find(&eventModels).Error; err != nil {
			tilog.Error(ctx, "event wal: get events from wal table failed", zap.Error(err))
			return err
		}

		labelValues := []string{topic.String(), w.appName}
		MetricsEventPendingTotal.WithLabelValues(labelValues...).Set(float64(len(eventModels)))

		if len(eventModels) == 0 {
			// no event was consumed
			tilog.Info(ctx, "event wal: no event to be sent", zap.String("topic", topic.String()))
			return nil
		}

		ids := lo.Map(eventModels, func(i EventModel, _ int) uint { return i.ID })
		uuids := lo.Map(eventModels, func(i EventModel, _ int) string { return i.UUID })

		// Update status.
		result := tx.
			Model(&EventModel{}).
			Where("id IN ?", ids).
			Update("status", StatusSent)
		if result.Error != nil {
			tilog.Error(ctx, "event wal: update event status to sent failed", zap.Error(result.Error))
			return result.Error
		}
		if result.RowsAffected != int64(len(eventModels)) {
			tilog.Warn(
				ctx,
				"event wal: lock event status to sent failed, rows affected not match",
				zap.Int64("rows_affected", result.RowsAffected),
				zap.Int("event_count", len(eventModels)),
			)
			return nil
		}

		tilog.Info(
			ctx,
			"event wal: get events from wal, try to publish",
			zap.Strings("uuids", uuids),
			zap.String("topic", string(topic)),
		)

		// Send events to eventbus.
		for i := range eventModels {
			meta := &Meta{
				Topic:       eventModels[i].Topic,
				UUID:        eventModels[i].UUID,
				TimestampMs: eventModels[i].Timestamp.UnixMilli(),
			}

			var ev Event
			switch eventModels[i].Topic {
			case TopicClusterEvent.String():
				ev = &ClusterEvent{}
			case TopicMaintenanceWindowEvent.String():
				ev = &MaintenanceWindowEvent{}
			case TopicDisplayEvent.String():
				ev = &DisplayEvent{}
			default:
				err := fmt.Errorf(
					"invalid event type, expect %s/%s/%s",
					TopicClusterEvent,
					TopicMaintenanceWindowEvent,
					TopicDisplayEvent,
				)
				tilog.Error(ctx, "event wal: invalid event type", zap.Error(err))
				return err
			}

			if err := json.Unmarshal([]byte(eventModels[i].Payload), ev); err != nil {
				tilog.Error(
					ctx,
					"event wal: unmarshal event failed",
					zap.Error(err),
					zap.String("uuid", eventModels[i].UUID),
				)
				return err
			}

			// This call is blocking until kafka acked, refer to tiinf implementation.
			if err := w.publishToEventbus(ctx, meta, ev); err != nil {
				tilog.Error(
					ctx,
					"event wal: publish event to eventbus failed",
					zap.Error(err),
					zap.String("uuid", eventModels[i].UUID),
				)
				return err
			}
		}

		return nil
	})
	if err != nil {
		tilog.Error(ctx, "handleEventsFromWAL failed: handle event from wal failed", zap.Error(err))
	}
}

func (w *dbWALWatcher) publishToEventbus(ctx context.Context, meta *Meta, ev Event) bizerr.BizError {
	labelValues := []string{
		meta.Topic,
		w.appName,
	}
	delay := time.Since(time.UnixMilli(meta.TimestampMs)).Seconds()

	err := w.handler.Publish(ctx, meta, ev)
	if err != nil {
		MetricsEventDelayDuration.WithLabelValues(labelValues...).Set(delay)
		return err
	}

	MetricsEventDelayDuration.DeleteLabelValues(labelValues...)
	return nil
}