package calls

import (
	"context"
	"fmt"
	"log/slog"
	"net/http"
	"path/filepath"
	"runtime"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"amadan.net/rastrillo/rastrillo/db"
)

// stuckHandler is a slog handler whose every write blocks until release
// is closed: a log sink that has stalled, as a full pipe does. It
// remembers what it was finally allowed to write.
type stuckHandler struct {
	release chan struct{}
	mu      *sync.Mutex
	msgs    *[]string
}

func (h stuckHandler) Enabled(context.Context, slog.Level) bool { return true }
func (h stuckHandler) Handle(_ context.Context, r slog.Record) error {
	<-h.release
	h.mu.Lock()
	*h.msgs = append(*h.msgs, r.Message)
	h.mu.Unlock()
	return nil
}
func (h stuckHandler) WithAttrs([]slog.Attr) slog.Handler { return h }
func (h stuckHandler) WithGroup(string) slog.Handler      { return h }

// funcSender adapts a function to mail.Sender.
type funcSender func(ctx context.Context, to string) error

func (f funcSender) Send(ctx context.Context, to, _, _ string) error { return f(ctx, to) }

func everyone(string) bool { return true }

// TestMailQueueSendNeverBlocksWhenFull: Send is the request path, so it
// may not block or do I/O even when the queue is full — not even to log
// the drop (Astra on oficina/slides #20, round 2, P1: a "queue full" log
// line on a stalled sink held the request, and saturation is
// admission-dependent). The worker is stuck on its first link and the
// log sink is stuck too; twice the queue's depth of Sends must still
// return at once, and the drops are logged by the worker afterwards.
func TestMailQueueSendNeverBlocksWhenFull(t *testing.T) {
	release := make(chan struct{})
	var once sync.Once
	free := func() { once.Do(func() { close(release) }) }
	stuck := funcSender(func(ctx context.Context, _ string) error {
		select {
		case <-release:
		case <-ctx.Done():
		}
		return nil
	})
	var mu sync.Mutex
	var msgs []string
	q := newMailQueue(stuck, everyone, nil, slog.New(stuckHandler{release, &mu, &msgs}))
	t.Cleanup(func() { free(); q.close() })

	done := make(chan struct{})
	go func() {
		for i := 0; i < 2*mailQueueDepth; i++ {
			_ = q.Send(context.Background(), fmt.Sprintf("p%d@example.test", i), "s", "b")
		}
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(2 * time.Second):
		free()
		<-done
		t.Fatal("Send blocked with the queue full: the request path waited on the queue or the log")
	}
	// The drops are the worker's to log, through the pump, once the
	// sink is free.
	free()
	q.close()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		mu.Lock()
		for _, m := range msgs {
			if strings.Contains(m, "links dropped") {
				mu.Unlock()
				return
			}
		}
		mu.Unlock()
		time.Sleep(5 * time.Millisecond)
	}
	t.Fatalf("no drop was logged with twice the queue's depth sent; the log has %q", msgs)
}

// TestMailQueueQueuesEveryAddress: the queue's fill level must not
// depend on admission, so Send queues a refused address's link exactly
// as it queues a permitted one; only the worker asks.
func TestMailQueueQueuesEveryAddress(t *testing.T) {
	release := make(chan struct{})
	stuck := funcSender(func(ctx context.Context, _ string) error {
		select {
		case <-release:
		case <-ctx.Done():
		}
		return nil
	})
	var asked sync.Map
	admits := func(to string) bool { asked.Store(to, true); return to == "first@example.test" }
	q := newMailQueue(stuck, admits, nil, slog.New(slog.DiscardHandler))
	t.Cleanup(func() { close(release); q.close() })
	_ = q.Send(context.Background(), "first@example.test", "s", "b")
	deadline := time.Now().Add(2 * time.Second)
	for _, ok := asked.Load("first@example.test"); !ok && time.Now().Before(deadline); _, ok = asked.Load("first@example.test") {
		time.Sleep(5 * time.Millisecond)
	}
	for i := 0; i < 5; i++ {
		_ = q.Send(context.Background(), fmt.Sprintf("refused%d@example.test", i), "s", "b")
	}
	if got := len(q.jobs); got != 5 {
		t.Fatalf("queued %d refused links, want all 5: fill level must not depend on admission", got)
	}
}

