package provision

import (
	"context"
	"errors"
	"log"
	"os"

	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/seeds"
)

// SeedHooks is what the provision pass needs to act on a seeded record
// (seeded-instances spec §3.2). Nil on ReconcileConfig: no seeded record is
// minted or acted on, which is the safe answer for a box that cannot place
// one.
type SeedHooks struct {
	// Data is this region's data bucket (seeds/ lives there).
	Data release.Store
	// LegacyOwners reads the operator assignment of legacy route names
	// (spec §6.4): the swap's only authority over a row filed under the
	// operator account. Nil, or an error, defers such a swap.
	LegacyOwners func(ctx context.Context) (*seeds.LegacyOwners, error)
	// NestedOK is the box's live nested probe, asked after the layout
	// intent says nested (spec §5.4 "Layout"). An error defers the swap to
	// the next pass; it never refuses (only the intent does). Nil: the
	// intent alone decides.
	NestedOK func() error
	// EnvBefore reports what the host's own env file holds, at the nested
	// path beside dbPath, as one of registry's EnvBefore values. The swap
	// asks it before the import's row exists and keeps the answer in the
	// journal, so a rollback can tell the instance key the import's first
	// launch minted from one that was there before (RetireKey). Nil, or an
	// error, defers the swap.
	EnvBefore func(dbPath, host string) (string, error)

	// The rollback's box effects (spec §5.6, seedrollback.go). All seven
	// or none: a box without them leaves every abandoned import journaled,
	// gated and untouched.
	//
	// FenceForRollback fences and stops host under the rollback's fence
	// generation, by the layout pass's rule: a fence already holding gen is
	// this rollback's own from an earlier pass. proceed false defers the
	// rollback, and the hook has lifted any fence it installed.
	FenceForRollback func(ctx context.Context, host string, gen uint64) (proceed bool, err error)
	// LiftFence lifts the fence gen, if it is the one host holds.
	LiftFence func(host string, gen uint64)
	// RemoveLocal removes the import's local data for its row (the
	// database, -wal, -shm, <db>.seed, litestream's metadata and any seed
	// fetch leftovers) through instancefs. Nothing there is not an error.
	RemoveLocal func(rt registry.Route) error
	// DeleteReplica deletes host's instances/ replica in the region data
	// bucket. Nothing there is not an error.
	DeleteReplica func(ctx context.Context, host string) error
	// NextFenceGen mints a fence generation no fence on this box holds,
	// never one it minted before, across agent restarts: the journal
	// stores it, and a resumed rollback recognises and lifts its fence by
	// it. An error, or 0, closes nothing this pass.
	NextFenceGen func() (uint64, error)
	// LocalSeedID is the seed id the <db>.seed marker beside dbPath names
	// ("" for none or no marker), read through instancefs. An error is a
	// marker that could not be read.
	LocalSeedID func(dbPath string) (string, error)
	// RetireKey retires the instance key the import created in the host's
	// own env file beside dbPath, by the journal's EnvBefore: the whole
	// file when there was none, its key lines when the file had no key,
	// nothing when the key was there before or the journal does not say.
	// It never removes a key or a line that predates the import, and it is
	// not part of RemoveLocal. Nothing there is not an error.
	RetireKey func(dbPath, host, envBefore string) error
}

type seedVerdict int

const (
	seedOrdinary seedVerdict = iota
	seedSkip
	seedFinishAbort
	seedSwap
)

