package activator

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

	"github.com/carlosframework/platform/internal/instancekey"
	"github.com/carlosframework/platform/internal/registry"
)

// bucketKey is ExecBackend.ConfiguredKey for a bucket whose configuration
// renders key ("" for none), counting its reads; down makes every read
// fail.
type bucketKey struct {
	key   string
	down  bool
	reads int
}

func (b *bucketKey) read(context.Context, registry.Route) (string, error) {
	b.reads++
	if b.down {
		return "", errors.New("bucket unreachable")
	}
	if b.key == "" {
		return "", nil
	}
	return instancekey.Digest(b.key), nil
}

// Final review I2 (Astra's P2), ruling 5: nothing is minted until the
// row's configuration is delivered. The files say what this box has
// delivered; the bucket says what the configuration sets. A key the bucket
// sets and the box has not delivered defers the mint; a bucket that cannot
// be read defers it the same way; a configuration that sets no key lets
// the mint proceed. The bucket is asked only when a mint is next: a key
// already delivered, in either file, costs no read.
func TestEnsureInstanceKeyAsksTheBucketBeforeItMints(t *testing.T) {
	tok := "CARLOS_ADMIN_TOKEN=" + strings.Repeat("c", 64) + "\n"
	for _, tc := range []struct {
		name        string
		own, config string
		bucket      *bucketKey
		want        InstanceKeyOutcome
		deferred    bool
		reads       int
	}{
		{"the configuration sets a key the box has not delivered: deferred", tok, "", &bucketKey{key: "key-A"}, InstanceKeyFailed, true, 1},
		{"the bucket cannot be read: deferred", tok, "", &bucketKey{down: true}, InstanceKeyFailed, true, 1},
		{"the configuration sets no key: minted", tok, "", &bucketKey{}, InstanceKeyMinted, false, 1},
		{"no configuration source: minted, nothing asked", tok, "", nil, InstanceKeyMinted, false, 0},
		{"the configured key is delivered: from config, nothing asked", tok, InstanceKeyVar + "=key-A\n", &bucketKey{key: "key-A"}, InstanceKeyFromConfig, false, 0},
		{"an own key: present, nothing asked", tok + InstanceKeyVar + "=" + strings.Repeat("K", 43) + "=\n", "", &bucketKey{key: "key-A"}, InstanceKeyPresent, false, 0},
	} {
		t.Run(tc.name, func(t *testing.T) {
			b := &ExecBackend{DataDir: t.TempDir()}
			if tc.bucket != nil {
				b.ConfiguredKey = tc.bucket.read
			}
			rt := registry.Route{Host: "shop.example.com", Kind: "instance"}
			own := filepath.Join(b.DataDir, rt.Host+".env")
			if err := os.WriteFile(own, []byte(tc.own), 0o600); err != nil {
				t.Fatal(err)
			}
			if tc.config != "" {
				if err := os.WriteFile(filepath.Join(b.DataDir, rt.Host+".config.env"), []byte(tc.config), 0o600); err != nil {
					t.Fatal(err)
				}
			}
			for _, call := range []string{"launch", "backfill"} {
				var got InstanceKeyOutcome
				var err error
				if call == "launch" {
					got, err = b.ensureInstanceKeyFor(t.Context(), rt, b.pathsFor(rt))
				} else {
					if err := os.WriteFile(own, []byte(tc.own), 0o600); err != nil {
						t.Fatal(err)
					}
					if tc.bucket != nil {
						tc.bucket.reads = 0
					}
					got, err = b.BackfillInstanceKey(t.Context(), rt)
				}
				if got != tc.want || errors.Is(err, ErrInstanceKeyDeferred) != tc.deferred || (err != nil) != tc.deferred {
					t.Fatalf("%s = %v, %v; want %v, deferred %v", call, got, err, tc.want, tc.deferred)
				}
				raw, _ := os.ReadFile(own)
				if minted := len(keyLines(raw)) > len(keyLines([]byte(tc.own))); minted != (tc.want == InstanceKeyMinted) {
					t.Fatalf("%s: a key line was appended: %v, want %v", call, minted, tc.want == InstanceKeyMinted)
				}
				if tc.bucket != nil && tc.bucket.reads != tc.reads {
					t.Fatalf("%s: the bucket was read %d times, want %d", call, tc.bucket.reads, tc.reads)
				}
			}
		})
	}
}

