package objstore

import (
	"context"
	"errors"
	"os"
	"strings"
	"testing"
	"time"

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

// queueFixture is a provisioned, scanning-enabled store with an armed engine
// and n objects the scanner knows about.
func queueFixture(t *testing.T, n int) (release.Store, *MemScanner, *QueueIndex, string) {
	t.Helper()
	s := release.NewDirStore(t.TempDir())
	_, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-west-1")
	rec := rotateFixture(t, s, pub, p)
	enableScanning(t, s, "bab", "titogo")

	sc := NewMemScanner()
	bucket := rec.Provisioned.Bucket
	for i := 0; i < n; i++ {
		sc.PutObject(bucket, objectName(i), "etag-"+objectName(i), time.Now().UTC(), 10)
	}
	return s, sc, NewQueueIndex(), bucket
}

func enqueueN(t *testing.T, s release.Store, idx *QueueIndex, n int) {
	t.Helper()
	for i := 0; i < n; i++ {
		if _, err := Enqueue(context.Background(), s, idx, "bab", "titogo", &QueueItem{
			Key: objectName(i), By: "app",
		}, time.Now()); err != nil {
			t.Fatal(err)
		}
	}
}

// Ids sort by time, so a prefix listing IS the queue in order.
func TestQueueIDSortsByTime(t *testing.T) {
	base := time.Date(2026, 9, 2, 8, 0, 0, 0, time.UTC)
	var prev string
	for i := 0; i < 200; i++ {
		id := NewQueueID(base.Add(time.Duration(i) * time.Millisecond))
		if id <= prev {
			t.Fatalf("id %q does not sort after %q", id, prev)
		}
		prev = id
	}
	// Two in the same millisecond must not collide.
	seen := map[string]bool{}
	for i := 0; i < 1000; i++ {
		id := NewQueueID(base)
		if seen[id] {
			t.Fatal("two ids in the same millisecond collided")
		}
		seen[id] = true
	}
}

// The object first, the index second. A crash between the two strands one item
// until the boot rebuild; the other order would mark an app whose item was
// never written, and a 202 sent in between would stop the client retrying.
// It is the ORDER that is under test, so the assertions have to be able to see
// it. Asserting only on the end state passes just as happily with the two
// swapped, which is the mutation that matters.
func TestEnqueueWritesObjectBeforeBump(t *testing.T) {
	ctx := context.Background()
	s, _, idx, _ := queueFixture(t, 1)

	// (a) At the instant the object is being written, the app is not yet
	// marked. Bump-first fails here.
	var markedDuringWrite int
	watching := &queueStore{Store: s, onPut: func(string) { markedDuringWrite = idx.Len() }}
	id, err := Enqueue(ctx, watching, idx, "bab", "titogo", &QueueItem{
		Key: "uploads/a", By: "app",
	}, time.Now())
	if err != nil {
		t.Fatal(err)
	}
	if markedDuringWrite != 0 {
		t.Fatal("the index was bumped before the item object was written")
	}
	if _, _, err := s.Get(ctx, QueueKey("bab", "titogo", id)); err != nil {
		t.Fatalf("item object was not written: %v", err)
	}
	if idx.Len() != 1 {
		t.Fatal("the app was not marked")
	}

	// (b) And a write that FAILS leaves the app unmarked, so nothing ever
	// points at an item that does not exist.
	fresh := NewQueueIndex()
	failing := &queueStore{Store: s, putErr: errors.New("the bucket said no")}
	if _, err := Enqueue(ctx, failing, fresh, "bab", "titogo", &QueueItem{
		Key: "uploads/b", By: "app",
	}, time.Now()); err == nil {
		t.Fatal("a failed write reported success")
	}
	if fresh.Len() != 0 {
		t.Fatal("the app was marked for an item that was never written")
	}
}

// THE test this design needed. The drain caps work per tick, so on a flooded
// app it deliberately leaves items behind with no enqueue to bump anything —
// and an earlier draft released the entry whenever the generation was
// unchanged, deleting the pointer to exactly those items. They would then sit
// until the next console boot, and the callback and long-poll wake would be
// lost for the app the cap exists for.
func TestIndexEntrySurvivesPartialDrain(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, _ := queueFixture(t, 5)
	enqueueN(t, s, idx, 5)

	done, listedEmpty, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 2, time.Now())
	if err != nil {
		t.Fatal(err)
	}
	if len(done) != 2 {
		t.Fatalf("drained %d, want the cap of 2", len(done))
	}
	if listedEmpty {
		t.Fatal("a capped listing reported itself empty")
	}
	if idx.Len() != 1 {
		t.Fatal("the entry pointing at the remaining items was released")
	}

	// Successive calls finish the job.
	for i := 0; i < 5 && idx.Len() > 0; i++ {
		if _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 2, time.Now()); err != nil {
			t.Fatal(err)
		}
	}
	if idx.Len() != 0 {
		t.Fatal("successive drains never emptied the queue")
	}
	left, err := s.List(ctx, QueuePrefix("bab", "titogo"))
	if err != nil || len(left) != 0 {
		t.Fatalf("items left behind: %v, %v", left, err)
	}
}

// An enqueue landing mid-drain must not have its entry released underneath it.
//
// Driven through DRAINAPP, not through Observe and Release by hand: what the
// generation counter protects is an ORDERING inside DrainApp — Observe before
// the listing — and a test that calls the three methods itself proves the
// methods work while leaving the ordering free to be wrong. Moving Observe
// below the listing passes such a test and fails this one.
func TestIndexEntrySurvivesEnqueueDuringDrain(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, _ := queueFixture(t, 2)
	enqueueN(t, s, idx, 1)

	// Drain the one item. The entry stays, because a consumed page may have
	// more behind it.
	if _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()); err != nil {
		t.Fatal(err)
	}
	if idx.Len() != 1 {
		t.Fatal("a consumed page released the entry before an empty listing confirmed it")
	}

	// Now the queue really is empty — except that an enqueue lands after the
	// listing has been taken and before the release decision.
	racing := &queueStore{Store: s}
	racing.afterList = func() {
		if _, err := Enqueue(ctx, s, idx, "bab", "titogo", &QueueItem{
			Key: objectName(1), By: "app",
		}, time.Now()); err != nil {
			t.Error(err)
		}
	}
	if _, listedEmpty, err := DrainApp(ctx, racing, sc, idx, "bab", "titogo", 10, time.Now()); err != nil || !listedEmpty {
		t.Fatalf("drain = %v, %v", listedEmpty, err)
	}
	if idx.Len() != 1 {
		t.Fatal("the entry for an item enqueued during the drain was released — nothing points at it until boot")
	}
	left, err := s.List(ctx, QueuePrefix("bab", "titogo"))
	if err != nil || len(left) != 1 {
		t.Fatalf("the late item = %v, %v", left, err)
	}
}

