//go:build torture

package spindle

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"math"
	"net/http"
	"net/http/cgi"
	"net/http/httptest"
	"os"
	"os/exec"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/bluesky-social/indigo/atproto/atcrypto"
	"github.com/bluesky-social/indigo/atproto/auth"
	"github.com/bluesky-social/indigo/atproto/identity"
	"github.com/bluesky-social/indigo/atproto/syntax"
	"github.com/sethvargo/go-envconfig"

	"tangled.org/core/api/org_tangled"
	"tangled.org/core/idresolver"
	"tangled.org/core/spindle/config"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/knotclient"
	"tangled.org/core/spindle/models"
	"tangled.org/core/spindle/quota"
	"tangled.org/core/tapc"
	"tangled.org/core/xrpc/serviceauth"
)

// one directory for every did, knots too, since a knot is did:web on loopback
type tortureDirectory struct {
	*identity.MockDirectory
	delay   atomic.Int64
	lookups atomic.Int64
}

func newTortureDirectory() *tortureDirectory {
	return &tortureDirectory{MockDirectory: identity.NewMockDirectory()}
}

func (d *tortureDirectory) LookupDID(ctx context.Context, did syntax.DID) (*identity.Identity, error) {
	d.lookups.Add(1)
	if err := sleepCtx(ctx, time.Duration(d.delay.Load())); err != nil {
		return nil, err
	}
	return d.MockDirectory.LookupDID(ctx, did)
}

func (d *tortureDirectory) Lookup(ctx context.Context, atid syntax.AtIdentifier) (*identity.Identity, error) {
	if did, err := atid.AsDID(); err == nil {
		return d.LookupDID(ctx, did)
	}
	return d.MockDirectory.Lookup(ctx, atid)
}

func sleepCtx(ctx context.Context, d time.Duration) error {
	if d <= 0 {
		return nil
	}
	t := time.NewTimer(d)
	defer t.Stop()
	select {
	case <-t.C:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

type tortureKnot struct {
	*fakeKnot
	did  syntax.DID
	priv atcrypto.PrivateKey
	git  *cgi.Handler
	root string

	delay    atomic.Int64
	gitDelay atomic.Int64
	down     atomic.Bool
	xrpc     atomic.Int64
	fetches  atomic.Int64
	inflight atomic.Int64

	readsMu sync.Mutex
	reads   map[string]int64
}

func (k *tortureKnot) readsOf(collection string) int64 {
	k.readsMu.Lock()
	defer k.readsMu.Unlock()
	return k.reads[collection]
}

func (k *tortureKnot) url() string { return k.srv.URL }

func (k *tortureKnot) middleware(records http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		k.inflight.Add(1)
		defer k.inflight.Add(-1)
		if strings.HasPrefix(r.URL.Path, "/did:") {
			k.serveGit(w, r)
			return
		}
		k.xrpc.Add(1)
		q := r.URL.Query()
		if collection := q.Get("collection"); collection != "" {
			k.readsMu.Lock()
			k.reads[collection]++
			k.readsMu.Unlock()
		}
		if sleepCtx(r.Context(), time.Duration(k.delay.Load())) != nil {
			return
		}
		if k.down.Load() {
			http.Error(w, "boom", http.StatusInternalServerError)
			return
		}
		records.ServeHTTP(w, r)
	})
}

func (k *tortureKnot) serveGit(w http.ResponseWriter, r *http.Request) {
	if strings.HasSuffix(r.URL.Path, "/info/refs") {
		k.fetches.Add(1)
	}
	if sleepCtx(r.Context(), time.Duration(k.gitDelay.Load())) != nil {
		return
	}
	k.git.ServeHTTP(w, r)
}

type tortureRepo struct {
	did   syntax.DID
	owner syntax.DID
	name  string
	knot  *tortureKnot
	shas  []string
}

