package hostops

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"slices"
	"strconv"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/controlname"
	"github.com/carlosframework/platform/internal/hostactors"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostevidence"
	"github.com/carlosframework/platform/internal/release"
)

var ErrWakeResetUncertain = errors.New("wake reset: stop outcome uncertain")

const WakeResetLifetime = 2 * time.Minute
const WakeResetPrefix = "control/wake-resets/"
const WakeResetArchivePrefix = "control/wake-reset-archive/"

// WakeResetRequest is an immutable signed instruction. Cancellation is a second
// signed object; its presence only prevents new dispatch, never proves no stop.
type WakeResetAdmission struct {
	Account      string            `json:"account"`
	App          string            `json:"app"`
	AppID        string            `json:"app_id"`
	Registration string            `json:"registration"`
	Epoch        uint64            `json:"epoch"`
	Owner        RegistrationOwner `json:"owner"`
}

type WakeResetRequest struct {
	Admission         WakeResetAdmission         `json:"admission"`
	V                 int                        `json:"v"`
	ID                string                     `json:"id"`
	Bucket            string                     `json:"bucket"`
	Console           string                     `json:"console"`
	ConsoleEnrollment string                     `json:"console_enrollment"`
	Namespace         string                     `json:"namespace"`
	Host              string                     `json:"host"`
	ClaimETag         string                     `json:"claim_etag"`
	ClaimGen          uint64                     `json:"claim_gen"`
	Incarnation       string                     `json:"incarnation"`
	Stamp             uint64                     `json:"stamp"`
	Session           uint64                     `json:"session"`
	Operation         string                     `json:"operation"`
	Phase             string                     `json:"phase"`
	Episode           string                     `json:"episode"`
	Attempt           hostclaim.WakeRetryAttempt `json:"attempt"`
	Round             uint64                     `json:"round,omitempty"`
	IssuedAt          time.Time                  `json:"issued_at"`
	Expires           time.Time                  `json:"expires"`
}

type WakeResetCancellation struct {
	V             int    `json:"v"`
	RequestSHA256 string `json:"request_sha256"`
}

type WakeResetTransport struct {
	Store     release.Store
	Directory hostactors.Directory
	Now       func() time.Time
}

func wakeResetID(r WakeResetRequest) string {
	r.ID = ""
	r.IssuedAt = time.Time{}
	r.Expires = time.Time{}
	raw, _ := json.Marshal(r)
	return evidenceDigest(raw)
}

func (r WakeResetRequest) Key() string { return WakeResetPrefix + r.Attempt.Box + "/" + r.ID + ".json" }
func (r WakeResetRequest) validate() error {
	if !controlname.ValidAccount(r.Admission.Account) || !controlname.ValidApp(r.Admission.App) || r.Admission.AppID == "" || r.Admission.Registration == "" || r.Admission.Epoch == 0 || r.Admission.Owner.Console != r.Console || r.Admission.Owner.Enrollment != r.ConsoleEnrollment || r.Admission.Owner.Controller == "" || r.V != 1 || !digestOK(r.ID) || !controlname.ValidHostShape(r.Host) || r.ClaimETag == "" || r.ClaimGen == 0 || !safeSegment(r.Bucket) || !safeSegment(r.Console) || r.ConsoleEnrollment == "" || r.Namespace == "" || !safeSegment(r.Incarnation) || r.Episode == "" || r.Attempt.ID == "" || !safeSegment(r.Attempt.Box) || r.Attempt.Enrollment == "" || r.Attempt.BootID == "" || r.Attempt.ControllerID == "" || r.IssuedAt.IsZero() || !r.Expires.After(r.IssuedAt) || r.Expires.Sub(r.IssuedAt) > WakeResetLifetime {
		return ErrIdentity
	}
	return nil
}

