package message

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

	"github.com/tidbcloud/AgentTeam/internal/coredb"
)

// ErrScopeFanoutDeferred marks a durable v2 delivery whose owner-local
// prerequisite state is still warming. It is retryable, but unlike a real
// delivery failure it must not consume the outbox's bounded failure budget.
var ErrScopeFanoutDeferred = errors.New("scope fanout delivery deferred")

type ScopeFanoutAdapter interface {
	Fanout(context.Context, ScopeFanoutOutboxRecord) error
}

// ScopeFanoutFailureObserver receives a best-effort callback only after the
// durable outbox row reaches its final failed state.
type ScopeFanoutFailureObserver interface {
	FanoutFailed(context.Context, ScopeFanoutOutboxRecord, error)
}

// ScopeFanoutWorkerOptions configures optional ownership bounds for an outbox
// worker. WorkspaceGroupID is set by workspace-profile servers so they only
// lease fanout rows for workspaces currently routed to their placement group.
// Empty keeps the monolith/all-profile global behavior.
type ScopeFanoutWorkerOptions struct {
	WorkspaceGroupID string
}

type ScopeFanoutWorker struct {
	db               coredb.DB
	adapter          ScopeFanoutAdapter
	leaseOwner       string
	workspaceGroupID string
	leaseDuration    time.Duration
	now              func() time.Time
	backoff          func(int) time.Duration
}

func NewScopeFanoutWorker(db coredb.DB, adapter ScopeFanoutAdapter, leaseOwner string) ScopeFanoutWorker {
	return NewScopeFanoutWorkerWithOptions(db, adapter, leaseOwner, ScopeFanoutWorkerOptions{})
}

func NewScopeFanoutWorkerWithOptions(db coredb.DB, adapter ScopeFanoutAdapter, leaseOwner string, options ScopeFanoutWorkerOptions) ScopeFanoutWorker {
	return newScopeFanoutWorker(db, adapter, leaseOwner, options, time.Now)
}

func NewScopeFanoutWorkerWithClock(db coredb.DB, adapter ScopeFanoutAdapter, leaseOwner string, now func() time.Time) ScopeFanoutWorker {
	return newScopeFanoutWorker(db, adapter, leaseOwner, ScopeFanoutWorkerOptions{}, now)
}

func newScopeFanoutWorker(db coredb.DB, adapter ScopeFanoutAdapter, leaseOwner string, options ScopeFanoutWorkerOptions, now func() time.Time) ScopeFanoutWorker {
	if now == nil {
		now = time.Now
	}
	return ScopeFanoutWorker{
		db:               db,
		adapter:          adapter,
		leaseOwner:       strings.TrimSpace(leaseOwner),
		workspaceGroupID: strings.TrimSpace(options.WorkspaceGroupID),
		leaseDuration:    time.Minute,
		now:              now,
		backoff:          scopeFanoutRetryBackoff,
	}
}

type ScopeFanoutRunResult struct {
	Leased     int
	Delivered  int
	Retried    int
	Failed     int
	Suppressed int
}

func (worker ScopeFanoutWorker) RunOnce(ctx context.Context, limit int) (ScopeFanoutRunResult, error) {
	if worker.db == nil {
		return ScopeFanoutRunResult{}, fmt.Errorf("scope fanout worker db is nil")
	}
	if worker.adapter == nil {
		return ScopeFanoutRunResult{}, fmt.Errorf("scope fanout adapter is nil")
	}
	if strings.TrimSpace(worker.leaseOwner) == "" {
		return ScopeFanoutRunResult{}, invalidArgument("scope fanout lease owner is required")
	}
	now := worker.now().UTC()
	leaseDuration := worker.leaseDuration
	if leaseDuration <= 0 {
		leaseDuration = time.Minute
	}
	repos := NewScopeRepositories(worker.db)
	leased, err := repos.Outbox.leaseDue(ctx, now, worker.leaseOwner, now.Add(leaseDuration), worker.workspaceGroupID, limit)
	if err != nil {
		return ScopeFanoutRunResult{}, err
	}
	result := ScopeFanoutRunResult{Leased: len(leased)}
	for _, outbox := range leased {
		processed, err := worker.processLeased(ctx, outbox)
		if err != nil {
			return result, err
		}
		result.Delivered += processed.Delivered
		result.Retried += processed.Retried
		result.Failed += processed.Failed
		result.Suppressed += processed.Suppressed
	}
	return result, nil
}

