package host

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io/fs"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// The final whole-branch review of Plan 4 (final-p4a-review.md, rulings in
// final-fix-rulings.md): the target's cleaning moves only what this op
// created, an abandoned build's location lease is released when the
// rollback starts, and the install's restore and install rights are granted
// at the phase that performs them.

// treeSnapshot records every entry under the named paths (relative to dir)
// by identity, mode and bytes, so a test can tell a path left in place from
// one moved away and back or rewritten.
func treeSnapshot(t *testing.T, dir string, names ...string) map[string]string {
	t.Helper()
	out := map[string]string{}
	for _, name := range names {
		err := filepath.WalkDir(filepath.Join(dir, name), func(p string, d fs.DirEntry, err error) error {
			if err != nil {
				return err
			}
			rel, _ := filepath.Rel(dir, p)
			fi, err := os.Lstat(p)
			if err != nil {
				return err
			}
			id, err := pathIdentity(p)
			if err != nil {
				return err
			}
			body := ""
			if fi.Mode().IsRegular() {
				b, err := os.ReadFile(p)
				if err != nil {
					return err
				}
				body = string(b)
			}
			out[rel] = fmt.Sprintf("%s %s %q", id, fi.Mode(), body)
			return nil
		})
		if err != nil {
			t.Fatalf("snapshot %s: %v", name, err)
		}
	}
	return out
}

func sameSnapshot(a, b map[string]string) bool {
	if len(a) != len(b) {
		return false
	}
	for k, v := range a {
		if b[k] != v {
			return false
		}
	}
	return true
}

// cleanTarget drives a rolling_back record from the decision to cleaned on
// the target: the release, the source's recovery, the cleaning.
func (h *rollbackHarness) cleanTarget(t *testing.T) *provision.MoveRollback {
	t.Helper()
	h.rollbackPass(t)
	if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReleased && rb.TargetPhase != provision.RbTargetReturned {
		t.Fatalf("the release: %+v", rb)
	}
	h.recovered(t)
	h.rollbackPass(t)
	rb := h.rb(t)
	if rb.TargetPhase != provision.RbTargetCleaned || rb.TargetParkedAt == nil {
		t.Fatalf("the cleaning did not complete: %+v (lease table %v)", rb, h.leases.holding())
	}
	return rb
}

// Finding 1: files the landing refused as foreign (an unregistered files
// directory and env file, RollbackFilesOccupied) stay where they are, byte
// for byte, through the rollback's cleaning, which still completes.
func TestRollbackCleaningKeepsWhatTheLandingRefused(t *testing.T) {
	t.Parallel()
	h := &rollbackHarness{landingHarness: newLandingHarness(t)}
	h.flip(t)
	h.sentinels(t)
	foreign := []string{p1Host, p1Host + ".env"}
	before := treeSnapshot(t, h.tdir, foreign...)
	h.land(t)
	if m := h.read(t, p1Host).Move; m.Reason != provision.RollbackFilesOccupied || m.LandingPhase != provision.LandLocal {
		t.Fatalf("the landing did not refuse the foreign files: %+v", m)
	}
	h.rollbackPass(t) // the decision converts the mark
	rolledBack(t, h.landingHarness, provision.RollbackFilesOccupied, provision.LandLocal)
	h.cleanTarget(t)
	h.sentinelsIntact(t, "after the cleaning")
	if after := treeSnapshot(t, h.tdir, foreign...); !sameSnapshot(before, after) {
		t.Fatalf("the foreign files changed:\nbefore %v\nafter  %v", before, after)
	}
	// Nothing of theirs is in the parked tree, and the op's journal is.
	var parkedForeign []string
	filepath.WalkDir(filepath.Join(h.tdir, ".parked"), func(p string, d fs.DirEntry, err error) error {
		if err == nil && (filepath.Base(p) == "sentinel.txt" || filepath.Base(p) == p1Host+".env") {
			parkedForeign = append(parkedForeign, p)
		}
		return nil
	})
	if len(parkedForeign) != 0 {
		t.Fatalf("foreign files parked: %v", parkedForeign)
	}
	if _, err := os.Lstat(moves.JournalDir(h.tdir, p1Host, p1ID)); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the op journal is still live: %v", err)
	}
}

