package host

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

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/provision"
)

// The landing's probing phase (moves spec §7.6, §7.7, §7.13; Plan 3 Task 6).
//
// restored to probing is the landing's step (movesland.go). At probing the
// pass makes the probe attempts T11 admitted (H10: the record's
// CutAttempts, written by MoveAdmit), at most moves.AttemptsFor("landing",
// n). Each attempt is counted on the record (provision.MoveProbeAttempt)
// and journalled (Journal.Attempt) before its call, then made through the
// probe wake (Task 5's activator.Service.ProbeWake behind moveCfg.Probe):
// one cold admission into the prepared generation (or a fresh one for a
// later attempt), restored, rendered, published and started by the claim
// (Group D), up to five starts inside the attempt's 30 s window, bounded by
// the landing's deadline (H8). The attempt's own journal steps
// (probe_opened_<k>, probe_start_<k>) are Task 5's movesProbeAttempts';
// probe_failed_<k> is this file's, written when the attempt is over.
//
// Outcomes. A pass: probing to probed, record then claim (the claim's step
// keeps the admitted writer session, H3); the commit is Task 7's. Task 5's
// ProbeInputsChanged: its recorded sentence, the configuration rollback. An
// attempt that is over (ErrProbeFailed, ErrAttemptOver): the next one, if
// any; none left: the attempts-exhausted rollback; the deadline passed: the
// deadline's. Rollbacks are marks (MoveMarkRollback); the one conversion to
// MoveRollBack is the pass's rollback decision (movesrollback.go). Any other
// error ends the pass and the same attempt is made again on the next.
//
// A crash restarts only the probe (§7.13): local and restored are never run
// again; a fresh worker finds probing and the counted attempt. One cold
// admission per budgeted attempt (§7.7, fix round 1, ruling 2): an attempt
// is admitted once Task 5's journalled session binding (probe_session_<s> =
// k) exists for it, and an admitted attempt is consumed on restart, failed
// unless it passed. Only a counted attempt with no binding is made again.
// Fix round 2: the claim is the source of truth for the admission itself.
// probe_admitting_<k> records the claim's writer before the call; a claim
// whose writer moved on since admitted under k though Reserve's binding
// never became durable. Before an admitted attempt with no probe_passed_<k>
// is consumed, the activator's live entry is asked: this move's, awake,
// with the claim's writer, it passed, and its pass is journalled then.
//
// A pass is journalled (probe_passed_<k>: the admitted session, stamp and
// generation) before the probing to probed record write (ruling 1). A pass
// that finds it does not probe again: it checks the claim's holder still
// carries that writer and the activator's live probe entry is this move's,
// then retries only the phase writes, record then claim. A writer that is
// gone consumes the attempt as failed.
//
// Unarmed in production: moveCfg.Probe is nil until claim Plan 11's launcher
// (R10) and Plan 5 Task 9's arming, so the landing waits at restored.

// probeStepFailed marks attempt k over in the op journal; probeStepPassed
// names attempt k's passing writer (probePassed).
const (
	probeStepFailed    = "probe_failed_"
	probeStepPassed    = "probe_passed_"
	probeStepAdmitting = "probe_admitting_"
)

// probeAdmitting is a probe_admitting_<k> step's value, written before the
// call: the claim's writer session and admission count the attempt started
// from. The claim is the source of truth for whether the attempt admitted
// (fix round 2, ruling 1).
type probeAdmitting struct {
	Session    uint64 `json:"session"`
	Admissions int    `json:"admissions"`
}

// probePassed is a probe_passed_<k> step's value: the admitted writer the
// passing attempt left serving.
type probePassed struct {
	Session    uint64 `json:"session"`
	Stamp      uint64 `json:"stamp"`
	Generation string `json:"generation"`
}

// probeWindowLen is the attempt's window, counted from its first start.
var probeWindowLen = time.Duration(moves.ProbeSeconds * float64(time.Second))

// MovesRecordAdmittedAttempts is Task 5's Admitted hook (H10): the probe
// attempts the record's admission priced, refused when it recorded none.
func MovesRecordAdmittedAttempts(_ context.Context, rec *provision.Record) (int, error) {
	if rec == nil || rec.Move == nil || rec.Move.CutAttempts < 1 {
		return 0, errors.New("moves: " + provision.RollbackProbeUnpriced)
	}
	return rec.Move.CutAttempts, nil
}

// attemptSpent reports whether counted attempt k is over: its failure is
// journalled, it was admitted (attemptAdmittedBy), or its window opened and
// has passed.
func (l *landingRun) attemptSpent(j *moves.Journal, k int, h *hostclaim.Holder) bool {
	if j.Done(probeStepFailed+strconv.Itoa(k)) || l.attemptAdmittedBy(j, k, h) {
		return true
	}
	raw, ok := j.Steps[probeStepOpened+strconv.Itoa(k)]
	if !ok {
		return false
	}
	opened, err := time.Parse(time.RFC3339Nano, raw)
	return err == nil && !l.cfg().Now().Before(opened.Add(probeWindowLen))
}

