package email

import (
	"context"
	"database/sql"
	"fmt"
	"log/slog"
	"math"
	"time"
)

// Queue stores outgoing emails durably and dispatches them via a Sender.
// It is safe for concurrent use; claim uses an atomic UPDATE ... WHERE
// status='pending' to lease rows without races.
type Queue struct {
	db     *sql.DB
	sender Sender
	logger *slog.Logger

	pollInterval time.Duration
	maxBackoff   time.Duration
}

// QueueRow is the on-disk representation of a queued email.
type QueueRow struct {
	ID             string
	NotificationID string
	ToAddress      string
	Subject        string
	Body           string
	Status         string
	Attempts       int
	MaxAttempts    int
	LastError      string
	NextAttemptAt  time.Time
	SentAt         *time.Time
	CreatedAt      time.Time
}

// NewQueue constructs a queue. sender may be nil; in that case emails stay
// pending forever (useful for tests that only inspect enqueue state).
func NewQueue(db *sql.DB, sender Sender, logger *slog.Logger) *Queue {
	if logger == nil {
		logger = slog.Default()
	}
	return &Queue{
		db:           db,
		sender:       sender,
		logger:       logger,
		pollInterval: 10 * time.Second,
		maxBackoff:   30 * time.Minute,
	}
}

// SetPollInterval overrides the worker poll interval. Must be called before
// Run. Useful for tests.
func (q *Queue) SetPollInterval(d time.Duration) {
	if d > 0 {
		q.pollInterval = d
	}
}

// SetMaxBackoff overrides the maximum retry backoff.
func (q *Queue) SetMaxBackoff(d time.Duration) {
	if d > 0 {
		q.maxBackoff = d
	}
}

// Enqueue inserts a new outgoing email. notificationID may be empty. The
// email is immediately eligible for delivery (next_attempt_at = now).
func (q *Queue) Enqueue(ctx context.Context, notificationID, to, subject, body string) (string, error) {
	if to == "" {
		return "", fmt.Errorf("enqueue: to address is required")
	}
	if subject == "" {
		return "", fmt.Errorf("enqueue: subject is required")
	}
	id, err := randomQueueID()
	if err != nil {
		return "", err
	}
	now := time.Now().UTC()
	if _, err := q.db.ExecContext(ctx, `
		INSERT INTO email_queue (id, notification_id, to_address, subject, body, status, attempts, max_attempts, next_attempt_at, created_at)
		VALUES (?, ?, ?, ?, ?, 'pending', 0, 3, ?, ?)
	`, id, nullableString(notificationID), to, subject, body, formatTime(now), formatTime(now)); err != nil {
		return "", fmt.Errorf("enqueue email: %w", err)
	}
	return id, nil
}

// MarkNotificationEmailed stamps notifications.emailed_at for the given
// notification id. Called by the worker once the email is dispatched.
func (q *Queue) MarkNotificationEmailed(ctx context.Context, notificationID string) error {
	if notificationID == "" {
		return nil
	}
	_, err := q.db.ExecContext(ctx, `
		UPDATE notifications SET emailed_at = ? WHERE id = ?
	`, formatTime(time.Now().UTC()), notificationID)
	if err != nil {
		return fmt.Errorf("mark notification emailed: %w", err)
	}
	return nil
}

// Run starts the worker loop. It blocks until ctx is cancelled.
func (q *Queue) Run(ctx context.Context) {
	q.logger.Info("email queue worker started", "poll_interval", q.pollInterval)
	ticker := time.NewTicker(q.pollInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			q.logger.Info("email queue worker stopped")
			return
		case <-ticker.C:
			if err := q.processBatch(ctx); err != nil {
				q.logger.Warn("email queue batch failed", "error", err)
			}
		}
	}
}

// processBatch claims and sends up to 25 pending emails per tick.
func (q *Queue) processBatch(ctx context.Context) error {
	if q.sender == nil {
		// No sender configured; leave rows pending so a future sender can
		// pick them up once wired.
		return nil
	}
	rows, err := q.claimBatch(ctx, 25)
	if err != nil {
		return fmt.Errorf("claim batch: %w", err)
	}
	for _, row := range rows {
		if err := q.deliver(ctx, row); err != nil {
			q.logger.Warn("deliver email failed", "id", row.ID, "error", err)
		}
	}
	return nil
}

// claimBatch atomically leases pending rows by flipping their status to
// 'pending' but bumping attempts so a concurrent worker cannot re-claim
// them until the next_attempt_at expires. We use a single UPDATE with a
// subquery to keep this race-free under SQLite.
func (q *Queue) claimBatch(ctx context.Context, limit int) ([]QueueRow, error) {
	tx, err := q.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, fmt.Errorf("begin claim: %w", err)
	}
	defer func() { _ = tx.Rollback() }()

	rows, err := tx.QueryContext(ctx, `
		SELECT id, COALESCE(notification_id, ''), to_address, subject, body, status, attempts, max_attempts,
		       COALESCE(last_error, ''), next_attempt_at, COALESCE(sent_at, ''), created_at
		FROM email_queue
		WHERE status = 'pending' AND next_attempt_at <= ?
		ORDER BY next_attempt_at ASC
		LIMIT ?
	`, formatTime(time.Now().UTC()), limit)
	if err != nil {
		return nil, fmt.Errorf("query pending: %w", err)
	}

	var claimed []QueueRow
	for rows.Next() {
		row, err := scanQueueRow(rows)
		if err != nil {
			rows.Close()
			return nil, err
		}
		claimed = append(claimed, row)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return nil, err
	}

	for _, row := range claimed {
		if _, err := tx.ExecContext(ctx, `
			UPDATE email_queue SET attempts = attempts + 1 WHERE id = ?
		`, row.ID); err != nil {
			return nil, fmt.Errorf("lease row %s: %w", row.ID, err)
		}
	}

	if err := tx.Commit(); err != nil {
		return nil, fmt.Errorf("commit claim: %w", err)
	}
	return claimed, nil
}