type tortureWorld struct {
	r   *tortureRun
	dir string
	js  *fakeJetstream
	hyd *fakeHydrant
	ids *tortureDirectory

	knots []*tortureKnot
	eng   *tortureEngine

	pdsMu sync.Mutex
	pds   map[syntax.DID]*tortureKnot

	// read at boot, set these before the world starts
	maxJobs               int
	admission             config.Admission
	qm                    *quota.Manager
	skew                  atomic.Int64
	repoPool, notifySlots int
	priorityKnots         []string

	s      *Spindle
	sp     atomic.Pointer[Spindle]
	api    *httptest.Server
	cancel context.CancelFunc
	done   chan error

	nextRepo atomic.Int64
}

func newTortureWorld(r *tortureRun) *tortureWorld {
	t := r.t
	t.Setenv("GIT_CONFIG_GLOBAL", os.DevNull)
	t.Setenv("GIT_CONFIG_SYSTEM", os.DevNull)
	t.Setenv("GIT_AUTHOR_NAME", "torture")
	t.Setenv("GIT_AUTHOR_EMAIL", "torture@tangled.org")
	t.Setenv("GIT_COMMITTER_NAME", "torture")
	t.Setenv("GIT_COMMITTER_EMAIL", "torture@tangled.org")
	gitExecPath(t)

	w := &tortureWorld{
		r:         r,
		dir:       t.TempDir(),
		ids:       newTortureDirectory(),
		pds:       map[syntax.DID]*tortureKnot{},
		eng:       newTortureEngine(),
		maxJobs:   8,
		admission: config.AdmissionOpen,
	}
	w.js = newFakeJetstream(t)
	w.hyd = &fakeHydrant{}
	w.hyd.onTrack = func(did syntax.DID) {
		if w.hyd.repoStatus(did) != synced(did).Status {
			w.backfill(did)
		}
	}
	w.hyd.onResync = func(did syntax.DID) {
		if w.hyd.repoStatus(did) != "" {
			w.backfill(did)
		}
	}
	w.ids.Insert(identity.Identity{DID: testOwner})

	r.gauge("queued_jobs", func() float64 { return w.rows("jobs") })
	r.gauge("pipelines", func() float64 { return w.rows("pipelines") })
	r.gauge("repo_retries", func() float64 { return w.rows("repo_retries") })
	r.gauge("db_mb", func() float64 {
		var total int64
		for _, suffix := range []string{"", "-wal", "-shm"} {
			if fi, err := os.Stat(w.dbPath() + suffix); err == nil {
				total += fi.Size()
			}
		}
		return float64(total) / (1 << 20)
	})
	r.gauge("engine_running", func() float64 { return float64(w.eng.running.Load()) })
	return w
}

func (w *tortureWorld) dbPath() string { return filepath.Join(w.dir, "spindle.db") }

func (w *tortureWorld) spindleDid() string {
	return config.Server{Hostname: faultHostname}.Did().String()
}

// nil once stopped
func (w *tortureWorld) spindle() *Spindle { return w.sp.Load() }

// NaN once stopped, since a `<=` ceiling check fails closed on it
func (w *tortureWorld) fromSpindle(read func(s *Spindle) float64) float64 {
	s := w.spindle()
	if s == nil {
		return math.NaN()
	}
	return read(s)
}

func (w *tortureWorld) countRows(query string) float64 {
	return w.fromSpindle(func(s *Spindle) float64 {
		var n float64
		if err := s.db.QueryRow(query).Scan(&n); err != nil {
			return math.NaN()
		}
		return n
	})
}

func (w *tortureWorld) newKeyedIdentity(did syntax.DID) (atcrypto.PrivateKey, identity.Identity) {
	priv, err := atcrypto.GeneratePrivateKeyP256()
	if err != nil {
		w.r.t.Fatal(err)
	}
	pub, err := priv.PublicKey()
	if err != nil {
		w.r.t.Fatal(err)
	}
	return priv, identity.Identity{
		DID:  did,
		Keys: map[string]identity.VerificationMethod{"atproto": {Type: "Multikey", PublicKeyMultibase: pub.Multibase()}},
	}
}

func (w *tortureWorld) probeTable() {
	if _, err := w.spindle().db.Exec(`create table if not exists torture_probe (n integer)`); err != nil {
		w.r.t.Fatal(err)
	}
}

