package objstore

import (
	"context"
	"crypto/md5"
	"encoding/hex"
	"errors"
	"os"
	"sort"
	"strings"
	"testing"
	"time"

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

// memLister lists a MemProvisioner bucket, paginated, with real content-hash
// ETags so the etag binding in Verdict is exercised rather than stubbed away.
type memLister struct {
	p        *MemProvisioner
	pageSize int
	calls    int
	// modified is each key's last-modified time. It is half of the version a
	// verdict binds to, so a lister that left it zero would make every sweep
	// test blind to that half.
	modified map[string]time.Time
	// before runs at the top of each ListPage, so a test can make the world
	// change underneath a sweep at the one moment that matters.
	before func()
}

func (l *memLister) touch(key string, at time.Time) {
	if l.modified == nil {
		l.modified = map[string]time.Time{}
	}
	l.modified[key] = at
}

func (l *memLister) ListPage(_ context.Context, bucket, startAfter string, _ int32) ([]ScanObject, bool, error) {
	l.calls++
	if l.before != nil {
		l.before()
	}
	keys := make([]string, 0, len(l.p.Buckets[bucket]))
	for k := range l.p.Buckets[bucket] {
		if k > startAfter {
			keys = append(keys, k)
		}
	}
	sort.Strings(keys)

	size := l.pageSize
	if size <= 0 {
		size = sweepPageSize
	}
	truncated := len(keys) > size
	if truncated {
		keys = keys[:size]
	}
	objs := make([]ScanObject, 0, len(keys))
	for _, k := range keys {
		body := l.p.Buckets[bucket][k]
		sum := md5.Sum(body)
		mod := l.modified[k]
		if mod.IsZero() {
			mod = time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)
		}
		objs = append(objs, ScanObject{
			Key:      k,
			ETag:     hex.EncodeToString(sum[:]),
			Modified: mod,
			Size:     int64(len(body)),
		})
	}
	return objs, truncated, nil
}

// refuseLister fails the test if it is ever asked for a page — for the cases
// where the right amount of work is none.
type refuseLister struct{ t *testing.T }

func (l *refuseLister) ListPage(context.Context, string, string, int32) ([]ScanObject, bool, error) {
	l.t.Fatal("listed a bucket that should never have been swept")
	return nil, false, nil
}

// sweepFixture stands up a provisioned, scanning-enabled store with n objects.
func sweepFixture(t *testing.T, n int) (release.Store, *MemProvisioner, *MemScanner, 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")

	bucket := rec.Provisioned.Bucket
	for i := 0; i < n; i++ {
		p.Buckets[bucket][objectName(i)] = []byte(strings.Repeat("x", i+1))
	}
	return s, p, NewMemScanner(), bucket
}

func objectName(i int) string {
	return "uploads/" + string(rune('a'+i)) + ".txt"
}

// sweepToCompletion drives RunSweep until a sweep finishes, so a paginated
// test does not have to guess how many ticks that takes. It bounds the loop:
// a sweep that never completes is a bug, not a slow test.
func sweepToCompletion(t *testing.T, s release.Store, l ObjectLister, sc Scanner) *ScanSidecar {
	t.Helper()
	var side *ScanSidecar
	for i := 0; i < 50; i++ {
		var err error
		side, err = RunSweep(context.Background(), s, l, sc, "bab", "titogo", 0)
		if err != nil {
			t.Fatalf("sweep %d: %v", i, err)
		}
		if side != nil && side.LastComplete != nil {
			return side
		}
	}
	t.Fatal("sweep never completed in 50 ticks")
	return nil
}

// A synchronous engine's FIRST pass reports objects as scanned. Counting them
// unscanned because "the next sweep reads the result" is GuardDuty's shape,
// and applying it to ClamAV would make the audit sentence false for a full
// re-sweep interval.
func TestSweepCountsSynchronousVerdicts(t *testing.T) {
	s, p, sc, _ := sweepFixture(t, 3)

	side, err := RunSweep(context.Background(), s, &memLister{p: p}, sc, "bab", "titogo", time.Minute)
	if err != nil {
		t.Fatal(err)
	}
	if side.LastComplete == nil {
		t.Fatal("a sweep that listed every object in one page did not complete")
	}
	got := side.LastComplete
	if got.Objects != 3 || got.Scanned != 3 || got.Unscanned != 0 {
		t.Fatalf("objects=%d scanned=%d unscanned=%d, want 3/3/0", got.Objects, got.Scanned, got.Unscanned)
	}
	if got.Engine != "clamav" || got.Signature == "" {
		t.Fatalf("no provenance recorded: engine=%q signature=%q", got.Engine, got.Signature)
	}
	if side.Current != nil {
		t.Fatal("a completed sweep left Current set")
	}
}

