package hostbox

import (
	"context"
	"errors"
	"sort"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/park"
)

// RegionalEnrollment must verify current enrollment, cloud identity, region,
// bucket access and retirement assignment on every call. Construction-time
// configuration is not authority.
type RegionalEnrollment interface {
	VerifyRegional(context.Context, string, string, time.Duration) error
}

// RegionalEvidence returns immutable, durable evidence covering every writer
// of residue represented by this exact claim observation. EvidenceAt is the
// original evidence time, never a retry time. LegacyIncarnation is required
// when the legacy instances/<host>/ namespace contains residue.
type RegionalEvidence interface {
	VerifyRetirement(context.Context, hostclaim.Observed, []lease.ObjectVersion) (legacyIncarnation string, evidenceAt time.Time, err error)
}

// RegionalObligationEvidence binds legacy publication chores before their
// first storage effect. It is required only while callers still publish
// obligations without the immutable ParkScope.
type RegionalObligationEvidence interface {
	VerifyObligation(context.Context, hostclaim.Observed, hostclaim.Obligation) (hostclaim.ParkScope, error)
}

// RegionalWorker is one pass of the regional retirement worker. Production
// wiring must provide cloud-roster verification and a durable evidence reader;
// this adapter deliberately has no permissive defaults.
type RegionalWorker struct {
	Regional   Regional
	Enrollment RegionalEnrollment
	Evidence   RegionalEvidence
}

func (w RegionalWorker) validate(ctx context.Context) error {
	e := w.Regional.Engine
	if ctx == nil || w.Enrollment == nil || w.Evidence == nil || e.Store == nil || e.Claims == nil || e.Bucket == "" || e.Region == "" {
		return ErrUnsupported
	}
	return w.Enrollment.VerifyRegional(ctx, e.Region, e.Bucket, RequestLimit)
}

// Census returns every host named by either regional data namespace. The
// VersionedStore contract makes each call a complete paginated listing.
func (w RegionalWorker) Census(ctx context.Context) ([]string, error) {
	if err := w.validate(ctx); err != nil {
		return nil, err
	}
	hosts := map[string]bool{}
	for _, prefix := range []string{"instances/", "moves/"} {
		versions, err := w.Regional.Engine.Store.ListVersions(ctx, w.Regional.Engine.Bucket, prefix)
		if err != nil {
			return nil, err
		}
		for _, v := range versions {
			if !strings.HasPrefix(v.Key, prefix) {
				return nil, ErrUnsupported
			}
			tail := strings.TrimPrefix(v.Key, prefix)
			host, rest, ok := strings.Cut(tail, "/")
			key, keyErr := hostclaim.Key(host)
			if !ok || host == "" || rest == "" || v.VersionID == "" || keyErr != nil || key == "" {
				return nil, ErrUnsupported
			}
			hosts[host] = true
		}
	}
	claimKeys, err := w.Regional.Engine.Claims.List(ctx, "control/hosts/")
	if err != nil {
		return nil, err
	}
	for _, key := range claimKeys {
		tail, ok := strings.CutPrefix(key, "control/hosts/")
		if !ok {
			return nil, ErrUnsupported
		}
		// List is recursive; evidence and exec objects are not claim records.
		if strings.Contains(tail, "/") {
			continue
		}
		if !strings.HasSuffix(tail, ".json") {
			return nil, ErrUnsupported
		}
		host := strings.TrimSuffix(tail, ".json")
		want, keyErr := hostclaim.Key(host)
		if keyErr != nil || want != key {
			return nil, ErrUnsupported
		}
		obs, readErr := hostclaim.Read(ctx, w.Regional.Engine.Claims, host)
		if readErr != nil {
			return nil, readErr
		}
		for _, o := range obs.Claim.Obligations {
			if w.Regional.Engine.OwnsRegion(o.Region) && o.Owner == "region:"+o.Region {
				hosts[host] = true
				break
			}
		}
	}
	out := make([]string, 0, len(hosts))
	for host := range hosts {
		out = append(out, host)
	}
	sort.Strings(out)
	return out, nil
}

// Run performs one bounded pass. Existing chores run in every claim state;
// serving/free claims start sweeps; holderless park_sweep:v1 operations admit
// residue from another region under the persisted operation. A missing claim is
// refused because legacy/backfill disposition is a separate authority.
func (w RegionalWorker) Run(ctx context.Context) error {
	hosts, err := w.Census(ctx)
	if err != nil {
		return err
	}
	var errs []error
	for _, host := range hosts {
		if err := ctx.Err(); err != nil {
			errs = append(errs, err)
			break
		}
		if err := w.runHost(ctx, host); err != nil {
			errs = append(errs, err)
		}
	}
	return errors.Join(errs...)
}