// Release needs BOTH conditions. Generation alone is the bug above.
func TestIndexReleasedOnlyOnEmptyListing(t *testing.T) {
	ref := AppRef{Account: "bab", App: "titogo"}
	idx := NewQueueIndex()
	idx.Bump(ref)
	observed := idx.Observe(ref)

	if idx.Release(ref, observed, false) {
		t.Fatal("released on a non-empty listing")
	}
	if !idx.Release(ref, observed, true) {
		t.Fatal("did not release on an empty listing with an unchanged generation")
	}
}

// The consent chain, re-checked at the drain because the record can change
// between enqueue and drain and THIS is where bytes are read.
func TestDrainAppRefusesAndDeletesWhenNotPermitted(t *testing.T) {
	ctx := context.Background()

	for name, break_ := range map[string]func(t *testing.T, s release.Store){
		"no attestation": func(t *testing.T, s release.Store) {
			rec, etag, _ := Get(ctx, s, "bab", "titogo")
			rec.Scanning.Requested = nil
			if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
				t.Fatal(err)
			}
		},
		"attested encrypted": func(t *testing.T, s release.Store) {
			rec, etag, _ := Get(ctx, s, "bab", "titogo")
			rec.Scanning.Requested.Encrypted = true
			if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
				t.Fatal(err)
			}
		},
		"engine not enabled for the store": func(t *testing.T, s release.Store) {
			rec, etag, _ := Get(ctx, s, "bab", "titogo")
			rec.Scanning.Enabled = nil
			if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
				t.Fatal(err)
			}
		},
		"app in the trash": func(t *testing.T, s release.Store) {
			if err := release.TrashApp(ctx, s, "bab", "titogo", "op@example.com", time.Now().UTC()); err != nil {
				t.Fatal(err)
			}
		},
	} {
		t.Run(name, func(t *testing.T) {
			s, sc, idx, _ := queueFixture(t, 2)
			enqueueN(t, s, idx, 2)
			break_(t, s)

			_, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
			if !errors.Is(err, ErrQueueNotPermitted) {
				t.Fatalf("err = %v, want ErrQueueNotPermitted", err)
			}
			// The items are DELETED, not kept: none of these can change back
			// in a way that makes them useful, and leaving them would block
			// the app's name for thirty days in the trash case.
			left, lerr := s.List(ctx, QueuePrefix("bab", "titogo"))
			if lerr != nil || len(left) != 0 {
				t.Fatalf("items survived a permanent refusal: %v, %v", left, lerr)
			}
		})
	}
}

// Disabled deployment-wide is REVERSIBLE, so the items are kept. The refusals
// above are not, so they are deleted. Getting these two the same way round
// would either strand a tenant's queue or lose it.
func TestDrainAppKeepsItemsWhenDeploymentDisabled(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, _ := queueFixture(t, 2)
	enqueueN(t, s, idx, 2)

	av, etag, err := release.GetAntivirus(ctx, s)
	if err != nil {
		t.Fatal(err)
	}
	av.Enabled = false
	if _, err := release.PutAntivirus(ctx, s, av, etag); err != nil {
		t.Fatal(err)
	}

	if _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()); !errors.Is(err, ErrScanningDisabled) {
		t.Fatalf("err = %v, want ErrScanningDisabled", err)
	}
	left, err := s.List(ctx, QueuePrefix("bab", "titogo"))
	if err != nil || len(left) != 2 {
		t.Fatalf("a reversible refusal dropped the queue: %v, %v", left, err)
	}
	if idx.Len() != 1 {
		t.Fatal("the index entry was released on a reversible refusal")
	}
}

// The drain re-issues failed and denied itself. Without this, an object the
// sweep last tagged failed would be answered from that tag forever and never
// rescanned on request — on-demand strictly less able than the hourly pass it
// exists to improve on.
func TestDrainAppReIssuesFailedAndDenied(t *testing.T) {
	ctx := context.Background()
	for _, stuck := range []Verdict{VerdictFailed, VerdictDenied} {
		t.Run(string(stuck), func(t *testing.T) {
			s, sc, idx, bucket := queueFixture(t, 1)
			key := objectName(0)
			// A stale verdict already on the object, at its current version.
			_, obj, err := sc.Status(ctx, bucket, key)
			if err != nil {
				t.Fatal(err)
			}
			sc.Put(bucket, key, obj.Version(), stuck)
			enqueueN(t, s, idx, 1)

			done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
			if err != nil {
				t.Fatal(err)
			}
			if len(done) != 1 {
				t.Fatalf("drained %d items", len(done))
			}
			if !done[0].Scanned {
				t.Fatalf("a %q verdict was answered from the tag rather than rescanned", stuck)
			}
			if done[0].Verdict != VerdictClean {
				t.Fatalf("verdict = %q, want the rescan's answer", done[0].Verdict)
			}
		})
	}
}

// A verdict already recorded at the current version costs a tag read and no
// fetch. That is the amplification guard: re-requesting a judged key must not
// re-stream its bytes.
func TestDrainAppReusesAJudgedVerdict(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, bucket := queueFixture(t, 1)
	key := objectName(0)
	_, obj, err := sc.Status(ctx, bucket, key)
	if err != nil {
		t.Fatal(err)
	}
	sc.Put(bucket, key, obj.Version(), VerdictClean)
	enqueueN(t, s, idx, 1)

	done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
	if err != nil {
		t.Fatal(err)
	}
	if len(done) != 1 || done[0].Scanned {
		t.Fatalf("a judged key was rescanned: %+v", done)
	}
	if done[0].Verdict != VerdictClean {
		t.Fatalf("verdict = %q", done[0].Verdict)
	}
}

