// The scan drain (spec §7.2): the goroutine that empties the on-demand queue.
//
// ITS OWN GOROUTINE AND ITS OWN CLOCK, which is the whole reason this file is
// not three functions inside scanpass.go. Draft 1 put a queue phase inside the
// existing pass; that cannot deliver seconds of latency, because scanPass
// sweeps apps sequentially and each gets a full budget (scanpass.go), so a
// queue phase behind it starves for up to N × budget. An app that uploads a
// file and waits for a verdict would be waiting on a walk of somebody else's
// bucket.
//
// The only state it shares with the sweep is object tags, which are idempotent
// per version: two writers agreeing on the same answer for the same bytes is
// not a race. It never touches the sidecar.
package console

import (
	"context"
	"errors"
	"log"
	"strings"
	"sync"
	"time"

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

// scanDrainInterval is the tick. 30s, against the sweep's hour: this path
// exists because waiting an hour for a verdict on an upload is useless.
const scanDrainInterval = 30 * time.Second

// scanDrainPerApp is §7.3's fairness cap — how many items ONE app gets in one
// tick, not how many the deployment gets.
//
// Each app gets a bounded turn; excess requests stay queued for later ticks.
const scanDrainPerApp = 50

// scanDrainAppBudget bounds one app's turn in one tick — the time half of the
// fairness rule, where scanDrainPerApp is the count half.
//
// Longer than a tick on purpose: a drain that overran its interval should
// finish the objects it started rather than be cut off mid-scan, and the
// ticker's next fire simply waits. What it must not do is run unbounded.
const scanDrainAppBudget = 90 * time.Second

// StartScanDrain rebuilds the index and then drains every 30s until ctx is
// done. serve() starts it; constructing a Server starts nothing.
//
// THE REBUILD RUNS FIRST AND BLOCKS THE FIRST TICK, because a tick before it
// would drain only the apps this process happens to have heard from — which,
// one second after a restart, is none of them.
func (s *Server) StartScanDrain(ctx context.Context) {
	if s.cfg.Releases == nil || s.cfg.StoreScanner == nil {
		return // no bucket, or no engine: nothing to drain into
	}
	s.scanDelivery = newScanDeliveryPool(ctx, s)
	defer func() {
		s.scanDelivery.stop()
		s.scanDelivery = nil
	}()
	rebuilt := true
	if err := s.rebuildScanQueue(ctx); err != nil {
		rebuilt = false
		// Not fatal. A console that cannot rebuild still drains what it is
		// told about from here on, and the next restart tries again; refusing
		// to start the goroutine would turn a listing hiccup into an outage
		// of the whole on-demand path.
		log.Printf("console: scan queue rebuild: %v", err)
	}
	rebuildDelay := scanDrainInterval
	nextRebuild := time.Now().Add(rebuildDelay)
	t := time.NewTicker(scanDrainInterval)
	defer t.Stop()
	for {
		// Retry failed discovery with backoff. An app held by another owner
		// must not cause a deployment-wide enumeration on every drain tick.
		if !rebuilt && !time.Now().Before(nextRebuild) {
			if err := s.rebuildScanQueue(ctx); err != nil {
				log.Printf("console: scan queue rebuild (retry): %v", err)
				rebuildDelay = min(rebuildDelay*2, 5*time.Minute)
				nextRebuild = time.Now().Add(rebuildDelay)
			} else {
				rebuilt = true
			}
		}
		if err := s.scanDrainTick(ctx); err != nil {
			log.Printf("console: scan drain: %v", err)
		}
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
	}
}

// rebuildScanQueue is the one full enumeration (§7.1): every app on the
// deployment, so that an item stranded by a crash between the object write and
// the index bump is found, and everything past its TTL is collected.
//
// Same enumeration as scanPass, and the same accepted cost — but ONCE at boot
// rather than hourly, so it is the cheap one of the two.
//
// One app's failure does not stop the others: this pass exists to recover what
// it can, and an account whose listing fails should not cost every account
// after it its queue.
func (s *Server) rebuildScanQueue(ctx context.Context) error {
	if s.cfg.Releases == nil {
		return nil
	}
	now := time.Now()
	acctPrefixes, err := s.cfg.Releases.ListDirs(ctx, "apps/")
	if err != nil {
		return err
	}
	var errs []error
	for _, acctPrefix := range acctPrefixes {
		acct := strings.TrimSuffix(strings.TrimPrefix(acctPrefix, "apps/"), "/")
		appPrefixes, err := s.cfg.Releases.ListDirs(ctx, acctPrefix)
		if err != nil {
			log.Printf("console: scan queue rebuild: %s: %v", acct, err)
			errs = append(errs, err)
			continue
		}
		for _, appPrefix := range appPrefixes {
			if err := ctx.Err(); err != nil {
				return err
			}
			app := strings.TrimSuffix(strings.TrimPrefix(appPrefix, acctPrefix), "/")
			if _, err := objstore.RebuildQueue(ctx, s.cfg.Releases, s.scanQueue, acct, app, now); err != nil {
				log.Printf("console: scan queue rebuild: %s/%s: %v", acct, app, err)
				errs = append(errs, err)
			}
			// The fallback results are collected in the SAME pass, because
			// this is the only place that visits an app which has stopped
			// enqueuing. The per-tick collector below only reaches apps that
			// still have work.
			s.collectScanResults(ctx, acct, app, now)
		}
	}
	// Joined, not swallowed: StartScanDrain retries until this returns nil, so
	// a partial enumeration must not look like a complete one.
	return errors.Join(errs...)
}

// scanDrainTick gives every app with work one capped turn.
//
// Round-robin is what iterating the index IS: every app named in it gets a
// turn each tick and none gets more than scanDrainPerApp, so an app cannot
// buy latency from another by queueing more.
//
// The deployment switch is read every tick, exactly as scanPass reads it, so
// `carlos system antivirus enable` takes effect without a restart. Disabled
// means this tick does nothing — not that there is no goroutine, which is what
// would make enabling need one.
func (s *Server) scanDrainTick(ctx context.Context) error {
	// An early-out for a direct caller, not a guarantee: StartScanDrain never
	// starts without an engine, and DrainApp refuses with ErrNoEngine before
	// it lists anything. Removing this line changes nothing observable — it
	// only stops a tick from logging one refusal per app. Kept because
	// scanPass guards the same way, and said here so nobody looks for the
	// test that pins it.
	if s.cfg.Releases == nil || s.cfg.StoreScanner == nil {
		return nil
	}
	enabled, err := s.antivirusEnabled(ctx)
	if err != nil {
		return err
	}
	if !enabled {
		// NOT THE SAFETY NET, and worth being exact about: DrainApp re-reads
		// consent for itself and refuses a disabled deployment on its own, so
		// deleting this check changes no verdict and loses no item. What it
		// changes is WORK — with it a disabled deployment costs one record
		// read per tick, without it every app with an index entry is listed
		// every 30s for as long as it stays off.
		//
		// Items are KEPT, not dropped: this refusal is reversible, and the
		// TTL is what eventually clears a deployment left off for a day.
		return nil
	}
	now := time.Now()
	deltas := map[string]usage.ScanRow{}
	for _, ref := range s.scanQueue.Apps() {
		if err := ctx.Err(); err != nil {
			return err
		}
		delta := s.drainScanApp(ctx, ref, now)
		if delta.Objects != 0 {
			row := deltas[ref.Account]
			row.Objects += delta.Objects
			row.Bytes += delta.Bytes
			deltas[ref.Account] = row
		}
	}
	return usage.UpsertScanDay(ctx, s.cfg.Releases, now, deltas)
}

// drainScanApp drains one app and tells whoever is waiting.
//
// Every outcome is logged and none is returned: one app's engine trouble must
// not end the tick for the apps after it in the round-robin, which is the
// posture scanPass already takes for the same reason.
func (s *Server) drainScanApp(ctx context.Context, ref objstore.AppRef, now time.Time) usage.ScanRow {
	var delta usage.ScanRow
	parent := ctx
	// ONE APP'S TURN IS BOUNDED. clamd's own read timeout is minutes, and the
	// object body stream carries no deadline of its own, so fifty slow objects
	// belonging to one tenant could hold the whole deployment's drain for
	// hours — starving every other tenant's single urgent item, which is the
	// precise failure §7.3's cap exists to prevent and which the cap alone
	// does not close. The sweep budgets per app for the same reason; this is
	// that budget. An app that runs out keeps its items and its index entry
	// and gets the next tick.
	ctx, cancel := context.WithTimeout(ctx, scanDrainAppBudget)
	defer cancel()

	startRec, _, _ := objstore.Get(ctx, s.cfg.Releases, ref.Account, ref.App)
	done, _, err := objstore.DrainApp(ctx, s.cfg.Releases, s.cfg.StoreScanner, s.scanQueue,
		ref.Account, ref.App, scanDrainPerApp, now)

	// NOTIFY BEFORE HANDLING THE ERROR. DrainApp returns the items it managed
	// before it stopped, and a waiter whose object was actually scanned should
	// be woken even though the app's next item failed. Dropping these on the
	// error path would make a long-poll wait out its full window for an answer
	// that already exists.
	for _, d := range done {
		if d.Scanned {
			delta.Objects++
			delta.Bytes += d.Size
		}
		s.scanNotify(ref, d.Key)
		for _, path := range d.CallbackPaths {
			if startRec != nil && startRec.Scanning != nil {
				s.scanDelivery.enqueue(scanDeliveryJob{ref: ref, done: d, path: path, gen: startRec.Scanning.WebhookGen})
			}
		}
	}

	switch {
	case err == nil:
	case errors.Is(err, objstore.ErrQueueNotPermitted):
		// Not an incident: consent was withdrawn, or the app was trashed, and
		// DrainApp has already deleted the items and the results. Logged at
		// all only because "the queue emptied without scanning" is a thing an
		// operator reading a support question wants to be able to see.
		log.Printf("console: scan drain: %s/%s: queue dropped — this store may no longer be scanned",
			ref.Account, ref.App)
	case errors.Is(err, objstore.ErrScanningDisabled):
		// A disable that landed between the tick's check and this app. The
		// items stay; the next enabled tick has them.
	default:
		log.Printf("console: scan drain: %s/%s: %v", ref.Account, ref.App, err)
	}
	cleanupCtx, cleanupCancel := context.WithTimeout(parent, storeTimeout)
	defer cleanupCancel()
	s.collectScanResults(cleanupCtx, ref.Account, ref.App, now)
	return delta
}

// collectScanResults ages out an app's fallback result objects.
//
// PER VISITED APP rather than in one deployment-wide sweep, so that a console
// which never restarts does not accumulate them — nothing else deletes a
// result for an object that stays permanently untaggable, because the drain
// only removes one when a later scan records a real verdict on the same key.
func (s *Server) collectScanResults(ctx context.Context, acct, app string, now time.Time) {
	n, err := objstore.CollectScanResults(ctx, s.cfg.Releases, acct, app, now.Add(-objstore.QueueTTL))
	switch {
	case errors.Is(err, objstore.ErrNoModLister):
		// A store that cannot report modification times, which is a test fake
		// rather than a deployment. Skipped, not failed — storagePass treats
		// its own optional interface the same way.
	case err != nil:
		log.Printf("console: scan result collect: %s/%s: %v", acct, app, err)
	case n > 0:
		log.Printf("console: scan result collect: %s/%s: dropped %d expired result(s)", acct, app, n)
	}
}

// ScanQueueDepth is how many apps have on-demand work waiting — the
// deployment-wide number §4.4 wants on `carlos system antivirus status`.
//
// APPS, not items, and named for what it counts. The per-app item count is
// objstore.QueueDepth, which costs a listing; this one is free and is the
// number that answers "is the drain keeping up".
func (s *Server) ScanQueueDepth() int {
	if s.scanQueue == nil {
		return 0
	}
	return s.scanQueue.Len()
}

// ---- waking the long-poll ---------------------------------------------------

// scanWaitKey is one object, for one app.
type scanWaitKey struct {
	ref objstore.AppRef
	key string
}

// scanWaiters is the in-process notification the status door's `wait` uses
// (§5.2), so a long-poll costs a channel rather than a tag read per second per
// waiter.
//
// IN-PROCESS ONLY, and that is a correctness statement rather than a caveat: a
// waiter on console A is never woken by a drain on console B. The door must
// therefore treat a wake as an OPTIMISATION and still read the tag when its
// timer fires — never as the only way an answer arrives.
type scanWaiters struct {
	mu sync.Mutex
	m  map[scanWaitKey]map[chan struct{}]struct{}
	// perApp and total are the two caps (§5.2). Counted rather than derived
	// from m, because m is keyed by OBJECT: one app parking a thousand
	// requests across a thousand keys is a thousand map entries of one, which
	// a per-key length would read as no load at all.
	perApp map[objstore.AppRef]int
	total  int
}

// scanWait registers interest in one object and returns the channel to select
// on, plus the cancel that must be deferred.
//
// The channel is BUFFERED to depth one and the notifier never blocks on it, so
// a waiter that has already given up cannot stall the drain — the goroutine
// whose latency this whole path exists to protect.
func (s *Server) scanWait(ref objstore.AppRef, key string) (ch <-chan struct{}, release func(), ok bool) {
	w := &s.scanWaits
	k := scanWaitKey{ref: ref, key: key}
	c := make(chan struct{}, 1)
	w.mu.Lock()
	defer w.mu.Unlock()
	if w.total >= scanWaitersGlobal || w.perApp[ref] >= scanWaitersPerApp {
		// Refused a PARK, not refused an answer. The caller reads the object
		// and replies truthfully; only the waiting is skipped, so a full
		// console degrades to polling rather than to errors.
		return nil, func() {}, false
	}
	if w.m == nil {
		w.m = map[scanWaitKey]map[chan struct{}]struct{}{}
		w.perApp = map[objstore.AppRef]int{}
	}
	if w.m[k] == nil {
		w.m[k] = map[chan struct{}]struct{}{}
	}
	w.m[k][c] = struct{}{}
	w.perApp[ref]++
	w.total++
	released := false
	return c, func() {
		w.mu.Lock()
		defer w.mu.Unlock()
		// IDEMPOTENT. The status door defers this and the test calls it early;
		// a second call that decremented again would leak capacity downwards
		// until the console stopped parking anything at all.
		if released {
			return
		}
		released = true
		if set := w.m[k]; set != nil {
			delete(set, c)
			if len(set) == 0 {
				// The map must not grow one entry per key ever waited on.
				delete(w.m, k)
			}
		}
		w.perApp[ref]--
		if w.perApp[ref] <= 0 {
			delete(w.perApp, ref)
		}
		w.total--
	}, true
}

// scanNotify wakes everyone waiting on one object. Never blocks.
func (s *Server) scanNotify(ref objstore.AppRef, key string) {
	w := &s.scanWaits
	w.mu.Lock()
	defer w.mu.Unlock()
	for ch := range w.m[scanWaitKey{ref: ref, key: key}] {
		select {
		case ch <- struct{}{}:
		default: // already woken and not yet read; one wake is all it needs
		}
	}
}