package host

import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/layout"
	"github.com/carlosframework/platform/internal/registry"
)

type fakeKeyBackfiller struct {
	got   map[string]activator.InstanceKeyOutcome
	err   map[string]error
	asked []string
}

func (f *fakeKeyBackfiller) BackfillInstanceKey(_ context.Context, rt registry.Route) (activator.InstanceKeyOutcome, error) {
	f.asked = append(f.asked, rt.Host)
	return f.got[rt.Host], f.err[rt.Host]
}

func TestBackfillInstanceKeysWritesOnlyReadyActivatorInstances(t *testing.T) {
	ready := func(host string) registry.Route {
		return registry.Route{Host: host, Kind: "instance", App: "shop", Status: "ready", Socket: "/run/carlos/" + host + ".sock"}
	}
	held := ready("moving.example.com")
	held.Status = "moving"
	landing := ready("landing.example.com")
	landing.Status = "landing"
	provisioning := ready("new.example.com")
	provisioning.Status = "provisioning"
	unit := ready("unit.example.com")
	unit.Backing = registry.BackingUnit
	alias := ready("alias.example.com")
	alias.Platform = true
	sidecar := ready("worker.example.com")
	sidecar.Kind = "sidecar"
	addr := ready("addr.example.com")
	addr.Socket, addr.Addr = "", "127.0.0.1:8444"
	replica := ready("shop-au.example.com")
	replica.Replica = true
	f := &fakeKeyBackfiller{
		got: map[string]activator.InstanceKeyOutcome{"shop.example.com": activator.InstanceKeyMinted, "old.example.com": activator.InstanceKeyPresent, "cfg.example.com": activator.InstanceKeyFromConfig},
		err: map[string]error{"bad.example.com": errors.New("read: permission denied")},
	}
	st := backfillInstanceKeys(t.Context(), []registry.Route{ready("shop.example.com"), ready("old.example.com"), ready("cfg.example.com"), ready("bad.example.com"), held, landing, provisioning, unit, alias, sidecar, addr, replica}, f)
	if len(f.asked) != 4 {
		t.Fatalf("asked %v, want only the four ready activator instances", f.asked)
	}
	if st != (instanceKeyBackfill{Minted: 1, Present: 1, FromConfig: 1, Failed: 1, Skipped: 8}) {
		t.Fatalf("stats = %+v", st)
	}
}

// A refusal (a fenced or conflicting row) answers InstanceKeyPresent
// together with its error (activator.BackfillInstanceKey). The error
// decides: the row is a failure, never "present", or the rollout check
// would read a host whose key was never written as ready.
func TestBackfillInstanceKeysCountsARefusalAsFailedNotPresent(t *testing.T) {
	rt := registry.Route{Host: "shop.example.com", Kind: "instance", Status: "ready", Socket: "/run/carlos/shop.example.com.sock"}
	f := &fakeKeyBackfiller{
		got: map[string]activator.InstanceKeyOutcome{rt.Host: activator.InstanceKeyPresent},
		err: map[string]error{rt.Host: errors.New("not backfilled: layout migration in flight")},
	}
	st := backfillInstanceKeys(t.Context(), []registry.Route{rt}, f)
	if st != (instanceKeyBackfill{Failed: 1}) {
		t.Fatalf("stats = %+v, want the refusal counted as failed only", st)
	}
}

// A host with no own env file has never been started on this box: nothing
// is created, and the pass counts it on its own line.
func TestBackfillInstanceKeysCountsAHostNeverStartedHere(t *testing.T) {
	rt := registry.Route{Host: "shop.example.com", Kind: "instance", Status: "ready", Socket: "/run/carlos/shop.example.com.sock"}
	f := &fakeKeyBackfiller{got: map[string]activator.InstanceKeyOutcome{rt.Host: activator.InstanceKeyNoEnvFile}}
	st := backfillInstanceKeys(t.Context(), []registry.Route{rt}, f)
	if st != (instanceKeyBackfill{NoEnvFile: 1}) {
		t.Fatalf("stats = %+v", st)
	}
}

