package schedule

import (
	"context"
	"errors"
	"net/http"
	"path/filepath"
	"sync/atomic"
	"testing"
	"time"

	"encoding/json"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"reflect"
)

type lifecycleWork struct{ valid *atomic.Bool }

func (w lifecycleWork) With(_ context.Context, _ hostwrite.Identity, fn func() error) error {
	if !w.valid.Load() {
		return errors.New("selection changed")
	}
	return fn()
}

type lifecycleSelector struct{ valid *atomic.Bool }

func (s lifecycleSelector) Select(context.Context, string, string) (hostwrite.Work, error) {
	return lifecycleWork{valid: s.valid}, nil
}

type pausedDelivery struct {
	started chan struct{}
	resume  chan struct{}
}

func (d pausedDelivery) Deliver(context.Context, registry.Route, *http.Request) (int, error) {
	close(d.started)
	<-d.resume
	return 204, nil
}

func TestRunnerChangedSelectionAfterDeliveryWritesNoBookkeeping(t *testing.T) {
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { reg.Close() })
	host := "app.bab.oncarlos.com"
	if err := reg.Upsert(registry.Route{Host: host, Kind: "instance", Account: "bab", App: "app", Status: "ready"}); err != nil {
		t.Fatal(err)
	}
	now := time.Unix(10_000, 0).UTC()
	if err := reg.UpsertTimer(registry.Timer{Host: host, Name: "tick", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/tick", NextAt: now.Unix()}); err != nil {
		t.Fatal(err)
	}
	valid := &atomic.Bool{}
	valid.Store(true)
	delivery := pausedDelivery{started: make(chan struct{}), resume: make(chan struct{})}
	events := atomic.Int64{}
	r := &Runner{
		StartDelivery: func(ctx context.Context, w hostwrite.Work, _ registry.Route, _ registry.Timer) (context.Context, func(), error) {
			err := w.With(ctx, hostwrite.Identity{}, func() error { return nil })
			return ctx, func() {}, err
		},
		Reg: reg, Deliver: delivery, Now: func() time.Time { return now }, Timeout: time.Minute,
		Token:      func(string) (string, error) { return "token", nil },
		Event:      func(registry.Route, string, string) { events.Add(1) },
		HostWrites: hostwrite.Config{Enabled: true, Selector: lifecycleSelector{valid: valid}},
	}
	r.Tick(t.Context())
	<-delivery.started
	valid.Store(false)
	close(delivery.resume)
	r.wait()
	row, err := reg.GetTimer(host, "tick")
	if err != nil {
		t.Fatal(err)
	}
	if row.LastAt != 0 || row.NextAt != now.Unix() || events.Load() != 0 {
		t.Fatalf("bookkeeping changed after authority loss: row=%+v events=%d", row, events.Load())
	}
}

func TestEnabledRunnerMissingLifecycleProviderRefusesTimer(t *testing.T) {
	d := &fakeDeliverer{status: 204}
	reg, r := runnerFixture(t, d)
	now := time.Unix(20_000, 0).UTC()
	r.Now = func() time.Time { return now }
	r.HostWrites = hostwrite.Config{Enabled: true}
	if err := reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "tick", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/tick", NextAt: now.Unix()}); err != nil {
		t.Fatal(err)
	}
	r.Tick(t.Context())
	r.wait()
	if len(d.calls) != 0 {
		t.Fatal("delivery ran without lifecycle provider")
	}
}

type rotatingSelector struct {
	current *atomic.Int64
	calls   int
}
type capturedEpoch struct {
	current  *atomic.Int64
	selected int64
}

func (s *rotatingSelector) Select(context.Context, string, string) (hostwrite.Work, error) {
	s.calls++
	return capturedEpoch{s.current, s.current.Load()}, nil
}
func (w capturedEpoch) With(_ context.Context, _ hostwrite.Identity, fn func() error) error {
	if w.current.Load() != w.selected {
		return errors.New("installation changed")
	}
	return fn()
}

type changeOnRead struct {
	release.Store
	key     string
	current *atomic.Int64
}

func (s changeOnRead) Get(ctx context.Context, key string) ([]byte, string, error) {
	b, e, err := s.Store.Get(ctx, key)
	if key == s.key {
		s.current.Add(1)
	}
	return b, e, err
}