// The mint's bucket read is bounded by configReadTimeout and by the
// launch's own context, whichever ends first.
func TestEnsureInstanceKeyBoundsTheBucketRead(t *testing.T) {
	b := &ExecBackend{DataDir: t.TempDir()}
	var deadline time.Time
	b.ConfiguredKey = func(ctx context.Context, _ registry.Route) (string, error) {
		deadline, _ = ctx.Deadline()
		return "", ctx.Err()
	}
	rt := registry.Route{Host: "shop.example.com", Kind: "instance"}
	if err := os.WriteFile(filepath.Join(b.DataDir, rt.Host+".env"), []byte("A=1\n"), 0o600); err != nil {
		t.Fatal(err)
	}
	if _, err := b.ensureInstanceKeyFor(context.Background(), rt, b.pathsFor(rt)); err != nil {
		t.Fatal(err)
	}
	if until := time.Until(deadline); until <= 0 || until > configReadTimeout {
		t.Fatalf("the read's deadline is %v away, want within %v", until, configReadTimeout)
	}
	if err := os.WriteFile(filepath.Join(b.DataDir, rt.Host+".env"), []byte("A=1\n"), 0o600); err != nil {
		t.Fatal(err) // keyless again, so the backfill reaches the read
	}
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	if _, err := b.BackfillInstanceKey(ctx, rt); !errors.Is(err, ErrInstanceKeyDeferred) || !errors.Is(err, context.Canceled) {
		t.Fatalf("a cancelled read = %v, want deferred by the cancellation", err)
	}
}

// A first start whose configuration is still on its way, end to end: the
// start is deferred and no app runs; nothing is minted into <host>.env;
// once the config sync has delivered <host>.config.env with the configured
// key, the start runs and the app seals with that key, the configured one,
// never a minted one.
func TestFirstStartWaitsForTheConfiguredKey(t *testing.T) {
	b := testExecBackend(t)
	bucket := &bucketKey{key: "key-A"}
	b.ConfiguredKey = bucket.read
	rt := registry.Route{Host: "shop.example.com", Kind: "instance", Socket: noControlSocket(t)}
	own := filepath.Join(b.DataDir, rt.Host+".env")
	tok := "CARLOS_ADMIN_TOKEN=" + strings.Repeat("c", 64) + "\n"
	if err := os.WriteFile(own, []byte(tok), 0o600); err != nil {
		t.Fatal(err)
	}
	out := stageHelperApp(t, b, rt.Host, "v1")
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	proc, _, err := b.Start(ctx, rt)
	if proc != nil {
		proc.Stop(context.Background())
	}
	if !errors.Is(err, ErrInstanceKeyDeferred) || !strings.Contains(err.Error(), rt.Host) {
		t.Fatalf("Start before delivery = %v, want ErrInstanceKeyDeferred naming the host", err)
	}
	if got := outcomeOf(out); got != "" {
		t.Fatalf("an app ran before delivery: %q", got)
	}
	if raw, _ := os.ReadFile(own); string(raw) != tok {
		t.Fatalf("the own file changed before delivery: %d bytes", len(raw))
	}
	// The config sync delivers the file.
	if err := os.WriteFile(filepath.Join(b.DataDir, rt.Host+".config.env"), []byte(InstanceKeyVar+"=key-A\n"), 0o600); err != nil {
		t.Fatal(err)
	}
	// The helper exits once it has recorded its outcome, so Start's health
	// wait never sees a live socket: the outcome is the proof, as in the
	// replica tests, and the error only must not be the deferral.
	ctx2, cancel2 := recordedCtx(2*time.Second, out) // ctx was spent waiting for the first outcome
	defer cancel2()
	proc, _, err = b.Start(ctx2, rt)
	if proc != nil {
		defer proc.Stop(context.Background())
	}
	if errors.Is(err, ErrInstanceKeyDeferred) {
		t.Fatalf("Start after delivery still deferred: %v", err)
	}
	if got := outcomeOf(out); got != "key "+instancekey.Digest("key-A") {
		t.Fatalf("the app's outcome = %q, want the configured key", got)
	}
	if raw, _ := os.ReadFile(own); string(raw) != tok {
		t.Fatal("a key was minted beside the configured one")
	}
}

