package hostexec

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"github.com/carlosframework/platform/internal/version"
	"os"
	"runtime"
	"slices"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostfloor"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostquarantine"
	"github.com/carlosframework/platform/internal/release"
	"path/filepath"
)

type Config struct {
	Binding        Binding
	Store          release.Store
	Guard          hostguard.Guard
	Directory      hostquarantine.ActorDirectory
	Clock          Clock
	UpdateGuardDir string
	UpdateJournal  string
	verifyRunning  func(context.Context) error
	LockDir        string
	// These close/cancel hooks must be nonblocking and safe to call repeatedly.
	CloseAdmission func()
	CancelWork     func()
	// Cleanup terminates only this host's processes and closes its replicators.
	// It must not wait for the host guard: uncertain disk writes retain that
	// guard to deny other processes, while termination must proceed independently.
	// Its failure remains a separate denial owned by the caller, even after CC9.
	Cleanup    func(context.Context) error
	writeLocal func(*localDir, State) error // test seam; nil uses durable descriptor-relative I/O
}

type Controller struct {
	cfg         Config
	mu          sync.Mutex
	state       State
	err         error
	cleanupErr  error
	releaseETag string
	cleanupOnce sync.Once
	cleanupDone chan struct{}
	ready       chan struct{}
	readyOnce   sync.Once
	lost        chan struct{}
	ctx         context.Context
	cancel      context.CancelFunc
	done        chan struct{}
	watchDone   chan struct{}
	unlock      func()
	unlive      func()
	closeMu     sync.Mutex
	closed      bool
}

