// design touchstones
//   simple to configure/implement
//     - minimal configurable options
//     - sparse use of commands/flags/env vars
//     - fast to start/join a cluster
//     - http-only api
//       api auth/encryption is best implemented by a reverse proxy (like caddy or nginx)
//     - no RPC (it's too annoying to deal with)
//   predictable state
//     - no runtime configuration allowed (no reload support)
//       for a gossip cluster to work, nodes in the cluster must
//       be the source of truth. we do not allow the user to change
//       the cluster's view of reality in any way - the user may
//       only inspect it.
//     - HTTP API only supports read-only commands
//       this ensures that cascade's starting state never
//       diverges from its running state.
//       (restarts are very fast + interruptionless)
//       this also helps with security
//   one massive global cluster
//     - for simplicity, operability, and because gossip can handle it
//   compatability
//     - attempt to have a consul-compatible API where it matters?
// todo
//   think about checks
//   think about DNS weights
//   dns resolver for services
//   read-write exception for maintenance mode?
//   opentelemetry metrics
//   how should cascade handle name conflicts? defaulting to just blowing up
//     for now, but we _could_ take the tailscale route & append -1 or whatever
//     to the node. that would be more user friendly.

// key differences from consul
//   - services are gossip'd

// in cascade, node == member, and catalog == agent
// agent api (http-only)
//   GET /v1/agent/metrics  -> opentelemetry metrics for this agent
//   GET /v1/agent/members  -> show serf cluster membership state
//   GET /v1/agent/services -> show services this agent owns
//     aliases: GET /v1/catalog/nodes
// catalog api (http & dns)
//   GET /v1/catalog/nodes
//   GET /v1/catalog/services
//     dns: dig @127.0.0.1 -p 8600 nostromo.node.cascade. ANY
//     dns: dig @127.0.0.1 -p 8600 redis.service.cascade. ANY

// consul uses /catalog because the masters maintain a service
// catalog separate than the agent state

// in cascade, the agents _are_ expected to maintain service catalogues,
// so we do away with the catalog concept entirely.

package agent

import (
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"path/filepath"
	"sync"
	"time"

	"j3s.sh/cascade/api"
	"github.com/hashicorp/serf/serf"
	"golang.org/x/exp/slog"
)

// antiEntropyInterval is how often each agent re-broadcasts its owned
// service state so peers that missed the original user event catch up.
const antiEntropyInterval = 60 * time.Second

// Agent starts and manages a Serf & adds service discovery (TODO)
type Agent struct {
	// this is the agent's configuration
	Config     *Config
	serfConfig *serf.Config

	// This is the underlying Serf we are wrapping
	serf *serf.Serf

	// This is the http api handler - see http.go and *_endpoint.go
	httpHandlers *HTTPHandlers

	// services holds the agent's own + gossipped service catalog.
	services *ServiceStore

	// dns is the optional DNS server that resolves
	// <node>.node.<domain> and <svc>.service.<domain>.
	dns *DNSServer

	// We receive serf events on this channel
	eventCh chan serf.Event
	logger  *slog.Logger

	// if this channel recieves a signal, cascade shuts down
	shutdown     bool
	shutdownCh   chan struct{}
	shutdownLock sync.Mutex
}

// New returns an agent lmao
func New(config *Config) *Agent {
	eventCh := make(chan serf.Event, 1024)

	// TODO: make log level configurable
	// XXX: reconsider this agent-scoped logger? what does consul do?
	// logLogger is a bridge from the log.Logger to slog.Logger
	h := slog.HandlerOptions{Level: slog.LevelDebug}.NewTextHandler(os.Stderr)
	logger := slog.New(h)
	slog.SetDefault(logger)

	// serf
	serfConfig := serf.DefaultConfig()
	serfConfig.EventCh = eventCh
	serfConfig.Logger = slog.NewLogLogger(h, slog.LevelInfo)
	// TODO: some of these serf settings were cargo-culted
	// from consul[1]. re-examine them eventually.
	serfConfig.EnableNameConflictResolution = false
	serfConfig.LeavePropagateDelay = 3 * time.Second
	serfConfig.MemberlistConfig.BindAddr = config.SerfBindAddr.IP.String()
	serfConfig.MemberlistConfig.BindPort = config.SerfBindAddr.Port
	serfConfig.MemberlistConfig.DeadNodeReclaimTime = 30 * time.Second
	serfConfig.MinQueueDepth = 4096
	serfConfig.NodeName = config.NodeName
	serfConfig.ProtocolVersion = uint8(serf.ProtocolVersionMax)
	serfConfig.QueueDepthWarning = 1000000
	serfConfig.ReconnectTimeout = 3 * 24 * time.Hour

	agent := Agent{
		Config:       config,
		serfConfig:   serfConfig,
		httpHandlers: &HTTPHandlers{},
		eventCh:      eventCh,
		logger:       logger,
		shutdownCh:   make(chan struct{}),
	}

	// we put the completed agent inside of httpHandlers because
	// it needs access to a bunch of Agent{} funcs. basically a hardlink.
	agent.httpHandlers.agent = &agent
	return &agent
}

