package hostprocess

import (
	"cmp"
	"context"
	"fmt"
	"log/slog"
	"slices"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/fsnotify/fsnotify"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/rest"

	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kcpapply"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kube"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/manifests"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/hostsvc"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kcpserver"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/conds"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/fswatch"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/supervise"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/logcap"
)

const (
	ManifestSet      = "hostprocesses"
	ProviderSet      = "provider"
	DefaultWorkspace = "root"
	RetryAfter       = 2 * time.Second
	MaxRetryAfter    = time.Minute
	MirrorPrefix     = "bootstrap-"
	MirrorLabel      = "socialweb.computer/placement"
	ConditionReady   = "Ready"
	ConditionLimits  = "LimitsEnforced"
)

var GVR = v1alpha1.HostProcessGVR

type Mirror struct {
	Role  string
	Unit  string
	Exec  []string
	Slice string
	After []string
	Wants []string
	Port  int
}

type Config struct {
	Kubeconfig   string
	Workspace    string
	Manager      hostsvc.Manager
	Prefix       string
	Prefixes     []string
	Bound        bool
	Resolve      func(v1alpha1.HostProcessAsset) (string, error)
	Mirrors      func() []Mirror
	ApplyTimeout time.Duration
	Selector     string
	KeepOrphans  bool
	Name         string
	Log          *slog.Logger
}

type Supervisor struct {
	c       Config
	client  kube.Resource
	placed  map[string]bool
	listed  bool
	backend hostsvc.Backend

	mu       sync.Mutex
	locks    sync.Map
	placedAt map[string]time.Time
	keys     map[string]string
	rekeyed  map[string]string
	slices   map[string]bool
	units    map[string]string
	backoff  map[string]time.Duration
	pending  map[string]bool
	pids     map[string]int64
	outputs  map[string]follower
	follows  sync.WaitGroup
	files    *fswatch.Watcher
	changed  chan string
}

type follower struct {
	key  string
	stop context.CancelFunc
}

func (s *Supervisor) track(name, unit string, gone bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if gone {
		for u, n := range s.units {
			if n == name {
				delete(s.units, u)
			}
		}
		return
	}
	if unit == "" {
		return
	}
	if s.units == nil {
		s.units = map[string]string{}
	}
	s.units[unit] = name
}

func (s *Supervisor) objectOf(unit string) (string, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	name, ok := s.units[unit]
	return name, ok
}

func (s *Supervisor) settle(ctx context.Context, name string, retry bool, pid int64) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if !retry {
		delete(s.backoff, name)
		delete(s.pids, name)
		return
	}
	if s.backoff == nil {
		s.backoff, s.pending, s.pids = map[string]time.Duration{}, map[string]bool{}, map[string]int64{}
	}
	if s.pids[name] != pid {
		delete(s.backoff, name)
		s.pids[name] = pid
	}
	if s.changed == nil || s.pending[name] {
		return
	}
	d := min(max(2*s.backoff[name], RetryAfter), MaxRetryAfter)
	s.backoff[name], s.pending[name] = d, true
	ch := s.changed
	time.AfterFunc(d, func() {
		s.mu.Lock()
		delete(s.pending, name)
		s.mu.Unlock()
		select {
		case ch <- name:
		case <-ctx.Done():
		}
	})
}

func (s *Supervisor) setPlaced(unit string, on bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if on {
		s.placed[unit] = true
		if s.placedAt == nil {
			s.placedAt = map[string]time.Time{}
		}
		s.placedAt[unit] = time.Now()
	} else {
		delete(s.placed, unit)
		delete(s.placedAt, unit)
		delete(s.keys, unit)
		delete(s.rekeyed, unit)
	}
}

func (s *Supervisor) setStartedKey(unit, key string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.keys == nil {
		s.keys = map[string]string{}
	}
	s.keys[unit] = key
}

func (s *Supervisor) setRekeyed(unit, key string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.rekeyed == nil {
		s.rekeyed = map[string]string{}
	}
	if key == "" {
		delete(s.rekeyed, unit)
		return
	}
	s.rekeyed[unit] = key
}

func (s *Supervisor) rekeyedFor(unit string) string {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.rekeyed[unit]
}

func (s *Supervisor) startedKey(unit string) (string, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	key, ok := s.keys[unit]
	return key, ok
}