// LastComplete survives the tick after completion. A single Sweep field would
// overwrite the audit answer with page-one partials every cycle — the
// compliance artifact destroyed on a timer.
func TestSweepLastCompleteSurvivesNextTick(t *testing.T) {
	s, p, sc, _ := sweepFixture(t, 5)
	l := &memLister{p: p, pageSize: 2}

	side := sweepToCompletion(t, s, l, sc)
	if side.LastComplete.Objects != 5 {
		t.Fatalf("completed sweep counted %d objects, want 5", side.LastComplete.Objects)
	}

	// One more tick: a fresh sweep starts and gets exactly one page in.
	next, err := RunSweep(context.Background(), s, l, sc, "bab", "titogo", 0)
	if err != nil {
		t.Fatal(err)
	}
	if next.Current == nil || next.Current.Objects != 2 {
		t.Fatalf("the next tick did not start a fresh partial sweep: %+v", next.Current)
	}
	if next.LastComplete == nil || next.LastComplete.Objects != 5 {
		t.Fatalf("page-one partials overwrote the audit answer: %+v", next.LastComplete)
	}
}

// Interrupted mid-bucket and re-run, a sweep reaches the same rollup without
// double-counting what it already saw.
func TestSweepResumesWithoutDoubleCounting(t *testing.T) {
	s, p, sc, _ := sweepFixture(t, 5)
	l := &memLister{p: p, pageSize: 2}

	side := sweepToCompletion(t, s, l, sc)
	if l.calls < 3 {
		t.Fatalf("the fixture did not actually paginate: %d listing calls", l.calls)
	}
	if side.LastComplete.Objects != 5 || side.LastComplete.Scanned != 5 {
		t.Fatalf("resumed sweep counted objects=%d scanned=%d, want 5/5",
			side.LastComplete.Objects, side.LastComplete.Scanned)
	}
}

// A fresh sweep over a bucket whose app is torn down between the listing and
// the write must leave NOTHING behind: an orphan scan.json under a purged
// prefix makes the app name permanently unclaimable, because the claim gate
// refuses any name whose prefix still holds a key and box-side auto-purge
// never reaches it.
//
// The first write of a fresh sweep is create-only by construction, so the CAS
// retry guard cannot catch this on its own — only the pre-write re-read can.
func TestSweepDoesNotResurrectTornDownApp(t *testing.T) {
	ctx := context.Background()
	s, p, sc, _ := sweepFixture(t, 3)

	l := &memLister{p: p}
	l.before = func() {
		// Teardown lands while this sweep is listing.
		if err := s.Delete(ctx, Key("bab", "titogo")); err != nil {
			t.Fatal(err)
		}
	}

	side, err := RunSweep(ctx, s, l, sc, "bab", "titogo", time.Minute)
	if err != nil {
		t.Fatalf("standing down is not an error: %v", err)
	}
	_ = side
	if _, _, err := s.Get(ctx, ScanKey("bab", "titogo")); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the sweep wrote an orphan sidecar under a purged app: %v", err)
	}
}

// A client-encrypted store is marked not_applicable and does no work. A
// scanner cannot read ciphertext, and reporting every object clean would be a
// true-looking sentence about bytes nobody examined.
func TestSweepSkipsEncryptedStore(t *testing.T) {
	ctx := context.Background()
	s, _, sc, _ := sweepFixture(t, 3)

	rec, etag, err := Get(ctx, s, "bab", "titogo")
	if err != nil {
		t.Fatal(err)
	}
	rec.Scanning.Requested = &ScanRequested{At: time.Now().UTC(), By: "m@example.com", Encrypted: true}
	if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
		t.Fatal(err)
	}

	side, err := RunSweep(ctx, s, &refuseLister{t: t}, sc, "bab", "titogo", time.Minute)
	if err != nil {
		t.Fatal(err)
	}
	if side == nil || !side.NotApplicable {
		t.Fatalf("an encrypted store was not marked not_applicable: %+v", side)
	}
	if side.LastComplete != nil {
		t.Fatal("coverage was claimed for a store nobody could read")
	}
}

