package host

import (
	"bytes"
	"context"
	"errors"
	"github.com/carlosframework/platform/internal/hostwrite"
	"log"
	"net/http"
	"net/http/httputil"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/edge"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/schedule"
	"github.com/carlosframework/platform/internal/version"
)

// Scheduled work, host side (spec §6.2, §6.3, §4.3).
//
// Two halves, one set of dependencies:
//
//   - the CONVERGER turns apps/<acct>/<app>/schedules.json into timer rows in
//     this box's registry. It runs on every serial move (so a `carlos schedule
//     set` lands in seconds, like every other console-mediated write), hourly
//     underneath that (the floor a missed bump falls back to), and once at
//     agent start (a box that booted after the bump must not wait an hour to
//     learn its schedules).
//   - the RUNNER fires the rows that are due, every 15s, and reports what
//     happened back to the bucket so `carlos schedule status` can answer
//     without a box command.
//
// Both are built ONCE, in the agent's startup, and shared: the converger's
// etag memory is what keeps an idle box at one GET per app per pass and makes
// a run signal exactly-once, so a second instance in another loop would
// re-seed that memory and either replay a run or swallow one.
//
// The two halves have different preconditions. The converger needs the deploy
// store (schedules live in the bucket). The runner does not: once-timers are
// registered by instances on their own control socket (spec §7.2) straight
// into this box's registry, so a box with an armed activator and no store
// still has rows to fire — conv is nil there, and nothing is reported.
type scheduleDeps struct {
	conv   *schedule.Converger // nil when the box has no deploy store
	runner *schedule.Runner
}

// newScheduleDeps assembles the converger and the runner around this agent's
// registry, deploy store and activator.
//
// store may be nil (no deploy store on this box): no converger is built and
// the runner reports nothing, but it still fires the once rows instances
// register. svc may be nil (no hibernation armed on this box): delivery then
// proxies straight to the instance's socket, which is exactly right — an
// instance that is not activator-managed is either already running or not
// this leg's problem. sink may be nil (no log store): the runner simply emits
// no member events. Callers must pass a genuinely nil interface for sink, not
// a nil *applog.Sink — see the trap agent.go documents where logSink is
// threaded.
func newScheduleDeps(reg *registry.Registry, store release.Store, svc *activator.Service, sink activator.EventSink, dd string) *scheduleDeps {
	box, err := os.Hostname()
	if err != nil || box == "" {
		box = "unknown"
	}
	var event func(rt registry.Route, level, msg string)
	if sink != nil {
		event = func(rt registry.Route, level, msg string) {
			sink.Event(release.RouteAccount(rt), rt.App, rt.Host, level, msg)
		}
	}
	// Zero when no activator is armed on this box: nothing hibernates, so no
	// schedule can hold anything awake, and holdsAwake reports false for all
	// of them.
	var idle time.Duration
	if svc != nil {
		idle = svc.IdleWindow()
	}
	sd := &scheduleDeps{
		runner: &schedule.Runner{
			Reg:     reg,
			Deliver: &activatorDeliverer{svc: svc},
			Token:   scheduleToken(dd, svc, reg),
			Skip:    scheduleSkip,
			Event:   event,
			Box:     box,
			Build:   version.Build,
			Idle:    idle,
		},
	}
	if store == nil {
		return sd
	}
	// The reporter is shared by both halves on purpose: a converge that adds a
	// row and a tick that fires it write the SAME status object, so whichever
	// ran last is what a member reads. Its own timeout, not the caller's — a
	// slow bucket must not hold a converge pass or a tick goroutine open.
	report := func(rt registry.Route, st schedule.Status) {
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		if err := schedule.WriteStatus(ctx, store, release.RouteAccount(rt), rt.App, st); err != nil {
			log.Printf("schedule: report %s: %v", rt.Host, err)
		}
	}
	sd.conv = schedule.NewConverger(reg, store, box, report)
	sd.conv.Idle = idle
	sd.runner.Report = report
	return sd
}

