netwatch/internal/reportbuf/reportbuf_test.go

package reportbuf_test

import (
	"encoding/json"
	"errors"
	"fmt"
	"io/fs"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"testing/synctest"
	"time"

	"sneak.berlin/go/netwatch/internal/config"
	"sneak.berlin/go/netwatch/internal/globals"
	"sneak.berlin/go/netwatch/internal/handlers"
	"sneak.berlin/go/netwatch/internal/healthcheck"
	"sneak.berlin/go/netwatch/internal/logger"
	"sneak.berlin/go/netwatch/internal/reportbuf"

	"github.com/klauspost/compress/zstd"
	"go.uber.org/fx"
	"go.uber.org/fx/fxtest"
)

// TestFlushOnShutdown proves the flush-on-shutdown path: a
// report appended after start but before the periodic flush
// window must reach disk when the fx lifecycle stops. This is
// the exact case that silent data loss on restart used to
// destroy.
func TestFlushOnShutdown(t *testing.T) {
	dir := t.TempDir()
	t.Setenv("DATA_DIR", dir)

	var buf *reportbuf.Buffer

	app := fxtest.New(t,
		fx.Provide(
			globals.New,
			logger.New,
			config.New,
			reportbuf.New,
		),
		fx.Populate(&buf),
	)

	app.RequireStart()

	err := buf.Append(map[string]string{"probe": "shutdown"})
	if err != nil {
		t.Fatalf("append report: %v", err)
	}

	// RequireStop runs the reportbuf OnStop hook, which is the
	// only code path that flushes buffered reports on shutdown.
	app.RequireStop()

	if !hasReportFile(t, dir) {
		t.Fatal("no report file on disk after shutdown; " +
			"the buffered report was lost")
	}
}

// TestFailedFinalFlushFailsStop proves a final flush that cannot
// write its file makes the stop fail, which makes the process
// exit non-zero instead of dropping the buffered reports silently.
func TestFailedFinalFlushFailsStop(t *testing.T) {
	dir := t.TempDir()
	t.Setenv("DATA_DIR", dir)

	var buf *reportbuf.Buffer

	app := fxtest.New(t,
		fx.Provide(
			globals.New,
			logger.New,
			config.New,
			reportbuf.New,
		),
		fx.Populate(&buf),
	)

	app.RequireStart()

	err := buf.Append(map[string]string{"probe": "shutdown"})
	if err != nil {
		t.Fatalf("append report: %v", err)
	}

	// Removing the data directory leaves the final flush nowhere to
	// write. A read-only directory would not do: tests run as root
	// in the backend image, and root ignores the read-only bit.
	err = os.RemoveAll(dir)
	if err != nil {
		t.Fatalf("remove data dir: %v", err)
	}

	err = app.Stop(t.Context())
	if !errors.Is(err, fs.ErrNotExist) {
		t.Fatalf("stop error = %v, want the final flush's error", err)
	}
}

// TestUncreatableDataDirStopsStart sets DATA_DIR below a regular file,
// where no directory can be created, even by root, as the tests run in
// the backend image. The start must fail with an error naming DATA_DIR
// and its value, as the errors for the other settings do.
func TestUncreatableDataDirStopsStart(t *testing.T) {
	file := filepath.Join(t.TempDir(), "file")
	writeBytes(t, file, 1)

	dir := filepath.Join(file, "reports")
	t.Setenv("DATA_DIR", dir)

	app := fxtest.New(t,
		fx.Provide(
			globals.New,
			logger.New,
			config.New,
			reportbuf.New,
		),
		fx.Invoke(func(*reportbuf.Buffer) {}),
	)

	err := app.Start(t.Context())
	if err == nil {
		t.Fatal("start succeeded with DATA_DIR below a regular file")
	}

	want := fmt.Sprintf("DATA_DIR %q: ", dir)
	if !strings.Contains(err.Error(), want) {
		t.Fatalf("start error = %q, want it to contain %q", err, want)
	}
}

// TestUnwritableDataDirStopsStart sets DATA_DIR to /proc, which exists
// and can be read, but where no file can be created, even by root, as
// the tests run in the backend image: root ignores a directory the test
// makes read-only. The start must fail with an error naming DATA_DIR and
// its value, rather than the first write of a report file.
func TestUnwritableDataDirStopsStart(t *testing.T) {
	dir := "/proc"
	t.Setenv("DATA_DIR", dir)

	app := fxtest.New(t,
		fx.Provide(
			globals.New,
			logger.New,
			config.New,
			reportbuf.New,
		),
		fx.Invoke(func(*reportbuf.Buffer) {}),
	)

	err := app.Start(t.Context())
	if err == nil {
		t.Fatal("start succeeded with DATA_DIR /proc")
	}

	want := fmt.Sprintf("DATA_DIR %q: ", dir)
	if !strings.Contains(err.Error(), want) {
		t.Fatalf("start error = %q, want it to contain %q", err, want)
	}
}

