package hostclaim

import (
	"fmt"
	"github.com/carlosframework/platform/internal/regionname"
	"reflect"
	"strconv"
)

const RootGeneration = "@root"

type Admission struct {
	ObservedOpID, ObservedPhase string // Empty values are compared literally.
	ObservedSession             uint64
	ObservedGeneration          string
	Box                         string
	Mode                        string
	NewGeneration               string
	PredecessorCertified        bool
	RemoteMaxTXID               uint64
	KeepStamp                   bool
	Coverage                    string
	Region                      string
	PredecessorRegion           string
	AdditionalWritesLost        []WritesLost
	LossID                      string
	lossVerified                bool
	fenceVerified               bool
}

// AdmitWriter is §6.4a's writer admission CAS within the current operation.
// Warm admission keeps the installation; cold admission installs a fresh
// generation and stamp. The caller checks the evidence required for warm mode
// and supplies PredecessorCertified.
func AdmitWriter(in Admission) func(*Claim) error {
	return func(c *Claim) error {
		if in.Region == "" {
			return fmt.Errorf("%w: an admission names its region", ErrWrongState)
		}
		if c.Holder == nil || c.Op.Exec == "" || c.Op.Exec != in.Box {
			return fmt.Errorf("%w: only the executor may admit a writer", ErrWrongState)
		}
		if c.Op.ID != in.ObservedOpID || c.Op.Phase != in.ObservedPhase {
			return fmt.Errorf("%w: operation or phase moved since it was observed", ErrWrongState)
		}
		switch c.Op.Kind {
		case KindCreating, KindDeleting, KindParking, KindPurging, KindPlacing:
			return fmt.Errorf("%w: op kind %q cannot admit a writer", ErrWrongState, c.Op.Kind)
		case KindRelocating:
			if err := relocatingAdmission(c, in); err != nil {
				return err
			}
			if err := relocationProbeOpening(c, in); err != nil {
				return err
			}
		}
		id, err := OperationLossID(c)
		if err != nil {
			return err
		}
		if id != "" && (in.Coverage != "" || (!OperationLossConsumed(c, id) && (!in.lossVerified || in.LossID != id))) {
			return ErrLossRecord
		}

		if c.Op.Kind == KindRestoring && c.Op.Journal["fence."+c.Op.Intent["target"]] != "" && in.Coverage != "" {
			return ErrFence
		}
		if _, pending := PendingMoveFenceLoss(c); pending && !in.fenceVerified {
			return ErrFence
		}
		h := c.Holder
		if in.LossID != "" && !in.lossVerified {
			return fmt.Errorf("%w: declaration-bound admission requires verified durable declaration", ErrWrongState)
		}
		if in.LossID != "" {
			for _, prior := range h.Admissions {
				if prior.LossID != in.LossID {
					continue
				}
				if prior.Box == in.Box && prior.Mode == in.Mode && prior.Generation == in.NewGeneration &&
					reflect.DeepEqual(prior.AdditionalWritesLost, in.AdditionalWritesLost) {
					return ErrLossAlreadyApplied
				}
				return fmt.Errorf("%w: declaration id already used by another admission", ErrWrongState)
			}
		}
		if c.Op.Kind == KindServing && h.WakeRetry != nil {
			r := h.WakeRetry
			if r.State != "attempting" || r.Attempt == nil || r.Attempt.Session != 0 || r.Attempt.Box != in.Box {
				return fmt.Errorf("%w: serving writer admission has no matching retry reservation", ErrWrongState)
			}
		}
		if h.Session != in.ObservedSession || h.Generation != in.ObservedGeneration {
			return fmt.Errorf("%w: writer session or generation moved since it was observed", ErrWrongState)
		}
		if in.KeepStamp && (in.Mode != "cold" || h.Stamp <= h.Session) {
			return fmt.Errorf("%w: keeping a stamp requires cold admission before its first writer", ErrWrongState)
		}
		switch in.Mode {
		case "warm":
			if !in.PredecessorCertified || in.NewGeneration != "" {
				return fmt.Errorf("%w: a warm admission requires a certificate and keeps the generation", ErrWrongState)
			}
		case "cold":
			if in.NewGeneration == "" || in.NewGeneration == h.Generation {
				return fmt.Errorf("%w: a cold recovery needs a new generation", ErrWrongState)
			}
		default:
			return fmt.Errorf("%w: unknown writer admission mode %q", ErrWrongState, in.Mode)
		}

		switch in.Coverage {
		case "", "sealed", "clean-return":
		default:
			return ErrWrongState
		}
		predecessor := AdmissionRecord{Session: h.Session, Box: h.WriterBox, Generation: h.Generation, Region: in.PredecessorRegion}
		if len(h.Admissions) > 0 {
			predecessor = h.Admissions[len(h.Admissions)-1]
		}
		if in.Mode == "warm" && !regionname.SameRegion(c.Op.Intent["home_region"], in.Region, predecessor.Region) {
			return fmt.Errorf("%w: a warm admission stays in its predecessor's region", ErrWrongState)
		}
		type writerSession struct {
			box     string
			session uint64
		}
		seen := make(map[writerSession]bool, len(in.AdditionalWritesLost)+1)
		if in.Mode == "cold" && !in.PredecessorCertified && in.Coverage == "" {
			seen[writerSession{predecessor.Box, predecessor.Session}] = true
		}
		for _, loss := range in.AdditionalWritesLost {
			writer := writerSession{loss.Box, loss.Session}
			if loss.Box == "" || (loss.Session == 0) != (loss.Generation == "") || seen[writer] {
				return ErrWrongState
			}
			seen[writer] = true
		}
		entry := AdmissionRecord{
			Session: c.Gen + 1, Box: in.Box, Generation: h.Generation,
			LossID:      in.LossID,
			Predecessor: predecessor.Generation, Mode: in.Mode, Coverage: in.Coverage,
			Region: in.Region, PredecessorRegion: predecessor.Region,
			Covered:              in.Mode == "warm",
			AdditionalWritesLost: append([]WritesLost(nil), in.AdditionalWritesLost...),
		}
		for _, ob := range c.Obligations {
			// Only a park of the holder's whole location retires the
			// predecessor; a scoped park of another location (an old
			// incarnation's residue, a staging prefix) does not.
			if ob.Kind == "park-replica" && regionname.SameRegion(c.Op.Intent["home_region"], ob.Region, entry.PredecessorRegion) &&
				(ob.Park == nil || ob.Park.Location == h.Location) {
				entry.Covered = true
				break
			}
		}
		if in.Mode == "cold" {
			entry.Generation = in.NewGeneration
			if !in.KeepStamp {
				h.Stamp = entry.Session
			}
			// A predecessor with no writer (a created incarnation's first
			// admission: writer_box "") acknowledged nothing, so there is
			// nothing to declare; every declaration names a box (§6.4a).
			if !in.PredecessorCertified && in.Coverage == "" && predecessor.Box != "" {
				entry.WritesLost = &WritesLost{Box: predecessor.Box, Generation: predecessor.Generation, Session: predecessor.Session, RemoteMaxTXID: in.RemoteMaxTXID}
			}
		}
		if err := pendingOrigin(c, entry); err != nil {
			return err
		}
		h.Admissions = append(h.Admissions, entry)
		for len(h.Admissions) > 8 && h.Admissions[0].Covered && admissionLossArchived(h.Admissions[0]) {
			h.Admissions = h.Admissions[1:]
		}
		h.Session, h.Generation, h.WriterBox = entry.Session, entry.Generation, in.Box
		if in.lossVerified && id != "" {
			setJournal(c, "loss_admitted."+id, strconv.FormatUint(entry.Session, 10))
		}
		return nil
	}
}

