//go:build torture

package spindle

import (
	"context"
	"fmt"
	"slices"
	"sync"
	"sync/atomic"
	"time"

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

// knots includes the caller's own knot
func poolFairShare(pool, knots int) int {
	return min(max(2, pool/knots), max(1, pool-poolReserve(pool)))
}

func init() {
	// one size per run to fit under MemoryMax, e.g. -torture.set pool=64,notify=16
	registerTorture("pool-sizing", "what a repo job pool size and a notify slot count cost idle and under a catch-up with pushes", func(r *tortureRun) {
		n := r.load("repos", 5000)
		repoPool := int(r.param("pool", 256))
		notifySlots := int(r.param("notify", 64))
		smallKnots := 4
		pushRepos := 40
		slowDelay := 2 * time.Second
		budget := r.duration("drain_budget", 30*time.Minute)

		w := newTortureWorld(r)
		w.repoPool, w.notifySlots = repoPool, notifySlots
		w.maxJobs = 8
		big := w.addKnot()
		small := w.addKnots(smallKnots)
		slow := w.addKnot()
		pick := func(i int) (*tortureKnot, syntax.DID) {
			owner := syntax.DID(fmt.Sprintf("did:plc:poolowner%05d", i))
			switch rem := i % 10; {
			case rem < 6:
				return big, owner
			case rem < 9:
				return small[i%smallKnots], owner
			}
			return slow, owner
		}
		repos := w.addRepos(n, pick, 0, nil)
		pushes := w.addRepos(pushRepos, func(i int) (*tortureKnot, syntax.DID) {
			return small[i%smallKnots], syntax.DID(fmt.Sprintf("did:plc:poolpush%03d", i))
		}, 2, map[string]string{"ci.yaml": tortureWorkflow("sleep 10ms")})
		r.metric("repo_pool", float64(repoPool))
		r.metric("notify_slots", float64(notifySlots))

		r.gauge("knots_inflight", func() float64 {
			var total int64
			for _, k := range w.knots {
				total += k.inflight.Load()
			}
			return float64(total)
		})
		r.gauge("slow_knot_inflight", func() float64 { return float64(slow.inflight.Load()) })
		r.gauge("notify_inflight", w.abuseNotifyInflight)
		r.gauge("notify_held_max_per_knot", func() float64 {
			most := 0.0
			for _, k := range w.knots {
				most = max(most, w.abuseHeld(k.host()))
			}
			return most
		})

		w.boot()
		time.Sleep(5 * time.Second)
		r.metric("idle_rss_mb", r.gaugeValue("rss_mb"))
		r.metric("idle_goroutines", r.gaugeValue("goroutines"))
		r.metric("idle_fds", r.gaugeValue("fds"))

		all := append(slices.Clone(repos), pushes...)
		w.seed(all, 10*time.Minute)
		r.metric("served_rss_mb", r.gaugeValue("rss_mb"))

		w.consumed(w.js.filler(), 30*time.Second)
		w.stop()
		var changed []syntax.DID
		var changedOn []*tortureRepo
		for i := 0; i < len(repos); i += 5 {
			who := syntax.DID(fmt.Sprintf("did:plc:poolnew%05d", i))
			ingPutCollaborator(repos[i], who)
			changed = append(changed, who)
			changedOn = append(changedOn, repos[i])
		}
		for _, k := range w.knots {
			k.delay.Store((20 * time.Millisecond).Nanoseconds())
		}
		slow.delay.Store(slowDelay.Nanoseconds())
		w.js.forget()
		w.js.filler()
		readsBefore := ingManifestReads(w.knots)

		w.boot()
		start := time.Now()
		ctx, cancel := context.WithTimeout(context.Background(), budget)
		defer cancel()
		var wg sync.WaitGroup

		var st deliveryStats
		ds := pushRanges(pushes, 0, 2)
		wg.Go(func() { w.flood(ctx, ds, min(2*notifySlots, 512), 40, &st) })

		var applied grantProbes
		stopLive := make(chan struct{})
		var liveSent atomic.Int64
		wg.Go(func() {
			for i := 0; ; i++ {
				select {
				case <-stopLive:
					return
				case <-time.After(250 * time.Millisecond):
				}
				w.probeGrant(repos[0], syntax.DID(fmt.Sprintf("did:plc:poollive%05d", i)), 2*time.Minute, &applied)
				liveSent.Add(1)
			}
		})

		// running jobs can stay over the share, only starting one over it is a violation
		slowHost := slow.host()
		var jobsPeak, violations atomic.Int64
		stopSample := make(chan struct{})
		sampling := make(chan struct{})
		go func() {
			defer close(sampling)
			prevRunning, prevOthers := 0, 0
			for {
				select {
				case <-stopSample:
					return
				case <-time.After(2 * time.Millisecond):
				}
				jobs := &w.spindle().retries.jobs
				jobs.mu.Lock()
				running := jobs.running[slowHost]
				others := 0
				for knot := range jobs.queued {
					if knot != slowHost {
						others++
					}
				}
				jobs.mu.Unlock()
				if others > 0 {
					if int64(running) > jobsPeak.Load() {
						jobsPeak.Store(int64(running))
					}
					if prevOthers > 0 && running > prevRunning && running > poolFairShare(repoPool, others+1) {
						violations.Add(1)
					}
				}
				prevRunning, prevOthers = running, others
			}
		}()

		w.probeTable()
		var probe latencies
		var probeErrors atomic.Int64
		stopProbe := w.probeWrites(50*time.Millisecond, &probe, &probeErrors)

		caughtUp := r.waitFor(budget, func() bool {
			pending, err := w.spindle().db.PendingCatchUp()
			return err == nil && len(pending) == 0
		})
		catchUp := time.Since(start)
		applied2 := r.waitFor(max(time.Until(start.Add(budget)), time.Second), func() bool {
			for i, who := range changed {
				if !w.isCollab(changedOn[i], who) {
					return false
				}
			}
			return true
		})
		elapsed := time.Since(start)
		close(stopLive)
		stopProbe()
		close(stopSample)
		cancel()
		wg.Wait()
		<-sampling
		applied.wg.Wait()

		// over budget there's no real time to report, only finished ones do
		if caughtUp {
			r.metric("catchup_s", catchUp.Seconds())
		}
		if applied2 {
			r.metric("apply_all_s", elapsed.Seconds())
		}
		r.metric("run_s", elapsed.Seconds())
		r.metric("refreshes_per_s", float64(ingManifestReads(w.knots)-readsBefore)/elapsed.Seconds())
		r.metric("live_events_per_s", float64(liveSent.Load())/elapsed.Seconds())
		r.metric("notify_accepted_per_s", float64(st.accepted.len())/elapsed.Seconds())
		r.metric("probe_write_errors", float64(probeErrors.Load()))
		r.metric("slow_knot_jobs_peak_while_others_wait", float64(jobsPeak.Load()))
		r.metric("fair_share_violations", float64(violations.Load()))
		r.latency("probe_write", &probe)
		r.latency("live_apply", &applied.latencies)
		r.metric("live_missed", float64(applied.missed.Load()))
		st.report(r, "notify_")
		r.note("rss, goroutines, fds and the knots' open requests are sampled all run, read them as _max in the metrics. busy and locked errors from spindle's own writes show up in errors by message, the probe write errors are only the probe's")

		r.check("catch-up finished within the budget", caughtUp, "the pending catch-up was not empty after %v", budget)
		r.check("every changed repo applied within the budget", applied2, "%d changed repos, not all applied %v after the boot", len(changed), budget)
		r.check("a slow knot never starts another repo job past its share while others wait", violations.Load() == 0,
			"the slow knot started %d jobs while over its share of %d workers with other knots waiting, its most running at once then was %d", violations.Load(), repoPool, jobsPeak.Load())
	})
}

func poolReserve(pool int) int { return max(2, pool/8) }

func init() {
	registerTorture("pool-priority", "a priority knot's repo jobs run at once while a hostile knot pins the shared repo pool", func(r *tortureRun) {
		hostileRepos := r.load("hostile_repos", 300)
		pool := int(r.param("pool", 16))
		hold := 4 * time.Second
		budget := r.duration("priority_budget", 3*time.Second)

		w := newTortureWorld(r)
		w.repoPool = pool
		prio, plain, hostile := w.addKnot(), w.addKnot(), w.addKnot()
		w.priorityKnots = []string{prio.host()}
		prioRepos := w.addRepos(10, func(i int) (*tortureKnot, syntax.DID) {
			return prio, syntax.DID(fmt.Sprintf("did:plc:prioowner%03d", i))
		}, 0, nil)
		plainRepos := w.addRepos(10, func(i int) (*tortureKnot, syntax.DID) {
			return plain, syntax.DID(fmt.Sprintf("did:plc:plainowner%03d", i))
		}, 0, nil)
		bad := w.addRepos(hostileRepos, func(i int) (*tortureKnot, syntax.DID) {
			return hostile, syntax.DID(fmt.Sprintf("did:plc:hostileowner%04d", i))
		}, 0, nil)
		all := append(append(append([]*tortureRepo{}, prioRepos...), plainRepos...), bad...)
		w.boot()
		w.seed(all, 5*time.Minute)
		w.consumed(w.js.filler(), 30*time.Second)
		w.stop()

		for i, repo := range bad {
			ingPutCollaborator(repo, syntax.DID(fmt.Sprintf("did:plc:hostilenew%04d", i)))
		}
		hostile.delay.Store(hold.Nanoseconds())
		w.js.forget()
		w.js.filler()
		w.boot()

		hostileHost := hostile.host()
		running := func() int {
			jobs := &w.spindle().retries.jobs
			jobs.mu.Lock()
			defer jobs.mu.Unlock()
			return jobs.running[hostileHost]
		}
		pinned := r.waitFor(time.Minute, func() bool { return running() >= pool-poolReserve(pool) })
		r.metric("hostile_jobs_running", float64(running()))
		r.check("the hostile knot pins the shared pool (setup)", pinned, "the hostile knot had %d of %d workers running, want %d", running(), pool, pool-poolReserve(pool))

		var prioProbe, plainProbe grantProbes
		w.probeGrant(prioRepos[0], syntax.DID("did:plc:prioprobe"), time.Minute, &prioProbe)
		w.probeGrant(plainRepos[0], syntax.DID("did:plc:plainprobe"), time.Minute, &plainProbe)
		prioProbe.wg.Wait()
		plainProbe.wg.Wait()
		missed := prioProbe.missed.Load() + plainProbe.missed.Load()
		r.latency("priority_knot_apply", &prioProbe.latencies)
		r.latency("plain_knot_apply", &plainProbe.latencies)
		r.metric("probes_missed", float64(missed))
		r.check("a priority knot's job runs at once while a hostile knot pins the pool", missed == 0 && prioProbe.quantile(1) <= budget,
			"priority knot's event applied after %v with %d missed, want within %v while the hostile knot ran %d workers", prioProbe.quantile(1), missed, budget, running())
	})
}