cairnquire/apps/server/internal/email/queue_test.go

package email

import (
	"context"
	"database/sql"
	"errors"
	"strings"
	"testing"
	"time"

	"github.com/tim/cairnquire/apps/server/internal/config"
	"github.com/tim/cairnquire/apps/server/internal/database"
)

func newTestDB(t *testing.T) *sql.DB {
	t.Helper()
	cfg := config.DatabaseConfig{Path: t.TempDir() + "/test.sqlite"}
	handle, err := database.Open(context.Background(), cfg)
	if err != nil {
		t.Fatalf("open db: %v", err)
	}
	t.Cleanup(func() { _ = handle.Close() })
	db := handle.SQL()
	if err := database.ApplyMigrations(context.Background(), db); err != nil {
		t.Fatalf("migrate: %v", err)
	}
	// Seed a user + notification so the queue's notification_id FK is
	// satisfiable and MarkNotificationEmailed can update a real row.
	_, err = db.Exec(`INSERT INTO users (id, email, display_name, role, created_at, last_seen_at) VALUES ('user:alice@example.com', 'alice@example.com', 'Alice', 'admin', ?, ?)`,
		time.Now().UTC().Format(time.RFC3339), time.Now().UTC().Format(time.RFC3339))
	if err != nil {
		t.Fatalf("seed user: %v", err)
	}
	_, err = db.Exec(`INSERT INTO notifications (id, user_id, type, resource_type, resource_id, message, created_at) VALUES (?, 'user:alice@example.com', 'file_changed', 'document', 'doc:test', 'test', ?)`,
		"notif:test-1", time.Now().UTC().Format(time.RFC3339))
	if err != nil {
		t.Fatalf("seed notification: %v", err)
	}
	return db
}

func TestQueue_EnqueueAndPendingCount(t *testing.T) {
	db := newTestDB(t)
	q := NewQueue(db, nil, testLogger(t))

	id, err := q.Enqueue(context.Background(), "notif:test-1", "alice@example.com", "Subject", "Body")
	if err != nil {
		t.Fatalf("Enqueue: %v", err)
	}
	if id == "" {
		t.Fatal("expected non-empty id")
	}
	count, err := q.PendingCount(context.Background())
	if err != nil {
		t.Fatalf("PendingCount: %v", err)
	}
	if count != 1 {
		t.Errorf("pending = %d, want 1", count)
	}
}

func TestQueue_EnqueueValidation(t *testing.T) {
	db := newTestDB(t)
	q := NewQueue(db, nil, testLogger(t))

	// Empty notification_id is allowed (the column is nullable).
	if _, err := q.Enqueue(context.Background(), "", "alice@example.com", "s", "b"); err != nil {
		t.Fatalf("empty notification_id should be allowed: %v", err)
	}
	if _, err := q.Enqueue(context.Background(), "notif:test-1", "", "s", "b"); err == nil {
		t.Fatal("expected error for empty to")
	}
	if _, err := q.Enqueue(context.Background(), "notif:test-1", "alice@example.com", "", "b"); err == nil {
		t.Fatal("expected error for empty subject")
	}
}

type recordingSender struct {
	sends []recordedSend
	err   error
}

type recordedSend struct {
	to      []string
	subject string
	body    string
}

func (r *recordingSender) Send(ctx context.Context, to []string, subject, body string) error {
	if r.err != nil {
		return r.err
	}
	r.sends = append(r.sends, recordedSend{to: to, subject: subject, body: body})
	return nil
}

