package objstore

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log"
	"strings"
	"time"

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

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

// scanS3API is the subset of *s3.Client the scanner calls, narrowed the same
// way s3API is so a test can stand a stub in with no network or credentials.
//
// It is deliberately separate from s3API: the provisioner creates and destroys
// buckets and never reads an object, the scanner reads objects and never
// touches a bucket. Keeping them apart means neither can quietly grow the
// other's powers.
type scanS3API interface {
	GetObject(ctx context.Context, in *awss3.GetObjectInput, optFns ...func(*awss3.Options)) (*awss3.GetObjectOutput, error)
	HeadObject(ctx context.Context, in *awss3.HeadObjectInput, optFns ...func(*awss3.Options)) (*awss3.HeadObjectOutput, error)
	GetObjectTagging(ctx context.Context, in *awss3.GetObjectTaggingInput, optFns ...func(*awss3.Options)) (*awss3.GetObjectTaggingOutput, error)
	PutObjectTagging(ctx context.Context, in *awss3.PutObjectTaggingInput, optFns ...func(*awss3.Options)) (*awss3.PutObjectTaggingOutput, error)
}

// streamScanner is the engine seam: what clamavScanner needs from a clamd.
// *clamd.Client satisfies it. The wire protocol is tested in internal/clamd,
// so tests here fake this interface rather than restaging a socket.
type streamScanner interface {
	ScanStream(ctx context.Context, r io.Reader) (clamd.Result, error)
	Version(ctx context.Context) (string, error)
}

// clamavScanner scans objects by streaming them past a local clamd and records
// the verdict as object tags.
//
// Nothing about the bucket changes: there is no per-bucket plan to enable, no
// quota to consume and no per-bucket cost. That is the whole reason ClamAV is
// the primary engine — the bought alternative caps at 25 protected buckets per
// account per Region and cannot be raised.
type clamavScanner struct {
	s3       scanS3API
	engine   streamScanner
	maxBytes int64
	// signer is what makes a tag a verdict. See scantag.go: the app can write
	// these tags on some gateways whatever the IAM policy says, so authenticity
	// and not write-permission is the property. Never nil in a running
	// deployment — the console refuses to construct a scanner without one.
	signer *TagSigner
}

var _ Scanner = (*clamavScanner)(nil)

// NewClamAVScanner builds a Scanner against S3 in region and a clamd on
// socket. maxBytes is the size ceiling above which an object is recorded as
// skipped without being fetched; it should mirror clamd's own StreamMaxLength.
//
// This does NOT ping clamd. An engine that is down at boot — mid signature
// update, say — must not take the console with it: object scanning is an
// opt-in feature of one subsystem, and failing the whole control plane closed
// because a scanner is restarting would be a far worse outage than the one it
// prevents. An unreachable engine surfaces through Provenance, which is what
// `carlos store scan status` reports, and fails the pass loudly per tick.
func NewClamAVScanner(ctx context.Context, region, endpoint, socket string, maxBytes int64, signer *TagSigner) (Scanner, error) {
	if signer == nil {
		// A scanner with no signer would write verdicts nothing can verify and
		// read every forgery as genuine. Refuse at construction, where an
		// operator sees it, rather than at the first tag.
		return nil, ErrNoTagKey
	}
	s3c, err := S3ClientFor(ctx, region, endpoint)
	if err != nil {
		return nil, fmt.Errorf("objstore: clamav scanner: %w", err)
	}
	return &clamavScanner{
		s3:       s3c,
		engine:   clamd.New(socket, clamd.Timeouts{}),
		maxBytes: maxBytes,
		signer:   signer,
	}, nil
}

// EnsureEnabled is a no-op. There is nothing on the bucket to configure: the
// engine is reached from the control plane, not attached to the store.
func (c *clamavScanner) EnsureEnabled(ctx context.Context, bucket string) error { return nil }

// EnsureDisabled is a no-op for the same reason. Nothing was attached, so
// nothing leaks when a bucket is destroyed.
func (c *clamavScanner) EnsureDisabled(ctx context.Context, bucket string) error { return nil }

