package host

// Quiet refusals (moves Plan 5 Task 9a; the Task 5 rulings). The claim's
// regional worker asks Task 5's provider about every host at rest, every
// minute, and an ordinary state refuses: a live move, residue the move's
// close does not vouch for, a joining claim with no holder to name a record
// (a free host's lanes are the claim's own free-host proof,
// claimFreeMoveProof). The worker reports each refusal, so the agent's log would repeat them every
// minute. movesRefusals keeps them quiet:
//
//   - report is the worker's report: a provider refusal (MovesSweepRefusal,
//     NotMovesEvidence) or a named dependency (ClaimDependencyError) is
//     logged at most once an hour per host and reason; anything else is
//     logged every time, as before;
//   - evidence wraps the provider: a refusal marked Anomaly (the history
//     contradicts itself, or something wrote where no move could) pages once
//     per host and reason through the box pager, as a warning: marked sent
//     only once delivered, never twice at once, a failure retried with
//     backoff a bounded number of times. The worker gets every error
//     unchanged.
//
// Both are wired only with CARLOS_MOVES_PHASE1 and CARLOS_HOST_CLAIM on;
// otherwise the worker's report is the plain log line it always was.

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/alerting"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/park"
)

// movesRefusalLogEvery is how often one expected refusal is logged.
const movesRefusalLogEvery = time.Hour

// movesPageAttempts bounds the tries one anomaly page gets.
const movesPageAttempts = 5

// movesRefusals is the agent's refusal log and anomaly pager for the moves
// evidence. Its memory is the process's: a restarted agent logs and pages
// each standing condition once more.
type movesRefusals struct {
	box  string
	page func(ctx context.Context, level, subject, body string) error
	logf func(format string, args ...any)
	now  func() time.Time

	mu     sync.Mutex
	logged map[string]time.Time
	pages  map[string]*movesPageState
}

// movesPageState is one anomaly's page: delivered once a send succeeds;
// while a send is in flight no other starts; a failed send is retried by a
// later observation after a doubling backoff, at most movesPageAttempts
// times in all.
type movesPageState struct {
	delivered, inflight bool
	failures            int
	next                time.Time
}

// movesPageBackoff is the wait after the first failed page; it doubles with
// each further failure, up to movesPageBackoffMax.
const (
	movesPageBackoff    = time.Minute
	movesPageBackoffMax = 30 * time.Minute
)

func newMovesRefusals(box string, page func(context.Context, string, string, string) error, logf func(string, ...any), now func() time.Time) *movesRefusals {
	return &movesRefusals{box: box, page: page, logf: logf, now: now, logged: map[string]time.Time{}, pages: map[string]*movesPageState{}}
}

// movesQuietEvidence is Task 5's provider with its anomalies paged. It is
// the regional evidence and the sweep verifier the claim registers.
type movesQuietEvidence struct {
	movesRetirementEvidence
	r *movesRefusals
}

var (
	_ hostbox.RegionalEvidence = movesQuietEvidence{}
	_ park.MoveSweepVerifier   = movesQuietEvidence{}
)

func (r *movesRefusals) evidence(e movesRetirementEvidence) movesQuietEvidence {
	return movesQuietEvidence{movesRetirementEvidence: e, r: r}
}

func (e movesQuietEvidence) VerifyRetirement(ctx context.Context, o hostclaim.Observed, versions []lease.ObjectVersion) (string, time.Time, error) {
	inc, at, err := e.movesRetirementEvidence.VerifyRetirement(ctx, o, versions)
	if err != nil {
		e.r.observe(ctx, err)
	}
	return inc, at, err
}

func (e movesQuietEvidence) VerifyMovesSweep(ctx context.Context, o hostclaim.Observed, versions []lease.ObjectVersion) (park.MoveSweepProof, error) {
	proof, err := e.movesRetirementEvidence.VerifyMovesSweep(ctx, o, versions)
	if err != nil {
		e.r.observe(ctx, err)
	}
	return proof, err
}

