package console

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/objstore"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/usage"
)

// drainFixture is a console whose release store is a PLAIN DirStore.
//
// The shared fixture wraps its store in a struct that embeds release.Store as
// an INTERFACE, which hides the two optional interfaces this whole file is
// about: PageLister, which is how the drain caps one app's turn (§7.3), and
// ModLister, which is how expired result objects are collected. A test on the
// wrapper would silently exercise the degraded fallback in both cases and
// prove nothing about the path a deployment actually runs.
//
// It returns the bucket root because two tests have to AGE a file, and mtime
// is what DirStore's ListMod reads.
func drainFixture(t *testing.T) (*consoleFixture, *objstore.MemScanner, string) {
	t.Helper()
	fx := newConsoleFixture(t)
	dir := t.TempDir()
	rel := release.NewDirStore(dir)
	fx.rel = rel
	fx.srv.cfg.Releases = rel
	sc := objstore.NewMemScanner()
	fx.srv.cfg.StoreScanner = sc
	return fx, sc, dir
}

func drainBucket(acct, app string) string { return "carlos-" + acct + "-" + app }

// enqueueFor puts one item in the bucket AND in the server's index, which is
// what the door does.
func enqueueFor(t *testing.T, fx *consoleFixture, acct, app, key string, at time.Time) {
	t.Helper()
	if _, err := objstore.Enqueue(context.Background(), fx.rel, fx.srv.scanQueue, acct, app,
		&objstore.QueueItem{Key: key, By: "test"}, at); err != nil {
		t.Fatalf("enqueue %s: %v", key, err)
	}
}

// enqueueStranded is the crash §7.1 describes: the item object is written and
// the process dies before the index is bumped. Modelled by bumping a THROWAWAY
// index, so the bucket has the item and this console has never heard of it.
func enqueueStranded(t *testing.T, fx *consoleFixture, acct, app, key string, at time.Time) {
	t.Helper()
	if _, err := objstore.Enqueue(context.Background(), fx.rel, objstore.NewQueueIndex(), acct, app,
		&objstore.QueueItem{Key: key, By: "test"}, at); err != nil {
		t.Fatalf("enqueue %s: %v", key, err)
	}
}

func queueLen(t *testing.T, fx *consoleFixture, acct, app string) int {
	t.Helper()
	n, err := objstore.QueueDepth(context.Background(), fx.rel, acct, app)
	if err != nil {
		t.Fatalf("queue depth: %v", err)
	}
	return n
}

func resultKeys(t *testing.T, fx *consoleFixture, acct, app string) []string {
	t.Helper()
	keys, err := fx.rel.List(context.Background(), objstore.ScanResultsPrefix(acct, app))
	if err != nil {
		t.Fatalf("list results: %v", err)
	}
	return keys
}

// countingStore counts listings, which is the only way to see the difference
// between "the drain refused" and "the drain did the work and objstore
// refused".
//
// It embeds the CONCRETE *release.DirStore rather than the interface, so
// ListMod and ListDirs are still promoted — embedding the interface is exactly
// the trap drainFixture's comment describes.
type countingStore struct {
	*release.DirStore
	lists atomic.Int64
}

type scanUsageCountingStore struct {
	*release.DirStore
	puts atomic.Int64
}

func (s *scanUsageCountingStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if strings.HasPrefix(key, usage.ScanPrefix) {
		s.puts.Add(1)
	}
	return s.DirStore.Put(ctx, key, body, ifMatch)
}

func (c *countingStore) List(ctx context.Context, prefix string) ([]string, error) {
	c.lists.Add(1)
	return c.DirStore.List(ctx, prefix)
}

func (c *countingStore) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) {
	c.lists.Add(1)
	return c.DirStore.ListPage(ctx, prefix, startAfter, limit)
}