func (a *Agent) eventLoop() {
	serfShutdownCh := a.serf.ShutdownCh()
	for {
		select {
		case e := <-a.eventCh:
			a.handleSerfEvent(e)

		case <-serfShutdownCh:
			a.logger.Info("detect serf shutdown, turning off event loop")
			a.Shutdown()
			return

		case <-a.shutdownCh:
			return
		}
	}
}

func (a *Agent) handleSerfEvent(e serf.Event) {
	switch ev := e.(type) {
	case serf.UserEvent:
		a.handleUserEvent(ev)
	case serf.MemberEvent:
		if ev.Type == serf.EventMemberFailed || ev.Type == serf.EventMemberLeave || ev.Type == serf.EventMemberReap {
			for _, m := range ev.Members {
				a.logger.Info("drop services for departed member", "node", m.Name, "event", ev.Type.String())
				a.services.DropNode(m.Name)
			}
		}
	default:
		a.logger.Debug("receive event", "event", e.String())
	}
}

func (a *Agent) handleUserEvent(ev serf.UserEvent) {
	switch ev.Name {
	case serfRegisterEvent:
		var p registerPayload
		if err := json.Unmarshal(ev.Payload, &p); err != nil {
			a.logger.Warn("decode svc-register event", "err", err)
			return
		}
		if p.Service == nil {
			return
		}
		a.services.ApplyRegister(p.Node, p.Service)
	case serfDeregisterEvent:
		var p deregisterPayload
		if err := json.Unmarshal(ev.Payload, &p); err != nil {
			a.logger.Warn("decode svc-deregister event", "err", err)
			return
		}
		a.services.ApplyDeregister(p.Node, p.ID)
	}
}

// RegisterService stores a service in the local agent and gossips the new
// state to the cluster.
func (a *Agent) RegisterService(svc *api.AgentService) error {
	if err := a.services.Register(svc); err != nil {
		return err
	}
	return a.broadcastRegister(svc)
}

// DeregisterService removes a service from the local agent and gossips the
// removal to the cluster.
func (a *Agent) DeregisterService(id string) error {
	if err := a.services.Deregister(id); err != nil {
		return err
	}
	payload, err := json.Marshal(deregisterPayload{Node: a.Config.NodeName, ID: id})
	if err != nil {
		return err
	}
	return a.serf.UserEvent(serfDeregisterEvent, payload, false)
}

func (a *Agent) broadcastRegister(svc *api.AgentService) error {
	payload, err := json.Marshal(registerPayload{Node: a.Config.NodeName, Service: svc})
	if err != nil {
		return err
	}
	return a.serf.UserEvent(serfRegisterEvent, payload, false)
}

// OwnedServices returns this agent's currently-owned service registrations.
func (a *Agent) OwnedServices() map[string]*api.AgentService {
	return a.services.Owned()
}

// CatalogServices returns the cluster-wide service-name -> tag-union view.
func (a *Agent) CatalogServices() map[string][]string {
	return a.services.AllServices()
}

// CatalogServiceInstances returns every known instance of the named service.
func (a *Agent) CatalogServiceInstances(name string) []NodeService {
	return a.services.ServiceInstances(name)
}

// CatalogAllInstances returns every gossipped service instance across the
// whole cluster as a flat slice.
func (a *Agent) CatalogAllInstances() []NodeService {
	return a.services.AllInstances()
}

// CatalogNodeServices returns the services owned by the named node.
func (a *Agent) CatalogNodeServices(node string) map[string]*api.AgentService {
	return a.services.NodeServices(node)
}

// CatalogNodes returns every node serf currently knows about, regardless of
// whether it has registered services. Nodes that have gossipped services
// but are not yet visible to serf (rare race during join) are also included.
func (a *Agent) CatalogNodes() []serf.Member {
	members := a.serf.Members()
	seen := map[string]struct{}{}
	for _, m := range members {
		seen[m.Name] = struct{}{}
	}
	for _, n := range a.services.Nodes() {
		if _, ok := seen[n]; ok {
			continue
		}
		members = append(members, serf.Member{Name: n, Status: serf.StatusNone})
		seen[n] = struct{}{}
	}
	return members
}

