// The scan queue: apps/<acct>/<app>/scan-queue/<id>.json, one object per item.
//
// ONE OBJECT PER ITEM, not one queue object. A shared queue object would put
// every enqueue in a compare-and-swap race with every other, and losing that
// race would be the normal case rather than the exception — the same problem
// the sidecar design avoided by having a single writer. Distinct keys never
// collide, so an enqueue needs no coordination at all.
//
// UNDER THE APP'S PREFIX, so PurgeApp removes the items with everything else
// and the name-claim gate can see them. A top-level scan-queue/<acct>/<app>/
// would be found by one listing instead of an index — but PurgeApp lists only
// the app prefix, and the claim gate never looks outside it, so items would
// survive app deletion invisibly. The index below is the price of keeping them
// where deletion can find them.
//
// The walk lives here and the goroutine lives in console, mirroring how
// RunSweep and scanPass already divide: objstore knows what to do to a bucket,
// console knows when.
package objstore

import (
	"context"
	"crypto/rand"
	"encoding/base32"
	"encoding/binary"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"os"
	"sort"
	"sync"
	"time"

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

// QueueSchema is pinned; an unknown schema refuses.
const QueueSchema = 1

// QueueTTL is how long an item may sit before it is dropped rather than
// scanned. One sweep cycle: after that the ordinary sweep has had its chance
// to reach the object, and a stale on-demand answer is worth less than none.
const QueueTTL = 24 * time.Hour

// QueueItem is one app's request that a named object be scanned soon.
type QueueItem struct {
	Schema int       `json:"schema"`
	Key    string    `json:"key"`
	At     time.Time `json:"at"`
	By     string    `json:"by"`
	// CallbackPath is the app's own path to POST the result to, validated at
	// the door. Never a host: the platform knows where the app lives, and a
	// caller-supplied host would be an SSRF against the console.
	CallbackPath string `json:"callback_path,omitempty"`
}

// AppRef names one app, for the index.
type AppRef struct{ Account, App string }

// queueIDEncoding is lowercase base32-HEX without padding.
//
// The hex alphabet, not the standard one, and the difference is the whole
// point: base32's usual alphabet puts the digits 2-7 at values 26-31, ABOVE
// the letters, while in ASCII digits sort BELOW them. Encoded ids would then
// not sort in the same order as the timestamps they encode, and a prefix
// listing would stop being the queue in order. base32-hex is monotonic —
// 0-9 then a-v, in both value and byte order.
var queueIDEncoding = base32.NewEncoding("0123456789abcdefghijklmnopqrstuv").WithPadding(base32.NoPadding)

// NewQueueID is a time-ordered id: milliseconds big-endian, then randomness.
//
// Hand-rolled because there is no ULID in this repo and adding a dependency is
// a decision, not a default. The properties that matter are that ids sort by
// time — so a prefix listing IS the queue in order — and that two enqueues in
// the same millisecond do not collide.
func NewQueueID(now time.Time) string {
	var b [16]byte
	binary.BigEndian.PutUint64(b[:8], uint64(now.UnixMilli()))
	if _, err := rand.Read(b[8:]); err != nil {
		// crypto/rand does not fail in practice; if it ever did, a
		// time-only id still sorts correctly and merely risks a collision
		// that the create-only write would catch.
		binary.BigEndian.PutUint64(b[8:], uint64(now.UnixNano()))
	}
	return queueIDEncoding.EncodeToString(b[:])
}

// QueuePrefix is where one app's items live.
func QueuePrefix(acct, app string) string {
	return release.AppPrefix(acct, app) + "scan-queue/"
}

// QueueKey is one item's object key.
func QueueKey(acct, app, id string) string { return QueuePrefix(acct, app) + id + ".json" }

// ---- the index --------------------------------------------------------------

// QueueIndex remembers which apps have work, so the drain lists only those
// prefixes instead of walking every app on the deployment every tick.
//
// SOFT STATE, deliberately. Losing it costs latency, never items: the boot
// rebuild finds anything stranded, and the ordinary sweep still reaches every
// object on its own cycle. What it must never do is claim an app is empty when
// it is not, which is what the generation counter below is for.
type QueueIndex struct {
	mu  sync.Mutex
	gen map[AppRef]uint64
}

func NewQueueIndex() *QueueIndex { return &QueueIndex{gen: map[AppRef]uint64{}} }

// Bump marks an app as having work. Called AFTER the item object is written:
// an index entry pointing at nothing is harmless, an item nothing points at is
// stranded until the next boot.
func (i *QueueIndex) Bump(ref AppRef) {
	i.mu.Lock()
	defer i.mu.Unlock()
	if i.gen == nil {
		i.gen = map[AppRef]uint64{}
	}
	i.gen[ref]++
}

// Observe reads the generation a drain is about to work against.
func (i *QueueIndex) Observe(ref AppRef) uint64 {
	i.mu.Lock()
	defer i.mu.Unlock()
	return i.gen[ref]
}

// Release drops an app's entry, and reports whether it did.
//
// TWO CONDITIONS, and the second is the one an earlier draft of this design
// got wrong. The generation must be unchanged — no enqueue landed while we
// worked — AND the listing must have come back empty. Generation alone is not
// enough: the drain caps how many items it handles per tick, so on a flooded
// app it deliberately leaves items behind with no enqueue to bump anything,
// and releasing there would delete the entry pointing at them.
func (i *QueueIndex) Release(ref AppRef, observed uint64, listedEmpty bool) bool {
	if !listedEmpty {
		return false
	}
	i.mu.Lock()
	defer i.mu.Unlock()
	if i.gen[ref] != observed {
		return false
	}
	delete(i.gen, ref)
	return true
}

// Apps lists the apps with work, for the drain's round-robin.
func (i *QueueIndex) Apps() []AppRef {
	i.mu.Lock()
	defer i.mu.Unlock()
	out := make([]AppRef, 0, len(i.gen))
	for ref := range i.gen {
		out = append(out, ref)
	}
	sort.Slice(out, func(a, b int) bool {
		if out[a].Account != out[b].Account {
			return out[a].Account < out[b].Account
		}
		return out[a].App < out[b].App
	})
	return out
}

// Len is how many apps have work — the deployment-wide queue depth surface.
func (i *QueueIndex) Len() int {
	i.mu.Lock()
	defer i.mu.Unlock()
	return len(i.gen)
}

// ---- enqueue ----------------------------------------------------------------

// Enqueue writes one item and then marks the app.
//
// ORDER MATTERS: the object first, the index second. A crash between the two
// strands that item until the next boot rebuild, which is recoverable. The
// other order would mark an app whose item was never written, and a caller
// answered 202 in between would stop retrying an item nothing points at.
//
// ENQUEUE OWNS THE TIMESTAMP, which is why now is a parameter and At is not a
// field the caller fills in. The id is derived from At and the TTL is measured
// from it, so a door that forgot to stamp it would mint an id sorting at year
// one and an item the very next drain discards as expired — a 202 that means
// nothing. One writer, one clock.
func claimRawEnqueue(ctx context.Context, s release.Store, idx *QueueIndex, acct, app string, item *QueueItem, now time.Time) (string, error) {
	item.Schema = QueueSchema
	item.At = now.UTC()
	body, err := json.MarshalIndent(item, "", "  ")
	if err != nil {
		return "", err
	}
	id := NewQueueID(item.At)
	if _, err := s.Put(ctx, QueueKey(acct, app, id), append(body, '\n'), ""); err != nil {
		return "", fmt.Errorf("objstore: enqueue %s/%s: %w", acct, app, err)
	}
	idx.Bump(AppRef{Account: acct, App: app})
	return id, nil
}

// ---- the boot rebuild -------------------------------------------------------

// RebuildQueue restores one app's index entry from what is actually in the
// bucket, and drops what expired while nobody was draining.
//
// THIS IS THE ONLY THING THAT COLLECTS A STRANDED ITEM, so it is not optional.
// Enqueue writes the object and then bumps the index; a crash between the two
// leaves an item in the bucket with nothing pointing at it, and the drain only
// ever looks at apps the index names. Nothing else would ever list that
// prefix again.
//
// UNCAPPED, unlike the drain's listing, and deliberately: the drain's cap is
// fairness between tenants within a tick, while this runs once and must see
// everything or the stranding it exists to fix survives it.
//
// live counts what a drain could still act on, and an item whose bytes could
// not be READ counts as live. An unreadable item is deleted (no tick will ever
// parse it), but a STORE error is the store having a bad moment: keeping the
// index entry costs one empty listing next tick, while dropping it strands the
// item until the next restart. Absence, never error — §7.5.
//
// Reads run outside any operation. The unreadable and expired items are
// collected afterwards in ONE app mutation, objstore.RebuildCollect, entered
// only when there is something to delete: the boot rebuild visits every app
// prefix, and an app with nothing to collect has nothing for it to write.
func RebuildQueue(ctx context.Context, s release.Store, idx *QueueIndex, acct, app string, now time.Time) (live int, err error) {
	keys, err := s.List(ctx, QueuePrefix(acct, app))
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			return 0, nil
		}
		return 0, err
	}
	// DEFERRED, so that an error partway through still points the index at what
	// was found. The first version bumped 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 — and since this rebuild is the only
	// thing that collects a stranded item, they would have sat until the next
	// restart. An index entry for an app with no work is harmless (one empty
	// listing, then it is released); a live item with no entry is not.
	defer func() {
		if live > 0 {
			idx.Bump(AppRef{Account: acct, App: app})
		}
	}()
	// A FAILED COLLECT MUST NOT ABORT THE INDEX PASS. The first version
	// returned on the first Delete error, and because ids sort by time the
	// EXPIRED items come first — so one refused delete returned before the
	// live items had even been looked at, leaving them with nothing pointing
	// at them until the next restart. Deleting is this pass's second job;
	// indexing is its first, and the first must finish.
	var stale []string
	for _, qk := range keys {
		item, rerr := readQueueItem(ctx, s, qk)
		switch {
		case errors.Is(rerr, os.ErrNotExist):
			continue
		case errors.Is(rerr, errQueueItemUnreadable):
			stale = append(stale, qk)
			continue
		case rerr != nil:
			log.Printf("objstore: scan queue rebuild: read %s: %v", qk, rerr)
			live++
			continue
		}
		if now.Sub(item.At) > QueueTTL {
			stale = append(stale, qk)
			continue
		}
		live++
	}
	if len(stale) == 0 {
		return live, nil
	}
	if err := release.WithAppMutation(ctx, s, acct, app, "objstore.RebuildCollect", func(ctx context.Context) error {
		return deleteQueueItems(ctx, s, stale)
	}); err != nil {
		// Still there, so still live: the next rebuild retries them.
		log.Printf("objstore: scan queue rebuild: collect %s/%s: %v", acct, app, err)
		return live + len(stale), err
	}
	return live, nil
}

