package provision

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

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

// rollbackImport is spec §5.6's rollback, driven by the journal and not the
// record (codex R2-7). Each step is idempotent, and the journal's state is
// where a later pass resumes:
//
//	swapped          1. close the gate (journal rolling-back, with a fence generation)
//	rolling-back     2. fence and stop the host under the journal's generation
//	                 3. retire the instance key the import created, then
//	                    remove the import's local data
//	                 4. put the original route back (journal cleanup-pending)
//	                 5. lift the fence
//	cleanup-pending  6. delete this seed's snapshot and the host's replica
//	                 7. delete the journal row
//	                 8. CAS the record to rolled-back, if it names this seed
//
// The route steps act only on the import's own row (importRowOf). Any other
// row at the host belongs to someone else, a later instance when a stale
// journal row came back from a snapshot, and the rollback holds there,
// touching nothing, until an operator looks. A host with no row at all is
// rolled back without a restore: whatever removed the import's row decided
// the host should not be served. Its local database at the host's default
// path is still removed when the database's <db>.seed marker names this
// seed and the host still has no row.
//
// Step 3 retires the import's instance key first (merge review S1). The
// key the import's first launch minted into the host's own env file would
// otherwise outlive the rollback and outrank the key a second import of
// the host configures. The journal's EnvBefore, recorded at the swap, says
// what was there before the import: only what the import created goes
// (SeedHooks.RetireKey). With the row gone the key is retired under the
// same marker rule as the database, and before it, so the marker that
// proves the installation is the import's is still there if the key's
// removal has to be tried again.
//
// rec is the pass's view of the journal's record (nil when gone). It chose
// the rollback; the destructive cleanup reads the record again (step 6).
func rollbackImport(ctx context.Context, s release.Store, reg *registry.Registry, cfg ReconcileConfig, j registry.ImportJournal, rec *Record) {
	h := cfg.Seeds
	if cfg.DryRun {
		log.Printf("provision: dry-run: would roll back %s's import %s", j.Host, j.SeedID)
		return
	}
	if !rollbackArmed(h) {
		log.Printf("provision: %s: import %s cannot roll back on this box: the activator is not wired; it stays journaled and gated", j.Host, j.SeedID)
		return
	}
	switch j.State {
	case registry.JournalSwapped, registry.JournalRollingBack:
		var ok bool
		if j, ok = rollBackRoute(ctx, s, reg, cfg, j); !ok {
			return
		}
	case registry.JournalCleanupPending:
	default:
		log.Printf("provision: %s: import %s is journaled in an unknown state %q; leaving it alone", j.Host, j.SeedID, j.State)
		return
	}
	cleanUpImport(ctx, s, reg, cfg, j)
}

// importRowOf reports whether rt is j's import row as the swap made it:
// the registry's test (gated, provisioned, naming the seed, whose ids are
// never reused), at the nested database path the swap gave it. The path is
// what step 3 removes at, so a row pointing anywhere else is not acted on.
func importRowOf(cfg ReconcileConfig, rt registry.Route, j registry.ImportJournal) bool {
	return registry.IsImportRow(rt, j) && rt.DBPath == NestedDBPath(cfg.DataDir, j.Host)
}

