package host

import (
	"context"
	"encoding/json"
	"slices"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/controlsock"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	execlease "github.com/carlosframework/platform/internal/hostexec"
	"github.com/carlosframework/platform/internal/hostwrite"
)

func executionExpectation(c *hostclaim.Claim) execlease.ExecutionRequest {
	return execlease.ExecutionRequest{Schema: 1, Host: c.Host, Incarnation: c.Holder.Incarnation, Stamp: c.Holder.Stamp, Session: c.Holder.Session, Op: c.Op.ID}
}

func (p *productionClaimLocal) executionTransport() func(context.Context, execlease.ExecutionRequest) (execlease.ExecutionResponse, error) {
	if p.executionCall != nil {
		return p.executionCall
	}
	return rootExecutionClient(controlsock.Path)
}

func (p *productionClaimLocal) prepareWith(ctx context.Context, actor hostbox.Actor, o hostclaim.Observed) error {
	if p.legacyParticipation {
		return p.prepareLegacyWith(ctx, actor, o)
	}
	c := o.Claim
	if c == nil {
		return execlease.ErrState
	}
	p.mu.Lock()
	old, lifetime, closed, assembly := p.slots[o.Host], p.lifetime, p.closed, p.assembly
	delete(p.serving, o.Host)
	p.mu.Unlock()
	if closed || lifetime == nil || assembly == nil {
		return hostwrite.ErrUnavailable
	}
	if p.Resources.Mode != "agent" {
		return p.checkRootParticipation(ctx, actor, c)
	}
	if c.Free() || c.Op.Kind == hostclaim.KindServing || !slices.Contains(c.Op.Participants, actor.Box) {
		if old != nil {
			select {
			case <-old.Ready:
			case <-ctx.Done():
				return ctx.Err()
			}
			if old.Controller == nil {
				return execlease.ErrLost
			}
			if err := old.Controller.Close(ctx); err != nil {
				return err
			}
			p.mu.Lock()
			if p.slots[o.Host] == old {
				delete(p.slots, o.Host)
			}
			p.mu.Unlock()
		}
		if c.Free() || c.Op.Kind != hostclaim.KindServing {
			return nil
		}
		q := executionExpectation(c)
		q.Operation, q.ID = "executionView", "serving-view"
		out, err := p.executionTransport()(ctx, q)
		if err != nil {
			return err
		}
		boot, err := execlease.BootID()
		if err != nil {
			return err
		}
		if out.Schema != 1 || out.Status != "serving" || out.Host != q.Host || out.Incarnation != q.Incarnation || out.Stamp != q.Stamp || out.Session != q.Session || out.Op != q.Op || out.BootID != boot || out.Token != "" || out.SafeUntil != 0 || out.LeaseETag != "" || !out.Expires.IsZero() {
			return execlease.ErrLost
		}
		if err := execlease.ClearRootClientMirror(p.Resources.DataDir, q.Host); err != nil {
			return err
		}
		p.mu.Lock()
		defer p.mu.Unlock()
		if p.closed || p.slots[o.Host] != nil {
			return execlease.ErrLost
		}
		if p.serving == nil {
			p.serving = make(map[string]execlease.ExecutionRequest)
		}
		p.serving[o.Host] = executionExpectation(c)
		return nil
	}
	if c.Holder == nil {
		return execlease.ErrState
	}
	if old != nil {
		select {
		case <-old.Ready:
		case <-ctx.Done():
			return ctx.Err()
		}
		if old.Err != nil {
			return old.Err
		}
		client, ok := old.Controller.(*execlease.RootClient)
		if !ok {
			return execlease.ErrLost
		}
		return client.Refresh(ctx)
	}
	boot, err := execlease.BootID()
	if err != nil {
		return err
	}
	slot := &claimParticipation{Ready: make(chan struct{}), Captured: o}
	binding := execlease.Binding{Box: actor.Box, Enrollment: actor.Enrollment, Host: o.Host, Incarnation: c.Holder.Incarnation, Op: c.Op.ID, BootID: boot}
	p.mu.Lock()
	if p.closed || p.slots[o.Host] != nil {
		p.mu.Unlock()
		return execlease.ErrControllerBusy
	}
	if p.slots == nil {
		p.slots = make(map[string]*claimParticipation)
	}
	p.slots[o.Host] = slot
	p.mu.Unlock()
	var captureMu sync.Mutex
	captured := o
	slot.Controller, slot.Err = execlease.OpenRootClient(lifetime, execlease.RootClientConfig{
		Binding: binding, Clock: execlease.SystemClock{}, Call: p.executionTransport(),
		Project: func(s execlease.State) error { return execlease.ProjectRootClient(p.Resources.DataDir, s) },
		Observe: func(ctx context.Context) (execlease.ExecutionRequest, error) {
			fresh, err := hostclaim.Read(ctx, p.Store, o.Host)
			if err != nil {
				return execlease.ExecutionRequest{}, err
			}
			if fresh.Absent || fresh.Claim == nil || fresh.Claim.Holder == nil {
				return execlease.ExecutionRequest{}, execlease.ErrLost
			}
			if fresh.Claim.Holder.Incarnation == binding.Incarnation && fresh.Claim.Op.ID == binding.Op {
				captureMu.Lock()
				captured = fresh
				captureMu.Unlock()
			}
			return executionExpectation(fresh.Claim), nil
		},
		CloseAdmission: func() { p.cancelWork(o.Host) },
		Cleanup: func(ctx context.Context) error {
			captureMu.Lock()
			current := captured
			captureMu.Unlock()
			return p.cleanup(ctx, current, binding)
		},
	})
	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) Check(ctx context.Context, actor hostbox.Actor, c *hostclaim.Claim, selection hostbox.Selection) error {
	if p.legacyParticipation {
		return p.checkLegacy(ctx, actor, c, selection)
	}
	if ctx.Err() != nil {
		return ctx.Err()
	}
	if c == nil || c.Holder == nil || actor.Box != selection.Box || actor.Region != selection.Region || c.Holder.Incarnation != selection.Captured.Incarnation || c.Holder.Stamp != selection.Captured.Stamp || slices.Contains(c.Lost, actor.Box) {
		return hostbox.ErrChanged
	}
	p.mu.Lock()
	slot, serving, closed := p.slots[c.Host], p.serving[c.Host], p.closed
	p.mu.Unlock()
	if closed {
		return execlease.ErrLost
	}
	if p.Resources.Mode != "agent" {
		return p.checkRootParticipation(ctx, actor, c)
	}
	if c.Op.Kind == hostclaim.KindServing {
		if slot != nil || serving != executionExpectation(c) {
			return execlease.ErrLost
		}
		return nil
	}
	if slot == nil {
		return execlease.ErrInactive
	}
	select {
	case <-slot.Ready:
	default:
		return execlease.ErrInactive
	}
	if slot.Err != nil {
		return slot.Err
	}
	client, ok := slot.Controller.(*execlease.RootClient)
	if !ok || slot.Binding.Box != actor.Box || slot.Binding.Enrollment != actor.Enrollment {
		return execlease.ErrLost
	}
	return client.Check(ctx, executionExpectation(c))
}