func (s *Supervisor) placedAfter(unit string, t time.Time) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	at, ok := s.placedAt[unit]
	return ok && at.After(t)
}

func (s *Supervisor) isPlaced(unit string) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.placed[unit]
}

func (s *Supervisor) lockObject(name string) func() {
	v, _ := s.locks.LoadOrStore(name, &sync.Mutex{})
	m := v.(*sync.Mutex)
	m.Lock()
	return m.Unlock
}

func (s *Supervisor) markAdopted(o v1alpha1.HostProcess, obs Observed) {
	if o.Spec.Placement == v1alpha1.HostProcessPlacementBootstrap || obs.Inspect != nil || s.isPlaced(o.Spec.Unit) {
		return
	}
	if obs.Unit.State == hostsvc.Active || obs.Unit.State == hostsvc.Activating {
		s.setPlaced(o.Spec.Unit, true)
	}
}

func (s *Supervisor) reconcileObject(ctx context.Context, o v1alpha1.HostProcess) (v1alpha1.HostProcess, bool) {
	unlock := s.lockObject(o.Name)
	defer unlock()
	cur := o.DeepCopy()
	s.track(o.Name, o.Spec.Unit, false)
	s.capOutput(o)
	obs := s.observe(ctx, o)
	s.markAdopted(o, obs)
	r := Reconcile(obs)
	next, failed := s.apply(ctx, o, r)
	err := s.writeStatus(ctx, cur, next)
	if err != nil && !apierrors.IsConflict(err) && !apierrors.IsNotFound(err) {
		s.c.log().Warn("writing status", "name", o.Name, "err", err.Error())
	}
	s.followOutput(ctx, next)
	rt := next.Status.Runtime
	listening := next.Status.Phase == v1alpha1.HostProcessPhaseRunning && rt != nil && rt.Pid > 0 && len(next.Status.Listeners) == 0
	var pid int64
	if listening {
		pid = rt.Pid
	}
	s.settle(ctx, o.Name, failed || r.Unknown || err != nil || listening, pid)
	return o, true
}

func (s *Supervisor) reconcileName(ctx context.Context, name string) {
	cur, err := kube.Get[v1alpha1.HostProcess](ctx, s.client, name)
	if err != nil {
		if !apierrors.IsNotFound(err) {
			s.c.log().Warn("reading a changed HostProcess", "name", name, "err", err.Error())
		}
		return
	}
	s.reconcileObject(ctx, *cur)
}

func (c Config) log() *slog.Logger {
	if c.Log == nil {
		return slog.New(slog.DiscardHandler)
	}
	if c.Name != "" {
		return c.Log.With("component", "hostprocess", "supervisor", c.Name)
	}
	return c.Log.With("component", "hostprocess")
}

func (c Config) prefixes() []string {
	if c.Prefix == "" {
		return c.Prefixes
	}
	return append([]string{c.Prefix}, c.Prefixes...)
}

func (c Config) prefixOf(unit string) string {
	ps := c.prefixes()
	for _, p := range ps {
		if strings.HasPrefix(unit, p) {
			return p
		}
	}
	if len(ps) == 0 {
		return ""
	}
	return ps[0]
}

func (c Config) workspace() string {
	if c.Workspace == "" {
		return DefaultWorkspace
	}
	return c.Workspace
}

func applySets(ctx context.Context, c Config, restCfg *rest.Config) error {
	ap, err := kcpapply.New(restCfg, kcpapply.Options{WaitTimeout: c.ApplyTimeout, Base: c.workspace(), Log: c.log()})
	if err != nil {
		return err
	}
	ix, err := manifests.Load()
	if err != nil {
		return err
	}
	for _, name := range []string{ProviderSet, ManifestSet} {
		set, ok := ix.Set(name)
		if !ok {
			return fmt.Errorf("hostprocess: deploy/sets.yaml names no %s set", name)
		}
		values, err := manifests.DecodeValues(set.Values, nil)
		if err != nil {
			return err
		}
		if _, err := ap.Apply(ctx, set, values); err != nil {
			return fmt.Errorf("hostprocess: applying the %s set: %w", name, err)
		}
	}
	return nil
}

