package host

import (
	"context"
	"errors"
	"fmt"
	"log"

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// The commit and the open on the target (moves spec §7.5, §7.9 steps 1 and
// 2; Plan 3 Task 7; preflight H6 to H9).
//
// probed: the commit is the record's CAS (provision.MoveCommit), which
// refuses a recorded rollback and a commit after the landing's deadline
// (H8); that refusal is the commit-deadline rollback mark, converted at the
// pass's one rollback site. Then the committed arm runs in the same pass.
//
// The committed arm (H9), routed for the target by the post-flip participant
// rule, runs every pass while the record is committed:
//   - the claim half: a claim lagging at probing is stepped to probed first
//     (platform-1e m-2), then hostclaim.Commit (probed only) under
//     KeepWriterSession, so the admitted probe session is the serving one
//     (H3); idempotent once committed. The source's location park is scoped
//     from the flip (H7), the staging lane from the seal's bound;
//   - the open, conditional: under the host's hold lock, with the record and
//     the claim re-read there and both committed, the probe's live activator
//     entry is marked open (MarkProbeOpen), and only then is the hold
//     released (ReleaseHeld), so an edge reload in between keeps the entry
//     (Task 5's open race). The entry from the probe is the serving process:
//     it is never restarted;
//   - MoveOpened, the first time only;
//   - the staging park and the close (§7.9 steps 4 and 5, Plan 5 Task 3,
//     movesparks.go): the lane's park through laneParkTarget at
//     landing/committed, then closeMove. A claim that already records this
//     move's release (a crash between the close's two halves) skips the
//     rest and only clears the record.
//
// Unarmed in production with the rest of the landing: landingArmed fails
// there until the wake-source gate opens (movesWakeSourceRegistered, Plan 5
// Task 9b), and the open needs
// ReleaseHeld.

// probed commits the landing: the record's CAS, then the committed arm.
func (l *landingRun) probed(ctx context.Context) (bool, error) {
	obs, err := l.syncClaimPhase(ctx)
	if err != nil {
		return false, err
	}
	if err := l.requirePhase(obs, provision.LandProbed); err != nil {
		return false, err
	}
	if err := l.fault("before-commit"); err != nil {
		return false, err
	}
	var pre provision.Record
	fresh, err := provision.SetRecordMove(ctx, l.s, l.rec.Account, l.rec.App, l.rec.Host, observedMove(&pre, provision.MoveCommit(l.g, l.cfg().Now())))
	if err == nil {
		l.w.logTransition(ctx, l.s, pre, *fresh)
	}
	switch {
	case errors.Is(err, provision.ErrCutDeadline):
		log.Printf("moves: %s: the commit of landing %s came after its deadline", l.rec.Host, l.g.ID)
		return l.markRollback(ctx, provision.RollbackCommitDeadline)
	case errors.Is(err, provision.ErrMoveConflict):
		// A replay finds committed; anything else is another writer's.
		if fresh, err = provision.GetRecord(ctx, l.s, l.rec.Account, l.rec.App, l.rec.Host); err != nil {
			return false, fmt.Errorf("re-read the record: %w", err)
		}
		if fresh == nil || fresh.Move == nil || fresh.Move.ID != l.g.ID || fresh.Move.State != provision.MoveCommitted {
			log.Printf("moves: %s: the commit of landing %s is refused; the record moved on", l.rec.Host, l.g.ID)
			return false, nil
		}
	case err != nil:
		return false, fmt.Errorf("the commit: %w", err)
	}
	log.Printf("moves: %s: landing %s is committed", l.rec.Host, l.g.ID)
	if err := l.fault("after-record-committed"); err != nil {
		return true, err
	}
	return true, l.w.committedTarget(ctx, l.s, *fresh)
}

// committedTarget is the committed arm on the target (see the file comment).
func (w *movesWorker) committedTarget(ctx context.Context, s release.Store, rec provision.Record) error {
	cfg := w.cfg
	m := rec.Move
	if m == nil || m.Legacy() || m.State != provision.MoveCommitted || m.To != cfg.Region || m.From == cfg.Region {
		return nil
	}
	if !landingArmed(cfg) {
		w.notice(rec.Host+"#committed", "moves: %s: the open of %s waits: this box's landing is not armed", rec.Host, m.ID)
		return nil
	}
	l := &landingRun{w: w, s: s, rec: rec, g: provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation}, guard: cfg.Guard}
	obs, err := hostclaim.Read(ctx, s, rec.Host)
	if err != nil {
		return fmt.Errorf("read the host claim: %w", err)
	}
	if !obs.Absent && hostclaim.RequireReleasedMove(m.ID)(obs.Claim) == nil {
		// A binary that closed claim first released the claim and stopped
		// before the record's half: only that is left (closeMove).
		return w.closeMove(ctx, s, rec, l.fault)
	}
	if err := l.commitClaim(ctx); err != nil {
		return err
	}
	if err := l.fault("after-claim-committed"); err != nil {
		return err
	}
	if cfg.ReleaseHeld == nil {
		w.notice(rec.Host+"#open", "moves: %s: the open of %s waits: this box cannot release the hold", rec.Host, m.ID)
	} else {
		stamped, err := l.open(ctx)
		if err != nil {
			return err
		}
		if stamped {
			log.Printf("moves: %s: landing %s is open", rec.Host, m.ID)
			nudgeSerial(ctx, s)
		}
	}
	// §7.9 step 4: the lane's park the commit queued (movesparks.go).
	if err := w.laneParkTarget(ctx, s, rec, hostclaim.KindLanding, hostclaim.PhaseCommitted); err != nil {
		return err
	}
	// §7.9 step 5: the close, once both stamps are in and nothing is owed.
	return w.closeMove(ctx, s, rec, l.fault)
}

