package provision

import (
	"context"
	"errors"
	"os"
	"path/filepath"
	"reflect"
	"strings"
	"testing"
	"time"

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

func seededRecord(t *testing.T, s release.Store, sqid, kind, state string) Record {
	t.Helper()
	rec := acmeRecord(sqid)
	id, _ := seeds.NewID()
	rec.Seed = &SeedSource{ID: id, Kind: kind, State: state, SHA256: "a", Bytes: 1, By: "t", At: time.Now()}
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	return rec
}

func nestedSeedCfg(data release.Store) ReconcileConfig {
	return ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data",
		Layout: func(context.Context, release.Store) (string, error) { return release.LayoutModeNested, nil },
		Seeds:  &SeedHooks{Data: data}}
}

func TestReconcileMintsAFromSeedNestedWithoutTheSeedCommand(t *testing.T) {
	s, sqid := setup(t) // the manifest declares Seed ["seed"]
	rec := seededRecord(t, s, sqid, seeds.KindReplica, seeds.StateReady)
	reg := regWith(t)
	if err := Reconcile(ctx, s, reg, nestedSeedCfg(dirStore(t))); err != nil {
		t.Fatal(err)
	}
	got := mustGet(t, reg, rec.Host)
	if got.Seed != "" || got.DBPath != NestedDBPath("/data", rec.Host) || got.SeedSource != SeedColumn(rec) {
		t.Fatalf("row = seed %q db %q col %q", got.Seed, got.DBPath, got.SeedSource)
	}
	// The record moves on (placed), and the next pass converges the column
	// and still never stamps the seed command.
	if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedPlace(rec.Seed.ID)); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, nestedSeedCfg(dirStore(t))); err != nil {
		t.Fatal(err)
	}
	got = mustGet(t, reg, rec.Host)
	row, _ := seeds.ParseRowSeed(got.SeedSource)
	if got.Seed != "" || row == nil || row.State != seeds.StatePlaced {
		t.Fatalf("after placed: seed %q col %+v", got.Seed, row)
	}
}

func TestReconcileDoesNotMintASeedFlatOrClaimed(t *testing.T) {
	s, sqid := setup(t)
	rec := seededRecord(t, s, sqid, seeds.KindReplica, seeds.StateReady)
	for name, cfg := range map[string]ReconcileConfig{
		"flat": {Region: "ie", Pools: []string{"tito"}, DataDir: "/data", Seeds: &SeedHooks{Data: dirStore(t)}},
		// The claim hook admits the write, so only the seed dispatch can
		// refuse it.
		"claimed": func() ReconcileConfig {
			c := nestedSeedCfg(dirStore(t))
			c.ClaimEnabled = true
			c.ClaimInstall = func(_ context.Context, _ ClaimRecord, write func(registry.ClaimOwnership) error) error {
				return write(registry.ClaimOwnership{ClaimIncarnation: "inc", ClaimStamp: 1, ClaimOwner: "owner", ClaimRowID: "row"})
			}
			return c
		}(),
		"unwired": {Region: "ie", Pools: []string{"tito"}, DataDir: "/data",
			Layout: func(context.Context, release.Store) (string, error) { return release.LayoutModeNested, nil }},
	} {
		reg := regWith(t)
		if err := Reconcile(ctx, s, reg, cfg); err != nil {
			t.Fatal(err)
		}
		if rt, _ := reg.Get(rec.Host); rt != nil {
			t.Errorf("%s: minted %+v", name, rt)
		}
	}
}

func TestReconcileSkipsSeedsThatAreNotReady(t *testing.T) {
	for _, c := range []struct{ kind, state string }{
		{seeds.KindReplica, seeds.StateReserving},
		{seeds.KindImport, seeds.StateAwaitingUpload},
		{seeds.KindImport, seeds.StateUploaded},
		{seeds.KindImport, seeds.StateRefused},
		{seeds.KindReplica, seeds.StateRolledBack},
	} {
		s, sqid := setup(t)
		rec := seededRecord(t, s, sqid, c.kind, c.state)
		reg := regWith(t)
		if err := Reconcile(ctx, s, reg, nestedSeedCfg(dirStore(t))); err != nil {
			t.Fatal(err)
		}
		if rt, _ := reg.Get(rec.Host); rt != nil {
			t.Errorf("%s %s: minted %+v", c.kind, c.state, rt)
		}
	}
}