// A verdict the object cannot carry goes to a result object instead, and the
// outcome says so — an infected object with Recorded false is NOT covered by
// the enforcement deny, because the deny matches a tag that is not there.
func TestDrainAppWritesResultWhenUnrecorded(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, bucket := queueFixture(t, 1)
	key := objectName(0)
	sc.Unrecordable(key)
	enqueueN(t, s, idx, 1)

	done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
	if err != nil {
		t.Fatal(err)
	}
	if len(done) != 1 || done[0].Recorded {
		t.Fatalf("outcome = %+v, want recorded false", done)
	}
	_, obj, err := sc.Status(ctx, bucket, key)
	if err != nil {
		t.Fatal(err)
	}
	res, err := ReadScanResult(ctx, s, "bab", "titogo", key, obj.Version())
	if err != nil || res == nil {
		t.Fatalf("no result object for an unrecorded verdict: %+v, %v", res, err)
	}
	if res.Verdict != string(VerdictClean) {
		t.Fatalf("result verdict = %q", res.Verdict)
	}
}

// And when a later scan DOES persist a verdict, the fallback goes — it must
// not outlive the real answer.
func TestDrainAppDeletesResultWhenLaterRecorded(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, bucket := queueFixture(t, 1)
	key := objectName(0)

	sc.Unrecordable(key)
	enqueueN(t, s, idx, 1)
	if _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()); err != nil {
		t.Fatal(err)
	}
	if _, _, err := s.Get(ctx, ScanResultKey("bab", "titogo", key)); err != nil {
		t.Fatalf("setup: no result object: %v", err)
	}

	// The tag limit clears, and the object is re-requested.
	sc2 := NewMemScanner()
	_, obj, _ := sc.Status(ctx, bucket, key)
	sc2.PutObject(bucket, key, obj.ETag, obj.Modified, obj.Size)
	if _, err := Enqueue(ctx, s, idx, "bab", "titogo", &QueueItem{
		Key: key, By: "app",
	}, time.Now()); err != nil {
		t.Fatal(err)
	}
	if _, _, err := DrainApp(ctx, s, sc2, idx, "bab", "titogo", 10, time.Now()); err != nil {
		t.Fatal(err)
	}
	if _, _, err := s.Get(ctx, ScanResultKey("bab", "titogo", key)); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the fallback outlived the real verdict: %v", err)
	}
}

// Expired items are dropped rather than scanned: after a sweep cycle the
// ordinary pass has had its chance, and a stale on-demand answer is worth less
// than none.
func TestDrainAppDropsExpiredItems(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, _ := queueFixture(t, 1)
	if _, err := Enqueue(ctx, s, idx, "bab", "titogo", &QueueItem{
		Key: objectName(0), By: "app",
	}, time.Now().Add(-48*time.Hour)); err != nil {
		t.Fatal(err)
	}

	done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
	if err != nil {
		t.Fatal(err)
	}
	if len(done) != 0 {
		t.Fatalf("an expired item was scanned: %+v", done)
	}
	left, err := s.List(ctx, QueuePrefix("bab", "titogo"))
	if err != nil || len(left) != 0 {
		t.Fatalf("expired item not collected: %v, %v", left, err)
	}
}

// The same key twice in one tick is scanned once. This is what stops one app's
// flood being amplified into repeated fetches of the same bytes.
//
// BUT DEDUPING THE WORK IS NOT DEDUPING THE PROMISES. Every enqueue was
// answered 202, and every one that named a callback path is owed a delivery.
// An earlier draft deleted the duplicate items along with their paths, so two
// requests for one key with different callbacks got one callback and the other
// app endpoint waited forever — a silent loss the drain's own test blessed by
// asserting only that one outcome came back.
func TestDrainAppDedupesWorkNotCallbacks(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, _ := queueFixture(t, 1)
	for _, path := range []string{"/hooks/a", "/hooks/b", "/hooks/a", ""} {
		if _, err := Enqueue(ctx, s, idx, "bab", "titogo", &QueueItem{
			Key: objectName(0), By: "app", CallbackPath: path,
		}, time.Now()); err != nil {
			t.Fatal(err)
		}
	}

	done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
	if err != nil {
		t.Fatal(err)
	}
	if len(done) != 1 {
		t.Fatalf("drained %d outcomes for one key — the object was read more than once", len(done))
	}
	if sc.Scans(objectName(0)) != 1 {
		t.Fatalf("the engine read the object %d times for four requests", sc.Scans(objectName(0)))
	}
	// Both distinct paths survive; the repeat does not become a second
	// delivery, and the request with no path adds nothing.
	got := done[0].CallbackPaths
	if len(got) != 2 || !containsString(got, "/hooks/a") || !containsString(got, "/hooks/b") {
		t.Fatalf("callback paths = %v, want both distinct paths and no duplicate", got)
	}
	left, err := s.List(ctx, QueuePrefix("bab", "titogo"))
	if err != nil || len(left) != 0 {
		t.Fatalf("duplicate items left behind: %v, %v", left, err)
	}
}

// An object deleted between enqueue and drain is dropped, not an error.
func TestDrainAppDropsGoneObjects(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, _ := queueFixture(t, 0) // the scanner knows about no objects
	if _, err := Enqueue(ctx, s, idx, "bab", "titogo", &QueueItem{
		Key: "uploads/vanished", By: "app",
	}, time.Now()); err != nil {
		t.Fatal(err)
	}

	done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
	if err != nil {
		t.Fatalf("a vanished object failed the drain: %v", err)
	}
	// IT IS AN OUTCOME, and this test used to assert the opposite. "There is
	// no such object" is the honest answer to the question the app asked, and
	// reporting nothing meant a long-poll parked on that key sat out its whole
	// window for an answer that had already been settled — the drain deleted
	// the item silently and woke nobody.
	if len(done) != 1 {
		t.Fatalf("outcomes = %+v, want the one that says the object is gone", done)
	}
	if !done[0].Gone {
		t.Errorf("outcome %+v is not marked Gone, so a caller cannot tell it from a verdict", done[0])
	}
	if done[0].Verdict != "" || done[0].Recorded || done[0].Scanned {
		t.Errorf("a gone object produced a verdict-shaped outcome: %+v", done[0])
	}
	left, _ := s.List(ctx, QueuePrefix("bab", "titogo"))
	if len(left) != 0 {
		t.Fatal("the item for a vanished object was kept")
	}
}

// An engine that goes away keeps the items AND the entry: it is a fact about
// the engine, not about these objects, and the next tick should retry.
func TestDrainAppKeepsItemsOnEngineFailure(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, _ := queueFixture(t, 1)
	enqueueN(t, s, idx, 1)
	sc.Fail("Scan", errors.New("dial unix /run/clamav/clamd.ctl: connection refused"))

	if _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()); err == nil {
		t.Fatal("an unreachable engine reported a successful drain")
	}
	left, err := s.List(ctx, QueuePrefix("bab", "titogo"))
	if err != nil || len(left) != 1 {
		t.Fatalf("an engine failure dropped the item: %v, %v", left, err)
	}
	if idx.Len() != 1 {
		t.Fatal("an engine failure released the index entry")
	}
}

