web/mailsqlite/queue_test.go

// SPDX-License-Identifier: MPL-2.0

package mailsqlite

import (
	"bytes"
	"context"
	"database/sql"
	"errors"
	"fmt"
	"path/filepath"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"gamertan.com/web/mail"
	_ "modernc.org/sqlite"
)

type testClock struct{ seconds atomic.Int64 }

func (clock *testClock) now() time.Time { return time.Unix(clock.seconds.Load(), 0).UTC() }
func (clock *testClock) advance(duration time.Duration) {
	clock.seconds.Add(int64(duration / time.Second))
}

type transportFunc func(context.Context, mail.Message) error

func (send transportFunc) Send(ctx context.Context, message mail.Message) error {
	return send(ctx, message)
}

func fixture(t *testing.T, capacity int) (*Queue, *sql.DB, *testClock) {
	t.Helper()
	db, err := sql.Open("sqlite", "file:"+filepath.ToSlash(filepath.Join(t.TempDir(), "mail.sqlite"))+"?_pragma=busy_timeout(5000)&_pragma=foreign_keys(1)&_pragma=journal_mode(WAL)")
	if err != nil {
		t.Fatal(err)
	}
	db.SetMaxOpenConns(8)
	t.Cleanup(func() { db.Close() })
	tx, err := db.Begin()
	if err != nil {
		t.Fatal(err)
	}
	if err = CreateSchema(context.Background(), tx); err != nil {
		t.Fatal(err)
	}
	if err = CreateSchema(context.Background(), tx); err != nil {
		t.Fatal(err)
	}
	if err = tx.Commit(); err != nil {
		t.Fatal(err)
	}
	clock := &testClock{}
	clock.seconds.Store(time.Date(2026, 9, 11, 6, 0, 0, 0, time.UTC).Unix())
	queue, err := New(db, Options{EncryptionKey: bytes.Repeat([]byte{17}, 32), MaxPending: capacity, Now: clock.now})
	if err != nil {
		t.Fatal(err)
	}
	return queue, db, clock
}

func messageAt(clock *testClock, number int) mail.Message {
	return mail.Message{ID: fmt.Sprintf("mail_message_%016d", number), To: "recipient@example.test", Subject: "Confirm your email", Text: "Private recovery token: only-in-ciphertext-123", CreatedAt: clock.now()}
}

func enqueue(t *testing.T, queue *Queue, clock *testClock, number int) mail.Message {
	t.Helper()
	message := messageAt(clock, number)
	if err := queue.Enqueue(context.Background(), message, clock.now().Add(time.Hour)); err != nil {
		t.Fatal(err)
	}
	return message
}

func storedPayload(t *testing.T, db *sql.DB, id string) []byte {
	t.Helper()
	var sealed []byte
	if err := db.QueryRow(`SELECT payload FROM gwf_mail_outbox WHERE id=?`, id).Scan(&sealed); err != nil {
		t.Fatal(err)
	}
	return sealed
}