// Capacity reports no limit. -1 is "unbounded", distinct from 0, which would
// mean a real engine with nothing left.
func (c *clamavScanner) Capacity(ctx context.Context) (remaining, capacity int, err error) {
	return -1, -1, nil
}

// Provenance names the engine and its signature database — the two facts an
// audit answer needs beside a count. A scan is only as good as the signatures
// it ran against, so "clean" without a database version is not evidence.
func (c *clamavScanner) Provenance(ctx context.Context) (engine, version string, err error) {
	v, err := c.engine.Version(ctx)
	if err != nil {
		return "clamav", "", err
	}
	return "clamav", v, nil
}

// Enforcement is the tag the store user's policy denies on.
func (c *clamavScanner) Enforcement() DenyTag {
	return DenyTag{Key: TagVerdict, Value: string(VerdictInfected)}
}

// Verdict reports what the engine previously said about this exact version of
// the object.
//
// A verdict is bound to the object VERSION it judged — see ObjectVersion: the
// ETag and the last-modified time together.
//
// Without that binding a key that once held something clean is a laundering
// channel: upload clean, get tagged clean, overwrite with the payload, keep
// the tag. The ETag alone is not quite enough for it, either. For a
// single-part upload the ETag is the content MD5, and chosen-prefix MD5
// collisions are practical on ordinary hardware — so an attacker who controls
// both uploads could produce a payload with the ETag of the clean object it
// replaces. Last-modified moves on every overwrite and is not theirs to
// choose. A tag whose recorded version does not match reads as unknown, and
// the sweep scans it again.
func (c *clamavScanner) Verdict(ctx context.Context, bucket, key, version string) (Verdict, error) {
	verdict, taggedVersion, at, stale, err := c.readSignedTag(ctx, bucket, key)
	if err != nil {
		return VerdictUnknown, err
	}
	if verdict == VerdictDenied && taggedVersion == "" {
		// The tag set itself was unreadable. A fact about the object, not a
		// transport failure, and it must not fail the pass.
		return VerdictDenied, nil
	}
	out := c.interpretForSweep(verdict, taggedVersion, version)
	if stale && out == verdict && verdict != "" {
		// A verdict issued under the previous key, still valid, still about
		// these bytes. Re-sign it so the old key can eventually be dropped —
		// WITHOUT rescanning, and WITHOUT moving carlos-scan-at, because a
		// re-sign judges nothing and must not claim a fresh scan.
		c.resign(ctx, bucket, key, taggedVersion, verdict, at)
	}
	return out, nil
}

// resign rewrites the signature of an existing verdict under the current key.
//
// Best effort, deliberately. The verdict is already valid under the previous
// key, so the cost of failing is that it is re-signed on some later tick;
// returning an error here would abandon the whole page over a housekeeping
// write and stall every object behind this one.
func (c *clamavScanner) resign(ctx context.Context, bucket, key, version string, v Verdict, at string) {
	if _, err := c.tagWith(ctx, bucket, key, version, v, at); err != nil {
		log.Printf("objstore: could not re-sign %s/%s under the current tag key (it stays valid under the previous one): %v", bucket, key, err)
	}
}

// readTag is the one place tags are read, shared by Verdict and Status so the
// two cannot drift on what a tag SAYS while differing, deliberately, on what
// it MEANS.
//
// A denied read comes back as (VerdictDenied, "", nil): unreadable tags are a
// fact about the object.
func (c *clamavScanner) readTag(ctx context.Context, bucket, key string) (Verdict, string, error) {
	v, ver, _, _, err := c.readSignedTag(ctx, bucket, key)
	return v, ver, err
}

