package spindle

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/http/cgi"
	"net/http/httptest"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"slices"
	"strings"
	"sync"
	"testing"

	"github.com/bluesky-social/indigo/atproto/identity"
	"github.com/bluesky-social/indigo/atproto/syntax"

	"tangled.org/core/gitutil"
	"tangled.org/core/spindle/config"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/knotclient"
)

const defsWorkflow = "engine: nixery\n"

type defsFile struct{ name, mode, body string }

// a knot whose git is a bare repo per did behind git http-backend; it logs
// smart http requests as "caps", "ls-refs <prefixes>" or "fetch <wants>"
type defsKnot struct {
	srv  *httptest.Server
	root string
	git  *cgi.Handler

	mu       sync.Mutex
	requests []string
}

func newDefsKnot(t *testing.T) *defsKnot {
	t.Helper()
	k := &defsKnot{root: t.TempDir()}
	k.git = &cgi.Handler{
		Path:       gitExecPath(t),
		Root:       "/",
		Env:        []string{"GIT_PROJECT_ROOT=" + k.root, "GIT_HTTP_EXPORT_ALL=1"},
		InheritEnv: []string{"HOME", "GIT_CONFIG_GLOBAL", "GIT_CONFIG_SYSTEM"},
	}
	k.srv = httptest.NewServer(http.HandlerFunc(k.serve))
	t.Cleanup(k.srv.Close)
	return k
}

func (k *defsKnot) host() string {
	u, _ := url.Parse(k.srv.URL)
	return u.Host
}

func (k *defsKnot) bare(t *testing.T, repo syntax.DID) string {
	t.Helper()
	dir := filepath.Join(k.root, repo.String())
	if _, err := os.Stat(dir); err == nil {
		return dir
	}
	gitInDir(t, k.root, "init", "-q", "--bare", dir)
	gitInDir(t, dir, "config", "uploadpack.allowFilter", "true")
	gitInDir(t, dir, "config", "uploadpack.allowAnySHA1InWant", "true")
	return dir
}

// a commit holding files under .tangled/workflows, made with plumbing so any
// mode goes; it stays fetchable by oid without a branch
func (k *defsKnot) commit(t *testing.T, repo syntax.DID, files ...defsFile) string {
	t.Helper()
	dir := k.bare(t, repo)
	env := append(os.Environ(), "GIT_INDEX_FILE="+filepath.Join(t.TempDir(), "index"))
	git := func(stdin string, args ...string) string {
		cmd := exec.Command("git", append([]string{"-C", dir}, args...)...)
		cmd.Env = env
		cmd.Stdin = strings.NewReader(stdin)
		out, err := cmd.CombinedOutput()
		if err != nil {
			t.Fatalf("git %v: %v\n%s", args, err, out)
		}
		return strings.TrimSpace(string(out))
	}
	git("", "read-tree", "--empty")
	for _, f := range files {
		path, mode := ".tangled/workflows/"+f.name, strings.TrimLeft(f.mode, "0")
		var oid string
		switch mode {
		case "160000":
			oid = strings.Repeat("5", 40)
		case "40000":
			path, mode = path+"/inner.yaml", "100644"
			fallthrough
		default:
			oid = git(f.body, "hash-object", "-w", "--stdin")
		}
		git("", "update-index", "--add", "--cacheinfo", mode+","+oid+","+path)
	}
	tree := git("", "write-tree")
	commit := git("", "-c", "user.name=ci", "-c", "user.email=ci@nel.pet", "commit-tree", tree, "-m", "wip")
	git("", "update-ref", "refs/keep/"+commit, commit)
	return commit
}

func (k *defsKnot) setBranch(t *testing.T, repo syntax.DID, name, commit string) {
	gitInDir(t, k.bare(t, repo), "update-ref", "refs/heads/"+name, commit)
}

func (k *defsKnot) setTag(t *testing.T, repo syntax.DID, name, commit string) {
	gitInDir(t, k.bare(t, repo), "-c", "user.name=ci", "-c", "user.email=ci@nel.pet", "tag", "-a", "-m", name, name, commit)
}

func (k *defsKnot) asked() []string {
	k.mu.Lock()
	defer k.mu.Unlock()
	return append([]string(nil), k.requests...)
}

func (k *defsKnot) count(kind string) int {
	n := 0
	for _, r := range k.asked() {
		if r == kind || strings.HasPrefix(r, kind+" ") {
			n++
		}
	}
	return n
}

