// The sweep (spec §7): a resumable reconciliation pass over one bucket.
//
// It is a reconciler, not a pipeline. Nothing in CARLOS sees a PUT — the
// control plane hands an app its own credentials and then never touches the
// data plane — so there is no upload to hook. What a reconciler can do that a
// pipeline cannot is COUNT WHAT IT HAS NOT SEEN: a pipeline's misses are
// invisible by construction, while a walk that lists the bucket knows exactly
// how many objects carry no verdict. That number is the whole point. An audit
// answer needs two sentences, and the second one is the honest one.
package objstore

import (
	"context"
	"errors"
	"fmt"
	"time"

	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"

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

// sweepPageSize is how many keys one listing page asks for. The page is the
// unit of durable progress: everything on it is scanned, then the sidecar is
// written once. Smaller pages mean more sidecar writes; larger ones mean more
// work lost when a tick's budget runs out mid-page.
const sweepPageSize = 1000

// ScanObject is one key as the listing reports it. Size, ETag and
// last-modified all come free on the listing page, which is why the sweep
// never needs a HeadObject: the size decides the ceiling before any fetch, and
// the other two bind the verdict.
type ScanObject struct {
	Key      string
	ETag     string
	Modified time.Time
	Size     int64
}

// Version is the token a verdict is bound to: this exact version of these
// exact bytes.
//
// Both halves earn their place. The ETag alone would be the content hash, and
// for a single-part upload that hash is MD5 — chosen-prefix MD5 collisions are
// practical, so an attacker controlling both uploads could give a payload the
// ETag of the clean object it replaces and inherit its verdict. Last-modified
// alone would be too coarse. Together they are cheap and neither is the
// uploader's to choose.
func (o ScanObject) Version() string {
	return normalizeETag(o.ETag) + "@" + o.Modified.UTC().Format(time.RFC3339)
}

// ObjectLister is the one listing call the sweep makes, narrowed so the walk
// is testable without S3 or credentials.
type ObjectLister interface {
	// ListPage returns up to limit keys ordered after startAfter, and whether
	// more remain.
	ListPage(ctx context.Context, bucket, startAfter string, limit int32) (objs []ScanObject, truncated bool, err error)
}

// s3Lister adapts an S3 client to ObjectLister.
type s3Lister struct{ s3 listPageAPI }

type listPageAPI interface {
	ListObjectsV2(ctx context.Context, in *awss3.ListObjectsV2Input, optFns ...func(*awss3.Options)) (*awss3.ListObjectsV2Output, error)
}

// NewObjectLister adapts an S3 client to the sweep's listing seam.
func NewObjectLister(c listPageAPI) ObjectLister { return &s3Lister{s3: c} }

func (l *s3Lister) ListPage(ctx context.Context, bucket, startAfter string, limit int32) ([]ScanObject, bool, error) {
	in := &awss3.ListObjectsV2Input{Bucket: &bucket, MaxKeys: &limit}
	// StartAfter, not ContinuationToken. A token's lifetime across processes
	// is unspecified and, on Ceph, unmeasured; a key is a portable resume
	// point that survives a console restart and a change of S3 vendor.
	if startAfter != "" {
		in.StartAfter = &startAfter
	}
	out, err := l.s3.ListObjectsV2(ctx, in)
	if err != nil {
		return nil, false, fmt.Errorf("objstore: list %s: %w", bucket, err)
	}
	objs := make([]ScanObject, 0, len(out.Contents))
	for _, o := range out.Contents {
		obj := ScanObject{Key: aws2s(o.Key), ETag: normalizeETag(aws2s(o.ETag))}
		if o.Size != nil {
			obj.Size = *o.Size
		}
		if o.LastModified != nil {
			obj.Modified = *o.LastModified
		}
		objs = append(objs, obj)
	}
	truncated := out.IsTruncated != nil && *out.IsTruncated
	return objs, truncated, nil
}

// errStandDown is not a failure. It is this sweep noticing that someone else —
// another driver, or a teardown — is ahead of it, and that the right move is
// to stop writing. RunSweep converts it to a nil error.
var errStandDown = errors.New("objstore: sweep standing down")

// The two sweep failures an operator can actually act on, named so a door can
// tell them apart from a transport error and say which it was. Everything else
// stays a logged detail: bucket names, ARNs and request ids do not belong in a
// client sentence.
var (
	// ErrEngineUnreachable is the engine not answering at all.
	ErrEngineUnreachable = errors.New("objstore: the scanning engine is not answering")
	// ErrScanningDisabled is the deployment switch being off. Its own sentinel
	// because a door must tell "nobody turned this on" apart from "the engine
	// is broken": one is a command away, the other is an incident.
	ErrScanningDisabled = errors.New("objstore: object scanning is disabled for this deployment")
	// ErrEngineDrift is a record naming a different engine than the one
	// configured. Refusing beats recounting another engine's verdicts as
	// unscanned, which would destroy the coverage history.
	ErrEngineDrift = errors.New("objstore: the record names a different scanning engine")
)

// RunSweep advances one bucket's coverage by up to budget of work and returns
// the sidecar as it now stands.
//
// A nil sidecar with a nil error means there was nothing to do: no store, no
// scanning enabled, or an app in the trash.
func claimRawRunSweep(ctx context.Context, s release.Store, lister ObjectLister, sc Scanner,
	acct, app string, budget time.Duration) (*ScanSidecar, error) {

	rec, _, err := Get(ctx, s, acct, app)
	if err != nil {
		return nil, err
	}
	if rec == nil || rec.Provisioned == nil || rec.Scanning == nil || rec.Scanning.Enabled == nil {
		return nil, nil
	}

	// A trashed store is destroyed within 30 days. Scanning it bills for
	// egress on data that is on its way out, and reports coverage for an app
	// nobody can reach.
	tomb, err := release.ReadAppTombstone(ctx, s, acct, app)
	if err != nil {
		return nil, err
	}
	if tomb != nil {
		return nil, nil
	}

	// A scanner cannot read ciphertext. Sweeping a client-encrypted store
	// would report every object clean, which is worse than reporting nothing:
	// it is a true-looking sentence about bytes nobody examined. Mark it and
	// stop.
	if rec.Scanning.Requested != nil && rec.Scanning.Requested.Encrypted {
		return markNotApplicable(ctx, s, acct, app)
	}

	// An engine enabled with no attestation on record is refused rather than
	// swept. The grant door will not create that state any more, so this is
	// unreachable in practice — but sweeping reads tenant object bytes, and
	// the attestation is the consent for that. Guessing "probably plaintext"
	// is not a decision this walk gets to make on the member's behalf.
	if rec.Scanning.Requested == nil {
		return nil, fmt.Errorf("objstore: %s/%s: scanning is enabled but no attestation is on record "+
			"— the member runs `carlos store scan request` to say whether these objects are readable",
			acct, app)
	}

	if sc == nil {
		return nil, fmt.Errorf("objstore: %s/%s has scanning enabled but no engine is configured "+
			"(set CARLOS_CLAMD_ADDR)", acct, app)
	}

	// The DEPLOYMENT switch, checked here and not only in the pass.
	//
	// A tick that goes quiet is not the same as scanning being off: the
	// on-demand sweep verb reaches this function by its own path, so without
	// this an operator could run `carlos system antivirus disable`, watch the
	// hourly pass stop, and still have `carlos store scan sweep` read tenant
	// bytes. Disable has to disable everywhere or it disables nothing.
	on, err := release.AntivirusEnabled(ctx, s)
	if err != nil {
		return nil, err
	}
	if !on {
		return nil, fmt.Errorf("%w (%s/%s)", ErrScanningDisabled, acct, app)
	}

	engine, signature, err := sc.Provenance(ctx)
	if err != nil {
		return nil, fmt.Errorf("%w (%s/%s): %v", ErrEngineUnreachable, acct, app, err)
	}
	// Engine drift refuses rather than guesses. Verdicts already in the bucket
	// were written by a different engine under its own conventions, and
	// recounting them as unscanned would destroy the coverage history. An
	// operator changes engine by revoking and re-granting, deliberately.
	if rec.Scanning.Enabled.Engine != engine {
		return nil, fmt.Errorf("%w (%s/%s): record says %q, configured engine is %q",
			ErrEngineDrift, acct, app, rec.Scanning.Enabled.Engine, engine)
	}

	side, etag, err := GetSidecar(ctx, s, acct, app)
	if err != nil {
		return nil, err
	}
	if side == nil {
		side = &ScanSidecar{}
	}
	side.NotApplicable = false
	if side.Current == nil {
		side.Current = &Sweep{StartedAt: time.Now().UTC()}
	}
	side.Current.Engine = engine
	side.Current.Signature = signature

	bucket := rec.Provisioned.Bucket
	deadline := time.Now().Add(budget)

	for {
		pageStart := side.Current.After
		objs, truncated, err := lister.ListPage(ctx, bucket, pageStart, sweepPageSize)
		if err != nil {
			return nil, err
		}
		// Verdicts reached on THIS page versus verdicts carried forward from an
		// earlier sweep. A clean tag is reused whenever the ETag still matches,
		// so a completed rollup can consist almost entirely of judgements made
		// months ago — against signatures that are not the ones the sidecar now
		// names. Counting the two apart is what stops "engine database" being
		// read as "every object below was judged against this".
		var fresh, reused int64

		for _, o := range objs {
			if err := ctx.Err(); err != nil {
				return nil, err
			}
			v, err := sc.Verdict(ctx, bucket, o.Key, o.Version())
			if errors.Is(err, ErrObjectGone) {
				continue // deleted since the listing; not ours to count
			}
			if err != nil {
				return nil, err
			}
			if v == VerdictUnknown {
				// A scan error here is the ENGINE failing, not this object:
				// the next object would fail identically. Abandon the page
				// without writing. Nothing is lost but the listing — every
				// verdict already reached is a tag in the bucket, so the
				// re-run reads them back without re-scanning a byte.
				//
				// Per-object facts do NOT arrive here as errors. An object we
				// are not allowed to read comes back as VerdictDenied and one
				// that has been deleted as ErrObjectGone, precisely so that
				// neither can stall the page: an error here abandons the page
				// unwritten, and a page that fails identically every tick
				// means the bucket's sweep never advances past it.
				// The recorded flag is the on-demand drain's business, not
				// the sweep's: Fresh and Reused count verdicts REACHED, and
				// the sweep re-lists every cycle anyway, so an unpersisted
				// verdict simply gets reached again.
				v, _, err = sc.Scan(ctx, bucket, o.Key, o.Version(), o.Size)
				if errors.Is(err, ErrObjectGone) {
					continue
				}
				if err != nil {
					return nil, err
				}
				if isCoverage(v) {
					fresh++
				}
			} else if isCoverage(v) {
				reused++
			}
			side.Current.count(v, o.Size)
		}

		side.Current.Fresh += fresh
		side.Current.Reused += reused
		side.Current.At = time.Now().UTC()
		if len(objs) > 0 {
			side.Current.After = objs[len(objs)-1].Key
		}

		last := !truncated
		if last {
			// The completed sweep becomes the answer, and Current clears. A
			// single field would have the next tick overwrite the compliance
			// artifact with page-one partials.
			side.Current.After = ""
			side.LastComplete = side.Current
			side.Current = nil
		}

		newETag, err := writeSidecar(ctx, s, acct, app, side, etag, pageStart)
		if errors.Is(err, errStandDown) {
			return side, nil
		}
		if err != nil {
			return nil, err
		}
		etag = newETag

		if last || time.Now().After(deadline) {
			return side, nil
		}
	}
}

// isCoverage is a verdict that counts as Scanned. Fresh and Reused split THAT
// number, not the object count: an object that was skipped or denied was never
// judged at all, so calling its verdict "fresh" would make the printed
// arithmetic — "N of M scanned, of those X judged now" — visibly wrong the
// moment a single object is too large.
func isCoverage(v Verdict) bool {
	return v == VerdictClean || v == VerdictInfected
}

// count folds one object's verdict into the rollup.
//
// Scanned and Unscanned are not complements, and the arithmetic here is the
// reason the audit answer needs two sentences. An infected object WAS scanned
// — we know what it is. A skipped one was not judged at all.
func (sw *Sweep) count(v Verdict, size int64) {
	sw.Objects++
	sw.Bytes += size
	switch v {
	case VerdictClean:
		sw.Scanned++
	case VerdictInfected:
		sw.Scanned++
		sw.Infected++
	case VerdictSkipped:
		sw.Skipped++
		sw.Unscanned++
	case VerdictDenied:
		sw.Denied++
		sw.Unscanned++
	case VerdictFailed:
		sw.Failed++
		sw.Unscanned++
	default:
		sw.Unscanned++
	}
}

// markNotApplicable records that no coverage is claimed for this store, and
// does no listing at all.
func markNotApplicable(ctx context.Context, s release.Store, acct, app string) (*ScanSidecar, error) {
	side, etag, err := GetSidecar(ctx, s, acct, app)
	if err != nil {
		return nil, err
	}
	if side != nil && side.NotApplicable && side.Current == nil {
		return side, nil // already recorded; no write, no CAS churn
	}
	if side == nil {
		side = &ScanSidecar{}
	}
	side.NotApplicable = true
	side.Current = nil
	if _, err := writeSidecar(ctx, s, acct, app, side, etag, ""); err != nil {
		if errors.Is(err, errStandDown) {
			return side, nil
		}
		return nil, err
	}
	return side, nil
}

// writeSidecar performs the pre-write re-read and the CAS retry guard.
//
// The re-read closes a hole the retry guard alone cannot. A FRESH sweep's
// first write is create-only by construction — no sidecar exists, so the etag
// is "". Sequence: the sweep reads an enabled store.json, spends seconds
// listing and scanning a page, a teardown runs and PurgeApp deletes the whole
// prefix, and the sweep then writes scan.json create-only, which SUCCEEDS. The
// result is an orphan under a purged app, and that orphan is expensive: the
// app-name claim gate refuses any name whose prefix still holds a key, rename
// is refused, and box-side auto-purge never reaches it because the tombstone
// was written last. The app name becomes permanently unclaimable.
//
// The one-round-trip race this still leaves is the same one
// deleteRecordIfUnchanged already accepts.
func writeSidecar(ctx context.Context, s release.Store, acct, app string,
	side *ScanSidecar, etag, pageStart string) (string, error) {

	rec, _, err := Get(ctx, s, acct, app)
	if err != nil {
		return "", err
	}
	if rec == nil || rec.Scanning == nil || rec.Scanning.Enabled == nil {
		return "", errStandDown
	}

	newETag, err := PutSidecar(ctx, s, acct, app, side, etag)
	if err == nil {
		return newETag, nil
	}
	if !errors.Is(err, release.ErrCASConflict) {
		return "", err
	}

	// Someone else wrote. A conditional write against a DELETED key also
	// reads as ErrCASConflict, so a blind retry would fall back to a
	// create-only write and resurrect a sidecar teardown just removed. Look
	// before retrying.
	current, freshETag, gerr := GetSidecar(ctx, s, acct, app)
	if gerr != nil {
		return "", gerr
	}
	if current == nil || current.Current == nil || current.Current.After != pageStart {
		// Deleted, completed by another driver, or advanced past us. In every
		// case this sweep is no longer the one making progress.
		return "", errStandDown
	}
	newETag, err = PutSidecar(ctx, s, acct, app, side, freshETag)
	if err != nil {
		if errors.Is(err, release.ErrCASConflict) {
			return "", errStandDown
		}
		return "", err
	}
	return newETag, nil
}
func RunSweep(ctx context.Context, s release.Store, lister ObjectLister, sc Scanner, acct, app string, budget time.Duration) (claimResult0 *ScanSidecar, claimResult1 error) {
	claimResult1 = release.WithAppMutation(ctx, s, acct, app, "objstore.RunSweep", func(claimContext context.Context) error {
		claimResult0, claimResult1 = claimRawRunSweep(claimContext, s, lister, sc, acct, app, budget)
		return claimResult1
	})
	return
}