package dbreads

import (
	"context"
	"crypto/ed25519"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"net"
	"os"
	"sync"
	"sync/atomic"
	"time"

	mrand "math/rand/v2"

	"github.com/carlosframework/platform/internal/dbdoor"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// AgentConfig is what the box's db-reads loop needs (spec §7, §8).
// internal/host fills it from the box's environment; tests fill it with
// stand-ins.
type AgentConfig struct {
	Store           release.Store     // the deployment bucket
	Region          string            // CARLOS_REGION
	Box             string            // this box's label, as its edges file names it
	Boot            string            // this agent process's id
	Pub             ed25519.PublicKey // CARLOS_RELEASE_PUBKEY
	HibernatePrefix string            // CARLOS_HIBERNATE_PREFIX
	// OpenReplica opens the read side of this box's own region bucket at a
	// prefix the agent derived (lease.ReadReplicaClient in production).
	OpenReplica func(ctx context.Context, prefix string) (dbdoor.ReplicaReader, error)
	// DialWorker connects to carlos-dbread.socket.
	DialWorker func(ctx context.Context) (net.Conn, error)
	// Units counts carlos-dbread@ units active, activating or deactivating.
	Units func(ctx context.Context) (int, error)
	// Free reads the filesystem holding /var/tmp.
	Free        func() (avail, total uint64, err error)
	MaxRestored int64 // DefaultMaxRestored when zero
	// MaxPull is a pull's cap on its planned LTX files: MaxPullBytes when
	// zero or larger, so no box ships more than a CLI accepts, and never
	// more than the worker holds (PlanBytesFactor times MaxRestored).
	MaxPull int64
	Now     func() time.Time
	Jitter  func() time.Duration // 0 to MaxJitter
	Logf    func(format string, args ...any)
	// Draining is this box's own published draining flag (its edges file's
	// draining field); a draining box takes no new read. Never draining
	// when nil.
	Draining func() bool

	// Test seams; zero is the spec's value.
	OwnerDeadline time.Duration
	PlanBudget    time.Duration
	CancelEvery   time.Duration
	ChunkEvery    time.Duration
	MaxOutput     int64
}

// Agent is the box's half of forwarded reads (spec §8.1): it lists the
// region's queue, verifies requests, elects an owner, runs the identity
// bracket around planning and download, streams bytes to a worker, and
// seals and writes output. It never opens SQLite, decodes pages, runs SQL or
// writes a file.
type Agent struct {
	cfg AgentConfig
	adm *admission

	mu      sync.Mutex
	decided map[string]time.Time

	probeOK    atomic.Bool
	probeAfter chan struct{}
	reads      sync.WaitGroup
}

func NewAgent(cfg AgentConfig) *Agent {
	if cfg.MaxRestored <= 0 {
		cfg.MaxRestored = DefaultMaxRestored
	}
	if cfg.MaxPull <= 0 || cfg.MaxPull > MaxPullBytes {
		cfg.MaxPull = MaxPullBytes
	}
	// The worker refuses a plan whose files total more than this (its
	// planWithinBounds), so a pull over it ends too-large here instead.
	if held := PlanBytesFactor * cfg.MaxRestored; cfg.MaxPull > held {
		cfg.MaxPull = held
	}
	if cfg.Now == nil {
		cfg.Now = time.Now
	}
	if cfg.Jitter == nil {
		cfg.Jitter = func() time.Duration { return mrand.N(MaxJitter) }
	}
	if cfg.Logf == nil {
		cfg.Logf = func(string, ...any) {}
	}
	if cfg.Draining == nil {
		cfg.Draining = func() bool { return false }
	}
	if cfg.OwnerDeadline == 0 {
		cfg.OwnerDeadline = OwnerDeadline
	}
	if cfg.PlanBudget == 0 {
		cfg.PlanBudget = PlanBudget
	}
	if cfg.CancelEvery == 0 {
		cfg.CancelEvery = CancelEvery
	}
	if cfg.ChunkEvery == 0 {
		cfg.ChunkEvery = ChunkEvery
	}
	if cfg.MaxOutput == 0 {
		cfg.MaxOutput = MaxOutputBytes
	}
	return &Agent{cfg: cfg, adm: newAdmission(cfg.Units, cfg.Free), decided: map[string]time.Time{},
		probeAfter: make(chan struct{}, 1)}
}

// Capable is whether this box may publish db-read-1: its last readiness
// probe passed (spec §7.1).
func (a *Agent) Capable() bool { return a.probeOK.Load() }

// Wait blocks until every read this agent owns has written its result.
func (a *Agent) Wait() { a.reads.Wait() }

func (a *Agent) now() time.Time { return a.cfg.Now().UTC() }

func (a *Agent) logf(format string, args ...any) { a.cfg.Logf(format, args...) }

func (a *Agent) decide(id string) {
	a.mu.Lock()
	defer a.mu.Unlock()
	a.decided[id] = a.now()
}

func (a *Agent) isDecided(id string) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	_, ok := a.decided[id]
	return ok
}

