package connector_webhooks

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/capabilities"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/inbound"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
)

type materializerKind string

const (
	materializerRetry          materializerKind = "retry"
	materializerPendingConfig  materializerKind = "pending_config"
	materializerConfigTerminal materializerKind = "config_terminal"
	materializerDead           materializerKind = "dead"
)

type materializerError struct {
	kind materializerKind
	err  error
}

func (e materializerError) Error() string {
	return e.err.Error()
}

func (e materializerError) Unwrap() error {
	return e.err
}

func materializerErrorKind(err error) materializerKind {
	var materializerErr materializerError
	if errors.As(err, &materializerErr) {
		return materializerErr.kind
	}
	return materializerRetry
}

func materializerOutcome(err error, okOutcome string) string {
	if err == nil {
		if okOutcome == "" {
			return "ok"
		}
		return okOutcome
	}
	outcome := string(materializerErrorKind(err))
	if outcome == "" {
		return "error"
	}
	return outcome
}

func materializerRetryError(err error) error {
	return materializerError{kind: materializerRetry, err: err}
}

func materializerPendingConfigError(err error) error {
	return materializerError{kind: materializerPendingConfig, err: err}
}

func materializerConfigTerminalError(err error) error {
	return materializerError{kind: materializerConfigTerminal, err: err}
}

func materializerDeadError(err error) error {
	return materializerError{kind: materializerDead, err: err}
}

type VerifiedDelivery struct {
	Namespace              string
	VerifiedPackID         string
	VerifiedVersion        string
	MaterializePackID      string
	MaterializePackVersion string
	SecretName             string
	RawBody                []byte
	EnqueueMetadata        map[string]string
	BeforeEnqueue          func(context.Context, MaterializationSnapshot) error
}

type MaterializationSnapshot struct {
	PackInstallID string
	PackVersion   string
	DeliveryID    string
	EventCount    int
	Metadata      map[string]string
}

type MaterializeResult struct {
	EventIDs       []string
	DuplicateFlags []bool
	DeliveryID     string
	Metadata       map[string]string
}

type validatedConnectorEvent struct {
	ev                     connectorcapabilities.ConnectorEvent
	routeVersion           int64
	envBytes               []byte
	scopedKey              string
	companionPackInstallID string
	companionFlowVersionID string
}

type preparedVerifiedDelivery struct {
	namespace      string
	packInstallID  string
	packVersion    string
	deliveryID     string
	events         []connector_events.BatchEvent
	baseMetadata   map[string]string
	metadata       map[string]string
	routeEventType []string
	sourceLaneIDs  []string
}

type materializerParserResolution struct {
	parser                  connectorcapabilities.EnvelopeParserHandler
	packInstallID           string
	packVersion             string
	fallbackMaterialization bool
}

func (p preparedVerifiedDelivery) snapshot() MaterializationSnapshot {
	return MaterializationSnapshot{
		PackInstallID: p.packInstallID,
		PackVersion:   p.packVersion,
		DeliveryID:    p.deliveryID,
		EventCount:    len(p.events),
		Metadata:      p.metadata,
	}
}

func (p preparedVerifiedDelivery) result(eventIDs []string, dups []bool) MaterializeResult {
	duplicateCount := 0
	for _, dup := range dups {
		if dup {
			duplicateCount++
		}
	}
	return MaterializeResult{
		EventIDs:       eventIDs,
		DuplicateFlags: dups,
		DeliveryID:     p.deliveryID,
		Metadata:       materializerMetadata(p.baseMetadata, p.deliveryID, len(eventIDs), duplicateCount, p.packInstallID, p.packVersion),
	}
}

func (d *Dispatcher) MaterializeVerifiedDelivery(ctx context.Context, in VerifiedDelivery) (result MaterializeResult, retErr error) {
	timer := newMaterializerStageTimer()
	defer func() {
		timer.finish(materializerOutcome(retErr, "ok"))
	}()

	prepared, err := d.prepareVerifiedDelivery(ctx, in, timer)
	if err != nil {
		return MaterializeResult{}, err
	}
	if len(prepared.events) == 0 {
		return prepared.result([]string{}, []bool{}), nil
	}
	if in.BeforeEnqueue != nil {
		if err := in.BeforeEnqueue(ctx, prepared.snapshot()); err != nil {
			timer.mark("before_enqueue", "retry")
			return MaterializeResult{}, materializerRetryError(err)
		}
		timer.mark("before_enqueue", "ok")
	}

	result, err = d.enqueuePreparedDelivery(ctx, prepared)
	if err != nil {
		timer.mark("enqueue_batch", materializerOutcome(err, "ok"))
		return MaterializeResult{}, err
	}
	timer.mark("enqueue_batch", "ok")
	return result, nil
}