// TestMailQueueSurvivesAPanic: a panic in the mailer is that message's
// failure; the worker recovers, logs, and sends the next one (Astra,
// round 2, P2).
func TestMailQueueSurvivesAPanic(t *testing.T) {
	var mu sync.Mutex
	var sent []string
	calls := 0
	s := funcSender(func(_ context.Context, to string) error {
		mu.Lock()
		defer mu.Unlock()
		calls++
		if calls == 1 {
			panic("mailer exploded")
		}
		sent = append(sent, to)
		return nil
	})
	q := newMailQueue(s, everyone, nil, slog.New(slog.DiscardHandler))
	t.Cleanup(q.close)
	_ = q.Send(context.Background(), "first@example.test", "s", "b")
	_ = q.Send(context.Background(), "second@example.test", "s", "b")
	q.wait()
	mu.Lock()
	defer mu.Unlock()
	if len(sent) != 1 || sent[0] != "second@example.test" {
		t.Fatalf("sent = %v, want the second link after the first panicked", sent)
	}
}

// TestMailQueueCloseIsBoundedAndStopsTheWorker: close drains within its
// deadline, then cancels the send in flight, abandons the rest, and the
// worker is gone — no goroutine outlives close (Astra, round 2, P2).
func TestMailQueueCloseIsBoundedAndStopsTheWorker(t *testing.T) {
	baseline := runtime.NumGoroutine()
	honours := funcSender(func(ctx context.Context, _ string) error {
		<-ctx.Done() // a relay that never answers, but obeys cancellation
		return ctx.Err()
	})
	q := newMailQueue(honours, everyone, nil, slog.New(slog.DiscardHandler))
	q.drainTimeout = 50 * time.Millisecond
	for i := 0; i < 3; i++ {
		_ = q.Send(context.Background(), fmt.Sprintf("p%d@example.test", i), "s", "b")
	}
	start := time.Now()
	q.close()
	if took := time.Since(start); took > 2*time.Second {
		t.Fatalf("close took %v, want about its 50ms deadline", took)
	}
	select {
	case <-q.done:
	default:
		t.Fatal("the worker was still running after close")
	}
	deadline := time.Now().Add(time.Second)
	for runtime.NumGoroutine() > baseline && time.Now().Before(deadline) {
		time.Sleep(5 * time.Millisecond)
	}
	if n := runtime.NumGoroutine(); n > baseline {
		t.Fatalf("%d goroutines after close, %d before: the queue leaked", n, baseline)
	}
}

// TestMailQueueCloseJoinsAnInProgressClose: close can be reached twice
// (a TestHandle's stop and a cleanup, say). The second close must wait
// for the first one's drain, not return because the queue is already
// marked closed, or a caller about to close the database or exit would
// leave an acknowledged link still at the relay (Astra on oficina/meet
// #25).
func TestMailQueueCloseJoinsAnInProgressClose(t *testing.T) {
	release, sending := make(chan struct{}), make(chan struct{})
	var sent atomic.Bool
	slow := funcSender(func(context.Context, string) error {
		close(sending)
		<-release
		sent.Store(true)
		return nil
	})
	q := newMailQueue(slow, everyone, nil, slog.New(slog.DiscardHandler))
	_ = q.Send(context.Background(), "paul@example.test", "s", "b")
	<-sending

	first := make(chan struct{})
	go func() { q.close(); close(first) }()
	for {
		q.mu.RLock()
		closed := q.closed
		q.mu.RUnlock()
		if closed {
			break
		}
		time.Sleep(time.Millisecond)
	}
	second := make(chan struct{})
	go func() { q.close(); close(second) }()
	select {
	case <-second:
		t.Fatal("a second close returned while the first was still draining")
	case <-time.After(100 * time.Millisecond):
	}
	close(release)
	for _, c := range []chan struct{}{first, second} {
		select {
		case <-c:
		case <-time.After(5 * time.Second):
			t.Fatal("close did not return after the send finished")
		}
	}
	if !sent.Load() {
		t.Fatal("the queued link was not sent before close returned")
	}
}

// mailWorkers counts goroutines running a sign-in mail queue's worker
// or log pump — the targeted leak check (goleak's idea, without the
// dependency) TestMain runs after the package's tests.
func mailWorkers() int {
	buf := make([]byte, 1<<22)
	stacks := string(buf[:runtime.Stack(buf, true)])
	return strings.Count(stacks, "(*mailQueue).run(") + strings.Count(stacks, "(*mailQueue).pump(")
}

