package kas

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"maps"
	"reflect"
	"regexp"
	"slices"
	"time"
	"unsafe"

	"github.com/getsentry/sentry-go"
	"github.com/redis/rueidis"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/api"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/module/modserver"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/module/modshared"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/errz"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/logz"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/syncz"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/pkg/event"
	"gitlab.com/gitlab-org/cluster-integration/tunnel/tool/retry"
	"go.opentelemetry.io/otel/trace"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/anypb"
)

var (
	// Help deduplicate errors like:
	//    read tcp 10.222.67.20:40272->10.216.1.45:11443: read: connection reset by peer
	// by removing the random egress port.

	ipv4 = `(?:\d+\.){3}\d+`

	removePortStdLibError = regexp.MustCompile(`(` + ipv4 + `:)\d+(->` + ipv4 + `:\d+)`)
)

const (
	redisAttemptInterval = 50 * time.Millisecond
	redisInitBackoff     = 100 * time.Millisecond
	redisMaxBackoff      = 10 * time.Second
	redisResetDuration   = 20 * time.Second
	redisBackoffFactor   = 2.0
	redisJitter          = 5.0

	eventsRedisChannel = "kas_events"
)

type SentryHub interface {
	CaptureEvent(event *sentry.Event) *sentry.EventID
}

type serverAPI struct {
	log               *slog.Logger
	hub               SentryHub
	redisClient       rueidis.Client
	gitPushEvent      syncz.SyncSubscriptions[*event.GitPushEvent]
	redisPollConfig   retry.PollConfig
	agentInfoResolver modserver.AgentInfoResolver
}

func newServerAPI(log *slog.Logger, hub SentryHub, redisClient rueidis.Client, agentInfoResolver modserver.AgentInfoResolver) *serverAPI {
	return &serverAPI{
		log:          log,
		hub:          hub,
		redisClient:  redisClient,
		gitPushEvent: *syncz.NewSyncSubscriptions[*event.GitPushEvent](),
		redisPollConfig: retry.PollConfig{
			Interval: redisAttemptInterval,
			Backoff: retry.NewExponentialBackoff(
				redisInitBackoff,
				redisMaxBackoff,
				redisResetDuration,
				redisBackoffFactor,
				redisJitter,
			),
		},
		agentInfoResolver: agentInfoResolver,
	}
}

func (a *serverAPI) HandleProcessingError(ctx context.Context, log *slog.Logger, msg string, err error, fields ...slog.Attr) {
	handleProcessingError(ctx, a.getHub, log, msg, err, a.agentInfoResolver, fields...)
}

func (a *serverAPI) getHub() (SentryHub, string) {
	return a.hub, ""
}

func (a *serverAPI) OnGitPushEvent(cb func(*event.GitPushEvent)) func() {
	return a.gitPushEvent.On(cb)
}

func (a *serverAPI) publishEvent(ctx context.Context, e proto.Message) error {
	payload, err := redisProtoMarshal(e)
	if err != nil {
		return fmt.Errorf("failed to marshal proto message to publish: %w", err)
	}
	publishCmd := a.redisClient.B().Publish().Channel(eventsRedisChannel).Message(rueidis.BinaryString(payload)).Build()
	return a.redisClient.Do(ctx, publishCmd).Error()
}

// subscribeToEvents subscribes to the events Redis channel
// and will dispatch each event to the registered callbacks.
func (a *serverAPI) subscribeToEvents(ctx context.Context) {
	_ = retry.PollWithBackoff(ctx, a.redisPollConfig, func(ctx context.Context) (error, retry.AttemptResult) { //nolint:staticcheck
		subCmd := a.redisClient.B().Subscribe().Channel(eventsRedisChannel).Build()
		err := a.redisClient.Receive(ctx, subCmd, func(msg rueidis.PubSubMessage) {
			protoMessage, err := redisProtoUnmarshal(msg.Message)
			if err != nil {
				a.HandleProcessingError(ctx, a.log, fmt.Sprintf("receiver message in channel %q cannot be unmarshalled into proto message", eventsRedisChannel), err)
				return
			}

			switch e := (protoMessage).(type) {
			case *event.GitPushEvent:
				a.gitPushEvent.Dispatch(e)
			default:
				a.HandleProcessingError(ctx, a.log, "Unable to handle received event", fmt.Errorf("failed to cast proto message of type %T to concrete type", e))
			}
		})
		switch err {
		case nil, context.Canceled, context.DeadlineExceeded: //nolint:errorlint
			return nil, retry.ContinueImmediately
		default:
			a.HandleProcessingError(ctx, a.log, "Error handling Redis SUBSCRIBE", err)
			return nil, retry.Backoff
		}
	})
}

func redisProtoMarshal(m proto.Message) ([]byte, error) {
	a, err := anypb.New(m) // use Any to capture type information so that a value can be instantiated in redisProtoUnmarshal()
	if err != nil {
		return nil, err
	}
	return proto.Marshal(a)
}

