package calls

import (
	"context"
	"fmt"
	"log/slog"
	"sync"
	"sync/atomic"
	"time"

	"amadan.net/rastrillo/rastrillo/mail"
)

// mailQueue is the sign-in mailer the auth plugin is given: Send queues
// the link and returns at once, and one background worker decides
// whether the address may have it and only then talks to SMTP.
//
// Both halves are the no-oracle guarantee, measured rather than
// reasoned about. When the send ran inside the request, a permitted
// address waited for the SMTP round trip before its `303 ?sent=1` and a
// denied one did not: 252 ms against 0.35 ms (Astra on oficina/slides
// #20). A response time is an answer like any other. So POST /signin
// does the same synchronous work for every address — the auth plugin's
// budgets, its link row, this enqueue — and the one admission-dependent
// step, the send, happens after the response, where nobody can time it.
//
// The request path is Send, and Send does no I/O and never blocks: not
// a send, not an admission check, not even a log line. Every address's
// link is queued, permitted or not, so how full the queue is never
// depends on admission either; the worker decides afterwards. A full
// queue drops the link and counts it in an atomic; the worker logs the
// count when it next runs. The person is answered exactly as when the
// queue is empty, because a different answer for a full queue would be
// one more thing to read. (Astra on #20, round 2.)
//
// The worker survives a panic in the mailer: it recovers per message,
// logs, and keeps draining. Delivery failures are logged, never
// surfaced: a failure only a permitted address can reach is the oracle
// again.
//
// Nothing but the log pump ever writes to the logger. The worker and
// close hand their lines to logf, which never blocks: a buffered
// channel, and a counted drop when it is full. A stalled log sink can
// therefore hold only the pump's one write — never a request, never the
// worker, never close (Astra on #20, round 3). The pump ends when the
// worker does.
//
// close drains what is queued within drainTimeout. Past it, the send
// in flight is cancelled through its context, everything still queued
// is abandoned and counted, and close waits a short grace for the
// worker to stop. A mailer that ignores its context is the only way a
// worker outlives close; that is counted in stuck and logged, best
// effort, too. close may be called more than once (the app's context
// ending, then main on its way out): a later call waits for the first
// one to finish, so whoever returns from close knows the drain is over.
type mailQueue struct {
	next   mail.Sender
	admits func(to string) bool
	// gate, when set, is read off the request's context at enqueue time
	// and travels with the link; the worker sends only if it said yes.
	gate   func(context.Context) bool
	logger *slog.Logger

	drainTimeout time.Duration
	sendTimeout  time.Duration

	mu      sync.RWMutex // held for reading by Send, for writing by close
	closed  bool
	jobs    chan mailJob
	dropped atomic.Int64
	pending sync.WaitGroup

	ctx    context.Context // cancelled when close gives up waiting
	cancel context.CancelFunc
	done   chan struct{}
	shut   chan struct{} // closed when the first close returns

	logMu      sync.RWMutex // logf reads logsClosed; the worker's exit sets it
	logsClosed bool
	logs       chan logLine // to the pump; logf never blocks on it
	unlogged   atomic.Int64 // lines handed over and not yet written
	lostLogs   atomic.Int64 // lines logf could not hand over
	stuck      atomic.Bool  // close gave up on a worker that ignored cancellation

	// refusedLog, when set, is the line the worker logs (through logf,
	// with a running count and never the address) for each link it
	// refuses.
	refusedLog string
	refusals   atomic.Int64
}

type logLine struct {
	msg  string
	args []any
}

type mailJob struct {
	to, subject, body string
	gated             bool
}

const (
	mailQueueDepth   = 64
	mailSendTimeout  = time.Minute
	mailDrainTimeout = 30 * time.Second
	mailAbandonGrace = 5 * time.Second
	mailLogDepth     = 32
)

func newMailQueue(next mail.Sender, admits func(string) bool, gate func(context.Context) bool, logger *slog.Logger) *mailQueue {
	return newMailQueueLogging(next, admits, gate, logger, "")
}

// newMailQueueLogging is newMailQueue with refusedLog set (see the
// field).
func newMailQueueLogging(next mail.Sender, admits func(string) bool, gate func(context.Context) bool, logger *slog.Logger, refusedLog string) *mailQueue {
	if logger == nil {
		logger = slog.Default()
	}
	ctx, cancel := context.WithCancel(context.Background())
	q := &mailQueue{next: next, admits: admits, gate: gate, logger: logger,
		drainTimeout: mailDrainTimeout, sendTimeout: mailSendTimeout,
		jobs: make(chan mailJob, mailQueueDepth), ctx: ctx, cancel: cancel,
		done: make(chan struct{}), shut: make(chan struct{}), logs: make(chan logLine, mailLogDepth), refusedLog: refusedLog}
	go q.pump()
	go q.run()
	return q
}

// Send queues one link. It never blocks, never does I/O and never
// fails: the request that called it answers the same way whatever
// happens next.
func (q *mailQueue) Send(ctx context.Context, to, subject, body string) error {
	job := mailJob{to: to, subject: subject, body: body, gated: true}
	if q.gate != nil {
		job.gated = q.gate(ctx)
	}
	q.mu.RLock()
	defer q.mu.RUnlock()
	if q.closed {
		q.dropped.Add(1)
		return nil
	}
	q.pending.Add(1)
	select {
	case q.jobs <- job:
	default:
		q.pending.Done()
		q.dropped.Add(1)
	}
	return nil
}

