summercms/modules/cooler/sitecache_test.go

package cooler

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"io"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"git.golem15.com/golem15/summercms/modules/backpack"
	"git.golem15.com/golem15/summercms/modules/compass"
	"git.golem15.com/golem15/summercms/modules/lagoon"
	_ "github.com/jackc/pgx/v5/stdlib"
	"github.com/testcontainers/testcontainers-go"
	"github.com/testcontainers/testcontainers-go/modules/postgres"
	"gorm.io/gorm"
)

const (
	purgeToken32 = "0123456789abcdef0123456789abcdef"
	purgeToken40 = "site-cache-unit-token-0123456789abcdefgh"
)

// captureLogs routes slog.Default to a text handler on a buffer for the
// test and returns the buffer.
func captureLogs(t *testing.T) *syncBuffer {
	t.Helper()
	buf := &syncBuffer{}
	prev := slog.Default()
	slog.SetDefault(slog.New(slog.NewTextHandler(buf, &slog.HandlerOptions{Level: slog.LevelDebug})))
	t.Cleanup(func() { slog.SetDefault(prev) })
	return buf
}

type syncBuffer struct {
	mu  sync.Mutex
	buf bytes.Buffer
}

func (b *syncBuffer) Write(p []byte) (int, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.buf.Write(p)
}

func (b *syncBuffer) lines() []string {
	b.mu.Lock()
	defer b.mu.Unlock()
	var out []string
	for _, line := range strings.Split(strings.TrimSpace(b.buf.String()), "\n") {
		if line != "" {
			out = append(out, line)
		}
	}
	return out
}