// Finding 1, the other half: what the op created is still parked by the
// cleaning, on its ownership evidence, and what it did not create is left.
// The landing placed its files and env (the placement receipt) and stopped
// before it registered them under F; a WAL and a rendered config were
// there that it never wrote. The cleaning parks the op's two and leaves the
// other two. Once the landing has installed its database and rendered its
// config (the restored CAS and the render's ack), the config and the
// database's WAL are the op's and are parked.
func TestRollbackCleaningParksWhatTheOpCreated(t *testing.T) {
	t.Parallel()
	t.Run("placed, not registered", func(t *testing.T) {
		h := &rollbackHarness{landingHarness: newLandingHarness(t)}
		h.flip(t)
		db := provision.DefaultDBPath(h.tdir, p1Host)
		for name, body := range map[string]string{db + "-wal": "a wal the landing never wrote", filepath.Join(h.tdir, p1Host+".config.env"): "OTHER=1\n"} {
			if err := os.WriteFile(name, []byte(body), 0o600); err != nil {
				t.Fatal(err)
			}
		}
		foreign := []string{p1Host + ".db-wal", p1Host + ".config.env"}
		before := treeSnapshot(t, h.tdir, foreign...)
		h.landAt = "after-extracted"
		h.crashLand(t)
		j := h.landJournal(t)
		if !j.Done(landStepFilesPlaced) || j.Done(landStepInventory) {
			t.Fatalf("the fixture: %v", j.Steps)
		}
		if c := h.claim(t); hostclaim.InventoryHasStamp(c, "uk-box", c.Holder.Stamp) {
			t.Fatal("the fixture registered the landing's paths")
		}
		h.rollBack(t)
		h.cleanTarget(t)
		for _, p := range []string{p1Host, p1Host + ".env"} {
			if _, err := os.Lstat(filepath.Join(h.tdir, p)); !errors.Is(err, os.ErrNotExist) {
				t.Fatalf("the landing's own %s is still live: %v", p, err)
			}
		}
		parked := parkedUnder(t, h.tdir)
		if !parkedHas(parked, "/artifacts/"+p1Host+"/upload.txt", "hello") {
			t.Fatalf("the landing's files were not parked: %v", parked)
		}
		if after := treeSnapshot(t, h.tdir, foreign...); !sameSnapshot(before, after) {
			t.Fatalf("what the op did not create changed:\nbefore %v\nafter  %v", before, after)
		}
	})
	t.Run("installed and rendered", func(t *testing.T) {
		h := newRollbackHarness(t)
		db := provision.DefaultDBPath(h.tdir, p1Host)
		config := filepath.Join(h.tdir, p1Host+".config.env")
		if err := os.WriteFile(db+"-wal", []byte("the landing's wal"), 0o600); err != nil {
			t.Fatal(err)
		}
		rendered := []byte("RENDERED=1\n")
		if err := os.WriteFile(config, rendered, 0o600); err != nil {
			t.Fatal(err)
		}
		sum := sha256.Sum256(rendered)
		if err := appconfig.WriteConfigAck(context.Background(), h.s, appconfig.ConfigAck{Schema: appconfig.ConfigAckSchema, Box: "uk-box", Host: p1Host,
			At: h.clock.Now(), MoveOp: p1ID, FileSHA256: hex.EncodeToString(sum[:])}); err != nil {
			t.Fatal(err)
		}
		h.rollBack(t)
		h.cleanTarget(t)
		for _, p := range sourceArtifactPaths(p1Host) {
			if _, err := os.Lstat(filepath.Join(h.tdir, p)); !errors.Is(err, os.ErrNotExist) {
				t.Fatalf("%s is still live: %v", p, err)
			}
		}
		parked := parkedUnder(t, h.tdir)
		if !parkedHas(parked, "/artifacts/"+p1Host+".db-wal", "the landing's wal") || !parkedHas(parked, "/artifacts/"+p1Host+".config.env", "RENDERED=1\n") {
			t.Fatalf("the op's leftovers were not parked: %v", parked)
		}
	})
	t.Run("a config the render did not write", func(t *testing.T) {
		h := newRollbackHarness(t)
		config := filepath.Join(h.tdir, p1Host+".config.env")
		if err := os.WriteFile(config, []byte("ANOTHER=1\n"), 0o600); err != nil {
			t.Fatal(err)
		}
		// The ack names this op, but another file's bytes.
		sum := sha256.Sum256([]byte("RENDERED=1\n"))
		if err := appconfig.WriteConfigAck(context.Background(), h.s, appconfig.ConfigAck{Schema: appconfig.ConfigAckSchema, Box: "uk-box", Host: p1Host,
			At: h.clock.Now(), MoveOp: p1ID, FileSHA256: hex.EncodeToString(sum[:])}); err != nil {
			t.Fatal(err)
		}
		before := treeSnapshot(t, h.tdir, p1Host+".config.env")
		h.rollBack(t)
		h.cleanTarget(t)
		if after := treeSnapshot(t, h.tdir, p1Host+".config.env"); !sameSnapshot(before, after) {
			t.Fatalf("a config the render did not write changed:\nbefore %v\nafter  %v", before, after)
		}
	})
}

