package objstore

import (
	"context"
	"sync"
	"time"
)

// MemScanner is the in-memory Scanner every walk test runs against, mirroring
// MemProvisioner: no network, no clamd, and a failure hook so a test can make
// any single method fail once and prove the walk survives it.
type MemScanner struct {
	// onStatus is OnStatus's hook; see its comment.
	onStatus func()
	mu       sync.Mutex
	enabled  map[string]bool
	verdicts map[string]scanRecord
	failures map[string]error
	scanAs   map[string]Verdict
	// objects is what a listing would have said about a key. Status needs it:
	// a caller holding only a key has no listing page to take a version from.
	objects map[string]ScanObject
	// unrecordable are keys whose verdict cannot be persisted — the ten-tag
	// cap, in production. A hook rather than a stub, because the case is the
	// reason Scan reports `recorded` at all.
	unrecordable map[string]bool
	// deniedHead are keys whose HeadObject is refused, so Status answers
	// denied with a ZERO ScanObject — the one shape that leaves a caller with
	// no version and no size.
	deniedHead map[string]bool
	// scans counts Scan calls per key, so a test can assert that deduping
	// really stopped the engine reading the same object twice rather than
	// merely folding two outcomes into one.
	scans map[string]int
}

// scanRecord is one object's verdict AND the object version it was judged
// against. The version is not decoration: Verdict answers Unknown when it
// disagrees, which is what stops a rewritten object inheriting the old bytes'
// verdict.
type scanRecord struct {
	version string
	verdict Verdict
}

func NewMemScanner() *MemScanner {
	return &MemScanner{
		enabled:  map[string]bool{},
		verdicts: map[string]scanRecord{},
		failures: map[string]error{},
	}
}

func memScanKey(bucket, key string) string { return bucket + "\x00" + key }

// Fail arms a one-shot failure for the named method, the same shape
// MemProvisioner.takeFailure gives the provisioner walks.
func (s *MemScanner) Fail(method string, err error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.failures[method] = err
}

// PutObject seeds what a listing would have said about a key, so Status can
// answer. Without it Status cannot tell an absent object from an unscanned
// one, and those are different answers.
func (s *MemScanner) PutObject(bucket, key, etag string, modified time.Time, size int64) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.objects == nil {
		s.objects = map[string]ScanObject{}
	}
	s.objects[memScanKey(bucket, key)] = ScanObject{
		Key: key, ETag: etag, Modified: modified, Size: size,
	}
}

// Unrecordable makes Scan reach a verdict it cannot persist, the way the
// ten-tag cap does in production.
func (s *MemScanner) Unrecordable(key string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.unrecordable == nil {
		s.unrecordable = map[string]bool{}
	}
	s.unrecordable[key] = true
}

// OnStatus installs a hook that runs at the top of Status, BEFORE any lock is
// taken, so a test can make something else happen while a caller is mid-read.
//
// It exists for one assertion that has no other seam: the status door
// registers its long-poll waiter before its first read, and the only way to
// catch the opposite order is to run a drain inside that read. The hook must
// not re-enter the scanner (Status will be called again beneath it), so a test
// uses sync.Once.
func (s *MemScanner) OnStatus(f func()) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.onStatus = f
}

// Status answers about the CURRENT version, and reports what the record SAYS
// — denied and failed included. The re-issue rule below is Verdict's, and
// applying it here would make a failed scan read as pending forever.
func (s *MemScanner) Status(ctx context.Context, bucket, key string) (Verdict, ScanObject, error) {
	// The hook runs unlocked, or anything it does that touches this scanner
	// deadlocks against the lock taken below.
	s.mu.Lock()
	hook := s.onStatus
	s.mu.Unlock()
	if hook != nil {
		hook()
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if err := s.takeFailure("Status"); err != nil {
		return VerdictUnknown, ScanObject{}, err
	}
	if s.deniedHead[memScanKey(bucket, key)] {
		// HeadObject itself refused: no version, no size, nothing to scan
		// with. The production shape is a bucket policy the console is not
		// covered by, as against the SSE-KMS case where the head succeeds and
		// the GET is what fails.
		return VerdictDenied, ScanObject{}, nil
	}
	obj, ok := s.objects[memScanKey(bucket, key)]
	if !ok {
		return VerdictUnknown, ScanObject{}, ErrObjectGone
	}
	rec, ok := s.verdicts[memScanKey(bucket, key)]
	if !ok || rec.version != obj.Version() {
		return VerdictUnknown, obj, nil
	}
	return rec.verdict, obj, nil
}

// Put seeds a verdict so a test can start from a partly-scanned bucket.
func (s *MemScanner) Put(bucket, key, version string, v Verdict) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.verdicts[memScanKey(bucket, key)] = scanRecord{version: version, verdict: v}
}