func TestSiteCachePurgeConfig(t *testing.T) {
	refused := []struct {
		name, url, token, key string
	}{
		{"only the URL", "http://127.0.0.1:3016/api/__site__/cache/purge", "", SiteCachePurgeTokenKey},
		{"only the token", "", purgeToken40, SiteCachePurgeURLKey},
		{"a bad scheme", "ftp://127.0.0.1/purge", purgeToken40, SiteCachePurgeURLKey},
		{"no scheme", "127.0.0.1:3016/purge", purgeToken40, SiteCachePurgeURLKey},
		{"no host", "http:///purge", purgeToken40, SiteCachePurgeURLKey},
		{"user info", "http://user:secret@127.0.0.1/purge", purgeToken40, SiteCachePurgeURLKey},
		{"a fragment", "http://127.0.0.1/purge#top", purgeToken40, SiteCachePurgeURLKey},
		{"an empty fragment", "http://127.0.0.1/purge#", purgeToken40, SiteCachePurgeURLKey},
		{"http to a non-loopback host", "http://example.test/purge", purgeToken40, SiteCachePurgeURLKey},
		{"http to a private address", "http://10.0.0.5:3016/purge", purgeToken40, SiteCachePurgeURLKey},
		{"a 31-character token", "http://127.0.0.1/purge", purgeToken32[:31], SiteCachePurgeTokenKey},
	}
	for _, tc := range refused {
		t.Run("refuses "+tc.name, func(t *testing.T) {
			p, err := NewSiteCachePurger(tc.url, tc.token)
			if err == nil || p != nil {
				t.Fatalf("NewSiteCachePurger(%q) = %v, %v; want an error", tc.url, p, err)
			}
			if !strings.Contains(err.Error(), tc.key) {
				t.Fatalf("error %q does not name %s", err, tc.key)
			}
			if tc.token != "" && strings.Contains(err.Error(), tc.token) {
				t.Fatalf("error %q carries the token", err)
			}
		})
	}

	t.Run("both empty disables the purge", func(t *testing.T) {
		p, err := NewSiteCachePurger("", "")
		if p != nil || err != nil {
			t.Fatalf("NewSiteCachePurger(\"\", \"\") = %v, %v; want nil, nil", p, err)
		}
	})

	accepted := []struct {
		url, endpoint string
	}{
		{"https://cache.example.test/api/__site__/cache/purge", "cache.example.test/api/__site__/cache/purge"},
		{"http://127.0.0.1:3016/api/__site__/cache/purge", "127.0.0.1:3016/api/__site__/cache/purge"},
		{"http://127.9.8.7/purge", "127.9.8.7/purge"},
		{"http://[::1]:3000/purge", "[::1]:3000/purge"},
		{"http://localhost:3000/purge?x=1", "localhost:3000/purge"},
	}
	for _, tc := range accepted {
		t.Run("accepts "+tc.url, func(t *testing.T) {
			p, err := NewSiteCachePurger(tc.url, purgeToken32)
			if err != nil || p == nil {
				t.Fatalf("NewSiteCachePurger(%q) = %v, %v; want a purger", tc.url, p, err)
			}
			if p.Endpoint() != tc.endpoint {
				t.Fatalf("Endpoint() = %q, want %q", p.Endpoint(), tc.endpoint)
			}
		})
	}

	t.Run("errors name the keys a caller passes", func(t *testing.T) {
		const urlKey, tokenKey = "acme.site_cache.purge_url", "acme.site_cache.purge_token"
		if _, err := NewSiteCachePurgerFor(urlKey, tokenKey, "http://example.test/purge", purgeToken40); err == nil || !strings.Contains(err.Error(), urlKey) {
			t.Fatalf("http to a remote host: %v, want an error naming %s", err, urlKey)
		}
		if _, err := NewSiteCachePurgerFor(urlKey, tokenKey, "http://127.0.0.1/purge", purgeToken32[:31]); err == nil || !strings.Contains(err.Error(), tokenKey) {
			t.Fatalf("short token: %v, want an error naming %s", err, tokenKey)
		}
	})

	t.Run("SiteCacheFromApp builds from the framework keys and publishes once", func(t *testing.T) {
		cfg := testConfig(t)
		app := backpack.New(cfg)
		if p := SiteCacheFromApp(app); p != nil {
			t.Fatalf("no keys: SiteCacheFromApp = %v, want nil", p)
		}
		if err := cfg.Set(SiteCachePurgeURLKey, "http://127.0.0.1:3000/api/__site__/cache/purge"); err != nil {
			t.Fatal(err)
		}
		if err := cfg.Set(SiteCachePurgeTokenKey, purgeToken40); err != nil {
			t.Fatal(err)
		}
		first := SiteCacheFromApp(app)
		if first == nil || first.Endpoint() != "127.0.0.1:3000/api/__site__/cache/purge" {
			t.Fatalf("SiteCacheFromApp = %v", first)
		}
		if again := SiteCacheFromApp(app); again != first {
			t.Fatal("a second call built another purger")
		}
		if got, ok := app.Lookup[*SiteCachePurger](); !ok || got != first {
			t.Fatal("the purger was not published")
		}
	})

	t.Run("SiteCacheFromApp logs an invalid configuration and returns nil", func(t *testing.T) {
		logs := captureLogs(t)
		cfg := testConfig(t)
		if err := cfg.Set(SiteCachePurgeURLKey, "http://example.test/purge"); err != nil {
			t.Fatal(err)
		}
		if err := cfg.Set(SiteCachePurgeTokenKey, purgeToken40); err != nil {
			t.Fatal(err)
		}
		if p := SiteCacheFromApp(backpack.New(cfg)); p != nil {
			t.Fatalf("SiteCacheFromApp = %v, want nil", p)
		}
		lines := logs.lines()
		if len(lines) != 1 || !strings.Contains(lines[0], "level=ERROR") || !strings.Contains(lines[0], SiteCachePurgeURLKey) {
			t.Fatalf("log = %q", lines)
		}
		assertNoToken(t, logs)
	})

	t.Run("Install fails on an invalid configuration and keeps a published purger", func(t *testing.T) {
		cfg := testConfig(t)
		if err := cfg.Set(SiteCachePurgeURLKey, "http://127.0.0.1/purge"); err != nil {
			t.Fatal(err)
		}
		if _, err := Install(context.Background(), backpack.New(cfg)); err == nil || !strings.Contains(err.Error(), SiteCachePurgeTokenKey) {
			t.Fatalf("Install with only the URL = %v, want an error naming %s", err, SiteCachePurgeTokenKey)
		}
		alias, err := NewSiteCachePurgerFor("acme.purge_url", "acme.purge_token", "http://127.0.0.1:4000/purge", purgeToken40)
		if err != nil {
			t.Fatal(err)
		}
		app := backpack.New(cfg)
		if err := app.Publish(alias); err != nil {
			t.Fatal(err)
		}
		h, err := Install(context.Background(), app)
		if err != nil {
			t.Fatalf("Install with a published purger: %v", err)
		}
		defer h.Close()
		if got := SiteCacheFromApp(app); got != alias {
			t.Fatal("Install replaced the published purger")
		}
	})

	var nilPurger *SiteCachePurger
	if nilPurger.Endpoint() != "" {
		t.Fatal("a nil purger has an endpoint")
	}
}

