//go:build torture

package spindle

import (
	"context"
	"fmt"
	"sync"
	"time"

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

	"tangled.org/core/api/org_tangled"
	"tangled.org/core/jetstream"
)

func init() {
	registerTorture("soak", "steady mixed traffic for a long time, to see what grows that shouldn't", func(r *tortureRun) {
		knots := r.load("knots", 4)
		perKnot := 3
		length := 3 * time.Minute
		interval := time.Second
		churn := 500 * time.Millisecond
		step := "lines 200; sleep " + r.duration("step", 100*time.Millisecond).String()

		commits := int(length/interval)/perKnot + 2
		w := newTortureWorld(r)
		w.maxJobs = int(r.param("workers", 8))
		rs := w.pushWorld(knots, knots*perKnot, commits, map[string]string{"ci.yaml": tortureWorkflow(step)})
		r.note("pushes per knot are serial and spaced like a busy knot, the engine is the fake one")

		ctx, cancel := context.WithTimeout(context.Background(), length)
		defer cancel()
		var st deliveryStats
		var wg sync.WaitGroup
		for _, k := range w.knots {
			var mine []*tortureRepo
			for _, repo := range rs {
				if repo.knot == k {
					mine = append(mine, repo)
				}
			}
			wg.Go(func() {
				next := make(map[*tortureRepo]int)
				tick := time.NewTicker(interval)
				defer tick.Stop()
				for i := 0; ; i++ {
					select {
					case <-ctx.Done():
						return
					case <-tick.C:
					}
					repo := mine[i%len(mine)]
					n := next[repo]
					if n >= len(repo.shas) {
						continue
					}
					next[repo] = n + 1
					w.knotOutbox(ctx, pushRange(repo, n, n+1), &st)
				}
			})
		}
		wg.Go(func() {
			tick := time.NewTicker(churn)
			defer tick.Stop()
			for i := 0; ; i++ {
				select {
				case <-ctx.Done():
					return
				case <-tick.C:
				}
				repo := rs[i%len(rs)]
				who := syntax.DID(fmt.Sprintf("did:plc:collab%02d", i%26))
				op := jetstream.CommitOperationCreate
				if (i/len(rs))%2 == 1 {
					op = jetstream.CommitOperationDelete
					repo.knot.erase(repo.did, org_tangled.GraphCollaboratorNSID, who.String())
				} else {
					repo.knot.put(repo.did, org_tangled.GraphCollaboratorNSID, who.String(), map[string]string{"createdAt": "2026-01-01T00:00:00Z"})
				}
				w.emitCollaborator(repo, who, op)
			}
		})
		wg.Wait()
		st.report(r, "notify_")
		// no heap check, HeapInuse swings with the gc phase and the fake jetstream holds every event
		for gauge, limit := range map[string]float64{"goroutines": 10, "fds": 2} {
			growth := r.growth(gauge)
			r.check(gauge+" stays flat", growth <= limit, "%s grows %.1f per minute over the second half, want <= %.0f", gauge, growth, limit)
		}

		w.checkDrained(2 * time.Minute)
		r.check("no delivery dropped", st.dropped.Load() == 0, "%d deliveries dropped", st.dropped.Load())
		r.check("retry table drains", r.waitFor(time.Minute, func() bool { return w.rows("repo_retries") == 0 }), "%v repo retries left", w.rows("repo_retries"))
		r.metric("pipelines_per_min", w.rows("pipelines")/length.Minutes())
		r.metric("db_mb_per_1k_pipelines", r.gaugeValue("db_mb")/w.rows("pipelines")*1000)
	})
}