// SPDX-License-Identifier: AGPL-3.0-or-later

package notify

import (
	"context"
	"fmt"
	"log"
	"math/rand"
	"time"

	"comail/internal/relaystore"
)

// DeliverFunc is the transport-agnostic dispatch the queue worker
// invokes for each claimed row. Implementations inspect
// Notification.Kind (see relaystore.NotificationKind*) and route to
// the appropriate sender — SMTP for member-facing mail, structured
// webhook for operator pings, and so on.
//
// Return nil on success; any error is recorded as last_error on the
// row and triggers exponential-backoff retry. After the configured
// max-attempts cap the row lands in the dead-letter table for operator
// review (see relaystore.ListDeadLetterNotifications).
type DeliverFunc func(ctx context.Context, n relaystore.Notification) error

// Default backoff parameters. Each failure reschedules the row for
// min(base * 2^attempts, cap) with ±JitterFraction jitter. The defaults
// ride through ~10 hours of outage before hitting the 10-attempt cap —
// long enough to survive a typical downstream incident, short enough
// that a genuinely-broken address doesn't clog the queue indefinitely.
const (
	DefaultBackoffBase    = 60 * time.Second
	DefaultBackoffCap     = time.Hour
	DefaultJitterFraction = 0.10 // ±10%
	DefaultBatchLimit     = 50
)

// QueueWorker drains pending_notifications on a fixed tick. One worker
// per process, please — running two against the same store will
// double-deliver because ClaimPendingNotifications does not lock rows.
//
// Lifecycle: NewQueueWorker → go worker.Run(ctx). Cancel the context
// to stop. Run returns ctx.Err() after a clean shutdown.
type QueueWorker struct {
	store          *relaystore.Store
	deliver        DeliverFunc
	tickInterval   time.Duration
	backoffBase    time.Duration
	backoffCap     time.Duration
	jitterFraction float64
	batchLimit     int
	rng            *rand.Rand
	// nowFn is swappable for deterministic tests. Defaults to time.Now.
	nowFn func() time.Time
}

// NewQueueWorker builds a worker with the default backoff + batch
// settings. The tickInterval controls how often the drain loop runs;
// values between 1s (development) and 30s (prod) are reasonable. Each
// tick claims up to DefaultBatchLimit rows so a backlog recovers
// without any manual intervention.
func NewQueueWorker(store *relaystore.Store, deliver DeliverFunc, tickInterval time.Duration) *QueueWorker {
	return &QueueWorker{
		store:          store,
		deliver:        deliver,
		tickInterval:   tickInterval,
		backoffBase:    DefaultBackoffBase,
		backoffCap:     DefaultBackoffCap,
		jitterFraction: DefaultJitterFraction,
		batchLimit:     DefaultBatchLimit,
		// Seeded from the wall clock — we don't care about
		// cryptographic randomness here, only about de-correlating
		// multiple workers (none currently, but future-proof).
		rng:   rand.New(rand.NewSource(time.Now().UnixNano())),
		nowFn: func() time.Time { return time.Now().UTC() },
	}
}

// NowFnForTest overrides the worker's clock. Exposed for tests in
// other packages that need to drive RunOnce across simulated backoff
// windows without wall-clock sleeps. Production code should leave the
// default in place.
func (q *QueueWorker) NowFnForTest(at time.Time) {
	q.nowFn = func() time.Time { return at }
}

// Run drains the queue on every tick until ctx is canceled. Returns
// ctx.Err() on cancel; never returns a nil error (use context.Canceled
// / context.DeadlineExceeded as the shutdown signal).
//
// Errors inside RunOnce (transport failures, DB hiccups) are logged
// and swallowed — the next tick retries from a clean slate. This keeps
// the goroutine long-lived across flaky upstream conditions.
func (q *QueueWorker) Run(ctx context.Context) error {
	if q.tickInterval <= 0 {
		return fmt.Errorf("queue worker: tickInterval must be > 0")
	}
	ticker := time.NewTicker(q.tickInterval)
	defer ticker.Stop()

	// First drain happens immediately so a process that's been down
	// for a while starts shipping queued rows as soon as it boots
	// (instead of waiting a full tick interval).
	q.RunOnce(ctx)

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-ticker.C:
			q.RunOnce(ctx)
		}
	}
}