// The re-review's finding: the bucket read runs outside adminTokenMu, so a
// bucket that is slow or blocked for one row's first wake never holds the
// lock every other row's mint waits on. Host A's read blocks until the
// test releases it; host B mints meanwhile; A then mints too.
func TestASlowBucketReadForOneHostDoesNotBlockAnothersMint(t *testing.T) {
	b := &ExecBackend{DataDir: t.TempDir()}
	release := make(chan struct{})
	entered := make(chan struct{})
	b.ConfiguredKey = func(ctx context.Context, rt registry.Route) (string, error) {
		if rt.Host == "a.example.com" {
			close(entered)
			select {
			case <-release:
			case <-ctx.Done():
				return "", ctx.Err()
			}
		}
		return "", nil
	}
	a := registry.Route{Host: "a.example.com", Kind: "instance"}
	bb := registry.Route{Host: "b.example.com", Kind: "instance"}
	for _, rt := range []registry.Route{a, bb} {
		if err := os.WriteFile(filepath.Join(b.DataDir, rt.Host+".env"), []byte("X=1\n"), 0o600); err != nil {
			t.Fatal(err)
		}
	}
	aDone := make(chan error, 1)
	go func() {
		_, err := b.ensureInstanceKeyFor(context.Background(), a, b.pathsFor(a))
		aDone <- err
	}()
	select {
	case <-entered: // A is inside its bucket read before B starts
	case <-time.After(2 * time.Second):
		t.Fatal("host A never reached its bucket read")
	}
	bDone := make(chan error, 1)
	go func() {
		_, err := b.ensureInstanceKeyFor(context.Background(), bb, b.pathsFor(bb))
		bDone <- err
	}()
	select {
	case err := <-bDone:
		if err != nil {
			t.Fatalf("host B's mint: %v", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("host B's mint waited on host A's bucket read")
	}
	select {
	case <-aDone:
		t.Fatal("host A minted before its bucket answered")
	default:
	}
	close(release)
	select {
	case err := <-aDone:
		if err != nil {
			t.Fatalf("host A's mint after the bucket answered: %v", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("host A never minted")
	}
	for _, rt := range []registry.Route{a, bb} {
		raw, _ := os.ReadFile(filepath.Join(b.DataDir, rt.Host+".env"))
		if len(keyLines(raw)) != 1 {
			t.Fatalf("%s: %d key lines, want one", rt.Host, len(keyLines(raw)))
		}
	}
}

// A key that appears between the two phases is never overwritten: the
// bucket read happens outside the lock, so the files it looked at may
// change before the decision. The locked read decides. Here the bucket
// reader itself, standing in for a concurrent writer, puts a key into the
// own file, or delivers the configuration with one, before it answers.
func TestAKeyThatAppearsDuringTheBucketReadIsKept(t *testing.T) {
	key := strings.Repeat("K", 43) + "="
	for _, tc := range []struct {
		name  string
		plant func(dataDir, host string)
		want  InstanceKeyOutcome
	}{
		{"into the own file", func(d, h string) {
			f, err := os.OpenFile(filepath.Join(d, h+".env"), os.O_APPEND|os.O_WRONLY, 0)
			if err != nil {
				t.Fatal(err)
			}
			f.WriteString(InstanceKeyVar + "=" + key + "\n")
			f.Close()
		}, InstanceKeyPresent},
		{"delivered in the configuration", func(d, h string) {
			if err := os.WriteFile(filepath.Join(d, h+".config.env"), []byte(InstanceKeyVar+"="+key+"\n"), 0o600); err != nil {
				t.Fatal(err)
			}
		}, InstanceKeyFromConfig},
	} {
		t.Run(tc.name, func(t *testing.T) {
			b := &ExecBackend{DataDir: t.TempDir()}
			rt := registry.Route{Host: "shop.example.com", Kind: "instance"}
			own := filepath.Join(b.DataDir, rt.Host+".env")
			if err := os.WriteFile(own, []byte("X=1\n"), 0o600); err != nil {
				t.Fatal(err)
			}
			reads := 0
			b.ConfiguredKey = func(context.Context, registry.Route) (string, error) {
				reads++
				tc.plant(b.DataDir, rt.Host)
				return "", nil // the bucket says: no configured key, mint away
			}
			got, err := b.ensureInstanceKeyFor(context.Background(), rt, b.pathsFor(rt))
			if err != nil || got != tc.want || reads != 1 {
				t.Fatalf("ensure = %v, %v after %d reads; want %v", got, err, reads, tc.want)
			}
			raw, _ := os.ReadFile(own)
			ownKeys := keyLines(raw)
			if tc.want == InstanceKeyPresent && (len(ownKeys) != 1 || ownKeys[0] != key) {
				t.Fatalf("own file keys %d, want the one that appeared, kept", len(ownKeys))
			}
			if tc.want == InstanceKeyFromConfig && len(ownKeys) != 0 {
				t.Fatal("a key was minted beside the one the configuration delivered")
			}
		})
	}
}

// The other side of the two phases: a mint that looks next only under the
// lock has no bucket verdict, and defers rather than mint on nothing. The
// first read saw a key in the own file (or, on the backfill, no file at
// all), so the bucket was not asked; by the locked read the key is gone
// (or a keyless file has appeared). Nothing is minted, the bucket is
// still unasked, and the next attempt asks it.
func TestAMintThatLooksNextOnlyUnderTheLockDefers(t *testing.T) {
	key := strings.Repeat("K", 43) + "="
	for _, tc := range []struct {
		name   string
		before string // the own file at the first read; "" is no file
		after  string // the own file at the locked read
		create bool
	}{
		{"a key seen by the first read is gone", "X=1\n" + InstanceKeyVar + "=" + key + "\n", "X=1\n", true},
		{"a keyless file appears after the backfill's first read", "", "X=1\n", false},
	} {
		t.Run(tc.name, func(t *testing.T) {
			b := &ExecBackend{DataDir: t.TempDir()}
			bucket := &bucketKey{}
			b.ConfiguredKey = bucket.read
			rt := registry.Route{Host: "shop.example.com", Kind: "instance"}
			own := filepath.Join(b.DataDir, rt.Host+".env")
			if tc.before != "" {
				if err := os.WriteFile(own, []byte(tc.before), 0o600); err != nil {
					t.Fatal(err)
				}
			}
			instanceKeyHooks.betweenPhases = func(string) {
				if err := os.WriteFile(own, []byte(tc.after), 0o600); err != nil {
					t.Error(err)
				}
			}
			t.Cleanup(func() { instanceKeyHooks.betweenPhases = nil })
			got, err := b.ensureInstanceKeyAt(context.Background(), rt, b.pathsFor(rt), tc.create, "minted")
			if got != InstanceKeyFailed || !errors.Is(err, ErrInstanceKeyDeferred) {
				t.Fatalf("ensure = %v, %v; want deferred", got, err)
			}
			if bucket.reads != 0 {
				t.Fatalf("the bucket was read %d times, want 0", bucket.reads)
			}
			raw, _ := os.ReadFile(own)
			if string(raw) != tc.after {
				t.Fatalf("own file changed by the deferred ensure (%d bytes, want %d)", len(raw), len(tc.after))
			}
		})
	}
}