// purgeHit is one request the test receiver got.
type purgeHit struct {
	method, path, auth, contentType string
	body                            map[string]string
}

func purgeReceiver(t *testing.T, handle func(w http.ResponseWriter, r *http.Request)) (*httptest.Server, func() []purgeHit) {
	t.Helper()
	var mu sync.Mutex
	var hits []purgeHit
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		raw, _ := io.ReadAll(r.Body)
		var body map[string]string
		_ = json.Unmarshal(raw, &body)
		mu.Lock()
		hits = append(hits, purgeHit{r.Method, r.URL.Path, r.Header.Get("Authorization"), r.Header.Get("Content-Type"), body})
		mu.Unlock()
		handle(w, r)
	}))
	t.Cleanup(srv.Close)
	return srv, func() []purgeHit {
		mu.Lock()
		defer mu.Unlock()
		return append([]purgeHit(nil), hits...)
	}
}

func newTestPurger(t *testing.T, srv *httptest.Server) *SiteCachePurger {
	t.Helper()
	p, err := NewSiteCachePurger(srv.URL+"/api/__site__/cache/purge", purgeToken40)
	if err != nil {
		t.Fatal(err)
	}
	return p
}

// waitPurges drains the purges queued on p, failing the test after 5 s.
func waitPurges(t *testing.T, p *SiteCachePurger) {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := p.Wait(ctx); err != nil {
		t.Fatalf("Wait: %v", err)
	}
}

func assertNoToken(t *testing.T, logs *syncBuffer) {
	t.Helper()
	for _, line := range logs.lines() {
		if strings.Contains(line, purgeToken40) || strings.Contains(line, "Bearer") || strings.Contains(line, "Authorization") {
			t.Fatalf("log line carries the token or the Authorization header: %s", line)
		}
	}
}

