package host

import (
	"context"
	"errors"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// countingClaims counts the claim reads made through it, and runs onGet,
// when set, before each.
type countingClaims struct {
	release.Store
	mu     sync.Mutex
	claims int
	onGet  func(key string)
}

func (s *countingClaims) Get(ctx context.Context, key string) ([]byte, string, error) {
	if strings.HasPrefix(key, "control/hosts/") {
		s.mu.Lock()
		s.claims++
		s.mu.Unlock()
	}
	if s.onGet != nil {
		s.onGet(key)
	}
	return s.Store.Get(ctx, key)
}

func (s *countingClaims) reads() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	n := s.claims
	s.claims = 0
	return n
}

func passRegistry(t *testing.T, hosts ...string) *registry.Registry {
	t.Helper()
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { reg.Close() })
	for _, host := range hosts {
		if err := reg.Upsert(registry.Route{Host: host, Status: "ready", Socket: "/run/" + host + ".sock"}); err != nil {
			t.Fatal(err)
		}
	}
	return reg
}

// Review M2: the completion-only pass of a box with nothing to finish reads
// each of its claims once and asks nothing more: no identity read, however
// many rows. A pass that finds pending completions asks which box this is
// once, however many it finds.
func TestCompletionPassAsksNothingMoreWithNothingPending(t *testing.T) {
	t.Parallel()
	ctx := t.Context()
	store := &countingClaims{Store: release.NewDirStore(t.TempDir())}
	for _, host := range []string{"a.example", "b.example", "c.example"} {
		passClaim(t, store.Store, host, hostclaim.Op{})
	}
	completeOn(t, store.Store, "b.example", "rel-b", "dst", true) // finished already
	reg := passRegistry(t, "a.example", "b.example", "c.example")
	identities := 0
	local := &productionClaimLocal{legacyParticipation: true, assembly: &ClaimAssembly{Claims: store, Guard: hostguard.Guard{DataDir: t.TempDir()}, stopLocal: func(context.Context) (claimStopLocal, error) {
		identities++
		return claimStopLocal{Actor: hostbox.Actor{Box: "dst"}}, nil
	}}}
	r, err := newCompletionRuntime(local, reg, &activator.Service{})
	if err != nil {
		t.Fatal(err)
	}
	store.reads()
	r.once(ctx, store)
	if n := store.reads(); identities != 0 || n != 3 {
		t.Fatalf("nothing pending: %d identity reads, %d claim reads (want 0 and 3)", identities, n)
	}
	for _, host := range []string{"d.example", "e.example"} {
		passClaim(t, store.Store, host, hostclaim.Op{})
		completeOn(t, store.Store, host, "rel-"+host[:1], "dst", false)
		if err := reg.Upsert(registry.Route{Host: host, Status: "ready", Socket: "/run/" + host + ".sock"}); err != nil {
			t.Fatal(err)
		}
	}
	f := &finishRecorder{}
	r.Finish = f.finish
	r.once(ctx, store)
	if n := store.reads(); identities != 1 || len(f.calls) != 2 || n != 5 {
		t.Fatalf("two pending: %d identity reads (want 1), finished %v, %d claim reads (want 5)", identities, f.calls, n)
	}
}

