package host

import (
	"context"
	"encoding/json"
	"fmt"
	"reflect"
	"strconv"

	"github.com/carlosframework/platform/internal/genselect"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostops"
	"github.com/carlosframework/platform/internal/park"
)

// The loss and fence rules of the source's recovery (moves-after-Plan-12
// Task 4: owed (a) and (b), Plan 4 Task 10's box half; rulings Q6, Q7's
// loss half and Q5 as platform-1e answered it on #236).
//
// A declared or fenced return (Q6). The claim journals a declaration of the
// target's loss under the op (hostclaim.OperationLossID) or a termination
// fence of the target (fence.<target>); the record's rollback cites it
// (Rollback.LossRef: the declaration's record key, or the fence record's
// key). Such a return is clean for want of a target and covered by nothing:
// its probe asks no coverage, for its first admission and every later one,
// since the declaration and the fence stay the op's after the first
// admission consumed them. A record and a claim that disagree, by presence
// or by identity, wait.
//
// The loss baseline validator (Q5, owed (b)). The claim's writer admits the
// first recovery admission under a pending loss through admitLoss or
// admitFencedLoss (hostbox/writer_loss.go), which call LossSources with the
// frozen record. No cross-region read exists (the box role reads only its
// own data bucket; platform-1e): each frozen source is vouched for from the
// maxima already recorded.
//
//   - replica:<region>/<gen>, a target session since the cut: the loss
//     boundary is the highest TXID of the tuple's own generation that the
//     restore contains (platform-1e's loss-boundary rule), never a number
//     carried across generations: a fresh target generation restarts its
//     TXIDs at [1, 1] (spec §5.1). It is genselect.GenerationMax over the
//     restore plan: a generation the plan does not contain (every target
//     generation behind a J4 return, which restores src_gen) is 0, every
//     write in it lost. Every tuple is kept, however many share 0. The
//     claim compares the declared maximum with this validated one
//     (hostclaim/lost.go AdmitWriterWithLoss).
//   - seal:<move>, the source's cut session when the seal did not cover the
//     source's replica: the cut seal the claim journals, its maximum N.
//
// The restore point is checked too: the source baseline's maximum, read
// from its own selected files by genselect.GenerationMax, the rule the
// claim's writer reads. A restore point that is not the recorded source
// baseline, or whose files do not reach its maximum, is a wait
// (hostbox.ErrChanged), within the probe's nine attempts; a frozen source
// this recovery cannot be the owner of is a refusal (hostclaim.ErrLossRecord).
//
// AdmissionEvidence's loss contract (Q7). The writer substitutes the
// validated predecessor's maximum for RemoteMaxTXID and the other validated
// sources for AdditionalWritesLost before it calls AdmissionEvidence. The
// recovery owners accept exactly that substitution, from the same
// LossSources call for the same observation, re-validated; every other
// check of the recovery's evidence, the lineage link and the source
// baseline's record before the CAS included, is unchanged.

// claimLossRefs are the keys of the target's loss the claim journals under
// the op: the declaration's record key, and the fence record's key.
func claimLossRefs(c *hostclaim.Claim) (map[string]bool, error) {
	refs := map[string]bool{}
	id, err := hostclaim.OperationLossID(c)
	if err != nil {
		return nil, err
	}
	if id != "" {
		key, err := hostclaim.LossRecordKey(id)
		if err != nil {
			return nil, err
		}
		refs[key] = true
	}
	if raw := c.Op.Journal["fence."+c.Op.Intent["target"]]; raw != "" {
		var ref hostclaim.FenceRef
		if err := json.Unmarshal([]byte(raw), &ref); err != nil || ref.Key == "" {
			return nil, fmt.Errorf("%w: the target's fence does not read", hostclaim.ErrFence)
		}
		refs[ref.Key] = true
	}
	return refs, nil
}

// recoveryDeclared reports whether a return is a declared or fenced one
// (Q6): the claim journals the target's loss and the record's rollback
// cites it (lossRef). A disagreement, by presence or by identity, is
// hostbox.ErrChanged: the probe waits for the two to agree.
func recoveryDeclared(c *hostclaim.Claim, lossRef string) (bool, error) {
	refs, err := claimLossRefs(c)
	if err != nil {
		return false, fmt.Errorf("%w: the claim's loss of the target: %v", hostbox.ErrChanged, err)
	}
	if len(refs) == 0 && lossRef == "" {
		return false, nil
	}
	if !refs[lossRef] {
		return false, fmt.Errorf("%w: the record and the claim disagree on the target's loss", hostbox.ErrChanged)
	}
	return true, nil
}

// lossValidation is one LossSources call's answer, kept for the
// AdmissionEvidence call of the same admission.
type lossValidation struct {
	etag   string
	record hostclaim.LossRecord
	valid  []hostclaim.ValidatedWriteSource
}

