package host

import (
	"context"
	"errors"
	"fmt"
	"log"
	"os"
	"sort"
	"time"

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

// The abort (moves spec §7.11, joint N13; Plan 5 Task 1 with rulings K5, K6,
// K7, K19 and K21), behind CARLOS_MOVES_PHASE1. It replaces Plan 1's interim
// abort, which released the claim at once and left the staging lane in the
// target bucket with no owner.
//
// Who retires what. The source cannot read the target's moves/* nor write
// its parked/ (M17), so it never touches the lane: the lane is parked by the
// target region's box. The source retires only its own state.
//
// The source, for a record aborting in its region (phase1MoveRound):
//
//  1. with the claim at this move's moving/"" and this box executing, it
//     takes the staging lease, so no staging writer is mid-round and none
//     can sign another request (K5);
//  2. it journals abort_bound once: the later of now + SignatureLife +
//     DefaultSkew and any seal's bound. Every staging request is signed at a
//     check that leaves its whole signature life before the lease's
//     SafeUntil, and none is signed after this moment while the lease is
//     held, so no request can land after the bound. A replay reads the
//     journalled bound and never computes a later one (revertClaim's
//     pattern);
//  3. it CASes hostclaim.Abort with moves.AbortScope(host, op, inc, bound):
//     moving/aborting, the lane's park-replica queued for the target region,
//     owned by it. Execution stays with the source, which serves throughout;
//  4. only then it retires its own state: the move's snapshots in its own
//     bucket, every version through DeleteVersion (K19: they are outside the
//     park grammar, so "the source parks its own snapshots" means it
//     retires them from its own bucket), then the receipts and the ledger,
//     and last the op's journal directory (the bound is in the claim by
//     then). It runs on every pass while the claim is aborting, before the
//     release, and is a no-op once everything is gone: nothing infers from
//     a missing journal that it already ran, and any error (a listing, a
//     delete, the journal directory) ends the pass unreleased;
//  5. it waits until no obligation under the op remains, then clears the
//     record with provision.MoveGuardedAbortComplete, and last releases
//     with hostclaim.ReleaseAbort (K7: from moving/aborting only, so never
//     before the lane park was queued). Record first, then the claim's
//     release (§7.5 and §19 K8 as Paul's ruling M4 amends them): a crash
//     between leaves the record at rest and the claim moving/aborting with
//     nothing owed, and the console's recovery pass releases from LastMove
//     (Plan 5 Task 7). A round that finds the claim already serving and
//     recording this move's release (a binary that closed claim first)
//     completes the record; any other serving claim is left alone.
//
// The target region's box, for a record aborting whose To is its region:
// movesParticipantRecord answers it as the target on the state alone (the
// record never flipped, so the target has neither the row nor a journal)
// and laneParkTarget runs the lane's park-replica through the park engine
// over its own bucket until the engine acknowledges it after the horizon,
// abort_bound + SignatureAge + RequestLimit. A late request found by the
// engine's verification listing is parked as a further retirement event.
//
// A dead target region (K21). If no box of the target region runs the lane
// park, the obligation stays owed, the claim stays moving/aborting and
// nothing releases. That blocks a delete (and any other operation) of the
// host until the region returns. Nothing needs to be done by hand; the
// overdue page is Task 8's.
//
// Arming. The source's half needs only what a preparing round already has.
// The target's lane park is unarmed in production, as the revert's is,
// until Task 9 wires the park engine's leases (revertLaneArmed); the tests
// inject it. Plan 1's DeletePrefix path stays for legacy records only
// (legacyMoveRound); this path never deletes from the target bucket.

// abortStepBound is the abort's journalled writers' horizon (K5).
const abortStepBound = "abort_bound"

// abFault is the abort's test seam after a side effect.
func (w *movesWorker) abFault(point string) error {
	if f := w.cfg.abFault; f != nil {
		return f(point)
	}
	return nil
}

// abortSource is phase1MoveRound's MoveAborting arm.
func (w *movesWorker) abortSource(ctx context.Context, s release.Store, rec provision.Record) error {
	cfg := w.cfg
	id := rec.Move.ID
	if cfg.Box == "" {
		return errors.New("the abort needs this box's label")
	}
	obs, err := hostclaim.Read(ctx, s, rec.Host)
	if err != nil {
		return fmt.Errorf("read the host claim: %w", err)
	}
	if obs.Absent {
		w.notice(rec.Host+"#abort", "moves: %s: aborting move %s has no host claim; left for the console pass", rec.Host, id)
		return nil
	}
	op := obs.Claim.Op
	mine := op.Kind == hostclaim.KindMoving && op.ID == id
	releaseAfter := false
	switch {
	case mine && op.Exec != cfg.Box:
		w.notice(rec.Host+"#abort", "moves: %s: aborting move %s: the claim's executor is %q, not this box; left for the console pass", rec.Host, id, op.Exec)
		return nil
	case mine && op.Phase == hostclaim.PhaseMoving:
		ok, err := w.abortClaim(ctx, s, rec)
		if err != nil || !ok {
			return err
		}
		if obs, err = hostclaim.Read(ctx, s, rec.Host); err != nil {
			return fmt.Errorf("read the host claim: %w", err)
		}
		if obs.Absent || obs.Claim.Op.ID != id || obs.Claim.Op.Phase != hostclaim.PhaseAborting {
			return nil // another writer moved it; the next pass reads it again
		}
		fallthrough
	case mine && op.Phase == hostclaim.PhaseAborting:
		// Every pass, before the close: the retirement is idempotent, and
		// a close never follows from inferring that it already ran (fix
		// round 1). Any error, the journal directory's included, ends the
		// pass with the record and the claim still aborting.
		if err := w.retireOwnMoveState(ctx, rec); err != nil {
			return err
		}
		// Nothing owed under the op, checked before the record's close
		// (ruling M4), as the console's grant checks it.
		if w.abortOwes(rec, obs) {
			return nil
		}
		releaseAfter = true
	case hostclaim.RequireReleasedMove(id)(obs.Claim) == nil:
		// A binary that closed claim first released before a crash:
		// finish the record. The own state went before the release;
		// retiring it again is a no-op.
		if err := w.retireOwnMoveState(ctx, rec); err != nil {
			return err
		}
	default:
		w.notice(rec.Host+"#abort", "moves: %s: aborting move %s: the claim is %s/%q (%s) and records no release of this move; left for the console pass", rec.Host, id, op.Kind, op.Phase, op.ID)
		return nil
	}
	g := provision.MoveGuard{ID: id, Incarnation: rec.Move.Incarnation}
	var pre provision.Record
	after, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, observedMove(&pre, provision.MoveGuardedAbortComplete(g, cfg.Now())))
	if err == nil {
		w.logTransition(ctx, s, pre, *after)
	}
	if err != nil {
		if errors.Is(err, provision.ErrMoveConflict) {
			log.Printf("moves: %s: abort of %s: the record moved on (%v)", rec.Host, id, err)
			return nil
		}
		return err
	}
	nudgeSerial(ctx, s)
	log.Printf("moves: %s: move %s aborted; the instance stays in %q", rec.Host, id, rec.Region)
	if !releaseAfter {
		return nil
	}
	if err := w.abFault("after-abort-record"); err != nil {
		return err
	}
	return w.releaseAbort(ctx, s, rec)
}