func (d *Dispatcher) prepareVerifiedDelivery(ctx context.Context, in VerifiedDelivery, timer *materializerStageTimer) (preparedVerifiedDelivery, error) {
	if d == nil || d.registry == nil || d.routes == nil || d.enqueueBatch == nil {
		return preparedVerifiedDelivery{}, materializerRetryError(fmt.Errorf("connector_webhooks: materializer is not fully wired"))
	}
	namespace := strings.ToLower(strings.TrimSpace(in.Namespace))
	secretName := strings.ToLower(strings.TrimSpace(in.SecretName))
	if namespace == "" || secretName == "" {
		return preparedVerifiedDelivery{}, materializerConfigTerminalError(fmt.Errorf("connector_webhooks: verified delivery missing namespace or secret"))
	}
	verifiedPackID := strings.TrimSpace(in.VerifiedPackID)
	verifiedVersion := strings.TrimSpace(in.VerifiedVersion)
	packInstallID := strings.TrimSpace(in.MaterializePackID)
	packVersion := strings.TrimSpace(in.MaterializePackVersion)
	pinnedMaterialization := packInstallID != ""
	if packInstallID == "" {
		packInstallID = verifiedPackID
		packVersion = verifiedVersion
	}
	if packInstallID == "" {
		return preparedVerifiedDelivery{}, materializerPendingConfigError(fmt.Errorf("connector_webhooks: verified delivery missing pack incarnation for namespace %q", namespace))
	}

	parserResolution, err := d.resolveVerifiedDeliveryParser(ctx, timer, namespace, packInstallID, packVersion, pinnedMaterialization)
	if err != nil {
		return preparedVerifiedDelivery{}, err
	}
	parser := parserResolution.parser
	packInstallID = parserResolution.packInstallID
	packVersion = parserResolution.packVersion
	fallbackMaterialization := parserResolution.fallbackMaterialization

	events, err := parser.Parse(ctx, &connectorcapabilities.ParserInput{
		Payload:    append([]byte(nil), in.RawBody...),
		SecretName: secretName,
	})
	if err != nil {
		if errors.Is(err, connectorcapabilities.ErrMalformedPayload) {
			if fallbackMaterialization {
				timer.mark("parse", "pending_config")
				return preparedVerifiedDelivery{}, materializerPendingConfigError(err)
			}
			timer.mark("parse", "dead")
			return preparedVerifiedDelivery{}, materializerDeadError(err)
		}
		if errors.Is(err, connectorcapabilities.ErrParserConfigDrift) {
			timer.mark("parse", "pending_config")
			return preparedVerifiedDelivery{}, materializerPendingConfigError(err)
		}
		timer.mark("parse", "retry")
		return preparedVerifiedDelivery{}, materializerRetryError(err)
	}
	timer.mark("parse", "ok")

	validated := make([]validatedConnectorEvent, 0, len(events))
	for _, ev := range events {
		preparedEvent, err := d.prepareMaterializedEvent(ctx, timer, namespace, packInstallID, packVersion, secretName, ev)
		if err != nil {
			return preparedVerifiedDelivery{}, err
		}
		validated = append(validated, preparedEvent)
	}

	batch, deliveryID := buildConnectorEventBatch(validated)
	timer.mark("build_batch", "ok")
	preEnqueueMetadata := materializerMetadata(in.EnqueueMetadata, deliveryID, len(batch), 0, packInstallID, packVersion)
	routeEventTypes := make([]string, 0, len(validated))
	sourceLaneIDs := make([]string, 0, len(validated))
	for _, ve := range validated {
		routeEventTypes = append(routeEventTypes, ve.ev.EventType)
		sourceLaneIDs = append(sourceLaneIDs, materializerSourceLaneID(ve.ev))
	}
	return preparedVerifiedDelivery{
		namespace:      namespace,
		packInstallID:  packInstallID,
		packVersion:    packVersion,
		deliveryID:     deliveryID,
		events:         batch,
		baseMetadata:   in.EnqueueMetadata,
		metadata:       preEnqueueMetadata,
		routeEventType: routeEventTypes,
		sourceLaneIDs:  sourceLaneIDs,
	}, nil
}