// seedAction is §3.2's dispatch, run before any other per-record handling.
// row is this box's row for the host, nil when it holds none. nested is
// asked only for a mint, through the pass's lazy layout reader. journalOK
// is whether this pass read the import journal: when it could not, no
// import is acted on, since the journal decides every import it holds a
// row for (spec §5.7) and the no-journal paths must never run while a
// journal row might exist.
func seedAction(rec Record, row *registry.Route, cfg ReconcileConfig, nested func() (bool, error), journalOK bool) seedVerdict {
	src := rec.Seed
	if src == nil {
		return seedOrdinary
	}
	if cfg.Seeds == nil {
		return seedSkip
	}
	if src.Import() && !journalOK {
		log.Printf("provision: %s: import %s left alone this pass: the import journal could not be read", rec.Host, src.ID)
		return seedSkip
	}
	hasRow := row != nil
	switch src.State {
	case seeds.StateAborted:
		return seedFinishAbort
	case seeds.StateRolledBack, seeds.StateReserving, seeds.StateAwaitingUpload, seeds.StateUploaded, seeds.StateRefused:
		return seedSkip
	case seeds.StateReady:
		if src.Import() {
			return seedSwap // seedswap.go
		}
		if hasRow {
			return seedOrdinary
		}
		if cfg.ClaimEnabled {
			log.Printf("provision: %s: not minting a seeded instance: claim control is on for this box", rec.Host)
			return seedSkip
		}
		// Only an activator-managed row is minted nested (routeFor): any
		// other row would start without its seed and without the seed
		// command, which is an empty database.
		if !rec.DerivesPaths() || !rec.HibernateOrDefault() {
			log.Printf("provision: %s: not minting a seeded instance: its record does not describe a hibernating instance", rec.Host)
			return seedSkip
		}
		if ok, err := nested(); err != nil || !ok {
			log.Printf("provision: %s: not minting a seeded instance: this box does not mint nested (%v)", rec.Host, err)
			return seedSkip
		}
		return seedOrdinary
	case seeds.StateSwapped, seeds.StatePlaced:
		// Before committed, an import's row is made only by the swap,
		// gated (spec §5.4, §8). A box with no row, or the swapping box
		// after its gated row was removed, must not mint one: routeFor
		// never sets the gate, so ordinary traffic would reach an
		// unverified import.
		if src.Import() && !hasRow {
			log.Printf("provision: %s: not minting import %s: it is %s but not committed, and only its swap makes its row", rec.Host, src.ID, src.State)
			return seedSkip
		}
		// The swap's row is gated until commit. An ungated row here was
		// put there by hand, and adopting or converging it would serve
		// the unverified import to ordinary traffic.
		if src.Import() && row.ImportGate == "" {
			log.Printf("provision: %s: not converging import %s: it is %s but not committed, and the row on this box has no import gate", rec.Host, src.ID, src.State)
			return seedSkip
		}
		return seedOrdinary
	case seeds.StateCommitted:
		return seedOrdinary
	}
	log.Printf("provision: %s: seed %s is in an unknown state %q; leaving it alone", rec.Host, src.ID, src.State)
	return seedSkip
}

// retireAbortedSeedRow removes the row of an aborted --from seed that this
// box minted (task-4 review I1). The seed can be aborted after its publish
// landed (a lost publish answer), and then a box may have minted the row and
// be running it, holding the host's lease and writing its replica. The
// console's finish deletes that replica, so the row must stop first. The
// record stays aborted this pass; a later pass that finds no row rolls the
// seed back (finishAbortedSeed).
//
// It removes the row exactly as the removal pass removes a row whose record
// is gone, with the same authority: a Provisioned row of this pass's own
// kind and scope, not a running operation's participant, through
// claimRetire (reg.Delete, which the activator follows by stopping it). It
// also requires the row to be this record's (account and app). A row it
// may not remove is left alone, logged, and the seed stays aborted.
//
// Only --from seeds: an import's row is the instance the import replaces,
// which its rollback restores (Plan 7) and never removes.
func retireAbortedSeedRow(ctx context.Context, reg *registry.Registry, cfg ReconcileConfig, rec Record, rt registry.Route) {
	if rt.Kind == SidecarKind || !rt.Provisioned || !cfg.inScope(rt.Region, rt.Pool, everywhereRoute(rt)) ||
		rt.Account != rec.Account || rt.App != rec.App {
		log.Printf("provision: %s: aborted seed %s has a row this pass may not remove; leaving the seed aborted", rec.Host, rec.Seed.ID)
		return
	}
	if cfg.KeepParticipant != nil && cfg.KeepParticipant(rt.Host) {
		log.Printf("provision: %s: aborted seed %s: not removing its row: participant of a running operation", rec.Host, rec.Seed.ID)
		return
	}
	if cfg.DryRun {
		log.Printf("provision: dry-run: would remove %s route %s (its seed %s was aborted)", rt.Kind, rt.Host, rec.Seed.ID)
		return
	}
	if err := cfg.claimRetire(ctx, reg, rt); err != nil {
		log.Printf("provision: remove %s route %s (its seed %s was aborted): %v", rt.Kind, rt.Host, rec.Seed.ID, err)
		return
	}
	log.Printf("provision: removed %s route %s (its seed %s was aborted); the next pass rolls it back", rt.Kind, rt.Host, rec.Seed.ID)
}