// armSchedule is the agent's arming switch for scheduled work, lifted out of
// agent.go so the decision itself can be tested rather than only its parts.
//
// storeErr is openDeployStore's error, and the three arms are not
// interchangeable:
//
//   - a deploy store: both halves. Declarations converge into rows and every
//     fire reports back to the bucket.
//   - NO store but an armed activator: the RUNNER ALONE, no converger, no
//     reports. This is the arm a review found missing (I6). A box that mounts
//     the control socket's schedule verbs accepts once-timers from its own
//     instances (spec §7.2) straight into the registry — and a box that
//     accepts a timer must fire it. Gating the runner on the store would have
//     left those rows sitting in the table forever.
//   - neither: nothing at all, since nothing could write a row or fire one.
//
// It returns nil for that last arm, which every caller (and scheduleLegFor,
// and converge) already tolerates.
func armSchedule(reg *registry.Registry, store release.Store, storeErr error, svc *activator.Service, sink activator.EventSink, dd string) *scheduleDeps {
	switch {
	case storeErr == nil:
		log.Printf("agent: scheduled work armed (converge on serial + hourly; runner every 15s)")
		return newScheduleDeps(reg, store, svc, sink, dd)
	case svc != nil:
		log.Printf("agent: schedules from the bucket are off (%v); once-timers registered by instances still run (runner every 15s)", storeErr)
		return newScheduleDeps(reg, nil, svc, sink, dd)
	default:
		log.Printf("agent: scheduled work disabled: %v", storeErr)
		return nil
	}
}

// scheduleSkip is the runner's Skip (harness Run 9, defect 8): a host whose
// row this box holds, and a host whose offline move's record names another
// region. After open the source keeps its row, held, until finishing, and
// open has lifted the fence that used to refuse the token; so the source's
// tick minted a fresh <host>.env (an admin token) inside the instance
// directory the move had just parked, a stray secret for a host served
// elsewhere, and recorded a delivery the activator refuses for a held row
// anyway (ServeScheduled). The token is the only instance file the runner
// writes; delivery wakes nothing for a held row, and the converger writes
// registry rows and the bucket status only. Any hold counts, not only an
// offline move's: no held row is ever served a tick, and a hold whose move
// has just ended is still a hold until reconcile removes or releases the
// row.
func scheduleSkip(rt registry.Route) bool {
	if registry.IsHold(rt.Status) {
		return true
	}
	if v := armedOfflineView(); v != nil {
		return v.elsewhere(rt.Host)
	}
	return false
}

// scheduleToken is the runner's Token: ensure-then-read on the activator's
// own <dataDir>/<host>.env (spec §6.4).
//
// Ensure, not just read, because the token is otherwise minted only when the
// activator spawns the instance — and the tick is what would spawn it. An
// instance asleep since before this build rolled has no token; a read-only
// Token would record "no instance token" five times and give up, and the
// wake that mints it would never come. EnsureAdminToken is the same call
// Start and Seed make, serialized with them, so a tick and a visitor racing
// on a fresh host mint ONE token.
//
// The error is redacted: it reaches the row's last_status and from there the
// bucket status object and `carlos schedule ls`, and a member must not learn
// the box's filesystem layout from it.
func scheduleToken(dd string, svc *activator.Service, reg *registry.Registry) func(host string) (string, error) {
	paths := envPathFor(dd, reg)
	read := schedule.TokenFromEnvPaths(paths)
	prepare := func(host string) (string, error) {
		if err := activator.EnsureAdminToken(paths(host)); err != nil {
			return "", errors.New(activator.RedactPaths(err.Error()))
		}
		tok, err := read(host)
		if err != nil {
			return "", errors.New(activator.RedactPaths(err.Error()))
		}
		return tok, nil
	}
	return func(host string) (string, error) {
		if svc == nil {
			return prepare(host)
		}
		return svc.PrepareScheduledToken(host, func() (string, error) {
			return prepare(host)
		})
	}
}

// envPathFor is scheduleToken's route to an instance's own env file: flat by
// default, and the nested layout's own <host>.env (design §7) when this
// box's registry classifies the host as a nested row. It is the SAME
// classification the exec backend used to decide where it spawned the
// instance in the first place (activator.Classify), so the runner and the
// token-ensure step ask the file the backend actually wrote rather than
// guessing the flat path for every host. reg nil (a test with no registry
// in hand) is every host flat, byte-identical to before this existed.
func envPathFor(dd string, reg *registry.Registry) func(host string) string {
	flat := func(host string) string { return filepath.Join(dd, host+".env") }
	if reg == nil {
		return flat
	}
	return func(host string) string {
		rt, err := reg.Get(host)
		if err != nil || rt == nil {
			return flat(host)
		}
		if layout, _ := activator.Classify(*rt, activator.LayoutInputs{DataDir: dd}); layout == activator.LayoutNested {
			return filepath.Join(provision.InstanceDir(dd, host), host+".env")
		}
		return flat(host)
	}
}