// A trashed app is not swept: it is destroyed within 30 days and scanning it
// bills for egress on data on its way out.
func TestSweepSkipsTrashedApp(t *testing.T) {
	ctx := context.Background()
	s, _, sc, _ := sweepFixture(t, 3)
	if err := release.TrashApp(ctx, s, "bab", "titogo", "op@example.com", time.Now().UTC()); err != nil {
		t.Fatal(err)
	}

	side, err := RunSweep(ctx, s, &refuseLister{t: t}, sc, "bab", "titogo", time.Minute)
	if err != nil {
		t.Fatal(err)
	}
	if side != nil {
		t.Fatalf("a trashed app was swept: %+v", side)
	}
}

// Skipped is not clean, and both are visible separately. An object over the
// size ceiling was never judged; folding it into Scanned would make the audit
// sentence false in exactly the direction that matters.
func TestSweepCountsSkippedSeparately(t *testing.T) {
	s, p, sc, _ := sweepFixture(t, 3)
	sc.ScanAs(objectName(1), VerdictSkipped)

	side, err := RunSweep(context.Background(), s, &memLister{p: p}, sc, "bab", "titogo", time.Minute)
	if err != nil {
		t.Fatal(err)
	}
	got := side.LastComplete
	if got.Objects != 3 || got.Scanned != 2 || got.Skipped != 1 {
		t.Fatalf("objects=%d scanned=%d skipped=%d, want 3/2/1", got.Objects, got.Scanned, got.Skipped)
	}
	if got.Unscanned != 1 {
		t.Fatalf("unscanned=%d, want 1 — a skipped object is not covered", got.Unscanned)
	}
}

// An infected object WAS scanned. We know exactly what it is; it is coverage,
// not a gap. This is the one place the two counts could plausibly have been
// wired the other way round.
func TestSweepCountsInfectedAsScanned(t *testing.T) {
	s, p, sc, _ := sweepFixture(t, 2)
	sc.ScanAs(objectName(0), VerdictInfected)

	side, err := RunSweep(context.Background(), s, &memLister{p: p}, sc, "bab", "titogo", time.Minute)
	if err != nil {
		t.Fatal(err)
	}
	got := side.LastComplete
	if got.Infected != 1 || got.Scanned != 2 || got.Unscanned != 0 {
		t.Fatalf("infected=%d scanned=%d unscanned=%d, want 1/2/0",
			got.Infected, got.Scanned, got.Unscanned)
	}
}

// A record naming another engine refuses rather than guessing. Verdicts in the
// bucket were written under conventions this engine does not read, and
// recounting them as unscanned would destroy the coverage history.
func TestSweepRefusesOnEngineDrift(t *testing.T) {
	ctx := context.Background()
	s, p, sc, _ := sweepFixture(t, 1)

	rec, etag, err := Get(ctx, s, "bab", "titogo")
	if err != nil {
		t.Fatal(err)
	}
	rec.Scanning.Enabled.Engine = "guardduty"
	if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
		t.Fatal(err)
	}

	_, err = RunSweep(ctx, s, &memLister{p: p}, sc, "bab", "titogo", time.Minute)
	if err == nil {
		t.Fatal("a sweep ran against a record naming a different engine")
	}
	if !strings.Contains(err.Error(), "guardduty") {
		t.Fatalf("the refusal does not name the engine on the record: %v", err)
	}
}

// An engine that goes away mid-page fails the pass and writes NOTHING. Every
// verdict already reached is a tag in the bucket, so the re-run reads them
// back without re-scanning a byte — but a half-written rollup would be a
// number nobody could interpret.
func TestSweepEngineFailureWritesNoRollup(t *testing.T) {
	ctx := context.Background()
	s, p, sc, _ := sweepFixture(t, 3)
	sc.Fail("Scan", errors.New("dial unix /run/clamav/clamd.ctl: connection refused"))

	if _, err := RunSweep(ctx, s, &memLister{p: p}, sc, "bab", "titogo", time.Minute); err == nil {
		t.Fatal("an unreachable engine reported a successful sweep")
	}
	if _, _, err := s.Get(ctx, ScanKey("bab", "titogo")); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("a failed sweep left a rollup behind: %v", err)
	}
}

// A store with no scanning enabled is not swept at all — which is every store
// on a deployment that has not opted in.
func TestSweepIgnoresStoreWithoutScanning(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	_, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-west-1")
	rotateFixture(t, s, pub, p)

	side, err := RunSweep(context.Background(), s, &refuseLister{t: t}, NewMemScanner(),
		"bab", "titogo", time.Minute)
	if err != nil {
		t.Fatal(err)
	}
	if side != nil {
		t.Fatalf("a store that never opted in was swept: %+v", side)
	}
}