// A row whose configuration is on its way (activator.ErrInstanceKeyDeferred)
// is counted apart from a failure: not minted by design, and found at its
// next start. The whole pass runs under one budget, so a bucket that is
// down costs the pass at most instanceKeyBackfillBudget, not one read's
// timeout per row.
func TestBackfillInstanceKeysCountsADeferredRowApartWithinOneBudget(t *testing.T) {
	ready := func(host string) registry.Route {
		return registry.Route{Host: host, Kind: "instance", Status: "ready", Socket: "/run/carlos/" + host + ".sock"}
	}
	var deadlines []time.Time
	f := keyBackfillFunc(func(ctx context.Context, rt registry.Route) (activator.InstanceKeyOutcome, error) {
		d, ok := ctx.Deadline()
		if !ok {
			t.Fatalf("%s: the backfill's context has no deadline", rt.Host)
		}
		deadlines = append(deadlines, d)
		if rt.Host == "waiting.example.com" {
			return activator.InstanceKeyFailed, fmt.Errorf("%w: its configuration sets CARLOS_INSTANCE_KEY", activator.ErrInstanceKeyDeferred)
		}
		return activator.InstanceKeyMinted, nil
	})
	st := backfillInstanceKeys(context.Background(), []registry.Route{ready("waiting.example.com"), ready("shop.example.com")}, f)
	if st != (instanceKeyBackfill{Minted: 1, Deferred: 1}) {
		t.Fatalf("stats = %+v, want one minted and one waiting", st)
	}
	if len(deadlines) != 2 || !deadlines[0].Equal(deadlines[1]) || time.Until(deadlines[0]) > instanceKeyBackfillBudget {
		t.Fatalf("deadlines %v, want one budget shared by both rows", deadlines)
	}
}

// keyBackfillFunc is a backfiller made of one function.
type keyBackfillFunc func(context.Context, registry.Route) (activator.InstanceKeyOutcome, error)

func (f keyBackfillFunc) BackfillInstanceKey(ctx context.Context, rt registry.Route) (activator.InstanceKeyOutcome, error) {
	return f(ctx, rt)
}

// openSlot is a preparation slot with no fence held: it runs prepare.
func openSlot(_ string, prepare func() (string, error)) (string, error) { return prepare() }

// routeOf answers the registry's Get from a fixed list.
func routeOf(routes ...registry.Route) func(string) (*registry.Route, error) {
	return func(host string) (*registry.Route, error) {
		for _, rt := range routes {
			if rt.Host == host {
				return &rt, nil
			}
		}
		return nil, nil
	}
}

// Task 6 ruling (K1), as the v0.26.1 hotfix changed it: the start sequence
// is activatorStart's, so the two guards runAgent cannot be tested for are
// pinned here. A claim box starts and nothing else: no sweep, no recovery,
// no backfill (ruling 7). A legacy box sweeps, recovers, publishes the
// pass and starts, in that order, and the backfill runs after Start, once
// every journal is settled. A route list that fails skips the backfill
// only.
func TestActivatorStartBackfillsOnlyALegacyBoxAfterRecoveryAndStart(t *testing.T) {
	ready := registry.Route{Host: "shop.example.com", Kind: "instance", Status: "ready", Socket: "/run/carlos/shop.example.com.sock"}
	for _, tc := range []struct {
		name    string
		claims  bool
		listErr error
		want    []string
	}{
		{"legacy box", false, nil, []string{"configure", "sweep", "recover", "publish", "start", "ready", "backfill"}},
		{"legacy box, routes unreadable", false, errors.New("db locked"), []string{"configure", "sweep", "recover", "publish", "start", "ready"}},
		{"claim box", true, nil, []string{"configure", "start", "ready"}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			var mu sync.Mutex
			var events []string
			note := func(e string) { mu.Lock(); events = append(events, e); mu.Unlock() }
			done := activatorStart{
				Configure: func(context.Context) { note("configure") },
				Ready:     func() { note("ready") },
				Claims:    tc.claims,
				Sweep:     func(context.Context) error { note("sweep"); return nil },
				Recover:   func(context.Context, bool) *layout.Pass { note("recover"); return nil },
				Routes:    func() ([]registry.Route, error) { return []registry.Route{ready}, tc.listErr },
				Route:     routeOf(ready),
				Keys: keyBackfillFunc(func(context.Context, registry.Route) (activator.InstanceKeyOutcome, error) {
					note("backfill")
					return activator.InstanceKeyMinted, nil
				}),
				Slot:    openSlot,
				Publish: func(*layout.Pass) { note("publish") },
				Start:   func() { note("start") },
			}.Run(context.Background())
			if tc.claims {
				select {
				case <-done:
				default:
					t.Fatal("claim-box startup became asynchronous")
				}
			}
			select {
			case <-done:
			case <-time.After(10 * time.Second):
				t.Fatal("the backfill never finished")
			}
			mu.Lock()
			defer mu.Unlock()
			if !reflect.DeepEqual(events, tc.want) {
				t.Fatalf("events = %v, want %v", events, tc.want)
			}
		})
	}
}