// observe pages an anomaly once per host and reason (fix round 1, I4): it
// is marked delivered only after the pager succeeds; a concurrent
// observation while a page is in flight sends nothing; a failed page is
// retried by a later observation after its backoff, at most
// movesPageAttempts times.
func (r *movesRefusals) observe(ctx context.Context, err error) {
	var refusal *MovesSweepRefusal
	if r == nil || r.page == nil || !errors.As(err, &refusal) || !refusal.Anomaly {
		return
	}
	key := refusal.Host + "\x00" + refusal.Reason
	now := r.now()
	r.mu.Lock()
	st := r.pages[key]
	if st == nil {
		st = &movesPageState{}
		r.pages[key] = st
	}
	if st.delivered || st.inflight || st.failures >= movesPageAttempts || now.Before(st.next) {
		r.mu.Unlock()
		return
	}
	st.inflight = true
	r.mu.Unlock()
	subject, body := movesAnomalyPage(refusal.Host, r.box)
	perr := r.page(ctx, alerting.LevelWarning, subject, body)
	r.mu.Lock()
	st.inflight = false
	if perr == nil {
		st.delivered = true
		r.mu.Unlock()
		return
	}
	st.failures++
	wait := movesPageBackoff << (st.failures - 1)
	if wait > movesPageBackoffMax || wait <= 0 {
		wait = movesPageBackoffMax
	}
	st.next = now.Add(wait)
	failures := st.failures
	r.mu.Unlock()
	if r.logf != nil {
		if failures >= movesPageAttempts {
			r.logf("moves: %s: the page for the refused cleanup failed %d times; no more are sent while this agent runs: %v", refusal.Host, failures, perr)
		} else {
			r.logf("moves: %s: the page for the refused cleanup failed (try %d of %d, again after %s): %v", refusal.Host, failures, movesPageAttempts, wait, perr)
		}
	}
}

// movesAnomalyPage is the anomaly's page: the subject (MV-N336) and the
// body (MV-N337). The reason stays in the agent's log, never in the page.
func movesAnomalyPage(host, box string) (subject, body string) {
	subject = fmt.Sprintf("carlos move cleanup refused: %s", host)
	body = fmt.Sprintf("The cleanup of %s on %s stopped: its move history does not add up, or something wrote where no move could. Nothing of %s is cleaned up in this region until an operator checks the agent's log on %s.",
		host, box, host, box)
	return subject, body
}

// report is the regional worker's report: each error of the pass, the
// expected refusals at most once an hour per host and reason.
func (r *movesRefusals) report(err error) {
	for _, leaf := range joinedErrors(err) {
		key, expected := movesExpectedRefusal(leaf)
		if !expected {
			r.logf("agent: regional claims: %v", leaf)
			continue
		}
		now := r.now()
		r.mu.Lock()
		last, seen := r.logged[key]
		if !seen || !now.Before(last.Add(movesRefusalLogEvery)) {
			r.logged[key] = now
			seen = false
		}
		r.mu.Unlock()
		if !seen {
			r.logf("agent: regional claims: %v (logged at most once an hour while it stands)", leaf)
		}
	}
}

// joinedErrors flattens errors.Join's tree into its leaves.
func joinedErrors(err error) []error {
	if err == nil {
		return nil
	}
	j, ok := err.(interface{ Unwrap() []error })
	if !ok {
		return []error{err}
	}
	var out []error
	for _, e := range j.Unwrap() {
		out = append(out, joinedErrors(e)...)
	}
	return out
}

// movesExpectedRefusal names a refusal that stands until its condition
// changes: the provider's typed refusals and a named dependency.
func movesExpectedRefusal(err error) (string, bool) {
	var refusal *MovesSweepRefusal
	var notMoves *NotMovesEvidence
	var dep *ClaimDependencyError
	switch {
	case errors.As(err, &refusal):
		return "sweep\x00" + refusal.Host + "\x00" + refusal.Reason, true
	case errors.As(err, &notMoves):
		return "not-moves\x00" + notMoves.Host + "\x00" + notMoves.Generation + "\x00" + notMoves.Reason, true
	case errors.As(err, &dep):
		return "dependency\x00" + strings.Join(dep.Contracts, "\x00"), true
	}
	return "", false
}