func TestTransactionalEncryptionAndDeduplication(t *testing.T) {
	queue, db, clock := fixture(t, 1)
	ctx := context.Background()
	message := messageAt(clock, 1)
	expires := clock.now().Add(time.Hour)
	tx, err := db.Begin()
	if err != nil {
		t.Fatal(err)
	}
	if _, err = tx.Exec(`CREATE TABLE domain_change(id TEXT)`); err != nil {
		t.Fatal(err)
	}
	if _, err = tx.Exec(`INSERT INTO domain_change VALUES ('changed')`); err != nil {
		t.Fatal(err)
	}
	if err = queue.EnqueueTx(ctx, tx, message, expires); err != nil {
		t.Fatal(err)
	}
	if err = tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	var count int
	if err = db.QueryRow(`SELECT COUNT(*) FROM gwf_mail_outbox`).Scan(&count); err != nil || count != 0 {
		t.Fatalf("rollback count=%d err=%v", count, err)
	}
	if err = db.QueryRow(`SELECT COUNT(*) FROM sqlite_master WHERE name='domain_change'`).Scan(&count); err != nil || count != 0 {
		t.Fatalf("domain rollback count=%d err=%v", count, err)
	}
	if err = queue.Enqueue(ctx, message, expires); err != nil {
		t.Fatal(err)
	}
	sealed := storedPayload(t, db, message.ID)
	if len(sealed) == 0 || bytes.Contains(sealed, []byte(message.To)) || bytes.Contains(sealed, []byte(message.Text)) {
		t.Fatal("payload is not encrypted")
	}
	if err = queue.Enqueue(ctx, message, expires); err != nil {
		t.Fatalf("idempotent at capacity: %v", err)
	}
	changed := message
	changed.Text = "Different content"
	if err = queue.Enqueue(ctx, changed, expires); !errors.Is(err, ErrConflict) {
		t.Fatalf("conflicting content: %v", err)
	}
	if err = queue.Enqueue(ctx, message, expires.Add(time.Minute)); !errors.Is(err, ErrConflict) {
		t.Fatalf("conflicting expiry: %v", err)
	}
	if err = queue.Enqueue(ctx, messageAt(clock, 2), expires); !errors.Is(err, ErrFull) {
		t.Fatalf("capacity: %v", err)
	}
	var calls int
	send := transportFunc(func(_ context.Context, got mail.Message) error {
		calls++
		if got.ID != message.ID || got.To != message.To || got.Text != message.Text {
			t.Fatal("message changed")
		}
		return nil
	})
	record, work, err := queue.ProcessOne(ctx, send)
	if err != nil || !work || record.State != "accepted" || record.Attempts != 1 || !record.NextAttemptAt.IsZero() {
		t.Fatalf("accept: %+v work=%v err=%v", record, work, err)
	}
	if storedPayload(t, db, message.ID) != nil {
		t.Fatal("accepted payload retained")
	}
	if err = queue.Enqueue(ctx, message, expires); err != nil {
		t.Fatal(err)
	}
	if _, work, err = queue.ProcessOne(ctx, send); err != nil || work || calls != 1 {
		t.Fatalf("completed ID sent again: calls=%d work=%v err=%v", calls, work, err)
	}
	records, err := queue.Recent(ctx, 10)
	if err != nil || len(records) != 1 || records[0].State != "accepted" || !records[0].NextAttemptAt.IsZero() {
		t.Fatalf("metadata: %+v %v", records, err)
	}
	if err = queue.Enqueue(ctx, messageAt(clock, 2), expires); err != nil {
		t.Fatalf("released capacity: %v", err)
	}
}

func TestConcurrentCapacity(t *testing.T) {
	queue, _, clock := fixture(t, 1)
	var group sync.WaitGroup
	var wins, full atomic.Int32
	for number := range 8 {
		group.Go(func() {
			err := queue.Enqueue(context.Background(), messageAt(clock, number), clock.now().Add(time.Hour))
			if err == nil {
				wins.Add(1)
			} else if errors.Is(err, ErrFull) {
				full.Add(1)
			} else {
				t.Errorf("enqueue: %v", err)
			}
		})
	}
	group.Wait()
	if wins.Load() != 1 || full.Load() != 7 {
		t.Fatalf("wins=%d full=%d", wins.Load(), full.Load())
	}
}

func TestRetryBoundsAndSafeDiagnostics(t *testing.T) {
	queue, db, clock := fixture(t, 0)
	message := enqueue(t, queue, clock, 1)
	send := transportFunc(func(context.Context, mail.Message) error {
		return &mail.Error{Stage: "recipient", Code: 450, Retryable: true}
	})
	for attempt := 1; attempt <= maxAttempts; attempt++ {
		record, work, err := queue.ProcessOne(context.Background(), send)
		if err != nil || !work || record.Attempts != attempt || record.FailureStage != "recipient" || record.SMTPCode != 450 {
			t.Fatalf("attempt %d: %+v %v %v", attempt, record, work, err)
		}
		if attempt == maxAttempts {
			if record.State != "failed" || storedPayload(t, db, message.ID) != nil {
				t.Fatal("exhausted payload not cleared")
			}
			break
		}
		delay := time.Minute * time.Duration(1<<(attempt-1))
		if record.State != "queued" || !record.NextAttemptAt.Equal(clock.now().Add(delay)) {
			t.Fatalf("retry schedule: %+v", record)
		}
		if _, work, err = queue.ProcessOne(context.Background(), send); err != nil || work {
			t.Fatalf("early retry: %v %v", work, err)
		}
		clock.advance(delay)
	}
	enqueue(t, queue, clock, 2)
	record, _, err := queue.ProcessOne(context.Background(), transportFunc(func(context.Context, mail.Message) error {
		return &mail.Error{Stage: "secret-recipient@example.test", Code: 9999, Retryable: false}
	}))
	if err != nil || record.State != "failed" || record.FailureStage != "transport" || record.SMTPCode != 0 {
		t.Fatalf("unsafe diagnostic: %+v %v", record, err)
	}
	enqueue(t, queue, clock, 3)
	record, _, err = queue.ProcessOne(context.Background(), transportFunc(func(context.Context, mail.Message) error { return errors.New("contains private token and address") }))
	if err != nil || record.State != "queued" || record.FailureStage != "transport" {
		t.Fatalf("raw transport error: %+v %v", record, err)
	}
}