// inserts on a timer, because a starved writer shows up as latency
func (w *tortureWorld) probeWrites(every time.Duration, lat *latencies, errs *atomic.Int64) (stop func()) {
	done := make(chan struct{})
	var wg sync.WaitGroup
	wg.Go(func() {
		tick := time.NewTicker(every)
		defer tick.Stop()
		for i := 0; ; i++ {
			select {
			case <-done:
				return
			case <-tick.C:
			}
			start := time.Now()
			if _, err := w.spindle().db.Exec(`insert into torture_probe (n) values (?)`, i); err != nil {
				errs.Add(1)
				continue
			}
			lat.since(start)
		}
	})
	return func() {
		close(done)
		wg.Wait()
	}
}

func (w *tortureWorld) addKnots(n int) []*tortureKnot {
	ks := make([]*tortureKnot, n)
	for i := range ks {
		ks[i] = w.addKnot()
	}
	return ks
}

func (w *tortureWorld) rows(table string) float64 {
	return w.countRows("select count(*) from " + table)
}

func (w *tortureWorld) addKnot() *tortureKnot {
	t := w.r.t
	k := &tortureKnot{reads: map[string]int64{}}
	k.fakeKnot = newFakeKnotBehind(t, k.middleware)
	k.root = filepath.Join(w.dir, fmt.Sprintf("knot%d", len(w.knots)))
	if err := os.MkdirAll(k.root, 0o755); err != nil {
		t.Fatal(err)
	}
	k.git = &cgi.Handler{
		Path:       gitExecPath(t),
		Root:       "/",
		Env:        []string{"GIT_PROJECT_ROOT=" + k.root, "GIT_HTTP_EXPORT_ALL=1"},
		InheritEnv: []string{"PATH", "HOME", "GIT_CONFIG_GLOBAL", "GIT_CONFIG_SYSTEM"},
	}
	k.did = serviceauth.DidWeb(k.host())
	var id identity.Identity
	k.priv, id = w.newKeyedIdentity(k.did)
	w.ids.Insert(id)
	w.knots = append(w.knots, k)
	return k
}

func tortureWorkflow(steps ...string) string {
	var b strings.Builder
	b.WriteString("engine: torture\nwhen:\n  - event: push\n    branch: main\nsteps:\n")
	for i, step := range steps {
		fmt.Fprintf(&b, "  - name: s%d\n    command: %q\n", i, step)
	}
	return b.String()
}

// no commits skips git, the repo is only records for membership load
func (w *tortureWorld) addRepo(k *tortureKnot, owner syntax.DID, commits int, workflows map[string]string) *tortureRepo {
	repo, err := w.tryAddRepo(k, owner, commits, workflows)
	if err != nil {
		w.r.t.Fatal(err)
	}
	return repo
}

// no t.Fatal here, addRepos calls this off the test goroutine
func (w *tortureWorld) tryAddRepo(k *tortureKnot, owner syntax.DID, commits int, workflows map[string]string) (*tortureRepo, error) {
	n := w.nextRepo.Add(1)
	repo := &tortureRepo{did: syntax.DID(fmt.Sprintf("did:plc:torture%06d", n)), owner: owner, name: fmt.Sprintf("repo%d", n), knot: k}
	if commits > 0 {
		if err := w.addHistory(repo, commits, workflows); err != nil {
			return nil, err
		}
	}
	w.ids.Insert(identity.Identity{DID: repo.did, Services: map[string]identity.ServiceEndpoint{
		"tangled_knot": {Type: "TangledKnot", URL: k.url()},
	}})
	k.put(repo.did, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, manifestRecord(owner, repo.name, w.spindleDid()))
	k.put(repo.did, org_tangled.GitRefsNSID, knotclient.RefsRkey, org_tangled.GitRefs{LexiconTypeID: org_tangled.GitRefsNSID, DefaultBranch: "main"})
	w.pdsOf(owner, k).declare(owner, repo)
	return repo, nil
}