// The CAS retry guard, which the pre-write re-read does NOT cover.
//
// Two writes are in play. The FIRST write of a fresh sweep is create-only, and
// the pre-write re-read is what protects that one. A LATER page's write is
// conditional on an etag, and losing that CAS means someone else moved the
// sidecar — another driver (the on-demand sweep verb races the hourly tick by
// design), or a teardown deleting it. A conditional write against a DELETED
// key also reads as ErrCASConflict, so a blind retry falls back to create-only
// and resurrects the object teardown just removed.
//
// Here the sidecar is deleted between this sweep's page-two write attempt and
// its conflict re-read. The sweep must stand down, not resurrect.
func TestSweepStandsDownWhenSidecarVanishesMidRetry(t *testing.T) {
	ctx := context.Background()
	s, p, sc, _ := sweepFixture(t, 5)
	l := &memLister{p: p, pageSize: 2}

	// One page, so a real sidecar with a Current exists and the next write is
	// a conditional replace rather than a create.
	if _, err := RunSweep(ctx, s, l, sc, "bab", "titogo", 0); err != nil {
		t.Fatal(err)
	}
	if _, _, err := s.Get(ctx, ScanKey("bab", "titogo")); err != nil {
		t.Fatalf("setup: no sidecar to conflict against: %v", err)
	}

	// Now make the next write lose its CAS *and* find the key gone. Writing a
	// different body under a new etag invalidates the caller's etag; deleting
	// it afterwards is the teardown.
	l.before = func() {
		l.before = nil // once: only the second page races
		if _, err := PutSidecar(ctx, s, "bab", "titogo",
			&ScanSidecar{Current: &Sweep{After: "somewhere-else"}}, ""); err == nil {
			t.Fatal("expected the create-only write to conflict with the existing sidecar")
		}
		// Delete it outright — the teardown case.
		if err := s.Delete(ctx, ScanKey("bab", "titogo")); err != nil {
			t.Fatal(err)
		}
	}

	side, err := RunSweep(ctx, s, l, sc, "bab", "titogo", 0)
	if err != nil {
		t.Fatalf("standing down is not an error: %v", err)
	}
	_ = side
	if _, _, err := s.Get(ctx, ScanKey("bab", "titogo")); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the sweep resurrected a sidecar that had been deleted: %v", err)
	}
}

// Fresh and reused verdicts are counted apart. A clean tag is reused for as
// long as the ETag matches, so a completed rollup can be built almost entirely
// from judgements made months ago — against a different signature database
// than the one the sidecar names. Without this split, "engine database" reads
// as a claim about every object in the rollup, and it is not one.
func TestSweepCountsFreshAndReusedVerdictsApart(t *testing.T) {
	ctx := context.Background()
	s, p, sc, _ := sweepFixture(t, 3)

	first, err := RunSweep(ctx, s, &memLister{p: p}, sc, "bab", "titogo", time.Minute)
	if err != nil {
		t.Fatal(err)
	}
	if first.LastComplete.Fresh != 3 || first.LastComplete.Reused != 0 {
		t.Fatalf("first sweep: fresh=%d reused=%d, want 3/0",
			first.LastComplete.Fresh, first.LastComplete.Reused)
	}

	// Nothing changed in the bucket, so the second sweep reaches no new
	// verdicts at all — it reads back the tags the first one wrote.
	second, err := RunSweep(ctx, s, &memLister{p: p}, sc, "bab", "titogo", time.Minute)
	if err != nil {
		t.Fatal(err)
	}
	if second.LastComplete.Fresh != 0 || second.LastComplete.Reused != 3 {
		t.Fatalf("second sweep: fresh=%d reused=%d, want 0/3 — every verdict was carried forward",
			second.LastComplete.Fresh, second.LastComplete.Reused)
	}
	if second.LastComplete.Scanned != 3 {
		t.Fatalf("scanned = %d, want 3 — reused verdicts are still coverage",
			second.LastComplete.Scanned)
	}
}

