fix(backend): rate-limit and cap report ingest, drop wildcard CORS (closes #20)

POST /api/v1/reports stays unauthenticated but is bounded. Each client
address, as the trusted-proxy logic resolves it, may send
REPORTS_PER_MINUTE reports a minute (default 60, counted by
go-chi/httprate over a sliding minute); past that it gets 429 with
Retry-After. reportbuf refuses a report that would take the report
files past DATA_DIR_MAX_BYTES (default 1 GiB), counting the files
already in DATA_DIR and unwritten reports at their uncompressed size;
the handler answers 507. CORS adds nothing unless CORS_ALLOWED_ORIGINS
lists origins. A limit that is not a positive number, or an origin
that is not a plain scheme://host[:port], stops the server from
starting.

Model: opus-5-5

backend/internal/reportbuf/reportbuf_test.go

@@ -1,11 +1,17 @@
 package reportbuf_test
 
 import (
+	"encoding/json"
 	"errors"
 	"io/fs"
 	"os"
+	"path/filepath"
+	"strconv"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"sneak.berlin/go/netwatch/internal/config"
 	"sneak.berlin/go/netwatch/internal/globals"
@@ -94,6 +100,253 @@ func TestFailedFinalFlushFailsStop(t *testing.T) {
 	}
 }
 
+// startBuffer starts a Buffer through fx, as main does, with the
+// DATA_DIR and DATA_DIR_MAX_BYTES the calling test has set.
+func startBuffer(t *testing.T) *reportbuf.Buffer {
+	t.Helper()
+
+	var buf *reportbuf.Buffer
+
+	app := fxtest.New(t,
+		fx.Provide(
+			globals.New,
+			logger.New,
+			config.New,
+			reportbuf.New,
+		),
+		fx.Populate(&buf),
+	)
+
+	app.RequireStart()
+	t.Cleanup(app.RequireStop)
+
+	return buf
+}
+
+// lineBytes is what one report takes in the buffer: its JSON and a
+// newline.
+func lineBytes(t *testing.T, report any) int {
+	t.Helper()
+
+	line, err := json.Marshal(report)
+	if err != nil {
+		t.Fatalf("marshal report: %v", err)
+	}
+
+	return len(line) + 1
+}
+
+func TestAppendPastCapIsRefused(t *testing.T) {
+	report := map[string]string{"id": "cap"}
+
+	t.Setenv("DATA_DIR", t.TempDir())
+	t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(lineBytes(t, report)))
+
+	buf := startBuffer(t)
+
+	err := buf.Append(report)
+	if err != nil {
+		t.Fatalf("report that fills the cap exactly: %v", err)
+	}
+
+	err = buf.Append(report)
+	if !errors.Is(err, reportbuf.ErrFull) {
+		t.Fatalf("report past the cap: error = %v, want ErrFull", err)
+	}
+}
+
+// TestCapCountsReportFilesAlreadyInDataDir starts on a data
+// directory holding a report file from an earlier run, and a file
+// that is not a report, which must not count.
+func TestCapCountsReportFilesAlreadyInDataDir(t *testing.T) {
+	const earlierBytes = 100
+
+	report := map[string]string{"id": "cap"}
+	dir := t.TempDir()
+
+	writeBytes(t, filepath.Join(dir, "reports-2026-01-01T00-00-00.000Z.jsonl.zst"),
+		earlierBytes)
+	writeBytes(t, filepath.Join(dir, "notes.txt"), 10*earlierBytes)
+
+	t.Setenv("DATA_DIR", dir)
+	t.Setenv("DATA_DIR_MAX_BYTES",
+		strconv.Itoa(earlierBytes+lineBytes(t, report)))
+
+	buf := startBuffer(t)
+
+	err := buf.Append(report)
+	if err != nil {
+		t.Fatalf("report that fills the cap exactly: %v", err)
+	}
+
+	err = buf.Append(report)
+	if !errors.Is(err, reportbuf.ErrFull) {
+		t.Fatalf("report past the cap: error = %v, want ErrFull", err)
+	}
+}
+
+// TestWrittenReportsCountAtFileSize checks that once reports are
+// written, they count as their compressed file, not their
+// uncompressed size, which frees room under the cap.
+func TestWrittenReportsCountAtFileSize(t *testing.T) {
+	// Repetitive, so its file is far smaller than its JSON.
+	report := map[string]string{"id": strings.Repeat("a", 1000)}
+	size := lineBytes(t, report)
+
+	t.Setenv("DATA_DIR", t.TempDir())
+	// Room for the report twice over only if the first one counts
+	// at its file's size by the time the second arrives.
+	t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(2*size-1))
+
+	buf := startBuffer(t)
+
+	err := buf.Append(report)
+	if err != nil {
+		t.Fatalf("first report: %v", err)
+	}
+
+	err = buf.Flush()
+	if err != nil {
+		t.Fatalf("flush: %v", err)
+	}
+
+	// The second report is written at shutdown, and must not land in
+	// the first file's millisecond (see TestWrittenReportsKeepCounting).
+	time.Sleep(time.Millisecond)
+
+	err = buf.Append(report)
+	if err != nil {
+		t.Fatalf("second report, after the first was written: %v", err)
+	}
+}
+
+// TestWrittenReportsKeepCounting writes one report file after another
+// under a small cap: each report must be taken while the files on disk
+// leave room for it, and refused once they do not.
+func TestWrittenReportsKeepCounting(t *testing.T) {
+	const maxBytes = 200
+
+	report := map[string]string{"id": "written"}
+	size := int64(lineBytes(t, report))
+	dir := t.TempDir()
+
+	t.Setenv("DATA_DIR", dir)
+	t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(maxBytes))
+
+	buf := startBuffer(t)
+
+	// Every file takes at least a byte, so they fill the cap within
+	// maxBytes rounds.
+	for range maxBytes {
+		used := reportFilesBytes(t, dir)
+
+		err := buf.Append(report)
+		if used+size > maxBytes {
+			if !errors.Is(err, reportbuf.ErrFull) {
+				t.Fatalf("with %d bytes of report files: error = %v, "+
+					"want ErrFull", used, err)
+			}
+
+			return
+		}
+
+		if err != nil {
+			t.Fatalf("with %d bytes of report files: %v", used, err)
+		}
+
+		// Report files are named to the millisecond; two in the same
+		// one collide (https://git.eeqj.de/sneak/netwatch/issues/61).
+		time.Sleep(time.Millisecond)
+
+		err = buf.Flush()
+		if err != nil {
+			t.Fatalf("flush: %v", err)
+		}
+	}
+
+	t.Fatal("the report files never filled the cap")
+}
+
+// TestConcurrentAppendsStopAtCap appends from many goroutines at once
+// with room for exactly roomFor reports: exactly that many must be
+// taken, which holds only if Append checks and counts each report
+// under one lock.
+func TestConcurrentAppendsStopAtCap(t *testing.T) {
+	const (
+		roomFor = 5
+		senders = 50
+	)
+
+	// Large, so each Append takes long enough for the senders to
+	// overlap while the cap is reached.
+	report := map[string]string{"id": strings.Repeat("a", 1_000_000)}
+
+	t.Setenv("DATA_DIR", t.TempDir())
+	t.Setenv("DATA_DIR_MAX_BYTES",
+		strconv.Itoa(roomFor*lineBytes(t, report)))
+
+	buf := startBuffer(t)
+
+	var (
+		taken atomic.Int64
+		wg    sync.WaitGroup
+	)
+
+	start := make(chan struct{})
+
+	for range senders {
+		wg.Go(func() {
+			<-start
+
+			err := buf.Append(report)
+			if err == nil {
+				taken.Add(1)
+			} else if !errors.Is(err, reportbuf.ErrFull) {
+				t.Errorf("append: %v", err)
+			}
+		})
+	}
+
+	close(start)
+	wg.Wait()
+
+	if got := taken.Load(); got != roomFor {
+		t.Fatalf("%d reports taken, want %d", got, roomFor)
+	}
+}
+
+// reportFilesBytes returns the total size of the report files in dir.
+func reportFilesBytes(t *testing.T, dir string) int64 {
+	t.Helper()
+
+	paths, err := filepath.Glob(filepath.Join(dir, "reports-*.jsonl.zst"))
+	if err != nil {
+		t.Fatalf("list report files: %v", err)
+	}
+
+	var total int64
+
+	for _, path := range paths {
+		info, statErr := os.Stat(path)
+		if statErr != nil {
+			t.Fatalf("stat %s: %v", path, statErr)
+		}
+
+		total += info.Size()
+	}
+
+	return total
+}
+
+func writeBytes(t *testing.T, path string, n int) {
+	t.Helper()
+
+	err := os.WriteFile(path, make([]byte, n), 0o600)
+	if err != nil {
+		t.Fatalf("write %s: %v", path, err)
+	}
+}
+
 func hasReportFile(t *testing.T, dir string) bool {
 	t.Helper()