// TestStopWaitsForPeriodicWrite holds open the write the periodic flush
// starts, then stops the app. The stop must wait until the write is
// released, and the report must then be in its file. The test runs in
// a synctest bubble, whose clock moves on at once when every goroutine
// in it is waiting, so the periodic flush does not take a minute.
func TestStopWaitsForPeriodicWrite(t *testing.T) {
	dir := t.TempDir()
	t.Setenv("DATA_DIR", dir)

	synctest.Test(t, func(t *testing.T) {
		buf, app := startApp(t)

		created := make(chan struct{})
		release := make(chan struct{})

		buf.OnFileCreated(func(*os.File) {
			created <- struct{}{}

			<-release
		})

		err := buf.Append(map[string]string{"id": "periodic"})
		if err != nil {
			t.Fatalf("append report: %v", err)
		}

		<-created

		err = stopWhileHeld(t, app, release)
		if err != nil {
			t.Fatalf("stop: %v", err)
		}

		files, err := readReportFiles(dir)
		if err != nil {
			t.Fatalf("read report files: %v", err)
		}

		want := `{"id":"periodic"}` + "\n"
		if len(files) != 1 || files[0] != want {
			t.Fatalf("report files = %q, want one holding %q", files, want)
		}
	})
}

// TestStopWaitsForWriteAtSizeThreshold holds open the write Append
// starts when the buffer reaches FlushSizeThreshold, then stops the
// app. The stop must wait until the write is released, and the reports
// must then be in its file.
func TestStopWaitsForWriteAtSizeThreshold(t *testing.T) {
	dir := t.TempDir()
	t.Setenv("DATA_DIR", dir)

	synctest.Test(t, func(t *testing.T) {
		buf, app := startApp(t)

		created := make(chan struct{})
		release := make(chan struct{})

		buf.OnFileCreated(func(*os.File) {
			created <- struct{}{}

			<-release
		})

		appended := appendToThreshold(t, buf)

		<-created

		err := stopWhileHeld(t, app, release)
		if err != nil {
			t.Fatalf("stop: %v", err)
		}

		files, err := readReportFiles(dir)
		if err != nil {
			t.Fatalf("read report files: %v", err)
		}

		if len(files) != 1 || files[0] != appended {
			t.Fatalf("%d report files, want one holding the %d bytes "+
				"appended", len(files), len(appended))
		}
	})
}

// TestFinalFlushWritesReportsOfFailedWrite holds open the write the
// periodic flush starts, makes it fail, then stops the app. The stop
// waits for the write, which puts its report back in the buffer, and
// the final flush writes it, so the stop succeeds with nothing lost.
func TestFinalFlushWritesReportsOfFailedWrite(t *testing.T) {
	dir := t.TempDir()
	t.Setenv("DATA_DIR", dir)

	synctest.Test(t, func(t *testing.T) {
		buf, app := startApp(t)

		created := make(chan *os.File)
		release := make(chan struct{})

		buf.OnFileCreated(func(f *os.File) {
			created <- f

			<-release
		})

		err := buf.Append(map[string]string{"id": "failed"})
		if err != nil {
			t.Fatalf("append report: %v", err)
		}

		held := <-created

		// Closing the file under the write makes the write fail.
		_ = held.Close()

		// The final flush's write goes through.
		buf.OnFileCreated(func(*os.File) {})

		err = stopWhileHeld(t, app, release)
		if err != nil {
			t.Fatalf("stop: %v", err)
		}

		files, err := readReportFiles(dir)
		if err != nil {
			t.Fatalf("read report files: %v", err)
		}

		want := `{"id":"failed"}` + "\n"
		if len(files) != 1 || files[0] != want {
			t.Fatalf("report files = %q, want one holding %q", files, want)
		}
	})
}

// stopWhileHeld stops app while a report file write is held open until
// release is closed, and fails the test if the stop returns before
// that. It then closes release and returns the stop's error. It must
// be called in a synctest bubble.
func stopWhileHeld(
	t *testing.T,
	app *fxtest.App,
	release chan<- struct{},
) error {
	t.Helper()

	stopped := make(chan error, 1)

	go func() { stopped <- app.Stop(t.Context()) }()

	// Wait returns once every other goroutine in the bubble is waiting:
	// the stop has got as far as it can while the write is held.
	synctest.Wait()

	if len(stopped) != 0 {
		t.Error("stop returned while a report file write was held open")
	}

	close(release)

	return <-stopped
}

// startApp starts a Buffer through fx, as cli.RunServer does, with the
// DATA_DIR and DATA_DIR_MAX_BYTES the calling test has set, and returns
// it with the app, for tests that stop the app themselves.
func startApp(t *testing.T) (*reportbuf.Buffer, *fxtest.App) {
	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, app
}

// startBuffer starts a Buffer through fx, as cli.RunServer does, with
// the DATA_DIR and DATA_DIR_MAX_BYTES the calling test has set.
func startBuffer(t *testing.T) *reportbuf.Buffer {
	t.Helper()

	buf, _ := startApp(t)

	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
}