func materializerSourceLaneID(event connectorcapabilities.ConnectorEvent) string {
	objectKind := event.ObjectKind
	sourceID := event.SourceID
	if strings.TrimSpace(event.SourceLaneObjectKind) != "" && strings.TrimSpace(event.SourceLaneID) != "" {
		// EnqueueBatch canonicalizes only the override object kind before
		// computing its lane key. Mirror that exact identity here.
		objectKind = strings.TrimSpace(event.SourceLaneObjectKind)
		sourceID = event.SourceLaneID
	}
	if objectKind == "" || sourceID == "" {
		return ""
	}
	return fmt.Sprintf("%d:%s:%d:%s", len(objectKind), objectKind, len(sourceID), sourceID)
}

func (d *Dispatcher) prepareMaterializedEvent(
	ctx context.Context,
	timer *materializerStageTimer,
	namespace string,
	packInstallID string,
	packVersion string,
	secretName string,
	ev connectorcapabilities.ConnectorEvent,
) (validatedConnectorEvent, error) {
	if strings.TrimSpace(ev.IdempotencyKey) == "" {
		return validatedConnectorEvent{}, materializerPendingConfigError(fmt.Errorf("parser returned event with empty IdempotencyKey"))
	}
	if strings.TrimSpace(ev.EventType) == "" {
		return validatedConnectorEvent{}, materializerPendingConfigError(fmt.Errorf("parser returned event with empty EventType"))
	}

	routeVersion, flowDeveloperName, routeErr := d.resolveMaterializerRoute(ctx, namespace, packInstallID, ev.EventType)
	if routeErr != nil {
		if errors.Is(routeErr, connector_inbound_routing.ErrRouteNotDeclared) {
			if _, maErr := d.materializerPackTransient(ctx, namespace); maErr != nil {
				timer.mark("resolve_route", "retry")
				return validatedConnectorEvent{}, materializerRetryError(maErr)
			}
			timer.mark("resolve_route", "pending_config")
			return validatedConnectorEvent{}, materializerPendingConfigError(routeErr)
		}
		timer.mark("resolve_route", "retry")
		return validatedConnectorEvent{}, materializerRetryError(routeErr)
	}
	timer.mark("resolve_route", "ok")

	var companionPackInstallID, companionFlowVersionID string
	if d.resolveCompanion != nil {
		companionIncarnation, isCrossPack, companionErr := d.resolveMaterializerCompanion(ctx, namespace, flowDeveloperName)
		if companionErr != nil {
			timer.mark("resolve_companion", "pending_config")
			return validatedConnectorEvent{}, materializerPendingConfigError(companionErr)
		}
		timer.mark("resolve_companion", "ok")
		if isCrossPack {
			if d.checkConsent == nil {
				timer.mark("check_consent", "retry")
				return validatedConnectorEvent{}, materializerRetryError(fmt.Errorf("connector_webhooks: companion consent checker is not wired"))
			}
			if consentErr := d.checkMaterializerConsent(ctx, companionIncarnation.CompanionPackName, companionIncarnation.CompanionPackInstallID, namespace, packVersion); consentErr != nil {
				timer.mark("check_consent", "pending_config")
				return validatedConnectorEvent{}, materializerPendingConfigError(consentErr)
			}
			timer.mark("check_consent", "ok")
			companionPackInstallID = companionIncarnation.CompanionPackInstallID
			companionFlowVersionID = companionIncarnation.CompanionFlowVersionID
		}
	}

	if strings.TrimSpace(ev.ObjectKind) == "" || strings.TrimSpace(ev.SourceID) == "" {
		return validatedConnectorEvent{}, materializerPendingConfigError(fmt.Errorf("parser returned event with empty typed fields"))
	}
	if (strings.TrimSpace(ev.SourceLaneObjectKind) == "") != (strings.TrimSpace(ev.SourceLaneID) == "") {
		return validatedConnectorEvent{}, materializerPendingConfigError(fmt.Errorf("parser returned partial source lane override"))
	}
	if validateErr := validateParserPayload(ev.Payload); validateErr != nil {
		return validatedConnectorEvent{}, materializerPendingConfigError(validateErr)
	}
	envBytes, encErr := connector_inbound.EncodeQueueEnvelope(ev.ObjectKind, ev.SourceID, json.RawMessage(ev.Payload))
	if encErr != nil {
		return validatedConnectorEvent{}, materializerPendingConfigError(encErr)
	}
	timer.mark("validate_encode", "ok")

	return validatedConnectorEvent{
		ev:                     ev,
		routeVersion:           routeVersion,
		envBytes:               envBytes,
		scopedKey:              scopeIdempotencyKey(secretName, ev.IdempotencyKey),
		companionPackInstallID: companionPackInstallID,
		companionFlowVersionID: companionFlowVersionID,
	}, nil
}