// readSignedTag is readTag with the two extra facts the re-signing path needs:
// the recorded timestamp, and whether the signature that vouched for all of it
// was issued under the PREVIOUS key.
//
// THIS IS THE TRUST BOUNDARY. Everything above it — Verdict, Status, the
// sweep, the drain, the status door — is reading a claim that has already been
// checked against the deployment's key. An unsigned or badly signed tag does
// not come back as a suspicious verdict to be handled carefully somewhere
// else; it comes back as no verdict at all, which is the same answer as a bare
// object and produces the same behaviour: scan it.
func (c *clamavScanner) readSignedTag(ctx context.Context, bucket, key string) (verdict Verdict, version, at string, stale bool, err error) {
	out, err := c.s3.GetObjectTagging(ctx, &awss3.GetObjectTaggingInput{
		Bucket: &bucket,
		Key:    &key,
	})
	if err != nil {
		if isGone(err) {
			return VerdictUnknown, "", "", false, ErrObjectGone
		}
		if isDenied(err) {
			return VerdictDenied, "", "", false, nil
		}
		return VerdictUnknown, "", "", false, fmt.Errorf("objstore: read tags %s/%s: %w", bucket, key, err)
	}

	var raw, sig string
	for _, t := range out.TagSet {
		switch aws2s(t.Key) {
		case TagVerdict:
			raw = aws2s(t.Value)
		case TagVersion:
			version = aws2s(t.Value)
		case TagAt:
			at = aws2s(t.Value)
		case TagSignature:
			sig = aws2s(t.Value)
		}
	}
	if raw == "" {
		return "", "", "", false, nil
	}
	ok, stale := c.signer.Verify(scanFacts{
		bucket: bucket, key: key, version: version, verdict: raw, at: at,
	}, sig)
	if !ok {
		// Somebody else wrote this. On several gateways an app CAN, whatever
		// the IAM policy says, so this is the expected shape of an attack and
		// not an anomaly to shout about. Report exactly what a bare object
		// reports.
		return "", "", "", false, nil
	}
	return Verdict(raw), version, at, stale, nil
}

// interpretForSweep applies the SWEEP's reading of a tag: unknown when it does
// not match the current version, and unknown for denied and failed so they are
// re-issued rather than inherited.
func (c *clamavScanner) interpretForSweep(verdict Verdict, taggedVersion, version string) Verdict {
	if verdict == "" || taggedVersion == "" {
		return VerdictUnknown
	}
	// denied and failed are RE-ISSUED every sweep, never carried forward.
	//
	// Both are transient by nature — a permissions fix, a Glacier restore, an
	// engine that was having a bad day — and the tag is an audit trace of what
	// happened last time, not a verdict to inherit. Reading them back as
	// answers would make them permanent: the object is not rewritten, so its
	// version never changes, so it would never be retried and the repair would
	// never take. An SSE-KMS denial is the case that bites, because the
	// tagging call needs no KMS access and therefore succeeds.
	switch verdict {
	case VerdictDenied, VerdictFailed:
		return VerdictUnknown
	}
	if taggedVersion != version {
		// The bytes changed under the verdict. Not a failure — just unknown
		// again, which is exactly what makes the next sweep re-scan it.
		return VerdictUnknown
	}
	return verdict
}

// Status answers about the CURRENT version of one object, finding that version
// itself, and reports what the tag SAYS rather than what the sweep should do
// about it. See the Scanner interface for why those are different questions.
func (c *clamavScanner) Status(ctx context.Context, bucket, key string) (Verdict, ScanObject, error) {
	head, err := c.s3.HeadObject(ctx, &awss3.HeadObjectInput{Bucket: &bucket, Key: &key})
	if err != nil {
		if isGone(err) {
			return VerdictUnknown, ScanObject{}, ErrObjectGone
		}
		if isDenied(err) {
			// No version and no size, so a caller must not go on to Scan.
			return VerdictDenied, ScanObject{}, nil
		}
		return VerdictUnknown, ScanObject{}, fmt.Errorf("objstore: head %s/%s: %w", bucket, key, err)
	}
	obj := ScanObject{Key: key, ETag: normalizeETag(aws2s(head.ETag))}
	if head.LastModified != nil {
		obj.Modified = *head.LastModified
	}
	if head.ContentLength != nil {
		obj.Size = *head.ContentLength
	}

	verdict, taggedVersion, err := c.readTag(ctx, bucket, key)
	if err != nil {
		return VerdictUnknown, obj, err
	}
	if verdict == VerdictDenied && taggedVersion == "" {
		return VerdictDenied, obj, nil
	}
	if verdict == "" || taggedVersion == "" || taggedVersion != obj.Version() {
		// No tag, or a tag about bytes that have since changed. Either way
		// nothing is known about what is there now.
		return VerdictUnknown, obj, nil
	}
	// Reported as written: denied and failed included. The re-issue rule is
	// the sweep's, and applying it here would make a failed scan read as
	// pending forever.
	return verdict, obj, nil
}

