OwnCord/Server/db/audit_writer_test.go

package db_test

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/J3vb/OwnCord/Server/db"
)

// fakeAuditStore records the batches handed to PersistAudits so tests can
// assert on batching behavior without a real database.
type fakeAuditStore struct {
	mu      sync.Mutex
	batches [][]db.AuditEntry
	entries []db.AuditEntry
	err     error // when non-nil, PersistAudits persists nothing
}

func (f *fakeAuditStore) PersistAudits(_ context.Context, entries []db.AuditEntry) (int, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.err != nil {
		return 0, f.err
	}
	cp := append([]db.AuditEntry(nil), entries...)
	f.batches = append(f.batches, cp)
	f.entries = append(f.entries, cp...)
	return len(entries), nil
}

func (f *fakeAuditStore) snapshot() (batches int, entries []db.AuditEntry) {
	f.mu.Lock()
	defer f.mu.Unlock()
	return len(f.batches), append([]db.AuditEntry(nil), f.entries...)
}

// waitForPersisted polls the writer's Stats until the persisted counter
// reaches want or the deadline passes.
func waitForPersisted(t *testing.T, w *db.AuditWriter, want uint64) {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		if persisted, _, _, _ := w.Stats(); persisted >= want {
			return
		}
		time.Sleep(time.Millisecond)
	}
	persisted, dropped, flushes, errs := w.Stats()
	t.Fatalf("timed out waiting for persisted=%d; stats: persisted=%d dropped=%d flushes=%d errors=%d",
		want, persisted, dropped, flushes, errs)
}

func TestAuditWriter_BatchFlush(t *testing.T) {
	store := &fakeAuditStore{}
	// flushEvery is huge so the only flush trigger is the batch filling up —
	// this pins that a full batch goes to the store as one PersistAudits call.
	w := db.NewAuditWriter(store, 16, 4, time.Hour)
	defer w.Stop(context.Background())

	// Enqueue before Start so the run loop sees all four immediately.
	for i := int64(1); i <= 4; i++ {
		w.Enqueue(i, fmt.Sprintf("action_%d", i), "user", i*10, fmt.Sprintf("detail_%d", i))
	}
	w.Start(context.Background())
	waitForPersisted(t, w, 4)

	batches, entries := store.snapshot()
	if batches != 1 {
		t.Errorf("store received %d batches, want 1 (batch-size flush)", batches)
	}
	if len(entries) != 4 {
		t.Fatalf("store received %d entries, want 4", len(entries))
	}
	// Field mapping and order must survive the queue round-trip.
	for i, e := range entries {
		n := int64(i + 1)
		if e.ActorID != n || e.Action != fmt.Sprintf("action_%d", n) ||
			e.TargetType != "user" || e.TargetID != n*10 || e.Detail != fmt.Sprintf("detail_%d", n) {
			t.Errorf("entry %d = %+v, want actor=%d action=action_%d target=user/%d detail=detail_%d",
				i, e, n, n, n*10, n)
		}
	}
}

func TestAuditWriter_DropOnFullQueueLogsError(t *testing.T) {
	store := &fakeAuditStore{}
	// Queue of one, not yet started: the second enqueue must drop.
	w := db.NewAuditWriter(store, 1, 50, time.Hour)

	out := captureLogs(t, func() {
		w.Enqueue(1, "kept_action", "user", 1, "kept detail")
		w.Enqueue(7, "dropped_action", "user", 42, "secret detail")
	})

	if _, dropped, _, _ := w.Stats(); dropped != 1 {
		t.Errorf("dropped counter = %d, want 1", dropped)
	}
	// D8: the drop must not be silent and must identify what was lost.
	for _, want := range []string{
		"audit log dropped",
		"action=dropped_action",
		"actor_id=7",
		"target_type=user",
		"target_id=42",
	} {
		if !strings.Contains(out, want) {
			t.Errorf("drop log missing %q; got: %s", want, out)
		}
	}
	// The detail string must not leak into logs.
	if strings.Contains(out, "secret detail") {
		t.Errorf("detail string leaked into drop log: %s", out)
	}

	// The queued entry must still land once the writer runs.
	w.Start(context.Background())
	w.Stop(context.Background())
	if persisted, _, _, _ := w.Stats(); persisted != 1 {
		t.Errorf("persisted = %d, want 1 (the non-dropped entry)", persisted)
	}
}

func TestAuditWriter_DrainOnStop(t *testing.T) {
	store := &fakeAuditStore{}
	// Neither flush trigger can fire (batch 50, ticker 1h): everything must
	// be flushed by Stop's drain.
	w := db.NewAuditWriter(store, 64, 50, time.Hour)
	w.Start(context.Background())
	for i := range int64(10) {
		w.Enqueue(i, "drain_action", "user", i, "")
	}
	w.Stop(context.Background())

	persisted, dropped, _, _ := w.Stats()
	if persisted != 10 || dropped != 0 {
		t.Errorf("persisted=%d dropped=%d, want 10/0", persisted, dropped)
	}
	if _, entries := store.snapshot(); len(entries) != 10 {
		t.Errorf("store received %d entries after Stop, want 10", len(entries))
	}
}