// setDeploymentScanning flips the deployment switch either way.
//
// enableDeploymentScanning is CREATE-ONLY — it returns early when a record
// already exists, which is right for a fixture arming a fresh deployment and
// useless for re-enabling one that was turned off.
func setDeploymentScanning(t *testing.T, fx *consoleFixture, on bool) {
	t.Helper()
	ctx := context.Background()
	av, etag, err := release.GetAntivirus(ctx, fx.rel)
	if err != nil {
		t.Fatal(err)
	}
	if av == nil {
		av = &release.AntivirusSettings{Engine: "clamav"}
	}
	av.Enabled = on
	av.At = time.Now().UTC()
	av.By = "test"
	if _, err := release.PutAntivirus(ctx, fx.rel, av, etag); err != nil {
		t.Fatal(err)
	}
}

// revokeScanning is the operator taking the per-store grant away AFTER the
// record exists, which enableStore cannot do: it writes with an empty etag and
// so conflicts with anything already there.
func revokeScanning(t *testing.T, fx *consoleFixture, acct, app string) {
	t.Helper()
	ctx := context.Background()
	rec, etag, err := objstore.Get(ctx, fx.rel, acct, app)
	if err != nil {
		t.Fatal(err)
	}
	if rec == nil {
		t.Fatalf("no record for %s/%s to revoke", acct, app)
	}
	rec.Scanning = nil
	if _, err := objstore.Put(ctx, fx.rel, acct, app, rec, etag); err != nil {
		t.Fatal(err)
	}
}

// TestBootRebuildFindsStrandedItem is the whole argument for the rebuild
// existing, and it is written so that the rebuild is the ONLY difference
// between the two halves.
//
// A tick runs first and must do nothing: the drain only visits apps the index
// names, and nothing named this one. Then the rebuild runs, and the same tick
// scans. Without the "before" half this would pass against a drain that walked
// every app on the deployment every 30s — which is the design the index exists
// to avoid.
func TestBootRebuildFindsStrandedItem(t *testing.T) {
	ctx := context.Background()
	fx, sc, _ := drainFixture(t)
	enableStore(t, fx, fixtureAcct, fixtureApp, true)
	sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/x.pdf", "etag-1", time.Now(), 10)
	enqueueStranded(t, fx, fixtureAcct, fixtureApp, "uploads/x.pdf", time.Now())

	if d := fx.srv.ScanQueueDepth(); d != 0 {
		t.Fatalf("index depth %d before the rebuild — the item was not stranded, so this test proves nothing", d)
	}
	if err := fx.srv.scanDrainTick(ctx); err != nil {
		t.Fatal(err)
	}
	if n := sc.Scans("uploads/x.pdf"); n != 0 {
		t.Fatalf("a tick scanned %d time(s) for an app the index never named", n)
	}

	if err := fx.srv.rebuildScanQueue(ctx); err != nil {
		t.Fatal(err)
	}
	if d := fx.srv.ScanQueueDepth(); d != 1 {
		t.Fatalf("index depth %d after the rebuild, want 1 — nothing would ever list that prefix again", d)
	}
	if err := fx.srv.scanDrainTick(ctx); err != nil {
		t.Fatal(err)
	}
	if n := sc.Scans("uploads/x.pdf"); n != 1 {
		t.Fatalf("scanned %d times after the rebuild, want 1", n)
	}
	if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 0 {
		t.Errorf("%d item(s) left after a successful drain", n)
	}
}