// Scan streams one object past the engine and records the result.
func (c *clamavScanner) Scan(ctx context.Context, bucket, key, version string, size int64) (Verdict, bool, error) {
	// The ceiling is checked BEFORE the fetch. The size is already on the
	// listing page that got us here, so paying egress to pull bytes clamd has
	// told us it will refuse is pure waste — and on a bucket full of large
	// media that waste is the entire bill.
	if c.maxBytes > 0 && size > c.maxBytes {
		recorded, err := c.tag(ctx, bucket, key, version, VerdictSkipped)
		if err != nil {
			return VerdictUnknown, false, err
		}
		return VerdictSkipped, recorded, nil
	}

	obj, err := c.s3.GetObject(ctx, &awss3.GetObjectInput{Bucket: &bucket, Key: &key})
	if err != nil {
		// An object we are not allowed to read is DENIED — a permanent fact
		// about this object that a permissions fix repairs, not a retry.
		//
		// This classification is load-bearing, not tidiness. Returning a plain
		// error here fails the whole page, unwritten; the next tick re-lists
		// the same page, meets the same object and fails identically, and the
		// bucket's sweep never advances past it. One SSE-KMS object the
		// console's key policy does not cover, or one object owned by another
		// account in an adopted bucket, would stall that store's coverage
		// forever while every surface reported a sweep merely "in progress".
		if isDenied(err) {
			// Tagging is best effort and its failure is deliberately ignored:
			// if we may not read the object we may well not be able to tag it,
			// and the verdict still stands for this sweep's count. The tag is
			// only an audit trace here — Verdict re-issues denied rather than
			// reading it back, so nothing depends on this landing.
			_, _ = c.tag(ctx, bucket, key, version, VerdictDenied)
			// recorded is false whether or not the tag landed: nothing may
			// depend on a trace we could not verify.
			return VerdictDenied, false, nil
		}
		// Deleted between the listing and the fetch. Routine on a bucket
		// anyone is using, and not a failure of anything — the sweep drops it
		// rather than counting an object that no longer exists.
		if isGone(err) {
			return VerdictUnknown, false, ErrObjectGone
		}
		return VerdictUnknown, false, fmt.Errorf("objstore: get %s/%s: %w", bucket, key, err)
	}
	defer obj.Body.Close()

	// Straight from S3 into the socket. The object is never buffered: a
	// gigabyte object costs a chunk of memory, not a gigabyte, and the console
	// holds no copy of tenant data at any point.
	res, err := c.engine.ScanStream(ctx, obj.Body)
	if err != nil {
		// The engine could not be reached or did not answer. That is a fact
		// about the ENGINE, not about this object, and it must not be written
		// as a per-object verdict: a clamd that is down for an hour would
		// otherwise leave thousands of objects permanently tagged failed,
		// with nothing to distinguish them from objects that genuinely broke
		// the scanner. Fail the pass instead; the object stays unknown and
		// the next sweep picks it up.
		return VerdictUnknown, false, fmt.Errorf("objstore: scan %s/%s: %w", bucket, key, err)
	}

	v := verdictFor(res.Status)
	recorded, err := c.tag(ctx, bucket, key, version, v)
	if err != nil {
		return VerdictUnknown, false, err
	}
	return v, recorded, nil
}

// verdictFor maps an engine status onto the recorded verdict.
//
// StatusSizeExceeded becomes VerdictSkipped, not VerdictFailed. Nothing
// failed: the object was simply never judged, and an audit that cannot tell
// "we looked and it was clean" from "we never looked" is not evidence.
func verdictFor(s clamd.Status) Verdict {
	switch s {
	case clamd.StatusClean:
		return VerdictClean
	case clamd.StatusInfected:
		return VerdictInfected
	case clamd.StatusSizeExceeded:
		return VerdictSkipped
	default:
		return VerdictFailed
	}
}

