package oracle

// This file is compiled into the pinned upstream internal/oracle package by
// tests/differential_oracle.py. It drives Stream through the real upstream
// simulator and holds real getRepo requests at the handler boundary long
// enough to make Atmos's 32-worker / 64-queued-job envelope observable.

import (
	"context"
	"log/slog"
	"math/rand/v2"
	"net/http"
	"net/http/httptest"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"sync"
	"sync/atomic"
	"syscall"
	"testing"
	"time"

	"github.com/bluesky-social/jetstream/internal/simulator/fanout"
	simhttp "github.com/bluesky-social/jetstream/internal/simulator/http"
	"github.com/bluesky-social/jetstream/internal/simulator/world"
	"github.com/stretchr/testify/require"
)

type resyncGetRepoGate struct {
	enabled atomic.Bool
	active  atomic.Int64
	maximum atomic.Int64
	total   atomic.Int64
	release chan struct{}
	once    sync.Once
}

func newResyncGetRepoGate() *resyncGetRepoGate {
	return &resyncGetRepoGate{release: make(chan struct{})}
}

func (g *resyncGetRepoGate) open() { g.once.Do(func() { close(g.release) }) }

func (g *resyncGetRepoGate) wrap(next http.Handler) http.Handler {
	return http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
		if g.enabled.Load() && r.URL.Path == "/xrpc/com.atproto.sync.getRepo" {
			active := g.active.Add(1)
			g.total.Add(1)
			for {
				seen := g.maximum.Load()
				if active <= seen || g.maximum.CompareAndSwap(seen, active) {
					break
				}
			}
			defer g.active.Add(-1)
			select {
			case <-g.release:
			case <-r.Context().Done():
				return
			}
		}
		next.ServeHTTP(rw, r)
	})
}