func New(ctx context.Context, c Config) (*Supervisor, error) {
	restCfg, err := kcpserver.FrontRestConfig(c.Kubeconfig)
	if err != nil {
		return nil, err
	}
	if !c.Bound {
		if err := applySets(ctx, c, restCfg); err != nil {
			return nil, err
		}
	}
	cfg := rest.CopyConfig(restCfg)
	cfg.Host = strings.TrimSuffix(cfg.Host, "/") + "/clusters/" + c.workspace()
	kc, err := kube.New(cfg)
	if err != nil {
		return nil, err
	}
	return &Supervisor{c: c, client: kc.Resource(GVR), placed: map[string]bool{}, backend: c.Manager.Capabilities().Backend,
		files: fswatch.New()}, nil
}

func MirrorName(role string) string { return MirrorPrefix + role }

func (s *Supervisor) syncMirrors(ctx context.Context) {
	if s.c.Mirrors == nil {
		return
	}
	for _, m := range s.c.Mirrors() {
		name := MirrorName(m.Role)
		want := v1alpha1.HostProcessSpec{Unit: m.Unit, Placement: v1alpha1.HostProcessPlacementBootstrap, Slice: m.Slice,
			After: m.After, Wants: m.Wants}
		if len(m.Exec) > 0 {
			want.Exec, want.Argv = m.Exec[0], m.Exec[1:]
		}
		cur, err := kube.Get[v1alpha1.HostProcess](ctx, s.client, name)
		if apierrors.IsNotFound(err) {
			o := v1alpha1.HostProcess{ObjectMeta: metav1.ObjectMeta{Name: name, Labels: map[string]string{MirrorLabel: string(v1alpha1.HostProcessPlacementBootstrap)}}, Spec: want}
			if _, err := kube.Create[v1alpha1.HostProcess](ctx, s.client, &o, metav1.CreateOptions{}); err != nil {
				s.c.log().Warn("writing a bootstrap mirror", "name", name, "err", err.Error())
			}
			continue
		}
		if err != nil {
			s.c.log().Warn("reading a bootstrap mirror", "name", name, "err", err.Error())
			continue
		}
		o := *cur
		if o.Spec.Unit != want.Unit || o.Spec.Exec != want.Exec || !slices.Equal(o.Spec.Argv, want.Argv) || o.Spec.Placement != want.Placement {
			o.Spec = want
			if _, err := kube.Update[v1alpha1.HostProcess](ctx, s.client, &o, metav1.UpdateOptions{}); err != nil {
				s.c.log().Warn("updating a bootstrap mirror", "name", name, "err", err.Error())
			}
		}
	}
}

func (s *Supervisor) observe(ctx context.Context, o v1alpha1.HostProcess) Observed {
	m := s.c.Manager
	obs := Observed{Object: o, Backend: s.backend, Prefix: s.c.prefixOf(o.Spec.Unit), Deleting: o.DeletionTimestamp != nil}
	if o.Spec.Placement == v1alpha1.HostProcessPlacementBootstrap {
		obs.Prefix = ""
	}
	obs.Unit, obs.Inspect = m.Inspect(ctx, o.Spec.Unit)
	if obs.Inspect != nil {
		return obs
	}
	if o.Spec.Asset != nil && s.c.Resolve != nil {
		if p, err := s.c.Resolve(*o.Spec.Asset); err == nil {
			obs.Exec = p
		}
	}
	rt := o.Status.Runtime
	if rt != nil && rt.InvocationID != "" && rt.Backend == string(s.backend) && o.Spec.Placement != v1alpha1.HostProcessPlacementBootstrap &&
		o.Status.Phase != v1alpha1.HostProcessPhaseLost {
		obs.StartedKey = s.rekeyedFor(o.Spec.Unit)
		if RestartedInPlace(o, obs.Unit) {
			obs.Restarted = true
			return obs
		}
		want := hostsvc.Identity{Unit: o.Spec.Unit, InvocationID: rt.InvocationID, PID: int(rt.Pid), StartTicks: uint64(rt.StartTicks),
			BootID: rt.HostBootID, Backend: hostsvc.Backend(rt.Backend), Manager: rt.Manager}
		c, err := m.Check(ctx, want)
		if err != nil {
			obs.Inspect = err
			return obs
		}
		obs.Check = &c
	}
	return obs
}