// parkedUnder maps every regular file under dir's .parked/ to its body.
func parkedUnder(t *testing.T, dir string) map[string]string {
	t.Helper()
	out := map[string]string{}
	err := filepath.WalkDir(filepath.Join(dir, ".parked"), func(p string, d fs.DirEntry, err error) error {
		if err != nil || d.IsDir() {
			return err
		}
		body, err := os.ReadFile(p)
		if err != nil {
			return err
		}
		out[p] = string(body)
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	return out
}

// Finding 2: a build that failed leaves the location's lease held across
// passes, as the landing designs; once the landing is rolled back that
// lease is abandoned and is released, whether the rollback is decided by
// this worker's landing pass or outside it. The same worker then cleans,
// and its park engine acquires the location's lease.
func TestRollbackReleasesTheAbandonedBuildLease(t *testing.T) {
	t.Parallel()
	failedBuild := func(t *testing.T) *rollbackHarness {
		t.Helper()
		h := &rollbackHarness{landingHarness: newLandingHarness(t)}
		h.flip(t)
		h.landAt = "after-build"
		h.crashLand(t)
		if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandBuilding {
			t.Fatalf("the fixture: %+v", m)
		}
		if held := h.leases.holding(); len(held) != 1 {
			t.Fatalf("the failed build holds %v, want the location's lease", held)
		}
		// The same worker carries on: its lease map is the build's.
		h.rw = h.lw
		return h
	}
	t.Run("decided in the landing pass", func(t *testing.T) {
		h := failedBuild(t)
		h.rollBack(t)
		if held := h.leases.holding(); len(held) != 0 {
			t.Fatalf("the rollback kept the build's lease: %v", held)
		}
		h.cleanTarget(t)
	})
	t.Run("decided outside the landing pass", func(t *testing.T) {
		h := failedBuild(t)
		// Another worker decides the rollback (the decision is not this
		// worker's pass); the target's next pass enters the rollback.
		h.clock.Advance(cutLandingWindow + time.Minute)
		other := &movesWorker{mirrors: map[string]lease.MirrorStore{}, cfg: h.rollbackCfg(t)}
		if done, err := other.rollBackLanding(context.Background(), h.s, h.read(t, p1Host)); !done || err != nil {
			t.Fatalf("the outside decision: %v %v", done, err)
		}
		if held := h.leases.holding(); len(held) != 1 {
			t.Fatalf("the fixture: %v", held)
		}
		h.rollbackPass(t)
		if held := h.leases.holding(); len(held) != 0 {
			t.Fatalf("the rollback's entry kept the build's lease: %v", held)
		}
		h.rollbackPass(t) // idempotent: nothing held, nothing to release
		h.cleanTarget(t)
	})
}

// policyProbe checks, from inside an installer's own write hooks, that each
// write it makes is one hostbox.Policy.Check grants with MovesClaimPolicy as
// Policy.Moves: the selection is made as hostbox.WorkSource.Select makes it
// (the live claim's op, kind and phase, the installation record's capture),
// the row's hold is the registry's, and the record is read from the store.
type policyProbe struct {
	t                    *testing.T
	s                    release.Store
	dataDir, box, region string
	hold                 func() string
	phase                func(*provision.Record) string
	granted, refused     []string
}

func (p *policyProbe) check(action string, id hostwrite.Identity) {
	ctx := context.Background()
	obs, err := hostclaim.Read(ctx, p.s, p1Host)
	if err != nil || obs.Absent {
		p.t.Errorf("read the claim: %v", err)
		return
	}
	captured, err := hostguard.Capture(p.dataDir, p1Host)
	if err != nil {
		p.t.Errorf("capture: %v", err)
		return
	}
	rec, err := provision.GetRecord(ctx, p.s, "bab", "app", p1Host)
	if err != nil {
		p.t.Errorf("read the record: %v", err)
		return
	}
	c := obs.Claim
	sel := hostbox.Selection{Host: p1Host, Action: action, Box: p.box, Region: p.region, Captured: captured, Kind: c.Op.Kind, Op: c.Op.ID,
		Phase: c.Op.Phase, Generation: c.Holder.Generation, Session: c.Holder.Session}
	pol := hostbox.Policy{Moves: MovesClaimPolicy(MovesRecordsFrom(p.s)), Held: func(context.Context, string) (string, error) { return p.hold(), nil }}
	cell := p.phase(rec) + "/" + action
	if err := pol.Check(ctx, c, sel, id); err != nil {
		p.refused = append(p.refused, cell+": "+err.Error())
		return
	}
	p.granted = append(p.granted, cell)
}

// wrap puts the probe in front of the installer's write hooks: the timers
// import (held-timers), the held render (held-config), the release gate
// (held-release) and the generation marker the database install writes
// inside its guard (restore and install).
func (p *policyProbe) wrap(c *moveCfg) {
	timers, render, ready, gen := c.ReplaceTimers, c.RenderHeld, c.ReleaseReady, c.WriteGen
	c.ReplaceTimers = func(ctx context.Context, host string, rows []registry.Timer) error {
		p.check(movesActionHeldTimers, hostwrite.Identity{})
		return timers(ctx, host, rows)
	}
	c.RenderHeld = func(ctx context.Context, s release.Store, rt registry.Route, op string, pin *provision.MoveInputs, before func(string) error) (appconfig.HeldRender, error) {
		hr, err := render(ctx, s, rt, op, pin, before)
		if err == nil {
			p.check(movesActionHeldConfig, HeldConfigIdentity(hr.BundleETags, hr.OverlayETag))
		}
		return hr, err
	}
	c.ReleaseReady = func(ctx context.Context, s release.Store, rec provision.Record, v string) moves.Check {
		p.check(movesActionHeldRelease, hostwrite.Identity{Release: v})
		return ready(ctx, s, rec, v)
	}
	if live := c.ReleaseLive; live != nil {
		c.ReleaseLive = func(ctx context.Context, s release.Store, rec provision.Record, v string) moves.Check {
			p.check(movesActionHeldRelease, hostwrite.Identity{Release: v})
			return live(ctx, s, rec, v)
		}
	}
	c.WriteGen = func(path, generation string) error {
		p.check(movesActionRestore, hostwrite.Identity{})
		p.check(movesActionInstall, hostwrite.Identity{})
		return gen(path, generation)
	}
}

func (p *policyProbe) want(t *testing.T, cells ...string) {
	t.Helper()
	if len(p.refused) != 0 {
		t.Fatalf("Policy.Check refused the installer's own writes:\n%s", strings.Join(p.refused, "\n"))
	}
	got := map[string]bool{}
	for _, c := range p.granted {
		got[c] = true
	}
	var missing []string
	for _, c := range cells {
		if !got[c] {
			missing = append(missing, c)
		}
	}
	sort.Strings(missing)
	if len(missing) != 0 {
		t.Fatalf("never checked %v (granted %v)", missing, p.granted)
	}
}

// Finding 3 (J24, G8): the dirty return's installer, end to end, writes
// only what hostbox.Policy.Check grants under MovesClaimPolicy at the phase
// it writes in, with its row held by the rollback id: held-timers at local;
// held-config, held-release, restore and install at configured, whose
// closing CAS writes restored.
func TestDirtyReturnWritesPassThePolicy(t *testing.T) {
	t.Parallel()
	h := newReturnHarness(t, true)
	p := &policyProbe{t: t, s: h.s, dataDir: h.sdir, box: "ie-box", region: "ie", hold: func() string { return h.srcRoute.HoldOp },
		phase: func(rec *provision.Record) string { return rec.Move.Rollback.SourcePhase }}
	p.wrap(&h.rcfg)
	pinned := h.read(t, p1Host).Move.Inputs
	if refused, err := h.install(t); err != nil || refused != "" {
		t.Fatalf("install: %q %v", refused, err)
	}
	h.installedReturn(t, pinned)
	p.want(t, "local/held-timers", "configured/held-config", "configured/held-release", "configured/restore", "configured/install")
}

// The landing's rows had the same off-by-one: its installer restores and
// installs inside configured too. Its writes, end to end, pass
// hostbox.Policy.Check under MovesClaimPolicy with the row held by the move.
func TestLandingWritesPassThePolicy(t *testing.T) {
	t.Parallel()
	h := newLandingHarness(t)
	h.flip(t)
	p := &policyProbe{t: t, s: h.s, dataDir: h.tdir, box: "uk-box", region: "uk", hold: func() string { return h.route.HoldOp },
		phase: func(rec *provision.Record) string { return rec.Move.LandingPhase }}
	if h.lw == nil {
		h.lw = &movesWorker{mirrors: map[string]lease.MirrorStore{}}
	}
	for i := 0; i < 4 && h.read(t, p1Host).Move.LandingPhase != provision.LandRestored; i++ {
		h.lw.cfg = h.landingCfg(t)
		p.wrap(&h.lw.cfg)
		if err := movesPass(context.Background(), h.s, h.lw); err != nil {
			t.Fatal(err)
		}
	}
	if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandRestored {
		t.Fatalf("the landing did not install: %+v", m)
	}
	p.want(t, "local/held-timers", "configured/held-config", "configured/held-release", "configured/restore", "configured/install")
}

// plantSiblings writes a quarantine file and a litestream shadow directory
// beside the database at the flat paths.
func plantSiblings(t *testing.T, dir, body string) []string {
	t.Helper()
	corrupt := p1Host + ".db.corrupt"
	shadow := ".activator-" + p1Host + ".db-litestream"
	if err := os.MkdirAll(filepath.Join(dir, shadow), 0o750); err != nil {
		t.Fatal(err)
	}
	for name, b := range map[string]string{corrupt: body, filepath.Join(shadow, "generation"): body} {
		if err := os.WriteFile(filepath.Join(dir, name), []byte(b), 0o600); err != nil {
			t.Fatal(err)
		}
	}
	return []string{corrupt, shadow}
}

// Finding 1, the re-review's residual (task-final-rereview-1.md): the
// quarantine file and the litestream shadow are the op's only when the
// install journalled them absent before it installed the database. One
// already there when the landing installed survives the cleaning byte for
// byte; ones made after the install are parked; with the journal entry
// missing (an install that recorded none), they are left.
func TestRollbackCleaningOwnsDatabaseSiblingsOnlyWhenRecordedAbsent(t *testing.T) {
	t.Parallel()
	t.Run("present before the install", func(t *testing.T) {
		h := &rollbackHarness{landingHarness: newLandingHarness(t)}
		h.flip(t)
		foreign := plantSiblings(t, h.tdir, "not the landing's")
		before := treeSnapshot(t, h.tdir, foreign...)
		h.land(t)
		h.installed(t)
		h.step(t, provision.LandRestored, provision.LandProbing)
		h.tpos = h.n
		h.rollBack(t)
		h.cleanTarget(t)
		if after := treeSnapshot(t, h.tdir, foreign...); !sameSnapshot(before, after) {
			t.Fatalf("the siblings the landing found changed:\nbefore %v\nafter  %v", before, after)
		}
	})
	t.Run("made after the install", func(t *testing.T) {
		h := newRollbackHarness(t)
		made := plantSiblings(t, h.tdir, "the landing's instance")
		h.rollBack(t)
		h.cleanTarget(t)
		for _, p := range made {
			if _, err := os.Lstat(filepath.Join(h.tdir, p)); !errors.Is(err, os.ErrNotExist) {
				t.Fatalf("%s is still live: %v", p, err)
			}
		}
		parked := parkedUnder(t, h.tdir)
		if !parkedHas(parked, "/artifacts/"+made[0], "the landing's instance") || !parkedHas(parked, "/artifacts/"+made[1]+"/generation", "the landing's instance") {
			t.Fatalf("the op's siblings were not parked: %v", parked)
		}
	})
	t.Run("no record of their absence", func(t *testing.T) {
		h := newRollbackHarness(t)
		j := h.landJournal(t)
		if _, ok := j.Steps[landStepPreInstall]; !ok {
			t.Fatalf("the install journalled no %s step: %v", landStepPreInstall, j.Steps)
		}
		j.Unmark(landStepPreInstall)
		if err := j.Commit(h.tdir); err != nil {
			t.Fatal(err)
		}
		made := plantSiblings(t, h.tdir, "unrecorded")
		before := treeSnapshot(t, h.tdir, made...)
		h.rollBack(t)
		h.cleanTarget(t)
		if after := treeSnapshot(t, h.tdir, made...); !sameSnapshot(before, after) {
			t.Fatalf("siblings with no recorded absence changed:\nbefore %v\nafter  %v", before, after)
		}
	})
}

// A crash after pre_install is journalled and before the database's rename:
// the replay looks again right before its rename, so a sibling that
// appeared in between is recorded present and survives the cleaning.
func TestPreInstallReplayLooksAgain(t *testing.T) {
	t.Parallel()
	h := &rollbackHarness{landingHarness: newLandingHarness(t)}
	h.flip(t)
	h.landAt = "after-" + landStepPreInstall
	h.crashLand(t)
	foreign := plantSiblings(t, h.tdir, "appeared after the first look")
	before := treeSnapshot(t, h.tdir, foreign...)
	h.freshTarget()
	h.land(t)
	h.installed(t)
	var pre preInstall
	if err := json.Unmarshal([]byte(h.landJournal(t).Steps[landStepPreInstall]), &pre); err != nil || len(pre.Absent) != 0 {
		t.Fatalf("pre_install after the replay: %+v %v", pre, err)
	}
	h.step(t, provision.LandRestored, provision.LandProbing)
	h.tpos = h.n
	h.rollBack(t)
	h.cleanTarget(t)
	if after := treeSnapshot(t, h.tdir, foreign...); !sameSnapshot(before, after) {
		t.Fatalf("siblings that appeared before the replay changed:\nbefore %v\nafter  %v", before, after)
	}
}