func TestQueue_DeliverSuccess(t *testing.T) {
	db := newTestDB(t)
	sender := &recordingSender{}
	q := NewQueue(db, sender, testLogger(t))
	q.SetPollInterval(20 * time.Millisecond)

	if _, err := q.Enqueue(context.Background(), "notif:test-1", "alice@example.com", "Hello", "World"); err != nil {
		t.Fatalf("Enqueue: %v", err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	go q.Run(ctx)

	if err := waitFor(ctx, func() bool {
		n, _ := q.PendingCount(ctx)
		return n == 0
	}); err != nil {
		t.Fatalf("email never sent: %v", err)
	}

	if len(sender.sends) != 1 {
		t.Fatalf("sends = %d, want 1", len(sender.sends))
	}
	if sender.sends[0].subject != "Hello" {
		t.Errorf("subject = %q", sender.sends[0].subject)
	}
	if sender.sends[0].body != "World" {
		t.Errorf("body = %q", sender.sends[0].body)
	}

	// The linked notification should now have emailed_at stamped.
	var emailedAt string
	if err := db.QueryRow(`SELECT COALESCE(emailed_at, '') FROM notifications WHERE id = ?`, "notif:test-1").Scan(&emailedAt); err != nil {
		t.Fatalf("query emailed_at: %v", err)
	}
	if emailedAt == "" {
		t.Error("expected notifications.emailed_at to be set after send")
	}
}

func TestQueue_RetryThenSucceed(t *testing.T) {
	db := newTestDB(t)
	// Fail twice, then succeed.
	sender := &flakySender{failTimes: 2}
	q := NewQueue(db, sender, testLogger(t))
	q.SetPollInterval(15 * time.Millisecond)
	q.SetMaxBackoff(50 * time.Millisecond)

	if _, err := q.Enqueue(context.Background(), "notif:test-1", "alice@example.com", "Retry", "Body"); err != nil {
		t.Fatalf("Enqueue: %v", err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	go q.Run(ctx)

	if err := waitFor(ctx, func() bool {
		return sender.successCount >= 1
	}); err != nil {
		t.Fatalf("email never succeeded: successCount=%d attempts=%d: %v", sender.successCount, sender.attempts, err)
	}

	// Verify the row is marked sent.
	var status string
	if err := db.QueryRow(`SELECT status FROM email_queue WHERE to_address = ?`, "alice@example.com").Scan(&status); err != nil {
		t.Fatalf("query status: %v", err)
	}
	if status != "sent" {
		t.Errorf("status = %q, want sent", status)
	}
}

func TestQueue_PermanentFailure(t *testing.T) {
	db := newTestDB(t)
	sender := &recordingSender{err: errors.New("permanent Postmark rejection")}
	q := NewQueue(db, sender, testLogger(t))
	q.SetPollInterval(15 * time.Millisecond)
	q.SetMaxBackoff(20 * time.Millisecond)

	if _, err := q.Enqueue(context.Background(), "notif:test-1", "alice@example.com", "Fail", "Body"); err != nil {
		t.Fatalf("Enqueue: %v", err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	go q.Run(ctx)

	if err := waitFor(ctx, func() bool {
		failed, _ := q.ListFailed(ctx, 10)
		return len(failed) == 1
	}); err != nil {
		failed, _ := q.ListFailed(ctx, 10)
		t.Fatalf("expected 1 failed row, got %d: %v", len(failed), err)
	}

	failed, _ := q.ListFailed(ctx, 10)
	if !strings.Contains(failed[0].LastError, "permanent") {
		t.Errorf("last_error = %q", failed[0].LastError)
	}
}

func TestQueue_NilSenderLeavesPending(t *testing.T) {
	db := newTestDB(t)
	q := NewQueue(db, nil, testLogger(t))
	q.SetPollInterval(15 * time.Millisecond)

	if _, err := q.Enqueue(context.Background(), "notif:test-1", "alice@example.com", "Pending", "Body"); err != nil {
		t.Fatalf("Enqueue: %v", err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
	defer cancel()
	go q.Run(ctx)

	<-ctx.Done()
	count, _ := q.PendingCount(context.Background())
	if count != 1 {
		t.Errorf("expected row to stay pending with nil sender, got %d", count)
	}
}

// flakySender fails the first N attempts then succeeds.
type flakySender struct {
	failTimes    int
	attempts     int
	successCount int
}

func (f *flakySender) Send(ctx context.Context, to []string, subject, body string) error {
	f.attempts++
	if f.attempts <= f.failTimes {
		return errors.New("transient failure")
	}
	f.successCount++
	return nil
}

func waitFor(ctx context.Context, cond func() bool) error {
	ticker := time.NewTicker(10 * time.Millisecond)
	defer ticker.Stop()
	for {
		if cond() {
			return nil
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-ticker.C:
		}
	}
}