// convergePassTimeout bounds ONE converge pass. The agent's three call sites
// (the seed and the serial leg in fastConvergeLoop, the hourly floor in
// adoptLoop) all hand this context.Background() — an unbounded context — and
// a pass is store-bound: a bucket that accepts a connection and then never
// answers would park the pass, and with it every caller queued on
// Converger's own mutex, for as long as the process lives. Generous on
// purpose: a pass reads one object per app plus the run signal, so minutes
// is far past any healthy bucket and still finite. The floors underneath
// (hourly, and the next serial move) are what recover a pass cut short.
//
// It bounds QUEUE PLUS WORK, not work alone: Pass takes the converger's
// mutex first, so a caller behind a parked pass can spend the budget waiting
// and then run with an expired context. Harmless — every app read fails,
// convergeApp is die-free and leaves its rows alone, no etag is poisoned —
// and it is the property that matters here: the leg returns, whatever the
// bucket is doing.
const convergePassTimeout = 5 * time.Minute

// converge runs one converger pass, or nothing on a box with no deploy
// store. The agent's three call sites go through this so none of them can
// reach into a nil conv — and so that all three get the timeout above
// rather than each remembering to apply it.
func (sd *scheduleDeps) converge(ctx context.Context) error {
	if sd == nil || sd.conv == nil {
		return nil
	}
	ctx, cancel := context.WithTimeout(ctx, convergePassTimeout)
	defer cancel()
	return sd.conv.Pass(ctx)
}

// scheduleLegFor is fastDeps.scheduleLeg for sd — a genuinely nil func when
// there is nothing to converge (no scheduled work armed, or no deploy store
// to read declarations from), so the tick skips the leg rather than calling
// a closure over a nil converger.
func scheduleLegFor(sd *scheduleDeps) func(ctx context.Context) {
	if sd == nil || sd.conv == nil {
		return nil
	}
	return func(ctx context.Context) {
		if err := sd.converge(ctx); err != nil {
			log.Printf("agent: schedule converge: %v", err)
		}
	}
}

// activatorDeliverer delivers one tick the way a member request reaches an
// instance from inside this process — through the activator, then the
// instance's unix socket — and never through the public edge (spec §6.3).
// Two properties come from that and only that:
//
//   - a hibernating instance WAKES for its tick, exactly as it would for a
//     visitor, and is held awake for the tick's duration because the
//     activator's in-flight count holds it open — but the tick earns no
//     activity credit, so it does not by itself keep the instance awake
//     afterwards. Separately, the sweeper's schedule guard (activator.go,
//     sweepOnce) holds a host resident whenever its next tick is due sooner
//     than the idle window, so a schedule firing more often than the idle
//     window stays resident by that guard's design rather than thrashing
//     through a wake/sleep pair on every tick. A schedule looser than the
//     idle window is not covered by the guard and does sleep between ticks,
//     waking again for the next one.
//   - the tick never leaves the box: no DNS, no TLS, no edge auth, nothing an
//     outsider could aim at the same path.
type activatorDeliverer struct {
	svc *activator.Service

	// proxies caches one unix proxy — and therefore one http.Transport, and
	// therefore one connection pool — per socket path. A fresh Transport per
	// delivery meant a fresh connection for every tick and a pool thrown
	// away unread; a schedule that fires every minute paid a dial (and the
	// runtime a reader goroutine) each time for nothing.
	// Keyed by socket path rather than host because the path is what the
	// Transport actually dials: a route re-pointed at a new socket gets a new
	// entry rather than a pool aimed at the old file.
	//
	// Bounded three ways, in the order they normally fire:
	//
	//   - the runner's daily prune leg hands this the sockets still routed
	//     here (RetainSockets) and everything else goes. That is the one that
	//     reclaims a RETIRED route: it stops receiving deliveries, so it can
	//     never error its own way out.
	//   - an entry is dropped when its delivery comes back as an error — a
	//     refused wake or an expired deadline. (A socket that will not dial
	//     is NOT one of these: ReverseProxy answers that itself with a 502
	//     and no error. Nothing is pooled behind a failed dial anyway.)
	//   - the whole map is dropped if it ever passes maxProxyCache, a safety
	//     net rather than a working mechanism.
	//
	// Every one of those closes the dropped transports' idle connections
	// rather than waiting out edge.UnixProxy's 90s IdleConnTimeout.
	mu      sync.Mutex
	proxies map[string]http.Handler

	// genPools is the delivery leg's own per-generation-socket pools, used
	// instead of proxies for an activator-managed instance once generations
	// are on (zero-downtime spec §7.1): a tick reaches the CURRENT
	// generation, asked per connection. Made on first use and retired
	// through the activator's hook, so a retired generation's pool closes
	// with it rather than waiting for the daily prune. Nil with generations
	// off, and nothing below reads it then.
	genOnce  sync.Once
	genPools *edge.GenPools
}

