package console

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

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

// geoipProbeAddrs is the country probe (spec §4.1 step 4): at least three
// must answer with a country. A variable so tests can use addresses the
// MaxMind fixtures hold (FACTS.md).
var geoipProbeAddrs = []string{"8.8.8.8", "1.1.1.1", "9.9.9.9", "208.67.222.222", "2001:4860:4860::8888"}

const geoipProbeNeed = 3

var (
	errGeoipBusy    = errors.New("a geoip publish is already in progress")
	errGeoipRefused = errors.New("geoip database refused")
)

func refused(category string) error { return fmt.Errorf("%w: %s", errGeoipRefused, category) }

// geoipPublisher is the console's single writer of control/geoip/ (spec §3).
// mu is held through cleanup by every publish, off, sweep and expiry, and a
// publish that finds it held answers errGeoipBusy rather than queueing.
type geoipPublisher struct {
	mu       sync.Mutex
	store    release.Store
	tmpRoot  string
	maxBytes int64
	deadline time.Duration
	now      func() time.Time
	// expired is D1's rule, a field so a test can apply it to the GeoIP2
	// fixtures (which never expire) without a GeoLite2 file in the repo.
	expired func(dbType string, build, now time.Time) bool
	// storeTimeout bounds a single small bucket read made while holding,
	// or about to hold, the publish slot.
	storeTimeout time.Duration
}

type geoipPublishOpts struct {
	Origin, Edition string
	Force           bool
	// RequireType refuses a database of any other type (the feed's check,
	// spec §10.3, including on the first publish). Monotonic refuses a
	// build that is not newer than the current pointer's when that pointer
	// came from the same Origin. The feed sets both and never Force.
	RequireType string
	Monotonic   bool
}

// newGeoipPublisher clears tmpRoot (the dedicated geoip-staging directory)
// before anything can stage into it: a console that crashed mid-upload or
// mid-job must not leave its 1 GiB behind for good (spec §4.2, codex R6-2).
func newGeoipPublisher(store release.Store, tmpRoot string) *geoipPublisher {
	if tmpRoot != "" {
		if err := os.RemoveAll(tmpRoot); err != nil {
			log.Printf("console: geoip staging directory could not be cleared: %v", err)
		}
		if err := os.MkdirAll(tmpRoot, 0o700); err != nil {
			log.Printf("console: geoip staging directory could not be created: %v", err)
		}
	}
	return &geoipPublisher{store: store, tmpRoot: tmpRoot, maxBytes: geoip.MaxBlobBytes,
		deadline: 15 * time.Minute, now: time.Now, expired: geoip.Expired, storeTimeout: 30 * time.Second}
}

func marshalPointer(p geoip.Pointer) ([]byte, error) { return json.MarshalIndent(p, "", "  ") }

// Current reads the pointer. present=false means no pointer (feature off).
func (p *geoipPublisher) Current(ctx context.Context) (geoip.Pointer, string, bool, error) {
	raw, etag, err := readPointerCapped(ctx, p.store)
	if errors.Is(err, os.ErrNotExist) {
		return geoip.Pointer{}, "", false, nil
	}
	if errors.Is(err, errPointerTooLarge) {
		// Present but not a pointer anyone wrote: replaceable, never decoded.
		return geoip.Pointer{}, etag, true, err
	}
	if err != nil {
		return geoip.Pointer{}, "", false, err
	}
	ptr, err := geoip.ParsePointer(raw)
	if err != nil {
		return geoip.Pointer{}, etag, true, err
	}
	return ptr, etag, true, nil
}

// geoipStaged is an upload on disk awaiting Commit. Whoever holds one owns
// the publish slot (p.mu) and the temporary directory: Commit or Abandon
// releases both (spec §4.2, codex R5-1).
type geoipStaged struct {
	dir, path, sha string
	size           int64
	// The pointer as it stood when the slot was taken (spec §3 step 1):
	// every comparison is against this read, and the pointer write is
	// conditional on its etag, so a foreign write during the upload is
	// refused rather than overwritten.
	cur        geoip.Pointer
	curEtag    string
	curPresent bool
	curErr     error
}