// RunOnce performs a single drain pass: claim ready rows, deliver each,
// record outcome. Exposed so tests can drive the state machine
// deterministically without spinning a goroutine.
//
// Delivery errors are recorded per-row and do not stop the pass —
// if row A fails, row B in the same batch still gets its chance.
func (q *QueueWorker) RunOnce(ctx context.Context) {
	if ctx.Err() != nil {
		return
	}
	now := q.nowFn().Unix()
	claimed, err := q.store.ClaimPendingNotifications(ctx, now, q.batchLimit)
	if err != nil {
		log.Printf("notify.queue.claim_error: error=%v", err)
		return
	}
	for _, n := range claimed {
		// Re-check ctx between deliveries so a shutdown doesn't block
		// behind a slow SMTP call at the end of a large batch.
		if ctx.Err() != nil {
			return
		}
		q.deliverOne(ctx, n)
	}
}

// deliverOne invokes the delivery function for one row and records the
// outcome. On success the row is DELETEd from the queue; on failure
// the retry schedule applies (or the row dead-letters if the cap is
// reached — handled inside MarkNotificationAttempted).
func (q *QueueWorker) deliverOne(ctx context.Context, n relaystore.Notification) {
	err := q.deliver(ctx, n)
	if err == nil {
		if mErr := q.store.MarkNotificationAttempted(ctx, n.ID, true, 0, ""); mErr != nil {
			log.Printf("notify.queue.mark_success_error: id=%d error=%v", n.ID, mErr)
		}
		log.Printf("notify.queue.delivered: id=%d kind=%s recipient=%s attempts=%d", n.ID, n.Kind, n.Recipient, n.Attempts+1)
		return
	}

	// Compute the next-attempt delay from n.Attempts (pre-increment).
	// The store increments attempts atomically inside
	// MarkNotificationAttempted, so passing the current value is fine
	// — we care about the schedule for the retry that follows this
	// failure, which is attempts=n.Attempts+1 by the time the worker
	// re-reads this row.
	backoff := q.computeBackoffSeconds(n.Attempts)
	log.Printf("notify.queue.deliver_failed: id=%d kind=%s recipient=%s attempts=%d next_retry_in=%ds error=%v",
		n.ID, n.Kind, n.Recipient, n.Attempts+1, backoff, err)
	if mErr := q.store.MarkNotificationAttempted(ctx, n.ID, false, backoff, err.Error()); mErr != nil {
		log.Printf("notify.queue.mark_fail_error: id=%d error=%v", n.ID, mErr)
	}
}

// computeBackoffSeconds returns the wait in seconds before the row
// should be tried again given the current attempts count (0 = about to
// make the first retry). Uses exponential backoff with a cap and
// bounded jitter — both standard for queue schedulers.
//
// Algorithm: base * 2^attempts, clamped to cap, then multiplied by
// (1 ± jitterFraction * rand). Jitter is symmetric around the nominal
// value so the mean delay is preserved.
func (q *QueueWorker) computeBackoffSeconds(attempts int) int {
	delay := q.backoffBase
	for i := 0; i < attempts && delay < q.backoffCap; i++ {
		delay *= 2
	}
	if delay > q.backoffCap {
		delay = q.backoffCap
	}
	// Symmetric jitter: [1-f, 1+f]. Using the full 2f span avoids a
	// systematic skew either direction.
	if q.jitterFraction > 0 {
		jitter := (q.rng.Float64()*2 - 1) * q.jitterFraction
		delay = time.Duration(float64(delay) * (1 + jitter))
	}
	secs := int(delay / time.Second)
	if secs < 1 {
		secs = 1
	}
	return secs
}