// fenceCtx is a context with the deadline FenceAndStop requires.
func fenceCtx(t *testing.T, d time.Duration) context.Context {
	ctx, cancel := context.WithTimeout(t.Context(), d)
	t.Cleanup(cancel)
	return ctx
}

// Off the startup path, the backfill shares the box with the layout pass
// and moves, which fence a host before they move its files. A host that
// is fenced when its turn comes is skipped and nothing is asked of it, and
// a row whose status changed after the list (it began a move) is skipped
// the same way; its next start mints the key. The rest are written from
// the row as it stands, not as it was listed.
func TestBackfillInSlotSkipsAFencedHostAndReadsTheRowAgain(t *testing.T) {
	ready := func(host string) registry.Route {
		return registry.Route{Host: host, Kind: "instance", Status: "ready", Socket: "/run/carlos/" + host + ".sock"}
	}
	fenced, moved, shop := ready("fenced.example.com"), ready("moved.example.com"), ready("shop.example.com")
	svc := &activator.Service{}
	if _, err := svc.FenceAndStop(fenceCtx(t, time.Second), fenced.Host, 1); err != nil && !errors.Is(err, activator.ErrStopUnproven) {
		t.Fatalf("fence: %v", err)
	}
	nowMoving := moved
	nowMoving.Status = "moving"
	nowShop := shop
	nowShop.Socket = "/run/carlos/nested/shop.example.com.sock"
	f := &fakeKeyBackfiller{got: map[string]activator.InstanceKeyOutcome{shop.Host: activator.InstanceKeyMinted}}
	var sockets []string
	g := keyBackfillInSlot{
		keys: keyBackfillFunc(func(ctx context.Context, rt registry.Route) (activator.InstanceKeyOutcome, error) {
			sockets = append(sockets, rt.Socket)
			return f.BackfillInstanceKey(ctx, rt)
		}),
		slot:  svc.PrepareScheduledToken,
		route: routeOf(fenced, nowMoving, nowShop),
	}
	st := backfillInstanceKeys(t.Context(), []registry.Route{fenced, moved, shop}, g)
	if !reflect.DeepEqual(f.asked, []string{shop.Host}) {
		t.Fatalf("asked %v, want only %s", f.asked, shop.Host)
	}
	if !reflect.DeepEqual(sockets, []string{nowShop.Socket}) {
		t.Fatalf("written from %v, want the row as it stands now", sockets)
	}
	if st != (instanceKeyBackfill{Minted: 1, Skipped: 2}) {
		t.Fatalf("stats = %+v, want one minted and two skipped", st)
	}
}