func TestSiteCachePurgeRequest(t *testing.T) {
	t.Run("sends the contract request and logs the purged count", func(t *testing.T) {
		logs := captureLogs(t)
		srv, hits := purgeReceiver(t, func(w http.ResponseWriter, _ *http.Request) {
			w.Header().Set("Content-Type", "application/json")
			_, _ = w.Write([]byte(`{"purged":7}`))
		})
		newTestPurger(t, srv).Purge(context.Background(), "category", "cordyceps")
		got := hits()
		if len(got) != 1 {
			t.Fatalf("%d requests, want 1", len(got))
		}
		h := got[0]
		if h.method != http.MethodPost || h.path != "/api/__site__/cache/purge" {
			t.Fatalf("request %s %s", h.method, h.path)
		}
		if h.auth != "Bearer "+purgeToken40 || !strings.HasPrefix(h.contentType, "application/json") {
			t.Fatalf("Authorization %q, Content-Type %q", h.auth, h.contentType)
		}
		if len(h.body) != 2 || h.body["kind"] != "category" || h.body["slug"] != "cordyceps" {
			t.Fatalf("body = %#v", h.body)
		}
		lines := logs.lines()
		if len(lines) != 1 || !strings.Contains(lines[0], "level=INFO") || !strings.Contains(lines[0], `msg="cooler: site cache purged"`) ||
			!strings.Contains(lines[0], "kind=category") || !strings.Contains(lines[0], "slug=cordyceps") ||
			!strings.Contains(lines[0], "status=200") || !strings.Contains(lines[0], "purged=7") {
			t.Fatalf("log = %q", lines)
		}
		assertNoToken(t, logs)
	})

	t.Run("an answer without purged logs the status only", func(t *testing.T) {
		logs := captureLogs(t)
		srv, _ := purgeReceiver(t, func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusNoContent) })
		newTestPurger(t, srv).Purge(context.Background(), "post", "hello")
		lines := logs.lines()
		if len(lines) != 1 || !strings.Contains(lines[0], "level=INFO") || !strings.Contains(lines[0], "status=204") || strings.Contains(lines[0], "purged=") {
			t.Fatalf("log = %q", lines)
		}
	})

	for _, status := range []int{http.StatusInternalServerError, http.StatusUnauthorized} {
		t.Run("a "+http.StatusText(status)+" answer logs one warning", func(t *testing.T) {
			logs := captureLogs(t)
			srv, hits := purgeReceiver(t, func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(status) })
			newTestPurger(t, srv).Purge(context.Background(), "post", "hello")
			if len(hits()) != 1 {
				t.Fatalf("%d requests, want 1", len(hits()))
			}
			lines := logs.lines()
			if len(lines) != 1 || !strings.Contains(lines[0], "level=WARN") || !strings.Contains(lines[0], `msg="cooler: site cache purge failed"`) ||
				!strings.Contains(lines[0], "kind=post") || !strings.Contains(lines[0], "slug=hello") {
				t.Fatalf("log = %q", lines)
			}
			if !strings.Contains(lines[0], "status="+strconv.Itoa(status)) {
				t.Fatalf("log %q lacks status=%d", lines[0], status)
			}
			assertNoToken(t, logs)
		})
	}

	t.Run("a slow receiver is abandoned after the timeout", func(t *testing.T) {
		logs := captureLogs(t)
		release := make(chan struct{})
		srv, _ := purgeReceiver(t, func(_ http.ResponseWriter, r *http.Request) {
			select {
			case <-release:
			case <-r.Context().Done():
			}
		})
		defer close(release)
		start := time.Now()
		newTestPurger(t, srv).Purge(context.Background(), "post", "slow")
		elapsed := time.Since(start)
		if elapsed < SiteCachePurgeTimeout-200*time.Millisecond || elapsed > SiteCachePurgeTimeout+1500*time.Millisecond {
			t.Fatalf("Purge returned after %s, want about %s", elapsed, SiteCachePurgeTimeout)
		}
		lines := logs.lines()
		if len(lines) != 1 || !strings.Contains(lines[0], "level=WARN") || !strings.Contains(lines[0], "error=timeout") {
			t.Fatalf("log = %q", lines)
		}
		assertNoToken(t, logs)
	})

	t.Run("an unreachable receiver logs one warning without the URL's token", func(t *testing.T) {
		logs := captureLogs(t)
		srv, _ := purgeReceiver(t, func(http.ResponseWriter, *http.Request) {})
		p := newTestPurger(t, srv)
		srv.Close()
		p.Purge(context.Background(), "category", "gone")
		lines := logs.lines()
		if len(lines) != 1 || !strings.Contains(lines[0], "level=WARN") || !strings.Contains(lines[0], "kind=category") {
			t.Fatalf("log = %q", lines)
		}
		assertNoToken(t, logs)
	})

	t.Run("a cancelled parent context still sends", func(t *testing.T) {
		captureLogs(t)
		srv, hits := purgeReceiver(t, func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte(`{"purged":1}`)) })
		ctx, cancel := context.WithCancel(context.Background())
		cancel()
		newTestPurger(t, srv).Purge(ctx, "post", "after-cancel")
		if got := hits(); len(got) != 1 || got[0].body["slug"] != "after-cancel" {
			t.Fatalf("requests = %#v, want one", got)
		}
	})

	t.Run("a nil purger sends nothing", func(t *testing.T) {
		logs := captureLogs(t)
		var p *SiteCachePurger
		p.Purge(context.Background(), "post", "nil")
		PurgeSiteCacheAfterCommit(context.Background(), nil, nil, "post", "nil")
		if lines := logs.lines(); len(lines) != 0 {
			t.Fatalf("log = %q, want nothing", lines)
		}
	})

	t.Run("outside a transaction the after-commit purge is queued at once", func(t *testing.T) {
		captureLogs(t)
		srv, hits := purgeReceiver(t, func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte(`{"purged":2}`)) })
		p := newTestPurger(t, srv)
		PurgeSiteCacheAfterCommit(context.Background(), nil, p, "category", "now")
		waitPurges(t, p)
		if got := hits(); len(got) != 1 || got[0].body["kind"] != "category" || got[0].body["slug"] != "now" {
			t.Fatalf("requests = %#v, want one", got)
		}
	})
}

// testConfig returns an empty config that ignores the process environment.
func testConfig(t *testing.T) *compass.Config {
	t.Helper()
	cfg, err := compass.Open(compass.Options{Dir: t.TempDir(), Environ: []string{}})
	if err != nil {
		t.Fatal(err)
	}
	return cfg
}