// Authority rechecks both enrollments, namespace delegation and termination
// fences from the current signed roster. An invitation alone grants nothing.
func (t WakeResetTransport) Authority(ctx context.Context, console, enrollment, namespace string) error {
	if t.Store == nil || t.Now == nil || t.Directory.Store == nil {
		return ErrContract
	}
	o, err := t.Directory.Read(ctx)
	if err != nil {
		return err
	}
	c, ok := o.Roster.Consoles[console]
	if !ok || c.Enrollment != enrollment || c.DeploymentBucket != t.Directory.Bucket || !slices.Contains(c.Namespaces, namespace) || c.Compute == nil || o.Roster.RequestLimitNS != int64(hostbox.RequestLimit) {
		return ErrIdentity
	}
	b, ok := o.Roster.Boxes[c.Compute.Box]
	if !ok || b.Enrollment != c.Compute.Enrollment || b.State != "active" {
		return ErrIdentity
	}
	fenced, err := t.Directory.TerminationFence(ctx, o, c.Compute.Box, t.Now())
	if err != nil {
		return err
	}
	if fenced {
		return ErrIdentity
	}
	return nil
}

func (t WakeResetTransport) Authorize(ctx context.Context, r WakeResetRequest) error {
	if err := r.validate(); err != nil {
		return err
	}
	if t.Store == nil || t.Now == nil || r.Bucket != t.Directory.Bucket || r.IssuedAt.After(t.Now().Add(5*time.Second)) || !t.Now().Before(r.Expires) {
		return ErrIdentity
	}
	if _, _, err := t.Store.Get(ctx, wakeResetArchiveKey(r)); err == nil {
		return ErrIdentity
	} else if !errors.Is(err, os.ErrNotExist) {
		return err
	}
	if err := t.Authority(ctx, r.Console, r.ConsoleEnrollment, r.Namespace); err != nil {
		return err
	}
	roster, err := t.Directory.Read(ctx)
	if err != nil {
		return err
	}
	b, ok := roster.Roster.Boxes[r.Attempt.Box]
	if !ok || b.State != "active" || b.Enrollment != r.Attempt.Enrollment || b.Namespace != r.Namespace {
		return ErrIdentity
	}
	fenced, err := t.Directory.TerminationFence(ctx, roster, r.Attempt.Box, t.Now())
	if err != nil {
		return err
	}
	if fenced {
		return ErrIdentity
	}
	o, err := hostclaim.Read(ctx, t.Store, r.Host)
	if err != nil {
		return err
	}
	if !r.matches(o) {
		return ErrIdentity
	}
	h := o.Claim.Holder
	if h.Account != r.Admission.Account || h.App != r.Admission.App {
		return ErrIdentity
	}
	admission, err := ReadAdmission(ctx, t.Store, r.Admission.Account, r.Admission.App, r.Admission.AppID)
	if err != nil {
		return err
	}
	reg, ok := admission.Value.Registrations[r.Admission.Registration]
	if !ok || admission.Value.Epoch != r.Admission.Epoch || reg.Epoch != r.Admission.Epoch || reg.Owner != r.Admission.Owner || reg.Kind != RegistrationWakeRetryReset || reg.Incarnation != r.Incarnation || reg.Host != r.Host || reg.Claim != SavedObservation(o) || !RegistrationLive(reg, t.Now()) {
		return ErrIdentity
	}
	return ctx.Err()
}
func (r WakeResetRequest) matches(o hostclaim.Observed) bool {
	if o.Claim == nil || o.Claim.Holder == nil || o.ETag != r.ClaimETag || o.Claim.Gen != r.ClaimGen || o.Claim.Op.Kind != hostclaim.KindServing || o.Claim.Op.ID != r.Operation || o.Claim.Op.Phase != r.Phase {
		return false
	}
	h := o.Claim.Holder
	wr := h.WakeRetry
	return h.Incarnation == r.Incarnation && h.Stamp == r.Stamp && h.Session == r.Session && wr != nil && wr.State == "attempting" && wr.EpisodeID == r.Episode && wr.Attempt != nil && *wr.Attempt == r.Attempt
}

