package activator

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"net"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
)

// syncBuffer is a bytes.Buffer two goroutines may write.
type syncBuffer struct {
	mu sync.Mutex
	b  bytes.Buffer
}

func (s *syncBuffer) Write(p []byte) (int, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.b.Write(p)
}
func (s *syncBuffer) String() string { s.mu.Lock(); defer s.mu.Unlock(); return s.b.String() }

func TestCmdVarsComposeTheRunEnvironment(t *testing.T) {
	b := &ExecBackend{Region: "ie"}
	rt := registry.Route{Host: "h.example.com"}
	file := map[string]string{
		"APP_KEY": "k", "CARLOS_ADMIN_TOKEN": "tok", "CARLOS_CONTROL_SOCKET": "/x", "CARLOS_PURGE": "header",
		"CARLOS_REGION": "us", "CARLOS_HOST": "liar.example.com", "CARLOS_RUN_ID": "planted",
	}
	got := b.cmdVars(rt, file, "/data/instances/h.example.com/app.db", "run_a", "/data/instances/h.example.com/.carlos-run/run_a")
	want := map[string]string{
		"APP_KEY": "k", "CARLOS_ADMIN_TOKEN": "tok", "CARLOS_REGION": "ie", "CARLOS_HOST": "h.example.com",
		"CARLOS_DB": "/data/instances/h.example.com/app.db", "CARLOS_RUN_ID": "run_a",
		"CARLOS_RUN_DIR": "/data/instances/h.example.com/.carlos-run/run_a",
	}
	if len(got) != len(want) {
		t.Fatalf("env = %v", got)
	}
	for k, v := range want {
		if got[k] != v {
			t.Errorf("%s = %q, want %q", k, got[k], v)
		}
	}
	b.Region = ""
	if _, ok := b.cmdVars(rt, file, "/db", "run_a", "/d")["CARLOS_REGION"]; ok {
		t.Error("an empty backend region must remove CARLOS_REGION (codex 7)")
	}
}

func startTestCmd(t *testing.T) (*ExecBackend, registry.Route, *fakeControl, *fakeUnit, CmdUnit, *syncBuffer, *syncBuffer) {
	t.Helper()
	ctl := &fakeControl{cmdInv: "inv1"}
	unit := &fakeUnit{}
	b, rt := nestedBackend(t, ctl, unit)
	b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" }
	out, errs := &syncBuffer{}, &syncBuffer{}
	dir := filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), CmdRunDirName, "run_a")
	u, err := b.StartCmd(context.Background(), rt, CmdSpec{RunID: "run_a", Args: []string{"import", "--dry-run"}, RunDir: dir, Stdout: out, Stderr: errs,
		genEnv: map[string]string{"CARLOS_ADMIN_TOKEN": "tok"}})
	if err != nil {
		t.Fatalf("StartCmd: %v", err)
	}
	return b, rt, ctl, unit, u, out, errs
}

// shrinkCmdAcceptGrace sets how long a cmd stream's Drain waits for a
// connection that never came, for one test. The fake units here never
// connect their streams, so every Drain waits the whole grace.
func shrinkCmdAcceptGrace(t *testing.T, d time.Duration) {
	t.Helper()
	old := cmdAcceptGrace
	cmdAcceptGrace = d
	t.Cleanup(func() { cmdAcceptGrace = old })
}

func TestStartCmdWritesTheLaunchFileAndStartsTheUnit(t *testing.T) {
	shrinkCmdAcceptGrace(t, 50*time.Millisecond) // the cleanup drains two unconnected streams
	b, rt, ctl, _, u, _, _ := startTestCmd(t)
	t.Cleanup(func() { u.StopAndSettle(context.Background()) })
	if u.Invocation() != "inv1" || !strings.Contains(ctl.joined(), "cmd "+rt.Host) {
		t.Fatalf("invocation %q, calls\n%s", u.Invocation(), ctl.joined())
	}
	raw, err := os.ReadFile(filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), "launch.cmd.json"))
	if err != nil {
		t.Fatal(err)
	}
	var l instancectl.Launch
	if err := json.Unmarshal(raw, &l); err != nil {
		t.Fatal(err)
	}
	if len(l.Argv) != 3 || l.Argv[0] != b.liveLink(rt.Host) || l.Argv[1] != "import" || l.Argv[2] != "--dry-run" {
		t.Errorf("argv = %q: exactly the binary and what followed --", l.Argv)
	}
	env := strings.Join(l.Env, "\n")
	for _, want := range []string{"CARLOS_RUN_ID=run_a", "CARLOS_HOST=" + rt.Host, "CARLOS_DB=" + rt.DBPath} {
		if !strings.Contains(env, want) {
			t.Errorf("env lacks %q", want)
		}
	}
	if strings.Contains(env, "CARLOS_CONTROL_SOCKET=") {
		t.Error("a command has no control socket")
	}
}

