package host

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io/fs"
	"log"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/alerting"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/provision"
)

// Paging for moves from the box (spec §13; Plan 6 Task 5; preflight L8). The
// box pages, through moveCfg.Page (PageOperator, the agent's alert sinks;
// Plan 5 Task 0), only what it alone sees:
//
//   - stuck, an incident, once per entry into stuck (pageStuck): the rule is
//     stuck-<epoch>, the record's recovery retry epoch, which only an
//     operator's retry advances, so a retried move that sticks again pages
//     again and a replay of the same mark does not;
//   - rolled_back, a warning, once per move, where this box closes a
//     rollback (recoverRun.finalize) or a revert (revertRun.close);
//   - a stop evidence refused for a process the move cannot account for
//     (StepStray), an incident, once per move, from the cut's stopping: the
//     page's body is M-C20, and the agent's log says M-C20 when the box
//     detects it. M-C20 is operator copy: the app's members are never told
//     (fix round 1, ruling 1);
//   - a refused start check (moves.StartCheck), an incident, once per move,
//     from the probe owners (movesProbeOwners.startCheck).
//
// The console's move pass (internal/console/movesrecovery.go, Plan 5 Task 7)
// keeps the rules whose facts it holds, and the box does not repeat them:
// rolling_back past release_deadline (an incident, once an hour), preparing
// with no promotable round for 60 minutes (a warning, once, measured from
// moves.PreparingSince, the six-hour abort's own start), an orphaned claim,
// an overdue close and an overdue abort. A move converted to relocating with
// no live box in its region (M-C22) is Plan 5 Task 7a's page.
//
// The page outbox (fix round 2, ruling 1). A page is one file,
// <DataDir>/.moves/.page-outbox/<host>/<move>/<rule>.json, created
// exclusively (outboxCreate, the move log's outbox's own helper) when its
// rule first fires: the file being there is the rule's once-per-entry limit,
// so a rule that fires again never pages again. The entry carries the page
// and its backoff. It is sent when created; one the pager refused is sent
// again by every moves pass (flushPageOutbox, first thing in the pass, before
// any record or fleet read) once its backoff is over (pageRetryFirst,
// doubling, up to pageRetryMax), until it is delivered, whatever the move
// and its journals have done since. A delivered entry stays as a tombstone,
// so a rule that fires again still finds it, until the move is retired on
// this box: no listed record carries it and its journal is gone
// (retirePageTombstones, only after a pass listed every record). The outbox
// depends on nothing in the move's journal.
//
// An index in memory (pageIndex, fix round 3) answers for what the disk
// cannot: a firing looks at the disk entry, then the index, and only a page
// in neither is created; only when that create fails is the page sent at
// once, kept in the index, and, if that fails too, sent again by every pass
// with a loud line in the agent's log. A delivery or a backoff the disk
// cannot record is kept in the index too, and a delivered page's tombstone
// is written again every pass until it lands. Only a restart while the index
// holds what the disk does not loses it.
//
// Unwired (nil Page, as production is until Plan 5 Task 9b wires it), a rule
// is never silent: the agent's log says so loudly, once per rule per move
// for as long as the agent runs, and nothing is refused or kept (fix round 1,
// ruling 2).

// pagingLogf is the agent's log for the paging code (a variable so tests
// can read it).
var pagingLogf = log.Printf

// pageFunc is the pager: moveCfg.Page's shape.
type pageFunc = func(ctx context.Context, level, subject, body string) error

// The rules' names in the outbox.
const (
	pageRuleRolledBack = "rolled_back"
	pageRuleStray      = "stray"
	pageRuleStartCheck = "start_check"
)

// pageRuleStuck is the stuck rule of one entry into stuck: its retry epoch.
func pageRuleStuck(m *provision.MoveState) string {
	return "stuck-" + strconv.FormatUint(m.RecoveryRetryEpoch, 10)
}

// pageOutcome is what became of one rule's firing.
type pageOutcome int

const (
	pageSent        pageOutcome = iota // handed to the pager without error
	pagePagedBefore                    // the outbox says it was delivered before
	pageUnwired                        // no pager is wired (logged loudly once)
	pageUndelivered                    // first firing, refused: kept to send again
	pagePending                        // fired before, not delivered yet: the passes send it again
)