func (s *Supervisor) apply(ctx context.Context, o v1alpha1.HostProcess, r Result) (v1alpha1.HostProcess, bool) {
	m := s.c.Manager
	log := s.c.log().With("hostprocess", o.Name, "unit", o.Spec.Unit)
	started, stopFailed, failed := false, false, false
	stopped := map[string]bool{}
	for _, op := range r.Ops {
		if stopFailed {
			break
		}
		switch op.Kind {
		case OpStart:
			wasPlaced := s.isPlaced(op.Unit)
			s.setPlaced(op.Unit, true)
			st, err := m.Start(ctx, op.Spec)
			if err == nil && st.Adopted && st.State != hostsvc.Active && st.State != hostsvc.Activating {
				log.Info("a loaded unit of this name is not running; resetting it before the start", "state", string(st.State), "result", st.Result)
				if err = m.Stop(ctx, op.Unit); err == nil {
					m.ResetFailed(ctx, op.Unit)
					st, err = m.Start(ctx, op.Spec)
				}
			}
			if err != nil {
				r.Phase, r.Message = v1alpha1.HostProcessPhaseFailed, err.Error()
				failed = true
				log.Warn("start", "err", err.Error())
				if !wasPlaced {
					s.setPlaced(op.Unit, false)
				}
				continue
			}
			s.setPlaced(op.Unit, true)
			s.setStartedKey(op.Unit, o.Spec.RestartKey)
			if stopped[op.Unit] {
				s.setRekeyed(op.Unit, o.Spec.RestartKey)
			} else {
				s.setRekeyed(op.Unit, "")
			}
			started = true
			log.Info("placed", "state", string(st.State), "adopted", st.Adopted, "restartKey", o.Spec.RestartKey)
		case OpStop:
			if err := m.Stop(ctx, op.Unit); err != nil {
				log.Warn("stop", "err", err.Error())
				stopFailed, failed = true, true
				continue
			}
			s.setPlaced(op.Unit, false)
			stopped[op.Unit] = true
			log.Info("stopped")
		}
	}
	next := o
	if r.Unknown {
		next.Status.Message = r.Message
		return next, failed
	}
	next.Status.Phase, next.Status.LostReason, next.Status.Message = r.Phase, r.LostReason, r.Message
	next.Status.ObservedGeneration = o.Generation
	if started {
		next.Status.RestartKey = o.Spec.RestartKey
	} else if key, ok := s.startedKey(o.Spec.Unit); ok {
		next.Status.RestartKey = key
	}
	if r.Identify {
		if id, err := m.Identify(ctx, o.Spec.Unit, 0); err == nil {
			st, _ := m.Inspect(ctx, o.Spec.Unit)
			next.Status.Runtime = &v1alpha1.HostProcessRuntime{Backend: string(id.Backend), Unit: id.Unit, InvocationID: id.InvocationID,
				Pid: int64(id.PID), StartTicks: int64(id.StartTicks), HostBootID: id.BootID, Manager: id.Manager}
			if !st.ActiveSince.IsZero() {
				next.Status.Runtime.ActiveSince = st.ActiveSince.UTC().Format(time.RFC3339Nano)
			}
			if o.Spec.Placement == v1alpha1.HostProcessPlacementSupervisor {
				s.setPlaced(o.Spec.Unit, true)
			}
		}
	}
	if next.Status.Runtime != nil && next.Status.Runtime.Pid > 0 && next.Status.Phase == v1alpha1.HostProcessPhaseRunning {
		next.Status.Listeners = listenersOf(int(next.Status.Runtime.Pid))
	} else {
		next.Status.Listeners = nil
	}
	if r.Phase == v1alpha1.HostProcessPhaseStopped || r.Phase == v1alpha1.HostProcessPhaseFailed || r.Phase == v1alpha1.HostProcessPhaseLost {
		if st, err := m.Inspect(ctx, o.Spec.Unit); err == nil && st.State != hostsvc.NotFound {
			next.Status.LastResult = &v1alpha1.HostProcessResult{Result: st.Result, ExitCode: st.ExecMainStatus,
				At: time.Now().UTC().Format(time.RFC3339)}
		}
	}
	conds.SetOn(&next, ConditionReady, next.Status.Phase == v1alpha1.HostProcessPhaseRunning, string(next.Status.Phase), next.Status.Message)
	caps := m.Capabilities()
	conds.SetOn(&next, ConditionLimits, caps.EnforcesLimits, string(caps.Backend), "limits are enforced only where the backend has cgroups")
	return next, failed
}