// managedByGenerations reports whether rt's ticks go through the current
// generation: generations on, and an activator-managed instance.
func (d *activatorDeliverer) managedByGenerations(rt registry.Route) bool {
	return d.svc != nil && d.svc.GenerationsOn() && rt.Kind == "instance" && rt.Hibernate
}

// genProxyFor is the AppSocket proxy for host over the leg's own pools.
func (d *activatorDeliverer) genProxyFor(host string) http.Handler {
	d.genOnce.Do(func() {
		d.genPools = edge.NewGenPools()
		d.svc.OnGenRetired(d.genPools.Retire)
	})
	return edge.AppSocketProxy(host, d.svc, d.genPools)
}

// maxProxyCache is that ceiling. Deliberately well above the number of
// exec instances a box runs (the 1000-per-host timer cap is per host, not
// per box) so the reset is a safety net, never a working mechanism.
const maxProxyCache = 512

// proxyFor is the cached unix proxy for socket.
func (d *activatorDeliverer) proxyFor(socket string) http.Handler {
	d.mu.Lock()
	defer d.mu.Unlock()
	if p, ok := d.proxies[socket]; ok {
		return p
	}
	if d.proxies == nil {
		d.proxies = map[string]http.Handler{}
	} else if len(d.proxies) >= maxProxyCache {
		for _, p := range d.proxies {
			closeIdle(p)
		}
		clear(d.proxies)
	}
	p := edge.UnixProxy(socket)
	d.proxies[socket] = p
	return p
}

// dropProxy forgets socket's proxy and closes the connections it was holding.
func (d *activatorDeliverer) dropProxy(socket string) {
	d.mu.Lock()
	defer d.mu.Unlock()
	if p, ok := d.proxies[socket]; ok {
		closeIdle(p)
		delete(d.proxies, socket)
	}
}

// RetainSockets keeps only the sockets that are still routed on this box and
// drops the rest — schedule.SocketRetainer, called by the runner's prune leg
// with the host list it has already read. Without it a retired route's entry
// would sit here until the maxProxyCache reset, since a route nothing
// delivers to can never error its way out.
func (d *activatorDeliverer) RetainSockets(live map[string]bool) {
	d.mu.Lock()
	defer d.mu.Unlock()
	for socket, p := range d.proxies {
		if !live[socket] {
			closeIdle(p)
			delete(d.proxies, socket)
		}
	}
}

// closeIdle returns a dropped proxy's pooled connections now rather than at
// the end of edge.UnixProxy's 90s IdleConnTimeout. Best-effort by design:
// both assertions are guarded, so a change to the shape edge builds costs
// the eager close and nothing else — the idle timeout is still the backstop.
// (schedule_wake_test.go pins the shape, so such a change fails a test
// rather than passing quietly.)
func closeIdle(h http.Handler) {
	if rp, ok := h.(*httputil.ReverseProxy); ok {
		if tr, ok := rp.Transport.(*http.Transport); ok {
			tr.CloseIdleConnections()
		}
	}
}

func (d *activatorDeliverer) Deliver(ctx context.Context, rt registry.Route, req *http.Request) (int, error) {
	return d.deliverSelected(ctx, rt, req, nil)
}
func (d *activatorDeliverer) DeliverClaim(ctx context.Context, rt registry.Route, req *http.Request, admit func(context.Context) (context.Context, func(), error)) (int, error) {
	return d.deliverSelected(ctx, rt, req, admit)
}
func (d *activatorDeliverer) deliverSelected(ctx context.Context, rt registry.Route, req *http.Request, admit func(context.Context) (context.Context, func(), error)) (int, error) {
	if rt.Socket == "" {
		// An addr-backed (or tunnel-backed) route has no socket to dial from
		// here. Converge only ever writes rows for exec-backed instances, so
		// this is a row that outlived its route's shape — an error, not a dial
		// to an empty path (which would be a connect-refused every 15s).
		return 0, errors.New("no socket for route " + rt.Host)
	}
	// Generations on: the proxy asks the activator for the current
	// generation's socket per connection (spec §7.1). Off: the cached proxy
	// on rt.Socket, exactly as before.
	gen := d.managedByGenerations(rt)
	var proxy http.Handler
	if gen {
		proxy = d.genProxyFor(rt.Host)
	} else {
		proxy = d.proxyFor(rt.Socket)
	}
	var admissionErr error
	if admit != nil {
		next := proxy
		proxy = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			call, finish, err := admit(r.Context())
			if err != nil {
				admissionErr = errors.Join(hostwrite.ErrUnavailable, err)
				http.Error(w, "instance timer authority changed", http.StatusServiceUnavailable)
				return
			}
			defer finish()
			next.ServeHTTP(w, r.WithContext(call))
		})
	}
	h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		// ServeScheduled, not ServeHTTP: a tick wakes the instance and is
		// held in flight for its duration, but must not reset the idle
		// clock. Reports false for a route it does not manage (not an
		// instance, or not hibernate-flagged); that route is proxied
		// straight, the same call shape the edge itself uses.
		if d.svc == nil || !d.svc.ServeScheduled(w, r, rt, proxy) {
			proxy.ServeHTTP(w, r)
		}
	})
	status, err := deliverVia(ctx, h, rt, req)
	if admissionErr != nil {
		return 0, admissionErr
	}
	if err != nil {
		// Anything that did not come back as an HTTP status — a wake the
		// activator refused, a delivery that ran past its deadline — retires
		// this socket's pool; the next tick dials fresh. Not the case a
		// reader expects here: a socket that will not dial is ReverseProxy's
		// own 502 with no error, and it has nothing pooled to retire anyway.
		// Retired ROUTES are reclaimed by RetainSockets, not from here. A
		// generation's pool is retired by the activator's hook instead.
		if !gen {
			d.dropProxy(rt.Socket)
		}
	}
	if err != nil && ctx.Err() == nil {
		// Redacted for the same reason as the token error: this string is
		// the row's last_status. The context's own error is left alone —
		// the runner matches it with errors.Is to tell a timeout apart.
		err = errors.New(activator.RedactPaths(err.Error()))
	}
	return status, err
}