// attemptAdmitted reports whether a session is journalled as bound to
// record attempt k (movesProbeAttempts.bindSession).
func attemptAdmitted(j *moves.Journal, k int) bool {
	want := strconv.Itoa(k)
	for step, v := range j.Steps {
		if strings.HasPrefix(step, probeStepSession) && v == want {
			return true
		}
	}
	return false
}

// attemptAdmittedBy reports whether attempt k admitted a writer: Task 5's
// binding exists for it, or probe_admitting_<k> was written and the claim
// has since moved on (another writer session, or a newer cold admission on
// this box). probe_admitting_<k> with the claim unchanged is an attempt that
// never admitted, and only that may reuse its index.
func (l *landingRun) attemptAdmittedBy(j *moves.Journal, k int, h *hostclaim.Holder) bool {
	if attemptAdmitted(j, k) {
		return true
	}
	raw, ok := j.Steps[probeStepAdmitting+strconv.Itoa(k)]
	if !ok || h == nil {
		return false
	}
	var a probeAdmitting
	if err := json.Unmarshal([]byte(raw), &a); err != nil {
		// Unreadable: never risk a second admission under k.
		return true
	}
	if h.Session != a.Session {
		return true
	}
	if len(h.Admissions) > a.Admissions {
		last := h.Admissions[len(h.Admissions)-1]
		return last.Box == l.cfg().Box && last.Mode == "cold"
	}
	return false
}

// livePass reports the activator's live probe entry when it is this move's
// and its writer is the claim's current one: an attempt that passed though
// its probe_passed_<k> never became durable (fix round 2, ruling 2).
func (l *landingRun) livePass(h *hostclaim.Holder) (activator.ProbeResult, bool) {
	live := l.cfg().ProbeLive
	if live == nil || h == nil || h.WriterBox != l.cfg().Box {
		return activator.ProbeResult{}, false
	}
	op, res, ok := live(l.rec.Host)
	if !ok || op != l.g.ID || res.Session != h.Session || res.Stamp != h.Stamp || res.Generation != h.Generation {
		return activator.ProbeResult{}, false
	}
	return res, true
}

// journalPassed writes probe_passed_<k> for the passing writer res.
func (l *landingRun) journalPassed(k int, res activator.ProbeResult) error {
	if err := l.fault("before-journal-passed"); err != nil {
		return err
	}
	raw, err := json.Marshal(probePassed{Session: res.Session, Stamp: res.Stamp, Generation: res.Generation})
	if err != nil {
		return err
	}
	j, err := l.journal()
	if err != nil {
		return err
	}
	j.Mark(probeStepPassed+strconv.Itoa(k), string(raw))
	return j.Commit(l.cfg().DataDir)
}

// failAttempt journals attempt k over.
func (l *landingRun) failAttempt(k int) error {
	j, err := l.journal()
	if err != nil {
		return err
	}
	if j.Done(probeStepFailed + strconv.Itoa(k)) {
		return nil
	}
	j.Mark(probeStepFailed+strconv.Itoa(k), l.cfg().Now().UTC().Format(time.RFC3339Nano))
	return j.Commit(l.cfg().DataDir)
}

// passed writes the phase steps of a passing attempt: record, then claim.
func (l *landingRun) passed(ctx context.Context) (bool, error) {
	if err := l.fault("before-record-probed"); err != nil {
		return false, err
	}
	return l.advance(ctx, provision.LandProbing, provision.LandProbed, nil)
}

// passedWriterLives checks a journalled pass against the claim's holder and
// the activator's live probe entry.
func (l *landingRun) passedWriterLives(obs hostclaim.Observed, raw string) bool {
	var p probePassed
	if err := json.Unmarshal([]byte(raw), &p); err != nil || p.Session == 0 {
		return false
	}
	h := obs.Claim.Holder
	if h == nil || h.Session != p.Session || h.Stamp != p.Stamp || h.Generation != p.Generation || h.WriterBox != l.cfg().Box {
		return false
	}
	live := l.cfg().ProbeLive
	if live == nil {
		return false
	}
	op, res, ok := live(l.rec.Host)
	return ok && op == l.g.ID && res.Session == p.Session && res.Stamp == p.Stamp && res.Generation == p.Generation
}