// purgeDB starts a Postgres container for the test and returns a GORM handle
// with a table to write in transactions. It skips under -short.
func purgeDB(t *testing.T) *gorm.DB {
	t.Helper()
	if testing.Short() {
		t.Skip("requires testcontainers postgres")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	t.Cleanup(cancel)
	ctr, err := postgres.Run(ctx,
		"postgres:16-alpine",
		postgres.WithDatabase("cooler"),
		postgres.WithUsername("cooler"),
		postgres.WithPassword("cooler"),
		postgres.BasicWaitStrategies(),
	)
	if err != nil {
		t.Fatalf("postgres: %v", err)
	}
	t.Cleanup(func() { _ = testcontainers.TerminateContainer(ctr) })
	dsn, err := ctr.ConnectionString(ctx, "sslmode=disable")
	if err != nil {
		t.Fatal(err)
	}
	db, err := sql.Open("pgx", dsn)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = db.Close() })
	if err := db.PingContext(ctx); err != nil {
		t.Fatal(err)
	}
	gdb, err := lagoon.Use(ctx, db)
	if err != nil {
		t.Fatalf("lagoon.Use: %v", err)
	}
	if err := gdb.Exec(`CREATE TABLE purge_posts (id serial PRIMARY KEY, slug text NOT NULL)`).Error; err != nil {
		t.Fatal(err)
	}
	return gdb
}

func countPosts(t *testing.T, gdb *gorm.DB, slug string) int64 {
	t.Helper()
	var n int64
	if err := gdb.Raw(`SELECT count(*) FROM purge_posts WHERE slug = ?`, slug).Scan(&n).Error; err != nil {
		t.Fatal(err)
	}
	return n
}

func TestSiteCacheAfterCommitOnly(t *testing.T) {
	gdb := purgeDB(t)
	captureLogs(t)
	srv, hits := purgeReceiver(t, func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte(`{"purged":1}`)) })
	p := newTestPurger(t, srv)
	ctx := context.Background()

	t.Run("a rolled-back save sends nothing", func(t *testing.T) {
		boom := errors.New("validation failed")
		err := lagoon.Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error {
			if err := tx.Exec(`INSERT INTO purge_posts (slug) VALUES ('rolled-back')`).Error; err != nil {
				return err
			}
			PurgeSiteCacheAfterCommit(ctx, tx, p, "post", "rolled-back")
			return boom
		})
		if !errors.Is(err, boom) {
			t.Fatalf("Transaction = %v, want the callback's error", err)
		}
		if n := countPosts(t, gdb, "rolled-back"); n != 0 {
			t.Fatalf("%d rows after rollback", n)
		}
		waitPurges(t, p)
		if got := hits(); len(got) != 0 {
			t.Fatalf("a rolled-back save sent %d purges", len(got))
		}
	})

	t.Run("a committed save sends one purge after the commit", func(t *testing.T) {
		var during int
		err := lagoon.Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error {
			if err := tx.Exec(`INSERT INTO purge_posts (slug) VALUES ('committed')`).Error; err != nil {
				return err
			}
			PurgeSiteCacheAfterCommit(ctx, tx, p, "post", "committed")
			during = len(hits())
			return nil
		})
		if err != nil {
			t.Fatal(err)
		}
		if during != 0 {
			t.Fatalf("%d purges were sent before the commit", during)
		}
		waitPurges(t, p)
		got := hits()
		if len(got) != 1 || got[0].body["kind"] != "post" || got[0].body["slug"] != "committed" {
			t.Fatalf("purges = %#v, want one for post committed", got)
		}
	})

	t.Run("a nested transaction purges once, after the outer commit", func(t *testing.T) {
		before := len(hits())
		var afterInner int
		err := lagoon.Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error {
			if err := lagoon.Transaction(ctx, tx, func(ctx context.Context, inner *gorm.DB) error {
				PurgeSiteCacheAfterCommit(ctx, inner, p, "category", "nested")
				return inner.Exec(`INSERT INTO purge_posts (slug) VALUES ('nested')`).Error
			}); err != nil {
				return err
			}
			afterInner = len(hits())
			return nil
		})
		if err != nil {
			t.Fatal(err)
		}
		if afterInner != before {
			t.Fatal("the purge ran when the inner transaction finished")
		}
		waitPurges(t, p)
		if got := hits(); len(got) != before+1 || got[len(got)-1].body["slug"] != "nested" {
			t.Fatalf("purges = %#v, want one more for nested", got)
		}
	})
}