// abortClaim is steps 1 to 3: the staging lease, the journalled bound and
// the Abort CAS. ok is false when the pass should wait (another worker holds
// the lease). Without a deployment bucket no round ever took the lease and
// none staged, so there is no lease to take and the bound alone holds.
func (w *movesWorker) abortClaim(ctx context.Context, s release.Store, rec provision.Record) (bool, error) {
	cfg := w.cfg
	host, id := rec.Host, rec.Move.ID
	if cfg.StagingLease != nil {
		_, rel, err := cfg.StagingLease(ctx, rec.Move.To, host)
		if err != nil {
			var he *lease.HeldError
			if errors.As(err, &he) {
				w.notice(host+"#abort", "moves: %s: aborting move %s waits: another worker holds the staging lease", host, id)
				return false, nil
			}
			return false, fmt.Errorf("staging lease: %w", err)
		}
		defer func() {
			rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
			defer cancel()
			if err := rel(rctx); err != nil {
				log.Printf("moves: %s: release the staging lease: %v", host, err)
			}
		}()
	}
	// Under the lease, the claim again: still this move's, before the seal.
	obs, err := hostclaim.Read(ctx, s, host)
	if err != nil {
		return false, fmt.Errorf("read the host claim: %w", err)
	}
	if obs.Absent || obs.Claim.Op.Kind != hostclaim.KindMoving || obs.Claim.Op.ID != id || obs.Claim.Op.Phase != hostclaim.PhaseMoving || obs.Claim.Op.Exec != cfg.Box {
		return true, nil
	}
	if obs.Claim.Holder == nil {
		return false, errors.New("the abort needs the claim's holder")
	}
	j, err := moves.ReadJournal(cfg.DataDir, host, id)
	if err != nil {
		return false, err
	}
	var bound time.Time
	if raw, ok := j.Steps[abortStepBound]; ok {
		if bound, err = time.Parse(time.RFC3339Nano, raw); err != nil {
			bound = time.Time{}
		}
	}
	if bound.IsZero() {
		bound = cfg.Now().Add(moves.SignatureLife + moves.DefaultSkew).UTC()
		if seal, err := journalledSeal(obs.Claim); err == nil && seal.Bound().After(bound) {
			bound = seal.Bound().UTC()
		}
		j.Mark(abortStepBound, bound.Format(time.RFC3339Nano))
		if err := j.Commit(cfg.DataDir); err != nil {
			return false, err
		}
		if err := w.abFault("after-abort-bound"); err != nil {
			return false, err
		}
	}
	scope := moves.AbortScope(host, id, obs.Claim.Holder.Incarnation, bound)
	if _, err := hostclaim.Update(ctx, s, obs, hostclaim.Abort(id, park.PlacementRegion(rec.Move.To), scope, obligationID)); err != nil {
		return false, fmt.Errorf("abort the claim: %w", err)
	}
	log.Printf("moves: %s: the claim aborts move %s; the staging lane in %s is queued for its park there", host, id, rec.Move.To)
	if err := w.abFault("after-abort-claim"); err != nil {
		return false, err
	}
	return true, nil
}