// commitClaim is the claim half of the commit (§7.5): written by whoever
// sees the record ahead. A claim already committed is done.
func (l *landingRun) commitClaim(ctx context.Context) error {
	cfg := l.cfg()
	id := l.g.ID
	obs, err := l.claim(ctx)
	if err != nil {
		return err
	}
	op := obs.Claim.Op
	if op.Kind != hostclaim.KindLanding {
		return fmt.Errorf("move %s: the claim is %s/%q, not landing", id, op.Kind, op.Phase)
	}
	if op.Phase == hostclaim.PhaseLandingCommitted {
		return nil
	}
	if cfg.Box == "" || op.Exec != cfg.Box {
		return fmt.Errorf("move %s: the claim's executor is %q, not this box", id, op.Exec)
	}
	if op.Phase == hostclaim.PhaseLandingProbing {
		// m-2: the commit follows probed; a claim lagging behind the
		// record's probed is stepped there on the same session.
		if obs, err = hostclaim.Update(ctx, l.s, obs, hostclaim.KeepWriterSession(id, hostclaim.SetMovePhase(id, hostclaim.KindLanding, hostclaim.PhaseLandingProbing, hostclaim.PhaseLandingProbed))); err != nil {
			return fmt.Errorf("move %s: the claim's probed before the commit: %w", id, err)
		}
	}
	source, staging, err := l.commitScopes(obs.Claim)
	if err != nil {
		return err
	}
	if _, err := hostclaim.Update(ctx, l.s, obs, hostclaim.KeepWriterSession(id, hostclaim.Commit(id, park.PlacementRegion(l.rec.Move.From), obs.Claim.Op.Intent["source"], cfg.parkRegion(), source, staging, obligationID))); err != nil {
		return fmt.Errorf("move %s: the claim's commit: %w", id, err)
	}
	log.Printf("moves: %s: the claim of landing %s is committed", l.rec.Host, id)
	return nil
}

