package fanout

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/google/uuid"

	"github.com/tidbcloud/AgentTeam/internal/store/message"
)

const (
	scopeFanoutLeaseDuration = time.Minute
	scopeFanoutRetryDelay    = time.Second
)

type scopeFanoutLeaseAttempt struct {
	Outbox message.ScopeFanoutOutboxRecord
}

type scopeFanoutLeaseAttemptContextKey struct{}

func withScopeFanoutLeaseAttempt(ctx context.Context, outbox message.ScopeFanoutOutboxRecord) context.Context {
	if ctx == nil {
		ctx = context.Background()
	}
	return context.WithValue(ctx, scopeFanoutLeaseAttemptContextKey{}, scopeFanoutLeaseAttempt{Outbox: outbox})
}

func scopeFanoutLeaseAttemptFromContext(ctx context.Context) (scopeFanoutLeaseAttempt, bool) {
	if ctx == nil {
		return scopeFanoutLeaseAttempt{}, false
	}
	attempt, ok := ctx.Value(scopeFanoutLeaseAttemptContextKey{}).(scopeFanoutLeaseAttempt)
	if !ok || strings.TrimSpace(attempt.Outbox.ID) == "" || strings.TrimSpace(attempt.Outbox.LeaseOwner) == "" {
		return scopeFanoutLeaseAttempt{}, false
	}
	return attempt, true
}

func (fanout DeliveryFanout) shouldCoordinateInlineTaskMessage(appMessage message.AppMessage, delivery message.MessageDelivery) bool {
	if fanout.scopeFanoutLeases == nil || delivery.IdempotentReplay {
		return false
	}
	return len(taskDeliveryTreeScopeIDs(appMessage, delivery)) > 0
}

// deliverInlineTaskMessage gives the synchronous request path the same
// exclusive outbox ownership used by the replay worker. If another dispatcher
// already owns or completed the row, this path performs no side effects.
//
// The second return value reports whether the coordinator handled the message.
// A missing outbox returns false so rolling legacy writers can retain the old
// direct delivery path without compromising rows created by new writers.
func (fanout DeliveryFanout) deliverInlineTaskMessage(
	ctx context.Context,
	workspaceID string,
	appMessage message.AppMessage,
	delivery message.MessageDelivery,
) (bool, error) {
	if !fanout.shouldCoordinateInlineTaskMessage(appMessage, delivery) {
		return false, nil
	}
	messageID := strings.TrimSpace(delivery.MessageID)
	if messageID == "" {
		messageID = strings.TrimSpace(appMessage.ID)
	}
	if messageID == "" {
		return true, fmt.Errorf("Task fanout message id is required")
	}

	owner := "inline:" + uuid.NewString()
	record, claimed, err := fanout.scopeFanoutLeases.ClaimScopeFanoutForMessage(
		ctx,
		messageID,
		owner,
		time.Now().UTC().Add(scopeFanoutLeaseDuration),
	)
	if err != nil {
		if errors.Is(err, message.ErrNotFound) {
			return false, nil
		}
		return true, err
	}
	if !claimed {
		return true, nil
	}
	if err := validateClaimedTaskFanout(record, workspaceID, appMessage, delivery); err != nil {
		retryErr := fanout.retryScopeFanoutLease(ctx, record, err)
		return true, errors.Join(err, retryErr)
	}

	attemptCtx := withScopeFanoutLeaseAttempt(ctx, record)
	if err := fanout.deliverCommittedMessage(attemptCtx, workspaceID, appMessage, delivery); err != nil {
		retryErr := fanout.retryScopeFanoutLease(ctx, record, err)
		return true, errors.Join(err, retryErr)
	}
	if err := fanout.scopeFanoutLeases.CompleteScopeFanoutLease(ctx, record.ID, record.LeaseOwner); err != nil {
		// Task deletion may suppress the row immediately after the guarded
		// effects finish. The deletion event is then terminal ownership.
		if errors.Is(err, message.ErrNotFound) {
			return true, nil
		}
		return true, err
	}
	return true, nil
}