// Stage reserves the one publish slot BEFORE reading any of src, then
// streams src to a temporary file, hashing and capping as it copies. On
// success the slot stays held for Commit; on failure it is released.
func (p *geoipPublisher) Stage(ctx context.Context, src io.Reader) (*geoipStaged, error) {
	if !p.mu.TryLock() {
		return nil, errGeoipBusy
	}
	dir, err := os.MkdirTemp(p.tmpRoot, "geoip-publish-*")
	if err != nil {
		p.mu.Unlock()
		return nil, err
	}
	st := &geoipStaged{dir: dir, path: filepath.Join(dir, "db.mmdb")}
	// The pointer read is bounded on its own: the caller's context may be
	// an HTTP request whose only deadline is on reading its body (code
	// review R4-1).
	rctx, rcancel := context.WithTimeout(ctx, p.storeTimeout)
	st.cur, st.curEtag, st.curPresent, st.curErr = p.Current(rctx)
	rcancel()
	if st.curErr != nil && !st.curPresent {
		p.Abandon(st)
		return nil, st.curErr // the store itself failed
	}
	st.sha, st.size, err = copyCapped(ctx, st.path, src, p.maxBytes)
	if err != nil {
		p.Abandon(st)
		return nil, err
	}
	return st, nil
}

// Abandon releases a staged upload that will not be committed.
func (p *geoipPublisher) Abandon(st *geoipStaged) {
	os.RemoveAll(st.dir)
	p.mu.Unlock()
}

var errPointerTooLarge = errors.New("the object at the geoip pointer key is larger than a pointer can be")

// readPointerCapped reads control/geoip/current.json streamed and capped at
// geoip.MaxPointerBytes, so an oversized object there can never be pulled
// into the console's memory (code review R1-3). The etag is the store's;
// DirStore cannot report one from Open, and its etag is defined as the
// content's sha256 (release.DirStore, etagOf), so it is computed here from
// the bytes, which are complete whenever they are under the cap.
func readPointerCapped(ctx context.Context, s release.Store) ([]byte, string, error) {
	return readCappedKey(ctx, s, geoip.PointerKey, geoip.MaxPointerBytes)
}

// readCappedKey is readPointerCapped for any small control object.
func readCappedKey(ctx context.Context, s release.Store, key string, max int) ([]byte, string, error) {
	rc, etag, err := release.OpenObject(ctx, s, key)
	if err != nil {
		return nil, "", err
	}
	defer rc.Close()
	b, err := io.ReadAll(io.LimitReader(rc, int64(max)+1))
	if err != nil {
		return nil, "", err
	}
	if len(b) > max {
		if etag == "" {
			// Without the store's etag a conditional replace cannot be
			// aimed at it; only off (unconditional delete) clears it.
			return nil, "", errPointerTooLarge
		}
		return nil, etag, errPointerTooLarge
	}
	if etag == "" {
		sum := sha256.Sum256(b)
		etag = hex.EncodeToString(sum[:])
	}
	return b, etag, nil
}

// Publish is Stage then Commit, under one deadline.
func (p *geoipPublisher) Publish(ctx context.Context, src io.Reader, o geoipPublishOpts) (geoip.Pointer, error) {
	ctx, cancel := context.WithTimeout(ctx, p.deadline)
	defer cancel()
	st, err := p.Stage(ctx, src)
	if err != nil {
		return geoip.Pointer{}, err
	}
	return p.Commit(ctx, st, o)
}