// finishAbortedSeed is §3.2's no-journal path for an aborted seed: remove
// a --from seed's local database (purgeAbortedFromLocal), delete this
// seed's snapshot (only this seed's id, codex R4-1), then CAS the record to
// rolled-back. Idempotent: an absent snapshot and a lost CAS are both done.
// The console finishes from rolled-back (Plan 8).
//
// A host or id that fails seeds.Check names no snapshot (no valid key
// could hold one), so nothing is deleted and the record is still rolled
// back.
func finishAbortedSeed(ctx context.Context, s release.Store, reg *registry.Registry, cfg ReconcileConfig, rec Record) {
	if cfg.DryRun {
		log.Printf("provision: dry-run: would finish aborted seed %s of %s", rec.Seed.ID, rec.Host)
		return
	}
	if rec.Seed.Kind == seeds.KindReplica && !purgeAbortedFromLocal(ctx, reg, cfg, rec) {
		return
	}
	if err := seeds.Check(rec.Host, rec.Seed.ID); err != nil {
		log.Printf("provision: %s: aborted seed %q names no snapshot: %v", rec.Host, rec.Seed.ID, err)
	} else {
		key := seeds.SnapshotKey(rec.Host, rec.Seed.ID)
		if err := cfg.Seeds.Data.Delete(ctx, key); err != nil && !errors.Is(err, os.ErrNotExist) {
			log.Printf("provision: %s: delete aborted seed's snapshot: %v", rec.Host, err)
			return
		}
	}
	if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedRolledBack(rec.Seed.ID)); err != nil && !errors.Is(err, ErrSeedConflict) {
		log.Printf("provision: %s: mark seed %s rolled back: %v", rec.Host, rec.Seed.ID, err)
		return
	}
	log.Printf("provision: %s: aborted seed %s rolled back (no box had swapped it)", rec.Host, rec.Seed.ID)
}

// rollbackArmed reports whether h has the rollback's seven box hooks.
func rollbackArmed(h *SeedHooks) bool {
	return h != nil && h.FenceForRollback != nil && h.LiftFence != nil && h.RemoveLocal != nil && h.DeleteReplica != nil &&
		h.NextFenceGen != nil && h.LocalSeedID != nil && h.RetireKey != nil
}

// purgeAbortedFromLocal removes an aborted --from seed's local database
// before the seed is rolled back (final review F2), on a pass that found
// no row for the host. The seed may have been aborted after its publish
// landed (a lost publish answer), and a wake then placed its database at
// the host's default nested path, where a later plain instance at the
// host would open it as a warm cache. The console's finish deletes the
// replica and the record, so nothing else would ever remove the file.
//
// It is Plan 7's rule for an import whose row is gone (removeOrphanedLocal),
// decided from the file: the database goes only when its <db>.seed marker
// names this seed and the host still has no row, with the host fenced and
// its stop proven. The marker is on disk, so this holds after a restart
// too. A marker that is absent or names another seed does not prove the
// data is this seed's: logged, and the seed rolls back with the file left.
// A marker that cannot be read, a fence that does not proceed, a row that
// appears or a removal that fails defers the rollback (ok false).
//
// A box without the rollback's hooks places no seed (they are armed with
// the seed reader, and only claim control leaves them off, which refuses
// every seed), so it has nothing of this seed's to remove: ok.
func purgeAbortedFromLocal(ctx context.Context, reg *registry.Registry, cfg ReconcileConfig, rec Record) (ok bool) {
	h := cfg.Seeds
	if !rollbackArmed(h) {
		log.Printf("provision: %s: aborted seed %s: this box cannot read seed markers, and places no seed; nothing local is removed", rec.Host, rec.Seed.ID)
		return true
	}
	db := NestedDBPath(cfg.DataDir, rec.Host)
	ours := func() (bool, bool) {
		id, err := h.LocalSeedID(db)
		if err != nil {
			log.Printf("provision: %s: aborted seed %s stays aborted: the seed marker beside %s cannot be read: %v", rec.Host, rec.Seed.ID, db, err)
			return false, false
		}
		if id != rec.Seed.ID {
			log.Printf("provision: %s: aborted seed %s rolls back with the database at %s left: its seed marker names %q, not this seed", rec.Host, rec.Seed.ID, db, id)
			return false, true
		}
		return true, true
	}
	if mine, ok := ours(); !mine {
		return ok
	}
	gen, err := h.NextFenceGen()
	if err == nil && gen == 0 {
		err = errors.New("the generation source answered 0")
	}
	if err != nil {
		log.Printf("provision: %s: aborted seed %s stays aborted: no fence generation: %v", rec.Host, rec.Seed.ID, err)
		return false
	}
	proceed, err := h.FenceForRollback(ctx, rec.Host, gen)
	defer h.LiftFence(rec.Host, gen) // a hook that failed may have fenced first
	if err != nil || !proceed {
		log.Printf("provision: %s: aborted seed %s stays aborted: the host is not fenced and stopped yet (%v)", rec.Host, rec.Seed.ID, err)
		return false
	}
	// Under the fence: still no row, and the marker still names this seed.
	row, err := reg.Get(rec.Host)
	if err != nil {
		log.Printf("provision: %s: aborted seed %s stays aborted: the route could not be read: %v", rec.Host, rec.Seed.ID, err)
		return false
	}
	if row != nil {
		log.Printf("provision: %s: aborted seed %s stays aborted: a route appeared at the host, so its local database is left", rec.Host, rec.Seed.ID)
		return false
	}
	if mine, ok := ours(); !mine {
		return ok
	}
	if err := h.RemoveLocal(registry.Route{Host: rec.Host, Kind: "instance", DBPath: db}); err != nil {
		log.Printf("provision: %s: aborted seed %s stays aborted: remove its local data: %v", rec.Host, rec.Seed.ID, err)
		return false
	}
	log.Printf("provision: %s: aborted seed %s has no route on this box: its local data is removed", rec.Host, rec.Seed.ID)
	return true
}