// commitScopes are the parks the commit queues: moves.CommitScopes, the one
// builder the console's pass uses too (Plan 5 Task 3, K9), so the box and
// the console write the same obligations. The source region's location is
// scoped from the flip (H7), the staging lane to the seal's bound.
func (l *landingRun) commitScopes(c *hostclaim.Claim) (source, staging hostclaim.ParkScope, err error) {
	if l.rec.Move == nil {
		return source, staging, errors.New("the commit's parks need the record's move")
	}
	return moves.CommitScopes(c, l.rec.Move.FlippedAt)
}

// open is the conditional open, proved (fix round 1, finding 3): under the
// host's hold lock, with the record and the claim re-read and both
// committed, the probe's entry is marked open, the hold released, and then
// opened_at stamped, in that order. A replay is only a row already ready and
// unheld; a release that changed nothing is re-read and accepted only when
// the row is ready and unheld; anything else refuses and stamps nothing.
// stamped reports that this call wrote opened_at.
func (l *landingRun) open(ctx context.Context) (stamped bool, err error) {
	cfg := l.cfg()
	host, id := l.rec.Host, l.g.ID
	provision.WithHostHoldLock(host, func() {
		var fresh *provision.Record
		if fresh, err = provision.GetRecord(ctx, l.s, l.rec.Account, l.rec.App, host); err != nil {
			return
		}
		if fresh == nil || fresh.Move == nil || fresh.Move.ID != id || fresh.Move.Incarnation != l.g.Incarnation || fresh.Move.State != provision.MoveCommitted {
			err = fmt.Errorf("move %s: the record is no longer committed", id)
			return
		}
		var obs hostclaim.Observed
		if obs, err = l.claim(ctx); err != nil {
			return
		}
		if op := obs.Claim.Op; op.Kind != hostclaim.KindLanding || op.Phase != hostclaim.PhaseLandingCommitted {
			err = fmt.Errorf("move %s: the claim is %s/%q, not committed", id, op.Kind, op.Phase)
			return
		}
		var rt *registry.Route
		if rt, err = cfg.Route(host); err != nil {
			return
		}
		switch {
		case rt == nil:
			err = fmt.Errorf("move %s: this box has no row to open", id)
			return
		case rt.HoldOp == "" && rt.Status == "ready":
			// open already: a replay
		case rt.HoldOp == "":
			err = fmt.Errorf("move %s: the row is unheld but %q, not open", id, rt.Status)
			return
		case rt.HoldOp != id:
			err = fmt.Errorf("move %s: the row is held %s by %q", id, rt.Status, rt.HoldOp)
			return
		default:
			if cfg.MarkProbeOpen != nil && !cfg.MarkProbeOpen(host, id) {
				log.Printf("moves: %s: no live probe entry for %s; the next request wakes it under the commit", host, id)
			}
			var released bool
			if released, err = cfg.ReleaseHeld(host, id); err != nil {
				return
			}
			if !released {
				if rt, err = cfg.Route(host); err != nil {
					return
				}
				if rt == nil || rt.Status != "ready" || rt.HoldOp != "" {
					err = fmt.Errorf("move %s: the release matched nothing and the row is not open", id)
					return
				}
			}
		}
		// The release is proved (the row is ready and unheld): the probe's
		// entry is opened, ending its idle protection (final review, ruling
		// 3). A failed release returned above and confirmed nothing.
		if cfg.ConfirmProbeOpen != nil {
			cfg.ConfirmProbeOpen(host, id)
		}
		if fresh.Move.OpenedAt == nil {
			var pre provision.Record
			var opened *provision.Record
			if opened, err = provision.SetRecordMove(ctx, l.s, l.rec.Account, l.rec.App, host, observedMove(&pre, provision.MoveOpened(l.g, cfg.Now()))); err != nil {
				err = fmt.Errorf("the record's opened_at: %w", err)
				return
			}
			l.w.logTransition(ctx, l.s, pre, *opened)
			stamped = true
		}
	})
	if err != nil {
		return false, fmt.Errorf("open the row: %w", err)
	}
	return stamped, nil
}