// Read checks the signed envelope and key binding before the worker uses a
// request. The deployment signing key authenticates only console requests;
// the box's separate enrolled evidence key authenticates the stop result.
func (t WakeResetTransport) Read(ctx context.Context, key string) (WakeResetRequest, error) {
	var r WakeResetRequest
	raw, _, err := t.Store.Get(ctx, key)
	if err != nil {
		return r, err
	}
	if len(raw) > 32<<10 || release.VerifyRootEnvelope(t.Directory.PublicKey, raw, &r) != nil || r.validate() != nil || r.ID != wakeResetID(r) || r.Key() != key {
		return r, ErrIdentity
	}
	return r, nil
}
func (t WakeResetTransport) cancelled(ctx context.Context, r WakeResetRequest) (bool, error) {
	raw, _, err := t.Store.Get(ctx, r.Key()+".cancel")
	if errors.Is(err, os.ErrNotExist) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	var cancel WakeResetCancellation
	body, _ := json.Marshal(r)
	if release.VerifyRootEnvelope(t.Directory.PublicKey, raw, &cancel) != nil || cancel.V != 1 || cancel.RequestSHA256 != evidenceDigest(body) {
		return false, ErrIdentity
	}
	return true, nil
}

// Pass invokes only the selected box's local stop owner. Positive completion
// comes exclusively from VerifyReceipt. Replay reuses root's durable stop
// fence; a failed call or publication never manufactures a completion record.
func (t WakeResetTransport) Pass(ctx context.Context, box string, stop func(context.Context, hostclaim.Observed, hostclaim.WakeRetryAttempt) error) error {
	_, err := t.Scan(ctx, box, stop)
	return err
}

// Scan is Pass that also reports how many request objects it listed, so an
// idle worker can poll the bucket less often.
func (t WakeResetTransport) Scan(ctx context.Context, box string, stop func(context.Context, hostclaim.Observed, hostclaim.WakeRetryAttempt) error) (int, error) {
	if t.Store == nil || t.Now == nil || stop == nil || !safeSegment(box) {
		return 0, ErrContract
	}
	keys, err := t.Store.List(ctx, WakeResetPrefix+box+"/")
	if err != nil {
		return 0, err
	}
	var errs []error
	listed := 0
	for _, key := range keys {
		if !strings.HasSuffix(key, ".json") {
			continue
		}
		listed++
		r, err := t.Read(ctx, key)
		if err != nil {
			errs = append(errs, err)
			continue
		}
		if r.Attempt.Box != box {
			errs = append(errs, ErrIdentity)
			continue
		}
		if !t.Now().Before(r.Expires) {
			continue
		}
		if err = t.VerifyReceipt(ctx, r); err == nil {
			continue
		}
		cancelled, err := t.cancelled(ctx, r)
		if err != nil {
			errs = append(errs, err)
			continue
		}
		if cancelled {
			continue
		}
		if err = t.Authorize(ctx, r); err != nil {
			errs = append(errs, err)
			continue
		}
		o, err := hostclaim.Read(ctx, t.Store, r.Host)
		if err != nil || !r.matches(o) {
			errs = append(errs, errors.Join(err, ErrIdentity))
			continue
		}
		request, done := context.WithDeadline(ctx, minResetDeadline(t.Now(), r.Expires))
		err = stop(request, o, r.Attempt)
		done()
		if err != nil {
			errs = append(errs, errors.Join(ErrWakeResetUncertain, err))
			continue
		}
		if err = t.VerifyReceipt(ctx, r); err != nil {
			errs = append(errs, err)
		}
	}
	return listed, errors.Join(errs...)
}