func Open(ctx context.Context, cfg Config) (*Controller, error) {
	if cfg.Clock == nil {
		cfg.Clock = SystemClock{}
	}
	if cfg.LockDir == "" {
		if cfg.UpdateGuardDir == "" {
			cfg.UpdateGuardDir = "/opt/carlos/system/guard"
			cfg.UpdateJournal = filepath.Join("/opt/carlos/system/router", "journal.json")
		}
		cfg.LockDir = "/run/carlos/hostexec"
	}
	var nonce [16]byte
	if _, err := rand.Read(nonce[:]); err != nil {
		return nil, err
	}
	cfg.Binding.ControllerID = hex.EncodeToString(nonce[:])
	if !cfg.Binding.valid() || cfg.Store == nil || cfg.Guard.Claims == nil || cfg.Directory == nil ||
		cfg.CloseAdmission == nil || cfg.CancelWork == nil || cfg.Cleanup == nil {
		return nil, ErrState
	}
	wall, boot, err := cfg.Clock.Now()
	if err != nil {
		return nil, err
	}
	process, err := currentControllerProcess()
	if err != nil {
		return nil, err
	}
	cctx, cancel := context.WithCancel(ctx)
	c := &Controller{cfg: cfg, state: State{Process: process, V: 1, Binding: cfg.Binding, State: "pending", Expires: wall.Add(TTL), SafeUntil: boot + int64(TTL-Margin)}, ctx: cctx, cancel: cancel,
		ready: make(chan struct{}), lost: make(chan struct{}), done: make(chan struct{}), watchDone: make(chan struct{}), cleanupDone: make(chan struct{})}
	// Hold the host guard across lifetime-lock acquisition and the pending
	// write. Acquire the live lock only after pending state is durable, so
	// orphaned active bytes cannot regain admission. Never wait for the
	// lifetime lock while holding the guard.
	var cleaned <-chan error
	pendingWritten := false
	for {
		err = cfg.Guard.WithLocal(ctx, cfg.Binding.Host, func() error {
			var lockErr error
			c.unlock, lockErr = tryAcquire(cfg.LockDir, cfg.Binding.Host)
			if lockErr != nil {
				return lockErr
			}
			// Start cleanup before any persistence can block. Exclusion stays
			// held until this independent cleanup has joined on every exit.
			result := make(chan error, 1)
			cleaned = result
			go func() { result <- cfg.Cleanup(context.Background()) }()
			cfg.CloseAdmission()
			cfg.CancelWork()
			d, openErr := openLocal(cfg.Guard.DataDir)
			if openErr != nil {
				return openErr
			}
			defer d.close()
			if writeErr := c.write(d, c.state); writeErr != nil {
				return writeErr
			}
			pendingWritten = true
			if cfg.UpdateGuardDir != "" {
				if err := cfg.checkUpdateIdentity(ctx); err != nil {
					return err
				}
			}
			c.unlive, lockErr = tryAcquireLive(cfg.LockDir, cfg.Binding.Host)
			return lockErr
		})
		if c.unlock != nil || !errors.Is(err, ErrControllerBusy) {
			break
		}
		select {
		case <-ctx.Done():
			err = ctx.Err()
		case <-time.After(10 * time.Millisecond):
			continue
		}
		break
	}
	if err == nil {
		err = <-cleaned
		cleaned = nil
	}
	if err == nil {
		err = c.authority(ctx)
	}
	if err != nil {
		pending := c.state
		c.state.State = "lost"
		hadLive := c.unlive != nil
		if c.unlive != nil {
			c.unlive()
			c.unlive = nil
		}
		if c.unlock != nil {
			// Before live-lock acquisition, no local state can authorize work.
			// A failed pending write (including an unsafe path) must not make
			// cleanup wait for another write to the same failed destination.
			if hadLive {
				err = errors.Join(err, c.save(c.state))
			}
			cleanupErr := error(nil)
			if cleaned != nil {
				cleanupErr = <-cleaned
				err = errors.Join(err, cleanupErr)
			}
			// No live lock or remote authority was issued. After positive joined
			// cleanup, abort only this Open's exact pending tuple while retaining
			// lifetime exclusion. Failed/uncertain cleanup retains its denial.
			if !hadLive && pendingWritten && cleanupErr == nil {
				err = errors.Join(err, cfg.Guard.WithLocal(context.Background(), pending.Host, func() error {
					d, e := openLocal(cfg.Guard.DataDir)
					if e != nil {
						return e
					}
					defer d.close()
					current, e := d.read(pending.Host)
					if e != nil {
						return e
					}
					// JSON retains the instant, not time.Time's monotonic/location
					// metadata. Compare that field semantically, all others exactly.
					if !current.Expires.Equal(pending.Expires) {
						return ErrState
					}
					current.Expires = pending.Expires
					if current != pending {
						return ErrState
					}
					return d.remove(pending.Host)
				}))
			}
			c.unlock()
		}
		cancel()
		return nil, err
	}
	go c.watch()
	go c.run()
	return c, nil
}

func (c *Controller) write(d *localDir, s State) error {
	if c.cfg.writeLocal != nil {
		return c.cfg.writeLocal(d, s)
	}
	return d.write(s)
}

// save retains the guard until lost state is durable, including failures before
// rename. The revoked live lock denies CheckLocal while the old file survives.
func (c *Controller) save(s State) error {
	return c.cfg.Guard.WithLocal(context.Background(), s.Host, func() error {
		for {
			d, err := openLocal(c.cfg.Guard.DataDir)
			if err == nil {
				err = c.write(d, s)
				d.close()
			}
			if err == nil || s.State != "lost" {
				return err
			}
			time.Sleep(100 * time.Millisecond)
		}
	})
}
func (c *Controller) authority(ctx context.Context) error {
	if err := c.cfg.checkUpdateIdentity(ctx); err != nil {
		return err
	}
	if err := c.cfg.checkRemoteUpdateIdentity(ctx); err != nil {
		return err
	}
	b := c.cfg.Binding
	if err := hostquarantine.Check(ctx, c.cfg.Directory, b.Box, b.Enrollment); err != nil {
		return err
	}
	obs, err := c.cfg.Guard.Claims.Read(ctx, b.Host)
	if err != nil {
		return err
	}
	if obs.Absent || obs.Claim == nil || obs.Host != b.Host || obs.Claim.Host != b.Host {
		return ErrInactive
	}
	q := obs.Claim
	if q.Validate() != nil || q.Holder == nil || q.Holder.Incarnation != b.Incarnation || q.Op.ID != b.Op ||
		q.Op.Kind == hostclaim.KindServing || q.Free() || !slices.Contains(q.Op.Participants, b.Box) || slices.Contains(q.Lost, b.Box) {
		return ErrInactive
	}
	return nil
}