// ---- the store wrapper the ordering and error tests need -------------------

// queueStore wraps a store so one operation can be failed, watched or raced.
//
// The optional listings are forwarded BY HAND. An embedded interface promotes
// only release.Store's own methods, so a wrapper that did not forward would
// quietly take the drain off PageLister and the collector off ModLister — and
// every test using it would then exercise the fallback path instead of the one
// it names.
type queueStore struct {
	release.Store
	putErr        error            // every Put fails with this
	onPut         func(key string) // runs BEFORE the write lands
	failGetPrefix string           // reads of keys under this prefix fail
	failGetKey    string           // reads of exactly this key fail
	afterList     func()           // runs after the listing, before the caller sees it
}

func (q *queueStore) Put(ctx context.Context, key string, body []byte, etag string) (string, error) {
	if q.onPut != nil {
		q.onPut(key)
	}
	if q.putErr != nil {
		return "", q.putErr
	}
	return q.Store.Put(ctx, key, body, etag)
}

func (q *queueStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if key == q.failGetKey || (q.failGetPrefix != "" && strings.HasPrefix(key, q.failGetPrefix)) {
		return nil, "", errors.New("objstore test: the bucket had a bad moment")
	}
	return q.Store.Get(ctx, key)
}

func (q *queueStore) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) {
	keys, err := q.Store.(release.PageLister).ListPage(ctx, prefix, startAfter, limit)
	if q.afterList != nil {
		f := q.afterList
		q.afterList = nil // once
		f()
	}
	return keys, err
}

func (q *queueStore) ListMod(ctx context.Context, prefix string) ([]release.KeyMod, error) {
	return q.Store.(release.ModLister).ListMod(ctx, prefix)
}

// Enqueue owns the timestamp, so a door cannot forget to stamp it. A zero At
// would mint an id sorting at year one and an item the very next drain
// discards as expired — a 202 that meant nothing.
func TestEnqueueStampsItsOwnTimestamp(t *testing.T) {
	ctx := context.Background()
	s, _, idx, _ := queueFixture(t, 1)
	now := time.Date(2026, 9, 2, 9, 0, 0, 0, time.UTC)

	item := &QueueItem{Key: "uploads/a", By: "app"} // no At
	id, err := Enqueue(ctx, s, idx, "bab", "titogo", item, now)
	if err != nil {
		t.Fatal(err)
	}
	if !item.At.Equal(now) {
		t.Fatalf("At = %v, want the clock the caller passed", item.At)
	}
	if want := NewQueueID(now); id[:12] != want[:12] {
		t.Fatalf("id %q does not carry the enqueue time (want the prefix of %q)", id, want)
	}
	stored, err := readQueueItem(ctx, s, QueueKey("bab", "titogo", id))
	if err != nil || stored.At.IsZero() {
		t.Fatalf("stored item = %+v, %v", stored, err)
	}
}

// A HeadObject the console may not make is not a recorded verdict.
//
// Saying recorded here would put "denied" in a callback beside a claim that
// the object carries it, when no tag was ever read or written — and the
// enforcement deny matches on a tag that is not there, so the app would be
// told a durable answer exists that no door can produce.
func TestDrainAppDeniedHeadIsNotRecorded(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, bucket := queueFixture(t, 1)
	sc.DenyHead(bucket, objectName(0))
	enqueueN(t, s, idx, 1)

	done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
	if err != nil {
		t.Fatal(err)
	}
	if len(done) != 1 {
		t.Fatalf("drained %d outcomes", len(done))
	}
	if done[0].Verdict != VerdictDenied {
		t.Fatalf("verdict = %q, want denied", done[0].Verdict)
	}
	if done[0].Recorded {
		t.Fatal("a denied HeadObject was reported as a recorded verdict")
	}
	if done[0].Scanned {
		t.Fatal("the engine was asked to read an object with no version")
	}
}

// §7.5: deletion is driven by ABSENCE, never by error. A bucket that fails one
// GET must not cost a tenant an accepted request.
func TestDrainAppKeepsItemOnTransientReadError(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, _ := queueFixture(t, 1)
	enqueueN(t, s, idx, 1)

	flaky := &queueStore{Store: s, failGetPrefix: QueuePrefix("bab", "titogo")}
	done, listedEmpty, err := DrainApp(ctx, flaky, sc, idx, "bab", "titogo", 10, time.Now())
	if err != nil {
		t.Fatalf("one unreadable item failed the whole drain: %v", err)
	}
	if len(done) != 0 {
		t.Fatalf("outcomes for an item that was never read: %+v", done)
	}
	if listedEmpty {
		t.Fatal("a page with an item in it reported itself empty")
	}
	left, err := s.List(ctx, QueuePrefix("bab", "titogo"))
	if err != nil || len(left) != 1 {
		t.Fatalf("a transient read error discarded the request: %v, %v", left, err)
	}
	if idx.Len() != 1 {
		t.Fatal("the entry pointing at the kept item was released")
	}

	// And it drains on the next tick, once the bucket is well again.
	done, _, err = DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
	if err != nil || len(done) != 1 {
		t.Fatalf("retry = %+v, %v", done, err)
	}
}

// The record itself failing to read is reversible too, and it is the SIBLING
// of the check above: QueuePermitted honours "absence, not error" for the
// record, and an earlier draft honoured it there and nowhere else.
func TestDrainAppKeepsItemsOnRecordReadError(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, _ := queueFixture(t, 1)
	enqueueN(t, s, idx, 1)

	broken := &queueStore{Store: s, failGetKey: Key("bab", "titogo")}
	if _, _, err := DrainApp(ctx, broken, sc, idx, "bab", "titogo", 10, time.Now()); err == nil {
		t.Fatal("an unreadable record reported a successful drain")
	}
	left, err := s.List(ctx, QueuePrefix("bab", "titogo"))
	if err != nil || len(left) != 1 {
		t.Fatalf("an unreadable record discarded the queue: %v, %v", left, err)
	}
	if idx.Len() != 1 {
		t.Fatal("an unreadable record released the index entry")
	}
}