func TestStartCmdRefusesAFlatInstanceAndAForeignRunDir(t *testing.T) {
	ctl := &fakeControl{cmdInv: "inv1"}
	b, rt := nestedBackend(t, ctl, &fakeUnit{})
	b.layoutFn = func(registry.Route) (Layout, string) { return LayoutFlat, "test" }
	if _, err := b.StartCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}); !errors.Is(err, ErrCmdNotNested) {
		t.Errorf("flat: %v, want ErrCmdNotNested", err)
	}
	b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" }
	if _, err := b.StartCmd(context.Background(), rt, CmdSpec{RunID: "run_a", RunDir: "/elsewhere"}); err == nil {
		t.Error("a run directory outside the instance's .carlos-run was accepted")
	}
	if strings.Contains(ctl.joined(), "cmd ") {
		t.Errorf("a refused start reached the control service:\n%s", ctl.joined())
	}
}

func TestStartCmdRefusesAnUnsafeRunID(t *testing.T) {
	for _, id := range []string{"../x", "", "a/b", ".", "..", "../../other.example.com", `a\b`} {
		ctl := &fakeControl{cmdInv: "inv1"}
		b, rt := nestedBackend(t, ctl, &fakeUnit{})
		b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" }
		dir := provision.InstanceDir(b.DataDir, rt.Host)
		// the RunDir that matches the id after cleaning: the caller is
		// "consistent", so only the id itself can be refused
		runDir := filepath.Join(dir, CmdRunDirName, id)
		if _, err := b.StartCmd(context.Background(), rt, CmdSpec{RunID: id, RunDir: runDir}); err == nil {
			t.Errorf("run id %q was accepted", id)
		}
		if strings.Contains(ctl.joined(), "cmd ") {
			t.Errorf("run id %q reached the control service", id)
		}
		if _, err := os.Stat(dir); !os.IsNotExist(err) {
			t.Errorf("run id %q: the instance directory was made (%v)", id, err)
		}
	}
}

func TestStopAndSettleDrainsBothStreams(t *testing.T) {
	t.Parallel()
	b, rt, ctl, unit, u, out, errs := startTestCmd(t)
	unit.set("active", "inv1", true, true)
	write := func(path, s string) {
		c, err := net.Dial("unix", path)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := c.Write([]byte(s)); err != nil {
			t.Fatal(err)
		}
		c.Close()
	}
	write(b.logSocketNamed(rt.Host, string(instancectl.RoleCmd)), "out-1\nout-2")
	write(b.logSocketNamed(rt.Host, cmdErrSocket), "err-1")
	go func() {
		time.Sleep(30 * time.Millisecond)
		unit.set("inactive", "inv1", false, false)
	}()
	if err := u.StopAndSettle(context.Background()); err != nil {
		t.Fatalf("StopAndSettle: %v", err)
	}
	if out.String() != "out-1\nout-2" || errs.String() != "err-1" {
		t.Errorf("stdout %q stderr %q", out.String(), errs.String())
	}
	if !strings.Contains(ctl.joined(), "stop "+rt.Host+" cmd") {
		t.Errorf("the cmd unit was not stopped:\n%s", ctl.joined())
	}
	if err := u.StopAndSettle(context.Background()); err != nil {
		t.Errorf("a second StopAndSettle: %v", err)
	}
}

// A connection the unit made at exec may still be queued in the listener's
// backlog when the unit has exited: Drain must take it and deliver its
// bytes (run spec §5.5; plan Review Focus 5).
func TestCmdStreamDrainTakesAConnectionStillInTheBacklog(t *testing.T) {
	dir, err := os.MkdirTemp("", "cs")
	if err != nil {
		t.Fatal(err)
	}
	defer os.RemoveAll(dir)
	for i := 0; i < 200; i++ {
		var w syncBuffer
		s, err := listenCmdStream(filepath.Join(dir, "s.sock"), &w)
		if err != nil {
			t.Fatal(err)
		}
		c, err := net.Dial("unix", filepath.Join(dir, "s.sock"))
		if err != nil {
			t.Fatal(err)
		}
		if _, err := c.Write([]byte("last words")); err != nil {
			t.Fatal(err)
		}
		c.Close()
		if err := s.Drain(time.Second); err != nil {
			t.Fatalf("round %d: Drain: %v", i, err)
		}
		if w.String() != "last words" {
			t.Fatalf("round %d: got %q", i, w.String())
		}
	}
}

func TestCmdStreamDrainOfAnUnconnectedStreamEndsAfterTheGrace(t *testing.T) {
	shrinkCmdAcceptGrace(t, 100*time.Millisecond)
	dir, err := os.MkdirTemp("", "cs")
	if err != nil {
		t.Fatal(err)
	}
	defer os.RemoveAll(dir)
	s, err := listenCmdStream(filepath.Join(dir, "s.sock"), nil)
	if err != nil {
		t.Fatal(err)
	}
	start := time.Now()
	if err := s.Drain(time.Second); err != nil {
		t.Fatal(err)
	}
	if d := time.Since(start); d > 3*cmdAcceptGrace {
		t.Errorf("took %s", d)
	}
}