func TestReconcileFinishesAnAbortedSeedWithNoJournal(t *testing.T) {
	s, sqid := setup(t)
	data := dirStore(t)
	rec := seededRecord(t, s, sqid, seeds.KindReplica, seeds.StateAborted)
	other, _ := seeds.NewID()
	mine := seeds.SnapshotKey(rec.Host, rec.Seed.ID)
	theirs := seeds.SnapshotKey(rec.Host, other)
	for _, k := range []string{mine, theirs} {
		if _, err := data.Put(ctx, k, []byte("db"), ""); err != nil {
			t.Fatal(err)
		}
	}
	reg := regWith(t)
	if err := Reconcile(ctx, s, reg, nestedSeedCfg(data)); err != nil {
		t.Fatal(err)
	}
	if _, _, err := data.Get(ctx, mine); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("this seed's snapshot: %v, want it deleted", err)
	}
	if _, _, err := data.Get(ctx, theirs); err != nil {
		t.Fatalf("another seed's snapshot was touched: %v", err)
	}
	got, err := GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
	if err != nil || got.Seed.State != seeds.StateRolledBack {
		t.Fatalf("record = %+v, %v; want rolled-back", got.Seed, err)
	}
	// Idempotent: a second pass over a rolled-back record does nothing.
	if err := Reconcile(ctx, s, reg, nestedSeedCfg(data)); err != nil {
		t.Fatal(err)
	}
}

// A record that would mint a flat row even on a nested box (it does not
// hibernate, so the activator never manages it) would start without its
// seed and without the seed command: it is not minted.
func TestReconcileDoesNotMintASeedThatWouldNotHibernate(t *testing.T) {
	s, sqid := setup(t)
	rec := acmeRecord(sqid)
	off := false
	rec.Hibernate = &off
	id, _ := seeds.NewID()
	rec.Seed = &SeedSource{ID: id, Kind: seeds.KindReplica, State: seeds.StateReady, SHA256: "a", Bytes: 1, By: "t", At: time.Now()}
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	if err := Reconcile(ctx, s, reg, nestedSeedCfg(dirStore(t))); err != nil {
		t.Fatal(err)
	}
	if rt, _ := reg.Get(rec.Host); rt != nil {
		t.Errorf("minted %+v", rt)
	}
}

// A --from seed aborted after this box minted its row (a lost publish
// answer): the row is removed first, the way a retired record's row is, and
// the seed stays aborted with its snapshot. The next pass, which finds no
// row, rolls it back.
func TestReconcileRemovesAnAbortedFromSeedsRowBeforeRollingBack(t *testing.T) {
	s, sqid := setup(t)
	data := dirStore(t)
	rec := seededRecord(t, s, sqid, seeds.KindReplica, seeds.StateReady)
	snap := seeds.SnapshotKey(rec.Host, rec.Seed.ID)
	if _, err := data.Put(ctx, snap, []byte("db"), ""); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	if err := Reconcile(ctx, s, reg, nestedSeedCfg(data)); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, rec.Host)
	if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbort(rec.Seed.ID)); err != nil {
		t.Fatal(err)
	}

	if err := Reconcile(ctx, s, reg, nestedSeedCfg(data)); err != nil {
		t.Fatal(err)
	}
	if rt, _ := reg.Get(rec.Host); rt != nil {
		t.Fatalf("the aborted seed's row is still up: %+v", rt)
	}
	got, err := GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
	if err != nil || got.Seed.State != seeds.StateAborted {
		t.Fatalf("record after the row's removal = %+v, %v; want still aborted", got.Seed, err)
	}
	if _, _, err := data.Get(ctx, snap); err != nil {
		t.Fatalf("the snapshot went with the row: %v", err)
	}

	if err := Reconcile(ctx, s, reg, nestedSeedCfg(data)); err != nil {
		t.Fatal(err)
	}
	got, err = GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
	if err != nil || got.Seed.State != seeds.StateRolledBack {
		t.Fatalf("record on the pass with no row = %+v, %v; want rolled-back", got.Seed, err)
	}
	if _, _, err := data.Get(ctx, snap); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("snapshot after the rollback: %v, want it deleted", err)
	}
	if rt, _ := reg.Get(rec.Host); rt != nil {
		t.Fatalf("a rolled-back seed was minted again: %+v", rt)
	}
}