// VerifyReceipt accepts only a signed root receipt for the full captured writer
// tuple and claim ETag. An expired request can recover a positive receipt: expiry
// closes dispatch, not evidence. Missing signatures and missing objects remain
// uncertain. The unsigned list is discovery, never authority.
func (t WakeResetTransport) VerifyReceipt(ctx context.Context, r WakeResetRequest) error {
	if r.validate() != nil || t.Store == nil || t.Now == nil {
		return ErrIdentity
	}
	roster, err := t.Directory.Read(ctx)
	if err != nil {
		return errors.Join(ErrWakeResetUncertain, err)
	}
	actor, ok := roster.Roster.Boxes[r.Attempt.Box]
	if !ok || actor.State != "active" || actor.Enrollment != r.Attempt.Enrollment {
		return ErrIdentity
	}
	fenced, err := t.Directory.TerminationFence(ctx, roster, r.Attempt.Box, t.Now())
	if err != nil || fenced {
		return errors.Join(ErrIdentity, err)
	}
	prefix := fmt.Sprintf("control/hosts/%s/serving-evidence/%s/", r.Host, r.Attempt.Box)
	keys, err := t.Store.List(ctx, prefix)
	if err != nil {
		return errors.Join(ErrWakeResetUncertain, err)
	}
	for _, key := range keys {
		if !strings.HasSuffix(key, ".json") {
			continue
		}
		raw, _, err := t.Store.Get(ctx, key)
		if err != nil {
			return errors.Join(ErrWakeResetUncertain, err)
		}
		var p StopReceipt
		if decode(raw, &p) != nil {
			continue
		}
		if p.Host != r.Host || p.Incarnation != r.Incarnation || p.InstallationStamp != r.Stamp || p.WriterSession != r.Session || p.OperationID != r.Operation || p.Phase != r.Phase || p.Box != r.Attempt.Box || p.Enrollment != r.Attempt.Enrollment || p.BootID != r.Attempt.BootID || p.ControllerID != r.Attempt.ControllerID || p.ClaimETag != r.ClaimETag || p.ClaimGen != r.ClaimGen {
			continue
		}
		revision, err := strconv.ParseUint(p.DirectoryRevision, 10, 64)
		if err != nil || revision == 0 || revision > roster.Roster.Revision || p.Region != actor.Region || p.V != 1 || !safeSegment(p.ID) || key != prefix+p.ID+".json" || p.Kind != "stopped" || p.ObligationID != "" || p.FenceID == "" || p.RequestLimitNS != int64(hostbox.RequestLimit) || p.EvidenceAt.IsZero() || p.EvidenceAt.After(t.Now().Add(5*time.Second)) || !digestOK(p.InventorySHA256) || !digestOK(p.LocalJournalSHA256) || !digestOK(p.ProducerDrainDigest) || len(p.CgroupIdentities) == 0 {
			continue
		}
		sig, _, err := t.Store.Get(ctx, key+".sig")
		if err != nil {
			continue
		}
		if hostevidence.Verify(actor.CensusPublicKey, key, raw, sig) == nil {
			return nil
		}
	}
	return ErrWakeResetUncertain
}
func (t WakeResetTransport) Wait(ctx context.Context, r WakeResetRequest, every time.Duration) error {
	if every <= 0 {
		return ErrContract
	}
	ticker := time.NewTicker(every)
	defer ticker.Stop()
	for {
		if err := ctx.Err(); err != nil {
			return errors.Join(ErrWakeResetUncertain, err)
		}
		if err := t.VerifyReceipt(ctx, r); err == nil {
			return nil
		}
		if !t.Now().Before(r.Expires) {
			return ErrWakeResetUncertain
		}
		select {
		case <-ctx.Done():
			return errors.Join(ErrWakeResetUncertain, ctx.Err())
		case <-ticker.C:
		}
	}
}