func TestExpiryAndCiphertextRecovery(t *testing.T) {
	queue, db, clock := fixture(t, 0)
	message := enqueue(t, queue, clock, 1)
	sealed := storedPayload(t, db, message.ID)
	wrong, err := New(db, Options{EncryptionKey: bytes.Repeat([]byte{18}, 32), Now: clock.now})
	if err != nil {
		t.Fatal(err)
	}
	var calls int
	send := transportFunc(func(context.Context, mail.Message) error { calls++; return nil })
	record, work, err := wrong.ProcessOne(context.Background(), send)
	if err != nil || !work || record.State != "queued" || record.FailureStage != "payload" || record.Attempts != 0 || calls != 0 {
		t.Fatalf("wrong key: %+v %v %v calls=%d", record, work, err, calls)
	}
	if !bytes.Equal(sealed, storedPayload(t, db, message.ID)) {
		t.Fatal("key error destroyed pending ciphertext")
	}
	clock.advance(5 * time.Minute)
	record, work, err = queue.ProcessOne(context.Background(), send)
	if err != nil || !work || record.State != "accepted" || record.Attempts != 1 || calls != 1 {
		t.Fatalf("key recovery: %+v %v %v calls=%d", record, work, err, calls)
	}
	message = enqueue(t, queue, clock, 2)
	sealed = storedPayload(t, db, message.ID)
	sealed[len(sealed)-1] ^= 1
	if _, err = db.Exec(`UPDATE gwf_mail_outbox SET payload=? WHERE id=?`, sealed, message.ID); err != nil {
		t.Fatal(err)
	}
	record, _, err = queue.ProcessOne(context.Background(), send)
	if err != nil || record.FailureStage != "payload" || calls != 1 {
		t.Fatalf("tamper sent: %+v %v calls=%d", record, err, calls)
	}
	clock.advance(time.Hour)
	if _, work, err = queue.ProcessOne(context.Background(), send); err != nil || work || calls != 1 {
		t.Fatalf("expired sent: work=%v err=%v calls=%d", work, err, calls)
	}
	if storedPayload(t, db, message.ID) != nil {
		t.Fatal("expired corrupt payload retained")
	}
	if _, err = queue.open(message.ID, sealed[:3], clock.now().Unix()); !errors.Is(err, ErrInvalid) {
		t.Fatal("short ciphertext accepted")
	}
	valid := enqueue(t, queue, clock, 3)
	if _, err = queue.open("another_message_identity", storedPayload(t, db, valid.ID), clock.now().Add(time.Hour).Unix()); !errors.Is(err, ErrInvalid) {
		t.Fatal("AAD identity not bound")
	}
	if _, err = queue.open(valid.ID, storedPayload(t, db, valid.ID), clock.now().Add(2*time.Hour).Unix()); !errors.Is(err, ErrInvalid) {
		t.Fatal("expiry not bound")
	}
}