// deliver attempts one email. On success it marks the row sent and stamps
// the linked notification. On failure it schedules a retry with exponential
// backoff or marks the row failed if attempts are exhausted.
func (q *Queue) deliver(ctx context.Context, row QueueRow) error {
	err := q.sender.Send(ctx, []string{row.ToAddress}, row.Subject, row.Body)
	if err == nil {
		if _, err := q.db.ExecContext(ctx, `
			UPDATE email_queue SET status = 'sent', sent_at = ?, last_error = '' WHERE id = ?
		`, formatTime(time.Now().UTC()), row.ID); err != nil {
			return fmt.Errorf("mark sent: %w", err)
		}
		if err := q.MarkNotificationEmailed(ctx, row.NotificationID); err != nil {
			q.logger.Warn("mark notification emailed", "id", row.NotificationID, "error", err)
		}
		return nil
	}

	// Failure: retry or give up.
	if row.Attempts >= row.MaxAttempts {
		if _, ferr := q.db.ExecContext(ctx, `
			UPDATE email_queue SET status = 'failed', last_error = ? WHERE id = ?
		`, truncateErr(err.Error()), row.ID); ferr != nil {
			return fmt.Errorf("mark failed: %w", ferr)
		}
		q.logger.Warn("email permanently failed", "id", row.ID, "attempts", row.Attempts, "error", err)
		return nil
	}

	backoff := q.backoff(row.Attempts)
	next := time.Now().UTC().Add(backoff)
	if _, err := q.db.ExecContext(ctx, `
		UPDATE email_queue SET status = 'pending', next_attempt_at = ?, last_error = ? WHERE id = ?
	`, formatTime(next), truncateErr(err.Error()), row.ID); err != nil {
		return fmt.Errorf("schedule retry: %w", err)
	}

	// TemporaryError (Postmark 5xx) is retried with the same backoff as
	// other errors; the max-attempts cap still bounds total work.
	q.logger.Info("email retry scheduled", "id", row.ID, "attempt", row.Attempts, "backoff", backoff, "error", err)
	return nil
}

func (q *Queue) backoff(attempts int) time.Duration {
	// 2^attempts seconds, capped: 2s, 4s, 8s, 16s, ... up to maxBackoff.
	d := time.Duration(math.Pow(2, float64(attempts))) * time.Second
	if d > q.maxBackoff {
		d = q.maxBackoff
	}
	if d < time.Second {
		d = time.Second
	}
	return d
}

// PendingCount returns the number of rows in pending status, primarily for
// tests and operator dashboards.
func (q *Queue) PendingCount(ctx context.Context) (int, error) {
	var count int
	err := q.db.QueryRowContext(ctx, `SELECT COUNT(1) FROM email_queue WHERE status = 'pending'`).Scan(&count)
	if err != nil {
		return 0, err
	}
	return count, nil
}

// ListFailed returns rows that have exhausted their retries, for operator
// inspection.
func (q *Queue) ListFailed(ctx context.Context, limit int) ([]QueueRow, error) {
	if limit <= 0 {
		limit = 50
	}
	rows, err := q.db.QueryContext(ctx, `
		SELECT id, COALESCE(notification_id, ''), to_address, subject, body, status, attempts, max_attempts,
		       COALESCE(last_error, ''), next_attempt_at, COALESCE(sent_at, ''), created_at
		FROM email_queue
		WHERE status = 'failed'
		ORDER BY created_at DESC
		LIMIT ?
	`, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var out []QueueRow
	for rows.Next() {
		row, err := scanQueueRow(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, row)
	}
	return out, rows.Err()
}

func scanQueueRow(rows *sql.Rows) (QueueRow, error) {
	var r QueueRow
	var sentAt, lastError string
	var nextAttempt, createdAt string
	if err := rows.Scan(&r.ID, &r.NotificationID, &r.ToAddress, &r.Subject, &r.Body, &r.Status,
		&r.Attempts, &r.MaxAttempts, &lastError, &nextAttempt, &sentAt, &createdAt); err != nil {
		return QueueRow{}, fmt.Errorf("scan queue row: %w", err)
	}
	r.LastError = lastError
	var err error
	r.NextAttemptAt, err = time.Parse(time.RFC3339, nextAttempt)
	if err != nil {
		return QueueRow{}, fmt.Errorf("parse next_attempt_at: %w", err)
	}
	r.CreatedAt, err = time.Parse(time.RFC3339, createdAt)
	if err != nil {
		return QueueRow{}, fmt.Errorf("parse created_at: %w", err)
	}
	if sentAt != "" {
		t, err := time.Parse(time.RFC3339, sentAt)
		if err == nil {
			r.SentAt = &t
		}
	}
	return r, nil
}

func formatTime(t time.Time) string {
	return t.UTC().Format(time.RFC3339)
}

func nullableString(s string) any {
	if s == "" {
		return nil
	}
	return s
}

func truncateErr(s string) string {
	if len(s) > 512 {
		return s[:512] + "..."
	}
	return s
}

// randomQueueID produces an id with enough entropy to avoid collisions. We
// keep a small helper in-package to avoid an import cycle on auth.
func randomQueueID() (string, error) {
	id, err := randomToken(18)
	if err != nil {
		return "", err
	}
	return "email:" + id, nil
}