// Loop runs a startup pass immediately, then on each serial notification or
// one-minute reconciliation tick. Cancelling ctx stops and joins the loop.
func (w RegionalWorker) Loop(ctx context.Context, serial <-chan struct{}, report func(error)) error {
	if ctx == nil {
		return ErrUnsupported
	}
	run := func() {
		if err := w.Run(ctx); err != nil && report != nil {
			report(err)
		}
	}
	run()
	ticker := time.NewTicker(time.Minute)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case _, ok := <-serial:
			if !ok {
				serial = nil
				continue
			}
			run()
		case <-ticker.C:
			run()
		}
	}
}

func (w RegionalWorker) runHost(ctx context.Context, host string) error {
	if err := w.Enrollment.VerifyRegional(ctx, w.Regional.Engine.Region, w.Regional.Engine.Bucket, RequestLimit); err != nil {
		return err
	}
	obs, err := hostclaim.Read(ctx, w.Regional.Engine.Claims, host)
	if err != nil {
		return err
	}
	if obs.Absent || obs.Claim == nil {
		return ErrUnsupported
	}
	pending := false
	works := append([]hostclaim.Obligation(nil), obs.Claim.Obligations...)
	for _, selected := range works {
		obs, err = hostclaim.Read(ctx, w.Regional.Engine.Claims, host)
		if err != nil {
			return err
		}
		var o hostclaim.Obligation
		found := false
		for _, current := range obs.Claim.Obligations {
			if current.ID == selected.ID && current.Op == selected.Op {
				o, found = current, true
				break
			}
		}
		if !found {
			continue
		}
		if o.Owner != "region:"+o.Region || !w.Regional.Engine.OwnsRegion(o.Region) {
			continue
		}
		if o.Kind != "park-replica" && o.Kind != "park-generation" {
			return ErrUnsupported
		}
		if o.Park != nil && o.Park.RequestLimit != RequestLimit {
			return ErrUnsupported
		}
		if o.Park == nil {
			provider, ok := w.Evidence.(RegionalObligationEvidence)
			if !ok {
				return ErrUnsupported
			}
			scope, err := provider.VerifyObligation(ctx, obs, o)
			if err != nil {
				return err
			}
			if scope.EvidenceAt.IsZero() || scope.RequestLimit != RequestLimit {
				return ErrUnsupported
			}
			if err := w.Enrollment.VerifyRegional(ctx, w.Regional.Engine.Region, w.Regional.Engine.Bucket, RequestLimit); err != nil {
				return err
			}
			if err := w.Regional.Engine.Bind(ctx, park.Work{Host: host, ID: o.ID, Op: o.Op}, scope); err != nil {
				return err
			}
			o.Park = &scope
		}
		if o.Kind == "park-replica" && o.Park != nil && strings.HasPrefix(o.Park.Location, "moves/"+host+"/") {
			// park-staging uses the existing park-replica shape and moves scope.
		}
		if err := w.Regional.Run(ctx, park.Work{Host: host, ID: o.ID, Op: o.Op}); err != nil {
			if errors.Is(err, park.ErrPending) {
				pending = true
				continue
			}
			return err
		}
	}

	// Chore execution may have changed or freed the claim. Re-read before a
	// possible sweep and require fresh enrollment plus evidence for that read.
	obs, err = hostclaim.Read(ctx, w.Regional.Engine.Claims, host)
	if err != nil {
		return err
	}
	joining := obs.Claim != nil && obs.Claim.Holder == nil && obs.Claim.Op.Kind == hostclaim.KindParking &&
		obs.Claim.Op.ID != "" && obs.Claim.Op.Exec == "" && obs.Claim.Op.Intent["park_sweep"] == "v1"
	if obs.Absent || obs.Claim == nil || (!obs.Claim.Free() && obs.Claim.Op.Kind != hostclaim.KindServing && !joining) {
		return nil
	}
	if pending {
		return nil
	}
	if err := w.Enrollment.VerifyRegional(ctx, w.Regional.Engine.Region, w.Regional.Engine.Bucket, RequestLimit); err != nil {
		return err
	}
	versions, err := w.Regional.Engine.Discover(ctx, host)
	if err != nil {
		return err
	}
	legacy, evidenceAt, err := w.Evidence.VerifyRetirement(ctx, obs, versions)
	if err != nil {
		return err
	}
	if evidenceAt.IsZero() {
		return ErrUnsupported
	}
	current, err := hostclaim.Read(ctx, w.Regional.Engine.Claims, host)
	if err != nil {
		return err
	}
	if current.Absent || current.ETag != obs.ETag {
		return ErrChanged
	}
	if err := w.Enrollment.VerifyRegional(ctx, w.Regional.Engine.Region, w.Regional.Engine.Bucket, RequestLimit); err != nil {
		return err
	}
	verified, err := w.Regional.observe(ctx, host)
	if err != nil {
		return err
	}
	if verified.ETag != obs.ETag {
		return ErrChanged
	}
	_, err = w.Regional.Engine.SweepObserved(ctx, obs, versions, legacy, evidenceAt, RequestLimit)
	return err
}