// tag writes the verdict, the ETag it applies to, and when it was reached.
//
// PutObjectTagging REPLACES the whole tag set, so anything already on the
// object is preserved explicitly here rather than assumed absent.
//
// An earlier version of this reasoned that nothing could be there to lose: the
// store user has no s3:PutObjectTagging, and AWS requires that same permission
// to set tags via x-amz-tagging on upload, so an app cannot tag its own object
// by either route. That holds for a bucket CARLOS minted. It does not hold for
// an ADOPTED one — `store grant --bucket` takes over a bucket whose objects
// may carry tags written long before CARLOS scoped any credentials, by tooling
// that is still using them for lifecycle rules or cost allocation. Replacing
// the set blindly would delete them, silently, on first scan.
func (c *clamavScanner) tag(ctx context.Context, bucket, key, version string, v Verdict) (recorded bool, err error) {
	// The timestamp is minted HERE, once, and threaded through — it goes in a
	// tag and into the signature over it, and two calls to time.Now() would
	// produce a signature over a moment the tag does not record.
	return c.tagWith(ctx, bucket, key, version, v, time.Now().UTC().Format(time.RFC3339))
}

// tagWith writes a verdict stamped with a GIVEN time, so that re-signing an
// existing verdict after a key rotation can keep the moment the bytes were
// actually judged. tag() supplies now; resign() supplies what was already
// there.
func (c *clamavScanner) tagWith(ctx context.Context, bucket, key, version string, v Verdict, at string) (recorded bool, err error) {
	scanTagSet := []s3types.Tag{
		{Key: s2aws(TagVerdict), Value: s2aws(string(v))},
		{Key: s2aws(TagVersion), Value: s2aws(version)},
		{Key: s2aws(TagAt), Value: s2aws(at)},
		// The one that makes the other three mean something.
		{Key: s2aws(TagSignature), Value: s2aws(c.signer.Sign(scanFacts{
			bucket: bucket, key: key, version: version, verdict: string(v), at: at,
		}))},
	}
	// OURS FIRST, so that truncating to the limit below drops foreign tags
	// and never one of the three this whole mechanism exists to write.
	set := append([]s3types.Tag(nil), scanTagSet...)
	// Read-merge-write. One extra call per object per LIFETIME, not per sweep:
	// an object whose verdict still matches its version never reaches Scan
	// again.
	existing, err := c.s3.GetObjectTagging(ctx, &awss3.GetObjectTaggingInput{
		Bucket: &bucket,
		Key:    &key,
	})
	if err != nil { //nolint:nestif // the three outcomes are the point
		// A tag set we cannot read must NOT be written over. PutObjectTagging
		// replaces the whole set, so proceeding here would destroy exactly the
		// foreign tags this merge exists to protect — silently, and on the one
		// path where we know least about what is there. The sweep read this
		// same object's tags moments ago through Verdict, so a failure now is
		// transient: fail the page and let the next tick retry.
		if isDenied(err) || isGone(err) {
			// Nothing we could ever write, and not a pass failure. NOT
			// recorded, which is what stops a caller believing the object
			// carries a verdict it does not.
			return false, nil
		}
		return false, fmt.Errorf("objstore: read tags before writing %s/%s: %w", bucket, key, err)
	}
	for _, t := range existing.TagSet {
		if !isScanTag(aws2s(t.Key)) {
			set = append(set, t)
		}
	}

	// S3 caps an object at ten tags, and an object already carrying eight
	// foreign ones cannot also hold our three. What happens next depends on
	// the verdict, and the two cases are genuinely different.
	//
	// CLEAN (or failed, or skipped): yield. Erroring would abandon the page
	// unwritten every single tick — the exact permanent stall the denied
	// classification was added to remove — and the only cost of not recording
	// is that the object is scanned again next time. A tenant's tags, which on
	// an ADOPTED bucket may predate CARLOS entirely, are worth more than one
	// avoided re-scan.
	//
	// INFECTED: do not yield. `scan enforce on` writes a Deny keyed on
	// `s3:ExistingObjectTag/carlos-scan = infected` (aws.go), so the Deny only
	// ever engages if this tag is WRITTEN. Yielding here means an app that
	// attaches eight tags of its own to an upload makes its objects
	// permanently unmarkable — verdict reached, write skipped, Deny never
	// matches, malware readable through the app's own credential, and nothing
	// anywhere alarms: the sweep counts the object as reached, the result
	// object says recorded:false, and the status door reads no tag and answers
	// unknown.
	//
	// So an infected verdict evicts foreign tags to fit. That is a real loss
	// of a tenant's metadata and it is the right trade against a known-
	// infected object that enforcement cannot touch. It takes the minimum the
	// limit forces — our three go in, as many foreign tags as still fit are
	// kept — because taking more than necessary would be gratuitous.
	if len(set) > maxObjectTags {
		if v != VerdictInfected {
			return false, nil
		}
		log.Printf("objstore: %s/%s is infected and carries %d foreign tags — evicting %d of them so the verdict can be recorded and enforcement can see it",
			bucket, key, len(set)-len(scanTagSet), len(set)-maxObjectTags)
		// set[:maxObjectTags] keeps OUR tags, which are first, and as many
		// foreign ones as still fit. The room left for a tenant is
		// maxObjectTags-len(scanTagSet) — derived, because it changed from
		// three to four the moment the signature tag landed and a hardcoded
		// number would have quietly started writing an eleven-tag set.
		set = set[:maxObjectTags]
	}

	if _, err := c.s3.PutObjectTagging(ctx, &awss3.PutObjectTaggingInput{
		Bucket:  &bucket,
		Key:     &key,
		Tagging: &s3types.Tagging{TagSet: set},
	}); err != nil {
		return false, fmt.Errorf("objstore: tag %s/%s: %w", bucket, key, err)
	}
	return true, nil
}