// QueueDepth is how many items are waiting for one app — the per-app half of
// the depth §4.4 and §10 both want on a status page.
//
// A full listing, not the index: the index counts APPS with work, which is the
// deployment-wide number, and says nothing about how much any one of them has.
// Absence is depth zero, not an error: an app that has never enqueued has no
// prefix.
func QueueDepth(ctx context.Context, s release.Store, acct, app string) (int, error) {
	keys, err := s.List(ctx, QueuePrefix(acct, app))
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			return 0, nil
		}
		return 0, err
	}
	return len(keys), nil
}

// ---- draining ---------------------------------------------------------------

// Drained is one item's outcome, everything console needs and nothing it does
// not: the long-poll wake, the callback body, and the per-tick meter fold.
type Drained struct {
	Key, Version, ETag string
	Verdict            Verdict
	// Recorded is whether the verdict reached the object itself. When false
	// the caller has written a result object instead, and the app must be told
	// so — an infected object with Recorded false is NOT covered by the
	// enforcement deny, because the deny matches on a tag that is not there.
	Recorded bool
	// Gone is an object that was not there to scan. It is an OUTCOME, not an
	// error: the app asked about a key, and "it does not exist" is the honest
	// answer to that. It rides in the same list so a waiter parked on the key
	// is woken — before this existed the drain deleted the items silently and
	// a long-poll sat out its whole window for an answer that had already
	// been settled.
	Gone bool
	// Scanned is whether the engine actually ran, as against the verdict being
	// read back from an existing tag. The meter counts on-demand work; an
	// answer the sweep already paid for is not the tenant's to be charged.
	Scanned   bool
	Size      int64
	Engine    string
	Signature string
	// CallbackPaths is EVERY path that asked about this key in this page, not
	// one. The drain dedupes by object key so the engine reads an object once,
	// and an earlier draft deleted the duplicate items along with their
	// callbacks: two requests for the same key with different paths returned
	// two 202s and delivered one callback. Deduplicate the WORK, never the
	// promises made about it.
	CallbackPaths []string
}