func (s *Supervisor) writeStatus(ctx context.Context, cur *v1alpha1.HostProcess, next v1alpha1.HostProcess) error {
	next.ResourceVersion = cur.ResourceVersion
	was, err := kube.Encode(cur)
	if err != nil {
		return err
	}
	u, err := kube.Encode(&next)
	if err != nil {
		return err
	}
	if equalStatus(was["status"], u["status"]) {
		return nil
	}
	_, err = kube.UpdateStatus[v1alpha1.HostProcess](ctx, s.client, u, metav1.UpdateOptions{})
	return err
}

func equalStatus(a, b any) bool { return fmt.Sprint(a) == fmt.Sprint(b) }

func (s *Supervisor) Pass(ctx context.Context) error {
	s.syncMirrors(ctx)
	listedAt := time.Now()
	list, err := kube.List[v1alpha1.HostProcess](ctx, s.client, metav1.ListOptions{LabelSelector: s.c.Selector})
	if err != nil {
		return err
	}
	seen := map[string]bool{}
	var liveSlices []string
	for i := range list {
		if o, ok := s.reconcileObject(ctx, list[i]); ok {
			seen[o.Spec.Unit] = true
			liveSlices = append(liveSlices, o.Spec.Slice)
		}
	}
	for _, prefix := range s.c.prefixes() {
		if s.c.KeepOrphans {
			break
		}
		units, err := s.c.Manager.List(ctx, prefix)
		if err == nil {
			for _, u := range units {
				if !seen[u.Name] && u.State == hostsvc.Failed && !s.placedAfter(u.Name, listedAt) {
					s.c.log().Info("the HostProcess of a failed unit is gone; resetting the unit", "unit", u.Name)
					if err := s.c.Manager.ResetFailed(ctx, u.Name); err != nil {
						s.c.log().Warn("resetting the failed unit of a gone HostProcess failed; the next pass retries", "unit", u.Name, "err", err.Error())
					} else {
						s.setPlaced(u.Name, false)
					}
					continue
				}
				if !s.isPlaced(u.Name) && s.listed {
					continue
				}
				if !seen[u.Name] && s.placedAfter(u.Name, listedAt) {
					continue
				}
				if !seen[u.Name] && (u.State == hostsvc.Active || u.State == hostsvc.Activating) {
					s.c.log().Info("the HostProcess of a placed unit is gone; stopping the unit", "unit", u.Name)
					if err := s.c.Manager.Stop(ctx, u.Name); err != nil {
						s.c.log().Warn("stopping the unit of a gone HostProcess failed; the next pass retries", "unit", u.Name, "err", err.Error())
					} else {
						s.setPlaced(u.Name, false)
					}
				}
			}
		}
	}
	s.listed = true
	if !s.c.KeepOrphans {
		s.sweepSlices(ctx, liveSlices)
	}
	return nil
}

const ClusterSlicePart = "cluster-"

func (s *Supervisor) sweepable(slice string) bool {
	if hostsvc.ValidateSlice(slice) != nil {
		return false
	}
	for _, p := range s.c.prefixes() {
		if strings.HasPrefix(slice, p+ClusterSlicePart) {
			return true
		}
	}
	return false
}

func (s *Supervisor) sweepSlices(ctx context.Context, live []string) {
	s.mu.Lock()
	if s.slices == nil {
		s.slices = map[string]bool{}
	}
	for _, sl := range live {
		if s.sweepable(sl) {
			s.slices[sl] = true
		}
	}
	var empty []string
	for sl := range s.slices {
		held := false
		for _, l := range live {
			if hostsvc.SliceContains(sl, l) {
				held = true
				break
			}
		}
		if !held {
			empty = append(empty, sl)
		}
	}
	s.mu.Unlock()
	for _, sl := range empty {
		if err := s.c.Manager.StopSlice(ctx, sl); err != nil {
			s.c.log().Warn("stopping the empty slice of gone HostProcesses failed; the next pass retries", "slice", sl, "err", err.Error())
			continue
		}
		s.mu.Lock()
		delete(s.slices, sl)
		s.mu.Unlock()
		s.c.log().Info("every HostProcess of the slice is gone; the empty slice is stopped", "slice", sl)
	}
}