// deliverVia is the pure core: run the handler against a recorder and turn
// what it wrote into the runner's (status, error) pair.
//
// The mapping is the whole point. The activator answers its own refusals with
// a 503 whose body names the instance ("instance is waking", "instance failed
// to wake", "instance is waking elsewhere"), which is indistinguishable at the
// status line from an app that answered 503 itself. Both retry — but the
// member-facing status must say WHICH, so a wake refusal comes back as an
// error (classified "503: instance is waking…") and an app's own 503 comes
// back as the status 503. Anything else is returned as the app said it.
func deliverVia(ctx context.Context, h http.Handler, rt registry.Route, req *http.Request) (int, error) {
	rec := &statusRecorder{}
	h.ServeHTTP(rec, req)
	// Checked AFTER the handler returns, and before the status: a tick that
	// ran past its deadline gets a 502 written under it by the proxy on the
	// way out, and reporting that 502 would retry a job that may well have
	// completed. The runner maps DeadlineExceeded to ErrTimeout, the one
	// outcome that does not retry.
	if err := ctx.Err(); err != nil {
		return 0, err
	}
	line := rec.firstLine()
	if rec.status == http.StatusServiceUnavailable && strings.HasPrefix(line, "instance") {
		return 0, errors.New(line)
	}
	if rec.status == 0 {
		// The handler returned without writing anything and the context is
		// fine — nothing an instance can legitimately do (even a failed dial
		// is a 502). Reporting status 0 would classify as "app refused" and
		// never retry; this says what actually happened and retries.
		return 0, errors.New("no response from " + rt.Host)
	}
	return rec.status, nil
}

// firstLineMax bounds what the recorder keeps of a body. A tick's response
// body is never delivered anywhere — only the first line of a wake refusal is
// ever read — so a chatty app cannot make this leg hold its output in memory.
const firstLineMax = 256

// statusRecorder is the ResponseWriter the activator and the proxy write
// through: it keeps the status and the first line of the body and discards
// everything else. Not httptest.ResponseRecorder — that one buffers the whole
// body, and this runs in the agent on every tick.
type statusRecorder struct {
	header  http.Header
	status  int
	line    []byte
	lineEnd bool // newline seen or cap reached — later writes are discarded
}

func (r *statusRecorder) Header() http.Header {
	if r.header == nil {
		r.header = http.Header{}
	}
	return r.header
}

func (r *statusRecorder) WriteHeader(code int) {
	if r.status == 0 {
		r.status = code
	}
}

func (r *statusRecorder) Write(b []byte) (int, error) {
	r.WriteHeader(http.StatusOK) // implicit 200, as on a real writer
	if !r.lineEnd {
		if i := bytes.IndexByte(b, '\n'); i >= 0 {
			r.line = append(r.line, b[:i]...)
			r.lineEnd = true
		} else {
			r.line = append(r.line, b...)
		}
		if len(r.line) >= firstLineMax {
			r.line = r.line[:firstLineMax]
			r.lineEnd = true
		}
	}
	return len(b), nil
}

// firstLine is the body's first line, trimmed — what the wake-refusal test
// above matches on.
func (r *statusRecorder) firstLine() string { return strings.TrimSpace(string(r.line)) }