// Copyright 2025 Emiliano Spinella (eminwux)
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// SPDX-License-Identifier: Apache-2.0

// Package daemon implements the kukeond process: it owns a controller.Exec,
// exposes the kukeonv1 API over a unix socket using net/rpc + jsonrpc, and
// manages its own lifecycle (listener, PID file, graceful shutdown).
package daemon

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net"
	"net/rpc"
	"net/rpc/jsonrpc"
	"os"
	"path/filepath"
	"sync"
	"time"

	"github.com/eminwux/kukeon/internal/client/local"
	"github.com/eminwux/kukeon/internal/controller"
	"github.com/eminwux/kukeon/internal/firewall"
	"github.com/eminwux/kukeon/pkg/api/kukeonv1"
)

// Options configures a Server.
type Options struct {
	// SocketPath is the unix socket path the daemon listens on.
	SocketPath string
	// SocketMode is the file mode applied to the socket after creation.
	SocketMode os.FileMode
	// SocketGID, when non-zero, is the group the listener socket is chowned
	// to (uid stays root) so non-root members of that group can dial the
	// daemon. Set by `kuke init` to the kukeon GID. Each Serve re-applies
	// this so a daemon restart does not lose group access.
	SocketGID int
	// PIDFile, when non-empty, is written on Serve and removed on Stop.
	PIDFile string
	// ReconcileInterval is the period of the background cell-reconciliation
	// loop. Zero or negative disables the loop — useful for tests and for
	// operators who explicitly opt out via `--reconcile-interval 0`.
	ReconcileInterval time.Duration
	// Controller is forwarded to controller.NewControllerExec.
	Controller controller.Options
}

// Server hosts the kukeonv1 API over JSON-RPC on a unix socket.
type Server struct {
	ctx    context.Context
	logger *slog.Logger
	opts   Options

	core *local.Client
	rpc  *rpc.Server

	// reconcileFn is the per-tick callable for the cell-reconciliation loop.
	// Defaults to core.ReconcileCells; tests in the daemon package overwrite
	// it before Serve so they exercise the ticker without a real controller.
	reconcileFn func() (controller.ReconcileResult, error)

	// forwardAdmissionFn re-asserts the host FORWARD admission chain once on
	// Serve startup. Defaults to reassertForwardAdmission; tests overwrite it
	// before Serve so they exercise the startup hook without touching iptables.
	forwardAdmissionFn func() error

	// stopCh is closed by Stop to terminate the reconcile loop independently
	// of s.ctx; loopWG lets Stop block until the loop exits, so core.Close
	// never races an in-flight reconcile pass.
	stopCh chan struct{}
	loopWG sync.WaitGroup

	mu       sync.Mutex
	listener net.Listener
	closed   bool
}

// NewServer constructs a Server. It does not open the listener; call Serve.
func NewServer(ctx context.Context, logger *slog.Logger, opts Options) *Server {
	if opts.SocketMode == 0 {
		opts.SocketMode = 0o600
	}
	core := local.New(ctx, logger, opts.Controller)
	srv := &Server{
		ctx:    ctx,
		logger: logger,
		opts:   opts,
		core:   core,
		stopCh: make(chan struct{}),
	}
	srv.reconcileFn = srv.core.ReconcileCells
	srv.forwardAdmissionFn = srv.reassertForwardAdmission
	return srv
}