// prune forgets decisions older than any listing can show again.
func (a *Agent) prune(now time.Time) {
	a.mu.Lock()
	defer a.mu.Unlock()
	for id, at := range a.decided {
		if now.Sub(at) > 2*time.Hour {
			delete(a.decided, id)
		}
	}
}

// askProbe asks the loop for a probe now: a read's worker failed to start
// (spec §7.1).
func (a *Agent) askProbe() {
	select {
	case a.probeAfter <- struct{}{}:
	default:
	}
}

// closeThenProbe gives the failed worker's connection slot back before
// asking for a probe, so admission sees the worker's unit as the orphan it
// now is and holds the probe until the unit has stopped (plan review
// finding 29). The deferred Close that follows is a no-op (countedConn).
func closeThenProbe(a *Agent, conn net.Conn) {
	conn.Close()
	a.askProbe()
}

// countedConn gives a worker connection's slot back exactly once.
type countedConn struct {
	net.Conn
	once sync.Once
	adm  *admission
}

func (c *countedConn) Close() error {
	err := c.Conn.Close()
	c.once.Do(c.adm.closed)
	return err
}

// dial opens a worker connection inside the three the socket allows.
func (a *Agent) dial(ctx context.Context) (net.Conn, error) {
	if !a.adm.opened() {
		return nil, errBusy
	}
	c, err := a.cfg.DialWorker(ctx)
	if err != nil {
		a.adm.closed()
		return nil, err
	}
	return &countedConn{Conn: c, adm: a.adm}, nil
}

// check is one half of the identity bracket (spec §8.2): the record read
// fresh, and refused unless it is the instance the envelope named, still a
// replicated instance, still in this box's region and not moving. A class,
// pool or fleet change after the console admitted the read is a record
// change.
func (a *Agent) check(ctx context.Context, req Request) (provision.Record, string, error) {
	rec, err := provision.GetRecord(ctx, a.cfg.Store, req.Account, req.App, req.Host)
	if errors.Is(err, os.ErrNotExist) {
		return provision.Record{}, CodeRecordChanged, nil
	}
	if err != nil {
		return provision.Record{}, "", err
	}
	inc, ierr := hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
	_, fleet := registry.FleetPlacement(rec.Region)
	switch {
	case ierr != nil, inc != req.Incarnation, rec.Account != req.Account, rec.App != req.App,
		rec.Host != req.Host, !provision.ValidHostShape(rec.Host), ClassRefusal(*rec) != "",
		rec.Pool != "", fleet, rec.Region != a.cfg.Region:
		return *rec, CodeRecordChanged, nil
	case rec.Move != nil:
		return *rec, CodeMoving, nil
	}
	return *rec, "", nil
}

func nonce() string {
	var b [16]byte
	rand.Read(b[:])
	return hex.EncodeToString(b[:])
}

// elect is J3: the first create of picked owns the read. A create that
// answers an error or a conflict is decided by reading picked back: this
// agent's own nonce owns it, anything else, or a read that fails, does not.
func (a *Agent) elect(ctx context.Context, id string) (Picked, bool) {
	p := Picked{Box: a.cfg.Box, Boot: a.cfg.Boot, Nonce: nonce(), Deadline: a.now().Add(a.cfg.OwnerDeadline)}
	body, err := json.Marshal(p)
	if err != nil {
		return Picked{}, false
	}
	if _, err := a.cfg.Store.Put(ctx, PickedKey(id), body, ""); err == nil {
		return p, true
	}
	raw, _, err := a.cfg.Store.Get(ctx, PickedKey(id))
	if err != nil {
		return Picked{}, false
	}
	var got Picked
	if json.Unmarshal(raw, &got) != nil || got.Nonce != p.Nonce {
		return Picked{}, false
	}
	return got, true
}