// Commit is spec §4.1 steps 3-6 on a staged upload. It always removes the
// staged directory and releases the slot.
func (p *geoipPublisher) Commit(ctx context.Context, st *geoipStaged, o geoipPublishOpts) (geoip.Pointer, error) {
	defer p.Abandon(st)
	if _, ok := ctx.Deadline(); !ok {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, p.deadline)
		defer cancel()
	}
	// The file is untrusted: it is parsed only in a child process (code
	// review R9), never here.
	db, err := verifyInChild(ctx, st.path, geoipProbeAddrs)
	switch {
	case ctx.Err() != nil:
		return geoip.Pointer{}, ctx.Err()
	case errors.Is(err, errGeoipVerifyTimeout):
		return geoip.Pointer{}, refused("the database took too long to verify")
	case errors.Is(err, errGeoipVerifyTree):
		return geoip.Pointer{}, refused("the database's search tree is malformed")
	case err != nil:
		return geoip.Pointer{}, refused("not a valid MaxMind-format database")
	}
	if db.Countries < geoipProbeNeed {
		return geoip.Pointer{}, refused("the database has no country data")
	}
	dbBuild := time.Unix(db.BuildEpoch, 0).UTC()
	// D1 at the door (codex R7-1): an already-expired GeoLite2 would replace
	// a usable database with one every edge must reject. force never
	// overrides it; the check runs again just before the pointer write.
	if p.expired(db.Source, dbBuild, p.now()) {
		return geoip.Pointer{}, refused("the database is already expired (GeoLite2, 30 days after build)")
	}
	if o.RequireType != "" && db.Source != o.RequireType {
		return geoip.Pointer{}, refused("the feed returned a " + db.Source + " database, not " + o.RequireType)
	}
	cur, etag, present, curErr := st.cur, st.curEtag, st.curPresent, st.curErr
	if o.Monotonic && present && curErr == nil && cur.Origin == o.Origin && !dbBuild.After(cur.Build()) {
		return geoip.Pointer{}, refused("not newer than the published edition")
	}
	switch {
	case curErr != nil:
		// A pointer exists but does not parse: replace it (etag still guards).
		log.Printf("console: geoip publish: the current pointer is unreadable and will be replaced")
	case present && !o.Force && cur.DatabaseType == db.Source && db.NodeCount < cur.NodeCount/2:
		return geoip.Pointer{}, refused("fewer than half the current database's nodes (use -force if intended)")
	}
	next := geoip.Pointer{V: 1, SHA256: st.sha, Key: geoip.BlobKey(st.sha), Size: st.size,
		DatabaseType: db.Source, BuildEpoch: db.BuildEpoch, NodeCount: db.NodeCount,
		Published: p.now().UTC().Truncate(time.Second), Origin: o.Origin, Edition: o.Edition}
	f, err := os.Open(st.path)
	if err != nil {
		return geoip.Pointer{}, err
	}
	defer f.Close()
	// A CAS conflict on the blob means an object already sits at this
	// content-addressed name. It is only the same bytes if it hashes to the
	// name: a truncated or foreign object there must never be published
	// (spec §3 step 2, plan review R1-6).
	if _, err := release.PutObjectFrom(ctx, p.store, next.Key, f, st.size, ""); errors.Is(err, release.ErrCASConflict) {
		if err := p.verifyExistingBlob(ctx, next.Key, st.sha, st.size); err != nil {
			return geoip.Pointer{}, err
		}
	} else if err != nil {
		return geoip.Pointer{}, err
	}
	body, err := marshalPointer(next)
	if err != nil {
		return geoip.Pointer{}, err
	}
	if len(body) > geoip.MaxPointerBytes {
		// Every reader caps the pointer at this size: one larger would
		// publish a database no edge can ever read (code review R7-1).
		return geoip.Pointer{}, refused("the pointer would be larger than readers accept")
	}
	if !present {
		etag = ""
	}
	if p.expired(next.DatabaseType, next.Build(), p.now()) {
		return geoip.Pointer{}, refused("the database expired while it was being published (GeoLite2, 30 days after build)")
	}
	if _, err := p.store.Put(ctx, geoip.PointerKey, body, etag); err != nil {
		return geoip.Pointer{}, fmt.Errorf("pointer write: %w", err)
	}
	if err := p.sweepLocked(ctx, next.Key); err != nil {
		log.Printf("console: geoip sweep after publish: %v", err)
	}
	log.Printf("console: geoip published %s (%s, %s)", st.sha[:8], next.DatabaseType, next.Origin)
	return next, nil
}