func (k *defsKnot) log(r *http.Request) {
	entry := "caps"
	if r.Method == http.MethodPost {
		body, _ := io.ReadAll(r.Body)
		r.Body = io.NopCloser(bytes.NewReader(body))
		var command string
		var args []string
		for _, line := range strings.Split(string(body), "\n") {
			if len(line) < 4 {
				continue
			}
			line = line[4:]
			if c, ok := strings.CutPrefix(line, "command="); ok {
				command = c
			} else if a, ok := strings.CutPrefix(line, "want "); ok {
				args = append(args, a)
			} else if a, ok := strings.CutPrefix(line, "ref-prefix "); ok {
				args = append(args, a)
			}
		}
		entry = strings.TrimSpace(command + " " + strings.Join(args, ","))
	}
	k.mu.Lock()
	k.requests = append(k.requests, entry)
	k.mu.Unlock()
}

func (k *defsKnot) serve(w http.ResponseWriter, r *http.Request) {
	if r.URL.Path == "/xrpc/com.atproto.repo.getRecord" {
		json.NewEncoder(w).Encode(map[string]any{"value": map[string]any{"defaultBranch": "main"}})
		return
	}
	if strings.HasPrefix(r.URL.Path, "/xrpc/") {
		w.WriteHeader(http.StatusNotFound)
		json.NewEncoder(w).Encode(map[string]string{"error": "MethodNotImplemented"})
		return
	}
	k.log(r)
	k.git.ServeHTTP(w, r)
}