func (d *Dispatcher) resolveVerifiedDeliveryParser(
	ctx context.Context,
	timer *materializerStageTimer,
	namespace string,
	packInstallID string,
	packVersion string,
	pinnedMaterialization bool,
) (materializerParserResolution, error) {
	parser, err := d.resolveMaterializerParser(ctx, namespace, packInstallID)
	if err == nil {
		timer.mark("resolve_parser", "ok")
		return materializerParserResolution{
			parser:        parser,
			packInstallID: packInstallID,
			packVersion:   packVersion,
		}, nil
	}
	if errors.Is(err, connectorcapabilities.ErrCapabilityIncarnationChanged) {
		return d.resolveCurrentParserAfterVerifiedMiss(ctx, timer, namespace, pinnedMaterialization, err, false)
	}
	if errors.Is(err, connectorcapabilities.ErrCapabilityNotAvailable) {
		return d.resolveCurrentParserAfterVerifiedMiss(ctx, timer, namespace, pinnedMaterialization, err, true)
	}
	timer.mark("resolve_parser", "retry")
	return materializerParserResolution{}, materializerRetryError(err)
}

func (d *Dispatcher) resolveCurrentParserAfterVerifiedMiss(
	ctx context.Context,
	timer *materializerStageTimer,
	namespace string,
	pinnedMaterialization bool,
	cause error,
	checkTransient bool,
) (materializerParserResolution, error) {
	if pinnedMaterialization {
		timer.mark("resolve_parser", "pending_config")
		return materializerParserResolution{}, materializerPendingConfigError(cause)
	}
	if checkTransient {
		if _, maErr := d.materializerPackTransient(ctx, namespace); maErr != nil {
			timer.mark("resolve_parser", "retry")
			return materializerParserResolution{}, materializerRetryError(maErr)
		}
	}
	parser, packInstallID, packVersion, fallbackErr := d.resolveCurrentMaterializerParser(ctx, namespace)
	if fallbackErr != nil {
		timer.mark("resolve_parser", materializerOutcome(fallbackErr, "ok"))
		return materializerParserResolution{}, fallbackErr
	}
	timer.mark("resolve_parser", "fallback")
	return materializerParserResolution{
		parser:                  parser,
		packInstallID:           packInstallID,
		packVersion:             packVersion,
		fallbackMaterialization: true,
	}, nil
}

func (d *Dispatcher) enqueuePreparedDelivery(ctx context.Context, prepared preparedVerifiedDelivery) (MaterializeResult, error) {
	enqueuedIDs, dups, err := d.enqueueBatch(ctx, prepared.namespace, prepared.packInstallID, prepared.deliveryID, prepared.events)
	if err != nil {
		return MaterializeResult{}, d.classifyPreparedEnqueueError(prepared, err)
	}
	return prepared.result(enqueuedIDs, dups), nil
}

func (d *Dispatcher) classifyPreparedEnqueueError(prepared preparedVerifiedDelivery, err error) error {
	if errors.Is(err, connector_events.ErrPackIncarnationChanged) {
		for _, eventType := range prepared.routeEventType {
			d.evictMaterializerRoute(prepared.namespace, prepared.packInstallID, eventType)
		}
		return materializerPendingConfigError(err)
	}
	if errors.Is(err, connector_events.ErrCompanionConsentMissing) ||
		errors.Is(err, connector_events.ErrCompanionVersionUnsatisfied) {
		return materializerPendingConfigError(err)
	}
	return materializerRetryError(err)
}