// Serve opens the listener and accepts connections until Stop is called or
// the context is cancelled. Returns nil on graceful shutdown.
func (s *Server) Serve() error {
	if err := s.prepareSocketDir(); err != nil {
		return err
	}
	if err := s.removeStaleSocket(); err != nil {
		return err
	}

	listener, err := net.Listen("unix", s.opts.SocketPath)
	if err != nil {
		return fmt.Errorf("listen on %s: %w", s.opts.SocketPath, err)
	}
	if s.opts.SocketGID > 0 {
		if chownErr := os.Chown(s.opts.SocketPath, 0, s.opts.SocketGID); chownErr != nil {
			s.logger.WarnContext(s.ctx,
				"chown socket to kukeon group failed; non-root operators will need sudo",
				"socket", s.opts.SocketPath,
				"gid", s.opts.SocketGID,
				"error", chownErr,
			)
		}
	}
	if err := os.Chmod(s.opts.SocketPath, s.opts.SocketMode); err != nil {
		_ = listener.Close()
		return fmt.Errorf("chmod socket: %w", err)
	}

	s.mu.Lock()
	s.listener = listener
	s.mu.Unlock()

	if err := s.writePIDFile(); err != nil {
		_ = listener.Close()
		return err
	}

	s.rpc = rpc.NewServer()
	svc := NewKukeonV1Service(s.ctx, s.logger, s.core)
	if err := s.rpc.RegisterName(kukeonv1.ServiceName, svc); err != nil {
		_ = listener.Close()
		return fmt.Errorf("register rpc service: %w", err)
	}

	s.logger.InfoContext(s.ctx, "kukeond listening", "socket", s.opts.SocketPath)
	// NewServer always wires forwardAdmissionFn; call it unguarded to match the
	// reconcileFn pattern (runReconcileOnce dereferences s.reconcileFn directly).
	if err := s.forwardAdmissionFn(); err != nil {
		// Best-effort: a flushed/unavailable FORWARD admission chain must
		// not block the daemon from serving. The next reboot self-heals
		// here; durable per-space egress + Docker-churn handling is #953.
		s.logger.WarnContext(s.ctx,
			"forward admission re-assert failed on startup; continuing",
			"error", err)
	}
	s.startReconcileLoop()
	s.acceptLoop(listener)
	return nil
}

// reassertForwardAdmission re-installs the KUKEON-FORWARD admission chain and
// its FORWARD jump on daemon startup so a host reboot — which flushes the
// kernel's iptables and strips the chain installed by `kuke init` — self-heals
// without a re-run of init. Without it, cells on a host whose FORWARD default
// policy is DROP (e.g. when Docker is installed) lose egress after a reboot:
// in-cell DNS times out, kuketty's boot stalls past the attach deadline, and
// `kuke run` fails with "attach ping timed out".
//
// Idempotent — every install step does -C before -I/-A, so on a healthy host
// this produces no rule churn. Best-effort: a missing iptables binary is a
// WARN-and-skip (the host owner opted out of kukeon-managed admission), and a
// transient iptables error is surfaced to the caller, which logs and continues
// rather than aborting Serve. Per-space egress re-assertion and Docker-churn
// ordering are out of scope here and tracked in #953.
func (s *Server) reassertForwardAdmission() error {
	if !firewall.IsIptablesAvailable() {
		s.logger.WarnContext(s.ctx,
			"iptables not found on PATH; skipping forward admission re-assert — "+
				"kukeon-bridge traffic may be blocked if FORWARD default policy is DROP",
			"chain", firewall.ForwardChainName)
		return nil
	}
	if err := firewall.NewInstaller(s.logger).Install(s.ctx); err != nil {
		return fmt.Errorf("re-assert forward admission chain: %w", err)
	}
	s.logger.InfoContext(s.ctx, "re-asserted forward admission chain on startup",
		"chain", firewall.ForwardChainName)
	return nil
}

// startReconcileLoop spawns the background cell-reconciliation ticker when
// ReconcileInterval > 0. Lifetime is bound to s.ctx — daemon shutdown stops
// the loop. Errors during a pass are logged and the loop continues. One log
// line per pass: brief on success, detailed on failure (per #161 AC).
func (s *Server) startReconcileLoop() {
	if s.opts.ReconcileInterval <= 0 {
		s.logger.InfoContext(s.ctx, "reconcile loop disabled",
			"interval", s.opts.ReconcileInterval)
		return
	}
	s.logger.InfoContext(s.ctx, "starting reconcile loop",
		"interval", s.opts.ReconcileInterval)
	s.loopWG.Add(1)
	go s.runReconcileLoop(s.opts.ReconcileInterval)
}

func (s *Server) runReconcileLoop(interval time.Duration) {
	defer s.loopWG.Done()

	// Run one pass immediately, before the first ticker tick, so a freshly
	// started daemon converges every cell's persisted Status to live
	// containerd reality at once instead of serving the pre-restart snapshot
	// for up to a full interval. The headline case is a host restart that
	// dropped containerd's tasks/records (#671): the on-disk metadata still
	// reads Ready and lists containers that no longer exist, so without this
	// eager pass `kuke get cell` shows stale state and AutoDelete cleanup of
	// now-phantom cells is delayed until the first tick. Gated on the same
	// shutdown signals as the loop so a daemon cancelled before the loop
	// starts does not run a pass.
	select {
	case <-s.ctx.Done():
		s.logger.InfoContext(s.ctx, "reconcile loop stopped")
		return
	case <-s.stopCh:
		s.logger.InfoContext(s.ctx, "reconcile loop stopped")
		return
	default:
		s.runReconcileOnce()
	}

	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-s.ctx.Done():
			s.logger.InfoContext(s.ctx, "reconcile loop stopped")
			return
		case <-s.stopCh:
			s.logger.InfoContext(s.ctx, "reconcile loop stopped")
			return
		case <-ticker.C:
			s.runReconcileOnce()
		}
	}
}