// TestAppendPastCapIsRefused fills the cap with a report not yet
// written. The next report is refused, and the report file already in
// DATA_DIR is kept: it is smaller than a report, so deleting it could
// not make room.
func TestAppendPastCapIsRefused(t *testing.T) {
	report := map[string]string{"id": "cap"}
	dir := t.TempDir()
	earlier := reportFilePath(dir, 1)

	writeBytes(t, earlier, 1)

	t.Setenv("DATA_DIR", dir)
	t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(1+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)
	}

	if !exists(t, earlier) {
		t.Fatal("report file deleted, though that could not make room")
	}
}

// TestOldestReportFileDeletedFirst starts on a data directory holding
// report files from an earlier run, and a file that is not a report,
// which neither counts nor is ever deleted. Nothing is deleted while
// there is room; then only the oldest report file is.
func TestOldestReportFileDeletedFirst(t *testing.T) {
	const fileBytes = 100

	report := map[string]string{"id": "oldest"}
	dir := t.TempDir()
	oldest := reportFilePath(dir, 1)
	kept := []string{
		reportFilePath(dir, 2),
		reportFilePath(dir, 3),
		filepath.Join(dir, "notes.txt"),
	}

	writeBytes(t, oldest, fileBytes)

	for _, path := range kept {
		writeBytes(t, path, fileBytes)
	}

	t.Setenv("DATA_DIR", dir)
	// Room for the three report files and one report.
	t.Setenv("DATA_DIR_MAX_BYTES",
		strconv.Itoa(3*fileBytes+lineBytes(t, report)))

	buf := startBuffer(t)

	err := buf.Append(report)
	if err != nil {
		t.Fatalf("report that fills the cap exactly: %v", err)
	}

	if !exists(t, oldest) {
		t.Fatal("oldest report file deleted while there was room")
	}

	err = buf.Append(report)
	if err != nil {
		t.Fatalf("report past the cap: %v", err)
	}

	if exists(t, oldest) {
		t.Fatal("oldest report file kept when room was needed")
	}

	for _, path := range kept {
		if !exists(t, path) {
			t.Fatalf("%s deleted; only the oldest report file should be", path)
		}
	}
}

// TestStartDeletesFilesPastCap starts on report files past the cap,
// as after the cap is lowered: the oldest are deleted until the rest
// fit.
func TestStartDeletesFilesPastCap(t *testing.T) {
	const fileBytes = 100

	dir := t.TempDir()
	oldest := reportFilePath(dir, 1)
	kept := []string{reportFilePath(dir, 2), reportFilePath(dir, 3)}

	writeBytes(t, oldest, fileBytes)

	for _, path := range kept {
		writeBytes(t, path, fileBytes)
	}

	t.Setenv("DATA_DIR", dir)
	t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(len(kept)*fileBytes))

	startBuffer(t)

	if exists(t, oldest) {
		t.Fatal("oldest report file kept, though the files were past the cap")
	}

	for _, path := range kept {
		if !exists(t, path) {
			t.Fatalf("%s deleted, though the rest fit without it", path)
		}
	}
}

// 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)
	dir := t.TempDir()

	t.Setenv("DATA_DIR", dir)
	// 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)
	}

	err = buf.Append(report)
	if err != nil {
		t.Fatalf("second report, after the first was written: %v", err)
	}

	if !hasReportFile(t, dir) {
		t.Fatal("first report's file deleted, though the second fit beside it")
	}
}

// TestWrittenFilesDeletedToMakeRoom writes one report file after
// another under a small cap. Every report must be taken; files are
// deleted only when the report would not fit beside them, and the
// files kept leave room for it.
func TestWrittenFilesDeletedToMakeRoom(t *testing.T) {
	const (
		maxBytes = 200
		// One file each, which take far more than maxBytes together.
		reports = 50
	)

	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)

	for range reports {
		before := reportFilesBytes(t, dir)

		err := buf.Append(report)
		if err != nil {
			t.Fatalf("with %d bytes of report files: %v", before, err)
		}

		after := reportFilesBytes(t, dir)

		if before+size <= maxBytes && after != before {
			t.Fatalf("files deleted, though the report fit beside "+
				"their %d bytes", before)
		}

		if after+size > maxBytes {
			t.Fatalf("%d bytes of report files kept, leaving no room "+
				"for the report", after)
		}

		err = buf.Flush()
		if err != nil {
			t.Fatalf("flush: %v", err)
		}
	}
}

// TestWrittenFilesDeletedOldestFirst writes three report files, then
// appends a report that fits only once one of them is deleted. The
// first one written must be the one deleted.
func TestWrittenFilesDeletedOldestFirst(t *testing.T) {
	const maxBytes = 1000

	report := map[string]string{"id": "written"}

	t.Setenv("DATA_DIR", t.TempDir())
	t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(maxBytes))

	buf := startBuffer(t)

	var written []string

	buf.OnFileCreated(func(f *os.File) { written = append(written, f.Name()) })

	for range 3 {
		err := buf.Append(report)
		if err != nil {
			t.Fatalf("append report: %v", err)
		}

		err = buf.Flush()
		if err != nil {
			t.Fatalf("flush: %v", err)
		}
	}

	// The three files hold the same report, so they are the same size.
	info, err := os.Stat(written[0])
	if err != nil {
		t.Fatalf("stat %s: %v", written[0], err)
	}

	// A report that fits beside two of the files, so deleting one makes
	// exactly the room it needs.
	pad := maxBytes - 2*int(info.Size()) -
		lineBytes(t, map[string]string{"id": ""})

	err = buf.Append(map[string]string{"id": strings.Repeat("a", pad)})
	if err != nil {
		t.Fatalf("report that needs room: %v", err)
	}

	if exists(t, written[0]) {
		t.Fatal("first report file written kept when room was needed")
	}

	for _, path := range written[1:] {
		if !exists(t, path) {
			t.Fatalf("%s deleted before the first report file written", path)
		}
	}
}