// A row this pass has no authority over (not Provisioned: hand-added) is
// left alone, and the seed stays aborted while it stands.
func TestReconcileLeavesAnAbortedSeedsUnownedRow(t *testing.T) {
	s, sqid := setup(t)
	rec := seededRecord(t, s, sqid, seeds.KindReplica, seeds.StateAborted)
	reg := regWith(t, registry.Route{Host: rec.Host, Socket: "/run/carlos/hand.sock", DBPath: "/data/hand.db",
		App: rec.App, Account: rec.Account, Region: "ie", Pool: "tito"})
	if err := Reconcile(ctx, s, reg, nestedSeedCfg(dirStore(t))); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, rec.Host)
	got, err := GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
	if err != nil || got.Seed.State != seeds.StateAborted {
		t.Fatalf("record = %+v, %v; want still aborted while the row stands", got.Seed, err)
	}
}

// An import's row is the instance it replaces: an aborted import never
// removes it, and rolls back as before (its swap's rollback is Plan 7's).
func TestReconcileNeverRemovesAnAbortedImportsRow(t *testing.T) {
	s, sqid := setup(t)
	rec := seededRecord(t, s, sqid, seeds.KindImport, seeds.StateAborted)
	reg := regWith(t, registry.Route{Host: rec.Host, Socket: DefaultSocket(rec.Host), DBPath: DefaultDBPath("/data", rec.Host),
		App: rec.App, Account: rec.Account, Region: "ie", Pool: "tito", Provisioned: true})
	if err := Reconcile(ctx, s, reg, nestedSeedCfg(dirStore(t))); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, rec.Host)
	got, err := GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
	if err != nil || got.Seed.State != seeds.StateRolledBack {
		t.Fatalf("record = %+v, %v; want rolled-back", got.Seed, err)
	}
}

// diskBox is the rollback's seven box hooks over a real data directory: the
// marker and the database are files, read and removed as the activator's
// ReadSeedMarker and PurgeSeedLocal do (their own tests are the activator's
// seedpurge_linux_test.go). It records the steps in order.
type diskBox struct {
	t       *testing.T
	s       release.Store
	rec     Record
	steps   []string
	proceed bool
	gen     uint64
	fenced  map[string]uint64
	markErr error
}

func newDiskBox(t *testing.T, s release.Store, rec Record) *diskBox {
	return &diskBox{t: t, s: s, rec: rec, proceed: true, fenced: map[string]uint64{}}
}

func (b *diskBox) hooks(h *SeedHooks) {
	h.NextFenceGen = func() (uint64, error) { b.gen++; return b.gen, nil }
	h.FenceForRollback = func(_ context.Context, host string, gen uint64) (bool, error) {
		b.steps = append(b.steps, "fence")
		if !b.proceed {
			return false, nil
		}
		b.fenced[host] = gen
		return true, nil
	}
	h.LiftFence = func(host string, gen uint64) {
		b.steps = append(b.steps, "lift")
		if b.fenced[host] == gen {
			delete(b.fenced, host)
		}
	}
	h.LocalSeedID = func(db string) (string, error) {
		b.steps = append(b.steps, "read-marker")
		if b.markErr != nil {
			return "", b.markErr
		}
		m, err := os.ReadFile(db + ".seed")
		if errors.Is(err, os.ErrNotExist) {
			return "", nil
		}
		return string(m), err
	}
	h.RemoveLocal = func(rt registry.Route) error {
		b.steps = append(b.steps, "remove-local")
		if _, ok := b.fenced[rt.Host]; !ok {
			b.t.Errorf("removed the local data of %s while it was not fenced and stopped", rt.Host)
		}
		// Before the rollback: the record is still aborted.
		if got, err := GetRecord(ctx, b.s, b.rec.Account, b.rec.App, b.rec.Host); err != nil || got.Seed.State != seeds.StateAborted {
			b.t.Errorf("removed the local data with the record at %+v, %v; want aborted", got.Seed, err)
		}
		for _, p := range []string{rt.DBPath, rt.DBPath + "-wal", rt.DBPath + ".seed"} {
			if err := os.Remove(p); err != nil && !errors.Is(err, os.ErrNotExist) {
				return err
			}
		}
		return nil
	}
	h.DeleteReplica = func(context.Context, string) error {
		b.t.Error("the no-journal path deleted a replica")
		return nil
	}
	h.RetireKey = func(string, string, string) error {
		b.t.Error("the no-journal path retired an instance key: only an import's rollback does")
		return nil
	}
}