// probing makes the landing's probe attempts (see the file comment).
func (l *landingRun) probing(ctx context.Context) (bool, error) {
	cfg := l.cfg()
	obs, err := l.syncClaimPhase(ctx)
	if err != nil {
		return false, err
	}
	if err := l.requirePhase(obs, provision.LandProbing); err != nil {
		return false, err
	}
	if cfg.Probe == nil {
		l.w.notice(l.rec.Host+"#probe", "moves: %s: landing %s waits at probing: this box has no probe wake", l.rec.Host, l.g.ID)
		return false, nil
	}
	m := l.rec.Move
	if m.CutAttempts < 1 {
		return l.markRollback(ctx, provision.RollbackProbeUnpriced)
	}
	n := moves.AttemptsFor(moves.ProbeLanding, m.CutAttempts)
	j, err := l.journal()
	if err != nil {
		return false, err
	}
	raw, ok := j.Steps[landStepInstall]
	if !ok {
		return false, fmt.Errorf("move %s: probing without an install manifest in the journal", l.g.ID)
	}
	var manifest moves.InstallManifest
	if err := json.Unmarshal([]byte(raw), &manifest); err != nil {
		return false, fmt.Errorf("move %s: the journalled install manifest: %w", l.g.ID, err)
	}
	counted := 0
	if m.Landing != nil {
		counted = m.Landing.Attempts
	}
	k := counted
	if k > 0 {
		if raw, ok := j.Steps[probeStepPassed+strconv.Itoa(k)]; ok {
			if l.passedWriterLives(obs, raw) {
				log.Printf("moves: %s: landing %s attempt %d passed; its phase writes are retried", l.rec.Host, l.g.ID, k)
				return l.passed(ctx)
			}
			log.Printf("moves: %s: landing %s attempt %d passed, but its writer is gone; the attempt is consumed", l.rec.Host, l.g.ID, k)
			if err := l.failAttempt(k); err != nil {
				return false, err
			}
			k++
		} else if l.attemptSpent(j, k, obs.Claim.Holder) {
			if !j.Done(probeStepFailed + strconv.Itoa(k)) {
				if res, ok := l.livePass(obs.Claim.Holder); ok {
					log.Printf("moves: %s: landing %s attempt %d passed (its live writer is the claim's); its phase writes follow", l.rec.Host, l.g.ID, k)
					if err := l.journalPassed(k, res); err != nil {
						return false, err
					}
					return l.passed(ctx)
				}
			}
			if err := l.failAttempt(k); err != nil {
				return false, err
			}
			k++
		}
	} else {
		k = 1
	}
	for ; k <= n; k++ {
		if l.pastDeadline() {
			return l.markRollback(ctx, provision.RollbackLandingDeadline)
		}
		if k > counted {
			ok, err := l.transition(ctx, "probe attempt "+strconv.Itoa(k), provision.MoveProbeAttempt(l.g, k))
			if !ok {
				return false, err
			}
			counted = k
		}
		// The claim's writer before the call: whether the attempt admitted is
		// the claim's to say after a crash (ruling 1).
		before, err := l.claim(ctx)
		if err != nil {
			return false, err
		}
		admitting, err := json.Marshal(probeAdmitting{Session: before.Claim.Holder.Session, Admissions: len(before.Claim.Holder.Admissions)})
		if err != nil {
			return false, err
		}
		// Under the journal's one lock (moves.UpdateJournal): the probe
		// owners and the paging rules write this journal too.
		if err := moves.UpdateJournal(cfg.DataDir, l.rec.Host, l.g.ID, func(j *moves.Journal) error {
			j.Phase, j.Attempt = provision.LandProbing, k
			j.Mark(probeStepAdmitting+strconv.Itoa(k), string(admitting))
			return nil
		}); err != nil {
			return false, err
		}
		if err := l.fault("before-probe-" + strconv.Itoa(k)); err != nil {
			return false, err
		}
		deadline, _ := landingDeadlineOf(m)
		if deadline == nil {
			return false, fmt.Errorf("move %s: the landing has no deadline", l.g.ID)
		}
		res, err := cfg.Probe(ctx, activator.ProbeConfig{Host: l.rec.Host, Op: l.g.ID, HoldOp: l.g.ID, Deadline: *deadline, Prepared: &manifest, Inputs: m.Inputs})
		var changed *ProbeInputsChanged
		switch {
		case err == nil:
			log.Printf("moves: %s: landing %s passed its probe on attempt %d of %d", l.rec.Host, l.g.ID, k, n)
			if err := l.journalPassed(k, res); err != nil {
				return false, err
			}
			return l.passed(ctx)
		case errors.As(err, &changed):
			log.Printf("moves: %s: landing %s probe attempt %d: %v", l.rec.Host, l.g.ID, k, err)
			return l.markRollback(ctx, changed.Reason)
		case errors.Is(err, activator.ErrProbeFailed) || errors.Is(err, activator.ErrAttemptOver):
			log.Printf("moves: %s: landing %s probe attempt %d of %d is over: %v", l.rec.Host, l.g.ID, k, n, err)
			if err := l.failAttempt(k); err != nil {
				return false, err
			}
		default:
			return false, fmt.Errorf("probe attempt %d: %w", k, err)
		}
	}
	if l.pastDeadline() {
		return l.markRollback(ctx, provision.RollbackLandingDeadline)
	}
	return l.markRollback(ctx, provision.RollbackProbeAttempts)
}