func TestSiteCacheFailureNeverFails(t *testing.T) {
	gdb := purgeDB(t)
	ctx := context.Background()

	t.Run("a 500 answer leaves the save committed", func(t *testing.T) {
		logs := captureLogs(t)
		srv, hits := purgeReceiver(t, func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusInternalServerError) })
		p := newTestPurger(t, srv)
		err := lagoon.Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error {
			PurgeSiteCacheAfterCommit(ctx, tx, p, "post", "server-error")
			return tx.Exec(`INSERT INTO purge_posts (slug) VALUES ('server-error')`).Error
		})
		if err != nil {
			t.Fatalf("a failing purge failed the save: %v", err)
		}
		waitPurges(t, p)
		if n := countPosts(t, gdb, "server-error"); n != 1 || len(hits()) != 1 {
			t.Fatalf("rows %d, purges %d; want 1 and 1", n, len(hits()))
		}
		if lines := logs.lines(); len(lines) != 1 || !strings.Contains(lines[0], "level=WARN") || !strings.Contains(lines[0], "status=500") {
			t.Fatalf("log = %q", lines)
		}
		assertNoToken(t, logs)
	})

	t.Run("a slow receiver is abandoned and the save stays committed", func(t *testing.T) {
		logs := captureLogs(t)
		release := make(chan struct{})
		srv, _ := purgeReceiver(t, func(_ http.ResponseWriter, r *http.Request) {
			select {
			case <-release:
			case <-r.Context().Done():
			}
		})
		defer close(release)
		p := newTestPurger(t, srv)
		start := time.Now()
		err := lagoon.Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error {
			PurgeSiteCacheAfterCommit(ctx, tx, p, "post", "slow")
			return tx.Exec(`INSERT INTO purge_posts (slug) VALUES ('slow')`).Error
		})
		if err != nil {
			t.Fatalf("a slow purge failed the save: %v", err)
		}
		if elapsed := time.Since(start); elapsed > 500*time.Millisecond {
			t.Fatalf("the save waited %s for a queued purge", elapsed)
		}
		if n := countPosts(t, gdb, "slow"); n != 1 {
			t.Fatalf("%d rows, want the save committed", n)
		}
		waitPurges(t, p)
		if lines := logs.lines(); len(lines) != 1 || !strings.Contains(lines[0], "error=timeout") {
			t.Fatalf("log = %q", lines)
		}
	})

	t.Run("an unreachable receiver leaves the save committed", func(t *testing.T) {
		captureLogs(t)
		srv, _ := purgeReceiver(t, func(http.ResponseWriter, *http.Request) {})
		p := newTestPurger(t, srv)
		srv.Close()
		err := lagoon.Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error {
			PurgeSiteCacheAfterCommit(ctx, tx, p, "post", "unreachable")
			return tx.Exec(`INSERT INTO purge_posts (slug) VALUES ('unreachable')`).Error
		})
		if err != nil || countPosts(t, gdb, "unreachable") != 1 {
			t.Fatalf("Transaction = %v, rows %d", err, countPosts(t, gdb, "unreachable"))
		}
		waitPurges(t, p)
	})
}

func TestSiteCacheFromApp(t *testing.T) {
	if p := SiteCacheFromApp(nil); p != nil {
		t.Fatal("SiteCacheFromApp(nil) must be nil")
	}
	if p := SiteCacheFromApp(backpack.New(nil)); p != nil {
		t.Fatal("an app without config has no purger")
	}

	cfg := testConfig(t)
	app := backpack.New(cfg)
	if p := SiteCacheFromApp(app); p != nil {
		t.Fatalf("unset keys: %v, want nil", p)
	}
	if _, ok := app.Lookup[*SiteCachePurger](); ok {
		t.Fatal("unset keys published a purger")
	}
	if err := cfg.Set(SiteCachePurgeURLKey, "https://www.example.test/api/__site__/cache/purge"); err != nil {
		t.Fatal(err)
	}
	if err := cfg.Set(SiteCachePurgeTokenKey, purgeToken40); err != nil {
		t.Fatal(err)
	}

	// Concurrent first calls build and publish one purger.
	var wg sync.WaitGroup
	got := make([]*SiteCachePurger, 20)
	for i := range got {
		wg.Add(1)
		go func() {
			defer wg.Done()
			got[i] = SiteCacheFromApp(app)
		}()
	}
	wg.Wait()
	for i, p := range got {
		if p == nil || p != got[0] {
			t.Fatalf("call %d = %v, want the one published purger", i, p)
		}
	}
	if pub, ok := app.Lookup[*SiteCachePurger](); !ok || pub != got[0] {
		t.Fatal("the purger was not published")
	}
	if got[0].Endpoint() != "www.example.test/api/__site__/cache/purge" {
		t.Fatalf("Endpoint = %q", got[0].Endpoint())
	}

	// A later config change does not rebuild the published purger.
	if err := cfg.Set(SiteCachePurgeURLKey, "https://other.example.test/purge"); err != nil {
		t.Fatal(err)
	}
	if again := SiteCacheFromApp(app); again != got[0] {
		t.Fatal("SiteCacheFromApp rebuilt a published purger")
	}
}