// ackRefreshBefore is how close to its expiry an execution acknowledgement
// may get before this box refreshes it. The console completes a create,
// delete or purge only while every participant's acknowledgement is
// unexpired (hostops FirstPhase: now < Expires - RegistrationSkew), and its
// only retry is recovery, five minutes after the op started. The participant
// loop runs every 30 s and the lease (execlease.TTL, 90 s) renews every 20 s,
// so a just-written acknowledgement has at least 70 s left. Two-thirds of the
// TTL (60 s) leaves the first pass after a write alone and refreshes on the
// next one. A refresh is due at most 60 s before expiry and the next pass
// comes 30 s later, so the acknowledgement stays usable as long as a pass
// starts within 60 - 30 - 5 = 25 s of its tick.
const ackRefreshBefore = 2 * execlease.TTL / 3

func (p *productionClaimLocal) wallNow() time.Time {
	if p.now != nil {
		return p.now()
	}
	return time.Now()
}

func (p *productionClaimLocal) acknowledgeOperation(ctx context.Context, actor hostbox.Actor, o hostclaim.Observed) error {
	c := o.Claim
	if c == nil || c.Holder == nil || (c.Op.Kind != hostclaim.KindCreating && c.Op.Kind != hostclaim.KindDeleting && c.Op.Kind != hostclaim.KindPurging) || !slices.Contains(c.Op.Participants, actor.Box) {
		return nil
	}
	if previous := c.Op.Journal["exec_ack."+actor.Box]; previous != "" {
		return p.refreshAcknowledgement(ctx, actor, o, previous)
	}
	p.mu.Lock()
	slot := p.slots[o.Host]
	p.mu.Unlock()
	if slot == nil {
		return execlease.ErrInactive
	}
	select {
	case <-slot.Ready:
	case <-ctx.Done():
		return ctx.Err()
	}
	if slot.Err != nil {
		return slot.Err
	}
	if slot.Controller == nil {
		return execlease.ErrInactive
	}
	state, err := slot.Controller.Ready(ctx)
	if err != nil {
		return err
	}
	if state.State != "active" || state.Host != o.Host || state.Incarnation != c.Holder.Incarnation || state.Op != c.Op.ID || state.Box != actor.Box || state.Enrollment != actor.Enrollment {
		return execlease.ErrLost
	}
	ack := hostclaim.ExecutionAcknowledgement{Op: state.Op, Enrollment: state.Enrollment, BootID: state.BootID, ControllerID: state.ControllerID, LeaseETag: state.LeaseETag, Expires: state.Expires}
	// Recorded before the write: an uncertain write that landed is still this
	// controller's acknowledgement on the next pass.
	p.mu.Lock()
	slot.AckController = state.ControllerID
	p.mu.Unlock()
	_, err = hostclaim.Update(ctx, p.Store, o, hostclaim.AcknowledgeExecution(actor.Box, ack))
	return err
}