// placeLocal writes a database and its <db>.seed marker naming id at db,
// as a wake that placed the seed leaves them.
func placeLocal(t *testing.T, db, id string) {
	t.Helper()
	if err := os.MkdirAll(filepath.Dir(db), 0o700); err != nil {
		t.Fatal(err)
	}
	for p, body := range map[string]string{db: "the seed's data", db + "-wal": "wal", db + ".seed": id} {
		if err := os.WriteFile(p, []byte(body), 0o600); err != nil {
			t.Fatal(err)
		}
	}
}

func diskSeedCfg(t *testing.T, data release.Store, box *diskBox) ReconcileConfig {
	cfg := nestedSeedCfg(data)
	cfg.DataDir = t.TempDir()
	box.hooks(cfg.Seeds)
	return cfg
}

func requireSeedState(t *testing.T, s release.Store, rec Record, state string) {
	t.Helper()
	got, err := GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
	if err != nil || got.Seed == nil || got.Seed.State != state {
		t.Fatalf("record = %+v, %v; want %s", got.Seed, err, state)
	}
}

// Final review F2: a --from seed whose publish landed but whose answer was
// lost. A box minted its row and a wake placed its database; then the abort
// won. The row goes first; the pass that finds no row removes the database
// (its marker names this seed) under a fence, before the record is rolled
// back. That pass runs as after an agent restart: nothing in memory, only
// the marker on disk. A plain instance created at the host afterwards is
// minted at the same path and finds no file there.
func TestReconcileAbortedFromSeedRemovesItsLocalDatabase(t *testing.T) {
	s, sqid := setup(t)
	data := dirStore(t)
	rec := seededRecord(t, s, sqid, seeds.KindReplica, seeds.StateReady)
	box := newDiskBox(t, s, rec)
	cfg := diskSeedCfg(t, data, box)
	reg := regWith(t)
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	db := mustGet(t, reg, rec.Host).DBPath
	if db != NestedDBPath(cfg.DataDir, rec.Host) {
		t.Fatalf("minted at %s", db)
	}
	placeLocal(t, db, rec.Seed.ID)
	if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbort(rec.Seed.ID)); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, _ := reg.Get(rec.Host); rt != nil {
		t.Fatalf("the aborted seed's row is still up: %+v", rt)
	}
	requireSeedState(t, s, rec, seeds.StateAborted)
	if _, err := os.Stat(db); err != nil {
		t.Fatalf("the database went with the row, before the no-row pass: %v", err)
	}

	restarted := newDiskBox(t, s, rec) // the agent restarted: no fences, no memory
	restarted.hooks(cfg.Seeds)
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	requireSeedState(t, s, rec, seeds.StateRolledBack)
	if want := []string{"read-marker", "fence", "read-marker", "remove-local", "lift"}; !reflect.DeepEqual(restarted.steps, want) {
		t.Fatalf("steps = %q, want %q", restarted.steps, want)
	}
	if len(restarted.fenced) != 0 {
		t.Fatalf("a fence was left: %v", restarted.fenced)
	}
	for _, p := range []string{db, db + "-wal", db + ".seed"} {
		if _, err := os.Lstat(p); !errors.Is(err, os.ErrNotExist) {
			t.Fatalf("%s survived the rollback: %v", p, err)
		}
	}

	// The console's finish deletes the record; a plain instance follows.
	if err := DeleteRecord(ctx, s, rec.Account, rec.App, rec.Host); err != nil {
		t.Fatal(err)
	}
	plain := acmeRecord(sqid)
	if err := CreateRecord(ctx, s, plain); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	next := mustGet(t, reg, plain.Host)
	if next.SeedSource != "" {
		t.Fatalf("the plain instance's row = %+v", next)
	}
	if _, err := os.Lstat(next.DBPath); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the plain instance at %s finds a database: %v", next.DBPath, err)
	}
}