// TestBootRebuildCollectsExpiredItemsAndResults covers BOTH prefixes on the
// same 24h TTL, because they expire for the same reason: after a sweep cycle
// the ordinary walk has had its chance at the object, and a stale on-demand
// answer is worth less than none.
//
// The result object matters more than the item. A stranded item is drained on
// the next tick; a stranded result is found by nothing, blocks the app-name
// claim gate, and would sit in the bucket forever.
func TestBootRebuildCollectsExpiredItemsAndResults(t *testing.T) {
	ctx := context.Background()
	fx, _, dir := drainFixture(t)
	enableStore(t, fx, fixtureAcct, fixtureApp, true)

	old := time.Now().Add(-objstore.QueueTTL - time.Hour)
	enqueueStranded(t, fx, fixtureAcct, fixtureApp, "uploads/stale.pdf", old)
	if err := objstore.WriteScanResult(ctx, fx.rel, fixtureAcct, fixtureApp, &objstore.ScanResult{
		Key: "uploads/untaggable.pdf", Verdict: "clean", Version: "etag-9@x",
		Engine: "clamav", Signature: "sig", At: old.UTC(),
	}); err != nil {
		t.Fatal(err)
	}
	// ListMod reads MTIME, not the At field inside the JSON, so the file has
	// to actually be old.
	resKey := objstore.ScanResultKey(fixtureAcct, fixtureApp, "uploads/untaggable.pdf")
	if err := os.Chtimes(filepath.Join(dir, filepath.FromSlash(resKey)), old, old); err != nil {
		t.Fatal(err)
	}

	if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 1 {
		t.Fatalf("%d item(s) staged, want 1", n)
	}
	if got := resultKeys(t, fx, fixtureAcct, fixtureApp); len(got) != 1 {
		t.Fatalf("%d result(s) staged, want 1", len(got))
	}

	if err := fx.srv.rebuildScanQueue(ctx); err != nil {
		t.Fatal(err)
	}
	if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 0 {
		t.Errorf("%d expired item(s) survived the rebuild", n)
	}
	if got := resultKeys(t, fx, fixtureAcct, fixtureApp); len(got) != 0 {
		t.Errorf("expired result(s) survived the rebuild: %v", got)
	}
	// And nothing is left pointing at an app with no work.
	if d := fx.srv.ScanQueueDepth(); d != 0 {
		t.Errorf("index depth %d after collecting everything, want 0", d)
	}
}

// TestRebuildKeepsAnItemInsideItsTTL is the other half of the test above, and
// it is here because "collects expired items" passes trivially against a
// rebuild that deletes everything it finds.
func TestRebuildKeepsAnItemInsideItsTTL(t *testing.T) {
	ctx := context.Background()
	fx, _, _ := drainFixture(t)
	enableStore(t, fx, fixtureAcct, fixtureApp, true)
	enqueueStranded(t, fx, fixtureAcct, fixtureApp, "uploads/fresh.pdf", time.Now().Add(-time.Hour))

	if err := fx.srv.rebuildScanQueue(ctx); err != nil {
		t.Fatal(err)
	}
	if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 1 {
		t.Fatalf("%d item(s) after the rebuild, want the one that is an hour old", n)
	}
	if d := fx.srv.ScanQueueDepth(); d != 1 {
		t.Errorf("index depth %d, want 1 — a kept item with no index entry is stranded again", d)
	}
}

// TestDrainRunsWhileSweepIsBlocked is §7.2 stated as something observable.
//
// "TestDrainIsItsOwnGoroutine" as a name would pass against anything at all —
// against a queue phase inside scanPass, even. So the sweep is PARKED, inside
// the lister, for the whole duration of a drain, and the drain has to complete
// anyway. That is the only formulation that fails if the two are ever folded
// together, which is what draft 1 proposed and what would cost this path the
// seconds of latency it exists for.
func TestDrainRunsWhileSweepIsBlocked(t *testing.T) {
	ctx := context.Background()
	fx, sc, _ := drainFixture(t)
	enableStore(t, fx, fixtureAcct, fixtureApp, true)
	sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/urgent.pdf", "etag-u", time.Now(), 10)
	enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/urgent.pdf", time.Now())

	entered := make(chan struct{}, 1)
	release := make(chan struct{})
	fx.srv.storeListerFor = func(context.Context, *objstore.Provisioned) (objstore.ObjectLister, error) {
		select {
		case entered <- struct{}{}:
		default:
		}
		<-release
		return &countingLister{}, nil
	}
	swept := make(chan struct{})
	go func() {
		_ = fx.srv.scanPass(ctx)
		close(swept)
	}()

	select {
	case <-entered:
	case <-time.After(5 * time.Second):
		close(release)
		t.Fatal("the sweep never reached the lister — the test never parked it")
	}

	// The sweep is now stopped dead inside scanOneApp. The drain must not be.
	if err := fx.srv.scanDrainTick(ctx); err != nil {
		t.Fatalf("drain tick while the sweep is parked: %v", err)
	}
	if n := sc.Scans("uploads/urgent.pdf"); n != 1 {
		t.Fatalf("scanned %d times while the sweep held its budget, want 1", n)
	}

	close(release)
	<-swept
}