// TestFailedDeletionStillCounts makes deleting the oldest report file
// fail. It is still there, so it still takes room, and the next oldest
// is deleted in its place.
func TestFailedDeletionStillCounts(t *testing.T) {
	const fileBytes = 100

	report := map[string]string{"id": "stuck"}
	dir := t.TempDir()
	stuck := reportFilePath(dir, 1)
	next := reportFilePath(dir, 2)
	newest := reportFilePath(dir, 3)

	for _, path := range []string{stuck, next, newest} {
		writeBytes(t, path, fileBytes)
	}

	t.Setenv("DATA_DIR", dir)
	t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(3*fileBytes))

	buf := startBuffer(t)

	// A directory that is not empty cannot be deleted, even by root,
	// which the tests run as in the backend image.
	err := os.Remove(stuck)
	if err != nil {
		t.Fatalf("remove %s: %v", stuck, err)
	}

	err = os.Mkdir(stuck, 0o750)
	if err != nil {
		t.Fatalf("make directory %s: %v", stuck, err)
	}

	writeBytes(t, filepath.Join(stuck, "file"), 1)

	err = buf.Append(report)
	if err != nil {
		t.Fatalf("report past the cap: %v", err)
	}

	if exists(t, next) {
		t.Fatal("next oldest report file kept: the failed deletion " +
			"counted as making room")
	}

	if !exists(t, newest) {
		t.Fatal("newest report file deleted, though deleting one made room")
	}
}

// TestFileDeletedByHandFreesRoom deletes the oldest report file by
// hand after start. When room is needed, its room counts as freed, so
// no other file is deleted.
func TestFileDeletedByHandFreesRoom(t *testing.T) {
	const fileBytes = 100

	report := map[string]string{"id": "by-hand"}
	dir := t.TempDir()
	gone := reportFilePath(dir, 1)
	kept := reportFilePath(dir, 2)

	writeBytes(t, gone, fileBytes)
	writeBytes(t, kept, fileBytes)

	t.Setenv("DATA_DIR", dir)
	t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(2*fileBytes))

	buf := startBuffer(t)

	err := os.Remove(gone)
	if err != nil {
		t.Fatalf("remove %s: %v", gone, err)
	}

	err = buf.Append(report)
	if err != nil {
		t.Fatalf("report past the cap: %v", err)
	}

	if !exists(t, kept) {
		t.Fatal("report file deleted, though the one deleted by hand " +
			"had made room")
	}
}

// TestFileBeingWrittenIsNeverDeleted holds the write of a report file
// open under a cap with room for two reports, then appends a second
// report and a third that needs room. The file being written is the
// only report file, and it is not deleted to make room, so the third
// report is refused. Once the write is complete, its file is deleted
// when room is needed.
func TestFileBeingWrittenIsNeverDeleted(t *testing.T) {
	report := map[string]string{"id": "writing"}

	t.Setenv("DATA_DIR", t.TempDir())
	t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(2*lineBytes(t, report)))

	buf := startBuffer(t)

	created := make(chan string)
	release := make(chan struct{})

	buf.OnFileCreated(func(f *os.File) {
		created <- f.Name()

		<-release
	})

	err := buf.Append(report)
	if err != nil {
		t.Fatalf("first report: %v", err)
	}

	flushed := make(chan error)

	go func() { flushed <- buf.Flush() }()

	writing := <-created

	// Errorf, not Fatalf, until the write is released, so that a
	// failure here does not leave it held.
	err = buf.Append(report)
	if err != nil {
		t.Errorf("second report: %v", err)
	}

	err = buf.Append(report)
	if !errors.Is(err, reportbuf.ErrFull) {
		t.Errorf("report that needs room: error = %v, want ErrFull", err)
	}

	if !exists(t, writing) {
		t.Error("report file deleted while it was being written")
	}

	// Only this write is held; the writes after it, the final one at
	// stop included, go through.
	buf.OnFileCreated(func(*os.File) {})

	close(release)

	err = <-flushed
	if err != nil {
		t.Fatalf("flush of the first report: %v", err)
	}

	// The second report still waits to be written, so this one fits
	// only once the file just written is deleted.
	err = buf.Append(report)
	if err != nil {
		t.Fatalf("report after the write was complete: %v", err)
	}

	if exists(t, writing) {
		t.Fatal("complete report file kept when room was needed")
	}
}