// ErrQueueNotPermitted is a queue that must not be drained: the record is
// gone, the app is in the trash, or the consent chain no longer holds. The
// caller deletes the items rather than retrying them.
var ErrQueueNotPermitted = errors.New("objstore: this store may not be scanned")

// ErrNoEngine is a drain asked to run with no scanner behind it.
var ErrNoEngine = errors.New("objstore: no scanning engine")

// DrainApp scans up to cap of one app's queued items.
//
// It owns Observe and Release. Console cannot: Observe must happen BEFORE the
// listing, and that ordering is exactly what the generation counter protects —
// a caller that could get it wrong would be a caller that eventually does.
//
// UNDER HOST CLAIMS, EVERY WRITE AND EVERY BACKEND CALL IS FENCED BY A SHORT
// EXCLUSIVE APP MUTATION OF ITS OWN, and nothing spans more than one object.
// Each scan group runs under objstore.DrainGroup, covering exactly the Scan
// call, its result-object write or delete, and the deletion of that group's
// items. Housekeeping (expired or unreadable items, a dropped queue) gets an
// operation only when there is something to delete. Reads — the listing, the
// consent chain, the items — stay outside.
//
// The shape matters more than it looks. One operation around the whole page
// made a failure on object 2 report object 1's finished, durable deletes as
// part of an uncertain effect, and held the app until an operator reconciled
// it; a scan group that completes is now complete on its own. A failure
// before any effect (a refused begin, a store read) settles not-applied and
// the app reopens; a Scan that fails after the boundary marked it issued
// settles uncertain and holds the app, which is the boundary's own ruling
// about a backend call whose outcome is unknown, and is not softened here.
// Known outcomes (object gone, consent withdrawn or the deployment disabled
// after the engine answered) are results of a completed effect, never the
// operation's error.
func DrainApp(ctx context.Context, s release.Store, sc Scanner, idx *QueueIndex,
	acct, app string, cap int, now time.Time) (done []Drained, listedEmpty bool, err error) {

	// In words rather than as a nil dereference, the way RunSweep refuses
	// (sweep.go). The console gates its goroutine on a wired engine, but a
	// package that dereferences whatever it is handed is one call site away
	// from taking the control plane down.
	if sc == nil {
		return nil, false, ErrNoEngine
	}
	ref := AppRef{Account: acct, App: app}
	observed := idx.Observe(ref)

	keys, err := listQueue(ctx, s, acct, app, cap)
	if err != nil {
		return nil, false, err
	}
	listedEmpty = len(keys) == 0
	defer func() {
		if err == nil {
			idx.Release(ref, observed, listedEmpty)
		}
	}()
	if listedEmpty {
		return nil, true, nil
	}

	// The consent chain, re-checked here and not only at the door: the record
	// can change between enqueue and drain, and THIS is where bytes are read.
	rec, refusal, err := QueuePermitted(ctx, s, acct, app)
	if err != nil {
		return nil, false, err
	}
	// The reversible refusal, kept apart from the permanent ones BY NAME now
	// rather than by which return slot it arrived in. It used to come back as
	// an error, so the `err != nil` line above caught it and the deletion
	// below never saw it — correct, and correct by accident: any later
	// refusal that returned an error would have been silently spared, and any
	// that did not would have been silently deleted on.
	if refusal == RefusalDisabled {
		return nil, false, ErrScanningDisabled
	}
	if refusal.Permanent() {
		// Not a retry — none of these can change back in a way that makes the
		// items useful. Drop them and say so.
		if derr := dropQueue(ctx, s, acct, app, keys); derr != nil {
			return nil, false, derr
		}
		return nil, false, ErrQueueNotPermitted
	}

	engine, signature, err := sc.Provenance(ctx)
	if err != nil {
		return nil, false, fmt.Errorf("%w (%s/%s): %v", ErrEngineUnreachable, acct, app, err)
	}
	if rec.Scanning.Enabled.Engine != engine {
		return nil, false, fmt.Errorf("%w (%s/%s): record says %q, configured engine is %q",
			ErrEngineDrift, acct, app, rec.Scanning.Enabled.Engine, engine)
	}

	// The page-level bucket is only used to fail fast; the loop below re-reads
	// the record and uses THAT bucket for every object.
	_ = rec.Provisioned.Bucket
	groups, stale := groupQueue(ctx, s, keys, now)
	if len(stale) > 0 {
		// Collected before any scan, in one operation, so a failure here
		// abandons the page before a verdict exists to lose.
		if err := release.WithAppMutation(ctx, s, acct, app, "objstore.DrainCollect", func(ctx context.Context) error {
			return deleteQueueItems(ctx, s, stale)
		}); err != nil {
			return nil, false, err
		}
	}
	for i, g := range groups {
		// CONSENT IS RE-READ FOR EVERY OBJECT, not once for the page.
		//
		// An external security review found this: the check above happens
		// once and the loop then reads up to fifty objects' bytes. A member
		// who re-attests their store as client-encrypted, an operator who
		// revokes the grant or disables the deployment, or a TrashApp landing
		// mid-page, would all be honoured only from the NEXT tick — and every
		// object still in this page would be fetched and scanned under a
		// consent that had already been withdrawn.
		//
		// The cost is three small metadata reads per object against a full
		// object body read and an AV scan, which is the right trade for the
		// one property this whole feature rests on. A pre-call check still
		// leaves an irreducible window between the check and the GET — that
		// race cannot be closed without a lease the object store does not
		// offer — but it is now one object wide instead of one page wide.
		rec, refusal, cerr := QueuePermitted(ctx, s, acct, app)
		if cerr != nil {
			// A read error is not a refusal: keep the rest of the page and
			// let the next tick have it.
			return done, false, cerr
		}
		if refusal == RefusalDisabled {
			return done, false, ErrScanningDisabled
		}
		if refusal.Permanent() {
			// Everything not yet handled goes, on the same rule the
			// page-level check applies — including this group, whose object
			// has not been read.
			if delErr := dropRemainingQueue(ctx, s, acct, app, groups[i:]); delErr != nil {
				return done, false, delErr
			}
			return done, false, ErrQueueNotPermitted
		}
		// Engine drift too, for the same reason: the record can be moved to a
		// different engine between one object and the next, and a verdict
		// recorded under the wrong provenance is worse than no verdict.
		if rec.Scanning.Enabled.Engine != engine {
			return done, false, fmt.Errorf("%w (%s/%s): record says %q, configured engine is %q",
				ErrEngineDrift, acct, app, rec.Scanning.Enabled.Engine, engine)
		}
		// ONE OPERATION PER GROUP. The known outcomes drainOne reports as
		// errors — the object is gone, consent went away or the deployment
		// was disabled after the engine answered — are results of a completed
		// effect, so the group function returns nil for them and hands them
		// out through outcome; the caller acts on them outside the operation.
		// Anything else drainOne returns is a real failure, and the boundary
		// settles it on its own evidence: not-applied before any effect, held
		// after one.
		var out Drained
		var outcome error
		err := release.WithAppMutation(ctx, s, acct, app, "objstore.DrainGroup", func(ctx context.Context) error {
			// Acquiring the permit can wait behind a revoke or reprovision.
			fresh, refusal, err := QueuePermitted(ctx, s, acct, app)
			if err != nil {
				return err
			}
			if refusal == RefusalDisabled {
				outcome = ErrScanningDisabled
				return nil
			}
			if refusal != RefusalNone {
				outcome = ErrQueueNotPermitted
				return nil
			}
			if fresh.Scanning.Enabled.Engine != engine {
				return fmt.Errorf("%w (%s/%s): record says %q, configured engine is %q",
					ErrEngineDrift, acct, app, fresh.Scanning.Enabled.Engine, engine)
			}
			o, derr := drainOne(ctx, s, sc, acct, app, fresh.Provisioned.Bucket, engine, signature, g)
			switch {
			case errors.Is(derr, ErrObjectGone):
				outcome = derr
				return deleteQueueItems(ctx, s, g.queueKeys)
			case errors.Is(derr, ErrQueueNotPermitted), errors.Is(derr, ErrScanningDisabled):
				// The items are the caller's to keep or drop.
				outcome = derr
				return nil
			case derr != nil:
				// An engine failure is about the ENGINE, not this item. Keep
				// the item and the index entry, and let the next tick retry.
				return derr
			}
			out = o
			return deleteQueueItems(ctx, s, g.queueKeys)
		})
		if err != nil {
			return done, false, err
		}
		switch {
		case errors.Is(outcome, ErrObjectGone):
			done = append(done, Drained{Key: g.key, Gone: true, CallbackPaths: g.callbackPaths})
			continue
		case errors.Is(outcome, ErrQueueNotPermitted):
			// Consent went away between this group's check and its write.
			// Same conclusion as the check above, so the error keeps meaning
			// one thing to the caller: the queue was dropped, not deferred.
			if delErr := dropRemainingQueue(ctx, s, acct, app, groups[i:]); delErr != nil {
				return done, false, delErr
			}
			return done, false, outcome
		case errors.Is(outcome, ErrScanningDisabled):
			return done, false, outcome
		}
		done = append(done, out)
	}
	// A page that was fully consumed may still have more behind it.
	listedEmpty = false
	return done, listedEmpty, nil
}