// An object that vanishes between the listing and the fetch is dropped, not
// counted. It is neither covered nor a gap — it is not there.
func TestSweepDropsObjectsThatDisappearMidPage(t *testing.T) {
	ctx := context.Background()
	s, p, _, _ := sweepFixture(t, 3)
	sc := &goneScanner{gone: map[string]bool{objectName(1): true}}

	side, err := RunSweep(ctx, s, &memLister{p: p}, sc, "bab", "titogo", time.Minute)
	if err != nil {
		t.Fatalf("a deleted object failed the sweep: %v", err)
	}
	if side.LastComplete.Objects != 2 {
		t.Fatalf("objects = %d, want 2 — the deleted one must not be counted",
			side.LastComplete.Objects)
	}
}

// goneScanner answers ErrObjectGone for named keys and clean for the rest.
type goneScanner struct {
	MemScanner
	gone map[string]bool
}

func (g *goneScanner) Verdict(context.Context, string, string, string) (Verdict, error) {
	return VerdictUnknown, nil
}

func (g *goneScanner) Scan(_ context.Context, _, key, _ string, _ int64) (Verdict, bool, error) {
	if g.gone[key] {
		return VerdictUnknown, false, ErrObjectGone
	}
	return VerdictClean, true, nil
}

func (g *goneScanner) Provenance(context.Context) (string, string, error) {
	return "clamav", "test/0/never", nil
}

// The sweep binds verdicts to the object VERSION, not the ETag alone.
//
// Same bytes, later last-modified: an overwrite that happened to produce the
// same content hash. A chosen-prefix MD5 collision is exactly this shape, and
// it is the case enforcement exists for — an app that accepts uploads and
// serves them back. Passing only the ETag here would let the payload inherit
// the clean verdict of the object it replaced, and every existing sweep test
// would still have passed.
func TestSweepRescansWhenOnlyLastModifiedChanges(t *testing.T) {
	ctx := context.Background()
	s, p, sc, bucket := sweepFixture(t, 1)
	l := &memLister{p: p}

	first, err := RunSweep(ctx, s, l, sc, "bab", "titogo", time.Minute)
	if err != nil {
		t.Fatal(err)
	}
	if first.LastComplete.Fresh != 1 {
		t.Fatalf("setup: fresh = %d, want 1", first.LastComplete.Fresh)
	}

	// The bytes are untouched — so the ETag is identical — but the object was
	// rewritten.
	_ = bucket
	l.touch(objectName(0), time.Date(2026, 9, 2, 0, 0, 0, 0, time.UTC))

	second, err := RunSweep(ctx, s, l, sc, "bab", "titogo", time.Minute)
	if err != nil {
		t.Fatal(err)
	}
	if second.LastComplete.Fresh != 1 || second.LastComplete.Reused != 0 {
		t.Fatalf("fresh=%d reused=%d — an overwritten object inherited its old verdict",
			second.LastComplete.Fresh, second.LastComplete.Reused)
	}
}

// Fresh and Reused split SCANNED, not the object count. An object that was
// skipped was never judged, so counting its verdict as "judged during this
// sweep" makes the printed arithmetic — "N of M scanned, of those X judged
// now" — visibly wrong the moment one object is too large.
func TestSweepFreshAndReusedSplitScannedNotObjects(t *testing.T) {
	s, p, sc, _ := sweepFixture(t, 3)
	sc.ScanAs(objectName(1), VerdictSkipped)

	side, err := RunSweep(context.Background(), s, &memLister{p: p}, sc, "bab", "titogo", time.Minute)
	if err != nil {
		t.Fatal(err)
	}
	got := side.LastComplete
	if got.Fresh+got.Reused != got.Scanned {
		t.Fatalf("fresh(%d)+reused(%d) = %d, but scanned = %d of %d objects",
			got.Fresh, got.Reused, got.Fresh+got.Reused, got.Scanned, got.Objects)
	}
	if got.Fresh != 2 {
		t.Fatalf("fresh = %d, want 2 — the skipped object was never judged", got.Fresh)
	}
}

// An engine enabled with no attestation on record is refused, not swept.
// Sweeping reads tenant object bytes and the attestation is the consent for
// that; guessing "probably plaintext" is not this walk's decision to make.
func TestSweepRefusesWhenNoAttestationIsOnRecord(t *testing.T) {
	ctx := context.Background()
	s, p, sc, _ := sweepFixture(t, 1)

	rec, etag, err := Get(ctx, s, "bab", "titogo")
	if err != nil {
		t.Fatal(err)
	}
	rec.Scanning.Requested = nil
	if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
		t.Fatal(err)
	}

	if _, err := RunSweep(ctx, s, &memLister{p: p}, sc, "bab", "titogo", time.Minute); err == nil {
		t.Fatal("a store with no attestation on record was swept anyway")
	}
}