// TestFailedWriteKeepsItsReports makes a write fail once its file is
// created, under a cap with room for one report. The part of the file
// written is removed, so it takes no room, and the report waits in
// memory for the next flush, which writes it. Once written, it counts
// as its file and no longer as a report waiting, so the cap still holds
// exactly: one more report is taken and the one after it is refused.
func TestFailedWriteKeepsItsReports(t *testing.T) {
	report := map[string]string{"id": "kept"}
	dir := t.TempDir()

	t.Setenv("DATA_DIR", dir)
	t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(lineBytes(t, report)))

	buf := startBuffer(t)

	var failed string

	// Closing the file under the write makes the write fail.
	buf.OnFileCreated(func(f *os.File) {
		failed = f.Name()

		_ = f.Close()
	})

	err := buf.Append(report)
	if err != nil {
		t.Fatalf("report that fills the cap exactly: %v", err)
	}

	err = buf.Flush()
	if err == nil {
		t.Fatal("flush succeeded, though its file was closed under it")
	}

	if exists(t, failed) {
		t.Fatal("file of the failed write kept")
	}

	// Writes from here on, the final one at stop included, succeed.
	buf.OnFileCreated(func(*os.File) {})

	err = buf.Flush()
	if err != nil {
		t.Fatalf("flush after the failed write: %v", err)
	}

	files, err := readReportFiles(dir)
	if err != nil {
		t.Fatalf("read report files: %v", err)
	}

	want := `{"id":"kept"}` + "\n"
	if len(files) != 1 || files[0] != want {
		t.Fatalf("report files = %q, want one holding %q", files, want)
	}

	// The file is deleted to make room if need be.
	err = buf.Append(report)
	if err != nil {
		t.Fatalf("report after the written one: %v", err)
	}

	err = buf.Append(report)
	if !errors.Is(err, reportbuf.ErrFull) {
		t.Fatalf("report past the cap: error = %v, want ErrFull", err)
	}
}

// TestFailedWriteKeepsReportsAppendedDuringIt holds a write open,
// appends a report while it is held, then makes the held write fail.
// Both reports are kept, and the next flush writes them in the order
// they were appended, the held write's first.
func TestFailedWriteKeepsReportsAppendedDuringIt(t *testing.T) {
	dir := t.TempDir()
	t.Setenv("DATA_DIR", dir)

	buf := startBuffer(t)

	created := make(chan *os.File)
	release := make(chan struct{})

	buf.OnFileCreated(func(f *os.File) {
		created <- f

		<-release
	})

	err := buf.Append(map[string]string{"id": "held"})
	if err != nil {
		t.Fatalf("report of the held write: %v", err)
	}

	flushed := make(chan error)

	go func() { flushed <- buf.Flush() }()

	held := <-created

	// Errorf, not Fatalf, until the held write is released, so that a
	// failure here does not leave it held.
	err = buf.Append(map[string]string{"id": "appended"})
	if err != nil {
		t.Errorf("report appended while the write was held: %v", err)
	}

	// Closing the file under the write makes the write fail.
	_ = held.Close()

	close(release)

	err = <-flushed
	if err == nil {
		t.Fatal("held write succeeded, though its file was closed under it")
	}

	// Writes from here on, the final one at stop included, succeed.
	buf.OnFileCreated(func(*os.File) {})

	err = buf.Flush()
	if err != nil {
		t.Fatalf("flush after the failed write: %v", err)
	}

	files, err := readReportFiles(dir)
	if err != nil {
		t.Fatalf("read report files: %v", err)
	}

	want := `{"id":"held"}` + "\n" + `{"id":"appended"}` + "\n"
	if len(files) != 1 || files[0] != want {
		t.Fatalf("report files = %q, want one holding %q", files, want)
	}
}

// TestFailedWritesKeepReportsInOrder holds a write open, then appends a
// report that takes the buffer to FlushSizeThreshold, so that Append
// starts a second write. The held write fails first, then the second,
// which waited for it and took both reports. The next flush writes them
// in the order they were appended, the held write's first.
func TestFailedWritesKeepReportsInOrder(t *testing.T) {
	dir := t.TempDir()
	t.Setenv("DATA_DIR", dir)

	buf := startBuffer(t)

	created := make(chan struct{})
	release := make(chan struct{})

	// Each write is held until release is closed, then fails, as its
	// file is closed under it.
	buf.OnFileCreated(func(f *os.File) {
		created <- struct{}{}

		<-release

		_ = f.Close()
	})

	err := buf.Append(map[string]string{"id": "held"})
	if err != nil {
		t.Fatalf("report of the held write: %v", err)
	}

	flushed := make(chan error)

	go func() { flushed <- buf.Flush() }()

	<-created

	big := strings.Repeat("a", reportbuf.FlushSizeThreshold)

	// Errorf, not Fatalf, until the held write is released, so that a
	// failure here does not leave it held.
	err = buf.Append(map[string]string{"id": big})
	if err != nil {
		t.Errorf("report that starts the second write: %v", err)
	}

	close(release)

	err = <-flushed
	if err == nil {
		t.Error("held write succeeded, though its file was closed under it")
	}

	// The second write has created its file, so it has taken the
	// reports.
	<-created

	// Writes from here on, the final one at stop included, succeed.
	buf.OnFileCreated(func(*os.File) {})

	// Flush waits for the second write to fail before it writes.
	err = buf.Flush()
	if err != nil {
		t.Fatalf("flush after the failed writes: %v", err)
	}

	files, err := readReportFiles(dir)
	if err != nil {
		t.Fatalf("read report files: %v", err)
	}

	want := `{"id":"held"}` + "\n" + `{"id":"` + big + `"}` + "\n"
	if len(files) != 1 || files[0] != want {
		t.Fatalf("%d report files, want one holding the held write's "+
			"report, then the second one", len(files))
	}
}