// maxObjectTags is S3's per-object tag limit.
const maxObjectTags = 10

// isScanTag is the set this scanner owns and may overwrite. Everything else on
// the object belongs to whoever put it there.
func isScanTag(key string) bool {
	return key == TagVerdict || key == TagVersion || key == TagAt || key == TagSignature
}

// normalizeETag strips the quotes S3 wraps an ETag in. The listing and the tag
// must compare equal or every object re-scans forever, and only one of the two
// arrives quoted.
func normalizeETag(etag string) string {
	return strings.Trim(strings.TrimSpace(etag), `"`)
}

func aws2s(p *string) string {
	if p == nil {
		return ""
	}
	return *p
}

func s2aws(s string) *string { return &s }

// ErrObjectGone is an object that vanished between the listing page and the
// fetch. The sweep drops it rather than counting it: it is neither covered nor
// a gap, it simply is not there any more.
var ErrObjectGone = errors.New("objstore: object no longer exists")

// deniedCodes are the S3 and KMS answers that mean "not allowed, and a retry
// will not change that". Matched on the API error code rather than the message
// so a wording change upstream does not silently reclassify them as failures.
var deniedCodes = map[string]bool{
	"AccessDenied":              true,
	"AccessDeniedException":     true, // KMS, on an SSE-KMS object
	"Forbidden":                 true,
	"AllAccessDisabled":         true,
	"KMS.AccessDeniedException": true,
	"InvalidObjectState":        true, // GLACIER / DEEP_ARCHIVE: unreadable until restored
}

func isDenied(err error) bool {
	var ae smithy.APIError
	if errors.As(err, &ae) {
		return deniedCodes[ae.ErrorCode()]
	}
	return false
}

// isGone is an object that is not there — deleted between the listing and the
// fetch, which happens routinely on a bucket in use.
func isGone(err error) bool {
	var nsk *s3types.NoSuchKey
	if errors.As(err, &nsk) {
		return true
	}
	var ae smithy.APIError
	if errors.As(err, &ae) {
		code := ae.ErrorCode()
		return code == "NoSuchKey" || code == "NotFound"
	}
	return false
}