package host

import (
	"context"
	"errors"
	"log"
	"os"
	"path/filepath"
	"slices"
	"strings"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/hostactors"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	execlease "github.com/carlosframework/platform/internal/hostexec"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/release"
)

type claimParticipation struct {
	Binding          execlease.Binding
	Controller       claimExecutionController
	Ready            chan struct{}
	Err              error
	Config           execlease.Config
	Captured         hostclaim.Observed
	CleanupSucceeded bool
	// AckController is the controller ID this slot's controller recorded in
	// its first execution acknowledgement. Only that acknowledgement is ever
	// refreshed: a restarted agent opens a new slot and leaves an older
	// controller's acknowledgement alone. Guarded by productionClaimLocal.mu.
	AckController string
}
type claimExecutionController interface {
	Ready(context.Context) (execlease.State, error)
	Close(context.Context) error
}

type productionClaimLocal struct {
	executionCall       func(context.Context, execlease.ExecutionRequest) (execlease.ExecutionResponse, error)
	serving             map[string]execlease.ExecutionRequest
	legacyParticipation bool // Only legacy-controller fixtures select this path.

	executionRead func(context.Context, *hostclaim.Claim) (execlease.ExecutionRecord, error)
	Resources     ClaimRuntimeResources
	Identity      hostbox.IdentityProvider
	Directory     hostactors.Directory
	Store         release.Store
	mu            sync.Mutex
	slots         map[string]*claimParticipation
	assembly      *ClaimAssembly
	lifetime      context.Context
	closed        bool
	loopCancel    context.CancelFunc
	loopDone      chan struct{}
	stopIdentity  *claimStopLocal
	// lockRoot is a test seam (region HA N12). Set, the participation's
	// hostguard and hostexec locks live under it and its controllers skip
	// the update-floor check (their UpdateGuardDir is empty; main's floor
	// tests cover it). Empty is production: /run/carlos's locks and the
	// floor under OptDir.
	lockRoot string
	// now is a test seam for the acknowledgement refresh's wall clock; nil
	// is time.Now.
	now func() time.Time
}

// lockDir is the named lock directory under lockRoot, or "" (the
// package's default) when lockRoot is unset.
func (p *productionClaimLocal) lockDir(name string) string {
	if p.lockRoot == "" {
		return ""
	}
	return filepath.Join(p.lockRoot, name)
}

// updateFloor is the controller's update-floor guard directory and
// journal: production's under OptDir, none under the N12 test seam. No
// production constructor sets lockRoot (OpenClaimOwners builds the owner
// without it; only tests assign it), so the floor is never bypassed in
// production.
func (p *productionClaimLocal) updateFloor() (string, string) {
	if p.lockRoot != "" {
		return "", ""
	}
	return filepath.Join(p.Resources.OptDir, "system", "guard"), filepath.Join(p.Resources.OptDir, "system", "router", "journal.json")
}