// RequestWakeReset is the sole console write door. Only a live exact reset
// registration can issue it; raw bucket handles never escape ControlWriter.
func (w *ControlWriter) RequestWakeReset(ctx context.Context, t WakeResetTransport, signer release.Signer, console, enrollment, namespace string, o hostclaim.Observed, attempt hostclaim.WakeRetryAttempt) (WakeResetRequest, error) {
	var r WakeResetRequest
	if w == nil || w.ready() != nil || o.Claim == nil || o.Claim.Holder == nil || o.Claim.Holder.WakeRetry == nil {
		return r, ErrContract
	}
	grants, _ := ctx.Value(lifecycleGrantKey{}).([]lifecycleGrant)
	allowed := false
	h := o.Claim.Holder
	var binding WakeResetAdmission
	for _, g := range grants {
		if g.w != w || g.registration != RegistrationWakeRetryReset || g.incarnation != h.Incarnation || g.account != h.Account || g.app != h.App || g.owner.Console != console || g.owner.Enrollment != enrollment {
			continue
		}
		live, err := grantLive(ctx, w.raw, g)
		if err != nil {
			return r, err
		}
		allowed = allowed || live
		if live {
			binding = WakeResetAdmission{Account: g.account, App: g.app, AppID: g.appID, Registration: g.operation, Epoch: g.epoch, Owner: g.owner}
		}
	}
	if !allowed {
		return r, ErrIdentity
	}
	r = WakeResetRequest{Admission: binding, V: 1, Bucket: t.Directory.Bucket, Console: console, ConsoleEnrollment: enrollment, Namespace: namespace, Host: o.Host, ClaimETag: o.ETag, ClaimGen: o.Claim.Gen, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Operation: o.Claim.Op.ID, Phase: o.Claim.Op.Phase, Episode: h.WakeRetry.EpisodeID, Attempt: attempt}
	// Stable identity binds the observed attempt, issuer, claim and round,
	// independent of retries or a console restart. Existing immutable bytes
	// retain their expiry. An archived round stays closed; a later issue for
	// the same claim ETag moves to the next round instead of pinning it.
	var receipt error
	checked := false
	// Rounds only accumulate one per expired lifetime, so the scan stays short.
	for round := uint64(0); ; round++ {
		if err := ctx.Err(); err != nil {
			return r, errors.Join(ErrWakeResetUncertain, receipt, err)
		}
		r.Round = round
		r.ID = wakeResetID(r)
		archived, err := t.readArchive(ctx, r)
		if err == nil {
			// Expiry closed dispatch, not evidence: a root receipt for the
			// tuple is still proof. Without one the outcome stays uncertain and
			// only a fresh round may dispatch again.
			if !checked {
				if receipt = t.VerifyReceipt(ctx, archived); receipt == nil {
					return archived, nil
				}
				checked = true
			}
			continue
		}
		if !errors.Is(err, os.ErrNotExist) {
			return r, err
		}
		existing, err := t.Read(ctx, r.Key())
		if err == nil {
			return existing, nil
		}
		if !errors.Is(err, os.ErrNotExist) {
			return r, err
		}
		r.IssuedAt = w.now()
		r.Expires = r.IssuedAt.Add(WakeResetLifetime)
		if err = t.Authorize(ctx, r); err != nil {
			return r, err
		}
		wire, err := release.SignRootEnvelope(signer, r)
		if err != nil {
			return r, err
		}
		if err = release.CreateExact(ctx, w.raw, r.Key(), wire); err != nil {
			saved, readErr := t.Read(ctx, r.Key())
			if readErr == nil && saved.ID == r.ID {
				return saved, nil
			}
			return r, errors.Join(ErrWakeResetUncertain, err, readErr)
		}
		return r, nil
	}
}