func (w *tortureWorld) pdsOf(owner syntax.DID, k *tortureKnot) *tortureKnot {
	w.pdsMu.Lock()
	defer w.pdsMu.Unlock()
	if pds, ok := w.pds[owner]; ok {
		return pds
	}
	w.pds[owner] = k
	w.ids.Insert(identity.Identity{DID: owner, Services: map[string]identity.ServiceEndpoint{
		"atproto_pds": {Type: "AtprotoPersonalDataServer", URL: k.url()},
	}})
	return k
}

func (w *tortureWorld) addHistory(repo *tortureRepo, commits int, workflows map[string]string) error {
	k := repo.knot
	bare := filepath.Join(k.root, repo.did.String())
	gitCmd := func(stdin io.Reader, args ...string) error {
		cmd := exec.Command("git", append([]string{"-C", bare}, args...)...)
		cmd.Stdin = stdin
		if out, err := cmd.CombinedOutput(); err != nil {
			return fmt.Errorf("git %v: %w\n%s", args, err, out)
		}
		return nil
	}
	if err := os.MkdirAll(bare, 0o755); err != nil {
		return err
	}
	for _, args := range [][]string{
		{"init", "-q", "--bare", "-b", "main"},
		{"config", "uploadpack.allowFilter", "true"},
		{"config", "uploadpack.allowAnySHA1InWant", "true"},
	} {
		if err := gitCmd(nil, args...); err != nil {
			return err
		}
	}

	var stream bytes.Buffer
	names := make([]string, 0, len(workflows))
	for name := range workflows {
		names = append(names, name)
	}
	sort.Strings(names)
	blob := func(path, contents string) {
		fmt.Fprintf(&stream, "M 644 inline %s\ndata %d\n%s\n", path, len(contents), contents)
	}
	for i := 1; i <= commits; i++ {
		fmt.Fprintf(&stream, "commit refs/heads/c%d\nmark :%d\ncommitter t <t@t> %d +0000\ndata 2\nc\n", i, i, 1700000000+i)
		if i > 1 {
			fmt.Fprintf(&stream, "from :%d\n", i-1)
			blob("n", repo.did.String()+" "+strconv.Itoa(i))
		} else {
			blob("n", repo.did.String()+" 1")
			for _, name := range names {
				blob(".tangled/workflows/"+name, workflows[name])
			}
		}
	}
	marks := filepath.Join(bare, "torture-marks")
	if err := gitCmd(&stream, "fast-import", "--quiet", "--export-marks="+marks); err != nil {
		return err
	}
	raw, err := os.ReadFile(marks)
	if err != nil {
		return err
	}
	repo.shas = make([]string, commits)
	for _, line := range strings.Split(strings.TrimSpace(string(raw)), "\n") {
		mark, sha, _ := strings.Cut(line, " ")
		i, _ := strconv.Atoi(strings.TrimPrefix(mark, ":"))
		repo.shas[i-1] = sha
	}
	return gitCmd(nil, "update-ref", "refs/heads/main", repo.shas[commits-1])
}

func (k *tortureKnot) declare(owner syntax.DID, repo *tortureRepo) {
	k.put(owner, org_tangled.RepoDeclarationNSID, strings.ToLower(repo.name), org_tangled.RepoDeclaration{LexiconTypeID: org_tangled.RepoDeclarationNSID, Repo: repo.did.String()})
}

// repos go in parallel because git is most of the setup time
func (w *tortureWorld) addRepos(n int, pick func(i int) (*tortureKnot, syntax.DID), commits int, workflows map[string]string) []*tortureRepo {
	repos := make([]*tortureRepo, n)
	var wg sync.WaitGroup
	var firstErr atomic.Pointer[error]
	next := atomic.Int64{}
	for range 8 {
		wg.Go(func() {
			for {
				i := int(next.Add(1)) - 1
				if i >= n {
					return
				}
				k, owner := pick(i)
				repo, err := w.tryAddRepo(k, owner, commits, workflows)
				if err != nil {
					firstErr.CompareAndSwap(nil, &err)
					return
				}
				repos[i] = repo
			}
		})
	}
	wg.Wait()
	if err := firstErr.Load(); err != nil {
		w.r.t.Fatal(*err)
	}
	return repos
}

