//go:build torture

package spindle

import (
	"context"
	"fmt"
	"math/rand/v2"
	"net/http"
	"strings"
	"sync"
	"sync/atomic"
	"time"

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

type pushDelivery struct {
	repo   *tortureRepo
	ref    string
	oldSha string
	newSha string
}

type deliveryStats struct {
	first    latencies
	accepted latencies
	attempts atomic.Int64
	dropped  atomic.Int64
	statuses sync.Map
}

func (st *deliveryStats) saw(status int) {
	n, _ := st.statuses.LoadOrStore(statusKind(status), new(atomic.Int64))
	n.(*atomic.Int64).Add(1)
}

func (st *deliveryStats) seen(kind string) int {
	n, ok := st.statuses.Load(kind)
	if !ok {
		return 0
	}
	return int(n.(*atomic.Int64).Load())
}

// 503 means busy, which doesn't count as an error
func (st *deliveryStats) serverErrors() int {
	total := 0
	st.statuses.Range(func(k, v any) bool {
		if kind := k.(string); strings.HasPrefix(kind, "http_5") && kind != "http_503" {
			total += int(v.(*atomic.Int64).Load())
		}
		return true
	})
	return total
}

func (st *deliveryStats) record(reply notifyReply, attempt int, start time.Time) {
	st.attempts.Add(1)
	st.saw(reply.status)
	if attempt == 1 {
		st.first.add(reply.elapsed)
	}
	if reply.status == http.StatusOK {
		st.accepted.since(start)
	}
}

func (st *deliveryStats) report(r *tortureRun, prefix string) {
	r.latency(prefix+"first_attempt", &st.first)
	r.latency(prefix+"accepted", &st.accepted)
	r.metric(prefix+"attempts", float64(st.attempts.Load()))
	r.metric(prefix+"dropped", float64(st.dropped.Load()))
	st.statuses.Range(func(k, v any) bool {
		r.count(prefix+k.(string), int(v.(*atomic.Int64).Load()))
		return true
	})
}

func (w *tortureWorld) knotOutbox(ctx context.Context, queue []pushDelivery, st *deliveryStats) {
	client := &http.Client{Timeout: time.Minute}
	backoff := time.Duration(0)
	for _, d := range queue {
		start := time.Now()
		for attempt := 1; ctx.Err() == nil; attempt++ {
			if resp, err := client.Get(w.api.URL + "/.well-known/did.json"); err == nil {
				resp.Body.Close()
			}
			reply := w.notifyWith(ctx, client, d)
			st.record(reply, attempt, start)
			if reply.status == http.StatusOK {
				backoff = 0
				break
			}
			if reply.status >= 400 && reply.status < 500 && reply.status != http.StatusRequestTimeout && reply.status != http.StatusTooManyRequests {
				st.dropped.Add(1)
				break
			}
			backoff = min(max(backoff*2, time.Second), time.Minute)
			sleepCtx(ctx, backoff)
		}
	}
}

func (w *tortureWorld) outboxes(ctx context.Context, queues map[*tortureKnot][]pushDelivery, st *deliveryStats) *latencies {
	var finished latencies
	start := time.Now()
	var wg sync.WaitGroup
	for _, queue := range queues {
		wg.Go(func() {
			w.knotOutbox(ctx, queue, st)
			finished.since(start)
		})
	}
	wg.Wait()
	return &finished
}

func (w *tortureWorld) flood(ctx context.Context, ds []pushDelivery, concurrency, maxAttempts int, st *deliveryStats) {
	var wg sync.WaitGroup
	work := make(chan pushDelivery)
	for range concurrency {
		wg.Go(func() {
			for d := range work {
				start := time.Now()
				wait := 50 * time.Millisecond
				for attempt := 1; ; attempt++ {
					reply := w.notifyWith(ctx, http.DefaultClient, d)
					st.record(reply, attempt, start)
					if reply.status == http.StatusOK {
						break
					}
					if (reply.status >= 0 && reply.status < 500) || attempt == maxAttempts || ctx.Err() != nil {
						st.dropped.Add(1)
						break
					}
					sleepCtx(ctx, wait+rand.N(wait))
					wait = min(wait*2, 2*time.Second)
				}
			}
		})
	}
	for _, d := range ds {
		work <- d
	}
	close(work)
	wg.Wait()
}

func (w *tortureWorld) notifyWith(ctx context.Context, client *http.Client, d pushDelivery) notifyReply {
	return w.post(ctx, client, w.token(d.repo.knot), w.notifyBody(d.repo, d.ref, d.oldSha, d.newSha))
}

func pushRange(repo *tortureRepo, from, to int) []pushDelivery {
	old := zeroSha
	if from > 0 {
		old = repo.shas[from-1]
	}
	var out []pushDelivery
	for _, sha := range repo.shas[from:to] {
		out = append(out, pushDelivery{repo: repo, ref: "refs/heads/main", oldSha: old, newSha: sha})
		old = sha
	}
	return out
}

func pushesOf(repo *tortureRepo, n int) []pushDelivery { return pushRange(repo, 0, n) }

func pushRanges(repos []*tortureRepo, from, to int) []pushDelivery {
	var out []pushDelivery
	for _, repo := range repos {
		out = append(out, pushRange(repo, from, to)...)
	}
	return out
}

func ownerFor(i, perOwner int) syntax.DID {
	return syntax.DID(fmt.Sprintf("did:plc:owner%04d", i/perOwner))
}

func (w *tortureWorld) checkDrained(within time.Duration) {
	drained := w.r.waitFor(within, func() bool {
		// check jobs first, a job leaves the table before it counts as running
		return w.rows("jobs") == 0 && w.eng.running.Load() == 0 && w.unfinishedWorkflows() == 0
	})
	w.r.check("queue drains", drained, "jobs=%v running=%d", w.rows("jobs"), w.eng.running.Load())
	w.r.check("nothing left unfinished", w.unfinishedWorkflows() == 0, "%v workflows not finished", w.unfinishedWorkflows())
}

func (w *tortureWorld) unfinishedWorkflows() float64 {
	return w.countRows(`
		select count(*) from workflow_statuses ws
		where ws.id = (
			select max(id) from workflow_statuses latest
			where latest.pipeline_id = ws.pipeline_id and latest.workflow = ws.workflow
		) and ws.status in ('pending', 'running')`)
}

func (w *tortureWorld) pushWorld(knots, repos, pushes int, workflows map[string]string) []*tortureRepo {
	w.addKnots(knots)
	rs := w.addRepos(repos, func(i int) (*tortureKnot, syntax.DID) {
		return w.knots[i%knots], ownerFor(i, 3)
	}, pushes, workflows)
	w.boot()
	w.seed(rs, 3*time.Minute)
	return rs
}

func (w *tortureWorld) checkPipelines(r *tortureRun, want int, what string) {
	got := int(w.rows("pipelines"))
	r.check("one pipeline per unique push", got == want, "%d pipelines for %d unique pushes (%s)", got, want, what)
}

func init() {
	registerTorture("notify-outbox-flush", "knots coming back with full outboxes, the way knot2 sends them", func(r *tortureRun) {
		knots := r.load("knots", 12)
		perKnot := 4
		pushes := 12
		dupRate := float64(0.1)
		step := "sleep " + r.duration("step", 50*time.Millisecond).String()

		w := newTortureWorld(r)
		w.maxJobs = int(r.param("workers", 8))
		rs := w.pushWorld(knots, knots*perKnot, pushes, map[string]string{"ci.yaml": tortureWorkflow(step)})
		baseGoroutines := r.gaugeValue("goroutines")

		queues := map[*tortureKnot][]pushDelivery{}
		unique := 0
		for _, repo := range rs {
			for _, d := range pushRange(repo, 0, pushes) {
				queues[repo.knot] = append(queues[repo.knot], d)
				unique++
				// a lost ack makes the knot resend, and some pushes get sent twice on purpose
				if rand.Float64() < dupRate {
					queues[repo.knot] = append(queues[repo.knot], d)
				}
			}
		}
		// only cross repo swaps, a repo's own pushes have to stay in order
		for _, q := range queues {
			rand.Shuffle(len(q), func(i, j int) {
				if q[i].repo != q[j].repo {
					q[i], q[j] = q[j], q[i]
				}
			})
		}

		var st deliveryStats
		finished := w.outboxes(context.Background(), queues, &st)
		st.report(r, "notify_")
		r.latency("knot_finished", finished)

		w.checkDrained(5 * time.Minute)
		r.check("no delivery dropped", st.dropped.Load() == 0, "%d deliveries dropped by the knot", st.dropped.Load())
		r.check("no server errors besides busy", st.serverErrors() == 0, "%d 5xx answers that weren't 503", st.serverErrors())
		w.checkPipelines(r, unique, fmt.Sprintf("%d deliveries", st.attempts.Load()))
		r.settles("goroutines", baseGoroutines+20, 30*time.Second)
	})

	registerTorture("notify-flood", "one knot that ignores the outbox discipline and blasts the caps", func(r *tortureRun) {
		repos := r.load("repos", 16)
		pushes := 10
		concurrency := 64
		replays := float64(0.5)
		step := "sleep " + r.duration("step", 20*time.Millisecond).String()

		w := newTortureWorld(r)
		w.maxJobs = int(r.param("workers", 8))
		rs := w.pushWorld(1, repos, pushes, map[string]string{"ci.yaml": tortureWorkflow(step)})
		baseGoroutines := r.gaugeValue("goroutines")

		ds := pushRanges(rs, 0, pushes)
		unique := len(ds)
		for _, d := range ds[:unique] {
			if rand.Float64() < replays {
				ds = append(ds, d)
			}
		}
		rand.Shuffle(len(ds), func(i, j int) { ds[i], ds[j] = ds[j], ds[i] })

		var st deliveryStats
		w.flood(context.Background(), ds, concurrency, 40, &st)
		st.report(r, "notify_")
		r.metric("busy_ratio", float64(st.seen("http_503"))/float64(st.attempts.Load()))

		w.checkDrained(5 * time.Minute)
		r.check("every delivery accepted", st.dropped.Load() == 0, "%d deliveries gave up", st.dropped.Load())
		r.check("no server errors besides busy", st.serverErrors() == 0, "%d 5xx answers that weren't 503", st.serverErrors())
		w.checkPipelines(r, unique, fmt.Sprintf("%d deliveries", len(ds)))
		r.settles("goroutines", baseGoroutines+20, 30*time.Second)
	})
}