// journalPass is spec §5.7: the box's import journal is read before the
// records are acted on, every pass, and each row drives its host (§5.4
// "Crashes", §5.6). handled hosts are skipped by the record loop;
// journaled hosts (every journal row) by the removal phase, since the
// journal owns the row until the row is gone.
//
// The journal is read whatever cfg.Seeds is (preflight M1): with seed
// hooks unarmed every journaled host is still protected and handled,
// with no action. With claim control on for this box it is the same:
// every journaled host is protected and handled, and nothing moves
// forward (no rollback, no redone record CAS, no cleared gate). ok is
// false when the journal cannot be read. Reconcile then treats the pass
// as a failed read for removal, and seedAction acts on no import.
//
// A journal row at swapped is decided by the record that answers to it,
// which is only ever the record under the journal's own key
// (journalRecord): another record for the same host can only vouch for
// the import, never revoke it.
//
//   - gone, no seed, another seed, or this seed aborted, refused or rolled
//     back: the import is abandoned, so it is rolled back. A refused
//     record with a journal row is a refusal that lost the race to this
//     box's swap: the journal wins;
//   - this seed at ready: the swap's record CAS is redone;
//   - this seed at committed: the gate is cleared and the journal row
//     dropped (CommitImport), and the loop converges the row;
//   - this seed at swapped or placed: the loop handles it as usual.
//
// The loop converges a host only when the pass's own record for it is the
// journal's: when another record won the host, or the journal's record is
// out of scope, the host stays handled, so no other record touches the
// row. Every other case leaves the host alone this pass, logged.
func journalPass(ctx context.Context, s release.Store, reg *registry.Registry, cfg ReconcileConfig,
	recs map[string]Record, readFailed bool, protect map[string]bool) (handled, journaled map[string]bool, ok bool) {
	handled, journaled = map[string]bool{}, map[string]bool{}
	rows, err := reg.ImportJournals()
	if err != nil {
		log.Printf("provision: the import journal could not be read: %v (no import is acted on and no route is removed this pass)", err)
		return handled, journaled, false
	}
	for _, j := range rows {
		journaled[j.Host] = true
		handled[j.Host] = true // cleared below for the two cases the loop converges
		if cfg.Seeds == nil {
			log.Printf("provision: %s: import %s is journaled, but this box cannot act on imports; its row is left alone", j.Host, j.SeedID)
			continue
		}
		// With claim control on, a journal row is only protected (merge
		// review S2). The box arms no swap and no rollback there, and the
		// claim console refuses commit, abort and finish, so nothing this
		// pass could do would be one step of a whole: redoing the swap's
		// record CAS writes a record, and clearing a committed gate opens
		// the row to ordinary traffic, each on a box whose claim has not
		// taken the host. The host stays handled (the loop converges
		// nothing) and journaled (the removal phase leaves its row), until
		// the claim's own seed algorithm exists (#250). Activation refuses
		// while a journal row exists, so this is an import that was caught
		// by a claim that came on anyway.
		if cfg.ClaimEnabled {
			log.Printf("provision: %s: import %s is journaled at %s and claim control is on for this box; its row is protected and nothing else is done", j.Host, j.SeedID, j.State)
			continue
		}
		rec, inPass, known := journalRecord(ctx, s, j, recs, readFailed, protect)
		if !known {
			log.Printf("provision: %s: import %s left alone this pass: its record could not be read", j.Host, j.SeedID)
			continue
		}
		// journalRecord answers only with the journal's own record; this
		// holds before any decision below, rollbacks first of all.
		if rec != nil && RecordKey(rec.Account, rec.App, rec.Host) != j.RecordKey {
			log.Printf("provision: %s: import %s left alone: the journal holds it under %s, not %s", j.Host, j.SeedID,
				j.RecordKey, RecordKey(rec.Account, rec.App, rec.Host))
			continue
		}
		switch j.State {
		case registry.JournalCleanupPending, registry.JournalRollingBack:
			startRollback(ctx, s, reg, cfg, j, rec)
			continue
		case registry.JournalSwapped:
		default:
			log.Printf("provision: %s: import %s is journaled in an unknown state %q; leaving it alone", j.Host, j.SeedID, j.State)
			continue
		}
		var seed *SeedSource
		if rec != nil {
			seed = rec.Seed
		}
		switch {
		case seed == nil || seed.ID != j.SeedID:
			startRollback(ctx, s, reg, cfg, j, rec)
		case seed.State == seeds.StateAborted || seed.State == seeds.StateRefused || seed.State == seeds.StateRolledBack:
			startRollback(ctx, s, reg, cfg, j, rec)
		case seed.State == seeds.StateReady:
			if cfg.DryRun {
				log.Printf("provision: dry-run: would redo %s's swap record CAS for import %s", j.Host, j.SeedID)
			} else {
				markSwapped(ctx, s, *rec)
			}
		case seed.State == seeds.StateCommitted:
			if cfg.DryRun {
				log.Printf("provision: dry-run: would clear %s's import gate", j.Host)
			} else if err := reg.CommitImport(j.Host, j.SeedID); err != nil {
				log.Printf("provision: %s: clear the import gate: %v", j.Host, err)
			} else {
				log.Printf("provision: %s: import %s committed: its gate is cleared and its journal row dropped", j.Host, j.SeedID)
			}
			handled[j.Host] = !inPass
		case seed.State == seeds.StateSwapped || seed.State == seeds.StatePlaced:
			handled[j.Host] = !inPass
		default:
			// reserving, awaiting-upload and uploaded come before any swap.
			log.Printf("provision: %s: import %s is journaled but its record says %s; leaving it alone", j.Host, j.SeedID, seed.State)
		}
	}
	return handled, journaled, true
}