// dropRemainingQueue deletes the items of every group not yet handled, and the
// app's fallback results with them.
//
// One function because two places reach the same conclusion — the per-object
// consent check, and a refusal observed inside drainOne after the scan — and
// the caller must not be able to tell them apart. ErrQueueNotPermitted means
// "the queue was dropped" wherever it comes from.
func dropRemainingQueue(ctx context.Context, s release.Store, acct, app string, left []*scanGroup) error {
	var keys []string
	for _, g := range left {
		keys = append(keys, g.queueKeys...)
	}
	return dropQueue(ctx, s, acct, app, keys)
}

// dropQueue deletes the named items and the app's fallback results in ONE
// operation, objstore.DropQueue. The results go with the items: a result
// object is a verdict no door may answer from once the app may not be
// scanned, and it is the worse orphan of the two — a stranded queue item is
// drained next tick, while a stranded result is found by nothing, blocks the
// app-name claim gate, and resurrects a deleted app on the next boot. The raw
// results delete is called so that one begin covers both.
func dropQueue(ctx context.Context, s release.Store, acct, app string, keys []string) error {
	return release.WithAppMutation(ctx, s, acct, app, "objstore.DropQueue", func(ctx context.Context) error {
		if err := deleteQueueItems(ctx, s, keys); err != nil {
			return err
		}
		return claimRawDeleteScanResults(ctx, s, acct, app)
	})
}