// MemberByName returns the serf member with the given name, or zero value
// if no such member is known.
func (a *Agent) MemberByName(name string) (serf.Member, bool) {
	for _, m := range a.serf.Members() {
		if m.Name == name {
			return m, true
		}
	}
	return serf.Member{}, false
}

// antiEntropyLoop periodically re-broadcasts this agent's owned services so
// peers that missed the original user event eventually catch up.
func (a *Agent) antiEntropyLoop() {
	t := time.NewTicker(antiEntropyInterval)
	defer t.Stop()
	for {
		select {
		case <-t.C:
			for _, svc := range a.services.Owned() {
				if err := a.broadcastRegister(svc); err != nil {
					a.logger.Warn("anti-entropy rebroadcast", "id", svc.ID, "err", err)
				}
			}
		case <-a.shutdownCh:
			return
		}
	}
}

// Start creates and starts an agent's serf client, eventloop, api, and more!
// The agent config should not be modified after an agent has been started.
func (a *Agent) Start() error {
	// open the service store before serf so we never gossip without it
	if err := os.MkdirAll(a.Config.DataDir, 0o700); err != nil {
		return fmt.Errorf("create data dir %q: %w", a.Config.DataDir, err)
	}
	dbPath := filepath.Join(a.Config.DataDir, a.Config.NodeName+".db")
	store, err := NewServiceStore(a.Config.NodeName, dbPath)
	if err != nil {
		return err
	}
	a.services = store

	// Create serf first
	srf, err := serf.Create(a.serfConfig)
	if err != nil {
		return err
	}
	a.serf = srf

	// Start event loop
	go a.eventLoop()
	go a.antiEntropyLoop()

	// Re-broadcast any services that survived restart so peers catch up
	// without waiting for the first anti-entropy tick.
	for _, svc := range a.services.Owned() {
		if err := a.broadcastRegister(svc); err != nil {
			a.logger.Warn("startup rebroadcast", "id", svc.ID, "err", err)
		}
	}

	// Start API Server
	apiHandler := a.httpHandlers.handler()
	go a.serveAPI(apiHandler)

	// Start DNS server
	a.dns = newDNSServer(a)
	if err := a.dns.Start(); err != nil {
		return err
	}

	return nil
}

func (a *Agent) serveAPI(handler http.Handler) {
	portStr := fmt.Sprintf("%s:%d", a.Config.HTTPBindAddr.IP.String(), a.Config.HTTPBindAddr.Port)
	if err := http.ListenAndServe(portStr, handler); err != nil {
		a.logger.Error("api error", err)
	}
}

// Shutdown closes this agent and all of its processes. Should be preceded
// by a Leave for a graceful shutdown.
func (a *Agent) Shutdown() error {
	a.shutdownLock.Lock()
	defer a.shutdownLock.Unlock()

	if a.shutdown {
		return nil
	}

	if a.serf == nil {
		goto EXIT
	}

	a.logger.Info("request serf shutdown")
	if err := a.serf.Shutdown(); err != nil {
		return err
	}

EXIT:
	a.logger.Info("complete serf shutdown")
	if a.dns != nil {
		a.dns.Shutdown()
	}
	if a.services != nil {
		if err := a.services.Close(); err != nil {
			a.logger.Warn("close service store", "err", err)
		}
	}
	a.shutdown = true
	close(a.shutdownCh)
	return nil
}

// Join asks the Serf instance to join. See the Serf.Join function.
func (a *Agent) Join(addrs []string) (n int, err error) {
	a.logger.Info("issue join request", "addrs", addrs)
	// we always ignore old events because cascade don't
	// care about the past
	n, err = a.serf.Join(addrs, true)
	if err != nil {
		return n, fmt.Errorf("Error joining: %v\n", err)
	}
	return n, err
}

// Leave prepares for a graceful shutdown of the agent and its processes
func (a *Agent) Leave() error {
	if a.serf == nil {
		return nil
	}

	a.logger.Info("request graceful serf leave")
	return a.serf.Leave()
}

// ShutdownCh returns a channel that can be selected to wait
// for the agent to perform a shutdown.
func (a *Agent) ShutdownCh() <-chan struct{} {
	return a.shutdownCh
}

func (a *Agent) GetCoordinate() error {
	return nil
}

// TODO: filter support
// Members returns all of the serf members in the gossip cluster.
func (a *Agent) Members() []serf.Member {
	// todo: move this to cascade/client perhaps
	return a.serf.Members()
}

// [1]: sources for consul serf tweaks
//      https://github.com/hashicorp/consul/blob/v1.14.4/agent/consul/config.go
//      https://github.com/hashicorp/consul/blob/v1.14.4/lib/serf/serf.go