package host

import (
	"container/list"
	"context"
	"fmt"
	"log"
	"math"
	"strconv"
	"sync"

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

// The move log and the member stream (spec §13, Plan 6 Task 3).
//
// Every record write a box makes for a move goes through observedMove, which
// keeps the record as the write found it; logTransition names what the write
// did (moves.TransitionOf, which the console's move pass shares): one event,
// one identity. Transitions another writer made (the console's request,
// manual cut, abort and retry; a declared loss) are seen by the box's own
// pass (observeMove) and named the same way. A recovery probe's starts are
// the attempt journal's (movesProbeAttempts.RecoveryAttempt,
// recoveryAttemptLogger). The identity is the log object's key, written
// create-only behind moveCfg.LogEvent (moveLogWriter in production), and
// only the writer that created the object tells the app's members (preflight
// L9, L27).
//
// A write that fails is kept, with the entry's exact bytes, in the worker's
// outbox on disk (moveslogoutbox.go), and every pass retries the outbox
// whatever the record now says, until the entry is created or found there;
// so neither a phase that moves on nor a close loses it (fix round 2,
// ruling 1).
//
// A worker remembers the identities it has handed off (written, found, or
// kept in the outbox), so a pass that sees the same transition again does
// not even try. The memory is bounded (moveLogMemoCap, oldest dropped), and
// a move is forgotten once its terminal outcome is logged or a pass sees its
// record with no move or another move (retireMoves): create-only storage
// still refuses a replay.

// observedMove wraps a record mutation so *pre keeps the record as the write
// found it: the transition's before, for the log.
func observedMove(pre *provision.Record, mut func(*provision.Record) error) func(*provision.Record) error {
	return func(rec *provision.Record) error {
		*pre = *rec
		pre.Move = copyMove(rec.Move)
		return mut(rec)
	}
}

// copyMove is a deep enough copy of a move for moves.TransitionOf: the fields
// it compares and reads, safe from a mutation that edits the journal in
// place.
func copyMove(m *provision.MoveState) *provision.MoveState {
	if m == nil {
		return nil
	}
	c := *m
	if m.Rollback != nil {
		rb := *m.Rollback
		c.Rollback = &rb
	}
	if m.Landing != nil {
		l := *m.Landing
		c.Landing = &l
	}
	if m.Source != nil {
		src := *m.Source
		c.Source = &src
	}
	return &c
}

// logTransition logs the transition one write made.
func (w *movesWorker) logTransition(ctx context.Context, s release.Store, before, after provision.Record) {
	if e, ok := moves.TransitionOf(before, after); ok {
		w.record(ctx, s, after, e)
	}
}

// observeMove logs the transitions another writer made that this box's pass
// sees: the console's request, manual cut, abort and retry, and a declared
// loss. Each is named as if it were this box's write from a record with no
// move (or, for a retry, by its epoch), so it has the same identity whoever
// logs it.
func (w *movesWorker) observeMove(ctx context.Context, s release.Store, rec provision.Record) {
	w.retireMoves(rec)
	m := rec.Move
	if m == nil || m.ID == "" {
		return
	}
	switch {
	case m.State == provision.MovePreparing, m.State == provision.MoveAborting,
		m.State == provision.MoveCutting && m.CutPhase == provision.CutRequested:
		w.logTransition(ctx, s, provision.Record{}, rec)
	case m.State == provision.MoveRollingBack && m.Rollback != nil && m.Rollback.TargetPhase == provision.RbTargetLost:
		// The console that performed the declaration logs it first (final
		// review I4); this is the second producer of the same identity.
		if e, ok := moves.DeclaredLost(m); ok {
			w.record(ctx, s, rec, e)
		}
	}
	if m.RecoveryRetryEpoch > 0 && m.State != provision.MoveStuck {
		w.record(ctx, s, rec, moves.Transition{Op: m.ID, Event: "retry", Identity: "retry-" + strconv.FormatUint(m.RecoveryRetryEpoch, 10),
			Fields: map[string]string{"from": m.From, "to": m.To}})
	}
}

// logRevPhase logs a revert's step, which is the claim's rev_phase, not the
// record's: reverting-<phase>, event rev_<phase>.
func (w *movesWorker) logRevPhase(ctx context.Context, s release.Store, rec provision.Record, before, after hostclaim.Observed) {
	phase := func(o hostclaim.Observed) string {
		if o.Claim == nil {
			return ""
		}
		return o.Claim.Op.Journal[hostclaim.JournalRevPhase]
	}
	p := phase(after)
	if p == "" || p == phase(before) || rec.Move == nil || rec.Move.ID == "" {
		return
	}
	w.record(ctx, s, rec, moves.Transition{Op: rec.Move.ID, Event: "rev_" + p, Identity: "reverting-" + p,
		Fields: map[string]string{"from": rec.Move.From, "to": rec.Move.To}})
}

// recoveryAttemptLogger is the attempt journal's RecoveryAttempt hook for
// this worker (set by probeAttempts): each recovery probe start the journal
// counts is probe_attempt-<epoch>-<n> (preflight L11), the starts of one wake
// included. s is the store the log is written to. A write that fails goes to
// the outbox like any other.
func (w *movesWorker) recoveryAttemptLogger(s release.Store) func(ctx context.Context, rec provision.Record, epoch uint64, n int) {
	return func(ctx context.Context, rec provision.Record, epoch uint64, n int) {
		if rec.Move == nil || rec.Move.ID == "" {
			return
		}
		w.record(ctx, s, rec, moves.Transition{Op: rec.Move.ID, Event: "probe_attempt",
			Identity: "probe_attempt-" + strconv.FormatUint(epoch, 10) + "-" + strconv.Itoa(n),
			Fields:   map[string]string{"from": rec.Move.From, "to": rec.Move.To}})
	}
}

// terminalEvents are the events that end a move's log.
var terminalEvents = map[string]bool{"done": true, "rolled_back": true, "aborted": true}

// record hands one entry to the LogEvent hook unless this worker has
// already handed it off. An entry whose write fails is kept in the outbox,
// for every later pass to retry; only an entry written, found there or kept
// is remembered. With no hook (a test's worker) the event is a log line.
func (w *movesWorker) record(ctx context.Context, s release.Store, rec provision.Record, e moves.Transition) {
	key := rec.Host + "/" + e.Op + "/" + e.Identity
	if w.logged.seen(key) {
		return
	}
	ev := moves.LogEvent{V: 1, Op: e.Op, Host: rec.Host, Event: e.Event, Identity: e.Identity, Actor: w.cfg.Box, Fields: e.Fields}
	if w.cfg.Now != nil {
		ev.At = w.cfg.Now().UTC()
	}
	if f := w.cfg.LogEvent; f != nil {
		if err := f(ctx, s, rec, ev); err != nil {
			if qerr := w.queueLog(rec, ev); qerr != nil {
				log.Printf("moves: %s: the log of %s at %s waits for the next sighting: %v; the outbox: %v", rec.Host, e.Op, e.Identity, err, qerr)
				return
			}
			log.Printf("moves: %s: the log of %s at %s is kept for the next pass: %v", rec.Host, e.Op, e.Identity, err)
		}
	} else {
		log.Printf("moves: %s: %s %s (%s)", rec.Host, e.Op, e.Event, e.Identity)
	}
	if terminalEvents[e.Event] {
		w.logged.forget(rec.Host, e.Op)
		return
	}
	w.logged.add(rec.Host, e.Op, key)
}

// retireMoves forgets the moves of rec's host other than the one its record
// names now: a move closed, or replaced, by another writer.
func (w *movesWorker) retireMoves(rec provision.Record) {
	keep := ""
	if rec.Move != nil {
		keep = rec.Move.ID
	}
	w.logged.retire(rec.Host, keep)
}

// moveLogMemoCap bounds a worker's memory of handed-off identities.
const moveLogMemoCap = 4096

// moveLogMemo is the identities a worker has handed off: at most
// moveLogMemoCap, the oldest dropped first, indexed by host and move so a
// move is forgotten whole. The zero value is ready.
type moveLogMemo struct {
	mu    sync.Mutex
	order *list.List                            // memoEntry, oldest first
	at    map[string]*list.Element              // key to its element
	moves map[string]map[string][]*list.Element // host to move to its elements
}

type memoEntry struct{ host, op, key string }

func (m *moveLogMemo) seen(key string) bool {
	m.mu.Lock()
	defer m.mu.Unlock()
	_, ok := m.at[key]
	return ok
}

func (m *moveLogMemo) len() int {
	m.mu.Lock()
	defer m.mu.Unlock()
	return len(m.at)
}

func (m *moveLogMemo) add(host, op, key string) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.order == nil {
		m.order, m.at, m.moves = list.New(), map[string]*list.Element{}, map[string]map[string][]*list.Element{}
	}
	if _, ok := m.at[key]; ok {
		return
	}
	for len(m.at) >= moveLogMemoCap {
		m.drop(m.order.Front())
	}
	el := m.order.PushBack(memoEntry{host, op, key})
	m.at[key] = el
	if m.moves[host] == nil {
		m.moves[host] = map[string][]*list.Element{}
	}
	m.moves[host][op] = append(m.moves[host][op], el)
}