func (worker ScopeFanoutWorker) processLeased(ctx context.Context, outbox ScopeFanoutOutboxRecord) (ScopeFanoutRunResult, error) {
	now := worker.now().UTC()
	repos := NewScopeRepositories(worker.db)
	// LeaseDue returns a payload snapshot. A message mutation can commit after
	// that lease and either refresh the create payload or suppress it on
	// withdrawal. Reload under the lease identity before any external effect so
	// the worker observes whichever transaction committed first.
	current, err := repos.Outbox.LoadLeasedForOwner(ctx, outbox.ID, outbox.LeaseOwner)
	if err != nil {
		if errors.Is(err, ErrNotFound) {
			return ScopeFanoutRunResult{Suppressed: 1}, nil
		}
		return ScopeFanoutRunResult{}, err
	}
	outbox = current
	activity, hasActivity, err := worker.activityForOutbox(ctx, repos, outbox)
	if err != nil {
		if !errors.Is(err, ErrNotFound) {
			return ScopeFanoutRunResult{}, err
		}
		if err := repos.Outbox.MarkSuppressed(ctx, outbox.ID, worker.leaseOwner, now, "activity_missing"); err != nil {
			return ScopeFanoutRunResult{}, err
		}
		return ScopeFanoutRunResult{Suppressed: 1}, nil
	}
	if hasActivity && activity.SuppressedAt != nil {
		if err := repos.Outbox.MarkSuppressed(ctx, outbox.ID, worker.leaseOwner, now, "activity_suppressed"); err != nil {
			return ScopeFanoutRunResult{}, err
		}
		return ScopeFanoutRunResult{Suppressed: 1}, nil
	}
	if shouldSuppressMissingFanoutPayload(outbox) {
		if err := repos.Outbox.MarkSuppressed(ctx, outbox.ID, worker.leaseOwner, now, "payload_missing"); err != nil {
			return ScopeFanoutRunResult{}, err
		}
		return ScopeFanoutRunResult{Suppressed: 1}, nil
	}

	if err := worker.adapter.Fanout(ctx, outbox); err != nil {
		nextAttemptAt := now.Add(worker.retryBackoff(outbox.Attempts))
		if errors.Is(err, ErrScopeFanoutDeferred) && outbox.Status == scopeFanoutOutboxStatusLeasedV2 {
			if markErr := repos.Outbox.MarkDeferred(ctx, outbox.ID, worker.leaseOwner, nextAttemptAt, err.Error()); markErr != nil {
				if errors.Is(markErr, ErrNotFound) {
					return ScopeFanoutRunResult{Suppressed: 1}, nil
				}
				return ScopeFanoutRunResult{}, markErr
			}
			return ScopeFanoutRunResult{Retried: 1}, nil
		}
		if markErr := repos.Outbox.MarkRetryOrFailed(ctx, outbox.ID, worker.leaseOwner, nextAttemptAt, err.Error()); markErr != nil {
			return ScopeFanoutRunResult{}, markErr
		}
		if outbox.Attempts >= outbox.MaxAttempts {
			if observer, ok := worker.adapter.(ScopeFanoutFailureObserver); ok {
				observer.FanoutFailed(ctx, outbox, err)
			}
			return ScopeFanoutRunResult{Failed: 1}, nil
		}
		return ScopeFanoutRunResult{Retried: 1}, nil
	}

	if err := repos.Outbox.MarkDelivered(ctx, outbox.ID, worker.leaseOwner, now); err != nil {
		// A lifecycle transaction may suppress the leased row after the worker's
		// final reload but before the adapter returns its clean no-op. Treat that
		// ownership loss as terminal suppression rather than failing the whole
		// worker batch or retrying a deleted Task.
		if errors.Is(err, ErrNotFound) {
			return ScopeFanoutRunResult{Suppressed: 1}, nil
		}
		return ScopeFanoutRunResult{}, err
	}
	return ScopeFanoutRunResult{Delivered: 1}, nil
}

func (worker ScopeFanoutWorker) activityForOutbox(ctx context.Context, repos ScopeRepositories, outbox ScopeFanoutOutboxRecord) (ScopeActivityEventRecord, bool, error) {
	if strings.TrimSpace(outbox.EventRefType) != DeliverySourceRefActivityEvent || strings.TrimSpace(outbox.EventRefID) == "" {
		return ScopeActivityEventRecord{}, false, nil
	}
	activity, err := repos.ActivityEvents.Load(ctx, outbox.EventRefID)
	if err != nil {
		return ScopeActivityEventRecord{}, false, err
	}
	return activity, true, nil
}

func shouldSuppressMissingFanoutPayload(outbox ScopeFanoutOutboxRecord) bool {
	return strings.TrimSpace(outbox.EventRefType) == DeliverySourceRefMessage &&
		strings.TrimSpace(outbox.FanoutType) == ScopeOutboxFanoutScope &&
		len(outbox.Payload) == 0
}

func (worker ScopeFanoutWorker) retryBackoff(attempt int) time.Duration {
	backoff := worker.backoff
	if backoff == nil {
		backoff = scopeFanoutRetryBackoff
	}
	return backoff(attempt)
}

func scopeFanoutRetryBackoff(attempt int) time.Duration {
	if attempt <= 0 {
		attempt = 1
	}
	if attempt > 6 {
		attempt = 6
	}
	return time.Duration(attempt*attempt) * time.Second
}