func TestSiteCachePurgeNoRedirect(t *testing.T) {
	logs := captureLogs(t)
	target, targetHits := purgeReceiver(t, func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte(`{"purged":1}`)) })
	srv, hits := purgeReceiver(t, func(w http.ResponseWriter, _ *http.Request) {
		w.Header().Set("Location", target.URL+"/api/__site__/cache/purge")
		w.WriteHeader(http.StatusTemporaryRedirect)
	})
	newTestPurger(t, srv).Purge(context.Background(), "post", "moved")
	if len(hits()) != 1 {
		t.Fatalf("%d requests to the purge URL, want 1", len(hits()))
	}
	if got := targetHits(); len(got) != 0 {
		t.Fatalf("the redirect target received %d requests: %#v", len(got), got)
	}
	lines := logs.lines()
	if len(lines) != 1 || !strings.Contains(lines[0], "level=WARN") || !strings.Contains(lines[0], `msg="cooler: site cache purge failed"`) || !strings.Contains(lines[0], "status=307") {
		t.Fatalf("log = %q, want one failed purge with status=307", lines)
	}
	assertNoToken(t, logs)
}

func TestSiteCachePurgeAsyncCoalesced(t *testing.T) {
	logs := captureLogs(t)
	release := make(chan struct{})
	srv, hits := purgeReceiver(t, func(w http.ResponseWriter, _ *http.Request) {
		<-release
		_, _ = w.Write([]byte(`{"purged":3}`))
	})
	p := newTestPurger(t, srv)
	released := false
	defer func() {
		if !released {
			close(release)
		}
		waitPurges(t, p)
	}()
	for i := range 20 {
		start := time.Now()
		PurgeSiteCacheAfterCommit(context.Background(), nil, p, "post", "slug-"+strconv.Itoa(i))
		if elapsed := time.Since(start); elapsed > 50*time.Millisecond {
			t.Fatalf("call %d blocked for %s", i, elapsed)
		}
	}
	close(release)
	released = true
	waitPurges(t, p)
	got := hits()
	if len(got) != 2 {
		t.Fatalf("%d requests, want 2 (one in flight, one coalesced follow-up)", len(got))
	}
	if got[0].body["slug"] != "slug-0" || got[1].body["slug"] != "slug-19" {
		t.Fatalf("slugs %q, %q; want slug-0 then the last, slug-19", got[0].body["slug"], got[1].body["slug"])
	}
	lines := logs.lines()
	if len(lines) != 2 || strings.Contains(lines[0], "coalesced=") || !strings.Contains(lines[1], "slug=slug-19") || !strings.Contains(lines[1], "coalesced=18") {
		t.Fatalf("log = %q, want the second purged line with coalesced=18", lines)
	}
	assertNoToken(t, logs)
}

// blockingReceiver answers once release is closed (or the request ends).
func blockingReceiver(t *testing.T) (*httptest.Server, func() []purgeHit, func()) {
	t.Helper()
	release := make(chan struct{})
	var once sync.Once
	srv, hits := purgeReceiver(t, func(w http.ResponseWriter, r *http.Request) {
		select {
		case <-release:
			_, _ = w.Write([]byte(`{"purged":1}`))
		case <-r.Context().Done():
		}
	})
	free := func() { once.Do(func() { close(release) }) }
	t.Cleanup(free)
	return srv, hits, free
}