// drop removes one element from every index.
func (m *moveLogMemo) drop(el *list.Element) {
	e := el.Value.(memoEntry)
	m.order.Remove(el)
	delete(m.at, e.key)
	els := m.moves[e.host][e.op]
	for i, x := range els {
		if x == el {
			els = append(els[:i], els[i+1:]...)
			break
		}
	}
	if len(els) > 0 {
		m.moves[e.host][e.op] = els
		return
	}
	delete(m.moves[e.host], e.op)
	if len(m.moves[e.host]) == 0 {
		delete(m.moves, e.host)
	}
}

// forget drops one move's identities.
func (m *moveLogMemo) forget(host, op string) {
	m.mu.Lock()
	defer m.mu.Unlock()
	for _, el := range append([]*list.Element(nil), m.moves[host][op]...) {
		m.drop(el)
	}
}

// retire drops the identities of host's moves other than keep.
func (m *moveLogMemo) retire(host, keep string) {
	m.mu.Lock()
	defer m.mu.Unlock()
	for op, els := range m.moves[host] {
		if op == keep {
			continue
		}
		for _, el := range append([]*list.Element(nil), els...) {
			m.drop(el)
		}
	}
}

// moveLogWriter is the production LogEvent: the object, create-only, under
// the move's report prefix in the deployment bucket, and the member event
// when this writer created it. A storage failure is returned, so the worker
// tries the transition again at its next sighting.
func moveLogWriter(sink activator.EventSink) func(context.Context, release.Store, provision.Record, moves.LogEvent) error {
	return func(ctx context.Context, s release.Store, rec provision.Record, ev moves.LogEvent) error {
		if s == nil {
			return fmt.Errorf("moves: no store for the log of %s", ev.Op)
		}
		created, err := moves.WriteLog(ctx, s, moves.ReportPrefix(rec.Account, rec.App, rec.Host, ev.Op), ev)
		if err != nil {
			return err
		}
		if !created || sink == nil {
			return nil
		}
		if line, ok := memberLine(ev, rec.Host); ok {
			sink.Event(rec.Account, rec.App, rec.Host, memberLevel(ev.Event), line)
		}
		return nil
	}
}