func newDefsSpindle(t *testing.T, knot *defsKnot, repos ...syntax.DID) *Spindle {
	t.Helper()
	ctx := context.Background()
	d, err := db.Make(ctx, filepath.Join(t.TempDir(), "spindle.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { d.Close() })
	for i, repo := range repos {
		if err := d.AddRepo(db.Repo{Knot: knot.host(), Owner: "did:plc:owner", Rkey: syntax.RecordKey(fmt.Sprintf("repo%d", i)), RepoDid: repo}); err != nil {
			t.Fatal(err)
		}
	}
	cfg := &config.Config{}
	cfg.Server.RepoDir = filepath.Join(t.TempDir(), "repos")
	return &Spindle{db: d, cfg: cfg, knots: knotclient.New(identity.NewMockDirectory(), http.DefaultClient, true)}
}

func regular(name string) defsFile { return defsFile{name, "0100644", defsWorkflow} }

// the origin is a dead path on purpose, like a clone whose remote is gone
func cloneWithCommit(t *testing.T, s *Spindle, repo syntax.DID) (src, commit string) {
	t.Helper()
	src = newSourceRepo(t, "sha1")
	writeTestFile(t, filepath.Join(src, ".tangled/workflows/local.yaml"), defsWorkflow)
	commitEverything(t, src)
	commit = gitInDir(t, src, "rev-parse", "HEAD")
	path := s.newRepoPath(repo)
	if err := gitutil.SparseSync(context.Background(), src, path, commit, sparseWorkflowDir); err != nil {
		t.Fatal(err)
	}
	gitInDir(t, path, "remote", "set-url", "origin", filepath.Join(t.TempDir(), "gone"))
	return src, commit
}

func definitionNames(t *testing.T, s *Spindle, repo syntax.DID, ref string) (names []string, commits []string) {
	t.Helper()
	defs, err := s.ListWorkflowDefinitions(context.Background(), repo, ref)
	if err != nil {
		t.Fatal(err)
	}
	for _, d := range defs {
		names = append(names, d.Name)
		commits = append(commits, d.Source.File.Commit)
	}
	return names, commits
}

func TestDefinitionMissReadsTheLocalCloneWithoutTheKnotOrAFetch(t *testing.T) {
	knot := newDefsKnot(t)
	repo := syntax.DID("did:plc:defsrepo")
	s := newDefsSpindle(t, knot, repo)
	_, commit := cloneWithCommit(t, s, repo)

	names, commits := definitionNames(t, s, repo, commit)
	if len(names) != 1 || names[0] != "local.yaml" || commits[0] != commit {
		t.Fatalf("definitions = %v at %v, want local.yaml at %s", names, commits, commit)
	}
	if got := knot.asked(); len(got) != 0 {
		t.Fatalf("a clone that had the commit still cost the knot %v", got)
	}
}

func TestDefinitionMissWithoutTheCommitLocallyReadsTheKnot(t *testing.T) {
	knot := newDefsKnot(t)
	repo := syntax.DID("did:plc:defsrepo")
	s := newDefsSpindle(t, knot, repo)
	cloneWithCommit(t, s, repo)
	other := knot.commit(t, repo, regular("remote.yaml"))

	names, commits := definitionNames(t, s, repo, other)
	if len(names) != 1 || names[0] != "remote.yaml" || commits[0] != other {
		t.Fatalf("definitions = %v at %v, want remote.yaml at %s", names, commits, other)
	}
}

func TestDefinitionMissWithoutACloneReadsTheKnotAndNeverRunsGit(t *testing.T) {
	knot := newDefsKnot(t)
	repo := syntax.DID("did:plc:defsrepo")
	s := newDefsSpindle(t, knot, repo)
	commit := knot.commit(t, repo, regular("ci.yaml"), regular("deploy.yaml"))
	t.Setenv("PATH", "")

	names, commits := definitionNames(t, s, repo, commit)
	if strings.Join(names, ",") != "ci.yaml,deploy.yaml" || commits[0] != commit {
		t.Fatalf("definitions = %v at %v", names, commits)
	}
	// the commit with its root, .tangled, then workflows with its blobs
	if knot.count("caps") != 1 || knot.count("ls-refs") != 0 || knot.count("fetch") != 3 {
		t.Fatalf("knot saw %v, want caps and three fetches without a ref lookup", knot.asked())
	}
	if _, err := os.Stat(s.newRepoPath(repo)); !os.IsNotExist(err) {
		t.Fatalf("a read made a clone dir: %v", err)
	}
}

func TestDefinitionKnotReadSkipsEverythingButRegularFiles(t *testing.T) {
	knot := newDefsKnot(t)
	repo := syntax.DID("did:plc:defsrepo")
	s := newDefsSpindle(t, knot, repo)
	commit := knot.commit(t, repo,
		regular("ci.yaml"),
		defsFile{"exec.yaml", "0100755", defsWorkflow},
		defsFile{"link.yaml", "0120000", "/etc/passwd"},
		defsFile{"sub", "0160000", ""},
		defsFile{"nested", "0040000", defsWorkflow},
	)
	t.Setenv("PATH", "")

	names, _ := definitionNames(t, s, repo, commit)
	if strings.Join(names, ",") != "ci.yaml,exec.yaml" {
		t.Fatalf("definitions = %v, want only the regular files", names)
	}
}

func TestDefinitionKnotReadRefusesMoreFilesThanSpindleRuns(t *testing.T) {
	knot := newDefsKnot(t)
	repo := syntax.DID("did:plc:defsrepo")
	s := newDefsSpindle(t, knot, repo)

	var files []defsFile
	for i := range maxWorkflowFiles + 1 {
		files = append(files, regular(fmt.Sprintf("w%03d.yaml", i)))
	}
	over := knot.commit(t, repo, files...)
	exact := knot.commit(t, repo, files[:maxWorkflowFiles]...)
	t.Setenv("PATH", "")
	if _, err := s.ListWorkflowDefinitions(context.Background(), repo, over); !errors.Is(err, errTooManyWorkflowFiles) {
		t.Fatalf("%d files read with err = %v, want errTooManyWorkflowFiles", len(files), err)
	}
	if names, _ := definitionNames(t, s, repo, exact); len(names) != maxWorkflowFiles {
		t.Fatalf("%d definitions, want %d", len(names), maxWorkflowFiles)
	}
}

func TestDefinitionKnotReadOfAMissingDirIsEmpty(t *testing.T) {
	knot := newDefsKnot(t)
	repo := syntax.DID("did:plc:defsrepo")
	s := newDefsSpindle(t, knot, repo)
	commit := knot.commit(t, repo)
	t.Setenv("PATH", "")

	names, _ := definitionNames(t, s, repo, commit)
	if len(names) != 0 {
		t.Fatalf("definitions = %v, want none", names)
	}
}

func TestDefinitionKnotReadOfAGoneRepoFails(t *testing.T) {
	knot := newDefsKnot(t)
	repo := syntax.DID("did:plc:defsrepo")
	s := newDefsSpindle(t, knot, repo)
	t.Setenv("PATH", "")

	// a 404 from the knot is no proof the dir is missing
	if _, err := s.ListWorkflowDefinitions(context.Background(), repo, strings.Repeat("a", 40)); err == nil {
		t.Fatal("a repo the knot doesn't have read as having no workflows")
	}
}

func TestDefinitionBranchRefsResolveToACommitBeforeReading(t *testing.T) {
	for _, ref := range []string{"HEAD", "", "main", "refs/heads/main"} {
		t.Run(ref, func(t *testing.T) {
			knot := newDefsKnot(t)
			repo := syntax.DID("did:plc:defsrepo")
			s := newDefsSpindle(t, knot, repo)
			commit := knot.commit(t, repo, regular("ci.yaml"))
			knot.setBranch(t, repo, "main", commit)
			knot.setBranch(t, repo, "other", knot.commit(t, repo, regular("other.yaml")))
			t.Setenv("PATH", "")

			names, commits := definitionNames(t, s, repo, ref)
			if len(names) != 1 || commits[0] != commit {
				t.Fatalf("definitions = %v at %v, want ci.yaml at %s", names, commits, commit)
			}
			asked := knot.asked()
			if i := slices.IndexFunc(asked, func(r string) bool { return strings.HasPrefix(r, "fetch ") }); i < 0 || asked[i] != "fetch "+commit {
				t.Fatalf("knot saw %v, want the first fetch to be the resolved commit", asked)
			}
			if knot.count("caps") != 1 {
				t.Fatalf("knot saw %v, want its caps asked once per read", knot.asked())
			}
		})
	}
}

func TestDefinitionBranchRefReadsLocallyAfterResolving(t *testing.T) {
	knot := newDefsKnot(t)
	repo := syntax.DID("did:plc:defsrepo")
	s := newDefsSpindle(t, knot, repo)
	src, commit := cloneWithCommit(t, s, repo)
	gitInDir(t, src, "push", "-q", knot.bare(t, repo), "HEAD:refs/heads/main")

	_, commits := definitionNames(t, s, repo, "main")
	if commits[0] != commit {
		t.Fatalf("commit = %s, want %s", commits[0], commit)
	}
	if knot.count("fetch") != 0 {
		t.Fatalf("knot saw %v, want only the branch lookup", knot.asked())
	}
}

func TestDefinitionTagRefsResolveToTheirCommitWithoutAClone(t *testing.T) {
	for _, ref := range []string{"v1.0", "refs/tags/v1.0"} {
		t.Run(ref, func(t *testing.T) {
			knot := newDefsKnot(t)
			repo := syntax.DID("did:plc:defsrepo")
			s := newDefsSpindle(t, knot, repo)
			commit := knot.commit(t, repo, regular("ci.yaml"))
			knot.setTag(t, repo, "v1.0", commit)
			t.Setenv("PATH", "")

			// annotated, so the read has to land on the peeled commit
			names, commits := definitionNames(t, s, repo, ref)
			if len(names) != 1 || commits[0] != commit {
				t.Fatalf("definitions = %v at %v, want ci.yaml at %s", names, commits, commit)
			}
		})
	}
}

func TestDefinitionBareRefIsABranchBeforeATag(t *testing.T) {
	knot := newDefsKnot(t)
	repo := syntax.DID("did:plc:defsrepo")
	s := newDefsSpindle(t, knot, repo)
	branchCommit := knot.commit(t, repo, regular("ci.yaml"))
	tagCommit := knot.commit(t, repo, regular("old.yaml"))
	knot.setBranch(t, repo, "release", branchCommit)
	knot.setTag(t, repo, "release", tagCommit)
	t.Setenv("PATH", "")

	if names, _ := definitionNames(t, s, repo, "release"); len(names) != 1 || names[0] != "ci.yaml" {
		t.Fatalf("definitions = %v, want the branch's", names)
	}
	if got := knot.count("ls-refs"); got != 1 {
		t.Fatalf("%d ref lookups, want one for both branch and tag", got)
	}
}

func TestDefinitionRefThatIsNeitherBranchNorTagFails(t *testing.T) {
	knot := newDefsKnot(t)
	repo := syntax.DID("did:plc:defsrepo")
	s := newDefsSpindle(t, knot, repo)
	knot.setBranch(t, repo, "main", knot.commit(t, repo, regular("ci.yaml")))
	t.Setenv("PATH", "")
	if _, err := s.ListWorkflowDefinitions(context.Background(), repo, "ghost"); err == nil {
		t.Fatal("an unknown ref resolved")
	}
}