// TestReportsRefusedWhileWritesFail posts reports through the report
// handler, as the page does. Once a write has failed, a report is
// refused with 500 until a write succeeds again; the reports taken
// before and after are stored, and the one refused is not.
func TestReportsRefusedWhileWritesFail(t *testing.T) {
	dir := t.TempDir()
	t.Setenv("DATA_DIR", dir)

	buf, h := startBufferWithHandlers(t)

	if code := postReport(t, h, "before"); code != http.StatusOK {
		t.Fatalf("report before the failed write: status = %d, want %d",
			code, http.StatusOK)
	}

	// With the data directory gone, the write has nowhere to go.
	err := os.Remove(dir)
	if err != nil {
		t.Fatalf("remove data dir: %v", err)
	}

	err = buf.Flush()
	if err == nil {
		t.Fatal("flush succeeded with the data directory gone")
	}

	if code := postReport(t, h, "refused"); code != http.StatusInternalServerError {
		t.Fatalf("report after the failed write: status = %d, want %d",
			code, http.StatusInternalServerError)
	}

	err = os.Mkdir(dir, 0o750)
	if err != nil {
		t.Fatalf("make data dir: %v", err)
	}

	err = buf.Flush()
	if err != nil {
		t.Fatalf("flush with the data directory back: %v", err)
	}

	if code := postReport(t, h, "after"); code != http.StatusOK {
		t.Fatalf("report after a write succeeded: status = %d, want %d",
			code, http.StatusOK)
	}

	err = buf.Flush()
	if err != nil {
		t.Fatalf("flush of the report after: %v", err)
	}

	files, err := readReportFiles(dir)
	if err != nil {
		t.Fatalf("read report files: %v", err)
	}

	stored := strings.Join(files, "")

	for _, clientID := range []string{"before", "after"} {
		if !strings.Contains(stored, `"clientId":"`+clientID+`"`) {
			t.Errorf("report %q not stored", clientID)
		}
	}

	if strings.Contains(stored, `"clientId":"refused"`) {
		t.Error("refused report stored")
	}
}

// TestHealthCheckFailsWhileWritesFail asks the health check handler
// for its answer around a failed write. It answers 200 with
// "status":"ok" at start, 503 with "status":"error" from the failed
// write until a write succeeds, and 200 with "status":"ok" again after.
func TestHealthCheckFailsWhileWritesFail(t *testing.T) {
	dir := t.TempDir()
	t.Setenv("DATA_DIR", dir)

	buf, h := startBufferWithHandlers(t)

	checkHealthCheck(t, h, "at start", http.StatusOK, "ok")

	// A flush writes a file only when a report is waiting.
	err := buf.Append(map[string]string{"id": "kept"})
	if err != nil {
		t.Fatalf("append: %v", err)
	}

	// With the data directory gone, the write has nowhere to go.
	err = os.Remove(dir)
	if err != nil {
		t.Fatalf("remove data dir: %v", err)
	}

	err = buf.Flush()
	if err == nil {
		t.Fatal("flush succeeded with the data directory gone")
	}

	checkHealthCheck(t, h, "after the failed write",
		http.StatusServiceUnavailable, "error")

	err = os.Mkdir(dir, 0o750)
	if err != nil {
		t.Fatalf("make data dir: %v", err)
	}

	err = buf.Flush()
	if err != nil {
		t.Fatalf("flush with the data directory back: %v", err)
	}

	checkHealthCheck(t, h, "after a write succeeded", http.StatusOK, "ok")
}

// startBufferWithHandlers starts a Buffer and the Handlers that store
// reports in it through fx, as cli.RunServer does, with the DATA_DIR
// the calling test has set.
func startBufferWithHandlers(
	t *testing.T,
) (*reportbuf.Buffer, *handlers.Handlers) {
	t.Helper()

	var (
		buf *reportbuf.Buffer
		h   *handlers.Handlers
	)

	app := fxtest.New(t,
		fx.Provide(
			config.New,
			globals.New,
			handlers.New,
			healthcheck.New,
			logger.New,
			reportbuf.New,
		),
		fx.Populate(&buf, &h),
	)

	app.RequireStart()
	t.Cleanup(app.RequireStop)

	return buf, h
}

// postReport posts a report from clientID to the report handler, as the
// page does, and returns the status of the answer.
func postReport(t *testing.T, h *handlers.Handlers, clientID string) int {
	t.Helper()

	rec := httptest.NewRecorder()
	req := httptest.NewRequestWithContext(t.Context(),
		http.MethodPost, "/api/v1/reports",
		strings.NewReader(`{"clientId":"`+clientID+`","hosts":[]}`),
	)
	req.Header.Set("Content-Type", "application/json")
	h.HandleReport().ServeHTTP(rec, req)

	return rec.Code
}

