#!/usr/bin/env bash
# PrivaPub under a crowd: the flood peer's twenty servers send signed activities at a set rate while a persona reads its
# home timeline, then the queue drains. Prints, and keeps in out/load/, what the crowd saw (PrivaPub's answers and how
# long they took), how long the activities waited in PrivaPub's queue and took to process, how long the queue took to
# drain, and how the home timeline answered meanwhile. Needs the pasture with the flood peer (run.sh add flood).
# usage: tools/pasture/load.sh [--rate=40] [--seconds=60] [--personas=5] [--follows=20] [--mix=create=70,like=20,follow=10]
set -uo pipefail
here="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
. "$here/lib/interop.sh"
rate=40; seconds=60; personas=5; follows=20; mix="create=70,like=20,follow=10"
for arg in "$@"; do
	case "$arg" in
		--rate=*) rate=${arg#*=} ;; --seconds=*) seconds=${arg#*=} ;; --personas=*) personas=${arg#*=} ;;
		--follows=*) follows=${arg#*=} ;; --mix=*) mix=${arg#*=} ;;
		*) echo "usage: $0 [--rate=N] [--seconds=N] [--personas=N] [--follows=N] [--mix=...]" >&2; exit 2 ;;
	esac
done
podman container exists pasture-flood || { echo "the flood peer is not up: run.sh add flood" >&2; exit 1; }
mongo() { podman exec pasture-mongo mongosh --quiet PrivaPub --eval "$1"; }

echo "load: $personas personas following $follows flood actors each; $rate activities a second for ${seconds}s ($mix)"
uris=(); tokens=(); : > "$here/.state/load-posts.txt"
for i in $(seq 0 $((personas - 1))); do
	token=$(privapub_token "load$i")
	[ -n "$token" ] || { echo "no token for load$i" >&2; exit 1; }
	tokens+=("$token")
	uris+=("$(curl -s -H "Authorization: Bearer $token" "$P/api/v1/accounts/verify_credentials" | j "print(d['url'].replace('/@', '/peasants/'))")")
	# each follows its own share of the crowd, so a note reaches some homes and not others
	for f in $(seq 0 $((follows - 1))); do
		n=$(( (i * follows + f) % 200 ))
		id=$(curl -s -H "Authorization: Bearer $token" "$P/api/v2/search?q=u$((n % 10))@flood$((n / 10 + 1)).test&resolve=true&type=accounts" | j "print(d['accounts'][0]['id'])")
		[ -n "$id" ] && curl -s -o /dev/null -X POST -H "Authorization: Bearer $token" "$P/api/v1/accounts/$id/follow"
	done
	for p in 1 2; do
		curl -s -X POST -H "Authorization: Bearer $token" "$P/api/v1/statuses" -d "status=a post for the crowd to like ($p)&visibility=public" | j "print(d['uri'])" >> "$here/.state/load-posts.txt"
	done
done
podman cp "$here/.state/load-posts.txt" pasture-flood:/state/posts.txt
# the crowd's Accepts arrive before the run starts
until_true 60 '[ "$(mongo "print(db.Following.countDocuments({TargetActorURI: /flood/, State: 1}))")" -ge $((personas * follows * 9 / 10)) ]' || true
echo "  follows accepted: $(mongo "print(db.Following.countDocuments({TargetActorURI: /flood/, State: 1}))")"

started=$(date -u +%Y-%m-%dT%H:%M:%S.000Z)
reader="${tokens[0]}"
sampler="$here/.state/load-api.txt"; : > "$sampler"
rm -f "$here/.state/load-stop"; echo 0 > "$here/.state/load-peak"
( while [ ! -e "$here/.state/load-stop" ]; do
	curl -s -o /dev/null -w '%{http_code} %{time_total}\n' -H "Authorization: Bearer $reader" "$P/api/v1/timelines/home?limit=40" >> "$sampler"
	sleep 0.25
done ) &
( peak=0; while [ ! -e "$here/.state/load-stop" ]; do
	q=$(mongo 'print(db.Job.countDocuments({State: {$in: [0, 1]}, RunAt: {$lte: new Date()}}))'); [ "$q" -gt "$peak" ] && peak=$q && echo "$peak" > "$here/.state/load-peak"; sleep 2
done ) &
crowd=$(podman exec pasture-flood /app/flood run --target=https://privapub.test --rate="$rate" --seconds="$seconds" \
	--personas="$(IFS=,; echo "${uris[*]}")" --posts=/state/posts.txt --mix="$mix")
sent_at=$(date +%s)
# (jobs that are due: retries waiting for a later attempt, a dead host's, are not the crowd's)
until_true 300 '[ "$(mongo "print(db.Job.countDocuments({State: {\$in: [0, 1]}, RunAt: {\$lte: new Date()}}))")" = "0" ]'
drain=$(( $(date +%s) - sent_at ))
touch "$here/.state/load-stop"; wait 2>/dev/null; rm -f "$here/.state/load-stop"

inbox=$(mongo "
	const events = db.InteractionEvent.find({Channel: 'in', Host: /^flood/, At: {\$gte: ISODate('$started')}}, {Outcome: 1, Reason: 1, WaitMs: 1, LatencyMs: 1}).toArray();
	const pct = (xs, p) => { const s = xs.filter(x => x != null).sort((a, b) => a - b); return s.length ? s[Math.min(s.length - 1, Math.ceil(p * s.length) - 1)] : null; };
	const processed = events.filter(e => e.Outcome != 'queued' && e.Outcome != 'refused');
	const outcomes = {}; events.forEach(e => outcomes[e.Outcome] = (outcomes[e.Outcome] || 0) + 1);
	print(JSON.stringify({ events: events.length, outcomes, waitP50Ms: pct(processed.map(e => e.WaitMs), 0.5), waitP95Ms: pct(processed.map(e => e.WaitMs), 0.95),
		processP50Ms: pct(processed.map(e => e.LatencyMs), 0.5), processP95Ms: pct(processed.map(e => e.LatencyMs), 0.95) }));")
api=$(python3 - "$sampler" <<'PY'
import json, sys
rows = [l.split() for l in open(sys.argv[1]) if l.strip()]
times = sorted(float(t) * 1000 for s, t in rows if s == "200")
pct = lambda p: round(times[min(len(times) - 1, max(0, int(-(-p * len(times) // 1)) - 1))], 1) if times else None
print(json.dumps({"requests": len(rows), "ok": len(times), "p50Ms": pct(0.5), "p95Ms": pct(0.95), "maxMs": round(times[-1], 1) if times else None}))
PY
)
mkdir -p "$here/out/load"
result=$(python3 -c 'import json, sys; print(json.dumps({"started": sys.argv[1], "rate": float(sys.argv[2]), "seconds": float(sys.argv[3]), "personas": int(sys.argv[4]),
	"follows": int(sys.argv[5]), "mix": sys.argv[6], "crowd": json.loads(sys.argv[7]), "inbox": json.loads(sys.argv[8]), "queuePeak": int(sys.argv[9]),
	"drainSeconds": int(sys.argv[10]), "homeTimeline": json.loads(sys.argv[11])}, indent=1))' \
	"$started" "$rate" "$seconds" "$personas" "$follows" "$mix" "$crowd" "$inbox" "$(cat "$here/.state/load-peak" 2>/dev/null || echo 0)" "$drain" "$api")
echo "$result" | tee "$here/out/load/$(date +%Y%m%d-%H%M%S).json"