type materializerStageTimer struct {
	start time.Time
	last  time.Time
}

func newMaterializerStageTimer() *materializerStageTimer {
	now := time.Now()
	return &materializerStageTimer{start: now, last: now}
}

func (t *materializerStageTimer) mark(stage, outcome string) {
	if t == nil {
		return
	}
	now := time.Now()
	recordConnectorMaterializerStageDuration(stage, outcome, now.Sub(t.last).Seconds())
	t.last = now
}

func (t *materializerStageTimer) finish(outcome string) {
	if t == nil {
		return
	}
	recordConnectorMaterializerStageDuration("total", outcome, time.Since(t.start).Seconds())
}

func buildConnectorEventBatch(validated []validatedConnectorEvent) ([]connector_events.BatchEvent, string) {
	deliveryKeys := make([]string, len(validated))
	for i, ve := range validated {
		deliveryKeys[i] = ve.scopedKey
	}
	sort.Strings(deliveryKeys)
	deliveryHash := sha256.New()
	for _, k := range deliveryKeys {
		deliveryHash.Write([]byte(k))
		deliveryHash.Write([]byte{0})
	}
	deliveryID := "dlv:" + hex.EncodeToString(deliveryHash.Sum(nil))

	batch := make([]connector_events.BatchEvent, len(validated))
	for i, ve := range validated {
		batch[i] = connector_events.BatchEvent{
			EventKind:              ve.ev.EventType,
			IdempotencyKey:         ve.scopedKey,
			Payload:                json.RawMessage(ve.envBytes),
			RouteVersion:           ve.routeVersion,
			DeliveryID:             deliveryID,
			CompanionPackInstallID: ve.companionPackInstallID,
			CompanionFlowVersionID: ve.companionFlowVersionID,
			SourceLaneObjectKind:   ve.ev.SourceLaneObjectKind,
			SourceLaneID:           ve.ev.SourceLaneID,
			DeliverySequence:       i,
		}
	}
	return batch, deliveryID
}

func materializerMetadata(base map[string]string, deliveryID string, eventCount, duplicateCount int, materializedPackID, materializedPackVersion string) map[string]string {
	out := make(map[string]string, len(base)+6)
	for k, v := range base {
		if strings.TrimSpace(k) != "" {
			out[k] = v
		}
	}
	out["router"] = "connector_ingress_receipt"
	out["deliveryId"] = deliveryID
	out["eventCount"] = fmt.Sprintf("%d", eventCount)
	out["duplicateCount"] = fmt.Sprintf("%d", duplicateCount)
	out["materializedPackId"] = materializedPackID
	out["materializedPackVersion"] = materializedPackVersion
	return out
}

func (d *Dispatcher) materializerPackTransient(ctx context.Context, namespace string) (bool, error) {
	if d == nil || d.packTransient == nil {
		return false, nil
	}
	return d.packTransient(ctx, namespace)
}

func (d *Dispatcher) resolveCurrentMaterializerParser(ctx context.Context, namespace string) (connectorcapabilities.EnvelopeParserHandler, string, string, error) {
	if d == nil || d.resolvePack == nil || d.registry == nil {
		return nil, "", "", materializerPendingConfigError(fmt.Errorf("connector_webhooks: current pack resolver is not wired for namespace %q", namespace))
	}
	row, ok, err := d.resolvePack(ctx, namespace)
	if err != nil {
		return nil, "", "", materializerRetryError(err)
	}
	if !ok || row == nil || strings.TrimSpace(row.InstallID) == "" {
		return nil, "", "", materializerPendingConfigError(fmt.Errorf("connector_webhooks: no active connector pack for namespace %q", namespace))
	}
	packInstallID := strings.TrimSpace(row.InstallID)
	parser, err := d.resolveMaterializerParser(ctx, namespace, packInstallID)
	if err != nil {
		if errors.Is(err, connectorcapabilities.ErrCapabilityIncarnationChanged) ||
			errors.Is(err, connectorcapabilities.ErrCapabilityNotAvailable) {
			return nil, "", "", materializerPendingConfigError(err)
		}
		return nil, "", "", materializerRetryError(err)
	}
	return parser, packInstallID, strings.TrimSpace(row.PackVersion), nil
}