func outputFiles(o v1alpha1.HostProcess) ([]string, int64) {
	out := o.Spec.Output
	if out == nil {
		return nil, 0
	}
	var files []string
	for _, f := range []string{out.Stdout, out.Stderr} {
		if f != "" && !slices.Contains(files, f) {
			files = append(files, f)
		}
	}
	return files, cmp.Or(out.CapBytes, logcap.DefaultCap)
}

const OutputDebounce = time.Second

func (s *Supervisor) followOutput(ctx context.Context, o v1alpha1.HostProcess) {
	files, capBytes := outputFiles(o)
	key := ""
	if o.DeletionTimestamp == nil && o.Status.Phase == v1alpha1.HostProcessPhaseRunning && len(files) > 0 {
		key = strings.Join(files, "\n") + "\n" + strconv.FormatInt(capBytes, 10)
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if cur, ok := s.outputs[o.Name]; ok {
		if cur.key == key {
			return
		}
		cur.stop()
		delete(s.outputs, o.Name)
	}
	if key == "" {
		return
	}
	w, err := s.files.Subscribe(files...)
	if err != nil {
		s.c.log().Warn("output files are capped only at reconcile: watching an output file failed", "name", o.Name, "files", strings.Join(files, ", "), "err", err.Error())
		return
	}
	fctx, cancel := context.WithCancel(ctx)
	if s.outputs == nil {
		s.outputs = map[string]follower{}
	}
	s.outputs[o.Name] = follower{key: key, stop: cancel}
	s.follows.Go(func() { s.rotateOnWrite(fctx, w, o, key) })
}

func (s *Supervisor) unfollowOutput(name string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if cur, ok := s.outputs[name]; ok {
		cur.stop()
		delete(s.outputs, name)
	}
}

func (s *Supervisor) rotateOnWrite(ctx context.Context, w *fswatch.Sub, o v1alpha1.HostProcess, key string) {
	defer w.Close()
	defer func() {
		s.mu.Lock()
		defer s.mu.Unlock()
		if cur, ok := s.outputs[o.Name]; ok && cur.key == key && ctx.Err() == nil {
			cur.stop()
			delete(s.outputs, o.Name)
		}
	}()
	for {
		select {
		case <-ctx.Done():
			return
		case _, ok := <-w.Errors():
			if !ok {
				return
			}
		case ev, ok := <-w.Events():
			if !ok || ev.Has(fsnotify.Remove) || ev.Has(fsnotify.Rename) {
				return
			}
		}
		unlock := s.lockObject(o.Name)
		s.capOutput(o)
		unlock()
		select {
		case <-ctx.Done():
			return
		case <-time.After(OutputDebounce):
		}
	}
}

func (s *Supervisor) capOutput(o v1alpha1.HostProcess) {
	files, capBytes := outputFiles(o)
	for _, f := range files {
		ok, err := logcap.Rotate(f, capBytes)
		if err != nil {
			s.c.log().Warn("an output file over its cap was not rotated", "name", o.Name, "file", f, "err", err.Error())
		}
		if ok {
			s.c.log().Info("an output file over its cap was rotated; it keeps its newest bytes in a .1 copy", "name", o.Name, "file", f, "cap", capBytes)
		}
	}
}

func (s *Supervisor) Run(ctx context.Context) error {
	s.c.log().Info("the HostProcess supervisor runs in the core", "workspace", s.c.workspace(), "backend", string(s.backend),
		"prefixes", s.c.prefixes())
	defer s.files.Close()
	defer s.follows.Wait()
	wake := make(chan struct{}, 1)
	changed := make(chan string, 1024)
	s.mu.Lock()
	s.changed = changed
	s.mu.Unlock()
	var wg sync.WaitGroup
	defer wg.Wait()
	for range EventWorkers {
		wg.Go(func() {
			for {
				select {
				case <-ctx.Done():
					return
				case name := <-changed:
					s.reconcileName(ctx, name)
				}
			}
		})
	}
	wg.Go(func() { s.watch(ctx, wake, changed) })
	for _, prefix := range s.c.prefixes() {
		wg.Go(func() { s.watchUnits(ctx, prefix, wake, changed) })
	}
	for {
		if err := s.Pass(ctx); err != nil {
			s.c.log().Warn("a reconcile pass failed; units keep running and the pass retries", "err", err.Error(), "after", RetryAfter.String())
			time.AfterFunc(RetryAfter, func() { poke(wake) })
		}
		select {
		case <-ctx.Done():
			return nil
		case <-wake:
			select {
			case <-ctx.Done():
				return nil
			case <-time.After(WakeDebounce):
			}
			select {
			case <-wake:
			default:
			}
		}
	}
}

const WakeDebounce = 100 * time.Millisecond

const EventWorkers = 4

func poke(wake chan<- struct{}) {
	select {
	case wake <- struct{}{}:
	default:
	}
}

func (s *Supervisor) watch(ctx context.Context, wake chan<- struct{}, changedCh chan<- string) {
	generations := map[string]int64{}
	failing := false
	for ctx.Err() == nil {
		events, stop, err := kube.Watch[v1alpha1.HostProcess](ctx, s.client, metav1.ListOptions{LabelSelector: s.c.Selector})
		if err != nil {
			if !failing {
				s.c.log().Warn("watching HostProcess objects failed; the supervisor retries the watch", "err", err.Error(), "after", RetryAfter.String())
				failing = true
			}
			select {
			case <-ctx.Done():
				return
			case <-time.After(RetryAfter):
			}
			continue
		}
		if failing {
			s.c.log().Info("watching HostProcess objects again")
			failing = false
		}
		poke(wake)
		for ev := range events {
			if ev.Object == nil {
				continue
			}
			name, gen, unit := ev.Object.Name, ev.Object.Generation, ev.Object.Spec.Unit
			s.track(name, unit, ev.Type == watch.Deleted)
			if ev.Type == watch.Deleted {
				s.unfollowOutput(name)
			}
			changed := ev.Type != watch.Modified || generations[name] != gen
			if ev.Type == watch.Deleted {
				delete(generations, name)
			} else {
				generations[name] = gen
			}
			if !changed {
				continue
			}
			if ev.Type != watch.Deleted {
				select {
				case changedCh <- name:
					continue
				default:
				}
			}
			poke(wake)
		}
		stop()
		select {
		case <-ctx.Done():
			return
		case <-time.After(RetryAfter):
		}
	}
}

func (s *Supervisor) watchUnits(ctx context.Context, prefix string, wake chan<- struct{}, changed chan<- string) {
	failing := false
	for ctx.Err() == nil {
		events, err := s.c.Manager.Watch(ctx, prefix)
		if err != nil {
			if !failing {
				s.c.log().Warn("watching units failed; the supervisor retries the watch", "prefix", prefix, "err", err.Error(), "after", RetryAfter.String())
				failing = true
			}
			select {
			case <-ctx.Done():
				return
			case <-time.After(RetryAfter):
			}
			continue
		}
		if failing {
			s.c.log().Info("watching units again", "prefix", prefix)
			failing = false
		}
		poke(wake)
		for ev := range events {
			name, ok := s.objectOf(ev.Name)
			if !ok {
				if !s.c.KeepOrphans {
					poke(wake)
				}
				continue
			}
			select {
			case changed <- name:
			default:
				poke(wake)
			}
		}
		select {
		case <-ctx.Done():
			return
		case <-time.After(RetryAfter):
		}
	}
}

func Supervise(ctx context.Context, c Config) {
	log := c.log()
	supervise.Restart(ctx, "hostprocess", supervise.Backoff{Min: RetryAfter, Max: RetryAfter, OnError: func(err error, after time.Duration) {
		log.Error("the HostProcess supervisor stopped; starting it again", "err", fmt.Sprint(err), "after", after.String())
	}}, func(ctx context.Context) error {
		s, err := New(ctx, c)
		if err != nil {
			return err
		}
		return s.Run(ctx)
	})
}

func listenersOf(pid int) []v1alpha1.HostProcessListener {
	ports, err := hostsvc.ListenPorts(pid)
	if err != nil || len(ports) == 0 {
		return nil
	}
	if len(ports) == 1 {
		return []v1alpha1.HostProcessListener{{Name: "tcp", Address: "127.0.0.1:" + strconv.Itoa(ports[0])}}
	}
	out := make([]v1alpha1.HostProcessListener, 0, len(ports))
	for _, p := range ports {
		out = append(out, v1alpha1.HostProcessListener{Name: "tcp-" + strconv.Itoa(p), Address: "127.0.0.1:" + strconv.Itoa(p)})
	}
	return out
}