// checkHealthCheck asks the health check handler for its answer, and
// fails the test, saying when it asked, unless the answer has status
// code wantCode and a "status" field of wantStatus.
func checkHealthCheck(
	t *testing.T,
	h *handlers.Handlers,
	when string,
	wantCode int,
	wantStatus string,
) {
	t.Helper()

	rec := httptest.NewRecorder()
	req := httptest.NewRequestWithContext(t.Context(),
		http.MethodGet, "/.well-known/healthcheck", http.NoBody)
	h.HandleHealthCheck().ServeHTTP(rec, req)

	var body struct {
		Status string `json:"status"`
	}

	err := json.Unmarshal(rec.Body.Bytes(), &body)
	if err != nil {
		t.Fatalf("health check %s: body not JSON: %v (%q)",
			when, err, rec.Body.String())
	}

	if rec.Code != wantCode || body.Status != wantStatus {
		t.Fatalf("health check %s: %d with status %q, want %d with %q",
			when, rec.Code, body.Status, wantCode, wantStatus)
	}
}

// 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)
	}
}

// TestTwoFlushesInOneMillisecond flushes twice within one millisecond,
// as a flush for size and the final flush at shutdown can: each flush
// must write a file of its own, and the files must hold every report.
func TestTwoFlushesInOneMillisecond(t *testing.T) {
	const flushes = 2

	dir := t.TempDir()
	t.Setenv("DATA_DIR", dir)

	buf := startBuffer(t)
	buf.StopClock(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC))

	for id := 1; id <= flushes; id++ {
		err := buf.Append(map[string]int{"id": id})
		if err != nil {
			t.Fatalf("append report %d: %v", id, err)
		}

		err = buf.Flush()
		if err != nil {
			t.Fatalf("flush %d: %v", id, err)
		}
	}

	files, err := readReportFiles(dir)
	if err != nil {
		t.Fatalf("read report files: %v", err)
	}

	if len(files) != flushes {
		t.Fatalf("%d report files after %d flushes", len(files), flushes)
	}

	for id := 1; id <= flushes; id++ {
		want := fmt.Sprintf(`{"id":%d}`+"\n", id)
		if !slices.Contains(files, want) {
			t.Fatalf("no report file holds report %d alone", id)
		}
	}
}

// TestReportFileHoldsTheLinesAppended flushes three reports and reads
// their file back: it must decompress to exactly their JSON lines, in
// the order they were appended.
func TestReportFileHoldsTheLinesAppended(t *testing.T) {
	dir := t.TempDir()
	t.Setenv("DATA_DIR", dir)

	buf := startBuffer(t)

	for _, id := range []int{1, 2, 3} {
		err := buf.Append(map[string]int{"id": id})
		if err != nil {
			t.Fatalf("append report %d: %v", id, err)
		}
	}

	err := buf.Flush()
	if err != nil {
		t.Fatalf("flush: %v", err)
	}

	files, err := readReportFiles(dir)
	if err != nil {
		t.Fatalf("read report files: %v", err)
	}

	want := `{"id":1}` + "\n" + `{"id":2}` + "\n" + `{"id":3}` + "\n"
	if len(files) != 1 || files[0] != want {
		t.Fatalf("report files = %q, want one holding %q", files, want)
	}
}

// TestFlushAtSizeThreshold appends reports until the buffer holds
// FlushSizeThreshold bytes. The append that gets it there must write
// them all to one report file, with no call to Flush and the periodic
// flush a minute away, and no earlier append may write one.
func TestFlushAtSizeThreshold(t *testing.T) {
	dir := t.TempDir()
	t.Setenv("DATA_DIR", dir)

	buf := startBuffer(t)

	appended := appendToThreshold(t, buf)

	// Append writes the file in the background, so wait for it.
	deadline := time.Now().Add(10 * time.Second)

	for {
		files, err := readReportFiles(dir)
		if err == nil && len(files) == 1 && files[0] == appended {
			return
		}

		if time.Now().After(deadline) {
			t.Fatalf("%d report files (error: %v), want one holding the "+
				"%d bytes appended", len(files), err, len(appended))
		}

		time.Sleep(10 * time.Millisecond)
	}
}

// TestEachSizeThresholdStartsAWrite reaches FlushSizeThreshold twice,
// with no call to Flush and the periodic flush a minute away. Append
// must start a write each time, so that the reports end up in two
// report files. The test runs in a synctest bubble, so that
// synctest.Wait returns once each write is done.
func TestEachSizeThresholdStartsAWrite(t *testing.T) {
	dir := t.TempDir()
	t.Setenv("DATA_DIR", dir)

	synctest.Test(t, func(t *testing.T) {
		buf := startBuffer(t)

		var appended string

		for range 2 {
			appended = appendToThreshold(t, buf)

			synctest.Wait()
		}

		files, err := readReportFiles(dir)
		if err != nil {
			t.Fatalf("read report files: %v", err)
		}

		if !slices.Equal(files, []string{appended, appended}) {
			t.Fatalf("%d report files, want two, each holding the %d "+
				"bytes appended to reach the threshold",
				len(files), len(appended))
		}
	})
}