func validateClaimedTaskFanout(
	record message.ScopeFanoutOutboxRecord,
	workspaceID string,
	appMessage message.AppMessage,
	delivery message.MessageDelivery,
) error {
	if strings.TrimSpace(record.WorkspaceID) != strings.TrimSpace(workspaceID) {
		return fmt.Errorf("claimed Task fanout workspace mismatch")
	}
	messageID := strings.TrimSpace(delivery.MessageID)
	if messageID == "" {
		messageID = strings.TrimSpace(appMessage.ID)
	}
	if strings.TrimSpace(record.MessageID) != messageID ||
		strings.TrimSpace(record.EventRefType) != message.DeliverySourceRefMessage ||
		strings.TrimSpace(record.EventRefID) != messageID ||
		strings.TrimSpace(record.FanoutType) != message.ScopeOutboxFanoutScope {
		return fmt.Errorf("claimed Task fanout source mismatch")
	}
	scopeIDs := taskDeliveryTreeScopeIDs(appMessage, delivery)
	if len(scopeIDs) != 1 || strings.TrimSpace(scopeIDs[0]) != strings.TrimSpace(record.ScopeID) {
		return fmt.Errorf("claimed Task fanout scope mismatch")
	}
	if len(delivery.FollowupDeliveries) > 0 {
		return fmt.Errorf("Task fanout outbox must own exactly one message delivery")
	}
	return nil
}

func (fanout DeliveryFanout) retryScopeFanoutLease(ctx context.Context, record message.ScopeFanoutOutboxRecord, fanoutErr error) error {
	if fanout.scopeFanoutLeases == nil {
		return nil
	}
	nextAttemptAt := time.Now().UTC().Add(scopeFanoutRetryDelay)
	var err error
	if errors.Is(fanoutErr, message.ErrScopeFanoutDeferred) && fanout.scopeFanoutDeferrer != nil {
		err = fanout.scopeFanoutDeferrer.DeferScopeFanoutLease(
			ctx,
			record.ID,
			record.LeaseOwner,
			nextAttemptAt,
			fanoutErr.Error(),
		)
	} else {
		// Older/test coordinators do not expose the optional deferrer. Keep
		// their existing lease-release behavior; production v2 stores implement
		// the branch above so prerequisite warming never spends failure budget.
		err = fanout.scopeFanoutLeases.RetryScopeFanoutLease(
			ctx,
			record.ID,
			record.LeaseOwner,
			nextAttemptAt,
			fanoutErr.Error(),
		)
	}
	if errors.Is(err, message.ErrNotFound) {
		// A concurrent irreversible Task deletion may suppress the lease.
		return nil
	}
	return err
}

func (fanout DeliveryFanout) renewScopeFanoutLease(ctx context.Context) error {
	attempt, ok := scopeFanoutLeaseAttemptFromContext(ctx)
	if !ok || fanout.scopeFanoutLeases == nil {
		return nil
	}
	return fanout.scopeFanoutLeases.RenewScopeFanoutLease(
		ctx,
		attempt.Outbox.ID,
		attempt.Outbox.LeaseOwner,
		time.Now().UTC().Add(scopeFanoutLeaseDuration),
	)
}

func scopeFanoutSourceIdentity(attempt scopeFanoutLeaseAttempt, delivery message.MessageDelivery) (string, string, error) {
	eventID := strings.TrimSpace(delivery.MessageID)
	if eventID == "" {
		eventID = strings.TrimSpace(delivery.Message.MessageID)
	}
	if eventID == "" {
		return "", "", fmt.Errorf("scope fanout Agent delivery event id is required")
	}
	eventType := message.DeliverySourceRefMessage
	if eventID == strings.TrimSpace(attempt.Outbox.EventRefID) {
		eventType = strings.TrimSpace(attempt.Outbox.EventRefType)
	}
	if eventType == "" {
		return "", "", fmt.Errorf("scope fanout Agent delivery event type is required")
	}
	return eventType, eventID, nil
}