package server

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

	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/gitlab"
	gapi "gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/gitlab/api"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/module/event_tracker"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/module/modserver"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/errz"
	"go.opentelemetry.io/otel/metric"
)

const (
	sentEventsCounterName        = "sent_events_counter"
	accumulatedEventsCounterName = "accumulated_events_counter"
)

type module struct {
	log                  *slog.Logger
	api                  modserver.API
	eventTracker         event_tracker.EventTrackerCollector
	gitLabClient         gitlab.ClientInterface
	eventReportingPeriod time.Duration
	meter                metric.Meter
	sentEvents           atomic.Int64
}

func newModule(config *modserver.Config, et event_tracker.EventTrackerCollector) (*module, error) {
	m := &module{
		log:                  config.Log,
		api:                  config.API,
		eventTracker:         et,
		gitLabClient:         config.GitLabClient,
		eventReportingPeriod: config.Config.Observability.EventReportingPeriod.AsDuration(),
		meter:                config.Meter,
	}

	err := m.constructSentEventsCounter()
	if err != nil {
		return nil, err
	}

	err = m.constructAccumulatedEventsCounter()
	if err != nil {
		return nil, err
	}

	return m, nil
}

func (m *module) Run(ctx context.Context) error {
	if m.eventReportingPeriod == 0 {
		return nil
	}

	ticker := time.NewTicker(m.eventReportingPeriod)
	defer ticker.Stop()
	done := ctx.Done()
	eventSizeLimit := m.eventTracker.GetEventSizeLimit()
	for {
		select {
		case <-done:
			ctxExit, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()
			// Flush events before exiting
			m.sendEvents(ctxExit) //nolint: contextcheck
			return nil
		case <-ticker.C:
			m.sendEvents(ctx)
		case <-eventSizeLimit:
			m.sendEvents(ctx)
			ticker.Reset(m.eventReportingPeriod)
		}
	}
}

func (m *module) sendEvents(ctx context.Context) {
	if err := m.sendEventsInternal(ctx); err != nil {
		if !errz.ContextDone(err) {
			m.api.HandleProcessingError(ctx, m.log, "Failed to send event data", err)
		}
	}
}

func (m *module) sendEventsInternal(ctx context.Context) error {
	ed := m.eventTracker.CloneEventData()
	if ed.IsEmpty() {
		return nil
	}
	gapiEd := convertEvents(ed)
	err := gapi.SendEvent(ctx, m.gitLabClient, gapiEd)
	// Discard events when it fails to send. This is to avoid trying to send them indefinitely
	m.eventTracker.Subtract(ed)
	if err != nil {
		return err
	}

	for _, v := range ed.Events {
		m.sentEvents.Add(int64(len(v)))
	}
	return nil
}

func (m *module) Name() string {
	return event_tracker.ModuleName
}

func convertEvents(ed *event_tracker.EventData) gapi.EventData {
	em := tool.TransformMap[string, gapi.Event, []event_tracker.Event, []gapi.Event, map[string][]event_tracker.Event, map[string][]gapi.Event](
		ed.Events,
		func(events []event_tracker.Event) int {
			return len(events)
		},
		func(events []event_tracker.Event, sink []gapi.Event) []gapi.Event {
			for _, event := range events {
				sink = append(sink, event)
			}
			return sink
		},
		func(events []gapi.Event) []gapi.Event {
			return events
		},
	)
	return gapi.EventData{
		Events: em,
	}
}

func (m *module) constructSentEventsCounter() error {
	_, err := m.meter.Int64ObservableCounter(
		sentEventsCounterName,
		metric.WithDescription("The total number of sent events from KAS to GitLab monolith"),
		metric.WithInt64Callback(
			func(ctx context.Context, observer metric.Int64Observer) error {
				observer.Observe(m.sentEvents.Load())
				return nil
			}),
	)
	return err
}

func (m *module) constructAccumulatedEventsCounter() error {
	_, err := m.meter.Int64ObservableUpDownCounter(
		accumulatedEventsCounterName,
		metric.WithDescription("The total number of accumulated events in KAS event tracker"),
		metric.WithInt64Callback(
			func(ctx context.Context, observer metric.Int64Observer) error {
				observer.Observe(m.eventTracker.AccumulatedEvents())
				return nil
			}),
	)
	return err
}