// TestMailQueueCloseIsBoundedWithAStalledLogger: close reports through
// nothing that can block, and never waits on the log pump. With the log
// sink stalled — every write hangs — and a relay that never answers but
// honours cancellation, and WHILE THE SINK IS STILL STALLED:
//
//   - close returns within its bound;
//   - the mail worker has exited;
//   - at most the one log pump remains, blocked in its write.
//
// A write blocked inside the logger cannot be interrupted from Go; it
// can only finish or die with the process, and shutdown deliberately
// does not wait for it. So only once the sink is released must the pump
// exit and the goroutine count return to its baseline (Astra on #20,
// rounds 3 and 4). The test fails if close waited on the pump, or if
// the pump could hold the worker.
func TestMailQueueCloseIsBoundedWithAStalledLogger(t *testing.T) {
	waitFor := func(cond func() bool) bool {
		deadline := time.Now().Add(2 * time.Second)
		for !cond() && time.Now().Before(deadline) {
			time.Sleep(5 * time.Millisecond)
		}
		return cond()
	}
	// Baselines, not zero: another test's queue may still be shutting
	// down beside this one.
	baseline, workers := runtime.NumGoroutine(), mailWorkers()
	release := make(chan struct{})
	var once sync.Once
	free := func() { once.Do(func() { close(release) }) }
	t.Cleanup(free)
	var mu sync.Mutex
	var msgs []string
	honours := funcSender(func(ctx context.Context, _ string) error {
		<-ctx.Done()
		return ctx.Err()
	})
	q := newMailQueue(honours, everyone, nil, slog.New(stuckHandler{release, &mu, &msgs}))
	q.drainTimeout = 50 * time.Millisecond
	for i := 0; i < 2*mailQueueDepth; i++ { // drops too, so there is plenty to log
		_ = q.Send(context.Background(), fmt.Sprintf("p%d@example.test", i), "s", "b")
	}
	closed := make(chan struct{})
	go func() { q.close(); close(closed) }()
	select {
	case <-closed:
	case <-time.After(2 * time.Second):
		free()
		<-closed
		t.Fatal("close hung on a stalled log sink")
	}

	// The sink is still stalled.
	select {
	case <-q.done:
	default:
		t.Fatal("the worker was still running after close, held by the stalled log sink")
	}
	if !waitFor(func() bool { return runtime.NumGoroutine() <= baseline+1 }) {
		t.Fatalf("%d goroutines with the sink stalled, %d before: more than the one blocked log pump remains", runtime.NumGoroutine(), baseline)
	}
	if n := mailWorkers(); n > workers+1 {
		t.Fatalf("%d mail queue goroutines with the sink stalled, %d before: want at most the log pump more", n, workers)
	}

	// Released, the pump finishes its write and exits.
	free()
	if !waitFor(func() bool { return mailWorkers() <= workers && runtime.NumGoroutine() <= baseline }) {
		t.Fatalf("after release: %d mail queue goroutines, %d goroutines against %d before", mailWorkers(), runtime.NumGoroutine(), baseline)
	}
}

// buildForTest is build on the test's context, with the sign-in mail
// queue closed in cleanup: the context ending no longer closes it (see
// build), so whoever builds an app closes its queue.
func buildForTest(t *testing.T, d *db.DB, cfg Config) (*http.ServeMux, *app, error) {
	t.Helper()
	mux, a, err := build(t.Context(), d, cfg)
	if err == nil {
		t.Cleanup(a.signinMail.close)
	}
	return mux, a, err
}

// TestShutdownStopsTheMailWorker: the app's context ending (SIGTERM)
// stops streams and background work but NOT mail intake, because
// Server.Shutdown is still letting accepted requests finish and their
// links must still go out (Astra on #25, round 2). The queue takes links
// until its owner closes it (Runtime.Close, or a TestHandle's stop), and
// that close stops the worker, so nothing that closes its app leaves
// one running.
func TestShutdownStopsTheMailWorker(t *testing.T) {
	logger := slog.New(slog.DiscardHandler)
	d, err := db.Open(filepath.Join(t.TempDir(), "calls.db"), logger)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { d.Close() })
	ctx, cancel := context.WithCancel(context.Background())
	_, a, err := build(ctx, d, Config{Env: EnvDevelopment, Origin: "http://app.test", Logger: logger})
	if err != nil {
		t.Fatal(err)
	}
	q := a.signinMail
	t.Cleanup(q.close)
	cancel()
	time.Sleep(100 * time.Millisecond)
	q.mu.RLock()
	closed := q.closed
	q.mu.RUnlock()
	if closed {
		t.Fatal("the app's context ending closed the sign-in mail queue while requests may still be finishing")
	}
	q.close()
	select {
	case <-q.done:
	case <-time.After(2 * time.Second):
		t.Fatal("the sign-in mail worker was still running after close")
	}
}