// Engine drift keeps the items too: the record naming a different engine is a
// thing an operator reverses, not a fact about these objects.
func TestDrainAppKeepsItemsOnEngineDrift(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, _ := queueFixture(t, 1)
	enqueueN(t, s, idx, 1)
	rec, etag, _ := Get(ctx, s, "bab", "titogo")
	rec.Scanning.Enabled.Engine = "someone-elses-scanner"
	if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
		t.Fatal(err)
	}

	_, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
	if !errors.Is(err, ErrEngineDrift) {
		t.Fatalf("err = %v, want ErrEngineDrift", err)
	}
	left, lerr := s.List(ctx, QueuePrefix("bab", "titogo"))
	if lerr != nil || len(left) != 1 {
		t.Fatalf("drift discarded the queue: %v, %v", left, lerr)
	}
	if idx.Len() != 1 {
		t.Fatal("drift released the index entry")
	}
}

// An engine that will not say what it is keeps them as well — the same
// reasoning as an engine that will not scan.
func TestDrainAppKeepsItemsOnProvenanceFailure(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, _ := queueFixture(t, 1)
	enqueueN(t, s, idx, 1)
	sc.Fail("Provenance", errors.New("dial: connection refused"))

	_, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
	if !errors.Is(err, ErrEngineUnreachable) {
		t.Fatalf("err = %v, want ErrEngineUnreachable", err)
	}
	left, lerr := s.List(ctx, QueuePrefix("bab", "titogo"))
	if lerr != nil || len(left) != 1 {
		t.Fatalf("an unreachable engine discarded the queue: %v, %v", left, lerr)
	}
}

// The one read failure that IS permanent: bytes no build will ever parse.
// Leaving it would hold the app's index entry open forever.
func TestDrainAppDeletesUnreadableItem(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, _ := queueFixture(t, 1)
	enqueueN(t, s, idx, 1)
	if _, err := s.Put(ctx, QueueKey("bab", "titogo", "0000000000000000000000000v"),
		[]byte("{not json"), ""); err != nil {
		t.Fatal(err)
	}

	done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
	if err != nil {
		t.Fatal(err)
	}
	if len(done) != 1 {
		t.Fatalf("the readable item did not drain alongside the broken one: %+v", done)
	}
	left, err := s.List(ctx, QueuePrefix("bab", "titogo"))
	if err != nil || len(left) != 0 {
		t.Fatalf("the unparseable item was kept: %v, %v", left, err)
	}
}

// A result object is a verdict a door may answer from. When the app may no
// longer be scanned at all, that verdict must go with the queue items — and it
// is the worse orphan of the two: an item is drained next tick, a result is
// found by nothing, blocks the app-name claim gate, and resurrects the app.
func TestDrainAppDeletesResultsWhenNotPermitted(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, _ := queueFixture(t, 1)
	enqueueN(t, s, idx, 1)
	if err := WriteScanResult(ctx, s, "bab", "titogo", &ScanResult{
		Key: objectName(0), Verdict: "clean", Version: "v1", At: time.Now().UTC(),
	}); err != nil {
		t.Fatal(err)
	}
	rec, etag, _ := Get(ctx, s, "bab", "titogo")
	rec.Scanning.Enabled = nil
	if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
		t.Fatal(err)
	}

	if _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()); !errors.Is(err, ErrQueueNotPermitted) {
		t.Fatalf("drain = %v, want ErrQueueNotPermitted", err)
	}
	left, err := s.List(ctx, ScanResultsPrefix("bab", "titogo"))
	if err != nil || len(left) != 0 {
		t.Fatalf("a result outlived the consent that allowed it: %v, %v", left, err)
	}
}

// A store that was never granted has no bucket to scan. The check has to be
// the FIRST one, because everything after it dereferences Provisioned.
func TestDrainAppRefusesWhenNotGranted(t *testing.T) {
	ctx := context.Background()

	for name, break_ := range map[string]func(t *testing.T, s release.Store){
		"record absent": func(t *testing.T, s release.Store) {
			if err := s.Delete(ctx, Key("bab", "titogo")); err != nil {
				t.Fatal(err)
			}
		},
		"declared but never granted": func(t *testing.T, s release.Store) {
			rec, etag, _ := Get(ctx, s, "bab", "titogo")
			rec.Provisioned = nil
			if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
				t.Fatal(err)
			}
		},
	} {
		t.Run(name, func(t *testing.T) {
			s, sc, idx, _ := queueFixture(t, 1)
			enqueueN(t, s, idx, 1)
			break_(t, s)

			_, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
			if !errors.Is(err, ErrQueueNotPermitted) {
				t.Fatalf("err = %v, want ErrQueueNotPermitted", err)
			}
			left, lerr := s.List(ctx, QueuePrefix("bab", "titogo"))
			if lerr != nil || len(left) != 0 {
				t.Fatalf("items survived a permanent refusal: %v, %v", left, lerr)
			}
		})
	}
}

// A drain with no engine refuses in words rather than panicking on a nil
// interface, the way RunSweep does. The goroutine gates on this too, but a
// package that dereferences whatever it is handed is one call site away from
// taking the console down.
func TestDrainAppRefusesWithoutAnEngine(t *testing.T) {
	ctx := context.Background()
	s, _, idx, _ := queueFixture(t, 1)
	enqueueN(t, s, idx, 1)

	_, _, err := DrainApp(ctx, s, nil, idx, "bab", "titogo", 10, time.Now())
	if !errors.Is(err, ErrNoEngine) {
		t.Fatalf("err = %v, want ErrNoEngine", err)
	}
	left, lerr := s.List(ctx, QueuePrefix("bab", "titogo"))
	if lerr != nil || len(left) != 1 {
		t.Fatalf("a missing engine discarded the queue: %v, %v", left, lerr)
	}
}

