krow_backend/go-api/internal/httpserver/maintenance_test.go

package httpserver_test

import (
	"context"
	"io"
	"log/slog"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/krow/krow-backend/go-api/internal/httpserver"
)

// Scheduler lifecycle.
//
// What is under test is the GOROUTINE, not the deletes — those are covered in
// internal/oauth and internal/ratelimit against real data. Here the questions
// are: does it start, does it do a pass, does it stop when told, does a failure
// take the process with it, and is running it twice safe.

/* ── Lifecycle ──────────────────────────────────────────────────────────── */

// It runs one pass IMMEDIATELY, before the first tick. A process that has been
// down should not carry a backlog for a further hour.
func TestMaintenanceRunsOnceImmediately(t *testing.T) {
	a := newOAuthAPI(t)
	m := a.srv.Maintenance()
	if m == nil {
		t.Fatal("a configured deployment returned no Maintenance")
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	done := make(chan struct{})
	go func() {
		httpserver.SweepMaintenance(ctx, m, slog.New(slog.NewTextHandler(io.Discard, nil)))
		close(done)
	}()

	// The immediate pass is the only one that will happen inside the test's
	// lifetime — the ticker is an hour. Give it a moment, then stop.
	time.Sleep(200 * time.Millisecond)
	cancel()

	select {
	case <-done:
	case <-time.After(5 * time.Second):
		t.Fatal("the sweeper did not stop within 5s of cancellation")
	}
}

// Cancellation must return promptly, or a shutdown hangs on a goroutine nobody
// is waiting for.
func TestMaintenanceStopsOnCancellation(t *testing.T) {
	a := newOAuthAPI(t)
	ctx, cancel := context.WithCancel(context.Background())

	done := make(chan struct{})
	go func() {
		httpserver.SweepMaintenance(ctx, a.srv.Maintenance(),
			slog.New(slog.NewTextHandler(io.Discard, nil)))
		close(done)
	}()

	time.Sleep(100 * time.Millisecond)
	start := time.Now()
	cancel()

	select {
	case <-done:
		if elapsed := time.Since(start); elapsed > 2*time.Second {
			t.Errorf("stopping took %v; shutdown would block on it", elapsed)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("the sweeper ignored cancellation")
	}
}

// An already-cancelled context must not run a pass and must return at once.
func TestMaintenanceWithAnAlreadyCancelledContextReturns(t *testing.T) {
	a := newOAuthAPI(t)
	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	done := make(chan struct{})
	go func() {
		httpserver.SweepMaintenance(ctx, a.srv.Maintenance(),
			slog.New(slog.NewTextHandler(io.Discard, nil)))
		close(done)
	}()

	select {
	case <-done:
	case <-time.After(5 * time.Second):
		t.Fatal("the sweeper did not return on an already-cancelled context")
	}
}

/* ── It does the work ───────────────────────────────────────────────────── */

// One pass removes dead rows and leaves live ones. The detailed retention rules
// are tested in internal/oauth; this asserts the scheduler is wired to them.
func TestMaintenanceSweepRemovesDeadRows(t *testing.T) {
	a := newOAuthAPI(t)
	ctx := context.Background()

	// A grant that is already past its expiry and its grace.
	if _, err := a.h.Pool.Exec(ctx,
		`INSERT INTO oauth_clients (client_id, client_name, redirect_uris)
		 VALUES ('sweep-client', 'Sweep', ARRAY['https://a.test/cb'])`); err != nil {
		t.Fatalf("client: %v", err)
	}
	userID, _ := seededUser(t, a.h.Pool)
	var orgID string
	if err := a.h.Pool.QueryRow(ctx,
		`SELECT org_id::text FROM users WHERE id = $1::uuid`, userID).Scan(&orgID); err != nil {
		t.Fatalf("org: %v", err)
	}

	if _, err := a.h.Pool.Exec(ctx,
		`INSERT INTO oauth_grants
		   (code_hash, client_id, user_id, org_id, redirect_uri, scopes, resource,
		    code_challenge, code_challenge_method, created_date, expires_at)
		 VALUES (repeat('a', 64), 'sweep-client', $1::uuid, $2::uuid, 'https://a.test/cb',
		         ARRAY['krow.read'], $3, repeat('B', 43), 'S256',
		         now() - interval '3 hours', now() - interval '3 hours' + interval '1 minute')`,
		userID, orgID, testMCPResource); err != nil {
		t.Fatalf("grant: %v", err)
	}

	// An expired rate-limit bucket.
	if _, err := a.h.Pool.Exec(ctx,
		`INSERT INTO rate_limits (bucket, window_start, count, expires_at)
		 VALUES ('test:old', now() - interval '3 hours', 5, now() - interval '2 hours')`); err != nil {
		t.Fatalf("bucket: %v", err)
	}
	// And a live one, which must survive.
	if _, err := a.h.Pool.Exec(ctx,
		`INSERT INTO rate_limits (bucket, window_start, count, expires_at)
		 VALUES ('test:live', now(), 1, now() + interval '1 hour')`); err != nil {
		t.Fatalf("bucket: %v", err)
	}

	result, err := a.srv.Maintenance().Sweep(ctx)
	if err != nil {
		t.Fatalf("Sweep: %v", err)
	}

	if result.Grants != 1 {
		t.Errorf("removed %d grants, want 1", result.Grants)
	}
	if result.RateLimits != 1 {
		t.Errorf("removed %d rate-limit rows, want 1", result.RateLimits)
	}
	if result.Total() != 2 {
		t.Errorf("Total() = %d, want 2", result.Total())
	}

	var live int
	if err := a.h.Pool.QueryRow(ctx,
		`SELECT count(*) FROM rate_limits WHERE bucket = 'test:live'`).Scan(&live); err != nil {
		t.Fatalf("count: %v", err)
	}
	if live != 1 {
		t.Error("the live rate-limit window was swept")
	}
}

// Running it repeatedly must be safe and must converge to removing nothing.
func TestRepeatedMaintenanceIsSafe(t *testing.T) {
	a := newOAuthAPI(t)
	ctx := context.Background()
	m := a.srv.Maintenance()

	for i := 0; i < 3; i++ {
		result, err := m.Sweep(ctx)
		if err != nil {
			t.Fatalf("pass %d: %v", i+1, err)
		}
		if i > 0 && result.Total() != 0 {
			t.Errorf("pass %d removed %d rows; a repeat pass should find nothing", i+1, result.Total())
		}
	}
}

// Two instances sweep concurrently with no coordination. Neither may error.
// Run with -race.
func TestConcurrentMaintenanceIsSafe(t *testing.T) {
	a := newOAuthAPI(t)
	ctx := context.Background()

	const instances = 4
	var wg sync.WaitGroup
	errs := make(chan error, instances)

	for i := 0; i < instances; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			if _, err := a.srv.Maintenance().Sweep(ctx); err != nil {
				errs <- err
			}
		}()
	}
	wg.Wait()
	close(errs)

	for err := range errs {
		t.Errorf("concurrent sweep errored: %v", err)
	}
}