// logf hands one error line to the pump. It never blocks: when the
// pump is behind, the line is counted in lostLogs instead.
func (q *mailQueue) logf(msg string, args ...any) {
	q.logMu.RLock()
	defer q.logMu.RUnlock()
	if q.logsClosed {
		q.lostLogs.Add(1)
		return
	}
	q.unlogged.Add(1)
	select {
	case q.logs <- logLine{msg: msg, args: args}:
	default:
		q.unlogged.Add(-1)
		q.lostLogs.Add(1)
	}
}

// pump is the only goroutine that writes to the logger. It ends when
// the worker closes logs on its way out.
//
// Its limit, stated rather than engineered around: a write blocked
// inside the logger (a stalled sink) cannot be interrupted from Go. It
// can only finish, or die with the process. Closing logs does not reach
// it mid-write, so a stalled sink leaves this one goroutine behind
// until the sink moves. Nothing else waits on it: not Send, not the
// worker, not close.
func (q *mailQueue) pump() {
	for l := range q.logs {
		if n := q.lostLogs.Swap(0); n > 0 {
			q.logger.Error("calls: sign-in mail log lines lost to a slow log sink", "count", n)
		}
		q.logger.Error(l.msg, l.args...)
		q.unlogged.Add(-1)
	}
}

func (q *mailQueue) run() {
	defer close(q.done)
	defer func() { // after the last logf below; ends the pump
		q.logMu.Lock()
		q.logsClosed = true
		close(q.logs)
		q.logMu.Unlock()
	}()
	abandoned := 0
	for job := range q.jobs {
		q.reportDrops()
		if q.ctx.Err() != nil {
			abandoned++
		} else {
			q.deliver(job)
		}
		q.pending.Done()
	}
	q.reportDrops()
	if abandoned > 0 {
		q.logf("calls: sign-in links abandoned at shutdown", "count", abandoned)
	}
}

// reportDrops logs the links Send had to drop since the last report.
// Here, on the worker, so the request path never writes a log line.
func (q *mailQueue) reportDrops() {
	if n := q.dropped.Swap(0); n > 0 {
		q.logf("calls: sign-in mail queue full; links dropped", "count", n)
	}
}

// deliver sends one link if it may go. A panic in the admission check
// or the mailer is this message's failure, not the process's.
func (q *mailQueue) deliver(job mailJob) {
	defer func() {
		if p := recover(); p != nil {
			q.logf("calls: sign-in mail worker recovered from a panic", "panic", fmt.Sprint(p))
		}
	}()
	if !job.gated || !q.admits(job.to) {
		// Not admitted: the link the auth plugin stored is never
		// delivered, so it is never redeemed. No address in the log.
		if q.refusedLog != "" {
			q.logf(q.refusedLog, "stage", "signin.gate", "refusals_since_boot", q.refusals.Add(1))
		}
		return
	}
	ctx, cancel := context.WithTimeout(q.ctx, q.sendTimeout)
	defer cancel()
	if err := q.next.Send(ctx, job.to, job.subject, job.body); err != nil {
		q.logf("calls: send sign-in link", "err", err)
	}
}

// wait blocks until every link queued so far has been sent or refused,
// and every line logged about them has been written. Tests use it to
// read the mailer, and the log, afterwards.
func (q *mailQueue) wait() {
	q.pending.Wait()
	for q.unlogged.Load() > 0 {
		time.Sleep(time.Millisecond)
	}
}

// close stops taking links and drains what is queued, within
// drainTimeout; then it cancels, abandons the rest, and stops the
// worker (see the type's comment). It never waits on the log pump:
// shutdown deliberately does not wait for a write blocked inside the
// logger, which can only finish or die with the process (see pump).
//
// A second close joins the first: it returns when that one does, never
// earlier, so main can wait for a drain the app's context began.
func (q *mailQueue) close() {
	q.mu.Lock()
	if q.closed {
		q.mu.Unlock()
		<-q.shut
		return
	}
	q.closed = true
	close(q.jobs)
	q.mu.Unlock()
	defer close(q.shut)
	defer q.cancel()
	select {
	case <-q.done:
		return
	case <-time.After(q.drainTimeout):
	}
	q.cancel()
	select {
	case <-q.done:
	case <-time.After(mailAbandonGrace):
		// Counted, not written: logging here could hang close on the
		// very sink that may be the trouble. The worker is still running,
		// so logf is still safe to call, and never blocks.
		q.stuck.Store(true)
		q.logf("calls: sign-in mail worker ignored cancellation; left running")
	}
}

// mayMail is the queue's gate: gateBegin's mayRequestLink answer for
// the request that asked for the link (ruling 12's budgets), carried
// through Begin on the request context. A context lookup, no I/O.
func mayMail(ctx context.Context) bool {
	may, _ := ctx.Value(mayMailKey{}).(bool)
	return may
}

// queueAttacher is a test mailer (another package's, so the method is
// exported) that wants to wait for the queue in front of it before it
// is read (callstest's recordingMailer).
type queueAttacher interface{ AttachSigninQueue(wait func()) }