// The purge race, at the drain. The re-read before writing a result is the
// only thing between an unrecordable verdict and an orphan under a purged
// prefix — and that orphan does not merely leak: the claim gate refuses the
// app's name for good, and materialize counts any key under the prefix and
// fails toward keeping the row, so the deleted app comes BACK on the next boot.
func TestResultObjectNotWrittenUnderPurgedPrefix(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, _ := queueFixture(t, 1)
	sc.Unrecordable(objectName(0)) // the ten-tag cap, in production
	enqueueN(t, s, idx, 1)

	// The teardown lands between the PER-OBJECT consent check (the second read
	// of store.json, added when a review pointed out that one check per page
	// is not one check per object) and drainOne's re-read before writing (the
	// third). The re-read is the guard being tested here; without it the
	// create-only write SUCCEEDS under a prefix nothing will ever list again.
	//
	// The count is 3 rather than 2 because there are now three reads of this
	// key in a drain, not two. If it ever drifts again, the tell is this test
	// failing with ErrQueueNotPermitted — that means the race moved to an
	// EARLIER guard, which is the case TestPurgeBeforeTheObjectIsReadIsRefused
	// below covers on purpose.
	staged := &recordRaceStore{Store: s, key: Key("bab", "titogo"), before: 3, race: func() {
		// What PurgeApp leaves behind: the prefix, emptied. Done directly
		// because PurgeApp itself refuses over a provisioned record, and the
		// teardown that clears it is not what this test is about.
		keys, err := s.List(ctx, release.AppPrefix("bab", "titogo"))
		if err != nil {
			t.Error(err)
			return
		}
		for _, k := range keys {
			if err := s.Delete(ctx, k); err != nil {
				t.Error(err)
			}
		}
	}}

	done, _, err := DrainApp(ctx, staged, sc, idx, "bab", "titogo", 10, time.Now())
	// REPORTED, not swallowed. The first version returned (out, nil) here, so
	// the drain read the group as a success: it deleted the items, put an
	// outcome in done, and carried on scanning the rest of the page knowing
	// consent was gone. The outcome it produced had Recorded false with no
	// result object behind it, which task 16 would have turned into a callback
	// saying "clean" about an object the status door answers "pending" about
	// forever.
	if !errors.Is(err, ErrQueueNotPermitted) {
		t.Fatalf("drain err = %v, want ErrQueueNotPermitted", err)
	}
	if len(done) != 0 {
		t.Errorf("done carries %d outcome(s) for a store that may not be scanned: %+v", len(done), done)
	}
	left, lerr := s.List(ctx, release.AppPrefix("bab", "titogo"))
	if lerr != nil {
		t.Fatal(lerr)
	}
	if len(left) != 0 {
		t.Fatalf("the purged app's prefix still holds %v — its name can never be claimed again", left)
	}
}

// TestPurgeBeforeTheObjectIsReadIsRefused is the guard the per-object consent
// check adds, and it is the better of the two outcomes: the teardown lands
// before this object's bytes are read at all, so nothing is fetched, nothing
// is scanned and no result can be stranded.
//
// The page-level check alone could not do this. It runs once and then the loop
// reads up to fifty objects, so a purge — or a revoked grant, or a member
// re-attesting the store as client-encrypted — was honoured only from the next
// tick, and every object still in the page was read under a consent that had
// already been withdrawn. Found by an external security review.
func TestPurgeBeforeTheObjectIsReadIsRefused(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, _ := queueFixture(t, 1)
	enqueueN(t, s, idx, 1)

	staged := &recordRaceStore{Store: s, key: Key("bab", "titogo"), before: 2, race: func() {
		keys, err := s.List(ctx, release.AppPrefix("bab", "titogo"))
		if err != nil {
			t.Error(err)
			return
		}
		for _, k := range keys {
			if err := s.Delete(ctx, k); err != nil {
				t.Error(err)
			}
		}
	}}

	_, _, err := DrainApp(ctx, staged, sc, idx, "bab", "titogo", 10, time.Now())
	if !errors.Is(err, ErrQueueNotPermitted) {
		t.Fatalf("err = %v, want ErrQueueNotPermitted — consent was gone before this object was read", err)
	}
	if n := sc.Scans(objectName(0)); n != 0 {
		t.Errorf("scanned %d time(s) after consent was withdrawn mid-page", n)
	}
	left, lerr := s.List(ctx, release.AppPrefix("bab", "titogo"))
	if lerr != nil {
		t.Fatal(lerr)
	}
	if len(left) != 0 {
		t.Fatalf("the purged app's prefix still holds %v", left)
	}
}

// recordRaceStore runs race() just before the Nth read of one key, so a test
// can put a competing writer at a chosen point in a walk. Same shape as
// walk_test.go's raceOnReadStore, which fires on the first read only.
type recordRaceStore struct {
	release.Store
	key    string
	before int // fire immediately before this read, 1-based
	race   func()
	seen   int
}

func (r *recordRaceStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if key == r.key {
		r.seen++
		if r.seen == r.before {
			r.race()
		}
	}
	return r.Store.Get(ctx, key)
}

func (r *recordRaceStore) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) {
	return r.Store.(release.PageLister).ListPage(ctx, prefix, startAfter, limit)
}

// THE CONSENT CHAIN IS ONE FUNCTION AND THIS IS THE TABLE THAT SAYS SO.
//
// Spec §5.1.1: the enqueue door and the drain must both refuse unless every
// link holds, and the reason they must not drift is that they have different
// consequences for the same answer — the door turns it into an HTTP state, the
// drain turns it into keep-or-delete (§7.5). A bool cannot carry that. It has
// already been a bool through ten tasks, and every caller reconstructed the
// reason from the record it was handed back, which is exactly the shape where
// two callers start disagreeing.
//
// So the refusal is named, and the two consequences are read off the name. The
// split this asserts is §7.5's, verbatim: everything upstream of the
// deployment switch is permanent, because none of it can change back in a way
// that makes THIS item useful; the switch itself is reversible, because an
// operator turns it on again and the item is still worth draining.
func TestQueueRefusalNamesEachLinkAndItsConsequence(t *testing.T) {
	ctx := context.Background()

	cases := []struct {
		name string
		// break mutates a fully permitted fixture to snap one link.
		snap      func(t *testing.T, s release.Store)
		want      QueueRefusal
		permanent bool
	}{
		{
			name: "nothing is broken",
			snap: func(*testing.T, release.Store) {},
			want: RefusalNone,
		},
		{
			name: "the store was never granted",
			snap: func(t *testing.T, s release.Store) {
				mutateRecord(t, s, func(r *Record) { r.Provisioned = nil })
			},
			want: RefusalNotGranted, permanent: true,
		},
		{
			name: "the app is in the trash",
			snap: func(t *testing.T, s release.Store) {
				if err := release.TrashApp(ctx, s, "bab", "titogo", "op@example.com", time.Now().UTC()); err != nil {
					t.Fatal(err)
				}
			},
			want: RefusalTrashed, permanent: true,
		},
		{
			name: "the member never requested scanning",
			snap: func(t *testing.T, s release.Store) {
				mutateRecord(t, s, func(r *Record) { r.Scanning.Requested = nil })
			},
			want: RefusalNotRequested, permanent: true,
		},
		{
			name: "the member attested the store client-encrypted",
			snap: func(t *testing.T, s release.Store) {
				mutateRecord(t, s, func(r *Record) { r.Scanning.Requested.Encrypted = true })
			},
			want: RefusalEncrypted, permanent: true,
		},
		{
			name: "the operator never granted scanning",
			snap: func(t *testing.T, s release.Store) {
				mutateRecord(t, s, func(r *Record) { r.Scanning.Enabled = nil })
			},
			want: RefusalNotEnabled, permanent: true,
		},
		{
			// The one reversible link, and the reason Permanent() exists at
			// all: an operator runs `carlos system antivirus enable` and every
			// item still in the queue is worth draining. Delete these and a
			// deployment that was off for ten minutes loses a tenant's
			// accepted requests.
			name: "the deployment switch is off",
			snap: func(t *testing.T, s release.Store) {
				disableDeployment(t, s)
			},
			want: RefusalDisabled, permanent: false,
		},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			s, _, _, _ := queueFixture(t, 1)
			tc.snap(t, s)

			rec, refusal, err := QueuePermitted(ctx, s, "bab", "titogo")
			if err != nil {
				t.Fatalf("QueuePermitted: %v", err)
			}
			if refusal != tc.want {
				t.Fatalf("refusal = %q, want %q", refusal, tc.want)
			}
			if refusal.Permanent() != tc.permanent {
				t.Fatalf("%q.Permanent() = %v, want %v — this decides whether the drain deletes the item or keeps it (§7.5)",
					refusal, refusal.Permanent(), tc.permanent)
			}
			if tc.want == RefusalNone && rec == nil {
				t.Fatal("a permitted app came back with no record")
			}
		})
	}
}