// TestDrainRoundRobinFairness: one app's flood must not delay another app's
// single item beyond one tick.
//
// FIFO across the deployment is the design this refuses. The flooder is given
// more items than the per-app cap on purpose, so that a drain which emptied
// one app before moving on would still be inside the flooder when the tick
// ended, and the other app's item would go unanswered.
func TestDrainRoundRobinFairness(t *testing.T) {
	ctx := context.Background()
	fx, sc, _ := drainFixture(t)
	enableStore(t, fx, fixtureAcct, fixtureApp, true)
	enableStore(t, fx, fixtureAcct, "flooder", true)

	flood := scanDrainPerApp + 10
	for i := 0; i < flood; i++ {
		key := fmt.Sprintf("uploads/flood-%03d.bin", i)
		sc.PutObject(drainBucket(fixtureAcct, "flooder"), key, "etag-f", time.Now(), 10)
		enqueueFor(t, fx, fixtureAcct, "flooder", key, time.Now())
	}
	sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/urgent.pdf", "etag-u", time.Now(), 10)
	enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/urgent.pdf", time.Now())

	if err := fx.srv.scanDrainTick(ctx); err != nil {
		t.Fatal(err)
	}
	if n := sc.Scans("uploads/urgent.pdf"); n != 1 {
		t.Fatalf("the urgent item scanned %d times in the tick that also met a flood of %d, want 1", n, flood)
	}
	// And the cap held: the flooder did not get the whole queue in one turn.
	left := queueLen(t, fx, fixtureAcct, "flooder")
	if left != flood-scanDrainPerApp {
		t.Errorf("flooder has %d item(s) left, want %d — the per-app cap is not being applied",
			left, flood-scanDrainPerApp)
	}
	// A SECOND tick, because the release rule needs one: DrainApp reports a
	// consumed page as "not empty" — a full page may have more behind it — so
	// an app's entry goes on the first tick that lists nothing, not on the
	// tick that emptied it. After tick two the urgent app is released and the
	// flooder, which still had a partial page, is not.
	if err := fx.srv.scanDrainTick(ctx); err != nil {
		t.Fatal(err)
	}
	if n := queueLen(t, fx, fixtureAcct, "flooder"); n != 0 {
		t.Errorf("flooder has %d item(s) after two ticks, want 0", n)
	}
	if d := fx.srv.ScanQueueDepth(); d != 1 {
		t.Errorf("index depth %d after two ticks, want 1 — the drained app should have been released and the flooder kept", d)
	}
}