// IsEnabled lets Teardown's test prove the scanner was disabled while the
// record that named it still existed.
func (s *MemScanner) IsEnabled(bucket string) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.enabled[bucket]
}

func (s *MemScanner) takeFailure(method string) error {
	if err, ok := s.failures[method]; ok {
		delete(s.failures, method)
		return err
	}
	return nil
}

func (s *MemScanner) EnsureEnabled(ctx context.Context, bucket string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if err := s.takeFailure("EnsureEnabled"); err != nil {
		return err
	}
	s.enabled[bucket] = true // already enabled is a success
	return nil
}

func (s *MemScanner) EnsureDisabled(ctx context.Context, bucket string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if err := s.takeFailure("EnsureDisabled"); err != nil {
		return err
	}
	delete(s.enabled, bucket) // already gone is a success
	return nil
}

func (s *MemScanner) Verdict(ctx context.Context, bucket, key, version string) (Verdict, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if err := s.takeFailure("Verdict"); err != nil {
		return VerdictUnknown, err
	}
	rec, ok := s.verdicts[memScanKey(bucket, key)]
	if !ok || rec.version != version {
		return VerdictUnknown, nil
	}
	// The SWEEP's reading: denied and failed are re-issued rather than
	// inherited, because both are transient and the object is not rewritten,
	// so its version never changes and the repair would never take. Status
	// deliberately does not do this — see the Scanner interface.
	switch rec.verdict {
	case VerdictDenied, VerdictFailed:
		return VerdictUnknown, nil
	}
	return rec.verdict, nil
}

func (s *MemScanner) Scan(ctx context.Context, bucket, key, version string, size int64) (Verdict, bool, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if err := s.takeFailure("Scan"); err != nil {
		return VerdictUnknown, false, err
	}
	if s.scans == nil {
		s.scans = map[string]int{}
	}
	s.scans[key]++
	v := VerdictClean
	if got, ok := s.scanAs[key]; ok {
		v = got
	}
	if s.unrecordable[key] {
		// A verdict reached and not persisted: the engine judged the object,
		// but nothing on the object says so.
		return v, false, nil
	}
	s.verdicts[memScanKey(bucket, key)] = scanRecord{version: version, verdict: v}
	return v, true, nil
}

// DenyHead makes Status refuse the head of one key.
func (s *MemScanner) DenyHead(bucket, key string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.deniedHead == nil {
		s.deniedHead = map[string]bool{}
	}
	s.deniedHead[memScanKey(bucket, key)] = true
}

// Scans is how many times the engine was asked to read one key.
func (s *MemScanner) Scans(key string) int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.scans[key]
}

// ScanAs makes Scan answer v for key rather than clean, so a test can stage a
// skipped or infected object without a real engine. Skipped in particular is
// not a hypothetical: it is every object over the size ceiling.
func (s *MemScanner) ScanAs(key string, v Verdict) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.scanAs == nil {
		s.scanAs = map[string]Verdict{}
	}
	s.scanAs[key] = v
}

func (s *MemScanner) Enforcement() DenyTag {
	return DenyTag{Key: TagVerdict, Value: string(VerdictInfected)}
}

func (s *MemScanner) Capacity(ctx context.Context) (int, int, error) { return -1, -1, nil }

// Provenance answers "clamav" rather than "memscanner" ON PURPOSE. The sweep
// refuses when the record's engine disagrees with the running one (spec §7.4),
// and test fixtures write Engine: "clamav" — a fake that named itself would
// make every one of those fixtures refuse instead of sweep.
func (s *MemScanner) Provenance(ctx context.Context) (string, string, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	// Armable like the others: an engine that will not say what it is, is a
	// state the drain and the sweep both have to survive, and a fake that
	// could never be in it left that path untested.
	if err := s.takeFailure("Provenance"); err != nil {
		return "", "", err
	}
	return "clamav", "MemScanner/0/never", nil
}

var _ Scanner = (*MemScanner)(nil)