// scanGroup is every queued request in this page that asks about ONE object
// key: the object is read once, and every request that asked gets its answer.
type scanGroup struct {
	key string
	// queueKeys is every item object standing for this key, all deleted
	// together once the group has an outcome.
	queueKeys []string
	// callbackPaths is one entry per DISTINCT path that asked. Empty when
	// every request was long-poll only.
	callbackPaths []string
}

// groupQueue reads a listed page into scan groups, naming what can never be
// acted on (stale: the caller deletes it) and KEEPING what a later tick might
// still read.
//
// The distinction is §7.5's, and getting it wrong loses a tenant's request for
// good: deletion is driven by ABSENCE and by content that no tick will ever
// parse, never by a store error. An earlier draft deleted on any read failure,
// so one transient 500 on a GET discarded an accepted request silently.
//
// It reads and never writes, so it runs outside any operation.
func groupQueue(ctx context.Context, s release.Store, keys []string, now time.Time) (groups []*scanGroup, stale []string) {
	byKey := map[string]*scanGroup{}
	for _, qk := range keys {
		item, err := readQueueItem(ctx, s, qk)
		switch {
		case errors.Is(err, os.ErrNotExist):
			continue // already gone; nothing to delete and nothing to do
		case errors.Is(err, errQueueItemUnreadable):
			// Unparseable, or a schema this build does not know. No tick will
			// ever read it, and leaving it would hold the app's index entry
			// open forever.
			stale = append(stale, qk)
			continue
		case err != nil:
			// A STORE error. Reversible: keep the item, keep the entry, and
			// let the next tick have it. The rest of the page still drains —
			// one bad round trip is not a reason to stall an app's queue.
			log.Printf("objstore: scan queue: read %s: %v", qk, err)
			continue
		}
		if now.Sub(item.At) > QueueTTL {
			stale = append(stale, qk)
			continue
		}
		g := byKey[item.Key]
		if g == nil {
			g = &scanGroup{key: item.Key}
			byKey[item.Key] = g
			groups = append(groups, g)
		}
		g.queueKeys = append(g.queueKeys, qk)
		if item.CallbackPath != "" && !containsString(g.callbackPaths, item.CallbackPath) {
			g.callbackPaths = append(g.callbackPaths, item.CallbackPath)
		}
	}
	return groups, stale
}