// TestDrainAgeSweepsResultObjects: per VISITED app, so a console that never
// restarts does not accumulate them.
//
// Nothing else deletes these. The drain removes a result only when a later
// scan records a real verdict on the same key, and for an object that is
// permanently untaggable — the ten-tag case — that never happens.
func TestDrainAgeSweepsResultObjects(t *testing.T) {
	ctx := context.Background()
	fx, sc, dir := drainFixture(t)
	enableStore(t, fx, fixtureAcct, fixtureApp, true)
	sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/live.pdf", "etag-l", time.Now(), 10)
	enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/live.pdf", time.Now())

	old := time.Now().Add(-objstore.QueueTTL - time.Hour)
	for _, k := range []string{"uploads/old.pdf", "uploads/recent.pdf"} {
		if err := objstore.WriteScanResult(ctx, fx.rel, fixtureAcct, fixtureApp, &objstore.ScanResult{
			Key: k, Verdict: "clean", Version: "etag-x@y", Engine: "clamav", Signature: "sig",
			At: time.Now().UTC(),
		}); err != nil {
			t.Fatal(err)
		}
	}
	oldPath := filepath.Join(dir, filepath.FromSlash(
		objstore.ScanResultKey(fixtureAcct, fixtureApp, "uploads/old.pdf")))
	if err := os.Chtimes(oldPath, old, old); err != nil {
		t.Fatal(err)
	}

	if err := fx.srv.scanDrainTick(ctx); err != nil {
		t.Fatal(err)
	}
	got := resultKeys(t, fx, fixtureAcct, fixtureApp)
	if len(got) != 1 {
		t.Fatalf("results after the tick = %v, want only the recent one", got)
	}
	// The one inside its TTL survives — a collector that dropped both would
	// pass a test that only counted the expired one.
	if want := objstore.ScanResultKey(fixtureAcct, fixtureApp, "uploads/recent.pdf"); got[0] != want {
		t.Errorf("survivor = %q, want %q", got[0], want)
	}
}

// TestQueueDepthReported: §4.4 wants it deployment-wide and §10 wants it per
// app, and they are different numbers measured in different ways.
func TestQueueDepthReported(t *testing.T) {
	ctx := context.Background()
	fx, _, _ := drainFixture(t)
	enableStore(t, fx, fixtureAcct, fixtureApp, true)
	enableStore(t, fx, fixtureAcct, "second", true)

	if d := fx.srv.ScanQueueDepth(); d != 0 {
		t.Fatalf("depth %d on an idle deployment, want 0", d)
	}
	enqueueFor(t, fx, fixtureAcct, fixtureApp, "a.pdf", time.Now())
	enqueueFor(t, fx, fixtureAcct, fixtureApp, "b.pdf", time.Now())
	enqueueFor(t, fx, fixtureAcct, "second", "c.pdf", time.Now())

	// Deployment-wide counts APPS with work, not items: two apps, three items.
	if d := fx.srv.ScanQueueDepth(); d != 2 {
		t.Errorf("deployment depth = %d, want 2 apps with work", d)
	}
	// Per app counts ITEMS.
	if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 2 {
		t.Errorf("per-app depth = %d, want 2 items", n)
	}
	if n := queueLen(t, fx, fixtureAcct, "second"); n != 1 {
		t.Errorf("per-app depth = %d, want 1 item", n)
	}
	// An app that has never enqueued has no prefix, and that is depth zero
	// rather than an error.
	if n := queueLen(t, fx, fixtureAcct, "never-used"); n != 0 {
		t.Errorf("depth %d for an app with no queue prefix, want 0", n)
	}
	_ = ctx
}