// TestAppendStartsOneWaitingWrite holds a write open, then appends two
// reports of FlushSizeThreshold bytes each. Append starts a write at
// the first, which waits for the held one, and none at the second,
// since that write is still waiting. Every write fails, so a write
// started at the second report would find the reports back in the
// buffer and try them again: once the held write is released, two
// writes must have been started, the held one and one more.
func TestAppendStartsOneWaitingWrite(t *testing.T) {
	t.Setenv("DATA_DIR", t.TempDir())

	synctest.Test(t, func(t *testing.T) {
		buf := startBuffer(t)

		var writes atomic.Int64

		release := make(chan struct{})

		// Each write is held until release is closed, then fails, as its
		// file is closed under it.
		buf.OnFileCreated(func(f *os.File) {
			writes.Add(1)

			<-release

			_ = f.Close()
		})

		err := buf.Append(map[string]string{"id": "one-waiting"})
		if err != nil {
			t.Fatalf("report of the held write: %v", err)
		}

		go func() { _ = buf.Flush() }()

		// Wait returns once the flush's write is held.
		synctest.Wait()

		big := map[string]string{
			"id": strings.Repeat("a", reportbuf.FlushSizeThreshold),
		}

		// Errorf, not Fatalf, until the held write is released, so that a
		// failure here does not leave it held.
		for range 2 {
			err = buf.Append(big)
			if err != nil {
				t.Errorf("report of FlushSizeThreshold bytes: %v", err)
			}
		}

		close(release)

		// Wait returns once every write started is done.
		synctest.Wait()

		started := writes.Load()

		// Writes from here on, the final one at stop included, succeed.
		buf.OnFileCreated(func(*os.File) {})

		if started != 2 {
			t.Fatalf("%d writes started, want 2: the held one and one "+
				"at the threshold", started)
		}
	})
}

// appendToThreshold appends reports to buf until it holds
// FlushSizeThreshold bytes, so that the last append starts writing them
// to a report file, and returns the lines appended.
func appendToThreshold(t *testing.T, buf *reportbuf.Buffer) string {
	t.Helper()

	// Large reports, so the threshold takes a few hundred appends.
	pad := strings.Repeat("a", 64<<10)

	var appended strings.Builder

	for id := 0; appended.Len() < reportbuf.FlushSizeThreshold; id++ {
		report := map[string]any{"id": id, "pad": pad}

		err := buf.Append(report)
		if err != nil {
			t.Fatalf("append report %d: %v", id, err)
		}

		line, err := json.Marshal(report)
		if err != nil {
			t.Fatalf("marshal report %d: %v", id, err)
		}

		appended.Write(line)
		appended.WriteByte('\n')
	}

	return appended.String()
}

// 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
}

// readReportFiles returns the decompressed contents of each report
// file in dir. A file still being written does not decompress, so it
// gives an error.
func readReportFiles(dir string) ([]string, error) {
	files := os.DirFS(dir)

	names, err := fs.Glob(files, "reports-*.jsonl.zst")
	if err != nil {
		return nil, fmt.Errorf("list report files: %w", err)
	}

	dec, err := zstd.NewReader(nil)
	if err != nil {
		return nil, fmt.Errorf("create zstd decoder: %w", err)
	}
	defer dec.Close()

	contents := make([]string, 0, len(names))

	for _, name := range names {
		compressed, readErr := fs.ReadFile(files, name)
		if readErr != nil {
			return nil, fmt.Errorf("read %s: %w", name, readErr)
		}

		data, decErr := dec.DecodeAll(compressed, nil)
		if decErr != nil {
			return nil, fmt.Errorf("decompress %s: %w", name, decErr)
		}

		contents = append(contents, string(data))
	}

	return contents, nil
}

// reportFilePath returns the path in dir of a report file named as
// written on the given day of January 2026, so that a lower day sorts
// as older.
func reportFilePath(dir string, day int) string {
	return filepath.Join(dir,
		fmt.Sprintf("reports-2026-01-%02dT00-00-00.000Z-1.jsonl.zst", day))
}

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 exists(t *testing.T, path string) bool {
	t.Helper()

	_, err := os.Stat(path)
	if errors.Is(err, fs.ErrNotExist) {
		return false
	}

	if err != nil {
		t.Fatalf("stat %s: %v", path, err)
	}

	return true
}

func hasReportFile(t *testing.T, dir string) bool {
	t.Helper()

	entries, err := os.ReadDir(dir)
	if err != nil {
		t.Fatalf("read data dir: %v", err)
	}

	for _, e := range entries {
		if strings.HasSuffix(e.Name(), ".jsonl.zst") {
			info, statErr := e.Info()
			if statErr != nil {
				t.Fatalf("stat %s: %v", e.Name(), statErr)
			}

			if info.Size() > 0 {
				return true
			}
		}
	}

	return false
}