func (w *tortureWorld) emitManifest(repo *tortureRepo) int64 {
	return w.js.manifest(repo.did, repo.knot.record(repo.did, org_tangled.RepoManifestNSID, knotclient.ManifestRkey))
}

func (w *tortureWorld) emitDeclaration(repo *tortureRepo) int64 {
	return w.js.declaration(repo.owner, repo.name, repo.did)
}

func (w *tortureWorld) emitCollaborator(repo *tortureRepo, who syntax.DID, op string) int64 {
	return w.js.collaborator(repo.did, who, op)
}

// production learns repos from jetstream and seed does too
func (w *tortureWorld) seed(repos []*tortureRepo, within time.Duration) {
	start := time.Now()
	var last int64
	for _, repo := range repos {
		w.emitDeclaration(repo)
		last = w.emitManifest(repo)
	}
	if !w.r.waitFor(within, func() bool { return w.servedCount(repos) == len(repos) && w.s.jc.LastSeenUs() > last }) {
		w.r.t.Fatalf("only %d of %d repos served after the jetstream seed", w.servedCount(repos), len(repos))
	}
	w.r.metric("seed_s", time.Since(start).Seconds())
}

func (w *tortureWorld) servedCount(repos []*tortureRepo) int {
	n := 0
	for _, repo := range repos {
		if w.served(repo) {
			n++
		}
	}
	return n
}

func (w *tortureWorld) served(repo *tortureRepo) bool {
	s := w.spindle()
	if s == nil {
		return false
	}
	rows, err := s.db.ReposByDid(repo.did)
	return err == nil && len(rows) > 0
}

func (w *tortureWorld) consumed(us int64, within time.Duration) bool {
	return w.r.waitFor(within, func() bool {
		s := w.spindle()
		return s != nil && s.jc.LastSeenUs() > us
	})
}

func (w *tortureWorld) backfill(did syntax.DID) {
	var events []tapc.RecordEventData
	for _, k := range w.knots {
		events = append(events, k.recordsOf(did)...)
	}
	w.hyd.backfill(did, events)
}

func (w *tortureWorld) boot() {
	t := w.r.t
	ctx := context.Background()
	d, err := db.Make(ctx, w.dbPath())
	if err != nil {
		t.Fatal(err)
	}
	cfg := &config.Config{}
	if err := envconfig.ProcessWith(ctx, &envconfig.Config{
		Target:   cfg,
		Lookuper: envconfig.MapLookuper(map[string]string{"SPINDLE_SERVER_HOSTNAME": faultHostname}),
	}); err != nil {
		t.Fatal(err)
	}
	cfg.Server.ListenAddr = "127.0.0.1:0"
	cfg.Server.MetricsListenAddr = "127.0.0.1:0"
	cfg.Server.DBPath = w.dbPath()
	cfg.Server.RepoDir = filepath.Join(w.dir, "repos")
	cfg.Server.LogDir = filepath.Join(w.dir, "logs")
	cfg.Server.JetstreamEndpoint = w.js.url()
	cfg.Server.PlcUrl = "http://127.0.0.1:1"
	cfg.Server.Hydrant = config.Hydrant{Dir: shortTempDir(t)}
	cfg.Server.Dev = true
	cfg.Server.Owner = testOwner.String()
	cfg.Server.Admission = w.admission
	cfg.Server.MaxJobCount = w.maxJobs
	if w.repoPool > 0 {
		cfg.Server.RepoJobWorkers = w.repoPool
	}
	if w.notifySlots > 0 {
		cfg.Server.NotifySlots = w.notifySlots
	}
	if len(w.priorityKnots) > 0 {
		cfg.Server.PriorityKnots = w.priorityKnots
	}
	cfg.ArtifactStores.Disk.Dir = filepath.Join(w.dir, "artifacts")
	for _, dir := range []string{cfg.Server.RepoDir, cfg.Server.LogDir, cfg.ArtifactStores.Disk.Dir} {
		if err := os.MkdirAll(dir, 0o755); err != nil {
			t.Fatal(err)
		}
	}

	connects := w.js.connects()
	s, err := New(ctx, cfg, d, map[string]models.Engine{"torture": w.eng}, w.qm)
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	dir := identity.NewCacheDirectory(w.ids, 250_000, 24*time.Hour, 2*time.Minute, 5*time.Minute)
	s.res = idresolver.NewMockResolver(dir)
	s.knots = knotclient.New(dir, knotclient.HTTPClient(true), true)
	s.backfill.start = w.hyd.start
	s.retries.now = func() time.Time { return time.Now().Add(time.Duration(w.skew.Load())) }
	w.s = s
	w.sp.Store(s)
	w.api = httptest.NewServer(s.Router())

	runCtx, cancel := context.WithCancel(ctx)
	w.cancel = cancel
	w.done = make(chan error, 1)
	go func() { w.done <- s.Start(runCtx) }()
	t.Cleanup(w.stop)
	if !w.r.waitFor(10*time.Second, func() bool { return w.js.connects() > connects }) {
		t.Fatal("jetstream never connected")
	}
}