// LossSources is the moves loss baseline validator
// (hostbox.LossAdmissionSources): the recovery owners' movesLossSources
// for a recovery, refused for every other kind.
func (m *movesProbeOwners) LossSources(ctx context.Context, o hostclaim.Observed, loss hostclaim.LossRecord, in hostclaim.Admission, baseline *genselect.Plan) ([]hostclaim.ValidatedWriteSource, error) {
	if !recoveryKind(o) {
		return nil, hostbox.ErrUnsupported
	}
	r := m.recovery()
	valid, err := r.movesLossSources(ctx, o, loss, baseline)
	if err != nil {
		return nil, err
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.lossValidated == nil {
		m.lossValidated = map[string]lossValidation{}
	}
	// One per host: a later call for the host replaces an earlier one.
	m.lossValidated[o.Host] = lossValidation{etag: o.ETag, record: loss, valid: append([]hostclaim.ValidatedWriteSource(nil), valid...)}
	return valid, nil
}

// movesLossSources vouches for each of loss's frozen sources (see the file
// comment), for the recovery's first admission while loss is the pending
// declaration or the pending fence of this op's target.
func (r movesRecoveryOwners) movesLossSources(ctx context.Context, o hostclaim.Observed, loss hostclaim.LossRecord, baseline *genselect.Plan) ([]hostclaim.ValidatedWriteSource, error) {
	v, err := r.recovery(ctx, o)
	if err != nil {
		return nil, err
	}
	c := o.Claim
	if v.kind != hostclaim.KindRestoring || v.dirty || !v.uncovered || !v.first {
		return nil, fmt.Errorf("%w: only a declared or fenced return's first admission declares a loss", hostclaim.ErrLossRecord)
	}
	if err := r.pendingLoss(c, v, loss); err != nil {
		return nil, err
	}
	if err := r.frozenMaximum(v, baseline); err != nil {
		return nil, err
	}
	cut, err := strconv.ParseUint(v.cut, 10, 64)
	if err != nil {
		return nil, fmt.Errorf("%w: the cut's session is not journalled", hostbox.ErrChanged)
	}
	out := make([]hostclaim.ValidatedWriteSource, 0, len(loss.WriteSources))
	for _, s := range loss.WriteSources {
		max, err := r.sourceMaximum(c, v, cut, baseline, s)
		if err != nil {
			return nil, err
		}
		out = append(out, hostclaim.ValidatedWriteSource{Reason: s.Reason, Box: s.Box, Generation: s.Generation, Session: s.Session, Region: s.Region, Baseline: s.Baseline, RemoteMaxTXID: max})
	}
	return out, nil
}

// pendingLoss requires loss to be what the claim's writer admits next: the
// fence's derived record while the consumed fence's target writer awaits
// its first admission, or else the op's unconsumed declaration; and either
// to be this op's, this holder's, of this op's target, and the one the
// record's rollback cites.
func (r movesRecoveryOwners) pendingLoss(c *hostclaim.Claim, v recoveryProbe, loss hostclaim.LossRecord) error {
	refused := func(why string) error { return fmt.Errorf("%w: %s", hostclaim.ErrLossRecord, why) }
	if loss.OpID != v.op || loss.Box != c.Op.Intent["target"] || loss.HolderIncarnation != v.holder.Incarnation {
		return refused("the loss is not this op's target's under this holder")
	}
	if fenced, pending := hostclaim.PendingMoveFenceLoss(c); pending {
		ref, _ := hostclaim.ConsumedMoveTargetFence(c)
		if loss.ID != fenced.ID || loss.Automatic == nil || ref.Key != v.rec.Move.Rollback.LossRef {
			return refused("the loss is not the pending fence the record cites")
		}
		return nil
	}
	id, err := hostclaim.OperationLossID(c)
	if err != nil {
		return err
	}
	if id == "" || hostclaim.OperationLossConsumed(c, id) || loss.ID != id {
		return refused("the loss is not the op's pending declaration")
	}
	if key, err := hostclaim.LossRecordKey(id); err != nil || key != v.rec.Move.Rollback.LossRef {
		return refused("the record does not cite the pending declaration")
	}
	return nil
}

// frozenMaximum checks the admission's restore point: the recovery's source
// baseline (src_gen, as the first admission selects it, and the recorded
// one once recorded), whose selected files of its own generation reach its
// maximum (genselect.GenerationMax over the plan, never a listing). A plan
// that disagrees waits.
func (r movesRecoveryOwners) frozenMaximum(v recoveryProbe, baseline *genselect.Plan) error {
	if baseline == nil || baseline.Generation != selectedGeneration(v.srcGen) {
		return fmt.Errorf("%w: the restore point is not the source baseline", hostbox.ErrChanged)
	}
	recorded, ok, err := r.recordedSourceBaseline(v.rec.Host, v)
	if err != nil {
		return err
	}
	if ok && (recorded.Generation != baseline.Generation || recorded.MaxTXID != baseline.MaxTXID) {
		return fmt.Errorf("%w: the restore point is not the recorded source baseline", hostbox.ErrChanged)
	}
	max, err := genselect.GenerationMax(baseline, v.holder.Location, baseline.Generation)
	if err != nil {
		return fmt.Errorf("%w: the source baseline's plan: %v", hostbox.ErrChanged, err)
	}
	if max != baseline.MaxTXID {
		return fmt.Errorf("%w: the source baseline's files reach %d, not the selection's %d", hostbox.ErrChanged, max, baseline.MaxTXID)
	}
	return nil
}

// sourceMaximum vouches for one frozen source (see the file comment) and
// returns its maximum.
func (r movesRecoveryOwners) sourceMaximum(c *hostclaim.Claim, v recoveryProbe, cut uint64, baseline *genselect.Plan, s hostclaim.WriteSource) (uint64, error) {
	m := r.m
	refused := func(why string) (uint64, error) {
		return 0, fmt.Errorf("%w: frozen source %s/%s/%d: %s", hostclaim.ErrLossRecord, s.Box, s.Generation, s.Session, why)
	}
	kind, region, identity, err := hostops.ParseMoveLossBaseline(s.Baseline)
	if err != nil {
		return refused("the baseline does not parse")
	}
	switch kind {
	case "replica":
		if s.Reason != "target" || s.Box != c.Op.Intent["target"] || s.Generation == "" || identity != s.Generation || s.Session <= cut {
			return refused("not a target session since the cut")
		}
		if !park.SameRegion(v.homeRegion, region, s.Region) || !park.SameRegion(v.homeRegion, s.Region, c.Op.Intent["to"]) {
			return refused("not the target region's replica")
		}
		// The highest TXID of the tuple's own generation the restore
		// contains: 0 for a generation outside it.
		max, err := genselect.GenerationMax(baseline, v.holder.Location, s.Generation)
		if err != nil {
			return 0, fmt.Errorf("%w: the restore plan: %v", hostbox.ErrChanged, err)
		}
		return max, nil
	case "seal":
		if s.Reason != "source-cut" || identity != v.op || s.Box != m.Box || selectedGeneration(s.Generation) != selectedGeneration(v.srcGen) || s.Session != cut ||
			!park.SameRegion(v.homeRegion, s.Region, m.Region) {
			return refused("not the source's cut session")
		}
		seal, err := journalledSeal(c)
		if err != nil {
			return 0, fmt.Errorf("%w: the cut seal: %v", hostbox.ErrChanged, err)
		}
		if seal.SrcCovered {
			return refused("the cut seal covered the source's replica")
		}
		return seal.N, nil
	}
	return refused("an unknown baseline")
}

// lossEvidence is AdmissionEvidence's loss contract: the admission is the
// writer's substitution of this observation's validated sources, which are
// validated again. It consumes the validation.
func (r movesRecoveryOwners) lossEvidence(ctx context.Context, o hostclaim.Observed, in hostclaim.Admission, baseline *genselect.Plan) error {
	m := r.m
	m.mu.Lock()
	kept, ok := m.lossValidated[o.Host]
	delete(m.lossValidated, o.Host)
	m.mu.Unlock()
	if !ok || kept.etag != o.ETag {
		return fmt.Errorf("%w: no loss sources were validated for this observation", hostbox.ErrChanged)
	}
	valid, err := r.movesLossSources(ctx, o, kept.record, baseline)
	if err != nil {
		return err
	}
	if len(valid) != len(kept.valid) || (len(valid) != 0 && !reflect.DeepEqual(valid, kept.valid)) {
		return fmt.Errorf("%w: the loss sources changed", hostbox.ErrChanged)
	}
	// The writer's substitution (writer_loss.go): the predecessor's
	// validated maximum, the rest as additional declarations; a fence
	// admits its one frozen target writer and nothing more.
	h := o.Claim.Holder
	p := hostclaim.AdmissionRecord{Box: h.WriterBox, Session: h.Session, Generation: h.Generation}
	if a := h.Admissions; len(a) > 0 {
		p = a[len(a)-1]
	}
	max := baseline.MaxTXID
	var more []hostclaim.WritesLost
	if _, fenced := hostclaim.PendingMoveFenceLoss(o.Claim); fenced {
		if len(valid) != 1 {
			return fmt.Errorf("%w: a fence freezes one target writer", hostclaim.ErrFence)
		}
		max = valid[0].RemoteMaxTXID
	} else {
		for _, s := range valid {
			if s.Box == p.Box && s.Session == p.Session && s.Generation == p.Generation {
				max = s.RemoteMaxTXID
			} else {
				more = append(more, hostclaim.WritesLost{Box: s.Box, Generation: s.Generation, Session: s.Session, RemoteMaxTXID: s.RemoteMaxTXID})
			}
		}
	}
	if in.RemoteMaxTXID != max || len(in.AdditionalWritesLost) != len(more) || (len(more) != 0 && !reflect.DeepEqual(in.AdditionalWritesLost, more)) {
		return fmt.Errorf("%w: the admission is not the validated loss sources' substitution", hostbox.ErrChanged)
	}
	return nil
}