func TestLeaseIsolationAndLateAcknowledgement(t *testing.T) {
	queue, db, clock := fixture(t, 0)
	message := enqueue(t, queue, clock, 1)
	entered, release := make(chan struct{}), make(chan struct{})
	type result struct {
		record Record
		work   bool
		err    error
	}
	done := make(chan result, 1)
	go func() {
		record, work, err := queue.ProcessOne(context.Background(), transportFunc(func(ctx context.Context, _ mail.Message) error {
			close(entered)
			select {
			case <-release:
				return nil
			case <-ctx.Done():
				return ctx.Err()
			}
		}))
		done <- result{record, work, err}
	}()
	<-entered
	if _, work, err := queue.ProcessOne(context.Background(), transportFunc(func(context.Context, mail.Message) error { t.Error("duplicate active lease"); return nil })); err != nil || work {
		t.Fatalf("leased work claimed: %v %v", work, err)
	}
	// SMTP cannot hold the SQLite writer lock while waiting on the network.
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if _, err := db.ExecContext(ctx, `CREATE TABLE independent_domain_write(id TEXT)`); err != nil {
		t.Fatal(err)
	}
	clock.advance(leaseDuration + time.Second)
	record, work, err := queue.ProcessOne(context.Background(), transportFunc(func(context.Context, mail.Message) error { return nil }))
	if err != nil || !work || record.State != "accepted" || record.Attempts != 2 {
		t.Fatalf("takeover: %+v %v %v", record, work, err)
	}
	close(release)
	previous := <-done
	if !previous.work || !errors.Is(previous.err, ErrConflict) {
		t.Fatalf("late acknowledgement: %+v", previous)
	}
	if storedPayload(t, db, message.ID) != nil {
		t.Fatal("late worker restored payload")
	}
}

func TestCancellationLeavesRecoverableLease(t *testing.T) {
	queue, db, clock := fixture(t, 0)
	message := enqueue(t, queue, clock, 1)
	ctx, cancel := context.WithCancel(context.Background())
	_, work, err := queue.ProcessOne(ctx, transportFunc(func(context.Context, mail.Message) error { cancel(); return nil }))
	if !work || !errors.Is(err, context.Canceled) {
		t.Fatalf("cancelled acknowledgement: %v %v", work, err)
	}
	if storedPayload(t, db, message.ID) == nil {
		t.Fatal("uncertain delivery destroyed recovery payload")
	}
	clock.advance(leaseDuration + time.Second)
	record, work, err := queue.ProcessOne(context.Background(), transportFunc(func(context.Context, mail.Message) error { return nil }))
	if err != nil || !work || record.State != "accepted" || record.Attempts != 2 {
		t.Fatalf("recovery: %+v %v %v", record, work, err)
	}
}

func TestValidation(t *testing.T) {
	queue, db, clock := fixture(t, 0)
	ctx := context.Background()
	for _, options := range []Options{{}, {EncryptionKey: make([]byte, 31)}, {EncryptionKey: make([]byte, 32), MaxPending: -1}, {EncryptionKey: make([]byte, 32), MaxPending: 10001}} {
		if _, err := New(db, options); !errors.Is(err, ErrInvalid) {
			t.Fatalf("invalid options: %v", err)
		}
	}
	if _, err := New(nil, Options{EncryptionKey: make([]byte, 32)}); !errors.Is(err, ErrInvalid) {
		t.Fatal("nil DB accepted")
	}
	if err := CreateSchema(ctx, nil); !errors.Is(err, ErrInvalid) {
		t.Fatal("nil migration accepted")
	}
	if err := queue.EnqueueTx(ctx, nil, messageAt(clock, 1), clock.now().Add(time.Hour)); !errors.Is(err, ErrInvalid) {
		t.Fatal("nil tx accepted")
	}
	for _, expiry := range []time.Time{clock.now(), clock.now().Add(-time.Minute), clock.now().Add(25 * time.Hour)} {
		if err := queue.Enqueue(ctx, messageAt(clock, 1), expiry); !errors.Is(err, ErrInvalid) {
			t.Fatalf("invalid expiry: %v", err)
		}
	}
	message := messageAt(clock, 1)
	message.CreatedAt = clock.now().Add(2 * time.Minute)
	if err := queue.Enqueue(ctx, message, clock.now().Add(time.Hour)); !errors.Is(err, ErrInvalid) {
		t.Fatal("future message accepted")
	}
	for _, limit := range []int{0, 101} {
		if _, err := queue.Recent(ctx, limit); !errors.Is(err, ErrInvalid) {
			t.Fatal("invalid metadata limit")
		}
	}
	if _, _, err := queue.ProcessOne(ctx, nil); !errors.Is(err, ErrInvalid) {
		t.Fatal("nil transport accepted")
	}
}