// String is the outcome as the agent's log says it.
func (o pageOutcome) String() string {
	switch o {
	case pageSent:
		return "paged"
	case pagePagedBefore:
		return "paged before"
	case pageUnwired:
		return "not paged: no pager is wired"
	case pagePending:
		return "not paged yet: it is sent again after a backoff"
	}
	return "not paged: the page was not delivered, and is sent again after a backoff"
}

// A refused page is tried again pageRetryFirst after its first attempt, and
// after each further attempt twice as long again, up to pageRetryMax.
const (
	pageRetryFirst = time.Minute
	pageRetryMax   = 30 * time.Minute
)

// pageBackoff is the wait after the tries-th attempt.
func pageBackoff(tries int) time.Duration {
	if tries < 1 {
		tries = 1
	}
	if tries > 6 {
		return pageRetryMax
	}
	if d := pageRetryFirst << (tries - 1); d < pageRetryMax {
		return d
	}
	return pageRetryMax
}

// pageEntry is one page in the outbox: the page, the attempts so far, when
// the next may start, and when it was delivered.
type pageEntry struct {
	V         int        `json:"v"`
	Host      string     `json:"host"`
	Op        string     `json:"op"`
	Rule      string     `json:"rule"`
	Level     string     `json:"level"`
	Subject   string     `json:"subject"`
	Body      string     `json:"body"`
	At        time.Time  `json:"at"`
	Tries     int        `json:"tries"`
	Next      time.Time  `json:"next"`
	Delivered *time.Time `json:"delivered,omitempty"`
}

// pageOutboxDir is the page outbox's root.
func pageOutboxDir(dataDir string) string {
	return filepath.Join(dataDir, ".moves", ".page-outbox")
}

// pageEntryDir is the directory of one move's pages.
func pageEntryDir(dataDir, host, op string) string {
	return filepath.Join(pageOutboxDir(dataDir), host, op)
}

// pageOutboxMu serializes every change to an outbox entry, and to the page
// index, in this process (the moves pass and the probe owners both fire
// pages); delivery itself is outside it, after the attempt is counted.
var pageOutboxMu sync.Mutex

// The page index (fix round 3): what this process knows of a page that the
// disk does not say, keyed by (outbox, host, move, rule). It holds a page
// the outbox could not keep (memory), a backoff deadline the disk could not
// record (next), and a delivery the disk does not record yet (delivered,
// whose tombstone every pass writes again until it lands). A firing looks at
// the disk and then the index, so one (host, move, rule) is one page however
// the storage fails; only a restart while the index holds a page the disk
// does not loses it.
type pageKey struct{ dataDir, host, op, rule string }

type pageIndexItem struct {
	e         pageEntry
	page      pageFunc // the pager that fired a memory page
	memory    bool     // the outbox could not keep it
	delivered bool
	next      time.Time
}

var pageIndex = map[pageKey]*pageIndexItem{}

func pageKeyOf(dataDir string, e pageEntry) pageKey {
	return pageKey{dataDir, e.Host, e.Op, e.Rule}
}

// readPageEntry reads one entry.
func readPageEntry(path string) (pageEntry, error) {
	var e pageEntry
	b, err := os.ReadFile(path)
	if err != nil {
		return e, err
	}
	if err := json.Unmarshal(b, &e); err != nil || e.V != 1 {
		return e, fmt.Errorf("%s is not a page outbox entry", path)
	}
	return e, nil
}

// writePageEntry replaces an entry.
func writePageEntry(dataDir string, e pageEntry) error {
	b, err := json.Marshal(e)
	if err != nil {
		return err
	}
	return outboxReplace(dataDir, pageEntryDir(dataDir, e.Host, e.Op), e.Rule+".json", b)
}

// pageEntryPath is an entry's file.
func pageEntryPath(dataDir, host, op, rule string) string {
	return filepath.Join(pageEntryDir(dataDir, host, op), rule+".json")
}