// copyCapped streams src into path, hashing, refusing more than max bytes
// as it copies (codex R1-7).
func copyCapped(ctx context.Context, path string, src io.Reader, max int64) (string, int64, error) {
	f, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
	if err != nil {
		return "", 0, err
	}
	defer f.Close()
	h := sha256.New()
	n, err := io.Copy(io.MultiWriter(f, h), io.LimitReader(geoipCtxReader{ctx, src}, max+1))
	if err != nil {
		return "", 0, err
	}
	if n > max {
		return "", 0, refused(fmt.Sprintf("larger than %d bytes", max))
	}
	if n == 0 {
		return "", 0, refused("empty upload")
	}
	if err := f.Sync(); err != nil {
		return "", 0, err
	}
	return hex.EncodeToString(h.Sum(nil)), n, nil
}

type geoipCtxReader struct {
	ctx context.Context
	r   io.Reader
}

func (c geoipCtxReader) Read(b []byte) (int, error) {
	if err := c.ctx.Err(); err != nil {
		return 0, err
	}
	return c.r.Read(b)
}

// verifyExistingBlob streams the object at key and checks it is exactly
// sha/size. Rare (a republish of the same file), so the read is acceptable.
func (p *geoipPublisher) verifyExistingBlob(ctx context.Context, key, sha string, size int64) error {
	rc, _, err := release.OpenObject(ctx, p.store, key)
	if err != nil {
		return fmt.Errorf("existing blob unreadable: %w", err)
	}
	defer rc.Close()
	h := sha256.New()
	n, err := io.Copy(h, io.LimitReader(rc, size+1))
	if err != nil {
		return fmt.Errorf("existing blob unreadable: %w", err)
	}
	if n != size || hex.EncodeToString(h.Sum(nil)) != sha {
		return refused("a different object already sits at this database's blob key")
	}
	return nil
}

// Off deletes the pointer, then every blob (spec §3). Unconditional delete:
// safe under the single-writer mutex (codex R2-6).
func (p *geoipPublisher) Off(ctx context.Context) error {
	// Never waits behind a publish (codex R9-2): a slow upload or job can
	// hold the slot for minutes, longer than any caller's timeout. The
	// operator gets a 409 and runs off again when the publish finishes.
	if !p.mu.TryLock() {
		return errGeoipBusy
	}
	defer p.mu.Unlock()
	ctx, cancel := context.WithTimeout(ctx, 2*time.Minute)
	defer cancel()
	if err := p.store.Delete(ctx, geoip.PointerKey); err != nil && !errors.Is(err, os.ErrNotExist) {
		return err
	}
	return p.sweepLocked(ctx, "")
}

// Sweep deletes every blob the pointer does not name (orphans from a crash
// between blob upload and pointer write). Run at startup and daily.
func (p *geoipPublisher) Sweep(ctx context.Context) error {
	p.mu.Lock()
	defer p.mu.Unlock()
	cur, _, present, err := p.Current(ctx)
	if err != nil {
		return err
	}
	keep := ""
	if present && !cur.Expired {
		keep = cur.Key
	}
	return p.sweepLocked(ctx, keep)
}

func (p *geoipPublisher) sweepLocked(ctx context.Context, keep string) error {
	keys, err := p.store.List(ctx, geoip.Prefix)
	if err != nil {
		return err
	}
	var firstErr error
	for _, k := range keys {
		if !strings.HasSuffix(k, ".mmdb") || k == keep {
			continue
		}
		if err := p.store.Delete(ctx, k); err != nil && !errors.Is(err, os.ErrNotExist) && firstErr == nil {
			firstErr = err
		}
	}
	return firstErr
}

// ExpireIfDue applies D1 to the bucket: a GeoLite2 pointer past expiry is
// rewritten with expired=true and its blob destroyed.
func (p *geoipPublisher) ExpireIfDue(ctx context.Context, now time.Time) (bool, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	cur, etag, present, err := p.Current(ctx)
	if err != nil || !present || cur.Expired || !geoip.Expired(cur.DatabaseType, cur.Build(), now) {
		return false, err
	}
	cur.Expired = true
	body, err := marshalPointer(cur)
	if err != nil {
		return false, err
	}
	if _, err := p.store.Put(ctx, geoip.PointerKey, body, etag); err != nil {
		return false, err
	}
	log.Printf("console: geoip database %s expired (GeoLite2, 30 days after build) and was removed", cur.SHA256[:8])
	return true, p.sweepLocked(ctx, "")
}