// abortOwes is step 5's wait: true, with a notice, while an obligation of
// the op remains (the lane park the target region has not acknowledged).
func (w *movesWorker) abortOwes(rec provision.Record, obs hostclaim.Observed) bool {
	id := rec.Move.ID
	for _, o := range obs.Claim.Obligations {
		if o.Op == id {
			w.notice(rec.Host+"#abort-wait", "moves: %s: aborting move %s waits for %s to park the staging lane (%s %s)", rec.Host, id, rec.Move.To, o.Kind, o.Region+o.Box)
			return true
		}
	}
	return false
}

// releaseAbort is step 5's second half: the record already says aborted, so
// hostclaim.ReleaseAbort from a fresh read. A failure is left to the
// console's recovery pass, which releases from LastMove.
func (w *movesWorker) releaseAbort(ctx context.Context, s release.Store, rec provision.Record) error {
	if err := releaseAfterClose(ctx, s, rec.Host, rec.Move.ID, hostclaim.ReleaseAbort(rec.Move.ID, w.cfg.Box)); err != nil {
		return err
	}
	return w.abFault("after-release-abort")
}

// retireOwnMoveState is step 4: the move's snapshots in this box's own
// bucket, every version through DeleteVersion (K19), then its receipts and
// ledger, and last its journal directory, which holds the bound. Idempotent.
func (w *movesWorker) retireOwnMoveState(ctx context.Context, rec provision.Record) error {
	cfg := w.cfg
	if cfg.Versioned == nil {
		return errors.New("the abort needs this box's versioned store to retire its snapshots")
	}
	vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
	if err != nil {
		return err
	}
	prefix := moves.SnapshotsPrefix(rec.Host, rec.Move.ID)
	vers, err := vs.ListVersions(ctx, cfg.OwnBucket, prefix)
	if err != nil {
		return fmt.Errorf("list the move's snapshots: %w", err)
	}
	for _, v := range vers {
		if err := vs.DeleteVersion(ctx, cfg.OwnBucket, v.Key, v.VersionID); err != nil {
			return fmt.Errorf("delete %s version %s: %w", v.Key, v.VersionID, err)
		}
	}
	own, err := w.mirror(ctx, cfg.OwnBucketRegion)
	if err != nil {
		return err
	}
	keys := []string{moves.ReceiptsKey(rec.Host, rec.Move.ID), moves.BuildStateReceiptsKey(rec.Host, rec.Move.ID), moves.LedgerKey(rec.Host, rec.Move.ID)}
	sort.Strings(keys)
	for _, k := range keys {
		if err := own.Delete(ctx, cfg.OwnBucket, k); err != nil && !errors.Is(err, lease.ErrMirrorNotFound) {
			return fmt.Errorf("delete %s: %w", k, err)
		}
	}
	if err := w.abFault("after-abort-own-state"); err != nil {
		return err
	}
	// The move's local directory: its journal and the archive kept for it.
	if err := os.RemoveAll(moves.JournalDir(cfg.DataDir, rec.Host, rec.Move.ID)); err != nil {
		return fmt.Errorf("remove the move's local directory: %w", err)
	}
	return w.abFault("after-abort-journal")
}