// TestDrainDoesNothingWhileTheDeploymentIsDisabled, and specifically KEEPS the
// items: the deployment switch is the one reversible refusal, and an operator
// running `enable` should find the queue still there. The TTL is what clears a
// deployment left off for a day, not the tick.
func TestDrainDoesNothingWhileTheDeploymentIsDisabled(t *testing.T) {
	ctx := context.Background()
	fx, sc, dir := drainFixture(t)
	counting := &countingStore{DirStore: release.NewDirStore(dir)}
	fx.rel = counting
	fx.srv.cfg.Releases = counting
	enableStore(t, fx, fixtureAcct, fixtureApp, true)
	sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/x.pdf", "etag-1", time.Now(), 10)
	enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/x.pdf", time.Now())
	setDeploymentScanning(t, fx, false)

	// The tick's own switch check is not what keeps the bytes safe — DrainApp
	// re-reads consent and refuses on its own, so removing the tick's check
	// changes no verdict and no item. What it changes is WORK: with it, a
	// disabled deployment costs one record read; without it, every app with a
	// queue entry is listed every 30s forever. Listings are therefore what
	// this asserts, because they are the only observable difference.
	counting.lists.Store(0)
	if err := fx.srv.scanDrainTick(ctx); err != nil {
		t.Fatal(err)
	}
	if n := counting.lists.Load(); n != 0 {
		t.Errorf("%d listing(s) on a disabled deployment — the tick is doing per-app work it can never use", n)
	}
	if n := sc.Scans("uploads/x.pdf"); n != 0 {
		t.Errorf("scanned %d times with the deployment switch off", n)
	}
	if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 1 {
		t.Fatalf("%d item(s) left, want the item KEPT — this refusal is reversible", n)
	}

	// And it drains once the operator turns it back on, without a restart.
	setDeploymentScanning(t, fx, true)
	if err := fx.srv.scanDrainTick(ctx); err != nil {
		t.Fatal(err)
	}
	if n := sc.Scans("uploads/x.pdf"); n != 1 {
		t.Errorf("scanned %d times after re-enabling, want 1", n)
	}
}

// TestDrainDropsTheQueueWhenConsentIsGone is §7.5's other side: the permanent
// refusals delete, because none of them can change back in a way that makes
// the item useful, and an item that outlives consent is a request to read
// bytes nobody may read.
func TestDrainDropsTheQueueWhenConsentIsGone(t *testing.T) {
	ctx := context.Background()
	fx, sc, _ := drainFixture(t)
	enableStore(t, fx, fixtureAcct, fixtureApp, true)
	enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/x.pdf", time.Now())
	// The operator revokes the per-store grant between enqueue and drain.
	revokeScanning(t, fx, fixtureAcct, fixtureApp)

	if err := fx.srv.scanDrainTick(ctx); err != nil {
		t.Fatal(err)
	}
	if n := sc.Scans("uploads/x.pdf"); n != 0 {
		t.Errorf("read bytes %d time(s) from a store that may not be scanned", n)
	}
	if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 0 {
		t.Errorf("%d item(s) survived a permanent refusal", n)
	}
}

// TestScanNotifyWakesAWaiter is the long-poll's half of the drain (§5.2). The
// door lands in task 13; the wake it selects on is delivered here, so it has
// something to be written against.
func TestScanNotifyWakesAWaiter(t *testing.T) {
	ctx := context.Background()
	fx, sc, _ := drainFixture(t)
	enableStore(t, fx, fixtureAcct, fixtureApp, true)
	sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/x.pdf", "etag-1", time.Now(), 10)
	enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/x.pdf", time.Now())

	ref := objstore.AppRef{Account: fixtureAcct, App: fixtureApp}
	woken, cancel, ok := fx.srv.scanWait(ref, "uploads/x.pdf")
	if !ok {
		t.Fatal("an idle console refused to park the first waiter")
	}
	defer cancel()
	// A waiter on a DIFFERENT key must not be woken by this drain.
	other, cancelOther, ok := fx.srv.scanWait(ref, "uploads/other.pdf")
	if !ok {
		t.Fatal("an idle console refused to park a second waiter")
	}
	defer cancelOther()

	if err := fx.srv.scanDrainTick(ctx); err != nil {
		t.Fatal(err)
	}
	select {
	case <-woken:
	default:
		t.Error("the drain scanned the object and nothing woke the waiter")
	}
	select {
	case <-other:
		t.Error("a waiter on another key was woken")
	default:
	}

	// Cancelling removes the registration rather than leaking one entry per
	// key ever waited on.
	cancel()
	fx.srv.scanWaits.mu.Lock()
	n := len(fx.srv.scanWaits.m[scanWaitKey{ref: ref, key: "uploads/x.pdf"}])
	fx.srv.scanWaits.mu.Unlock()
	if n != 0 {
		t.Errorf("%d waiter(s) still registered after cancel", n)
	}
}