// firePage is a rule firing (fix round 3, ruling 1). The disk entry is
// looked at first: there, delivered or not, the rule sends nothing (the
// passes retry a refused one). Then the index: there, it sends nothing.
// Only a (host, move, rule) in neither is created, its attempt counted, and
// sent; and only when that create fails is the page sent directly, and kept
// in the index if that fails too. Unwired, it warns once per rule per move
// (warned is the memory of that) and keeps nothing.
func firePage(ctx context.Context, page pageFunc, warned *sync.Map, dataDir, host, op, rule string, now time.Time, level, subject, body string) pageOutcome {
	if page == nil {
		if _, seen := warned.LoadOrStore(host+"/"+op+"/"+rule, true); !seen {
			pagingLogf("moves: %s: WARNING: no operator pager is wired on this box, so the %s page of move %s is not sent; the move goes on (Plan 5 Task 9b wires the pager)", host, rule, op)
		}
		return pageUnwired
	}
	e := pageEntry{V: 1, Host: host, Op: op, Rule: rule, Level: level, Subject: subject, Body: body, At: now.UTC(), Tries: 1, Next: now.UTC().Add(pageBackoff(1))}
	if !outboxSegment(host) || !outboxSegment(op) || !outboxSegment(rule) || dataDir == "" {
		pagingLogf("moves: %s: WARNING: the %s page of move %s has no usable outbox name, so it is sent now", host, rule, op)
		if err := page(ctx, level, subject, body); err != nil {
			pagingLogf("moves: %s: WARNING: the %s page of move %s was not delivered: %v", host, rule, op, err)
			return pageUndelivered
		}
		return pageSent
	}
	key := pageKeyOf(dataDir, e)
	pageOutboxMu.Lock()
	if kept, err := readPageEntry(pageEntryPath(dataDir, host, op, rule)); err == nil {
		pageOutboxMu.Unlock()
		if kept.Delivered != nil {
			return pagePagedBefore
		}
		return pagePending
	} else if !errors.Is(err, fs.ErrNotExist) {
		if _, serr := os.Lstat(pageEntryPath(dataDir, host, op, rule)); serr == nil {
			// There, but not readable now: it is the rule's entry all the same.
			pageOutboxMu.Unlock()
			return pagePending
		}
	}
	if item := pageIndex[key]; item != nil {
		delivered := item.delivered
		pageOutboxMu.Unlock()
		if delivered {
			return pagePagedBefore
		}
		return pagePending
	}
	b, err := json.Marshal(e)
	created := false
	if err == nil {
		created, err = outboxCreate(dataDir, pageEntryDir(dataDir, host, op), rule+".json", b)
	}
	if err == nil && !created {
		pageOutboxMu.Unlock()
		return pagePending
	}
	if err != nil {
		// The outbox cannot keep it: the index does, before the send, so no
		// re-fire sends it again.
		item := &pageIndexItem{e: e, page: page, memory: true, next: e.Next}
		pageIndex[key] = item
		pageOutboxMu.Unlock()
		pagingLogf("moves: %s: WARNING: the page outbox cannot keep the %s page of move %s, so it is sent now: %v", host, rule, op, err)
		if perr := page(ctx, level, subject, body); perr != nil {
			pagingLogf("moves: %s: WARNING: the %s page of move %s was not delivered either (%v); it is kept in memory only and sent again every pass, and is lost if the agent restarts first", host, rule, op, perr)
			return pageUndelivered
		}
		pageOutboxMu.Lock()
		item.delivered = true
		pageOutboxMu.Unlock()
		return pageSent
	}
	pageOutboxMu.Unlock()
	return deliverPageEntry(ctx, page, dataDir, e, now)
}

// deliverPageEntry sends an entry whose attempt is already counted, and marks
// it delivered once the pager took it: on disk, or, when the disk cannot
// record it, in the index, whose tombstone the passes write again until it
// lands. A refused page stays for the passes.
func deliverPageEntry(ctx context.Context, page pageFunc, dataDir string, e pageEntry, now time.Time) pageOutcome {
	if err := page(ctx, e.Level, e.Subject, e.Body); err != nil {
		pagingLogf("moves: %s: the %s page of move %s was not delivered (try %d), and is sent again after %s: %v", e.Host, e.Rule, e.Op, e.Tries, pageBackoff(e.Tries), err)
		return pageUndelivered
	}
	pageOutboxMu.Lock()
	defer pageOutboxMu.Unlock()
	if err := markPageDelivered(dataDir, e, now); err != nil {
		pageIndex[pageKeyOf(dataDir, e)] = &pageIndexItem{e: e, delivered: true}
		pagingLogf("moves: %s: the %s page of move %s was sent; the outbox cannot say so yet, so this agent remembers it and writes it again every pass: %v", e.Host, e.Rule, e.Op, err)
	}
	return pageSent
}