// The other half of "nothing moves a file under it": a fence asked for
// while a row is being written (the layout pass or a move about to move
// the host's files) is installed only once the write has returned.
func TestBackfillInSlotHoldsOffAFenceUntilTheWriteReturns(t *testing.T) {
	t.Parallel()
	rt := registry.Route{Host: "shop.example.com", Kind: "instance", Status: "ready", Socket: "/run/carlos/shop.example.com.sock"}
	svc := &activator.Service{}
	writing, release := make(chan struct{}), make(chan struct{})
	var wrote atomic.Bool
	g := keyBackfillInSlot{
		keys: keyBackfillFunc(func(context.Context, registry.Route) (activator.InstanceKeyOutcome, error) {
			close(writing)
			<-release
			wrote.Store(true)
			return activator.InstanceKeyMinted, nil
		}),
		slot:  svc.PrepareScheduledToken,
		route: routeOf(rt),
	}
	done := make(chan instanceKeyBackfill)
	go func() { done <- backfillInstanceKeys(t.Context(), []registry.Route{rt}, g) }()
	<-writing
	fenced := make(chan bool)
	go func() {
		_, err := svc.FenceAndStopIfUnfenced(fenceCtx(t, 10*time.Second), rt.Host, 1)
		if err != nil && !errors.Is(err, activator.ErrStopUnproven) {
			t.Errorf("fence: %v", err)
		}
		fenced <- wrote.Load()
	}()
	select {
	case <-fenced:
		t.Fatal("the fence was installed while the backfill was writing the host")
	case <-time.After(200 * time.Millisecond):
	}
	close(release)
	if !<-fenced {
		t.Fatal("the fence was installed before the write returned")
	}
	if st := <-done; st != (instanceKeyBackfill{Minted: 1}) {
		t.Fatalf("stats = %+v", st)
	}
}

// v0.26.1 hotfix: the v0.26.0 roll to a 127-instance box failed its update
// gate because the backfill ran serially before Start, one bucket read per
// keyless row, and the edge never listened inside the gate's 20 seconds.
// On a box with hundreds of keyless rows and a slow bucket, Start is
// called, and Run returns (runAgent goes on to the agent socket and the
// edge), without waiting for the backfill.
func TestActivatorStartDoesNotWaitForTheBackfillOnALargeBox(t *testing.T) {
	const rows, perRow, bound = 300, 30 * time.Millisecond, 2 * time.Second
	var routes []registry.Route
	for i := range rows {
		host := fmt.Sprintf("shop%03d.example.com", i)
		routes = append(routes, registry.Route{Host: host, Kind: "instance", Status: "ready", Socket: "/run/carlos/" + host + ".sock"})
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	slowBucket := keyBackfillFunc(func(ctx context.Context, rt registry.Route) (activator.InstanceKeyOutcome, error) {
		select {
		case <-time.After(perRow):
			return activator.InstanceKeyMinted, nil
		case <-ctx.Done():
			return activator.InstanceKeyFailed, fmt.Errorf("%w: %w", activator.ErrInstanceKeyDeferred, ctx.Err())
		}
	})
	began := time.Now()
	started := make(chan time.Duration, 1)
	done := activatorStart{
		Sweep:   func(context.Context) error { return nil },
		Recover: func(context.Context, bool) *layout.Pass { return nil },
		Routes:  func() ([]registry.Route, error) { return routes, nil },
		Route:   routeOf(routes...),
		Keys:    slowBucket,
		Slot:    (&activator.Service{}).PrepareScheduledToken,
		Publish: func(*layout.Pass) {},
		Start:   func() { started <- time.Since(began) },
	}.Run(ctx)
	returned := time.Since(began)
	select {
	case startedAfter := <-started:
		if startedAfter > bound {
			t.Errorf("Start took %v, want within %v", startedAfter, bound)
		}
	case <-time.After(bound):
		t.Fatal("Start waited for the backfill")
	}
	if returned > bound {
		t.Errorf("Run returned after %v, want within %v: the edge waits for Run", returned, bound)
	}
	select {
	case <-done:
		t.Fatal("the backfill had finished when Run returned; it should still be running")
	default:
	}
	// Agent shutdown is not this test's: cancel the pass and let it end.
	cancel()
	select {
	case <-done:
	case <-time.After(30 * time.Second):
		t.Fatal("the backfill did not end once its context was cancelled")
	}
}