// RefusalNone must never be Permanent: "keep going" is not a deletion reason,
// and a Permanent() written as `r != RefusalDisabled` would say it was.
func TestPermittedIsNotAPermanentRefusal(t *testing.T) {
	if RefusalNone.Permanent() {
		t.Fatal("RefusalNone.Permanent() is true — the drain would delete every item it was allowed to scan")
	}
}

// A READ ERROR IS NOT A REFUSAL. It has no name in the chain because nothing
// about consent was learned: §7.5's "deletion is driven by absence, never by
// error". The error comes back as an error, and the refusal stays empty — so a
// caller that forgets to check err cannot mistake a bad bucket moment for
// permission, and equally cannot mistake it for a permanent no.
func TestQueuePermittedReportsAReadErrorAsAnError(t *testing.T) {
	s, _, _, _ := queueFixture(t, 1)
	broken := &queueStore{Store: s, failGetKey: "apps/bab/titogo/store.json"}

	rec, refusal, err := QueuePermitted(context.Background(), broken, "bab", "titogo")
	if err == nil {
		t.Fatal("a failing store.json read came back as a clean answer")
	}
	if refusal != RefusalNone {
		t.Fatalf("refusal = %q on a read error, want the empty refusal — a named refusal here would be deleted on (§7.5)", refusal)
	}
	if rec != nil {
		t.Fatal("a record came back from a read that failed")
	}
}

// mutateRecord reads, edits and writes back one store.json with the real
// etag, so a test can snap exactly one link of the consent chain and leave
// every other link intact.
func mutateRecord(t *testing.T, s release.Store, edit func(*Record)) {
	t.Helper()
	ctx := context.Background()
	rec, etag, err := Get(ctx, s, "bab", "titogo")
	if err != nil || rec == nil {
		t.Fatalf("re-read the record: %+v, %v", rec, err)
	}
	edit(rec)
	if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
		t.Fatal(err)
	}
}

// disableDeployment turns the deployment switch off the way `carlos system
// antivirus disable` does — the record, not an env var.
func disableDeployment(t *testing.T, s release.Store) {
	t.Helper()
	ctx := context.Background()
	av, etag, err := release.GetAntivirus(ctx, s)
	if err != nil || av == nil {
		t.Fatalf("antivirus record: %+v, %v", av, err)
	}
	av.Enabled = false
	if _, err := release.PutAntivirus(ctx, s, av, etag); err != nil {
		t.Fatal(err)
	}
}

// readErrorStore fails Get for one exact key, and behaves for everything else.
// A transient bad moment on one object, which is the case §7.5 separates from
// absence.
type readErrorStore struct {
	release.Store
	key string
	err error
}

func (r *readErrorStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if key == r.key {
		return nil, "", r.err
	}
	return r.Store.Get(ctx, key)
}

// TestRebuildKeepsAnItemItCouldNotRead pins the claim RebuildQueue's own
// comment makes and nothing tested: a STORE error is the store having a bad
// moment, so the item counts as live and the index still points at it.
//
// The opposite — treating an unreadable item as gone — silently strands it,
// because this rebuild is the only thing that would ever have listed that
// prefix again. §7.5's rule is that deletion is driven by absence, never by
// error, and this is that rule on the read side.
func TestRebuildKeepsAnItemItCouldNotRead(t *testing.T) {
	ctx := context.Background()
	s, _, idx, _ := queueFixture(t, 1)
	enqueueN(t, s, idx, 1)

	keys, err := s.List(ctx, QueuePrefix("bab", "titogo"))
	if err != nil || len(keys) != 1 {
		t.Fatalf("staging: %v %v", keys, err)
	}
	broken := &readErrorStore{Store: s, key: keys[0], err: errors.New("s3 had a moment")}

	fresh := NewQueueIndex()
	live, err := RebuildQueue(ctx, broken, fresh, "bab", "titogo", time.Now())
	if err != nil {
		t.Fatalf("a read error failed the whole rebuild: %v", err)
	}
	if live != 1 {
		t.Errorf("live = %d, want 1 — an item we could not read is not an item we know is gone", live)
	}
	if fresh.Len() != 1 {
		t.Error("the index does not point at the app, so nothing will ever list that prefix again")
	}
	left, lerr := s.List(ctx, QueuePrefix("bab", "titogo"))
	if lerr != nil || len(left) != 1 {
		t.Errorf("the item was deleted on a store error: %v %v", left, lerr)
	}
}