func containsString(list []string, want string) bool {
	for _, s := range list {
		if s == want {
			return true
		}
	}
	return false
}

func deleteQueueItems(ctx context.Context, s release.Store, keys []string) error {
	for _, k := range keys {
		if err := s.Delete(ctx, k); err != nil && !errors.Is(err, os.ErrNotExist) {
			return err
		}
	}
	return nil
}

// drainOne scans a single object and records the outcome wherever it can go.
func drainOne(ctx context.Context, s release.Store, sc Scanner, acct, app, bucket, engine, signature string,
	g *scanGroup) (Drained, error) {

	verdict, obj, err := sc.Status(ctx, bucket, g.key)
	if err != nil {
		return Drained{}, err
	}
	// PROVENANCE IS NOT FILLED IN YET, and that is the whole of the fix below.
	// engine and signature describe the engine THIS TICK is configured with;
	// whether they describe the thing that judged this object is not known
	// until we know whether the engine ran.
	out := Drained{
		Key: g.key, Version: obj.Version(), ETag: obj.ETag, Size: obj.Size,
		CallbackPaths: g.callbackPaths,
	}

	// The drain applies the re-issue rule ITSELF: denied and failed are
	// transient, and answering an on-demand request from such a tag forever
	// would make this path strictly less able than the hourly sweep it exists
	// to improve on.
	needsScan := verdict == VerdictUnknown || verdict == VerdictFailed || verdict == VerdictDenied
	if !needsScan {
		// The object carries this verdict, at this version, right now — but
		// NOTHING HERE KNOWS WHAT PRODUCED IT. Engine and signature stay
		// empty, and the reason is worth stating because filling them in is
		// the obvious-looking thing to do.
		//
		// The tag may have been written by the hourly sweep last week under a
		// different signature database, and this deployment's engine may have
		// been replaced since. Stamping today's provenance onto it produces a
		// SIGNED callback body naming a signature database that never judged
		// this object — the status door already refuses to make exactly that
		// guess for a tag verdict (apiscanstatus.go), so the two surfaces
		// would disagree and the one lying would be the audit record.
		//
		// A verdict with no provenance is honest and useful. A verdict with
		// invented provenance is the failure this whole feature exists to
		// avoid. The tag does carry its own `at`, which readSignedTag parses
		// and Status discards; widening that seam is how these fields get
		// filled in truthfully, and it is a separate change.
		out.Verdict, out.Recorded = verdict, true
		// And any fallback for this key is now superseded — §5.2.1 says the
		// drain drops it as soon as a real verdict is persisted, and the
		// sweep can be what persisted it. Without this the result object
		// survives until the 24h age sweep even though the object it stands
		// in for has carried a real answer since. Delete is idempotent, so
		// the common case (no fallback) costs one cheap call.
		if err := DeleteScanResult(ctx, s, acct, app, g.key); err != nil {
			return Drained{}, err
		}
		return out, nil
	}
	if obj == (ScanObject{}) {
		// HeadObject itself was denied, so there is no version and no size to
		// scan with — and NOTHING WAS RECORDED. Saying recorded here would put
		// "denied" in a callback beside a claim that the object carries it,
		// when the console never even saw a tag; the enforcement deny matches
		// on a tag that is not there, and the app would be told a durable
		// answer exists that no door can produce.
		out.Verdict, out.Recorded = verdict, false
		return out, nil
	}

	v, recorded, err := sc.Scan(ctx, bucket, g.key, obj.Version(), obj.Size)
	if err != nil {
		return Drained{}, err
	}
	// The engine ran, here, now — so this is the one place the configured
	// engine and signature database really do describe what judged the bytes.
	out.Verdict, out.Recorded = v, recorded
	// Scanned is whether the bytes REACHED the engine, which the call alone
	// does not say: skipped is an object over the size ceiling that was never
	// read, and denied is a GET the console was refused. Counting either as
	// scanned would meter a tenant for a full object nothing looked at. Clean,
	// infected and failed are the answers an engine gives after reading.
	out.Scanned = v == VerdictClean || v == VerdictInfected || v == VerdictFailed
	out.Engine, out.Signature = engine, signature

	if recorded {
		// A real verdict is on the object now, so any fallback must go.
		if err := DeleteScanResult(ctx, s, acct, app, g.key); err != nil {
			return Drained{}, err
		}
		return out, nil
	}

	// The verdict could not be written to the object. Re-read the record
	// first: the writeSidecar race applies, and a result object stranded under
	// a purged prefix is found by nothing and resurrects the app.
	// ANY refusal stops the write, permanent or not: the question here is not
	// "will this item ever be useful" but "may this app be scanned right now",
	// and a result object written after the answer turned to no is a verdict
	// no door may answer from.
	//
	// AND IT IS RETURNED AS AN ERROR, which the first version got wrong. It
	// returned (out, nil), so DrainApp read the group as a SUCCESS: it deleted
	// the items, appended an outcome to done, and carried on scanning the rest
	// of the page with positive knowledge that consent was gone. Three things
	// followed, and the third is the worst — on a reversible `disabled` the
	// items were deleted against §7.5, the remaining objects were read anyway,
	// and `done` carried a verdict with Recorded false and NO result object
	// behind it, so task 16 would deliver a callback saying "clean" about an
	// object the status door will answer "pending" about forever. That is
	// exactly the disagreement §5.2.1 introduced the result object to prevent.
	if _, refusal, err := QueuePermitted(ctx, s, acct, app); err != nil {
		return Drained{}, err
	} else if refusal == RefusalDisabled {
		return Drained{}, ErrScanningDisabled
	} else if refusal != RefusalNone {
		return Drained{}, ErrQueueNotPermitted
	}
	if err := WriteScanResult(ctx, s, acct, app, &ScanResult{
		Key: g.key, Verdict: string(v), Version: obj.Version(),
		Engine: engine, Signature: signature, At: time.Now().UTC(),
	}); err != nil {
		return Drained{}, err
	}
	return out, nil
}