func redisProtoUnmarshal(payload string) (proto.Message, error) {
	var a anypb.Any
	// Avoid creating a temporary copy
	payloadBytes := unsafe.Slice(unsafe.StringData(payload), len(payload)) //nolint: gosec
	err := proto.Unmarshal(payloadBytes, &a)
	if err != nil {
		return nil, err
	}
	return a.UnmarshalNew()
}

func handleProcessingError(ctx context.Context, hub func() (SentryHub, string), log *slog.Logger, msg string, err error, agentInfoResolver modserver.AgentInfoResolver, fields ...slog.Attr) {
	// NOTE: in case the msg is empty we'll use a generic msg and rely on the err being descriptive enough for the user.
	if msg == "" {
		msg = "An error occurred"
	}

	slogAttrs := make([]slog.Attr, 0, len(fields)+1)
	slogAttrs = append(slogAttrs, fields...)
	slogAttrs = append(slogAttrs, logz.Error(err))

	var ue errz.UserError
	isUserError := errors.As(err, &ue)
	if isUserError {
		// TODO Don't log it, send it somewhere the user can see it https://gitlab.com/gitlab-org/gitlab/-/issues/277323
		// Log at Info for now.
		log.LogAttrs(ctx, slog.LevelInfo, msg, slogAttrs...)
	} else {
		// FIXME: this should be moved to capture once we do not resolve agent information implicitly for all HandleProcessError calls.
		agentKey := func() api.AgentKey {
			if i := slices.IndexFunc(fields, func(a slog.Attr) bool { return a.Key == api.AgentKeyFieldName }); i != -1 {
				return fields[i].Value.Any().(api.AgentKey)
			}
			return modshared.NoAgentKey
		}()

		extraData := make(map[string]any, len(fields))
		for _, f := range fields {
			v := f.Value.Any()
			if v == nil {
				continue
			}
			extraData[f.Key] = v
		}

		// FIXME: once we have usage data available to map agent key to agent version we don't need to send this implicitly anymore.
		if agentKey != modshared.NoAgentKey {
			if agentInfoResolver == nil {
				panic("There is no AgentInfoResolver registered in the server api to add extra data to errors. This is a programming error. Please open an issue in https://gitlab.com/gitlab-org/cluster-integration/gitlab-agent/-/issues")
			}
			resolverCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 1*time.Second) // preserve context values for tracing
			defer cancel()
			agentInfo, airErr := agentInfoResolver.Get(resolverCtx, agentKey)
			if airErr != nil {
				log.Error("Failed to retrieve extra agent information to send to sentry", logz.Error(airErr))
				// we still continue here to not lose the error that we wanted to send. It'll just not contain the extra agent information.
			}
			maps.Copy(extraData, agentInfo)
		}

		log.LogAttrs(ctx, slog.LevelError, msg, slogAttrs...)

		h, transaction := hub()
		capture(ctx, h, transaction, agentKey, msg, err, extraData)
	}
}

func capture(ctx context.Context, hub SentryHub, transaction string, agentKey api.AgentKey, msg string, err error, extra map[string]any) {
	errStr := removeRandomPort(err.Error())

	e := sentry.NewEvent()
	if agentKey != modshared.NoAgentKey {
		e.User.ID = agentKey.String()
	}
	e.Level = sentry.LevelError
	e.Exception = []sentry.Exception{
		{
			Type:       reflect.TypeOf(err).String(),
			Value:      fmt.Sprintf("%s: %s", msg, errStr),
			Stacktrace: sentry.ExtractStacktrace(err),
		},
	}
	tc := trace.SpanContextFromContext(ctx)
	traceID := tc.TraceID()
	if traceID.IsValid() {
		e.Tags[modserver.SentryFieldTraceID] = traceID.String()
		sampled := "false"
		if tc.IsSampled() {
			sampled = "true"
		}
		e.Tags[modserver.SentryFieldTraceSampled] = sampled
	}
	e.Transaction = transaction
	e.Extra = extra
	hub.CaptureEvent(e)
}

func removeRandomPort(err string) string {
	return removePortStdLibError.ReplaceAllString(err, "${1}x${2}")
}

func ErrReporterWithoutAgentInfos(hub SentryHub) errz.ErrReporter {
	return &errReporterWithoutAgentInfos{hub: hub}
}

type errReporterWithoutAgentInfos struct {
	hub SentryHub
}

func (r *errReporterWithoutAgentInfos) HandleProcessingError(ctx context.Context, log *slog.Logger, msg string, err error, fields ...slog.Attr) {
	// we can safely pass a nil AgentInfoResolver here, because it won't be used with NoAgentKey.
	handleProcessingError(ctx, func() (SentryHub, string) { return r.hub, "" }, log, msg, err, nil, fields...)
}