// TestDrainNotifiesEvenWhenALaterItemFails: DrainApp returns what it managed
// before it stopped, and a waiter whose object WAS scanned must be woken even
// though the app's next item failed. Notifying after the error check would
// make a long-poll wait out its whole window for an answer that already
// exists on the object.
//
// Reaching that state takes some care with the fake. The FIRST object is
// pre-tagged at its current version, so Status answers from the tag and Scan
// is never called for it — which leaves the one-shot Scan failure below for
// the SECOND object. done therefore carries the first while err carries the
// second, which is exactly the shape being asserted.
func TestDrainNotifiesEvenWhenALaterItemFails(t *testing.T) {
	ctx := context.Background()
	fx, sc, _ := drainFixture(t)
	enableStore(t, fx, fixtureAcct, fixtureApp, true)
	bucket := drainBucket(fixtureAcct, fixtureApp)

	mod := time.Now().Truncate(time.Second)
	first := objstore.ScanObject{Key: "uploads/first.pdf", ETag: "etag-1", Modified: mod, Size: 10}
	sc.PutObject(bucket, first.Key, first.ETag, mod, first.Size)
	sc.Put(bucket, first.Key, first.Version(), objstore.VerdictClean)
	sc.PutObject(bucket, "uploads/second.pdf", "etag-2", mod, 10)

	// Ids sort by time, so the first item is drained first.
	enqueueFor(t, fx, fixtureAcct, fixtureApp, first.Key, time.Now())
	enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/second.pdf", time.Now().Add(2*time.Millisecond))

	ref := objstore.AppRef{Account: fixtureAcct, App: fixtureApp}
	woken, cancel, ok := fx.srv.scanWait(ref, first.Key)
	if !ok {
		t.Fatal("an idle console refused to park a waiter")
	}
	defer cancel()

	sc.Fail("Scan", errors.New("engine had a bad moment"))

	if err := fx.srv.scanDrainTick(ctx); err != nil {
		t.Fatal(err) // the tick logs per-app failures rather than returning them
	}
	if sc.Scans(first.Key) != 0 {
		t.Fatal("the pre-tagged object was rescanned, so the Scan failure did not reach the second item")
	}
	select {
	case <-woken:
	default:
		t.Error("the first object was answered and its waiter was not woken, because a later item failed")
	}
	// The failed item is KEPT — an engine failure is about the engine, not
	// about whether this request was legitimate.
	if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 1 {
		t.Errorf("%d item(s) left, want the failed one kept for the next tick", n)
	}
}

// TestStartScanDrainRebuildsBeforeItsFirstTick covers the goroutine itself,
// which the tick-level tests above do not: they call rebuildScanQueue and
// scanDrainTick by hand, so a StartScanDrain that never ran the rebuild would
// pass every one of them and strand every item across every restart.
//
// It is also the only test of "the first tick is immediate". A loop that
// waited a full interval before its first pass would make every restart cost
// 30s of latency on work that was already queued.
func TestStartScanDrainRebuildsBeforeItsFirstTick(t *testing.T) {
	fx, sc, _ := drainFixture(t)
	enableStore(t, fx, fixtureAcct, fixtureApp, true)
	sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/x.pdf", "etag-1", time.Now(), 10)
	enqueueStranded(t, fx, fixtureAcct, fixtureApp, "uploads/x.pdf", time.Now())

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	stopped := make(chan struct{})
	go func() {
		fx.srv.StartScanDrain(ctx)
		close(stopped)
	}()

	deadline := time.Now().Add(5 * time.Second)
	for sc.Scans("uploads/x.pdf") == 0 {
		if time.Now().After(deadline) {
			cancel()
			<-stopped
			t.Fatal("the goroutine did not rebuild and drain a stranded item within 5s — well inside its first tick")
		}
		time.Sleep(2 * time.Millisecond)
	}
	cancel()
	select {
	case <-stopped:
	case <-time.After(5 * time.Second):
		t.Fatal("StartScanDrain did not return when its context was cancelled")
	}
}