// markPageDelivered writes e's entry as delivered: the entry on disk marked,
// or, when there is none, a tombstone created. The caller holds
// pageOutboxMu.
func markPageDelivered(dataDir string, e pageEntry, now time.Time) error {
	at := now.UTC()
	kept, err := readPageEntry(pageEntryPath(dataDir, e.Host, e.Op, e.Rule))
	switch {
	case err == nil:
		if kept.Delivered != nil {
			// Marked before, perhaps by a write whose sync failed: it
			// counts once it is durable (final review I2).
			return outboxSettle(dataDir, pageEntryDir(dataDir, e.Host, e.Op), pageEntryPath(dataDir, e.Host, e.Op, e.Rule))
		}
		kept.Delivered = &at
		return writePageEntry(dataDir, kept)
	case errors.Is(err, fs.ErrNotExist):
		t := e
		t.Delivered = &at
		b, err := json.Marshal(t)
		if err != nil {
			return err
		}
		_, err = outboxCreate(dataDir, pageEntryDir(dataDir, e.Host, e.Op), e.Rule+".json", b)
		return err
	}
	return err
}

// flushPageOutbox is every moves pass's delivery half of paging (fix round 3,
// ruling 4: it runs first, before any record or fleet read, so a bucket that
// cannot be read never holds a page back). The index's pages are sent again
// once their backoff is over and its delivered pages' tombstones written
// again; then each disk entry not yet delivered is sent again once its
// backoff (the later of the disk's and the index's) is over, its attempt
// counted and the next put off before the send, so two senders in one
// window send it once. A deadline the disk cannot record is kept in the
// index (fix round 3, ruling 2), with its attempt count, until a write of
// the entry lands (re-review 3). Unwired, the disk's pages wait.
func (w *movesWorker) flushPageOutbox(ctx context.Context) {
	page, dd := w.cfg.Page, w.cfg.DataDir
	now := w.now()
	flushPageIndex(ctx, now)
	if page == nil || dd == "" {
		return
	}
	root := pageOutboxDir(dd)
	err := walkPageOutbox(root, func(p string) error {
		if err := ctx.Err(); err != nil {
			return err
		}
		w.flushPageEntry(ctx, page, dd, p, now)
		return nil
	})
	if err != nil {
		pagingLogf("moves: the page outbox: %v", err)
	}
}

// walkPageOutbox calls fn with each entry file under root.
func walkPageOutbox(root string, fn func(p string) error) error {
	return filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
		if errors.Is(err, fs.ErrNotExist) && p == root {
			return fs.SkipAll
		}
		if err != nil {
			return err
		}
		if d.IsDir() || !strings.HasSuffix(d.Name(), ".json") || strings.HasPrefix(d.Name(), ".") {
			return nil
		}
		return fn(p)
	})
}

