//go:build torture

package spindle

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"slices"
	"strings"
	"sync"
	"sync/atomic"
	"time"

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

	"tangled.org/core/api/org_tangled"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/quota"
	"tangled.org/core/sqlite"
)

type admUser struct {
	did  syntax.DID
	priv atcrypto.PrivateKey
}

// has to run after addRepo, which overwrites the owner's entry
func (w *tortureWorld) admAddUser(did syntax.DID, pds *tortureKnot) *admUser {
	priv, id := w.newKeyedIdentity(did)
	if pds != nil {
		id.Services = map[string]identity.ServiceEndpoint{
			"atproto_pds": {Type: "AtprotoPersonalDataServer", URL: pds.url()},
		}
	}
	w.ids.Insert(id)
	return &admUser{did: did, priv: priv}
}

func (w *tortureWorld) admToken(u *admUser, nsid string) string { return w.sign(u.did, u.priv, nsid) }

// same setup as Start, which keeps the timeouts real
func (w *tortureWorld) admServe() string {
	srv := newHTTPServer("", w.s.Router())
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		w.r.t.Fatal(err)
	}
	go srv.Serve(ln)
	w.r.t.Cleanup(func() { srv.Close() })
	return ln.Addr().String()
}

func admFdKinds() string {
	targets, err := fdTargets()
	if err != nil {
		return "unreadable"
	}
	kinds := map[string]int{}
	for _, target := range targets {
		if target == "" {
			continue
		}
		kind, _, found := strings.Cut(target, ":")
		if !found {
			kind = "file"
		}
		kinds[kind]++
	}
	return fmt.Sprint(kinds)
}

// built before the db exists, the real store gets swapped in after boot
type admQuotaStore struct {
	qs       atomic.Pointer[db.QuotaStore]
	reserves atomic.Int64
}

func (s *admQuotaStore) Reserve(ctx context.Context, req quota.ReserveRequest) (quota.Reservation, error) {
	s.reserves.Add(1)
	return s.qs.Load().Reserve(ctx, req)
}

func (s *admQuotaStore) CheckDenied(ctx context.Context, req quota.ReserveRequest) (quota.Reservation, bool, error) {
	return s.qs.Load().CheckDenied(ctx, req)
}

func (s *admQuotaStore) BeginCommit(ctx context.Context, id string) error {
	return s.qs.Load().BeginCommit(ctx, id)
}

func (s *admQuotaStore) Commit(ctx context.Context, id string) error {
	return s.qs.Load().Commit(ctx, id)
}

func (s *admQuotaStore) Release(ctx context.Context, id string) error {
	return s.qs.Load().Release(ctx, id)
}

type admQuotaObserver struct {
	depth atomic.Int64
}

func (o *admQuotaObserver) RecordDecision(kind, resource string, allowed, temporary bool, reason string) {
}

func (o *admQuotaObserver) SetWaitDepth(kind string, depth int64) {
	if kind == string(quota.KindWorkflow) {
		o.depth.Store(depth)
	}
}

func (o *admQuotaObserver) RecordWait(kind, resource string, duration time.Duration) {}