// Review M1: the completion-only pass reads the hosts its own op journals
// name beside its registry rows, so a host with a journal and no row is
// read; a journal it cannot read is said, and the rest are still read.
func TestCompletionPassReadsTheHostsItsJournalsName(t *testing.T) {
	ctx := t.Context()
	store := release.NewDirStore(t.TempDir())
	passClaim(t, store, "row.example", hostclaim.Op{})
	passClaim(t, store, "rowless.example", hostclaim.Op{})
	completeOn(t, store, "rowless.example", "rel-1", "dst", false)
	data := t.TempDir()
	marker, err := relocationProbeFencePath(data, "rel-1")
	if err != nil {
		t.Fatal(err)
	}
	if _, err := relocationProbeFenceOwner(marker, "rowless.example", 1); err != nil {
		t.Fatal(err)
	}
	bad := filepath.Join(data, "relocate", "ops", "rel-2", "attempts.json")
	if err := os.MkdirAll(filepath.Dir(bad), 0o700); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(bad, []byte("not a journal"), 0o600); err != nil {
		t.Fatal(err)
	}
	reg := passRegistry(t, "row.example")
	local := &productionClaimLocal{legacyParticipation: true, assembly: &ClaimAssembly{Claims: store, Guard: hostguard.Guard{DataDir: data}, stopLocal: func(context.Context) (claimStopLocal, error) {
		return claimStopLocal{Actor: hostbox.Actor{Box: "dst"}}, nil
	}}}
	r, err := newCompletionRuntime(local, reg, &activator.Service{})
	if err != nil {
		t.Fatal(err)
	}
	hosts, err := r.Hosts()
	if len(hosts) != 2 || hosts[0] != "row.example" || hosts[1] != "rowless.example" || err == nil {
		t.Fatalf("hosts %v, err %v: want the row, the journal's host and the unreadable journal", hosts, err)
	}
	buf := captureRacyLog(t)
	f := &finishRecorder{}
	r.Finish = f.finish
	r.once(ctx, store)
	if len(f.calls) != 1 || f.calls[0] != "rowless.example:rel-1" || !strings.Contains(buf.String(), "relocate: list hosts:") {
		t.Fatalf("finished %v, log:\n%s", f.calls, buf.String())
	}
}

// Review M8: an alert is one per host, step and cause. Two steps failing
// with one cause are both said; a cause that clears and returns is said
// again; a host the pass no longer reads is forgotten.
func TestRelocatePassAlertsPerStepAndForgetsClearedCauses(t *testing.T) {
	ctx := t.Context()
	store := release.NewDirStore(t.TempDir())
	passClaim(t, store, "a.example", relocatingOp("rel-a"))
	same := errors.New("the roster could not be read")
	var order []string
	participant := &flakyStepper{name: "participant", order: &order, err: same}
	r := &RelocationRuntime{Worker: stepRecorder{name: "worker", order: &order, err: same}, Participant: participant, Box: "dst"}
	listed := []string{"a.example"}
	r.Hosts = func() ([]string, error) { return listed, nil }
	buf := captureRacyLog(t)
	r.once(ctx, store)
	r.once(ctx, store)
	if strings.Count(buf.String(), "relocate a.example: participant: "+same.Error()) != 1 || strings.Count(buf.String(), "relocate a.example: worker: "+same.Error()) != 1 {
		t.Fatalf("not one alert per step:\n%s", buf.String())
	}
	participant.err = nil // the cause clears
	r.once(ctx, store)
	participant.err = same // and returns
	r.once(ctx, store)
	if n := strings.Count(buf.String(), "relocate a.example: participant: "+same.Error()); n != 2 {
		t.Fatalf("a returning cause said %d times, want twice:\n%s", n, buf.String())
	}
	if r.alerts.size() != 2 {
		t.Fatalf("held %d causes, want 2", r.alerts.size())
	}
	listed = nil
	r.once(ctx, store)
	if r.alerts.size() != 0 {
		t.Fatalf("held %d causes for hosts no longer read", r.alerts.size())
	}
}

// flakyStepper is a stepRecorder whose error can change between passes.
type flakyStepper struct {
	name  string
	order *[]string
	err   error
}

func (s *flakyStepper) Step(_ context.Context, host string) (hostclaim.Observed, error) {
	*s.order = append(*s.order, s.name+":"+host)
	return hostclaim.Observed{}, s.err
}