/* ── Failure isolation ──────────────────────────────────────────────────── */

// A failing sweep must be logged and survived, not fatal. The database is
// closed underneath the sweeper, which is the closest thing to a real outage a
// test can arrange.
func TestMaintenanceSurvivesADatabaseFailure(t *testing.T) {
	a := newOAuthAPI(t)
	m := a.srv.Maintenance()

	var logged strings.Builder
	log := slog.New(slog.NewTextHandler(&logged, &slog.HandlerOptions{Level: slog.LevelDebug}))

	// A cancelled context makes every statement fail immediately.
	dead, cancel := context.WithCancel(context.Background())
	cancel()

	if _, err := m.Sweep(dead); err == nil {
		t.Log("note: the sweep reported no error on a cancelled context")
	}

	// The goroutine wrapper must not panic or exit the process on that.
	ctx, stop := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() {
		httpserver.SweepMaintenance(ctx, m, log)
		close(done)
	}()
	time.Sleep(150 * time.Millisecond)
	stop()

	select {
	case <-done:
	case <-time.After(5 * time.Second):
		t.Fatal("the sweeper did not stop")
	}
}

/* ── It runs wherever there is something to retain ──────────────────────── */

// This used to assert the opposite: no OAuth meant nothing to sweep, so no
// ticker. Long-term memory changed the premise. Memories carry an expiry that
// is a retention promise about personal data, and a promise enforced only when
// an unrelated feature happens to be switched on is not a promise. So the
// sweeper now exists wherever the database does.
func TestMaintenanceRunsForMemoryEvenWithoutOAuth(t *testing.T) {
	a := newAPI(t) // the standard fixture: no OAuth configuration

	m := a.srv.Maintenance()
	if m == nil {
		t.Fatal("no sweeper, so expired memories would be retained forever")
	}

	// It must still do a pass without OAuth configured rather than failing on
	// the half that is absent.
	if _, err := m.Sweep(context.Background()); err != nil {
		t.Errorf("a sweep without OAuth failed: %v", err)
	}
}

// And the runner still returns immediately when there is genuinely nothing,
// rather than ticking for the life of the process.
func TestSweepMaintenanceReturnsImmediatelyWithNothingToSweep(t *testing.T) {
	done := make(chan struct{})
	go func() {
		httpserver.SweepMaintenance(context.Background(), nil,
			slog.New(slog.NewTextHandler(io.Discard, nil)))
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal("the sweeper ticked despite having nothing to sweep")
	}
}