// memberLine is what the app's members read for an event (preflight L9: M-C1,
// M-C5, M-C6, M-C7, M-C8, M-C9, and for stuck MV-N380, a line with no
// operator command in it), from the shared templates (moves/operatorcopy.go,
// L12), built from the entry alone, so an entry the outbox writes later says
// the same. A region the record names as the home's "" is
// moves.HomeRegionText. ok is false for an event members are not told of.
// Its level is memberLevel's.
func memberLine(ev moves.LogEvent, host string) (line string, ok bool) {
	from, to := moves.RegionText(ev.Fields["from"]), moves.RegionText(ev.Fields["to"])
	switch ev.Event {
	case "requested":
		return moves.MoveStartingLine(host, from, to), true
	case "admitted":
		budget, err := strconv.Atoi(ev.Fields["cut_budget_s"])
		if err != nil || budget <= 0 {
			return "", false
		}
		return moves.CutStartingLine(to, host, moves.CutBudgetText(budget)), true
	case "flipped":
		return moves.FlippedLine(from, to), true
	case "done":
		offline, err := strconv.ParseFloat(ev.Fields["offline_s"], 64)
		if err != nil {
			return "", false
		}
		return moves.MovedLine(host, to, int(math.Round(offline))), true
	case "rolling_back":
		reason := ev.Fields["reason"]
		if reason == "" {
			return "", false
		}
		return moves.MovingBackLine(to, host, reason, from), true
	case "rolled_back":
		return moves.MovedBackLine(host, from), true
	case "stuck":
		return moves.StuckMemberLine(host, to, from), true
	}
	return "", false
}

// memberLevel is the level of an event's member line.
func memberLevel(event string) string {
	switch event {
	case "rolling_back", "rolled_back":
		return "warn"
	case "stuck":
		return "error"
	}
	return "info"
}