// Review M6: the sweep reads every claim before it takes the journals'
// lock exclusively, so a wake's journal write never waits on a bucket
// read; and what it decided unlocked it removes only as it finds it under
// the lock. A directory that held no journal when it was read, and holds
// one by the time the lock is held, is kept.
func TestSweepReadsClaimsUnlockedAndChecksAgainUnderTheLock(t *testing.T) {
	ctx := t.Context()
	data := t.TempDir()
	ops := filepath.Join(data, "relocate", "ops")
	for _, dir := range []string{"aaa", "bbb"} {
		if err := os.MkdirAll(filepath.Join(ops, dir), 0o700); err != nil {
			t.Fatal(err)
		}
	}
	// aaa: a write's temporary only. bbb: b.example's journal, its claim gone.
	if err := os.WriteFile(filepath.Join(ops, "aaa", ".attempts.json.1"), []byte("{"), 0o600); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(filepath.Join(ops, "bbb", "install.json"), []byte(`{"v":1,"host":"b.example"}`), 0o600); err != nil {
		t.Fatal(err)
	}
	locked := false
	store := &countingClaims{Store: release.NewDirStore(t.TempDir())}
	store.onGet = func(key string) {
		if relocationJournals.TryRLock() {
			relocationJournals.RUnlock()
		} else {
			locked = true
		}
		// A write lands in aaa after the sweep read it as empty.
		if key == "control/hosts/b.example.json" {
			if err := os.WriteFile(filepath.Join(ops, "aaa", "attempts.json"), []byte(`{"v":1,"host":"a.example","op":"aaa"}`), 0o600); err != nil {
				t.Error(err)
			}
		}
	}
	if err := sweepRelocationJournals(ctx, data, store); err != nil {
		t.Fatal(err)
	}
	if locked {
		t.Fatal("a claim was read under the journals' exclusive lock")
	}
	if store.reads() == 0 {
		t.Fatal("the sweep read no claim")
	}
	if _, err := os.Stat(filepath.Join(ops, "bbb")); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the ended op's journals survived: %v", err)
	}
	if _, err := os.Stat(filepath.Join(ops, "aaa", "attempts.json")); err != nil {
		t.Fatalf("a journal committed after the decision was swept: %v", err)
	}
}

// closingStepper closes stepped on its first Step.
type closingStepper struct {
	once    *sync.Once
	stepped chan struct{}
}

func (s closingStepper) Step(context.Context, string) (hostclaim.Observed, error) {
	s.once.Do(func() { close(s.stepped) })
	return hostclaim.Observed{}, nil
}

// Review M3 (run with -race): the pass's loop owns the runtime's fields
// once it starts, and its first pass replaces them with the full pass's
// (tryUpgrade). The launch says what it started from what it decided
// before the loop, never by reading the runtime after it.
func TestLaunchRelocationPassReadsNothingTheLoopOwns(t *testing.T) {
	store := release.NewDirStore(t.TempDir())
	passClaim(t, store, "a.example", relocatingOp("rel-a"))
	reg := passRegistry(t)
	local := &productionClaimLocal{legacyParticipation: true, assembly: &ClaimAssembly{Claims: store, Guard: hostguard.Guard{DataDir: t.TempDir()}, stopLocal: func(context.Context) (claimStopLocal, error) {
		return claimStopLocal{Actor: hostbox.Actor{Box: "dst"}}, nil
	}}}
	stepped := make(chan struct{})
	step := closingStepper{once: &sync.Once{}, stepped: stepped}
	r, err := relocationPassFor(local, reg, &activator.Service{}, func(context.Context) (*RelocationRuntime, error) {
		return &RelocationRuntime{Worker: step, Participant: step, Box: "dst", Finish: func(context.Context, string, string) error { return nil }}, nil
	})
	if err != nil {
		t.Fatal(err)
	}
	buf := captureRacyLog(t)
	ctx, cancel := context.WithCancel(t.Context())
	done := launchRelocationPass(ctx, r, store, true)
	<-stepped
	cancel()
	<-done
	if !strings.Contains(buf.String(), "relocation pass started (completions only: false)") || !strings.Contains(buf.String(), "agent: region HA: relocation pass started") {
		t.Fatalf("not said:\n%s", buf.String())
	}
}