// TestStartScanDrainIsInertWithoutAnEngine: scanning is opt-in, and the
// ordinary deployment has no clamd. The goroutine must return rather than tick
// forever over a deployment that can never drain anything.
func TestStartScanDrainIsInertWithoutAnEngine(t *testing.T) {
	fx, _, _ := drainFixture(t)
	fx.srv.cfg.StoreScanner = nil

	stopped := make(chan struct{})
	go func() {
		fx.srv.StartScanDrain(context.Background())
		close(stopped)
	}()
	select {
	case <-stopped:
	case <-time.After(5 * time.Second):
		t.Fatal("StartScanDrain kept ticking on a deployment with no engine")
	}
}

func TestScanUsageCountsOnDemandOnly(t *testing.T) {
	ctx := context.Background()
	fx, sc, _ := drainFixture(t)
	enableStore(t, fx, fixtureAcct, fixtureApp, true)
	bucket := drainBucket(fixtureAcct, fixtureApp)
	mod := time.Now().Truncate(time.Second)

	tagged := objstore.ScanObject{Key: "uploads/tagged.pdf", ETag: "etag-tagged", Modified: mod, Size: 100}
	sc.PutObject(bucket, tagged.Key, tagged.ETag, tagged.Modified, tagged.Size)
	sc.Put(bucket, tagged.Key, tagged.Version(), objstore.VerdictClean)
	sc.PutObject(bucket, "uploads/new.pdf", "etag-new", mod, 25)
	enqueueFor(t, fx, fixtureAcct, fixtureApp, tagged.Key, mod)
	enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/new.pdf", mod.Add(time.Millisecond))

	if err := fx.srv.scanDrainTick(ctx); err != nil {
		t.Fatal(err)
	}
	raw, _, err := fx.rel.Get(ctx, usage.ScanDayKey(usage.DayOf(time.Now())))
	if err != nil {
		t.Fatal(err)
	}
	var day usage.ScanDay
	if err := json.Unmarshal(raw, &day); err != nil {
		t.Fatal(err)
	}
	if got := day.Accounts[fixtureAcct]; got != (usage.ScanRow{Objects: 1, Bytes: 25}) {
		t.Fatalf("usage = %+v, want only the newly scanned object", got)
	}
}

func TestScanUsageFoldsPerTick(t *testing.T) {
	ctx := context.Background()
	fx, sc, dir := drainFixture(t)
	counting := &scanUsageCountingStore{DirStore: release.NewDirStore(dir)}
	fx.rel = counting
	fx.srv.cfg.Releases = counting
	enableStore(t, fx, fixtureAcct, fixtureApp, true)
	enableStore(t, fx, fixtureAcct, "second", true)
	for app, key := range map[string]string{fixtureApp: "uploads/a.pdf", "second": "uploads/b.pdf"} {
		sc.PutObject(drainBucket(fixtureAcct, app), key, "etag-1", time.Now(), 10)
		enqueueFor(t, fx, fixtureAcct, app, key, time.Now())
	}

	if err := fx.srv.scanDrainTick(ctx); err != nil {
		t.Fatal(err)
	}
	if got := counting.puts.Load(); got != 1 {
		t.Fatalf("usage writes = %d, want one for the whole tick", got)
	}
	raw, _, err := counting.Get(ctx, usage.ScanDayKey(usage.DayOf(time.Now())))
	if err != nil {
		t.Fatal(err)
	}
	var day usage.ScanDay
	if err := json.Unmarshal(raw, &day); err != nil {
		t.Fatal(err)
	}
	if got := day.Accounts[fixtureAcct]; got != (usage.ScanRow{Objects: 2, Bytes: 20}) {
		t.Fatalf("folded usage = %+v", got)
	}
}