// loseLocked latches denial before attempting disk I/O or process cleanup.
func (c *Controller) loseLocked(err error) {
	if c.err != nil {
		return
	}
	c.err = err
	c.state.State = "lost"
	// Revoke cross-process admission before publishing loss or starting I/O.
	// This close needs no disk space. The lifetime lock remains held.
	if c.unlive != nil {
		c.unlive()
		c.unlive = nil
	}
	close(c.lost)
	c.cancel()
	c.cfg.CloseAdmission()
	c.cfg.CancelWork()
	c.readyOnce.Do(func() { close(c.ready) })
	s := c.state
	c.cleanupOnce.Do(func() {
		go func() {
			// Neither a blocked disk nor blocked cleanup can prevent the other.
			cleaned := make(chan error, 1)
			go func() { cleaned <- c.cfg.Cleanup(context.Background()) }()
			persistErr := c.save(s)
			cleanupErr := <-cleaned
			c.mu.Lock()
			c.cleanupErr = errors.Join(persistErr, cleanupErr)
			c.mu.Unlock()
			close(c.cleanupDone)
		}()
	})
}
func (c *Controller) checkLocked() error {
	if c.err != nil {
		return c.err
	}
	_, boot, err := c.cfg.Clock.Now()
	if err != nil {
		c.loseLocked(err)
	} else if boot >= c.state.SafeUntil {
		c.loseLocked(ErrLost)
	}
	return c.err
}
func (c *Controller) Lost() <-chan struct{} { return c.lost }
func (c *Controller) Ready(ctx context.Context) (State, error) {
	select {
	case <-ctx.Done():
		return State{}, ctx.Err()
	case <-c.ready:
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	if err := c.checkLocked(); err != nil {
		return State{}, err
	}
	if c.state.State != "active" {
		return State{}, ErrState
	}
	return c.state, nil
}
func (c *Controller) CleanupError() error { c.mu.Lock(); defer c.mu.Unlock(); return c.cleanupErr }

func (c *Controller) watch() {
	defer close(c.watchDone)
	for {
		c.mu.Lock()
		err := c.checkLocked()
		c.mu.Unlock()
		if err != nil {
			return
		}
		// CLOCK_BOOTTIME is sampled on every wake and every admission. A normal Go
		// timer does not itself prove that a suspended machine has remained stopped.
		select {
		case <-c.ctx.Done():
			c.mu.Lock()
			c.loseLocked(c.ctx.Err())
			c.mu.Unlock()
			return
		case <-c.cfg.Clock.After(100 * time.Millisecond):
		}
	}
}
func (c *Controller) run() {
	defer close(c.done)
	key, _ := Key(c.cfg.Binding.Host, c.cfg.Binding.Box)
	body, etag, err := c.cfg.Store.Get(c.ctx, key)
	if errors.Is(err, os.ErrNotExist) {
		err = nil
		etag = ""
	} else if err == nil {
		var r Record
		if decode(body, &r) != nil || r.Op == "" || r.Expires.IsZero() || etag == "" {
			err = ErrState
		}
	}
	if err != nil {
		c.mu.Lock()
		c.loseLocked(err)
		c.mu.Unlock()
		return
	}
	for {
		if err = c.authority(c.ctx); err != nil {
			c.mu.Lock()
			c.loseLocked(err)
			c.mu.Unlock()
			return
		}
		wall, start, clockErr := c.cfg.Clock.Now()
		if clockErr != nil {
			c.mu.Lock()
			c.loseLocked(clockErr)
			c.mu.Unlock()
			return
		}
		c.mu.Lock()
		err = c.checkLocked()
		captured := c.state.Binding
		c.mu.Unlock()
		if err != nil {
			return
		}
		proposed := Record{Op: captured.Op, Expires: wall.Add(TTL)}
		bytes, _ := json.Marshal(proposed)
		next, putErr := c.cfg.Store.Put(c.ctx, key, bytes, etag)
		c.mu.Lock()
		// A late acknowledgement can only contribute its exact ETag to release.
		if captured != c.state.Binding {
			c.mu.Unlock()
			return
		}
		if putErr == nil && next != "" {
			c.releaseETag = next
			etag = next
		}
		if err = c.checkLocked(); err == nil && (putErr != nil || next == "") {
			if putErr == nil {
				putErr = ErrState
			}
			c.loseLocked(putErr)
			err = putErr
		}
		c.mu.Unlock()
		if err != nil {
			return
		}
		// Network reads never hold either the controller mutex or the host guard.
		if err = c.authority(c.ctx); err != nil {
			c.mu.Lock()
			c.loseLocked(err)
			c.mu.Unlock()
			return
		}
		c.mu.Lock()
		if err = c.checkLocked(); err != nil {
			c.mu.Unlock()
			return
		}
		nextState := c.state
		nextState.State = "active"
		nextState.LeaseETag = next
		nextState.Expires = proposed.Expires
		nextState.SafeUntil = start + int64(TTL-Margin)
		// Never use a response-relative deadline, including when the wall clock moved.
		_, now, clockErr := c.cfg.Clock.Now()
		if clockErr != nil || now >= nextState.SafeUntil {
			c.loseLocked(ErrLost)
			c.mu.Unlock()
			return
		}
		c.mu.Unlock()
		// Publication and any intervening loss are serialized under the same host
		// guard, so another process never observes a stale active overwrite.
		if err = c.publish(nextState); err != nil {
			c.mu.Lock()
			c.loseLocked(err)
			c.mu.Unlock()
			return
		}
		_, now, clockErr = c.cfg.Clock.Now()
		if clockErr != nil {
			c.mu.Lock()
			c.loseLocked(clockErr)
			c.mu.Unlock()
			return
		}
		delay := time.Duration(start + int64(Renewal) - now)
		if delay < 0 {
			delay = 0
		}
		select {
		case <-c.ctx.Done():
			return
		case <-c.cfg.Clock.After(delay):
		}
	}
}

// publish leaves mu available to the watchdog during filesystem calls. If a
// write has an uncertain outcome, it retains the host guard until denial is
// durable; releasing it over possibly active bytes would admit another process.
func (c *Controller) publish(next State) error {
	return c.cfg.Guard.WithLocal(context.Background(), next.Host, func() error {
		d, err := openLocal(c.cfg.Guard.DataDir)
		if err != nil {
			return err
		}
		defer d.close()
		c.mu.Lock()
		err = c.checkLocked()
		if err == nil && next.Binding != c.state.Binding {
			c.loseLocked(ErrState)
			err = ErrState
		}
		c.mu.Unlock()
		if err != nil {
			return err
		}
		writeErr := c.write(d, next)
		c.mu.Lock()
		if writeErr != nil {
			c.loseLocked(writeErr)
		}
		err = c.checkLocked()
		if err == nil {
			_, boot, clockErr := c.cfg.Clock.Now()
			if clockErr != nil || boot >= next.SafeUntil {
				c.loseLocked(ErrLost)
				err = c.err
			}
		}
		if err == nil {
			c.state = next
			c.readyOnce.Do(func() { close(c.ready) })
			c.mu.Unlock()
			return nil
		}
		denied := c.state
		denied.State = "lost"
		c.mu.Unlock()
		// Cancellation does not release the guard over uncertain active bytes.
		// Close may time out; it keeps the lifetime lock until this join completes.
		for {
			if persistErr := c.write(d, denied); persistErr == nil {
				return err
			}
			time.Sleep(100 * time.Millisecond)
		}
	})
}

// Close cancels and joins all lease requests before conditional release. It
// retains the lifetime lock on a timed-out join; call Close again to finish.
func (c *Controller) Close(ctx context.Context) error {
	c.closeMu.Lock()
	defer c.closeMu.Unlock()
	if c.closed {
		return nil
	}
	c.mu.Lock()
	c.loseLocked(ErrLost)
	c.mu.Unlock()
	for _, done := range []<-chan struct{}{c.done, c.watchDone, c.cleanupDone} {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-done:
		}
	}
	c.mu.Lock()
	etag := c.releaseETag
	c.mu.Unlock()
	var err error
	if etag != "" {
		key, _ := Key(c.cfg.Binding.Host, c.cfg.Binding.Box)
		err = release.DeleteIfMatch(ctx, c.cfg.Store, key, etag)
	}
	c.closed = true
	c.unlock()
	return errors.Join(err, c.CleanupError())
}