// refreshAcknowledgement extends this controller's own acknowledgement for
// the op while it waits on the console, once fewer than ackRefreshBefore
// remain. It writes only from the controller's current Ready state, under the
// same checks as the first acknowledgement, and never on behalf of another
// boot or controller: an acknowledgement this slot did not write is left to
// expire.
//
// The root broker rotates the controller token on every renewal, so the
// current state's ControllerID changes every 20 s. The slot is what
// identifies the same controller; the refreshed acknowledgement keeps the
// ControllerID the first one bound and carries the current lease ETag and
// expiry.
func (p *productionClaimLocal) refreshAcknowledgement(ctx context.Context, actor hostbox.Actor, o hostclaim.Observed, previous string) error {
	c := o.Claim
	var prior hostclaim.ExecutionAcknowledgement
	if json.Unmarshal([]byte(previous), &prior) != nil || prior.Op != c.Op.ID || prior.Enrollment != actor.Enrollment || prior.BootID == "" || prior.ControllerID == "" {
		return nil
	}
	if !p.wallNow().Add(ackRefreshBefore).After(prior.Expires) {
		return nil
	}
	p.mu.Lock()
	slot := p.slots[o.Host]
	p.mu.Unlock()
	if slot == nil {
		return nil
	}
	select {
	case <-slot.Ready:
	case <-ctx.Done():
		return ctx.Err()
	}
	p.mu.Lock()
	owned := slot.AckController
	p.mu.Unlock()
	if slot.Err != nil || slot.Controller == nil || owned == "" || owned != prior.ControllerID {
		return nil
	}
	state, err := slot.Controller.Ready(ctx)
	if err != nil {
		return err
	}
	if state.State != "active" || state.Host != o.Host || state.Incarnation != c.Holder.Incarnation || state.Op != c.Op.ID || state.Box != actor.Box || state.Enrollment != actor.Enrollment {
		return execlease.ErrLost
	}
	if state.BootID != prior.BootID || state.LeaseETag == "" || !state.Expires.After(prior.Expires) {
		return nil
	}
	ack := hostclaim.ExecutionAcknowledgement{Op: prior.Op, Enrollment: prior.Enrollment, BootID: prior.BootID, ControllerID: prior.ControllerID, LeaseETag: state.LeaseETag, Expires: state.Expires}
	_, err = hostclaim.Update(ctx, p.Store, o, hostclaim.AcknowledgeExecution(actor.Box, ack))
	return err
}