func init() {
	// twice the read timeout, which means only the server cutting them off ends it
	registerTorture("http-slowloris", "connections that never finish their headers or body against a real listener", func(r *tortureRun) {
		conns := r.load("conns", 600)
		hold := r.duration("hold", 2*newHTTPServer("", nil).ReadTimeout)
		drip := time.Second

		w := newTortureWorld(r)
		w.maxJobs = 2
		user := w.admAddUser("did:plc:slowuser", nil)
		w.boot()
		addr := w.admServe()
		// warm up files and pools so the baseline doesn't pay for them
		for range 20 {
			if resp, err := http.Get("http://" + addr + "/_health"); err == nil {
				io.Copy(io.Discard, resp.Body)
				resp.Body.Close()
			}
		}
		http.DefaultClient.CloseIdleConnections()
		time.Sleep(2 * time.Second)
		baseGoroutines := r.gaugeValue("goroutines")
		baseFds := r.gaugeValue("fds")
		token := w.admToken(user, org_tangled.SecretAddSecretNSID)

		var headerOpen, bodyOpen, headerCut, bodyCut atomic.Int64
		stop := make(chan struct{})
		var attackers sync.WaitGroup
		for i := range conns {
			attackers.Go(func() {
				c, err := net.DialTimeout("tcp", addr, 5*time.Second)
				if err != nil {
					r.count("dial", 1)
					return
				}
				defer c.Close()
				body := i%2 == 1
				opening := "GET /_health HTTP/1.1\r\nHost: spindle\r\n"
				next := "X-a: b\r\n"
				if body {
					bodyOpen.Add(1)
					opening = fmt.Sprintf("POST /xrpc/%s HTTP/1.1\r\nHost: spindle\r\nAuthorization: Bearer %s\r\nContent-Type: application/json\r\nContent-Length: 1000000\r\n\r\n{\"repo\":\"", org_tangled.SecretAddSecretNSID, token)
					next = "a"
				} else {
					headerOpen.Add(1)
				}
				if _, err := c.Write([]byte(opening)); err != nil {
					return
				}
				tick := time.NewTicker(drip)
				defer tick.Stop()
				buf := make([]byte, 1)
				for {
					select {
					case <-stop:
						return
					case <-tick.C:
					}
					c.SetWriteDeadline(time.Now().Add(time.Second))
					if _, err := c.Write([]byte(next)); err == nil {
						c.SetReadDeadline(time.Now().Add(time.Millisecond))
						n, rerr := c.Read(buf)
						var ne net.Error
						if n == 0 && errors.As(rerr, &ne) && ne.Timeout() {
							continue
						}
					}
					if body {
						bodyCut.Add(1)
					} else {
						headerCut.Add(1)
					}
					return
				}
			})
		}

		var legit latencies
		var legitFailed atomic.Int64
		probing := make(chan struct{})
		go func() {
			defer close(probing)
			tick := time.NewTicker(250 * time.Millisecond)
			defer tick.Stop()
			for {
				select {
				case <-stop:
					return
				case <-tick.C:
				}
				client := &http.Client{Timeout: 3 * time.Second, Transport: &http.Transport{DisableKeepAlives: true}}
				start := time.Now()
				resp, err := client.Get("http://" + addr + "/_health")
				if err != nil {
					legitFailed.Add(1)
					continue
				}
				io.Copy(io.Discard, resp.Body)
				resp.Body.Close()
				if resp.StatusCode != http.StatusOK {
					legitFailed.Add(1)
					continue
				}
				legit.since(start)
			}
		}()

		time.Sleep(hold)
		r.metric("fds_during", r.gaugeValue("fds"))
		r.metric("goroutines_during", r.gaugeValue("goroutines"))
		close(stop)
		attackers.Wait()
		<-probing

		r.metric("header_conns", float64(headerOpen.Load()))
		r.metric("body_conns", float64(bodyOpen.Load()))
		r.metric("header_cut_by_server", float64(headerCut.Load()))
		r.metric("body_cut_by_server", float64(bodyCut.Load()))
		r.latency("legit_health", &legit)
		r.metric("legit_failed", float64(legitFailed.Load()))

		r.check("conns that never finish their headers get dropped (A1)", headerCut.Load() == headerOpen.Load(), "%d of %d header-dripping conns cut by the server within %v, want all", headerCut.Load(), headerOpen.Load(), hold)
		r.check("conns that never finish their body get dropped (A1)", bodyCut.Load() == bodyOpen.Load(), "%d of %d body-dripping conns cut by the server within %v, want all", bodyCut.Load(), bodyOpen.Load(), hold)
		r.check("legit requests keep getting through", legitFailed.Load() == 0, "%d health probes failed or timed out, want 0", legitFailed.Load())
		r.note("fds after the attackers left: %s, baseline %.0f", admFdKinds(), baseFds)
		r.settles("goroutines", baseGoroutines+50, 30*time.Second)
		r.settles("fds", baseFds+30, 30*time.Second)
	})

	// the 50ms quota tick writes for every waiting owner, that's the cost here
	registerTorture("quota-churn", "many owners at a tight workflow limit, what the quota manager costs while they wait", func(r *tortureRun) {
		owners := r.load("owners", 40)
		queued := int(r.param("queued", 2))
		limit := int64(r.param("limit", 1))
		hold := r.duration("hold", 20*time.Second)
		window := 4 * time.Second

		obs := &admQuotaObserver{}
		store := &admQuotaStore{}
		qm := quota.NewManager(store, 50*time.Millisecond, obs)
		r.t.Cleanup(qm.Close)

		w := newTortureWorld(r)
		w.qm = qm
		w.maxJobs = owners*queued + 8
		r.gauge("quota_waiters", func() float64 { return float64(obs.depth.Load()) })
		r.gauge("quota_reservations", func() float64 { return w.rows("quota_reservations") })

		flow := map[string]string{"ci.yaml": tortureWorkflow("sleep " + hold.String())}
		knot := w.addKnot()
		repos := w.addRepos(owners, func(i int) (*tortureKnot, syntax.DID) { return knot, ownerFor(i, 1) }, 2*queued, flow)
		fresh := w.addRepos(1, func(int) (*tortureKnot, syntax.DID) { return knot, "did:plc:fresh" }, 1, flow)
		w.boot()
		w.seed(append(slices.Clone(repos), fresh...), 3*time.Minute)
		defaults := quota.Defaults{quota.ScopeUser: quota.Resources{quota.ResourceWorkflows: limit}}
		store.qs.Store(db.NewQuotaStore(w.s.db, defaults))
		ctx := context.Background()

		w.probeTable()
		var quiet latencies
		for i := range 50 {
			start := time.Now()
			if _, err := w.s.db.Exec(`insert into torture_probe (n) values (?)`, -i); err != nil {
				r.t.Fatal(err)
			}
			quiet.since(start)
		}
		var probe latencies
		var probeErrors atomic.Int64
		stopProbe := w.probeWrites(100*time.Millisecond, &probe, &probeErrors)

		ds := pushRanges(repos, 0, queued)
		var st deliveryStats
		w.flood(ctx, ds, 32, 20, &st)
		st.report(r, "notify_")

		want := int64(owners) * (int64(queued) - limit)
		formed := r.waitFor(60*time.Second, func() bool { return obs.depth.Load() >= want*9/10 })
		r.check("waiters reach the quota queue (setup)", formed, "queue depth %d, want at least 90%% of %d", obs.depth.Load(), want)

		r0 := store.reserves.Load()
		t0 := time.Now()
		time.Sleep(window / 2)
		freshAt := time.Now()
		w.notifyWith(ctx, http.DefaultClient, pushRange(fresh[0], 0, 1)[0])
		freshStarted := r.waitFor(30*time.Second, func() bool { return slices.Contains(w.eng.startOrder(), "did:plc:fresh") })
		freshTook := time.Since(freshAt)
		if rest := window - time.Since(t0); rest > 0 {
			time.Sleep(rest)
		}
		secs := time.Since(t0).Seconds()
		reserves := store.reserves.Load() - r0
		stopProbe()

		r.metric("reserve_per_s", float64(reserves)/secs)
		r.metric("fresh_owner_start_ms", float64(freshTook.Milliseconds()))
		r.metric("probe_errors", float64(probeErrors.Load()))
		r.latency("probe_write", &probe)

		perOwner := float64(reserves) / secs / float64(owners)
		r.check("idle waiters don't poll the db (A7)", perOwner <= 1, "%.1f Reserve write txs per waiting owner per second while nothing was released, want at most 1", perOwner)
		quietP99 := quiet.quantile(0.99)
		r.check("unrelated writes aren't starved by the quota tick (A7)", probe.quantile(0.99) <= 10*quietP99+50*time.Millisecond && probeErrors.Load() == 0, "probe insert p99 %v and %d failed, %v when quiet, want within 10x plus 50ms and 0 failed", probe.quantile(0.99), probeErrors.Load(), quietP99)
		r.check("an unrelated owner with quota starts without waiting on the queue (A7)", freshStarted && freshTook < hold/2, "started=%v after %v with %d waiters queued, want started in under half of one workflow's %v hold", freshStarted, freshTook, obs.depth.Load(), hold)

		w.checkDrained(5 * time.Minute)
		leaked := r.waitFor(10*time.Second, func() bool { return w.rows("quota_reservations") == 0 })
		r.check("no reservations leaked after the run (A7)", leaked, "%v quota_reservations rows left, want 0", w.rows("quota_reservations"))
		r.settles("quota_waiters", 0, 30*time.Second)

		half := repos[:max(1, owners/2)]
		ds2 := pushRanges(half, queued, 2*queued)
		var st2 deliveryStats
		w.flood(ctx, ds2, 32, 20, &st2)
		holding := r.waitFor(30*time.Second, func() bool {
			return w.rows("quota_reservations") > 0 && obs.depth.Load() > 0
		})
		r.note("stopped with reservations held and waiters queued: %v", holding)
		w.stop()

		raw, err := sqlite.Open(w.dbPath())
		if err != nil {
			r.t.Fatal(err)
		}
		var left int
		if err := raw.QueryRow(`select count(*) from quota_reservations`).Scan(&left); err != nil {
			r.t.Fatal(err)
		}
		raw.Close()
		r.metric("reservations_after_clean_stop", float64(left))
		r.check("a clean stop leaves no reservations (A7)", left == 0, "%d quota_reservations rows after stopping mid-run, want 0", left)

		// Recover isn't called by Start, only server.go's Run does it
		d, err := db.Make(ctx, w.dbPath())
		if err != nil {
			r.t.Fatal(err)
		}
		qs := db.NewQuotaStore(d, nil)
		leaks := owners * queued
		for i := range leaks {
			id := fmt.Sprintf("crash-%d", i)
			if _, err := qs.Reserve(ctx, quota.ReserveRequest{
				ID:        id,
				Kind:      quota.KindWorkflow,
				Key:       id,
				Identity:  quota.Identity{OwnerDID: ownerFor(i, 1).String(), RepoDID: "did:plc:crashrepo"},
				Resources: quota.Resources{quota.ResourceWorkflows: 1},
			}); err != nil {
				r.t.Fatal(err)
			}
		}
		_, recoverErr := qs.Recover(ctx, nil)
		var after int
		if err := d.QueryRow(`select count(*) from quota_reservations`).Scan(&after); err != nil {
			r.t.Fatal(err)
		}
		d.Close()
		r.check("recover clears what a crash leaves (A7)", recoverErr == nil && after == 0, "err=%v and %d of %d crash reservations still there, want no error and 0", recoverErr, after, leaks)
	})

	// the first workflow waits on it because resumeWipes runs before the workers start
	registerTorture("ban-restart", "a restart with a long ban list, time to the first workflow", func(r *tortureRun) {
		bans := r.load("bans", 5000)

		w := newTortureWorld(r)
		w.maxJobs = 2
		knot := w.addKnot()
		repo := w.addRepo(knot, "did:plc:banowner", 3, map[string]string{"ci.yaml": tortureWorkflow("sleep 10ms")})
		w.boot()
		w.seed([]*tortureRepo{repo}, time.Minute)
		ctx := context.Background()
		pushes := pushRange(repo, 0, 3)

		w.notifyWith(ctx, http.DefaultClient, pushes[0])
		if !r.waitFor(30*time.Second, func() bool { return w.eng.started.Load() >= 1 }) {
			r.t.Fatal("the first workflow never started")
		}

		restart := func(i int) time.Duration {
			before := w.eng.started.Load()
			start := time.Now()
			w.boot()
			reply := w.notifyWith(ctx, http.DefaultClient, pushes[i])
			if reply.status != http.StatusOK {
				r.t.Fatalf("push %d after restart answered %d: %s", i, reply.status, reply.body)
			}
			if !r.waitFor(5*time.Minute, func() bool { return w.eng.started.Load() > before }) {
				r.t.Fatalf("no workflow started within 5m of restart %d", i)
			}
			return time.Since(start)
		}

		w.stop()
		quiet := restart(1)
		w.stop()

		d, err := db.Make(ctx, w.dbPath())
		if err != nil {
			r.t.Fatal(err)
		}
		for i := range bans {
			if err := d.PutBan(db.BanEntry{SubjectDid: syntax.DID(fmt.Sprintf("did:plc:banned%06d", i))}); err != nil {
				r.t.Fatal(err)
			}
		}
		d.Close()
		banned := restart(2)

		r.metric("restart_no_bans_s", quiet.Seconds())
		r.metric("restart_with_bans_s", banned.Seconds())
		r.check("a restart doesn't wait on the ban list (A10)", banned <= quiet+10*time.Second, "first workflow after boot took %v with %d bans and %v without, want at most 10s more", banned, bans, quiet)
	})

	// unknown repo lookups hit the knot, only lookupSlots of them run at once
	// well under the knot client's 30s timeout
	registerTorture("unknown-repo-stripes", "made-up repo dids on a stalling knot against legit refreshes", func(r *tortureRun) {
		attackers := r.load("attackers", 200)
		probes := 40
		stall := 2 * time.Second
		const lookupSlots = 2

		w := newTortureWorld(r)
		w.maxJobs = 2
		slow := w.addKnot()
		fast := w.addKnot()
		evilRepos := w.addRepos(attackers, func(int) (*tortureKnot, syntax.DID) { return slow, "did:plc:evil" }, 0, nil)
		baseRepos := w.addRepos(probes, func(int) (*tortureKnot, syntax.DID) { return fast, "did:plc:legit" }, 0, nil)
		hitRepos := w.addRepos(probes, func(int) (*tortureKnot, syntax.DID) { return fast, "did:plc:legit" }, 0, nil)
		// they look unknown since the web made them before spindle saw the event
		webRepos := w.addRepos(10, func(int) (*tortureKnot, syntax.DID) { return fast, "did:plc:legit" }, 0, nil)
		evil := w.admAddUser("did:plc:evil", slow)
		legit := w.admAddUser("did:plc:legit", fast)
		w.boot()
		w.seed(append(append([]*tortureRepo{}, baseRepos...), hitRepos...), 2*time.Minute)

		ctx := context.Background()
		client := &http.Client{Timeout: 3 * time.Minute}
		listSecrets := func(u *admUser, repo *tortureRepo) notifyReply {
			token := w.admToken(u, org_tangled.SecretListSecretsNSID)
			return w.call(ctx, client, http.MethodGet, org_tangled.SecretListSecretsNSID, "repo="+repo.did.String(), token, nil)
		}
		probe := func(repos []*tortureRepo, into *latencies) {
			var wg sync.WaitGroup
			for _, repo := range repos {
				wg.Go(func() {
					reply := listSecrets(legit, repo)
					into.add(reply.elapsed)
					if reply.status != http.StatusOK {
						r.count("probe_"+statusKind(reply.status), 1)
					}
				})
			}
			wg.Wait()
		}

		var base, attacked latencies
		probe(baseRepos, &base)

		slow.delay.Store(int64(stall))
		var held sync.WaitGroup
		var attackersDone, attackersBusy atomic.Int64
		for _, repo := range evilRepos {
			held.Go(func() {
				reply := listSecrets(evil, repo)
				attackersDone.Add(1)
				if reply.status == http.StatusServiceUnavailable {
					attackersBusy.Add(1)
				}
				r.count("attacker_"+statusKind(reply.status), 1)
			})
		}
		time.Sleep(time.Second)
		stillStalled := int64(attackers) - attackersDone.Load()
		busy := attackersBusy.Load()
		var web latencies
		var webBusy atomic.Int64
		var webWG sync.WaitGroup
		for _, repo := range webRepos {
			webWG.Go(func() {
				reply := listSecrets(legit, repo)
				web.add(reply.elapsed)
				if reply.status == http.StatusServiceUnavailable {
					webBusy.Add(1)
				}
			})
		}
		probe(hitRepos, &attacked)
		webWG.Wait()
		held.Wait()
		r.metric("attackers_stalled_during_probes", float64(stillStalled))
		r.metric("attackers_refused_busy", float64(busy))
		r.metric("web_created_repos_refused_busy", float64(webBusy.Load()))
		r.latency("web_created_repo_lookup", &web)
		r.check("attackers are refused busy at once and only the lookup slots stall (setup)", stillStalled <= lookupSlots && busy >= int64(attackers)-lookupSlots,
			"%d of %d attacker requests still in flight after 1s and %d answered busy, want at most %d in flight and the rest refused busy", stillStalled, attackers, busy, lookupSlots)
		r.note("a repo the web just created and spindle hasn't heard of takes a lookup slot like an attacker's repo, so while the attackers hold them %d of %d such requests were refused busy", webBusy.Load(), len(webRepos))

		baseP90 := base.quantile(0.9)
		attackedP90 := attacked.quantile(0.9)
		r.check("a served repo's request isn't queued behind stalled lookups (A3)", attackedP90 <= baseP90+time.Second, "p90 %v under attack and %v alone with %d attacker dids on a stalling knot, want within 1s of alone", attackedP90, baseP90, attackers)
	})
}