// CheckLocal must run inside Guard.With, before each bounded host write. It
// does not grant authority and never clears quarantine or cleanup denial.
func CheckLocal(dataDir string, expected Binding, clock Clock, lockDir string) error {
	if !expected.valid() || clock == nil {
		return ErrState
	}
	if lockDir == "" {
		lockDir = "/run/carlos/hostexec"
	}
	// Guard.With serializes these probes with other writers and Open. A free
	// live lock denies even if ENOSPC left a future-dated active file behind.
	for _, probe := range []func(string, string) (func(), error){probeLifetime, probeLive} {
		unlock, err := probe(lockDir, expected.Host)
		if err == nil {
			unlock()
			return ErrLost
		}
		if errors.Is(err, os.ErrNotExist) {
			return ErrLost
		}
		if !errors.Is(err, ErrControllerBusy) {
			return err
		}
	}
	d, err := openLocal(dataDir)
	if err != nil {
		return err
	}
	defer d.close()
	s, err := d.read(expected.Host)
	if err != nil {
		return err
	}
	if s.Binding != expected || s.State != "active" {
		return ErrLost
	}
	_, boot, err := clock.Now()
	if err != nil {
		return err
	}
	if boot >= s.SafeUntil {
		return ErrLost
	}
	return nil
}

// ClearStale performs CC9 with the lifetime lock held, after any old controller
// has joined. A failed or mismatched claim read leaves local denial intact.
func ClearStale(ctx context.Context, cfg Config) error {
	lockDir := cfg.LockDir
	if lockDir == "" {
		lockDir = "/run/carlos/hostexec"
	}
	if cfg.Guard.Claims == nil || !validID(cfg.Binding.Box) || !validID(cfg.Binding.Enrollment) {
		return ErrState
	}
	var unlock func()
	var err error
	for {
		err = cfg.Guard.WithLocal(ctx, cfg.Binding.Host, func() error {
			var lockErr error
			unlock, lockErr = tryAcquire(lockDir, cfg.Binding.Host)
			if lockErr != nil {
				return lockErr
			}
			denied := false
			defer func() {
				if !denied {
					unlock()
					unlock = nil
				}
			}()
			d, openErr := openLocal(cfg.Guard.DataDir)
			if openErr != nil {
				return openErr
			}
			defer d.close()
			old, readErr := d.read(cfg.Binding.Host)
			if errors.Is(readErr, os.ErrNotExist) {
				denied = true
				return nil
			}
			if readErr != nil {
				return readErr
			}
			if old.Box != cfg.Binding.Box || old.Enrollment != cfg.Binding.Enrollment ||
				old.Op != cfg.Binding.Op || old.Incarnation != cfg.Binding.Incarnation {
				return ErrState
			}
			// Holding the lifetime lock must never make an orphaned active record
			// appear to belong to a running controller during the claim read.
			old.State = "lost"
			writeErr := d.write(old)
			denied = writeErr == nil
			return writeErr
		})
		if !errors.Is(err, ErrControllerBusy) {
			break
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(10 * time.Millisecond):
		}
	}
	if err != nil {
		return err
	}
	defer unlock()
	obs, err := cfg.Guard.Claims.Read(ctx, cfg.Binding.Host)
	if err != nil {
		return err
	}
	if obs.Absent || obs.Claim == nil || obs.Host != cfg.Binding.Host || obs.Claim.Host != cfg.Binding.Host || obs.Claim.Validate() != nil {
		return ErrState
	}
	q := obs.Claim
	return cfg.Guard.WithLocal(ctx, cfg.Binding.Host, func() error {
		d, err := openLocal(cfg.Guard.DataDir)
		if err != nil {
			return err
		}
		defer d.close()
		old, err := d.read(cfg.Binding.Host)
		if errors.Is(err, os.ErrNotExist) {
			return nil
		}
		if err != nil {
			return err
		}
		if old.Box != cfg.Binding.Box || old.Enrollment != cfg.Binding.Enrollment ||
			old.Op != cfg.Binding.Op || old.Incarnation != cfg.Binding.Incarnation {
			return ErrState
		}
		// Only the caller's expected operation is cleared. A fresh successor
		// operation may exclude this box; that is also CC9 discharge evidence.
		if !q.Free() && q.Op.Kind != hostclaim.KindServing &&
			slices.Contains(q.Op.Participants, old.Box) {
			return ErrInactive
		}
		return d.remove(old.Host)
	})
}