// rollBackRoute is steps 1 to 5. It returns the journal as it now stands
// and whether the route steps are done (the journal is cleanup-pending).
func rollBackRoute(ctx context.Context, s release.Store, reg *registry.Registry, cfg ReconcileConfig, j registry.ImportJournal) (registry.ImportJournal, bool) {
	h := cfg.Seeds
	// Nothing is closed, fenced or removed for a row that is not the
	// import's. The registry refuses the gate close and the restore for
	// one as well; this check keeps the fence and the removal off it.
	row, err := reg.Get(j.Host)
	if err != nil {
		log.Printf("provision: %s: import %s rollback deferred: the route could not be read: %v", j.Host, j.SeedID, err)
		return j, false
	}
	if row != nil && !importRowOf(cfg, *row, j) {
		// A fence an earlier pass kept for this rollback must not hold the
		// host for whoever owns it now. LiftFence lifts only this
		// generation.
		if j.State == registry.JournalRollingBack {
			h.LiftFence(j.Host, j.FenceGen)
		}
		log.Printf("provision: %s: import %s is not rolled back: the route at the host is not the import's, so nothing there is touched", j.Host, j.SeedID)
		return j, false
	}

	// Step 1. The generation an earlier close stored is the one the fence
	// is known by, so the journal is read back rather than trusted.
	if j.State == registry.JournalSwapped {
		gen, err := h.NextFenceGen()
		if err == nil && gen == 0 {
			err = errors.New("the generation source answered 0")
		}
		if err != nil {
			log.Printf("provision: %s: import %s rollback deferred: no fence generation: %v", j.Host, j.SeedID, err)
			return j, false
		}
		if err := reg.CloseImportGate(j.Host, j.SeedID, gen); err != nil {
			log.Printf("provision: %s: import %s: close the import gate: %v", j.Host, j.SeedID, err)
			return j, false
		}
		cur, err := reg.ImportJournal(j.Host)
		if err != nil || cur == nil || cur.SeedID != j.SeedID || cur.State != registry.JournalRollingBack {
			log.Printf("provision: %s: import %s: the journal after the gate closed is %+v (%v)", j.Host, j.SeedID, cur, err)
			return j, false
		}
		j = *cur
	}

	// Step 2. The host is fenced and stopped before anything is removed,
	// so no wake can write the database or the replica afterwards. The
	// hook lifts any fence it installs when it does not proceed.
	proceed, err := h.FenceForRollback(ctx, j.Host, j.FenceGen)
	if err != nil || !proceed {
		if err != nil {
			h.LiftFence(j.Host, j.FenceGen) // a hook that failed may have fenced first
		}
		log.Printf("provision: %s: import %s rollback deferred: the host is not fenced and stopped yet (%v)", j.Host, j.SeedID, err)
		return j, false
	}

	// Step 3, under the fence. A failure from here on keeps the fence: the
	// next pass finds it its own by the journal's generation and resumes.
	row, err = reg.Get(j.Host)
	if err != nil {
		log.Printf("provision: %s: import %s: read the route under the fence: %v", j.Host, j.SeedID, err)
		return j, false
	}
	switch {
	case row == nil:
		if !removeOrphanedLocal(reg, cfg, j) {
			return j, false
		}
	case !importRowOf(cfg, *row, j) || row.ImportGate != registry.ImportGateClosed:
		h.LiftFence(j.Host, j.FenceGen)
		log.Printf("provision: %s: import %s is not rolled back: the route changed under the fence, so nothing there is touched", j.Host, j.SeedID)
		return j, false
	default:
		if err := h.RetireKey(row.DBPath, j.Host, j.EnvBefore); err != nil {
			log.Printf("provision: %s: import %s: retire its instance key: %v", j.Host, j.SeedID, err)
			return j, false
		}
		if err := h.RemoveLocal(*row); err != nil {
			log.Printf("provision: %s: import %s: remove its local data: %v", j.Host, j.SeedID, err)
			return j, false
		}
	}

	// Step 4: the original route back, and the journal at cleanup-pending,
	// in one transaction. Step 5: the fence goes.
	if err := reg.RestoreImportRoute(j.Host, j.SeedID); err != nil {
		log.Printf("provision: %s: import %s: restore the original route: %v", j.Host, j.SeedID, err)
		return j, false
	}
	h.LiftFence(j.Host, j.FenceGen)
	j.State = registry.JournalCleanupPending
	if row != nil {
		log.Printf("provision: %s: import %s rolled back to its original route", j.Host, j.SeedID)
	}
	return j, true
}

// cleanUpImport is steps 6 to 8, run from cleanup-pending. It never fences
// or stops anything: the route is the original row by now (or none), and
// its process is the app's own.
func cleanUpImport(ctx context.Context, s release.Store, reg *registry.Registry, cfg ReconcileConfig, j registry.ImportJournal) {
	h := cfg.Seeds
	valid := seeds.Check(j.Host, j.SeedID) == nil
	if !valid {
		log.Printf("provision: %s: import %q names no snapshot and no replica; only its journal row is cleaned up", j.Host, j.SeedID)
	} else if err := h.Data.Delete(ctx, seeds.SnapshotKey(j.Host, j.SeedID)); err != nil && !errors.Is(err, os.ErrNotExist) {
		log.Printf("provision: %s: import %s: delete its snapshot: %v", j.Host, j.SeedID, err)
		return
	}

	// The replica is keyed by host, not by seed, so the record is read
	// again right before it: the pass's copy may be stale.
	rec, err := journalRecordByKey(ctx, s, j)
	if err != nil {
		log.Printf("provision: %s: import %s cleanup deferred: its record cannot be read: %v", j.Host, j.SeedID, err)
		return
	}
	if valid {
		ours, why, err := replicaIsTheImports(ctx, s, j, rec)
		switch {
		case err != nil:
			log.Printf("provision: %s: import %s cleanup deferred: %v", j.Host, j.SeedID, err)
			return
		case !ours:
			log.Printf("provision: %s: import %s: the host's replica is left alone: %s", j.Host, j.SeedID, why)
		default:
			if err := h.DeleteReplica(ctx, j.Host); err != nil {
				log.Printf("provision: %s: import %s: delete the host's replica: %v", j.Host, j.SeedID, err)
				return
			}
		}
	}

	if err := reg.DeleteImportJournal(j.Host, j.SeedID); err != nil {
		log.Printf("provision: %s: import %s: retire its journal row: %v", j.Host, j.SeedID, err)
		return
	}
	log.Printf("provision: %s: import %s is cleaned up and its journal row retired", j.Host, j.SeedID)

	// Step 8. Only from aborted (SeedRolledBack): a record already
	// refused or rolled back, or gone, is done. An aborted record whose
	// CAS fails is finished by the no-journal path on a later pass.
	if rec == nil || rec.Seed == nil || rec.Seed.ID != j.SeedID {
		return
	}
	if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedRolledBack(j.SeedID)); err != nil &&
		!errors.Is(err, ErrSeedConflict) && !errors.Is(err, os.ErrNotExist) {
		log.Printf("provision: %s: mark import %s rolled back: %v", j.Host, j.SeedID, err)
	}
}