func (s *Server) runReconcileOnce() {
	defer func() {
		if r := recover(); r != nil {
			s.logger.ErrorContext(s.ctx, "reconcile pass panicked; loop continues",
				"panic", r)
		}
	}()
	res, err := s.reconcileFn()
	switch {
	case err != nil:
		s.logger.ErrorContext(s.ctx, "reconcile pass failed",
			"error", err,
			"cells_scanned", res.CellsScanned,
			"cells_updated", res.CellsUpdated,
			"cells_deleted", res.CellsDeleted,
			"cells_errored", res.CellsErrored,
			"errors", res.Errors)
	case len(res.Errors) > 0:
		s.logger.WarnContext(s.ctx, "reconcile pass completed with errors",
			"cells_scanned", res.CellsScanned,
			"cells_updated", res.CellsUpdated,
			"cells_deleted", res.CellsDeleted,
			"cells_errored", res.CellsErrored,
			"errors", res.Errors)
	default:
		s.logger.InfoContext(s.ctx, "reconcile ok",
			"cells_scanned", res.CellsScanned,
			"cells_updated", res.CellsUpdated,
			"cells_deleted", res.CellsDeleted)
	}
}

func (s *Server) acceptLoop(listener net.Listener) {
	for {
		conn, err := listener.Accept()
		if err != nil {
			s.mu.Lock()
			closed := s.closed
			s.mu.Unlock()
			if closed || errors.Is(err, net.ErrClosed) {
				return
			}
			if s.ctx.Err() != nil {
				return
			}
			s.logger.WarnContext(s.ctx, "accept error", "error", err)
			continue
		}
		go s.rpc.ServeCodec(jsonrpc.NewServerCodec(conn))
	}
}

// Stop closes the listener, releases the controller, and removes the socket
// and PID file. Safe to call multiple times.
func (s *Server) Stop() error {
	s.mu.Lock()
	if s.closed {
		s.mu.Unlock()
		return nil
	}
	s.closed = true
	listener := s.listener
	s.listener = nil
	s.mu.Unlock()

	// Signal the reconcile loop to exit and wait for any in-flight pass to
	// return before tearing down the controller — otherwise a tick already
	// inside reconcileFn would race s.core.Close.
	close(s.stopCh)
	s.loopWG.Wait()

	var firstErr error
	if listener != nil {
		if err := listener.Close(); err != nil {
			firstErr = err
		}
	}
	if s.core != nil {
		if err := s.core.Close(); err != nil && firstErr == nil {
			firstErr = err
		}
	}
	if s.opts.SocketPath != "" {
		_ = os.Remove(s.opts.SocketPath)
	}
	if s.opts.PIDFile != "" {
		_ = os.Remove(s.opts.PIDFile)
	}
	return firstErr
}

func (s *Server) prepareSocketDir() error {
	dir := filepath.Dir(s.opts.SocketPath)
	if dir == "" || dir == "." {
		return nil
	}
	return os.MkdirAll(dir, 0o755)
}

func (s *Server) removeStaleSocket() error {
	err := os.Remove(s.opts.SocketPath)
	if err == nil || errors.Is(err, os.ErrNotExist) {
		return nil
	}
	return fmt.Errorf("remove stale socket: %w", err)
}

func (s *Server) writePIDFile() error {
	if s.opts.PIDFile == "" {
		return nil
	}
	if err := os.MkdirAll(filepath.Dir(s.opts.PIDFile), 0o755); err != nil {
		return fmt.Errorf("pid file dir: %w", err)
	}
	content := fmt.Sprintf("%d\n", os.Getpid())
	if err := os.WriteFile(s.opts.PIDFile, []byte(content), 0o644); err != nil {
		return fmt.Errorf("write pid file: %w", err)
	}
	return nil
}