// startRollback hands an abandoned import to its rollback, which a dry
// run never starts.
func startRollback(ctx context.Context, s release.Store, reg *registry.Registry, cfg ReconcileConfig, j registry.ImportJournal, rec *Record) {
	if cfg.DryRun {
		log.Printf("provision: dry-run: would roll back %s's import %s", j.Host, j.SeedID)
		return
	}
	rollbackImport(ctx, s, reg, cfg, j, rec)
}

// journalRecord is the record under the journal's own key, nil when it is
// gone, and known false when this pass cannot tell. inPass says it is the
// record the pass holds for the host, so the loop may converge with it.
//
// recs holds the first in-scope record per host that read cleanly, which
// need not be the journal's: another app's or account's record can win a
// duplicate host, or hold the host while the journal's record is out of
// this box's scope. So recs answers only when its record's key is the
// journal's; otherwise the record is read by the journal's key, and only
// os.ErrNotExist means gone. A protected host, a failed read, a journal
// row with no key, and a body under the key that names another host or
// key all decide nothing this pass: a committed import must never be
// rolled back on a guess.
func journalRecord(ctx context.Context, s release.Store, j registry.ImportJournal, recs map[string]Record,
	readFailed bool, protect map[string]bool) (rec *Record, inPass, known bool) {
	if protect[j.Host] || j.RecordKey == "" {
		return nil, false, false
	}
	if r, ok := recs[j.Host]; ok && RecordKey(r.Account, r.App, r.Host) == j.RecordKey {
		return &r, true, true
	}
	if readFailed {
		return nil, false, false
	}
	r, err := journalRecordByKey(ctx, s, j)
	if err != nil {
		log.Printf("provision: %s: the import's record %s is not one this pass can use (%v)", j.Host, j.RecordKey, err)
		return nil, false, false
	}
	return r, false, true
}