package dbreads

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"
)

var (
	errBusy    = errors.New("dbreads: this box is running its two reads")
	errOrphans = errors.New("dbreads: a carlos-dbread@ unit this agent is not running is still active")
	errNoRoom  = errors.New("dbreads: not enough disk for the read")
)

// admission is the agent's one table (spec §5.7, J5): how many worker
// connections it holds, which reads it is running, and every disk
// reservation. The agent is the only process that starts workers, so one
// mutex covers it all.
type admission struct {
	units func(context.Context) (int, error)
	free  func() (avail, total uint64, err error)

	mu      sync.Mutex
	conns   int
	running map[string]bool
	held    map[string]int64
	probing bool
}

func newAdmission(units func(context.Context) (int, error), free func() (uint64, uint64, error)) *admission {
	return &admission{units: units, free: free, running: map[string]bool{}, held: map[string]int64{}}
}

// reconcileLocked is the refusal-based reconciliation (spec §5.7): while any
// carlos-dbread@ unit is active, activating or deactivating beyond the
// connections this agent holds, nothing is admitted. A finished read's
// reservation is released only when no unit is left at all: then every
// stopped unit's scratch has been removed by systemd.
func (a *admission) reconcileLocked(ctx context.Context) error {
	n, err := a.units(ctx)
	if err != nil {
		return fmt.Errorf("dbreads: list carlos-dbread@ units: %w", err)
	}
	if n > a.conns {
		return errOrphans
	}
	if n == 0 {
		for id := range a.held {
			if !a.running[id] {
				delete(a.held, id)
			}
		}
	}
	return nil
}

// fitsLocked: after subtracting every other reservation and need, the
// filesystem holding /var/tmp keeps DiskFloor and DiskFreePercent free.
func (a *admission) fitsLocked(id string, need int64) bool {
	avail, total, err := a.free()
	if err != nil {
		return false
	}
	left := int64(avail) - need
	for k, v := range a.held {
		if k != id {
			left -= v
		}
	}
	return left >= DiskFloor && left >= int64(total)*DiskFreePercent/100
}

// admitRead admits a read before it enters the election: a free read slot,
// no orphan, and room for the floor reservation.
func (a *admission) admitRead(ctx context.Context, id string) error {
	a.mu.Lock()
	defer a.mu.Unlock()
	if err := a.reconcileLocked(ctx); err != nil {
		return err
	}
	if len(a.running) >= ReadsPerBox {
		return errBusy
	}
	if !a.fitsLocked(id, DiskFloor) {
		return errNoRoom
	}
	a.running[id] = true
	a.held[id] = DiskFloor
	return nil
}

// grow raises a running read's reservation to need (never below the floor)
// once its plan is known.
func (a *admission) grow(id string, need int64) error {
	a.mu.Lock()
	defer a.mu.Unlock()
	need = max(need, DiskFloor)
	if !a.fitsLocked(id, need) {
		return errNoRoom
	}
	a.held[id] = need
	return nil
}

// finish ends a read's slot. Its reservation stays until reconcileLocked
// sees no unit left.
func (a *admission) finish(id string) {
	a.mu.Lock()
	defer a.mu.Unlock()
	delete(a.running, id)
}

// abandon gives back a read this agent admitted and then did not run (it
// lost the election, or stopped before it): no worker was started for it,
// so its reservation goes at once rather than waiting for the units to drain.
func (a *admission) abandon(id string) {
	a.mu.Lock()
	defer a.mu.Unlock()
	delete(a.running, id)
	delete(a.held, id)
}

// retry is a read starting a fresh plan after a planned file vanished (spec
// §8.2). Its previous worker may still hold scratch until its unit stops, so
// that reservation moves to a retired key, held until no unit is left (J5),
// and the read starts again from the floor (plan review finding 10).
func (a *admission) retry(id string, attempt int) {
	a.mu.Lock()
	defer a.mu.Unlock()
	if v, ok := a.held[id]; ok {
		a.held[fmt.Sprintf("%s#%d", id, attempt)] = v
	}
	a.held[id] = DiskFloor
}

// settle waits until no carlos-dbread@ unit is running beyond this agent's
// open connections, so a retry opens its next worker only once the last one
// is gone, and never into the probe's connection. It releases retired
// reservations once no unit is left at all.
func (a *admission) settle(ctx context.Context, poll time.Duration) error {
	for {
		a.mu.Lock()
		err := a.reconcileLocked(ctx)
		a.mu.Unlock()
		if !errors.Is(err, errOrphans) {
			return err
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(poll):
		}
	}
}

// admitProbe admits the readiness probe: no orphan and no other probe. The
// probe's connection is the third (opened checks the count).
func (a *admission) admitProbe(ctx context.Context) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	if a.probing || a.reconcileLocked(ctx) != nil {
		return false
	}
	a.probing = true
	return true
}

func (a *admission) probeDone() {
	a.mu.Lock()
	defer a.mu.Unlock()
	a.probing = false
}

// opened takes a connection slot; the agent never opens a fourth.
func (a *admission) opened() bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	if a.conns >= MaxConnections {
		return false
	}
	a.conns++
	return true
}

func (a *admission) closed() {
	a.mu.Lock()
	defer a.mu.Unlock()
	a.conns--
}