// slowAuditStore models a store whose flushes take a while (a slow/stalled
// disk). It tracks whether any flush ran after Close was called — mirroring
// main.go closing the database right after AuditWriter.Stop returns.
type slowAuditStore struct {
	mu           sync.Mutex
	delay        time.Duration
	closed       bool
	persisted    int
	flushAfter   int // entries flushed after Close (a correctness violation)
	flushesAfter int
}

func (s *slowAuditStore) PersistAudits(_ context.Context, entries []db.AuditEntry) (int, error) {
	s.mu.Lock()
	closedAtEntry := s.closed
	delay := s.delay
	s.mu.Unlock()

	if delay > 0 {
		time.Sleep(delay)
	}

	s.mu.Lock()
	defer s.mu.Unlock()
	if closedAtEntry || s.closed {
		s.flushesAfter++
		s.flushAfter += len(entries)
	}
	s.persisted += len(entries)
	return len(entries), nil
}

func (s *slowAuditStore) close() {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.closed = true
}

func (s *slowAuditStore) stats() (persisted, flushAfter, flushesAfter int) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.persisted, s.flushAfter, s.flushesAfter
}

// TestAuditWriter_StopWaitsForGoroutineExit pins the fixed contract: Stop must
// not return until the run goroutine has finished its in-flight flush, even
// when the Stop context expires first. The store flush (200ms) far outlasts
// the Stop ctx (20ms); the old select{done|ctx.Done} would have returned at
// ~20ms with nothing persisted. The fix must return only after the flush
// completes, with every entry persisted.
func TestAuditWriter_StopWaitsForGoroutineExit(t *testing.T) {
	store := &slowAuditStore{delay: 200 * time.Millisecond}
	// Neither the batch (50) nor the ticker (1h) can flush; only Stop's drain
	// flushes, so the in-flight flush is deterministic.
	w := db.NewAuditWriter(store, 64, 50, time.Hour)
	w.Start(context.Background())
	for i := range int64(5) {
		w.Enqueue(i, "slow_action", "user", i, "")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
	defer cancel()

	start := time.Now()
	w.Stop(ctx)
	elapsed := time.Since(start)

	if elapsed < 150*time.Millisecond {
		t.Errorf("Stop returned after %v, want it to block for the ~200ms flush "+
			"(it must not abandon the goroutine when ctx expires)", elapsed)
	}
	persisted, flushAfter, _ := store.stats()
	if persisted != 5 {
		t.Errorf("persisted=%d, want 5 (Stop must wait for the flush to finish)", persisted)
	}
	if flushAfter != 0 {
		t.Errorf("flushAfter=%d, want 0 (nothing was closed yet)", flushAfter)
	}
	if p, _, _, _ := w.Stats(); p != 5 {
		t.Errorf("writer persisted counter = %d, want 5", p)
	}
}

// TestAuditWriter_EnqueueAfterStopDropsLoudly locks D8 ("a drop is never
// silent") against the post-shutdown race: main.go's deferred Stop can
// return while a WS handler on another goroutine is still mid-flight (hub
// GracefulStop and http.Server.Shutdown do not wait for hijacked WebSocket
// conns), so Enqueue can be called after run() has fully exited. Before the
// fix that entry landed silently in the still-buffered, now-unread queue: no
// dropped-counter bump, no log line.
func TestAuditWriter_EnqueueAfterStopDropsLoudly(t *testing.T) {
	store := &fakeAuditStore{}
	w := db.NewAuditWriter(store, 64, 50, time.Hour)
	w.Start(context.Background())
	w.Stop(context.Background())

	_, droppedBefore, _, _ := w.Stats()

	out := captureLogs(t, func() {
		w.Enqueue(9, "post_stop_action", "user", 99, "secret detail")
	})

	_, droppedAfter, _, _ := w.Stats()
	if droppedAfter != droppedBefore+1 {
		t.Errorf("dropped counter = %d, want %d (post-Stop Enqueue must count as a loud drop)", droppedAfter, droppedBefore+1)
	}
	for _, want := range []string{
		"audit log dropped",
		"action=post_stop_action",
		"actor_id=9",
		"target_type=user",
		"target_id=99",
	} {
		if !strings.Contains(out, want) {
			t.Errorf("post-Stop drop log missing %q; got: %s", want, out)
		}
	}
	if strings.Contains(out, "secret detail") {
		t.Errorf("detail string leaked into post-Stop drop log: %s", out)
	}

	// The entry must genuinely never reach the store — no goroutine is left
	// to read the queue.
	if batches, _ := store.snapshot(); batches != 0 {
		t.Errorf("post-Stop entry reached the store via %d batch(es); want it dropped, not written", batches)
	}
}

// TestAuditWriter_StopDrainsBeforeStoreClose reproduces main.go's LIFO
// shutdown ordering (AuditWriter.Stop, then database.Close) and asserts the
// fix: because Stop returns only after the goroutine exits, no flush can run
// after the store is closed — so a slow-disk shutdown never writes into a
// closed pool (the D8 audit-loss race).
func TestAuditWriter_StopDrainsBeforeStoreClose(t *testing.T) {
	store := &slowAuditStore{delay: 200 * time.Millisecond}
	w := db.NewAuditWriter(store, 64, 50, time.Hour)
	w.Start(context.Background())
	for i := range int64(5) {
		w.Enqueue(i, "shutdown_action", "user", i, "")
	}

	// Stop ctx expires long before the flush finishes, as in a >5s disk stall.
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
	defer cancel()
	w.Stop(ctx)

	// Mirror main.go: close the store immediately after Stop returns.
	store.close()

	persisted, flushAfter, flushesAfter := store.stats()
	if flushAfter != 0 || flushesAfter != 0 {
		t.Errorf("%d entries across %d flushes ran after store close; want 0 "+
			"(Stop must fully drain before the DB is closed)", flushAfter, flushesAfter)
	}
	if persisted != 5 {
		t.Errorf("persisted=%d, want 5 before store close", persisted)
	}
}

func TestAuditWriter_FlushFailureCountsAndLogs(t *testing.T) {
	store := &fakeAuditStore{err: errors.New("disk on fire")}
	w := db.NewAuditWriter(store, 16, 50, time.Hour)

	out := captureLogs(t, func() {
		w.Start(context.Background())
		w.Enqueue(1, "lost_action", "user", 1, "")
		w.Enqueue(2, "lost_action", "user", 2, "")
		w.Stop(context.Background())
	})

	if _, _, _, errs := w.Stats(); errs != 2 {
		t.Errorf("errors counter = %d, want 2", errs)
	}
	for _, want := range []string{"flush lost audit entries", "disk on fire"} {
		if !strings.Contains(out, want) {
			t.Errorf("flush-failure log missing %q; got: %s", want, out)
		}
	}
}

func TestAuditWriter_NilReceiverIsSafe(t *testing.T) {
	var w *db.AuditWriter
	w.Enqueue(1, "a", "user", 1, "") // must not panic
	w.Stop(context.Background())     // must not panic
}

func TestNewAuditWriter_PanicsOnNilStore(t *testing.T) {
	defer func() {
		if recover() == nil {
			t.Fatal("NewAuditWriter(nil, ...) did not panic")
		}
	}()
	db.NewAuditWriter(nil, 0, 0, 0)
}

func TestAuditWriter_ConcurrentEnqueue(t *testing.T) {
	store := &fakeAuditStore{}
	w := db.NewAuditWriter(store, 4096, 32, time.Millisecond)
	w.Start(context.Background())

	const goroutines, perGoroutine = 8, 250
	var wg sync.WaitGroup
	for g := range goroutines {
		wg.Add(1)
		go func(g int) {
			defer wg.Done()
			for i := range perGoroutine {
				w.Enqueue(int64(g), "concurrent_action", "user", int64(i), "")
			}
		}(g)
	}
	wg.Wait()
	w.Stop(context.Background())

	persisted, dropped, _, _ := w.Stats()
	if persisted+dropped != goroutines*perGoroutine {
		t.Errorf("persisted(%d)+dropped(%d) = %d, want %d",
			persisted, dropped, persisted+dropped, goroutines*perGoroutine)
	}
	if _, entries := store.snapshot(); uint64(len(entries)) != persisted {
		t.Errorf("store received %d entries, want %d (persisted counter)", len(entries), persisted)
	}
}

// ─── PersistAudits (batch insert + per-row fallback) ─────────────────────────

func TestPersistAudits_SingleTransaction(t *testing.T) {
	database := newAdminTestDB(t)
	uid := seedUser(t, database, "batchactor")

	entries := []db.AuditEntry{
		{ActorID: uid, Action: "first", TargetType: "user", TargetID: 1, Detail: "d1"},
		{ActorID: uid, Action: "second", TargetType: "channel", TargetID: 2, Detail: "d2"},
		{ActorID: uid, Action: "third", TargetType: "server", TargetID: 0, Detail: ""},
	}
	persisted, err := database.PersistAudits(context.Background(), entries)
	if err != nil {
		t.Fatalf("PersistAudits() error: %v", err)
	}
	if persisted != 3 {
		t.Fatalf("PersistAudits() = %d, want 3", persisted)
	}

	got, err := database.GetAuditLog(context.Background(), 10, 0)
	if err != nil {
		t.Fatalf("GetAuditLog() error: %v", err)
	}
	if len(got) != 3 {
		t.Fatalf("GetAuditLog() = %d entries, want 3", len(got))
	}
	// Newest-first: the last inserted row comes back first.
	if got[0].Action != "third" || got[2].Action != "first" {
		t.Errorf("unexpected order: got[0]=%q got[2]=%q, want third/first", got[0].Action, got[2].Action)
	}
	if got[1].Detail != "d2" || got[1].TargetType != "channel" || got[1].TargetID != 2 {
		t.Errorf("middle entry = %+v, want action=second target=channel/2 detail=d2", got[1])
	}
}

func TestPersistAudits_PoisonRowFallsBackPerRow(t *testing.T) {
	database := newAdminTestDB(t)
	uid := seedUser(t, database, "poisonactor")

	// The admin test schema declares actor_id REFERENCES users(id) and Open
	// enables foreign_keys, so a nonexistent actor poisons the transaction.
	entries := []db.AuditEntry{
		{ActorID: uid, Action: "good_one", TargetType: "user", TargetID: 1},
		{ActorID: 999999, Action: "poison", TargetType: "user", TargetID: 2},
		{ActorID: uid, Action: "good_two", TargetType: "user", TargetID: 3},
	}
	persisted, err := database.PersistAudits(context.Background(), entries)
	if err == nil {
		t.Error("PersistAudits() error = nil, want the poison row's error")
	}
	if persisted != 2 {
		t.Fatalf("PersistAudits() = %d, want 2 (good rows land despite poison row)", persisted)
	}

	got, dbErr := database.GetAuditLog(context.Background(), 10, 0)
	if dbErr != nil {
		t.Fatalf("GetAuditLog() error: %v", dbErr)
	}
	if len(got) != 2 {
		t.Fatalf("GetAuditLog() = %d entries, want 2", len(got))
	}
	if got[0].Action != "good_two" || got[1].Action != "good_one" {
		t.Errorf("surviving actions = %q, %q; want good_two, good_one", got[0].Action, got[1].Action)
	}
}

func TestPersistAudits_EmptyBatch(t *testing.T) {
	database := newAdminTestDB(t)
	persisted, err := database.PersistAudits(context.Background(), nil)
	if err != nil || persisted != 0 {
		t.Errorf("PersistAudits(nil) = (%d, %v), want (0, nil)", persisted, err)
	}
}

// ─── WriteAudit routing (sync fallback vs installed writer) ──────────────────

// TestWriteAudit_SynchronousWithoutWriter pins the token CLI contract: a bare
// *DB with no writer installed writes audit entries synchronously, so the
// entry is visible the moment WriteAudit returns.
func TestWriteAudit_SynchronousWithoutWriter(t *testing.T) {
	database := newAdminTestDB(t)
	uid := seedUser(t, database, "syncactor")

	db.WriteAudit(context.Background(), database, uid, "cli_action", "api_token", 5, "label")

	got, err := database.GetAuditLog(context.Background(), 10, 0)
	if err != nil {
		t.Fatalf("GetAuditLog() error: %v", err)
	}
	if len(got) != 1 || got[0].Action != "cli_action" {
		t.Fatalf("GetAuditLog() = %+v, want exactly the synchronously written cli_action", got)
	}
}

// TestWriteAudit_AsyncWithInstalledWriter verifies the seam: once main.go
// installs a writer on the *DB, WriteAudit enqueues instead of inserting —
// the row only lands when the writer flushes (here forced via Stop's drain).
func TestWriteAudit_AsyncWithInstalledWriter(t *testing.T) {
	database := newAdminTestDB(t)
	uid := seedUser(t, database, "asyncactor")

	// Neither flush trigger can fire before Stop, making "not yet written"
	// deterministic rather than a timing accident.
	w := db.NewAuditWriter(database, 16, 50, time.Hour)
	w.Start(context.Background())
	database.SetAuditWriter(w)

	db.WriteAudit(context.Background(), database, uid, "async_action", "user", uid, "detail")

	got, err := database.GetAuditLog(context.Background(), 10, 0)
	if err != nil {
		t.Fatalf("GetAuditLog() error: %v", err)
	}
	if len(got) != 0 {
		t.Fatalf("entry visible before flush: %+v — WriteAudit did not take the async path", got)
	}

	w.Stop(context.Background())
	got, err = database.GetAuditLog(context.Background(), 10, 0)
	if err != nil {
		t.Fatalf("GetAuditLog() after Stop error: %v", err)
	}
	if len(got) != 1 || got[0].Action != "async_action" || got[0].Detail != "detail" {
		t.Fatalf("GetAuditLog() after Stop = %+v, want the drained async_action entry", got)
	}
}