// flushPageEntry is one disk entry of flushPageOutbox.
func (w *movesWorker) flushPageEntry(ctx context.Context, page pageFunc, dd, p string, now time.Time) {
	pageOutboxMu.Lock()
	e, err := readPageEntry(p)
	if err != nil || e.Rule+".json" != filepath.Base(p) || e.Op != filepath.Base(filepath.Dir(p)) || e.Host != filepath.Base(filepath.Dir(filepath.Dir(p))) {
		pageOutboxMu.Unlock()
		pagingLogf("moves: the page outbox: %s is not an entry, and is left for an operator: %v", p, err)
		return
	}
	key := pageKeyOf(dd, e)
	item := pageIndex[key]
	if e.Delivered != nil {
		if item != nil && !item.memory {
			delete(pageIndex, key) // the disk says it all now
		}
		pageOutboxMu.Unlock()
		return
	}
	if item != nil && (item.delivered || now.Before(item.next)) {
		pageOutboxMu.Unlock()
		return // delivered (its mark is the index's to write) or backing off
	}
	if item == nil && now.Before(e.Next) {
		pageOutboxMu.Unlock()
		return
	}
	// The index's attempts and deadline, when the disk could not record
	// them, stand over the disk's until a write lands (re-review 3).
	if item != nil && item.e.Tries > e.Tries {
		e.Tries = item.e.Tries
	}
	e.Tries++
	e.Next = now.UTC().Add(pageBackoff(e.Tries))
	if err := writePageEntry(dd, e); err != nil {
		pageIndex[key] = &pageIndexItem{e: e, next: e.Next}
		pagingLogf("moves: %s: the %s page of move %s is sent with its backoff kept in memory: the outbox cannot record it: %v", e.Host, e.Rule, e.Op, err)
	} else if item != nil {
		delete(pageIndex, key)
	}
	pageOutboxMu.Unlock()
	deliverPageEntry(ctx, page, dd, e, now)
}

// flushPageIndex sends each memory page whose backoff is over (a delivered
// one is marked delivered, a refused one waits for its next attempt), and
// writes each delivered page's tombstone, dropping it from the index once
// it lands (fix round 3, ruling 3).
func flushPageIndex(ctx context.Context, now time.Time) {
	pageOutboxMu.Lock()
	var due []*pageIndexItem
	for key, item := range pageIndex {
		switch {
		case item.delivered:
			if err := markPageDelivered(key.dataDir, item.e, now); err == nil {
				delete(pageIndex, key)
			}
		case item.memory && !now.Before(item.next):
			item.e.Tries++
			item.next = now.UTC().Add(pageBackoff(item.e.Tries))
			due = append(due, item)
		}
	}
	pageOutboxMu.Unlock()
	for _, item := range due {
		e := item.e
		if err := item.page(ctx, e.Level, e.Subject, e.Body); err != nil {
			pagingLogf("moves: %s: WARNING: the %s page of move %s, kept in memory only, was not delivered (try %d): %v", e.Host, e.Rule, e.Op, e.Tries, err)
			continue
		}
		pagingLogf("moves: %s: the %s page of move %s, kept in memory, is delivered", e.Host, e.Rule, e.Op)
		pageOutboxMu.Lock()
		item.delivered = true
		pageOutboxMu.Unlock()
	}
}

// retirePageTombstones removes each delivered disk entry whose move is
// retired on this box: no record in recs carries it and its journal is gone.
// Only a pass that listed every record calls it.
func (w *movesWorker) retirePageTombstones(recs []provision.Record) {
	dd := w.cfg.DataDir
	if dd == "" {
		return
	}
	live := map[string]bool{}
	for _, rec := range recs {
		if rec.Move != nil && rec.Move.ID != "" {
			live[rec.Host+"/"+rec.Move.ID] = true
		}
	}
	root := pageOutboxDir(dd)
	err := walkPageOutbox(root, func(p string) error {
		pageOutboxMu.Lock()
		defer pageOutboxMu.Unlock()
		e, err := readPageEntry(p)
		if err != nil || e.Delivered == nil || live[e.Host+"/"+e.Op] || moveJournalPresent(dd, e.Host, e.Op) {
			return nil
		}
		if _, held := pageIndex[pageKeyOf(dd, e)]; held {
			return nil
		}
		if err := os.Remove(p); err != nil {
			pagingLogf("moves: the page outbox: %v", err)
		}
		return nil
	})
	if err != nil {
		pagingLogf("moves: the page outbox: %v", err)
	}
	pruneEmptyDirs(root)
}

// firePage is the worker's firePage: its pager, outbox and clock.
func (w *movesWorker) firePage(ctx context.Context, host, op, rule, level, subject, body string) pageOutcome {
	return firePage(ctx, w.cfg.Page, &w.pageWarned, w.cfg.DataDir, host, op, rule, w.now(), level, subject, body)
}