// TestRebuildStillIndexesWhenADeleteFails is the blocker an adversarial review
// found: the bump used to happen only on the success path, so one failed
// Delete of an EXPIRED item returned early and left the app's LIVE items with
// nothing pointing at them — until the next restart, since this pass is the
// only thing that collects a stranded item.
func TestRebuildStillIndexesWhenADeleteFails(t *testing.T) {
	ctx := context.Background()
	s, _, idx, _ := queueFixture(t, 2)

	// One expired item and one live one.
	if _, err := Enqueue(ctx, s, idx, "bab", "titogo",
		&QueueItem{Key: objectName(0), By: "app"}, time.Now().Add(-QueueTTL-time.Hour)); err != nil {
		t.Fatal(err)
	}
	if _, err := Enqueue(ctx, s, idx, "bab", "titogo",
		&QueueItem{Key: objectName(1), By: "app"}, time.Now()); err != nil {
		t.Fatal(err)
	}
	broken := &deleteErrorStore{Store: s, err: errors.New("s3 refused the delete")}

	fresh := NewQueueIndex()
	live, err := RebuildQueue(ctx, broken, fresh, "bab", "titogo", time.Now())
	if err == nil {
		t.Fatal("the failed delete was not reported")
	}
	_ = live
	if fresh.Len() != 1 {
		t.Fatalf("index holds %d app(s) after a failed delete, want 1 — the live item is stranded until the next restart", fresh.Len())
	}
}

// deleteErrorStore fails every Delete.
type deleteErrorStore struct {
	release.Store
	err error
}

func (d *deleteErrorStore) Delete(context.Context, string) error { return d.err }

// TestReusedVerdictCarriesNoProvenance is the one an adversarial review found
// missing, and the defect it pins was live: the outcome was stamped with the
// engine and signature database THIS TICK is configured with, before anything
// knew whether the engine had run.
//
// The scenario is ordinary. The hourly sweep tags an object clean under
// signature database 27100. A week later the app asks for a scan; the drain
// reads the tag, does not scan, and — before this — reported engine "clamav"
// and today's database 27412. That outcome becomes a SIGNED callback body,
// so the platform would have put its name to a claim about which database
// judged an object when no such judgement happened. The status door already
// refuses to make that guess, so the two surfaces disagreed and the signed
// one was the one lying.
//
// Empty is the honest answer. A verdict with no provenance is useful; a
// verdict with invented provenance is the failure the feature exists to stop.
func TestReusedVerdictCarriesNoProvenance(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, bucket := queueFixture(t, 1)

	// Pre-tag the object at its current version, as a sweep would have.
	obj := ScanObject{Key: objectName(0), ETag: "etag-" + objectName(0), Size: 10}
	status, seen, err := sc.Status(ctx, bucket, objectName(0))
	if err != nil {
		t.Fatal(err)
	}
	_ = status
	obj = seen
	sc.Put(bucket, objectName(0), obj.Version(), VerdictClean)
	enqueueN(t, s, idx, 1)

	done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
	if err != nil {
		t.Fatal(err)
	}
	if len(done) != 1 {
		t.Fatalf("outcomes = %+v, want one", done)
	}
	got := done[0]
	if got.Verdict != VerdictClean || !got.Recorded {
		t.Fatalf("outcome = %+v, want the tag's clean verdict", got)
	}
	if got.Scanned {
		t.Fatal("the engine ran for an object that already carried a verdict")
	}
	if got.Engine != "" || got.Signature != "" {
		t.Errorf("reused verdict claims engine %q signature %q — neither judged this object",
			got.Engine, got.Signature)
	}
}

// TestScannedVerdictCarriesProvenance is the other half, without which the
// test above passes against a drain that never reports provenance at all.
func TestScannedVerdictCarriesProvenance(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, _ := queueFixture(t, 1)
	enqueueN(t, s, idx, 1)

	done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
	if err != nil {
		t.Fatal(err)
	}
	if len(done) != 1 || !done[0].Scanned {
		t.Fatalf("outcomes = %+v, want one that was actually scanned", done)
	}
	if done[0].Engine == "" {
		t.Error("an outcome the engine really produced carries no engine name")
	}
}

// TestDrainMetersOnlyBytesTheEngineRead pins Scanned to the bytes reaching
// the engine, not to the Scan call having been made. A skipped object is one
// over the size ceiling that was never read, and a denied one is a GET the
// console was refused; both used to count their full size against the tenant.
// The meter is unpriced and best-effort, but it must not overstate.
func TestDrainMetersOnlyBytesTheEngineRead(t *testing.T) {
	ctx := context.Background()
	s, sc, idx, _ := queueFixture(t, 4)
	sc.ScanAs(objectName(1), VerdictSkipped)
	sc.ScanAs(objectName(2), VerdictDenied)
	sc.ScanAs(objectName(3), VerdictInfected)
	enqueueN(t, s, idx, 4)

	done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
	if err != nil {
		t.Fatal(err)
	}
	byKey := map[string]Drained{}
	for _, d := range done {
		byKey[d.Key] = d
	}
	if len(byKey) != 4 {
		t.Fatalf("outcomes = %+v, want four", done)
	}
	for _, key := range []string{objectName(1), objectName(2)} {
		d := byKey[key]
		if n := sc.Scans(key); n != 1 {
			t.Fatalf("%s: the engine was asked %d time(s), want once — this test is about an answer the call gave", key, n)
		}
		if d.Scanned {
			t.Errorf("%s answered %q and is metered as scanned: %d bytes nothing read would be charged", key, d.Verdict, d.Size)
		}
	}
	for _, key := range []string{objectName(0), objectName(3)} {
		d := byKey[key]
		if !d.Scanned || d.Size != 10 {
			t.Errorf("%s answered %q with Scanned=%v Size=%d, want a scanned object of 10 bytes", key, d.Verdict, d.Scanned, d.Size)
		}
	}
}

// TestRebuildWithNothingStaleDeletesNothing: the rebuild reads every item and
// enters its collect only when it found something to collect. Over a store
// whose every Delete fails, a prefix of live items rebuilds without error,
// which is only possible if no delete was attempted.
func TestRebuildWithNothingStaleDeletesNothing(t *testing.T) {
	ctx := context.Background()
	s, _, idx, _ := queueFixture(t, 2)
	enqueueN(t, s, idx, 2)
	broken := &deleteErrorStore{Store: s, err: errors.New("no delete may happen here")}

	fresh := NewQueueIndex()
	live, err := RebuildQueue(ctx, broken, fresh, "bab", "titogo", time.Now())
	if err != nil {
		t.Fatalf("a rebuild with nothing to collect reached a delete: %v", err)
	}
	if live != 2 || fresh.Len() != 1 {
		t.Fatalf("live = %d, indexed apps = %d, want 2 and 1", live, fresh.Len())
	}
}