func (p *productionClaimLocal) bind(ctx context.Context, a *ClaimAssembly) error {
	local, err := a.localStopIdentity(ctx)
	if err != nil {
		return err
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	p.assembly = a
	p.lifetime = ctx
	p.stopIdentity = &local
	return nil
}

// Prepare performs lease I/O before WorkSource takes hostguard. Serving never
// opens a lease. A failed controller is retained as denial until cleanup and
// CC9 positively discharge it, never replaced by age.
func (p *productionClaimLocal) Prepare(ctx context.Context, o hostclaim.Observed) error {
	if o.Claim == nil || (!o.Claim.Free() && o.Claim.Holder == nil) {
		return hostbox.ErrChanged
	}
	actor, err := p.Identity.Verify(ctx)
	if err != nil {
		return err
	}
	return p.prepareWith(ctx, actor, o)
}
func (p *productionClaimLocal) prepareLegacyWith(ctx context.Context, actor hostbox.Actor, o hostclaim.Observed) error {
	var err error
	c := o.Claim
	p.mu.Lock()
	old := p.slots[o.Host]
	a := p.assembly
	lifetime := p.lifetime
	closed := p.closed
	p.mu.Unlock()
	if closed || a == nil || lifetime == nil {
		return hostwrite.ErrUnavailable
	}
	if p.Resources.Mode != "agent" {
		return p.checkRootParticipation(ctx, actor, c)
	}
	if c.Op.Kind == hostclaim.KindServing || c.Free() || !slices.Contains(c.Op.Participants, actor.Box) {

		if old == nil {
			state, err := execlease.InspectLocal(p.Resources.DataDir, o.Host)
			if errors.Is(err, os.ErrNotExist) {
				return nil
			}
			if err != nil {
				return err
			}
			return p.clearUntracked(ctx, actor, o, state, execlease.Config{Binding: state.Binding, Guard: hostguard.Guard{DataDir: p.Resources.DataDir, LockDir: p.lockDir("hostguard"), Claims: hostguard.StoreReader{Store: p.Store}}, LockDir: p.lockDir("hostexec")})
		}
		select {
		case <-old.Ready:
		case <-ctx.Done():
			return ctx.Err()
		}
		if old.Controller != nil {
			// Close joins before it reports cleanup or conditional-release
			// errors. A timed-out join retains the slot and lifetime lock.
			closeErr := old.Controller.Close(ctx)
			if ctx.Err() != nil {
				return errors.Join(closeErr, ctx.Err())
			}
		}
		// A previous failed cleanup remains denial until this fresh positive
		// proof succeeds. Its historical error must not veto CC9 forever.
		if !old.CleanupSucceeded {
			if err = p.cleanup(ctx, old.Captured, old.Config.Binding); err != nil {
				return err
			}
		}

		if err = execlease.ClearStale(ctx, old.Config); err != nil {
			return err
		}
		p.mu.Lock()
		if p.slots[o.Host] == old {
			delete(p.slots, o.Host)
		}
		p.mu.Unlock()
		return nil
	}
	if old != nil {
		select {
		case <-old.Ready:
		case <-ctx.Done():
			return ctx.Err()
		}
		if old.Err != nil {
			return old.Err
		}
		if old.Binding.Op != c.Op.ID || old.Binding.Incarnation != c.Holder.Incarnation {
			return execlease.ErrLost
		}
		_, err = old.Controller.Ready(ctx)
		return err
	}
	boot, err := os.ReadFile("/proc/sys/kernel/random/boot_id")
	if err != nil {
		return err
	}
	slot := &claimParticipation{Ready: make(chan struct{}), Captured: o}
	p.mu.Lock()
	if p.closed || p.slots[o.Host] != nil {
		p.mu.Unlock()
		return execlease.ErrControllerBusy
	}
	p.slots[o.Host] = slot
	p.mu.Unlock()
	binding := execlease.Binding{Box: actor.Box, Enrollment: actor.Enrollment, Host: o.Host, Incarnation: c.Holder.Incarnation, Op: c.Op.ID, BootID: strings.TrimSpace(string(boot))}
	floor, journal := p.updateFloor()
	slot.Config = execlease.Config{Binding: binding, Store: p.Store, Guard: hostguard.Guard{DataDir: p.Resources.DataDir, LockDir: p.lockDir("hostguard"), Claims: hostguard.StoreReader{Store: p.Store}}, Directory: p.Directory, Clock: execlease.SystemClock{}, UpdateGuardDir: floor, UpdateJournal: journal, LockDir: p.lockDir("hostexec"),
		CloseAdmission: func() { p.cancelWork(o.Host) }, CancelWork: func() { p.cancelWork(o.Host) }, Cleanup: func(ctx context.Context) error {
			err := p.cleanup(ctx, o, binding)
			slot.CleanupSucceeded = err == nil
			return err
		}}
	slot.Controller, slot.Err = execlease.Open(lifetime, slot.Config)
	if slot.Err == nil {
		var state execlease.State
		state, slot.Err = slot.Controller.Ready(ctx)
		slot.Binding = state.Binding
	}
	close(slot.Ready)
	return slot.Err
}
func (p *productionClaimLocal) cancelWork(host string) {
	p.mu.Lock()
	a := p.assembly
	p.mu.Unlock()
	if a != nil && a.Wake != nil {
		if pending := a.Wake.get(host); pending != nil {
			pending.cancel()
		}
	}
}
func (p *productionClaimLocal) cleanup(ctx context.Context, o hostclaim.Observed, binding execlease.Binding) error {
	p.mu.Lock()
	a := p.assembly
	local := p.stopIdentity
	p.mu.Unlock()
	if a != nil && local != nil {
		copy := *a
		saved := *local
		copy.stopLocal = func(context.Context) (claimStopLocal, error) { return saved, nil }
		a = &copy
	}
	if a == nil || a.Wake == nil || a.Launches == nil || a.Reconcile == nil || a.Reconcile.Census == nil {
		return hostwrite.ErrUnavailable
	}
	p.cancelWork(o.Host)
	if !p.legacyParticipation {
		if done, err := a.completedRetirementDischarge(ctx, o, binding); done || err != nil {
			return err
		}
	}
	// Claim ask C10 (region HA rulings 56, 66): a relocation that completed
	// on this box is discharged by the persisted participation binding and
	// its exact completed probe evidence, before any ordinary cleanup, so
	// no serving-shaped stop ever replaces the probe's fence. The captured
	// observation is not consulted: after a restart it is the current one.
	if done, err := a.completedRelocationDischarge(ctx, binding); done || err != nil {
		return err
	}
	// Cleanup uses the journal and producer set directly. It must never call
	// WorkSource.Select: Open is still waiting for this cleanup to finish.
	has, err := a.Launches.HasHost(ctx, o.Host)
	if err != nil {
		return err
	}
	if has {
		if o.Claim == nil || o.Claim.Holder == nil || o.Claim.Holder.Session == 0 {
			return ErrClaimStopInventory
		}
		return a.stopCapturedSession(ctx, o)
	}
	if o.Claim == nil || o.Claim.Holder == nil {
		return ErrClaimStopInventory
	}
	if !p.legacyParticipation && a.Launches.Remote() && a.Reconcile.Inventory.Remote() {
		return ctx.Err()
	}
	return a.Reconcile.Census.CheckLocal(ctx, o, hostbox.RecordSelection{Host: o.Host, Incarnation: o.Claim.Holder.Incarnation})
}
func (p *productionClaimLocal) checkLegacy(ctx context.Context, a hostbox.Actor, c *hostclaim.Claim, s hostbox.Selection) error {
	if ctx.Err() != nil {
		return ctx.Err()
	}
	if c == nil || c.Holder == nil || a.Box != s.Box || a.Region != s.Region || c.Holder.Incarnation != s.Captured.Incarnation || slices.Contains(c.Lost, a.Box) {
		return hostbox.ErrChanged
	}
	p.mu.Lock()
	slot := p.slots[c.Host]
	closed := p.closed
	p.mu.Unlock()
	if closed {
		return execlease.ErrLost
	}
	if p.Resources.Mode != "agent" {
		return p.checkRootParticipation(ctx, a, c)
	}
	if c.Op.Kind == hostclaim.KindServing {
		if slot != nil {
			return execlease.ErrLost
		}
		_, err := execlease.InspectLocal(p.Resources.DataDir, c.Host)
		if !errors.Is(err, os.ErrNotExist) {
			return errors.Join(execlease.ErrLost, err)
		}
		return nil
	}
	if slot == nil {
		state, err := execlease.InspectLocal(p.Resources.DataDir, c.Host)
		if err != nil {
			return err
		}
		if state.Op != c.Op.ID || state.Incarnation != c.Holder.Incarnation || state.Box != a.Box || state.Enrollment != a.Enrollment {
			return execlease.ErrState
		}
		return execlease.CheckLocal(p.Resources.DataDir, state.Binding, execlease.SystemClock{}, "")
	}
	select {
	case <-slot.Ready:
	default:
		return execlease.ErrInactive
	}
	if slot.Err != nil {
		return slot.Err
	}
	if slot.Binding.Op != c.Op.ID || slot.Binding.Incarnation != c.Holder.Incarnation || slot.Binding.Box != a.Box || slot.Binding.Enrollment != a.Enrollment {
		return execlease.ErrLost
	}
	return execlease.CheckLocal(p.Resources.DataDir, slot.Binding, execlease.SystemClock{}, "")
}
func (p *productionClaimLocal) Close(ctx context.Context) error {
	p.mu.Lock()
	p.closed = true
	cancel, done := p.loopCancel, p.loopDone
	slots := make([]*claimParticipation, 0, len(p.slots))
	for _, s := range p.slots {
		slots = append(slots, s)
	}
	p.mu.Unlock()
	if cancel != nil {
		cancel()
		select {
		case <-done:
		case <-ctx.Done():
			return ctx.Err()
		}
	}
	var errs []error
	for _, slot := range slots {
		select {
		case <-slot.Ready:
		case <-ctx.Done():
			return errors.Join(errors.Join(errs...), ctx.Err())
		}
		if slot.Controller != nil {
			errs = append(errs, slot.Controller.Close(ctx))
		}
		errs = append(errs, slot.Err)
	}
	return errors.Join(errs...)
}

// start observes participants even when they have no pending local writes.
// Root and adopt only inspect the agent's controller and never own a lease.
func (p *productionClaimLocal) start(ctx context.Context) {
	if p.Resources.Mode != "agent" {
		return
	}
	worker, cancel := context.WithCancel(ctx)
	p.mu.Lock()
	if p.closed || p.loopCancel != nil {
		p.mu.Unlock()
		cancel()
		return
	}
	p.loopCancel, p.loopDone = cancel, make(chan struct{})
	done := p.loopDone
	p.mu.Unlock()
	go func() {
		defer close(done)
		ticker := time.NewTicker(30 * time.Second)
		defer ticker.Stop()
		for {
			err := p.participate(worker)
			if err != nil && worker.Err() == nil {
				log.Printf("host claim participation: %v", err)
			}
			select {
			case <-worker.Done():
				return
			case <-ticker.C:
			}
		}
	}()
}
func (p *productionClaimLocal) participate(ctx context.Context) error {
	identity, finishIdentity := context.WithTimeout(ctx, 30*time.Second)
	actor, err := p.Identity.Verify(identity)
	finishIdentity()
	if err != nil {
		return err
	}
	discovery, finishDiscovery := context.WithTimeout(ctx, 30*time.Second)
	keys, err := p.Store.List(discovery, "control/hosts/")
	finishDiscovery()
	if err != nil {
		return err
	}
	var errs []error
	for _, key := range keys {
		host := strings.TrimSuffix(strings.TrimPrefix(key, "control/hosts/"), ".json")
		canonical, err := hostclaim.Key(host)
		if err != nil || canonical != key {
			continue
		}
		// Each host gets a fresh budget from the worker, including its claim read.
		// One slow earlier host cannot consume every later host's next pass.
		bounded, cancel := context.WithTimeout(ctx, hostbox.RequestLimit)
		err = p.participateHost(bounded, actor, host)
		cancel()
		if err != nil {
			errs = append(errs, err)
		}
		if ctx.Err() != nil {
			return errors.Join(errors.Join(errs...), ctx.Err())
		}
	}
	return errors.Join(errs...)
}
func (p *productionClaimLocal) participateHost(ctx context.Context, actor hostbox.Actor, host string) error {
	o, err := hostclaim.Read(ctx, p.Store, host)
	if err != nil {
		return err
	}
	if o.Absent || o.Claim == nil {
		return nil
	}
	if p.legacyParticipation {
		p.mu.Lock()
		tracked := p.slots[host] != nil
		p.mu.Unlock()
		if !tracked && (o.Claim.Op.Kind == hostclaim.KindServing || !slices.Contains(o.Claim.Op.Participants, actor.Box)) {
			if _, err := execlease.InspectLocal(p.Resources.DataDir, host); errors.Is(err, os.ErrNotExist) {
				return nil
			}
		}
	}
	if err := p.prepareWith(ctx, actor, o); err != nil {
		return err
	}
	if p.legacyParticipation {
		return nil
	}
	return p.acknowledgeOperation(ctx, actor, o)
}