func TestConvergerCapturesBeforeDeclarationAndRunReads(t *testing.T) {
	for _, key := range []string{DeclarationKey("bab", "app"), RunKey("bab", "app")} {
		t.Run(key, func(t *testing.T) {
			reg, store := convergeFixture(t)
			decl := []byte(`{"schedules":[{"name":"tick","every":"1h","path":"/tick"}]}`)
			if _, err := store.Put(t.Context(), DeclarationKey("bab", "app"), decl, ""); err != nil {
				t.Fatal(err)
			}
			epoch := &atomic.Int64{}
			selector := &rotatingSelector{current: epoch}
			c := NewConverger(reg, changeOnRead{store, key, epoch}, "box", nil)
			c.HostWrites = hostwrite.Config{Enabled: true, Selector: selector}
			if err := c.Pass(t.Context()); err == nil {
				t.Fatal("old inputs acquired successor authority")
			}
			rows, err := reg.TimersFor("a.bab.oncarlos.com")
			if err != nil || len(rows) != 0 {
				t.Fatalf("stale rows=%v %v", rows, err)
			}
			if selector.calls != 4 {
				t.Fatalf("selection refreshed: %d", selector.calls)
			}
		})
	}
}

type replaceAtWork struct {
	n, at   int
	replace func()
}

func (w *replaceAtWork) With(_ context.Context, _ hostwrite.Identity, fn func() error) error {
	w.n++
	if w.n == w.at {
		w.replace()
	}
	return fn()
}
func TestConvergerRegistrationMutationsPreserveReplacement(t *testing.T) {
	for _, kind := range []string{"delete", "unit-delete", "status", "run", "upsert"} {
		t.Run(kind, func(t *testing.T) {
			reg, _ := convergeFixture(t)
			host := "a.bab.oncarlos.com"
			original := registry.Timer{Host: host, Name: "tick", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/tick", NextAt: 100, DueAt: 100}
			if err := reg.UpsertTimer(original); err != nil {
				t.Fatal(err)
			}
			rt, _ := reg.Get(host)
			var decl Declaration
			if err := json.Unmarshal([]byte(`{"schedules":[{"name":"tick","every":"1h","path":"/tick"}]}`), &decl); err != nil {
				t.Fatal(err)
			}
			fire := ""
			at := 2
			switch kind {
			case "delete":
				decl = Declaration{}
			case "unit-delete":
				rt.Backing = registry.BackingUnit
				decl = Declaration{}
			case "status":
				decl = Declaration{}
				if err := json.Unmarshal([]byte(`{"schedules":[{"name":"tick","cron":"0 0 31 2 *","path":"/tick"}]}`), &decl); err != nil {
					t.Fatal(err)
				}
				at = 3
			case "run":
				fire = "tick"
				at = 4
			}
			var replacement *registry.Timer
			w := &replaceAtWork{at: at, replace: func() {
				new := original
				new.Path = "/successor"
				new.NextAt = 999
				new.DueAt = 999
				if err := reg.UpsertTimer(new); err != nil {
					t.Fatal(err)
				}
				replacement, _ = reg.GetTimer(host, "tick")
			}}
			ctx := context.WithValue(t.Context(), timerWorkKey{}, timerWork{ctx: t.Context(), work: w})
			c := NewConverger(reg, nil, "box", nil)
			c.HostWrites.Enabled = true
			_, _ = c.convergeHost(ctx, *rt, decl, fire, time.Unix(1000, 0))
			got, err := reg.GetTimer(host, "tick")
			if replacement == nil || err != nil || !reflect.DeepEqual(got, replacement) {
				t.Fatalf("%s replaced row: got=%+v want=%+v err=%v work=%d", kind, got, replacement, err, w.n)
			}
		})
	}
}

func (d pausedDelivery) DeliverClaim(ctx context.Context, rt registry.Route, req *http.Request, admit func(context.Context) (context.Context, func(), error)) (int, error) {
	child, finish, err := admit(ctx)
	if err != nil {
		return 0, err
	}
	defer finish()
	return d.Deliver(child, rt, req)
}