// checkUpdateIdentity reads the executing inode, never the replaceable current
// path. Run it after pending is durable and before each acknowledgement.
func (cfg Config) checkUpdateIdentity(ctx context.Context) error {
	if cfg.UpdateGuardDir == "" {
		return nil
	}
	if err := hostfloor.UpdateIdle(cfg.UpdateGuardDir, cfg.UpdateJournal); err != nil {
		return err
	}
	if cfg.verifyRunning != nil {
		return cfg.verifyRunning(ctx)
	}
	pub, err := release.ParsePublicKey(os.Getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
	if err != nil {
		return err
	}
	d := hostfloor.Disk{Dir: cfg.UpdateGuardDir, Bucket: os.Getenv("CARLOS_DEPLOYMENT_BUCKET"), PublicKey: pub}
	running, err := d.Running(version.Build, runtime.GOOS+"-"+runtime.GOARCH)
	if err != nil || !running.GuardReady {
		if err != nil {
			return err
		}
		return hostfloor.ErrFloor
	}
	_, floor, err := d.Load()
	if err != nil || running.FloorSHA256 != floor.Digest {
		return hostfloor.ErrFloor
	}
	return hostfloor.UpdateIdle(cfg.UpdateGuardDir, cfg.UpdateJournal)
}

// authority calls this outside the controller mutex and host guard. Recheck
// the executing identity against the exact local digest compared remotely.
func (cfg Config) checkRemoteUpdateIdentity(ctx context.Context) error {
	if cfg.UpdateGuardDir == "" || cfg.verifyRunning != nil {
		return nil
	}
	pub, err := release.ParsePublicKey(os.Getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
	if err != nil {
		return err
	}
	d := hostfloor.Disk{Dir: cfg.UpdateGuardDir, Bucket: os.Getenv("CARLOS_DEPLOYMENT_BUCKET"), PublicKey: pub}
	running, err := d.Running(version.Build, runtime.GOOS+"-"+runtime.GOARCH)
	if err != nil {
		return err
	}
	_, floor, err := d.Load()
	if err != nil || !running.GuardReady || running.FloorSHA256 != floor.Digest {
		return hostfloor.ErrFloor
	}
	return hostfloor.CheckRemote(ctx, cfg.Store, pub, d.Bucket, floor)
}