// The marker decides, as for an import whose row is gone (Plan 7): a
// marker naming another seed, or none, rolls back with the file left; a
// marker that cannot be read, or a fence that does not proceed, keeps the
// seed aborted with everything in place.
func TestReconcileAbortedFromSeedDecidesItsLocalDataFromTheMarker(t *testing.T) {
	other, _ := seeds.NewID()
	for _, c := range []struct {
		name    string
		marker  string // "" writes no marker; "this" names the seed
		markErr error
		noFence bool
		state   string
		removed bool
		steps   []string
		logged  string
	}{
		{name: "this seed", marker: "this", state: seeds.StateRolledBack, removed: true,
			steps: []string{"read-marker", "fence", "read-marker", "remove-local", "lift"}, logged: "its local data is removed"},
		{name: "another seed", marker: other, state: seeds.StateRolledBack,
			steps: []string{"read-marker"}, logged: "its seed marker names \"" + other + "\", not this seed"},
		{name: "no marker", state: seeds.StateRolledBack,
			steps: []string{"read-marker"}, logged: "its seed marker names \"\", not this seed"},
		{name: "unreadable marker", marker: "this", markErr: errors.New("permission denied"), state: seeds.StateAborted,
			steps: []string{"read-marker"}, logged: "the seed marker beside"},
		{name: "not fenced", marker: "this", noFence: true, state: seeds.StateAborted,
			steps: []string{"read-marker", "fence", "lift"}, logged: "is not fenced and stopped yet"},
	} {
		t.Run(c.name, func(t *testing.T) {
			s, sqid := setup(t)
			data := dirStore(t)
			rec := seededRecord(t, s, sqid, seeds.KindReplica, seeds.StateAborted)
			snap := seeds.SnapshotKey(rec.Host, rec.Seed.ID)
			if _, err := data.Put(ctx, snap, []byte("db"), ""); err != nil {
				t.Fatal(err)
			}
			box := newDiskBox(t, s, rec)
			box.markErr, box.proceed = c.markErr, !c.noFence
			cfg := diskSeedCfg(t, data, box)
			db := NestedDBPath(cfg.DataDir, rec.Host)
			switch c.marker {
			case "":
				placeLocal(t, db, "")
				if err := os.Remove(db + ".seed"); err != nil {
					t.Fatal(err)
				}
			case "this":
				placeLocal(t, db, rec.Seed.ID)
			default:
				placeLocal(t, db, c.marker)
			}
			logged := captureLog(t)
			if err := Reconcile(ctx, s, regWith(t), cfg); err != nil {
				t.Fatal(err)
			}
			requireSeedState(t, s, rec, c.state)
			if !reflect.DeepEqual(box.steps, c.steps) {
				t.Fatalf("steps = %q, want %q", box.steps, c.steps)
			}
			if !strings.Contains(logged.String(), c.logged) {
				t.Fatalf("not logged %q:\n%s", c.logged, logged)
			}
			if _, err := os.Lstat(db); errors.Is(err, os.ErrNotExist) != c.removed {
				t.Fatalf("database at %s: %v; want removed %v", db, err, c.removed)
			}
			_, _, err := data.Get(ctx, snap)
			if gone := errors.Is(err, os.ErrNotExist); gone != (c.state == seeds.StateRolledBack) {
				t.Fatalf("snapshot: %v; want it deleted only with the rollback", err)
			}
		})
	}
}