func TestSiteCachePurgeWaitAndClose(t *testing.T) {
	t.Run("nil and idle purgers return at once", func(t *testing.T) {
		var nilPurger *SiteCachePurger
		if err := nilPurger.Wait(context.Background()); err != nil {
			t.Fatalf("nil Wait = %v", err)
		}
		srv, _ := purgeReceiver(t, func(http.ResponseWriter, *http.Request) {})
		ctx, cancel := context.WithCancel(context.Background())
		cancel()
		if err := newTestPurger(t, srv).Wait(ctx); err != nil {
			t.Fatalf("idle Wait under a cancelled context = %v, want nil", err)
		}
	})

	t.Run("Wait gives up with the context and the purge is still sent", func(t *testing.T) {
		captureLogs(t)
		srv, hits, free := blockingReceiver(t)
		p := newTestPurger(t, srv)
		PurgeSiteCacheAfterCommit(context.Background(), nil, p, "post", "pending")
		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
		defer cancel()
		if err := p.Wait(ctx); !errors.Is(err, context.DeadlineExceeded) {
			t.Fatalf("Wait = %v, want context.DeadlineExceeded", err)
		}
		free()
		waitPurges(t, p)
		if got := hits(); len(got) != 1 || got[0].body["slug"] != "pending" {
			t.Fatalf("requests = %#v, want the queued purge", got)
		}
	})

	install := func(t *testing.T, srv *httptest.Server) (*Handle, *SiteCachePurger) {
		t.Helper()
		cfg := testConfig(t)
		if err := cfg.Set(SiteCachePurgeURLKey, srv.URL+"/api/__site__/cache/purge"); err != nil {
			t.Fatal(err)
		}
		if err := cfg.Set(SiteCachePurgeTokenKey, purgeToken40); err != nil {
			t.Fatal(err)
		}
		app := backpack.New(cfg)
		h, err := Install(context.Background(), app)
		if err != nil {
			t.Fatal(err)
		}
		p := SiteCacheFromApp(app)
		if p == nil {
			t.Fatal("Install published no purger")
		}
		return h, p
	}

	t.Run("Handle.Close waits for a queued purge", func(t *testing.T) {
		captureLogs(t)
		srv, hits := purgeReceiver(t, func(w http.ResponseWriter, _ *http.Request) {
			time.Sleep(200 * time.Millisecond)
			_, _ = w.Write([]byte(`{"purged":1}`))
		})
		h, p := install(t, srv)
		PurgeSiteCacheAfterCommit(context.Background(), nil, p, "post", "at-shutdown")
		if err := h.Close(); err != nil {
			t.Fatal(err)
		}
		if got := hits(); len(got) != 1 || got[0].body["slug"] != "at-shutdown" {
			t.Fatalf("requests after Close = %#v, want the queued purge", got)
		}
	})

	t.Run("Handle.Close is bounded when the site hangs", func(t *testing.T) {
		captureLogs(t)
		srv, _, _ := blockingReceiver(t)
		h, p := install(t, srv)
		PurgeSiteCacheAfterCommit(context.Background(), nil, p, "post", "hung")
		start := time.Now()
		if err := h.Close(); err != nil {
			t.Fatal(err)
		}
		if elapsed := time.Since(start); elapsed > SiteCachePurgeTimeout+time.Second+500*time.Millisecond {
			t.Fatalf("Close took %s, want at most %s plus a second", elapsed, SiteCachePurgeTimeout)
		}
		waitPurges(t, p)
	})
}

func TestSiteCacheBulkCommitCoalesced(t *testing.T) {
	gdb := purgeDB(t)
	captureLogs(t)
	ctx := context.Background()

	run := func(t *testing.T, save func(p *SiteCachePurger) error, last string) {
		t.Helper()
		srv, hits, free := blockingReceiver(t)
		p := newTestPurger(t, srv)
		start := time.Now()
		if err := save(p); err != nil {
			t.Fatal(err)
		}
		if elapsed := time.Since(start); elapsed > 500*time.Millisecond {
			t.Fatalf("the saves took %s while the site was blocked, want under 500ms", elapsed)
		}
		free()
		waitPurges(t, p)
		got := hits()
		if len(got) != 2 {
			t.Fatalf("%d requests, want 2 (one in flight, one coalesced follow-up)", len(got))
		}
		if got[1].body["slug"] != last {
			t.Fatalf("follow-up slug = %q, want the last, %q", got[1].body["slug"], last)
		}
	}

	t.Run("25 purges in one transaction", func(t *testing.T) {
		run(t, func(p *SiteCachePurger) error {
			return lagoon.Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error {
				for i := range 25 {
					slug := "bulk-" + strconv.Itoa(i)
					if err := tx.Exec(`INSERT INTO purge_posts (slug) VALUES (?)`, slug).Error; err != nil {
						return err
					}
					PurgeSiteCacheAfterCommit(ctx, tx, p, "post", slug)
				}
				return nil
			})
		}, "bulk-24")
	})

	t.Run("10 separate transactions", func(t *testing.T) {
		run(t, func(p *SiteCachePurger) error {
			for i := range 10 {
				slug := "each-" + strconv.Itoa(i)
				if err := lagoon.Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error {
					PurgeSiteCacheAfterCommit(ctx, tx, p, "post", slug)
					return tx.Exec(`INSERT INTO purge_posts (slug) VALUES (?)`, slug).Error
				}); err != nil {
					return err
				}
			}
			return nil
		}, "each-9")
	})
}