// the db stays on disk for the next boot to reopen
func (w *tortureWorld) stop() {
	if w.cancel == nil {
		return
	}
	w.api.Close()
	w.cancel()
	w.cancel = nil
	select {
	case err := <-w.done:
		if err != nil {
			w.r.t.Errorf("Start: %v", err)
		}
	case <-time.After(60 * time.Second):
		w.r.check("clean shutdown", false, "spindle did not stop within 60s")
	}
	w.sp.Store(nil)
	w.s.db.Close()
	w.s = nil
}

type notifyReply struct {
	status  int
	body    string
	elapsed time.Duration
}

func (w *tortureWorld) notifyBody(repo *tortureRepo, ref, oldSha, newSha string) []byte {
	actor := repo.owner.String()
	body, err := json.Marshal(org_tangled.EventNotifyEvent_Input{
		Repo: repo.did.String(),
		Event: &org_tangled.EventNotifyEvent_Input_Event{EventPush: &org_tangled.EventPush{
			Repo: repo.did.String(), Ref: ref, OldSha: oldSha, NewSha: newSha, Actor: &actor,
		}},
	})
	if err != nil {
		panic(err)
	}
	return body
}

func (w *tortureWorld) sign(did syntax.DID, priv atcrypto.PrivateKey, nsid string) string {
	lxm := syntax.NSID(nsid)
	token, err := auth.SignServiceAuth(did, w.spindleDid(), time.Minute, &lxm, priv)
	if err != nil {
		panic(err)
	}
	return token
}

func (w *tortureWorld) token(k *tortureKnot) string {
	return w.sign(k.did, k.priv, org_tangled.EventNotifyEventNSID)
}

func (w *tortureWorld) call(ctx context.Context, client *http.Client, method, nsid, query, token string, body []byte) notifyReply {
	start := time.Now()
	target := w.api.URL + "/xrpc/" + nsid
	if query != "" {
		target += "?" + query
	}
	req, err := http.NewRequestWithContext(ctx, method, target, bytes.NewReader(body))
	if err != nil {
		panic(err)
	}
	req.Header.Set("Content-Type", "application/json")
	if token != "" {
		req.Header.Set("Authorization", "Bearer "+token)
	}
	resp, err := client.Do(req)
	if err != nil {
		return notifyReply{status: -1, body: err.Error(), elapsed: time.Since(start)}
	}
	defer resp.Body.Close()
	raw, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<20))
	return notifyReply{status: resp.StatusCode, body: string(raw), elapsed: time.Since(start)}
}

func (w *tortureWorld) post(ctx context.Context, client *http.Client, token string, body []byte) notifyReply {
	return w.call(ctx, client, http.MethodPost, org_tangled.EventNotifyEventNSID, "", token, body)
}

func statusKind(status int) string {
	if status < 0 {
		return "transport"
	}
	return "http_" + strconv.Itoa(status)
}