func PublishGeneration(observedOpID, observedPhase string, observedSession uint64, publishedGeneration, region string, newID func() string) func(*Claim) error {
	return func(c *Claim) error {
		if c.Holder == nil || c.Op.ID != observedOpID || c.Op.Phase != observedPhase ||
			c.Holder.Session != observedSession || c.Holder.Generation != publishedGeneration || region == "" {
			return ErrWrongState
		}
		switch c.Op.Kind {
		case KindCreating, KindDeleting, KindParking, KindPurging, KindPlacing:
			return ErrWrongState
		case KindRelocating:
			// §6.4a item 5: recognizing the kind grants nothing; only a
			// phase where a relocating writer session lives may publish.
			if !SessionPhase(KindRelocating, c.Op.Phase) {
				return ErrWrongState
			}
		}

		h := c.Holder
		admitted := false
		for _, record := range h.Admissions {
			if record.Generation == publishedGeneration {
				admitted = true
				break
			}
		}
		if !admitted {
			return ErrWrongState
		}
		activeGeneration := publishedGeneration
		if activeGeneration == "" {
			activeGeneration = RootGeneration
		}
		op := c.Op.ID
		if op == "" {
			op = fmt.Sprintf("session:%d", h.Session)
		}
		for i := range h.Admissions {
			record := &h.Admissions[i]
			if record.Covered {
				continue
			}
			predecessor := record.Predecessor
			if predecessor == "" {
				predecessor = RootGeneration
			}
			if predecessor == activeGeneration {
				continue
			}
			if record.PredecessorRegion == "" {
				record.Covered = true
				continue
			}
			if !regionname.SameRegion(c.Op.Intent["home_region"], record.PredecessorRegion, region) {
				continue
			}
			queued := false
			for _, ob := range c.Obligations {
				if ob.Kind == "park-generation" && regionname.SameRegion(c.Op.Intent["home_region"], ob.Region, region) && ob.Generation == predecessor {
					queued = true
					break
				}
			}
			if !queued {
				c.Obligations = append(c.Obligations, Obligation{
					ID: newID(), Op: op, Kind: "park-generation", Region: region,
					Generation: predecessor, Retirement: newID(), Owner: "region:" + region,
				})
			}
			record.Covered = true
		}
		return nil
	}
}