// pageStuck pages an operator once a MoveMarkStuck has landed (moves spec
// §13: stuck is an incident; K11), once per entry into stuck. role is this
// box's part in the move, "source" or "target", and reason the recorded
// stuck sentence. The page names the host, the move, the box and the
// reason, and nothing else: no path, no error text, no secret.
func (w *movesWorker) pageStuck(ctx context.Context, rec provision.Record, role, reason string) {
	m := rec.Move
	if m == nil || m.ID == "" {
		return
	}
	subject, body := stuckPage(rec.Host, m.ID, w.cfg.Box, role, reason)
	w.firePage(ctx, rec.Host, m.ID, pageRuleStuck(m), alerting.LevelIncident, subject, body)
}

// pageRolledBack pages a warning once this box has closed a move rolled
// back (§13): rec is the record as the close found it.
func (w *movesWorker) pageRolledBack(ctx context.Context, rec provision.Record) {
	if rec.Move == nil || rec.Move.ID == "" {
		return
	}
	subject, body := rolledBackPage(rec.Host, rec.Move.ID, w.cfg.Box)
	w.firePage(ctx, rec.Host, rec.Move.ID, pageRuleRolledBack, alerting.LevelWarning, subject, body)
}

// strayRefused reports whether a stop evidence report was refused for a
// process the move cannot account for: its StepStray ran and failed. A step
// that never ran is not a refusal.
func strayRefused(rep moves.StoppedReport) bool {
	c, ok := rep.Steps[moves.StepStray]
	return ok && !c.OK
}

// pageStray pages an incident once per move for a stop evidence refused by a
// stray process (§13). The page's body is M-C20, and the agent's log says
// M-C20, with the move, when the box first detects it (the firing that keeps
// the page), whether or not the page is delivered then (fix round 2,
// ruling 1). rec is the cut's record.
func (w *movesWorker) pageStray(ctx context.Context, rec provision.Record) {
	if rec.Move == nil || rec.Move.ID == "" {
		return
	}
	subject, body := strayPage(rec.Host, w.cfg.Box)
	switch w.firePage(ctx, rec.Host, rec.Move.ID, pageRuleStray, alerting.LevelIncident, subject, body) {
	case pageSent, pageUndelivered:
		pagingLogf("moves: %s: move %s: %s", rec.Host, rec.Move.ID, strayLine(w.cfg.Box, rec.Host))
	}
}

// pageStartRefused pages an incident once per move for a refused §5.1 start
// check on this box (§13), and returns what became of the page. Its outbox
// is the box's, which the moves pass retries.
func (m *movesProbeOwners) pageStartRefused(ctx context.Context, host, op string) pageOutcome {
	now := time.Now()
	if m.Now != nil {
		now = m.Now()
	}
	subject, body := startRefusedPage(host, op, m.Box)
	return firePage(ctx, m.Page, &m.pageWarned, m.DataDir, host, op, pageRuleStartCheck, now, alerting.LevelIncident, subject, body)
}

// rolledBackPage is the rolled_back page: the subject (MV-N385) and the
// body (MV-N386).
func rolledBackPage(host, id, box string) (subject, body string) {
	subject = fmt.Sprintf("carlos move rolled back: %s", host)
	body = fmt.Sprintf("The move %s of %s rolled back, and %s is serving on %s again.", id, host, host, box)
	return subject, body
}

// strayPage is the stray process's page: the subject (MV-N387) and the
// body, M-C20.
func strayPage(host, box string) (subject, body string) {
	subject = fmt.Sprintf("carlos move stray process: %s", host)
	return subject, strayLine(box, host)
}

// strayLine is M-C20, the operators' sentence for the stray process: the
// stray page's body and the agent's log line.
func strayLine(box, host string) string {
	return fmt.Sprintf("%s is running a process for %s that this move cannot account for. An operator has been paged.", box, host)
}

// startRefusedPage is the refused start check's page: the subject
// (MV-N389) and the body (MV-N390).
func startRefusedPage(host, id, box string) (subject, body string) {
	subject = fmt.Sprintf("carlos move start refused: %s", host)
	body = fmt.Sprintf("%s refused to start %s for the move %s: what it holds of %s is not the generation the host claim names. Nothing was started.",
		box, host, id, host)
	return subject, body
}