// QueuePermitted is the consent chain, in one place so the door and the drain
// cannot answer it differently.
//
// EXPORTED because the door lives in another package. That is the whole
// reason, and it is worth stating: the alternative — console re-deriving the
// same six conditions from a record it fetched itself — is the exact shape
// §5.1.1 exists to forbid. One implementation, two callers, and the only
// difference between them is what they do with the answer.
func QueuePermitted(ctx context.Context, s release.Store, acct, app string) (*Record, QueueRefusal, error) {
	rec, _, err := Get(ctx, s, acct, app)
	if err != nil {
		return nil, RefusalNone, err
	}
	if rec == nil || rec.Provisioned == nil {
		return nil, RefusalNotGranted, nil
	}
	tomb, err := release.ReadAppTombstone(ctx, s, acct, app)
	if err != nil {
		return nil, RefusalNone, err
	}
	if tomb != nil {
		return nil, RefusalTrashed, nil
	}
	switch {
	case rec.Scanning == nil || rec.Scanning.Requested == nil:
		return nil, RefusalNotRequested, nil
	case rec.Scanning.Requested.Encrypted:
		return nil, RefusalEncrypted, nil
	case rec.Scanning.Enabled == nil:
		return nil, RefusalNotEnabled, nil
	}
	on, err := release.AntivirusEnabled(ctx, s)
	if err != nil {
		return nil, RefusalNone, err
	}
	if !on {
		// The record comes back because this refusal is REVERSIBLE and the
		// caller may want to say something about the store it is still
		// refusing to scan. Everything above returns nil: there is nothing
		// there to talk about.
		return rec, RefusalDisabled, nil
	}
	return rec, RefusalNone, nil
}