// CoverRegionPredecessors covers predecessors retired by a region's park of
// the whole location. Callers compose it into the CAS that queues that park
// (a committed move's source cleanup). It queues no chores itself.
func CoverRegionPredecessors(observedOpID, observedPhase string, region string) func(*Claim) error {
	return func(c *Claim) error {
		if c.Holder == nil || c.Op.ID != observedOpID || c.Op.Phase != observedPhase || region == "" {
			return ErrWrongState
		}
		for i := range c.Holder.Admissions {
			record := &c.Holder.Admissions[i]
			if !record.Covered && regionname.SameRegion(c.Op.Intent["home_region"], record.PredecessorRegion, region) {
				record.Covered = true
			}
		}
		return nil
	}
}

// CoverParkedPredecessors covers only the predecessors retired by a sweep.
// Callers compose it into the CAS that queues the sweep and the CAS that
// acknowledges it. It queues no chores itself.
func CoverParkedPredecessors(observedOpID, observedPhase string, region string, generations []string) func(*Claim) error {
	parked := make(map[string]bool, len(generations))
	for _, generation := range generations {
		parked[generation] = true
	}
	return func(c *Claim) error {
		if c.Holder == nil || c.Op.ID != observedOpID || c.Op.Phase != observedPhase || region == "" {
			return ErrWrongState
		}
		for i := range c.Holder.Admissions {
			record := &c.Holder.Admissions[i]
			predecessor := record.Predecessor
			if predecessor == "" {
				predecessor = RootGeneration
			}
			if !record.Covered && regionname.SameRegion(c.Op.Intent["home_region"], record.PredecessorRegion, region) && parked[predecessor] {
				record.Covered = true
			}
		}
		return nil
	}
}

// relocatingAdmission is §4's relocating phase table for a writer admission,
// as claim Plan 11 accepted it (R1): only op.exec's cold probe, in probing,
// of an operation that is not stuck. Every other phase admits no writer.
// The source's admission in starting or releasing needs its admission-fence
// authority, which no claim code supplies yet, so it stays refused; this
// check only ever adds refusals to AdmitWriter's own.
func relocatingAdmission(c *Claim, in Admission) error {
	if c.Op.Journal["stuck"] != "" {
		return fmt.Errorf("%w: relocation is stuck", ErrWrongState)
	}
	if c.Op.Phase != PhaseRelocatingProbing {
		return fmt.Errorf("%w: relocating/%s admits no writer", ErrWrongState, c.Op.Phase)
	}
	if in.Mode != "cold" || in.Box != c.Op.Exec || in.Coverage != "" {
		return fmt.Errorf("%w: a relocation probe is op.exec's cold admission", ErrWrongState)
	}
	return nil
}

// relocationProbeOpening is region HA's prepared generation P on top of
// relocatingAdmission (§5.8, L9, rev 12), and it only adds refusals: the
// probe admission that opens P installs P itself, and P is opened at most
// once. The opening is journaled with the admission, so it outlives the
// admission list's trim. A certified opening must be backed by this
// operation's journaled release, naming the exact predecessor session, and
// a pin reaching that release's final position: that is the only way a
// relocation probe certifies its predecessor (claim Plan 11 grants no
// relocation coverage of its own, so the release is expressed as the
// generic PredecessorCertified, never as a new coverage value).
func relocationProbeOpening(c *Claim, in Admission) error {
	p, opened := c.Op.Journal["prepared"], preparedOpened(c)
	if p == "" {
		// No prepared generation journaled: main's cold probe as claim Plan
		// 11 accepted it (region HA's ToProbing always journals P first).
		// With no P there is no release to back a certification, so a
		// certified relocating probe is refused.
		if in.PredecessorCertified {
			return fmt.Errorf("%w: a relocating probe certifies only P's opening", ErrWrongState)
		}
		return nil
	}
	if (!opened && in.NewGeneration != p) || (opened && in.NewGeneration == p) {
		return fmt.Errorf("%w: the first probe opens P, and only it", ErrWrongState)
	}
	if opened {
		if in.PredecessorCertified {
			// The release certifies only the session P's opening succeeds.
			return fmt.Errorf("%w: only P's opening certifies its predecessor", ErrWrongState)
		}
		return nil
	}
	if in.PredecessorCertified {
		var r Release
		var pin RecoveryPin
		if !readJournal(c.Op, "release", &r) || !readJournal(c.Op, "pin", &pin) || pin.MaxTXID < r.FinalTXID || pin.FilesSeq < r.ManifestSeq {
			return fmt.Errorf("%w: a certified opening needs this operation's release and a pin reaching it", ErrWrongState)
		}
		h := c.Holder
		predecessor := AdmissionRecord{Session: h.Session, Box: h.WriterBox, Generation: h.Generation}
		if len(h.Admissions) > 0 {
			predecessor = h.Admissions[len(h.Admissions)-1]
		}
		if predecessor.Box != r.ReleasedBy || predecessor.Session != r.Session || predecessor.Generation != r.Generation {
			return fmt.Errorf("%w: a certified opening names another predecessor", ErrWrongState)
		}
	}
	ensureJournal(c)
	c.Op.Journal["prepared_opened"] = p
	return nil
}