func TestStreamResyncCapacityOracle(t *testing.T) {
	require.Equal(t, streamDifferentialPin, os.Getenv("STREAM_ORACLE_EXPECTED_PIN"))
	streamBin, err := filepath.Abs(os.Getenv("STREAM_ORACLE_BIN"))
	require.NoError(t, err)
	require.FileExists(t, streamBin)

	const accounts = 96 // 32 active workers + the exact 64-job queue.
	simCfg := world.DefaultConfig()
	simCfg.DataDir = filepath.Join(t.TempDir(), "simulator")
	simCfg.Seed = 0x32_64_2048
	simCfg.Accounts = accounts
	simCfg.InitialRecords = 128
	simCfg.CommitsPerSec = 1
	simCfg.FirehoseHistory = 4096

	w, err := world.New(t.Context(), simCfg)
	require.NoError(t, err)
	defer func() { require.NoError(t, w.Close()) }()
	_, err = w.EnsureSeed()
	require.NoError(t, err)
	require.NoError(t, w.Bootstrap(t.Context(), slog.Default()))
	fan := fanout.New(8192)
	require.NoError(t, w.AttachRuntime(
		rand.New(rand.NewPCG(simCfg.Seed^0xfeedf00d, simCfg.Seed^0xc0ffee)),
		fan,
	))

	gate := newResyncGetRepoGate()
	simSrv := httptest.NewServer(nil)
	simSrv.Config.Handler = gate.wrap(simhttp.NewHandler(w, simSrv.URL))
	defer func() {
		// Release held handlers before Close waits for them. This also keeps a
		// failed anti-vacuity assertion from hanging the test process.
		gate.open()
		simSrv.Close()
	}()

	port := freeStreamOraclePort(t)
	baseURL := "http://127.0.0.1:" + strconv.Itoa(port)
	dataDir := filepath.Join(t.TempDir(), "stream")
	logs := &synchronizedBuffer{}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	cmd := exec.CommandContext(ctx, streamBin,
		"--port="+strconv.Itoa(port),
		"--data-dir="+dataDir,
		"--relay-url="+simSrv.URL,
		"--plc-url="+simSrv.URL,
		"--max-segment-bytes=4194304",
		"--compaction-interval=0",
		"--retry-interval=0",
	)
	cmd.Stdout = logs
	cmd.Stderr = logs
	require.NoError(t, cmd.Start())
	stopped := false
	defer func() {
		if stopped {
			return
		}
		_ = cmd.Process.Signal(syscall.SIGTERM)
		waitDone := make(chan struct{})
		go func() {
			_ = cmd.Wait()
			close(waitDone)
		}()
		select {
		case <-waitDone:
		case <-time.After(10 * time.Second):
			_ = cmd.Process.Kill()
			<-waitDone
		}
	}()

	waitForStreamOracleServing(t, cmd, baseURL, logs)
	_, err = w.GenerateIdentityForTest(t.Context(), 0, false)
	require.NoError(t, err)
	waitForStreamPipelineDrain(t, baseURL, w.CurrentSeq(), logs)

	// First sighting is accepted by both implementations and seeds durable
	// chain state. The next commit silently mutates its predecessor, forcing
	// one authentic whole-repo repair for every distinct DID.
	for i := range accounts {
		_, _, err = w.GenerateRecordOpForTest(t.Context(), i, "create", "app.bsky.feed.post", "capacity-baseline")
		require.NoError(t, err)
	}
	waitForStreamPipelineDrain(t, baseURL, w.CurrentSeq(), logs)

	gate.enabled.Store(true)
	for i := range accounts {
		_, err = w.GenerateSilentMutationThenCommitForTest(t.Context(), i)
		require.NoError(t, err)
	}

	var saturated string
	deadline := time.Now().Add(45 * time.Second)
	for time.Now().Before(deadline) {
		saturated = scrapeStreamOracleMetrics(t, baseURL)
		active, activeOK := metricValue(saturated, "stream_resync_workers_active")
		queued, queueOK := metricValue(saturated, "stream_resync_queue_depth")
		if activeOK && queueOK && active == 32 && queued == 64 && gate.active.Load() == 32 {
			break
		}
		time.Sleep(20 * time.Millisecond)
	}
	requireStreamMetric(t, saturated, "stream_resync_workers_active", 32)
	requireStreamMetric(t, saturated, "stream_resync_queue_depth", 64)
	require.Equal(t, int64(32), gate.active.Load(), "all real fetch workers must reach getRepo")
	require.Equal(t, int64(32), gate.maximum.Load())
	rssAtSaturation, ok := metricValue(saturated, "process_resident_memory_bytes")
	require.True(t, ok)

	gate.open()
	var drained string
	var peakRSS = rssAtSaturation
	deadline = time.Now().Add(90 * time.Second)
	for time.Now().Before(deadline) {
		drained = scrapeStreamOracleMetrics(t, baseURL)
		if rss, ok := metricValue(drained, "process_resident_memory_bytes"); ok && rss > peakRSS {
			peakRSS = rss
		}
		completed, completedOK := metricValue(drained, `stream_resync_total{result="completed"}`)
		active, activeOK := metricValue(drained, "stream_resync_workers_active")
		queued, queueOK := metricValue(drained, "stream_resync_queue_depth")
		if completedOK && activeOK && queueOK && completed == accounts && active == 0 && queued == 0 {
			break
		}
		time.Sleep(20 * time.Millisecond)
	}
	requireStreamMetric(t, drained, `stream_resync_total{result="completed"}`, accounts)
	requireStreamMetric(t, drained, `stream_resync_total{result="failed"}`, 0)
	requireStreamMetric(t, drained, "stream_resync_workers_active", 0)
	requireStreamMetric(t, drained, "stream_resync_queue_depth", 0)
	require.Equal(t, int64(accounts), gate.total.Load(), "every repair must fetch a real CAR")

	require.NoError(t, cmd.Process.Signal(syscall.SIGTERM))
	require.NoErrorf(t, cmd.Wait(), "Stream did not shut down cleanly:\n%s", logs.String())
	stopped = true

	events, err := ObserveSegments(dataDir)
	require.NoError(t, err)
	require.NoError(t, CheckInvariants(events))
	ground, err := GroundTruthFromWorld(w)
	require.NoError(t, err)
	reconstructed, err := Reconstruct(EventsSortedBySeq(events))
	require.NoError(t, err)
	require.NoError(t, Compare(ground, reconstructed),
		"all 96 saturated repairs must converge to the simulator MST")
	t.Logf("resync capacity receipt: workers=32 queue=64 real_getRepo=%d saturation_rss=%.0f peak_rss=%.0f",
		gate.total.Load(), rssAtSaturation, peakRSS)
}