// consider decides one in-window queue entry (spec §5.5 step 4): read and
// verify the request against its entry, check admission, wait the jitter,
// and enter the election. A refused or forged request is decided, and so
// logged once. A read that fails admission is not decided: a later pass may
// admit it while the window lasts.
func (a *Agent) consider(ctx context.Context, e QueueEntry) {
	raw, _, err := a.cfg.Store.Get(ctx, RequestKey(e.ID))
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			a.decide(e.ID)
			a.logf("dbreads: %s: a queue entry with no request; ignored", e.ID)
		}
		return
	}
	req, err := Verify(a.cfg.Pub, raw, e.ID, a.cfg.Region, a.now())
	if err == nil && !QueueMatches(e, req) {
		err = ErrEnvelope
	}
	if err != nil {
		a.decide(e.ID)
		a.logf("dbreads: %s: refused the request: %v", e.ID, err)
		return
	}
	// A request whose console has already given up (its request ended
	// before it was announced) is never picked (spec §5.5 step 4).
	if _, _, err := a.cfg.Store.Get(ctx, CancelKey(e.ID)); err == nil {
		a.decide(e.ID)
		return
	}
	// A draining box does not enter the election (spec §7.1, D2).
	if a.cfg.Draining() {
		return
	}
	if err := a.adm.admitRead(ctx, e.ID); err != nil {
		return
	}
	if d := a.cfg.Jitter(); d > 0 {
		select {
		case <-ctx.Done():
			a.adm.abandon(e.ID)
			return
		case <-time.After(d):
		}
	}
	// It may have started draining during the jitter.
	if a.cfg.Draining() {
		a.adm.abandon(e.ID)
		return
	}
	p, owned := a.elect(ctx, e.ID)
	a.decide(e.ID)
	if !owned {
		a.adm.abandon(e.ID)
		return
	}
	a.reads.Add(1)
	go a.serve(ctx, req, p)
}

var errSandbox = errors.New("dbreads: the worker can see more than its sandbox allows")

// probeRetry is how soon a probe that admission skipped is tried again.
var probeRetry = time.Second

// Run is the db-reads loop: a probe at start, a pass on every nudge (a
// serial move) and every ListEvery, a probe every ProbeEvery and whenever a
// read's worker failed to start (spec §5.5 step 4, §7.1). A probe that
// admission skips (the failed worker's unit is still stopping) stays owed
// and is tried every probeRetry until it runs (plan review finding 29).
func (a *Agent) Run(ctx context.Context, nudge <-chan struct{}) {
	var retry <-chan time.Time
	tryProbe := func() {
		retry = nil
		if !a.Probe(ctx) {
			retry = time.After(probeRetry)
		}
	}
	tryProbe()
	list := time.NewTicker(ListEvery)
	defer list.Stop()
	probe := time.NewTicker(ProbeEvery)
	defer probe.Stop()
	for {
		select {
		case <-ctx.Done():
			a.reads.Wait()
			return
		case <-nudge:
			a.Pass(ctx)
		case <-list.C:
			a.Pass(ctx)
		case <-probe.C:
			tryProbe()
		case <-a.probeAfter:
			tryProbe()
		case <-retry:
			tryProbe()
		}
	}
}

// Pass lists this region's queue for this hour and the last, and considers
// every in-window entry it has not decided. An out-of-window entry is
// skipped on its key alone. A box whose capability is withdrawn takes no
// new read; one it already holds runs on (spec §11.2).
func (a *Agent) Pass(ctx context.Context) {
	if !a.Capable() {
		return
	}
	now := a.now()
	a.prune(now)
	for _, hour := range []time.Time{now.Add(-time.Hour), now} {
		keys, err := a.cfg.Store.List(ctx, QueueHourPrefix(a.cfg.Region, hour))
		if err != nil {
			a.logf("dbreads: list the %s queue: %v", a.cfg.Region, err)
			continue
		}
		for _, key := range keys {
			e, ok := ParseQueueKey(key)
			if !ok || e.Region != a.cfg.Region || !InWindow(e.Issued, now) || a.isDecided(e.ID) {
				continue
			}
			a.consider(ctx, e)
		}
	}
}

// Probe is the readiness probe (spec §7.1): a hello on its own connection,
// so a busy box is never mistaken for a broken one. While admission is
// blocked (an orphan unit, or no unit listing) it is skipped, the last
// result stands, and it answers false so Run keeps it owed.
func (a *Agent) Probe(ctx context.Context) bool {
	if !a.adm.admitProbe(ctx) {
		return false
	}
	defer a.adm.probeDone()
	err := a.probe(ctx)
	if err != nil {
		a.logf("dbreads: readiness probe: %v", err)
	}
	a.probeOK.Store(err == nil)
	return true
}

func (a *Agent) probe(ctx context.Context) error {
	pctx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	conn, err := a.dial(pctx)
	if err != nil {
		return err
	}
	defer conn.Close()
	defer context.AfterFunc(pctx, func() { conn.Close() })()
	if err := writeJSONFrame(conn, frameHello, hello{V: ProtocolVersion}); err != nil {
		return err
	}
	typ, p, err := readFrame(conn)
	if err != nil {
		return err
	}
	var h HelloReply
	if typ != frameHelloReply || json.Unmarshal(p, &h) != nil {
		return errProtocol
	}
	if h.V != ProtocolVersion {
		return fmt.Errorf("dbreads: the worker speaks protocol %d, this agent %d", h.V, ProtocolVersion)
	}
	if len(h.Errors) != 0 || len(h.Interfaces) != 1 || h.Interfaces[0] != "lo" || h.DataEntries != 0 || h.EtcReadable {
		return fmt.Errorf("%w: %+v", errSandbox, h)
	}
	return nil
}