// readArchive returns an archived request only if it is the signed request
// for exactly r's id; anything else at the archive key is an identity error.
func (t WakeResetTransport) readArchive(ctx context.Context, r WakeResetRequest) (WakeResetRequest, error) {
	var a WakeResetRequest
	raw, _, err := t.Store.Get(ctx, wakeResetArchiveKey(r))
	if err != nil {
		return a, err
	}
	if len(raw) > 32<<10 || release.VerifyRootEnvelope(t.Directory.PublicKey, raw, &a) != nil || a.validate() != nil || a.ID != wakeResetID(a) || a.ID != r.ID {
		return a, ErrIdentity
	}
	return a, nil
}
func (w *ControlWriter) CancelWakeReset(ctx context.Context, t WakeResetTransport, signer release.Signer, r WakeResetRequest) error {
	if w == nil || w.ready() != nil || t.Store == nil || r.validate() != nil {
		return ErrContract
	}
	if !w.now().Before(r.Expires) {
		return ErrIdentity
	}
	if _, _, err := t.Store.Get(ctx, wakeResetArchiveKey(r)); err == nil {
		return ErrIdentity
	} else if !errors.Is(err, os.ErrNotExist) {
		return err
	}
	request, done := context.WithDeadline(ctx, minResetDeadline(w.now(), r.Expires))
	defer done()
	if err := t.Authority(request, r.Console, r.ConsoleEnrollment, r.Namespace); err != nil {
		return err
	}
	raw, _ := json.Marshal(r)
	wire, err := release.SignRootEnvelope(signer, WakeResetCancellation{V: 1, RequestSHA256: evidenceDigest(raw)})
	if err != nil {
		return err
	}
	return release.CreateExact(request, w.raw, r.Key()+".cancel", wire)
}

func minResetDeadline(now, expires time.Time) time.Time {
	limit := now.Add(hostbox.RequestLimit / 4)
	if expires.Before(limit) {
		return expires
	}
	return limit
}
func wakeResetArchiveKey(r WakeResetRequest) string {
	return WakeResetArchivePrefix + r.Attempt.Box + "/" + r.ID + ".json"
}

// SweepWakeResets archives signed expired requests before conditionally removing
// them from dispatch. The immutable archive prevents deterministic-ID reuse;
// it is never a stop proof. Root evidence remains independently recoverable.
func (w *ControlWriter) SweepWakeResets(ctx context.Context, t WakeResetTransport) error {
	if w == nil || w.ready() != nil || t.Store == nil || t.Now == nil {
		return ErrContract
	}
	keys, err := t.Store.List(ctx, WakeResetPrefix)
	if err != nil {
		return err
	}
	var errs []error
	for _, key := range keys {
		if !strings.HasSuffix(key, ".json") {
			continue
		}
		r, err := t.Read(ctx, key)
		if err != nil {
			errs = append(errs, err)
			continue
		}
		if !w.now().After(r.Expires.Add(hostbox.RequestLimit)) {
			continue
		}
		raw, etag, err := w.raw.Get(ctx, key)
		if err != nil {
			errs = append(errs, err)
			continue
		}
		var again WakeResetRequest
		if release.VerifyRootEnvelope(w.pub, raw, &again) != nil || again.ID != r.ID || again.ID != wakeResetID(again) || again.validate() != nil || !again.Expires.Equal(r.Expires) {
			errs = append(errs, ErrIdentity)
			continue
		}
		if err = release.CreateExact(ctx, w.raw, wakeResetArchiveKey(r), raw); err != nil {
			errs = append(errs, err)
			continue
		}
		// Keep the request discoverable if cancellation cleanup fails. Archival
		// closes issuance first; cancellation is removed before dispatch.
		if cancelled, cancelErr := t.cancelled(ctx, r); cancelErr != nil {
			errs = append(errs, cancelErr)
			continue
		} else if cancelled {
			_, tag, cancelErr := w.raw.Get(ctx, key+".cancel")
			if cancelErr == nil {
				cancelErr = release.DeleteIfMatch(ctx, w.raw, key+".cancel", tag)
			}
			if cancelErr != nil {
				errs = append(errs, cancelErr)
				continue
			}
		}
		if err = release.DeleteIfMatch(ctx, w.raw, key, etag); err != nil {
			errs = append(errs, err)
		}
	}
	return errors.Join(errs...)
}