// QueueRefusal is why the consent chain said no, and RefusalNone is the only
// value that means yes.
//
// It is a NAME rather than a bool because the enqueue door and the drain draw
// different conclusions from the same answer — the door turns it into an HTTP
// state (§5.1), the drain turns it into keep-or-delete (§7.5) — and a bool
// forces each of them to re-derive the reason from the record they were handed
// back. Two callers re-deriving the same fact from the same data is the exact
// shape in which they start disagreeing, and the whole point of §5.1.1 is that
// they must not.
//
// A read ERROR is not a refusal and never gets a name here. Nothing about
// consent was learned, and §7.5's "deletion is driven by absence, never by
// error" depends on those two staying separate: give a bad bucket moment a
// name and something will eventually delete on it.
type QueueRefusal string

const (
	RefusalNone         QueueRefusal = ""
	RefusalNotGranted   QueueRefusal = "not_granted"
	RefusalTrashed      QueueRefusal = "trashed"
	RefusalNotRequested QueueRefusal = "not_requested"
	RefusalEncrypted    QueueRefusal = "encrypted"
	RefusalNotEnabled   QueueRefusal = "not_enabled"
	RefusalDisabled     QueueRefusal = "disabled"
)

// Permanent reports whether this refusal will still be true in a way that
// makes the queued item worthless — §7.5's delete-vs-keep split, in the one
// place both callers read it from.
//
// Everything upstream of the deployment switch is permanent: a store that was
// never granted, an app in the trash, scanning never requested, an attestation
// saying the bytes are ciphertext, an operator who never enabled it. None of
// those can change back in a way that makes THIS item useful — if the state is
// later repaired the sweep covers the object anyway, and the requester is long
// gone.
//
// The deployment switch is not. An operator runs `carlos system antivirus
// enable` and every item still in the queue is worth draining. A deployment
// disabled for longer than the TTL drops its queue, which is right: both doors
// answered `disabled` at the time.
//
// Written as an explicit list rather than `r != RefusalDisabled`, so that a
// refusal added later is permanent only because somebody said so. The default
// for a new name is "keep the item", which is the survivable mistake.
func (r QueueRefusal) Permanent() bool {
	switch r {
	case RefusalNotGranted, RefusalTrashed, RefusalNotRequested,
		RefusalEncrypted, RefusalNotEnabled:
		return true
	}
	return false
}

func listQueue(ctx context.Context, s release.Store, acct, app string, cap int) ([]string, error) {
	prefix := QueuePrefix(acct, app)
	if pl, ok := s.(release.PageLister); ok {
		return pl.ListPage(ctx, prefix, "", cap)
	}
	// A store that cannot page still works, just less cheaply — the same
	// degradation storagePass accepts for SizeLister.
	keys, err := s.List(ctx, prefix)
	if err != nil {
		return nil, err
	}
	sort.Strings(keys)
	if cap > 0 && len(keys) > cap {
		keys = keys[:cap]
	}
	return keys, nil
}

// errQueueItemUnreadable marks an item whose BYTES are the problem — bad JSON,
// or a schema this build does not know. It is the one read failure that is
// permanent, and the drain deletes on it; every other read failure is the
// store having a bad moment and the item stays. Wrapped rather than compared
// so the caller can tell the two apart with errors.Is and nothing else.
var errQueueItemUnreadable = errors.New("objstore: unreadable queue item")

func readQueueItem(ctx context.Context, s release.Store, key string) (*QueueItem, error) {
	raw, _, err := s.Get(ctx, key)
	if err != nil {
		return nil, err
	}
	var item QueueItem
	if err := json.Unmarshal(raw, &item); err != nil {
		return nil, fmt.Errorf("%w %s: %v", errQueueItemUnreadable, key, err)
	}
	if item.Schema != QueueSchema {
		return nil, fmt.Errorf("%w %s: schema %d is not %d", errQueueItemUnreadable, key, item.Schema, QueueSchema)
	}
	return &item, nil
}
func Enqueue(ctx context.Context, s release.Store, idx *QueueIndex, acct, app string, item *QueueItem, now time.Time) (claimResult0 string, claimResult1 error) {
	claimResult1 = release.WithAppMutation(ctx, s, acct, app, "objstore.Enqueue", func(claimContext context.Context) error {
		claimResult0, claimResult1 = claimRawEnqueue(claimContext, s, idx, acct, app, item, now)
		return claimResult1
	})
	return
}