// removeOrphanedLocal is step 3 for an import whose row something removed,
// run under the fence. Its local database is still at the host's default
// nested path (routeFor's), where a later instance at the host would open
// it as a warm cache and replicate it as its own. It is decided from the
// file, whoever holds the host's record: while the journal row exists no
// other row can be minted at the host, so under the fence nothing else
// has that file open. The database is removed only when its <db>.seed
// marker names this seed and the host still has no row. A marker that is
// absent or names another seed does not prove the data is the import's,
// so the file is left. The import's instance key is retired by the same
// rule, just before the database, and left with it. ok false defers the
// rollback, with the fence lifted.
func removeOrphanedLocal(reg *registry.Registry, cfg ReconcileConfig, j registry.ImportJournal) (ok bool) {
	h := cfg.Seeds
	db := NestedDBPath(cfg.DataDir, j.Host)
	id, err := h.LocalSeedID(db)
	if err != nil {
		h.LiftFence(j.Host, j.FenceGen)
		log.Printf("provision: %s: import %s rollback deferred: the seed marker beside %s cannot be read: %v", j.Host, j.SeedID, db, err)
		return false
	}
	if id != j.SeedID {
		log.Printf("provision: %s: import %s has no route on this box; the database at %s is left: its seed marker names %q, not this seed", j.Host, j.SeedID, db, id)
		return true
	}
	row, err := reg.Get(j.Host)
	if err != nil {
		h.LiftFence(j.Host, j.FenceGen)
		log.Printf("provision: %s: import %s rollback deferred: the route could not be read: %v", j.Host, j.SeedID, err)
		return false
	}
	if row != nil {
		// Not expected while journaled. The next pass judges the row.
		h.LiftFence(j.Host, j.FenceGen)
		log.Printf("provision: %s: import %s: a route appeared at the host, so its local database is left and the rollback waits", j.Host, j.SeedID)
		return false
	}
	// The key before the database: the marker just read is what says this
	// installation is the import's, and RemoveLocal removes it.
	if err := h.RetireKey(db, j.Host, j.EnvBefore); err != nil {
		log.Printf("provision: %s: import %s: retire its instance key: %v", j.Host, j.SeedID, err)
		return false
	}
	if err := h.RemoveLocal(registry.Route{Host: j.Host, Kind: "instance", DBPath: db}); err != nil {
		log.Printf("provision: %s: import %s: remove its local data: %v", j.Host, j.SeedID, err)
		return false
	}
	log.Printf("provision: %s: import %s has no route on this box: its local data is removed and no route is put back", j.Host, j.SeedID)
	return true
}

// replicaIsTheImports decides step 6's replica delete. The host's replica
// is the import's to delete only while the journal's record, if there is
// one, still names this seed in an abandoned state, and no other record
// holds the host. A record gone with no other record for the host counts.
// A record naming another seed, or none, or another record for the host,
// means the replica is someone else's: ours is false, and why says so.
// Anything this pass cannot tell is an error, and the cleanup waits.
func replicaIsTheImports(ctx context.Context, s release.Store, j registry.ImportJournal, rec *Record) (ours bool, why string, err error) {
	if rec != nil {
		switch {
		case rec.Seed == nil || rec.Seed.ID != j.SeedID:
			return false, "its record no longer names this seed", nil
		case rec.Seed.State != seeds.StateAborted && rec.Seed.State != seeds.StateRefused && rec.Seed.State != seeds.StateRolledBack:
			return false, "", fmt.Errorf("its record names this seed as %s, not abandoned", rec.Seed.State)
		}
	}
	known, failed, err := knownAccounts(ctx, s)
	if err != nil {
		return false, "", err
	}
	if failed {
		return false, "", errors.New("an account could not be read, so another record for the host cannot be ruled out")
	}
	keys, partial, err := listAppSubtree(ctx, s, known, "instances/")
	if err != nil {
		return false, "", err
	}
	if partial {
		return false, "", errors.New("the records could not all be listed, so another record for the host cannot be ruled out")
	}
	for _, key := range keys {
		if _, _, host, ok := ParseRecordKey(key, known); ok && host == j.Host && key != j.RecordKey {
			return false, "the record " + key + " holds the host", nil
		}
	}
	return true, "", nil
}

// journalRecordByKey reads the record under the journal's own key: nil when
// it is gone, an error when it cannot be read, does not decode, or is not
// the journal's (another host or key in its body).
func journalRecordByKey(ctx context.Context, s release.Store, j registry.ImportJournal) (*Record, error) {
	raw, _, err := s.Get(ctx, j.RecordKey)
	if errors.Is(err, os.ErrNotExist) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	r, err := DecodeRecord(raw)
	if err != nil {
		return nil, err
	}
	if r.Host != j.Host || RecordKey(r.Account, r.App, r.Host) != j.RecordKey {
		return nil, fmt.Errorf("the record under %s is %s's", j.RecordKey, RecordKey(r.Account, r.App, r.Host))
	}
	return &r, nil
}