wx-win-agent/control-plane/server.go

package controlplane

import (
	"context"
	"crypto/rand"
	"crypto/sha256"
	"crypto/subtle"
	"crypto/tls"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"mime"
	"net/http"
	"net/url"
	"path"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	webassets "git.ipao.vip/rogee/wx-win-agent/control-plane/web"
)

type ServerConfig struct {
	ListenAddr                 string
	DataFile                   string
	AccountDataDir             string
	NodeTokens                 map[string]string
	WebUsers                   map[string]string
	LeaseTTL                   time.Duration
	HeartbeatTimeout           time.Duration
	SessionTTL                 time.Duration
	TLSCertFile                string
	TLSKeyFile                 string
	MTLSClientCAFile           string
	MTLSRequireNodeCert        bool
	MTLSRevokedCertsFile       string
	BackupDir                  string
	BackupCount                int
	BackupInterval             time.Duration
	TaskRetention              time.Duration
	EventRetention             time.Duration
	AuditRetention             time.Duration
	AccountRetention           time.Duration
	AccountMaxShardBytes       int64
	AccountMaxBatchBytes       int64
	AccountMaintenanceInterval time.Duration
	AIBaseURL                  string
	AIAPIKey                   string
	AIModel                    string
	AITimeout                  time.Duration
	AISchedulerInterval        time.Duration
	AIProvider                 AIProvider
}

func DefaultServerConfig() ServerConfig {
	return ServerConfig{
		ListenAddr:                 "127.0.0.1:8090",
		DataFile:                   "control-plane-data.json",
		NodeTokens:                 map[string]string{},
		WebUsers:                   map[string]string{},
		LeaseTTL:                   30 * time.Second,
		HeartbeatTimeout:           45 * time.Second,
		SessionTTL:                 8 * time.Hour,
		BackupCount:                7,
		BackupInterval:             5 * time.Minute,
		TaskRetention:              30 * 24 * time.Hour,
		EventRetention:             30 * 24 * time.Hour,
		AuditRetention:             90 * 24 * time.Hour,
		AccountRetention:           30 * 24 * time.Hour,
		AccountMaxShardBytes:       512 * 1024 * 1024,
		AccountMaxBatchBytes:       maxDataBatchBytes,
		AccountMaintenanceInterval: 5 * time.Minute,
		AIModel:                    "gpt-4o-mini",
		AITimeout:                  60 * time.Second,
		AISchedulerInterval:        5 * time.Second,
	}
}

type Server struct {
	config          ServerConfig
	store           *Store
	accountStores   *AccountStoreManager
	tlsConfig       *tls.Config
	nodeTokenHashes map[string][32]byte
	userPasswords   map[string][32]byte
	sessionMu       sync.Mutex
	sessions        map[string]session
	aiProvider      AIProvider
	aiCtx           context.Context
	aiCancel        context.CancelFunc
	aiQueue         chan string
	aiWG            sync.WaitGroup
	closeOnce       sync.Once
	closeErr        error
}

type session struct {
	Username string
	Expires  time.Time
}

type requestError struct {
	status  int
	code    string
	message string
}

func (e requestError) Error() string { return e.message }

func NewServer(config ServerConfig) (*Server, error) {
	defaults := DefaultServerConfig()
	if config.ListenAddr == "" {
		config.ListenAddr = defaults.ListenAddr
	}
	if config.DataFile == "" {
		config.DataFile = defaults.DataFile
	}
	if config.AccountDataDir == "" {
		config.AccountDataDir = config.DataFile + ".accounts"
	}
	if config.BackupDir == "" {
		config.BackupDir = config.DataFile + ".backups"
	}
	if config.BackupCount == 0 {
		config.BackupCount = defaults.BackupCount
	}
	if config.BackupInterval == 0 {
		config.BackupInterval = defaults.BackupInterval
	}
	if config.TaskRetention == 0 {
		config.TaskRetention = defaults.TaskRetention
	}
	if config.EventRetention == 0 {
		config.EventRetention = defaults.EventRetention
	}
	if config.AuditRetention == 0 {
		config.AuditRetention = defaults.AuditRetention
	}
	if config.AccountRetention == 0 {
		config.AccountRetention = defaults.AccountRetention
	}
	if config.AccountMaxShardBytes == 0 {
		config.AccountMaxShardBytes = defaults.AccountMaxShardBytes
	}
	if config.AccountMaxBatchBytes == 0 {
		config.AccountMaxBatchBytes = defaults.AccountMaxBatchBytes
	}
	if config.AccountMaintenanceInterval == 0 {
		config.AccountMaintenanceInterval = defaults.AccountMaintenanceInterval
	}
	if config.LeaseTTL <= 0 {
		config.LeaseTTL = defaults.LeaseTTL
	}
	if config.HeartbeatTimeout <= 0 {
		config.HeartbeatTimeout = defaults.HeartbeatTimeout
	}
	if config.SessionTTL <= 0 {
		config.SessionTTL = defaults.SessionTTL
	}
	if config.AIModel == "" {
		config.AIModel = defaults.AIModel
	}
	if config.AITimeout <= 0 {
		config.AITimeout = defaults.AITimeout
	}
	if config.AISchedulerInterval <= 0 {
		config.AISchedulerInterval = defaults.AISchedulerInterval
	}
	if config.AIProvider == nil && config.AIBaseURL != "" {
		config.AIProvider = &OpenAICompatibleProvider{BaseURL: config.AIBaseURL, APIKey: config.AIAPIKey, Model: config.AIModel}
	}
	if config.NodeTokens == nil {
		config.NodeTokens = map[string]string{}
	}
	if config.WebUsers == nil {
		config.WebUsers = map[string]string{}
	}
	if config.BackupCount < 0 || config.BackupInterval < 0 || config.TaskRetention < 0 || config.EventRetention < 0 || config.AuditRetention < 0 || config.AccountRetention < 0 || config.AccountMaxShardBytes < 0 || config.AccountMaxBatchBytes < 0 || config.AccountMaintenanceInterval < 0 || config.AITimeout < 0 || config.AISchedulerInterval < 0 {
		return nil, errors.New("backup count, account budgets, and retention settings must be non-negative")
	}
	tlsConfig, err := newTLSConfig(config)
	if err != nil {
		return nil, err
	}
	store, err := OpenStore(config.DataFile, StoreOptions{
		BackupDir: config.BackupDir, BackupCount: config.BackupCount, BackupInterval: config.BackupInterval,
		TaskRetention: config.TaskRetention, EventRetention: config.EventRetention, AuditRetention: config.AuditRetention,
	})
	if err != nil {
		return nil, err
	}
	accountStores, err := OpenAccountStoreManager(config.AccountDataDir, AccountStoreManagerOptions{
		Retention:     config.AccountRetention,
		MaxShardBytes: config.AccountMaxShardBytes,
		MaxBatchBytes: config.AccountMaxBatchBytes,
	})
	if err != nil {
		_ = store.Close()
		return nil, err
	}
	aiCtx, aiCancel := context.WithCancel(context.Background())
	s := &Server{
		config:          config,
		store:           store,
		accountStores:   accountStores,
		tlsConfig:       tlsConfig,
		nodeTokenHashes: map[string][32]byte{},
		userPasswords:   map[string][32]byte{},
		sessions:        map[string]session{},
		aiProvider:      config.AIProvider,
		aiCtx:           aiCtx,
		aiCancel:        aiCancel,
		aiQueue:         make(chan string, 1000),
	}
	for nodeID, token := range config.NodeTokens {
		if validIdentifier(nodeID, 200) && token != "" {
			s.nodeTokenHashes[nodeID] = sha256.Sum256([]byte(token))
		}
	}
	for username, password := range config.WebUsers {
		if username != "" && password != "" {
			s.userPasswords[username] = sha256.Sum256([]byte(password))
		}
	}
	s.aiWG.Add(3)
	go s.aiWorker()
	go s.aiScheduler()
	go s.accountMaintenance()
	return s, nil
}

func (s *Server) Handler() http.Handler { return s }

func (s *Server) ListenAndServe(ctx context.Context) error {
	server := &http.Server{Addr: s.config.ListenAddr, Handler: s, ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 15 * time.Second, WriteTimeout: 15 * time.Second, IdleTimeout: 60 * time.Second}
	go func() {
		<-ctx.Done()
		shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = server.Shutdown(shutdownCtx)
	}()
	defer func() { _ = s.Close() }()
	var err error
	if s.tlsConfig != nil {
		server.TLSConfig = s.tlsConfig
		err = server.ListenAndServeTLS("", "")
	} else {
		err = server.ListenAndServe()
	}
	if errors.Is(err, http.ErrServerClosed) {
		return nil
	}
	return err
}

func (s *Server) accountMaintenance() {
	defer s.aiWG.Done()
	run := func() {
		_, _ = s.accountStores.RunMaintenance(s.aiCtx, time.Now().UTC())
	}
	run()
	interval := s.config.AccountMaintenanceInterval
	if interval <= 0 {
		return
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-s.aiCtx.Done():
			return
		case <-ticker.C:
			run()
		}
	}
}

// Close releases the active/passive store lock and stops background work.
func (s *Server) Close() error {
	s.closeOnce.Do(func() {
		s.aiCancel()
		s.aiWG.Wait()
		s.closeErr = errors.Join(s.accountStores.Close(), s.store.Close())
	})
	return s.closeErr
}

func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	correlationID := r.Header.Get("X-Correlation-Id")
	if !validIdentifier(correlationID, 128) {
		correlationID = randomID()
	}
	w.Header().Set("X-Correlation-Id", correlationID)
	w.Header().Set("Cache-Control", "no-store")
	w.Header().Set("X-Content-Type-Options", "nosniff")
	w.Header().Set("Referrer-Policy", "no-referrer")
	if r.Method == http.MethodOptions {
		w.WriteHeader(http.StatusNoContent)
		return
	}

	var err error
	switch {
	case r.URL.Path == "/" && r.Method == http.MethodGet:
		s.serveFrontend(w, "dist/index.html")
		return
	case strings.HasPrefix(r.URL.Path, "/assets/") && r.Method == http.MethodGet:
		s.serveFrontend(w, path.Join("dist", strings.TrimPrefix(r.URL.Path, "/")))
		return
	case r.URL.Path == "/healthz" && r.Method == http.MethodGet:
		writeJSON(w, http.StatusOK, map[string]any{"status": "ok", "protocol_version": ProtocolVersion, "correlation_id": correlationID})
		return
	case r.URL.Path == "/readyz" && r.Method == http.MethodGet:
		err = s.ready(w, correlationID)
	case r.URL.Path == "/v1/auth/login" && r.Method == http.MethodPost:
		err = s.login(w, r, correlationID)
	case r.URL.Path == "/v1/nodes" && r.Method == http.MethodGet:
		err = s.listNodes(w, r)
	case r.URL.Path == "/v1/nodes/register" && r.Method == http.MethodPost:
		err = s.registerNode(w, r, correlationID)
	case r.URL.Path == "/v1/data/batches" && r.Method == http.MethodPost:
		err = s.ingestDataBatch(w, r, correlationID)
	case strings.HasPrefix(r.URL.Path, "/v1/data/accounts/"):
		err = s.dataRoute(w, r, correlationID)
	case strings.HasPrefix(r.URL.Path, "/v1/nodes/"):
		err = s.nodeRoute(w, r, correlationID)
	case strings.HasPrefix(r.URL.Path, "/v1/reads/"):
		err = s.readRoute(w, r, correlationID)
	case strings.HasPrefix(r.URL.Path, "/v1/ai/"):
		err = s.aiRoute(w, r, correlationID)
	case r.URL.Path == "/v1/tasks" || strings.HasPrefix(r.URL.Path, "/v1/tasks/"):
		err = s.taskRoute(w, r, correlationID)
	case r.URL.Path == "/v1/events":
		err = s.eventRoute(w, r, correlationID)
	case r.URL.Path == "/v1/audit" && r.Method == http.MethodGet:
		err = s.auditRoute(w, r, correlationID)
	default:
		err = requestError{status: http.StatusNotFound, code: "NotFound", message: "Resource was not found."}
	}
	if err != nil {
		writeError(w, err, correlationID)
	}
}

func (s *Server) ready(w http.ResponseWriter, correlationID string) error {
	if err := s.store.Read(func(PersistedState) error { return nil }); err != nil {
		return requestError{status: http.StatusServiceUnavailable, code: "NotReady", message: "The control plane store is not ready."}
	}
	writeJSON(w, http.StatusOK, map[string]any{"status": "ready", "protocol_version": ProtocolVersion, "correlation_id": correlationID})
	return nil
}

func (s *Server) login(w http.ResponseWriter, r *http.Request, correlationID string) error {
	var request struct {
		Username string `json:"username"`
		Password string `json:"password"`
	}
	if err := decodeJSON(r, &request, 8*1024); err != nil {
		return err
	}
	password, exists := s.userPasswords[request.Username]
	provided := sha256.Sum256([]byte(request.Password))
	if !exists || subtle.ConstantTimeCompare(password[:], provided[:]) != 1 {
		return requestError{status: http.StatusUnauthorized, code: "Unauthorized", message: "Authentication failed."}
	}
	token := randomID()
	s.sessionMu.Lock()
	s.sessions[token] = session{Username: request.Username, Expires: time.Now().UTC().Add(s.config.SessionTTL)}
	s.sessionMu.Unlock()
	_ = s.appendAudit(request.Username, "login", "session", correlationID, "success")
	writeJSON(w, http.StatusOK, map[string]any{"access_token": token, "token_type": "Bearer", "expires_in": int(s.config.SessionTTL.Seconds()), "correlation_id": correlationID})
	return nil
}

func (s *Server) registerNode(w http.ResponseWriter, r *http.Request, correlationID string) error {
	nodeID, err := s.authenticateNode(r)
	if err != nil {
		return err
	}
	var request NodeRegistration
	if err := decodeJSON(r, &request, 128*1024); err != nil {
		return err
	}
	if request.NodeID != nodeID {
		return requestError{status: http.StatusForbidden, code: "NodeIdentityMismatch", message: "The token is not assigned to this node."}
	}
	if (request.ConnectionID != "" && !validIdentifier(request.ConnectionID, 200)) || !validIdentifier(request.AgentVersion, 80) || request.ProtocolVersion != ProtocolVersion || len(request.Capabilities) > 100 || !validCapabilities(request.Capabilities) || !validAccountSummaries(request.Accounts) {
		return requestError{status: http.StatusBadRequest, code: "InvalidRegistration", message: "Node registration is invalid."}
	}
	if err := s.registerDataAccounts(nodeID, request); err != nil {
		if errors.Is(err, ErrAccountBindingConflict) {
			return requestError{status: http.StatusConflict, code: "AccountBindingConflict", message: "The registered account source conflicts with an existing binding."}
		}
		return requestError{status: http.StatusInternalServerError, code: "AccountRegistrationFailed", message: "The platform could not register account data storage."}
	}
	now := time.Now().UTC()
	if err := s.store.Mutate(func(state *PersistedState) error {
		node := state.Nodes[nodeID]
		if node.NodeID != "" && node.ConnectionID != request.ConnectionID {
			quarantineNodeTasks(state, nodeID, now)
		}
		node.NodeID = nodeID
		node.ConnectionID = request.ConnectionID
		node.AgentVersion = request.AgentVersion
		node.ProtocolVersion = request.ProtocolVersion
		node.Capabilities = append([]string(nil), request.Capabilities...)
		node.Status = NodeOnline
		node.LastHeartbeatAt = &now
		node.ReportingConfigVersion = request.ReportingConfigVersion
		node.Accounts = append([]AccountSummary(nil), request.Accounts...)
		node.LastCorrelationID = correlationID
		state.Nodes[nodeID] = node
		state.Audit = appendAudit(state.Audit, "node:"+nodeID, "node.register", nodeID, correlationID, "success", now)
		return nil
	}); err != nil {
		return requestError{status: http.StatusInternalServerError, code: "PersistenceFailed", message: "The control plane could not persist the node registration."}
	}
	writeJSON(w, http.StatusOK, NodeResponse{NodeID: nodeID, ConnectionID: request.ConnectionID, Status: NodeOnline, Authenticated: true, CorrelationID: correlationID})
	return nil
}

func (s *Server) nodeRoute(w http.ResponseWriter, r *http.Request, correlationID string) error {
	parts := pathParts(r.URL.Path)
	if len(parts) < 3 || parts[0] != "v1" || parts[1] != "nodes" {
		return requestError{status: http.StatusNotFound, code: "NotFound", message: "Resource was not found."}
	}
	nodeID, err := s.authenticateNode(r)
	if err != nil {
		return err
	}
	if nodeID != parts[2] {
		return requestError{status: http.StatusForbidden, code: "NodeIdentityMismatch", message: "The token is not assigned to this node."}
	}
	if len(parts) == 3 && r.Method == http.MethodGet {
		return s.nodeDetails(w, nodeID, correlationID)
	}
	if len(parts) < 4 {
		return requestError{status: http.StatusNotFound, code: "NotFound", message: "Resource was not found."}
	}
	switch parts[3] {
	case "heartbeat":
		if r.Method != http.MethodPost || len(parts) != 4 {
			return requestError{status: http.StatusMethodNotAllowed, code: "MethodNotAllowed", message: "Method is not allowed."}
		}
		return s.heartbeat(w, r, nodeID, correlationID)
	case "tasks":
		return s.nodeTaskRoute(w, r, nodeID, parts[4:], correlationID)
	case "events":
		if r.Method != http.MethodPost || len(parts) != 4 {
			return requestError{status: http.StatusMethodNotAllowed, code: "MethodNotAllowed", message: "Method is not allowed."}
		}
		return s.ingestEvent(w, r, nodeID, correlationID)
	case "data":
		return s.nodeDataRoute(w, r, nodeID, parts[4:], correlationID)
	default:
		return requestError{status: http.StatusNotFound, code: "NotFound", message: "Resource was not found."}
	}
}

func (s *Server) listNodes(w http.ResponseWriter, r *http.Request) error {
	if _, err := s.authenticateWeb(r); err != nil {
		return err
	}
	var nodes []Node
	if err := s.store.Mutate(func(state *PersistedState) error {
		now := time.Now().UTC()
		for nodeID, value := range state.Nodes {
			node := value
			nodeChanged := false
			if node.LastHeartbeatAt == nil || now.Sub(*node.LastHeartbeatAt) > s.config.HeartbeatTimeout {
				if node.Status != NodeOffline {
					node.Status = NodeOffline
					nodeChanged = true
				}
			}
			if nodeChanged {
				quarantineNodeTasks(state, nodeID, now)
				state.Audit = appendAudit(state.Audit, "node:"+nodeID, "node.offline", nodeID, "", "success", now)
				state.Nodes[nodeID] = node
			}
			nodes = append(nodes, node)
		}
		return nil
	}); err != nil {
		return err
	}
	sort.Slice(nodes, func(i, j int) bool { return nodes[i].NodeID < nodes[j].NodeID })
	writeJSON(w, http.StatusOK, map[string]any{"nodes": nodes})
	return nil
}

func (s *Server) nodeDetails(w http.ResponseWriter, nodeID, _ string) error {
	var node Node
	if err := s.store.Read(func(state PersistedState) error {
		value, ok := state.Nodes[nodeID]
		if !ok {
			return requestError{status: http.StatusNotFound, code: "NodeNotFound", message: "Node was not registered."}
		}
		node = value
		return nil
	}); err != nil {
		return err
	}
	writeJSON(w, http.StatusOK, node)
	return nil
}

func (s *Server) heartbeat(w http.ResponseWriter, r *http.Request, nodeID, correlationID string) error {
	var request Heartbeat
	if err := decodeJSON(r, &request, 32*1024); err != nil {
		return err
	}
	if request.NodeID != nodeID || (request.ConnectionID != "" && !validIdentifier(request.ConnectionID, 200)) || request.ProtocolVersion != ProtocolVersion || !validIdentifier(request.AgentVersion, 80) || !validNodeStatus(request.NodeStatus) || request.QueueLength < 0 || request.QueueLength > 100000 {
		return requestError{status: http.StatusBadRequest, code: "InvalidHeartbeat", message: "Heartbeat is invalid."}
	}
	now := time.Now().UTC()
	if err := s.store.Mutate(func(state *PersistedState) error {
		node := state.Nodes[nodeID]
		if node.NodeID == "" {
			return requestError{status: http.StatusConflict, code: "NodeNotRegistered", message: "Register the node before sending a heartbeat."}
		}
		if node.ConnectionID != request.ConnectionID {
			return requestError{status: http.StatusConflict, code: "StaleConnection", message: "The heartbeat belongs to an older Client connection."}
		}
		node.AgentVersion = request.AgentVersion
		node.ProtocolVersion = request.ProtocolVersion
		node.Status = request.NodeStatus
		node.WechatRunning = request.WechatRunning
		node.WechatLoggedIn = request.WechatLoggedIn
		node.SessionLocked = request.SessionLocked
		node.ActiveAccountID = request.ActiveAccountID
		node.QueueLength = request.QueueLength
		node.ReportingConfigVersion = request.ReportingConfigVersion
		node.LastErrorCode = request.LastErrorCode
		node.LastHeartbeatAt = &now
		node.LastCorrelationID = correlationID
		state.Nodes[nodeID] = node
		state.Audit = appendAudit(state.Audit, "node:"+nodeID, "node.heartbeat", nodeID, correlationID, "success", now)
		return nil
	}); err != nil {
		return err
	}
	writeJSON(w, http.StatusOK, NodeResponse{NodeID: nodeID, ConnectionID: request.ConnectionID, Status: request.NodeStatus, LastHeartbeatAt: &now, CorrelationID: correlationID})
	return nil
}

func (s *Server) nodeTaskRoute(w http.ResponseWriter, r *http.Request, nodeID string, parts []string, correlationID string) error {
	if len(parts) == 0 && r.Method == http.MethodGet {
		return s.pollTasks(w, r, nodeID, correlationID)
	}
	if len(parts) != 2 || r.Method != http.MethodPost {
		return requestError{status: http.StatusNotFound, code: "NotFound", message: "Resource was not found."}
	}
	taskID := parts[0]
	switch parts[1] {
	case "ack":
		return s.ackTask(w, r, nodeID, taskID, correlationID)
	case "start":
		return s.startTask(w, r, nodeID, taskID, correlationID)
	case "renew":
		return s.renewTask(w, r, nodeID, taskID, correlationID)
	case "result":
		return s.recordTaskResult(w, r, nodeID, taskID, correlationID)
	default:
		return requestError{status: http.StatusNotFound, code: "NotFound", message: "Resource was not found."}
	}
}

func (s *Server) pollTasks(w http.ResponseWriter, r *http.Request, nodeID, correlationID string) error {
	waitSeconds := queryWaitSeconds(r.URL.Query().Get("wait_seconds"))
	deadline := time.Now().Add(time.Duration(waitSeconds) * time.Second)
	for {
		tasks, err := s.claimTasks(r, nodeID, correlationID)
		if err != nil {
			return err
		}
		if len(tasks) > 0 || waitSeconds == 0 || !time.Now().Before(deadline) {
			writeJSON(w, http.StatusOK, TaskBatch{Tasks: tasks})
			return nil
		}
		select {
		case <-r.Context().Done():
			return nil
		case <-time.After(250 * time.Millisecond):
		}
	}
}

func (s *Server) claimTasks(r *http.Request, nodeID, correlationID string) ([]Task, error) {
	accountID := r.URL.Query().Get("account_id")
	if accountID != "" && !validIdentifier(accountID, 200) {
		return nil, requestError{status: http.StatusBadRequest, code: "InvalidAccount", message: "account_id is invalid."}
	}
	now := time.Now().UTC()
	var tasks []Task
	if err := s.store.Mutate(func(state *PersistedState) error {
		for taskID, value := range state.Tasks {
			task := value
			if task.NodeID != nodeID || (accountID != "" && task.AccountID != accountID) || terminal(task.Status) {
				continue
			}
			changed := false
			if task.NotAfter != nil && !now.Before(*task.NotAfter) {
				if task.Status == TaskPending {
					task.Status = TaskExpired
					task.StateVersion++
					task.UpdatedAt = now
					task.LeaseOwner = ""
					task.LeaseExpiresAt = nil
					changed = true
				} else if task.Status == TaskAccepted || task.Status == TaskRunning {
					markUnconfirmed(&task, "TaskDeadlineReached", now)
					changed = true
				}
			}
			if terminal(task.Status) {
				if changed {
					state.Tasks[taskID] = task
				}
				continue
			}
			if task.CancelRequestedAt != nil && task.Status == TaskPending && !leaseActive(task, now) {
				task.Status = TaskCancelled
				task.StateVersion++
				task.UpdatedAt = now
				task.LeaseOwner = ""
				task.LeaseExpiresAt = nil
				changed = true
			}
			if terminal(task.Status) {
				if changed {
					state.Tasks[taskID] = task
				}
				continue
			}
			if task.Status == TaskPending {
				if leaseActive(task, now) && task.LeaseOwner != nodeID {
					continue
				}
				if !leaseActive(task, now) {
					task.LeaseGeneration++
					expires := now.Add(s.config.LeaseTTL)
					task.LeaseExpiresAt = &expires
					task.LeaseOwner = nodeID
					task.UpdatedAt = now
					task.LastCorrelationID = correlationID
					changed = true
				}
			}
			if task.Status == TaskAccepted || task.Status == TaskRunning {
				if task.LeaseOwner != nodeID || !leaseActive(task, now) {
					if task.LeaseOwner == nodeID && !leaseActive(task, now) {
						markUnconfirmed(&task, "LeaseExpired", now)
						changed = true
					}
					if changed {
						state.Tasks[taskID] = task
					}
					continue
				}
			}
			if changed {
				state.Tasks[taskID] = task
			}
			if task.Status == TaskPending || task.Status == TaskAccepted || task.Status == TaskRunning {
				tasks = append(tasks, task)
			}
		}
		state.Audit = appendAudit(state.Audit, "node:"+nodeID, "task.poll", nodeID, correlationID, "success", now)
		return nil
	}); err != nil {
		return nil, err
	}
	sort.Slice(tasks, func(i, j int) bool { return tasks[i].CreatedAt.Before(tasks[j].CreatedAt) })
	return tasks, nil
}

func (s *Server) ackTask(w http.ResponseWriter, r *http.Request, nodeID, taskID, correlationID string) error {
	var request TaskAck
	if err := decodeJSON(r, &request, 8*1024); err != nil {
		return err
	}
	if request.TaskID != taskID {
		return requestError{status: http.StatusBadRequest, code: "TaskIdentityMismatch", message: "Task identity does not match the path."}
	}
	var task Task
	err := s.store.Mutate(func(state *PersistedState) error {
		value, ok := state.Tasks[taskID]
		if !ok || value.NodeID != nodeID {
			return requestError{status: http.StatusNotFound, code: "TaskNotFound", message: "Task was not found."}
		}
		task = value
		if task.AccountID != request.AccountID {
			return requestError{status: http.StatusForbidden, code: "AccountMismatch", message: "Task account does not match."}
		}
		if terminal(task.Status) {
			return nil
		}
		if task.LeaseGeneration != request.LeaseGeneration || task.LeaseOwner != nodeID || !leaseActive(task, time.Now().UTC()) {
			return requestError{status: http.StatusConflict, code: "LeaseMismatch", message: "The task lease is no longer valid."}
		}
		if task.Status != TaskPending {
			if task.Status == TaskAccepted {
				return nil
			}
			return requestError{status: http.StatusConflict, code: "InvalidTaskState", message: "The task cannot be accepted in its current state."}
		}
		if task.CancelRequestedAt != nil {
			task.Status = TaskCancelled
			task.StateVersion++
			task.LeaseOwner = ""
			task.LeaseExpiresAt = nil
		} else {
			task.Status = TaskAccepted
			task.StateVersion++
		}
		task.UpdatedAt = time.Now().UTC()
		task.LastCorrelationID = correlationID
		state.Tasks[taskID] = task
		state.Audit = appendAudit(state.Audit, "node:"+nodeID, "task.accept", taskID, correlationID, "success", task.UpdatedAt)
		return nil
	})
	if err != nil {
		return err
	}
	writeJSON(w, http.StatusOK, task)
	return nil
}

func (s *Server) startTask(w http.ResponseWriter, r *http.Request, nodeID, taskID, correlationID string) error {
	var request TaskAck
	if err := decodeJSON(r, &request, 8*1024); err != nil {
		return err
	}
	if request.TaskID != taskID {
		return requestError{status: http.StatusBadRequest, code: "TaskIdentityMismatch", message: "Task identity does not match the path."}
	}
	var task Task
	err := s.store.Mutate(func(state *PersistedState) error {
		value, ok := state.Tasks[taskID]
		if !ok || value.NodeID != nodeID {
			return requestError{status: http.StatusNotFound, code: "TaskNotFound", message: "Task was not found."}
		}
		task = value
		if task.AccountID != request.AccountID || task.LeaseGeneration != request.LeaseGeneration || task.LeaseOwner != nodeID {
			return requestError{status: http.StatusConflict, code: "LeaseMismatch", message: "The task lease is no longer valid."}
		}
		now := time.Now().UTC()
		if terminal(task.Status) {
			return nil
		}
		if task.Status != TaskAccepted {
			return requestError{status: http.StatusConflict, code: "InvalidTaskState", message: "The task must be accepted before execution."}
		}
		if task.CancelRequestedAt != nil {
			task.Status = TaskCancelled
			task.StateVersion++
			task.LeaseOwner = ""
			task.LeaseExpiresAt = nil
		} else if task.NotAfter != nil && !now.Before(*task.NotAfter) {
			task.Status = TaskExpired
			task.StateVersion++
			task.LeaseOwner = ""
			task.LeaseExpiresAt = nil
		} else if !leaseActive(task, now) {
			return requestError{status: http.StatusConflict, code: "LeaseExpired", message: "The task lease expired before execution started."}
		} else {
			task.Status = TaskRunning
			task.StateVersion++
		}
		task.UpdatedAt = now
		task.LastCorrelationID = correlationID
		state.Tasks[taskID] = task
		state.Audit = appendAudit(state.Audit, "node:"+nodeID, "task.start", taskID, correlationID, "success", now)
		return nil
	})
	if err != nil {
		return err
	}
	writeJSON(w, http.StatusOK, task)
	return nil
}

func (s *Server) renewTask(w http.ResponseWriter, r *http.Request, nodeID, taskID, correlationID string) error {
	var request TaskAck
	if err := decodeJSON(r, &request, 8*1024); err != nil {
		return err
	}
	if request.TaskID != taskID {
		return requestError{status: http.StatusBadRequest, code: "TaskIdentityMismatch", message: "Task identity does not match the path."}
	}
	var task Task
	if err := s.store.Mutate(func(state *PersistedState) error {
		value, ok := state.Tasks[taskID]
		if !ok || value.NodeID != nodeID {
			return requestError{status: http.StatusNotFound, code: "TaskNotFound", message: "Task was not found."}
		}
		task = value
		if task.AccountID != request.AccountID || task.LeaseGeneration != request.LeaseGeneration || task.LeaseOwner != nodeID {
			return requestError{status: http.StatusConflict, code: "LeaseMismatch", message: "The task lease is no longer valid."}
		}
		if task.CancelRequestedAt != nil {
			return requestError{status: http.StatusConflict, code: "CancelRequested", message: "The task lease cannot be renewed after cancellation was requested."}
		}
		now := time.Now().UTC()
		if task.Status != TaskAccepted && task.Status != TaskRunning || !leaseActive(task, now) {
			return requestError{status: http.StatusConflict, code: "LeaseExpired", message: "The task lease is no longer renewable."}
		}
		expires := now.Add(s.config.LeaseTTL)
		task.LeaseExpiresAt = &expires
		task.StateVersion++
		task.UpdatedAt = now
		task.LastCorrelationID = correlationID
		state.Tasks[taskID] = task
		state.Audit = appendAudit(state.Audit, "node:"+nodeID, "task.renew", taskID, correlationID, "success", now)
		return nil
	}); err != nil {
		return err
	}
	writeJSON(w, http.StatusOK, task)
	return nil
}

func (s *Server) recordTaskResult(w http.ResponseWriter, r *http.Request, nodeID, taskID, correlationID string) error {
	var result TaskResult
	if err := decodeJSON(r, &result, MaxTaskResultBytes+32*1024); err != nil {
		return err
	}
	if result.TaskID != taskID || !terminal(result.Status) {
		return requestError{status: http.StatusBadRequest, code: "InvalidTaskResult", message: "Task result or terminal status is invalid."}
	}
	if len(result.Content) > MaxTaskResultBytes {
		return requestError{status: http.StatusRequestEntityTooLarge, code: "ContentTooLarge", message: "The task result content is too large."}
	}
	var task Task
	err := s.store.Mutate(func(state *PersistedState) error {
		value, ok := state.Tasks[taskID]
		if !ok || value.NodeID != nodeID {
			return requestError{status: http.StatusNotFound, code: "TaskNotFound", message: "Task was not found."}
		}
		task = value
		if task.AccountID != result.AccountID {
			return requestError{status: http.StatusForbidden, code: "AccountMismatch", message: "Task account does not match."}
		}
		if hasTaskContent(result.Content) && !isReadTaskKind(task.Kind) {
			return requestError{status: http.StatusForbidden, code: "ContentNotAllowed", message: "Only read task results may contain content."}
		}
		if terminal(task.Status) {
			if task.Result != nil && task.Result.Status == result.Status && task.Result.ErrorCode == result.ErrorCode {
				return nil
			}
			return requestError{status: http.StatusConflict, code: "TerminalState", message: "The task already has a different terminal result."}
		}
		if task.LeaseGeneration != result.LeaseGeneration || task.LeaseOwner != nodeID {
			return requestError{status: http.StatusConflict, code: "LeaseMismatch", message: "The result belongs to an expired task lease."}
		}
		if task.Status != TaskAccepted && task.Status != TaskRunning {
			return requestError{status: http.StatusConflict, code: "InvalidTaskState", message: "The task is not awaiting a final result."}
		}
		result.CorrelationID = firstNonEmpty(result.CorrelationID, correlationID)
		task.Result = &result
		task.Status = result.Status
		task.StateVersion++
		task.UpdatedAt = time.Now().UTC()
		task.LeaseOwner = ""
		task.LeaseExpiresAt = nil
		task.LastCorrelationID = correlationID
		state.Tasks[taskID] = task
		state.Audit = appendAudit(state.Audit, "node:"+nodeID, "task.result", taskID, correlationID, "success", task.UpdatedAt)
		return nil
	})
	if err != nil {
		return err
	}
	if task.Kind == "read-messages" {
		go s.handleAITaskResult(taskID)
	}
	writeJSON(w, http.StatusOK, task)
	return nil
}

func (s *Server) readRoute(w http.ResponseWriter, r *http.Request, correlationID string) error {
	username, err := s.authenticateWeb(r)
	if err != nil {
		return err
	}
	parts := pathParts(r.URL.Path)
	if len(parts) != 3 || parts[0] != "v1" || parts[1] != "reads" || r.Method != http.MethodPost {
		return requestError{status: http.StatusNotFound, code: "NotFound", message: "Resource was not found."}
	}
	kind := "read-" + parts[2]
	if !isReadTaskKind(kind) {
		return requestError{status: http.StatusNotFound, code: "NotFound", message: "Resource was not found."}
	}
	var request ReadTaskSubmission
	if err := decodeJSON(r, &request, 64*1024); err != nil {
		return err
	}
	payload, err := readTaskPayload(kind, request)
	if err != nil {
		return err
	}
	return s.createTaskSubmission(w, TaskSubmission{
		NodeID: request.NodeID, AccountID: request.AccountID, Kind: kind,
		IdempotencyKey: request.IdempotencyKey, Payload: payload, NotAfter: request.NotAfter,
	}, username, correlationID)
}

type readSessionsPayload struct {
	Limit  int `json:"limit"`
	Offset int `json:"offset"`
}

type readContactsPayload struct {
	Limit      int    `json:"limit"`
	Offset     int    `json:"offset"`
	Contains   string `json:"contains,omitempty"`
	GroupsOnly bool   `json:"groups_only"`
}

type readMessagesPayload struct {
	Limit          int    `json:"limit"`
	Offset         int    `json:"offset"`
	ChatID         string `json:"chat_id"`
	IncludeContent bool   `json:"include_content"`
}

func isReadTaskKind(kind string) bool {
	switch kind {
	case "read-sessions", "read-contacts", "read-messages":
		return true
	default:
		return false
	}
}

func readTaskPayload(kind string, request ReadTaskSubmission) (jsonRaw, error) {
	limit := request.Limit
	if limit == 0 {
		limit = 50
	}
	if limit < 1 || limit > 200 || request.Offset < 0 {
		return nil, requestError{status: http.StatusBadRequest, code: "InvalidPagination", message: "limit must be 1..200 and offset must be non-negative."}
	}
	if request.Contains != "" && !validIdentifier(request.Contains, 200) {
		return nil, requestError{status: http.StatusBadRequest, code: "InvalidReadRequest", message: "contains is invalid."}
	}
	switch kind {
	case "read-sessions":
		if request.Contains != "" || request.GroupsOnly != nil || request.ChatID != "" || request.IncludeContent {
			return nil, requestError{status: http.StatusBadRequest, code: "InvalidReadRequest", message: "The sessions read request contains unsupported fields."}
		}
		return json.Marshal(readSessionsPayload{Limit: limit, Offset: request.Offset})
	case "read-contacts":
		if request.GroupsOnly == nil || request.ChatID != "" || request.IncludeContent {
			return nil, requestError{status: http.StatusBadRequest, code: "InvalidReadRequest", message: "contacts requires groups_only and does not accept chat_id or include_content."}
		}
		return json.Marshal(readContactsPayload{Limit: limit, Offset: request.Offset, Contains: request.Contains, GroupsOnly: *request.GroupsOnly})
	case "read-messages":
		if !validIdentifier(request.ChatID, 512) || request.GroupsOnly != nil || request.Contains != "" {
			return nil, requestError{status: http.StatusBadRequest, code: "InvalidReadRequest", message: "messages requires a stable chat_id and does not accept contains or groups_only."}
		}
		return json.Marshal(readMessagesPayload{Limit: limit, Offset: request.Offset, ChatID: request.ChatID, IncludeContent: request.IncludeContent})
	default:
		return nil, requestError{status: http.StatusNotFound, code: "NotFound", message: "Resource was not found."}
	}
}

func validTaskPayload(kind string, payload jsonRaw) bool {
	if kind == "send-text" {
		return validSendTextPayload(payload)
	}
	if kind == "sync-data" {
		var value struct {
			StreamKey string `json:"stream_key"`
			Reason    string `json:"reason"`
		}
		return decodeRaw(payload, &value) && value.StreamKey == "messages" && (value.Reason == "" || validIdentifier(value.Reason, 80))
	}
	if !isReadTaskKind(kind) || len(payload) == 0 || len(payload) > 64*1024 || !json.Valid(payload) {
		return false
	}
	switch kind {
	case "read-sessions":
		var value readSessionsPayload
		return decodeRaw(payload, &value) && validPagination(value.Limit, value.Offset)
	case "read-contacts":
		var value readContactsPayload
		return decodeRaw(payload, &value) && validPagination(value.Limit, value.Offset) && (value.Contains == "" || validIdentifier(value.Contains, 200))
	case "read-messages":
		var value readMessagesPayload
		return decodeRaw(payload, &value) && validPagination(value.Limit, value.Offset) && validIdentifier(value.ChatID, 512)
	default:
		return false
	}
}

func decodeRaw(payload jsonRaw, target any) bool {
	decoder := json.NewDecoder(strings.NewReader(string(payload)))
	decoder.DisallowUnknownFields()
	if decoder.Decode(target) != nil {
		return false
	}
	var extra any
	return decoder.Decode(&extra) == io.EOF
}

func validPagination(limit, offset int) bool { return limit >= 1 && limit <= 200 && offset >= 0 }

func hasTaskContent(content jsonRaw) bool { return len(content) != 0 && string(content) != "null" }

func (s *Server) taskRoute(w http.ResponseWriter, r *http.Request, correlationID string) error {
	username, err := s.authenticateWeb(r)
	if err != nil {
		return err
	}
	parts := pathParts(r.URL.Path)
	if len(parts) == 2 && parts[0] == "v1" && parts[1] == "tasks" {
		switch r.Method {
		case http.MethodGet:
			return s.listTasks(w, r)
		case http.MethodPost:
			return s.createTask(w, r, username, correlationID)
		default:
			return requestError{status: http.StatusMethodNotAllowed, code: "MethodNotAllowed", message: "Method is not allowed."}
		}
	}
	if len(parts) == 3 && r.Method == http.MethodGet {
		return s.getTask(w, parts[2])
	}
	if len(parts) == 4 && parts[3] == "cancel" && r.Method == http.MethodPost {
		return s.cancelTask(w, parts[2], username, correlationID)
	}
	if len(parts) == 4 && parts[3] == "resume" && r.Method == http.MethodPost {
		return s.resumeTask(w, parts[2], username, correlationID)
	}
	return requestError{status: http.StatusNotFound, code: "NotFound", message: "Resource was not found."}
}

func (s *Server) createTask(w http.ResponseWriter, r *http.Request, username, correlationID string) error {
	var request TaskSubmission
	if err := decodeJSON(r, &request, 128*1024); err != nil {
		return err
	}
	return s.createTaskSubmission(w, request, username, correlationID)
}

func (s *Server) createTaskSubmission(w http.ResponseWriter, request TaskSubmission, username, correlationID string) error {
	if !validIdentifier(request.NodeID, 200) || !validIdentifier(request.AccountID, 200) || !validIdentifier(request.Kind, 80) || !validIdentifier(request.IdempotencyKey, 128) {
		return requestError{status: http.StatusBadRequest, code: "InvalidTask", message: "Task identity fields are invalid."}
	}
	if !validTaskPayload(request.Kind, request.Payload) {
		return requestError{status: http.StatusBadRequest, code: "UnsupportedTask", message: "The task kind or payload is not supported."}
	}
	if request.NotAfter != nil && !request.NotAfter.After(time.Now().UTC()) {
		return requestError{status: http.StatusBadRequest, code: "InvalidDeadline", message: "not_after must be in the future."}
	}
	var response TaskSubmissionResponse
	err := s.store.Mutate(func(state *PersistedState) error {
		node, registered := state.Nodes[request.NodeID]
		if !registered {
			return requestError{status: http.StatusConflict, code: "NodeNotReady", message: "The target node is not registered."}
		}
		if !nodeHasAccount(node, request.AccountID) {
			return requestError{status: http.StatusConflict, code: "AccountNotReady", message: "The target account is not currently registered on the node."}
		}
		payloadKey := string(request.Payload)
		for _, existing := range state.Tasks {
			if existing.NodeID != request.NodeID || existing.AccountID != request.AccountID || existing.IdempotencyKey != request.IdempotencyKey {
				continue
			}
			if existing.Kind != request.Kind || string(existing.Payload) != payloadKey {
				return requestError{status: http.StatusConflict, code: "IdempotencyConflict", message: "The idempotency key is already bound to different task parameters."}
			}
			response = TaskSubmissionResponse{TaskID: existing.TaskID, Status: existing.Status, Duplicate: true, StateVersion: existing.StateVersion}
			return nil
		}
		now := time.Now().UTC()
		status := TaskPending
		if !nodeAvailable(node, now, s.config.HeartbeatTimeout) {
			status = TaskWaitingForClient
		}
		task := Task{
			TaskID: randomID(), NodeID: request.NodeID, AccountID: request.AccountID, Kind: request.Kind,
			IdempotencyKey: request.IdempotencyKey, Payload: append(jsonRaw(nil), request.Payload...), NotAfter: request.NotAfter,
			Status: status, StateVersion: 1, CreatedAt: now, UpdatedAt: now, LastCorrelationID: correlationID,
		}
		state.Tasks[task.TaskID] = task
		state.Audit = appendAudit(state.Audit, "user:"+username, "task.create", task.TaskID, correlationID, "success", now)
		response = TaskSubmissionResponse{TaskID: task.TaskID, Status: task.Status, Duplicate: false, StateVersion: task.StateVersion}
		return nil
	})
	if err != nil {
		return err
	}
	writeJSON(w, http.StatusAccepted, response)
	return nil
}

func (s *Server) listTasks(w http.ResponseWriter, r *http.Request) error {
	nodeID := r.URL.Query().Get("node_id")
	accountID := r.URL.Query().Get("account_id")
	limit := queryLimit(r.URL.Query().Get("limit"))
	var tasks []Task
	if err := s.store.Read(func(state PersistedState) error {
		for _, task := range state.Tasks {
			if nodeID != "" && task.NodeID != nodeID || accountID != "" && task.AccountID != accountID {
				continue
			}
			tasks = append(tasks, task)
		}
		return nil
	}); err != nil {
		return err
	}
	sort.Slice(tasks, func(i, j int) bool { return tasks[i].CreatedAt.After(tasks[j].CreatedAt) })
	if len(tasks) > limit {
		tasks = tasks[:limit]
	}
	writeJSON(w, http.StatusOK, map[string]any{"tasks": tasks})
	return nil
}

func (s *Server) getTask(w http.ResponseWriter, taskID string) error {
	if !validIdentifier(taskID, 200) {
		return requestError{status: http.StatusBadRequest, code: "InvalidTask", message: "Task ID is invalid."}
	}
	var task Task
	if err := s.store.Read(func(state PersistedState) error {
		value, ok := state.Tasks[taskID]
		if !ok {
			return requestError{status: http.StatusNotFound, code: "TaskNotFound", message: "Task was not found."}
		}
		task = value
		return nil
	}); err != nil {
		return err
	}
	writeJSON(w, http.StatusOK, task)
	return nil
}

func (s *Server) cancelTask(w http.ResponseWriter, taskID, username, correlationID string) error {
	var task Task
	err := s.store.Mutate(func(state *PersistedState) error {
		value, ok := state.Tasks[taskID]
		if !ok {
			return requestError{status: http.StatusNotFound, code: "TaskNotFound", message: "Task was not found."}
		}
		task = value
		if terminal(task.Status) {
			return nil
		}
		now := time.Now().UTC()
		if task.CancelRequestedAt == nil {
			task.CancelRequestedAt = &now
			task.StateVersion++
			if (task.Status == TaskPending || task.Status == TaskWaitingForClient) && !leaseActive(task, now) {
				task.Status = TaskCancelled
				task.LeaseOwner = ""
				task.LeaseExpiresAt = nil
			}
			task.UpdatedAt = now
			state.Audit = appendAudit(state.Audit, "user:"+username, "task.cancel-request", taskID, correlationID, "success", now)
			state.Tasks[taskID] = task
		}
		return nil
	})
	if err != nil {
		return err
	}
	writeJSON(w, http.StatusOK, task)
	return nil
}

func (s *Server) resumeTask(w http.ResponseWriter, taskID, username, correlationID string) error {
	var task Task
	err := s.store.Mutate(func(state *PersistedState) error {
		value, ok := state.Tasks[taskID]
		if !ok {
			return requestError{status: http.StatusNotFound, code: "TaskNotFound", message: "Task was not found."}
		}
		task = value
		if terminal(task.Status) {
			return requestError{status: http.StatusConflict, code: "TerminalState", message: "The task cannot be resumed after it reached a terminal state."}
		}
		if task.Status != TaskWaitingForClient {
			return requestError{status: http.StatusConflict, code: "InvalidTaskState", message: "Only tasks waiting for a Client can be resumed."}
		}
		if task.CancelRequestedAt != nil {
			return requestError{status: http.StatusConflict, code: "CancelRequested", message: "The task was cancelled and cannot be resumed."}
		}
		node, exists := state.Nodes[task.NodeID]
		now := time.Now().UTC()
		if !exists || !nodeAvailable(node, now, s.config.HeartbeatTimeout) || !nodeHasAccount(node, task.AccountID) {
			return requestError{status: http.StatusConflict, code: "ClientNotReady", message: "The target Client is not connected with the requested account."}
		}
		if task.NotAfter != nil && !now.Before(*task.NotAfter) {
			task.Status = TaskExpired
			task.StateVersion++
			task.UpdatedAt = now
		} else {
			task.Status = TaskPending
			task.StateVersion++
			task.UpdatedAt = now
			task.LastCorrelationID = correlationID
		}
		state.Tasks[taskID] = task
		state.Audit = appendAudit(state.Audit, "user:"+username, "task.resume", taskID, correlationID, "success", now)
		return nil
	})
	if err != nil {
		return err
	}
	writeJSON(w, http.StatusOK, task)
	return nil
}

func (s *Server) eventRoute(w http.ResponseWriter, r *http.Request, _ string) error {
	if r.Method != http.MethodGet {
		return requestError{status: http.StatusMethodNotAllowed, code: "MethodNotAllowed", message: "Method is not allowed."}
	}
	if _, err := s.authenticateWeb(r); err != nil {
		return err
	}
	return s.listEvents(w, r)
}

func (s *Server) ingestEvent(w http.ResponseWriter, r *http.Request, nodeID, correlationID string) error {
	var event MessageEvent
	if err := decodeJSON(r, &event, 64*1024); err != nil {
		return err
	}
	if event.NodeID != nodeID || !validIdentifier(event.AccountID, 200) || !validIdentifier(event.ChatID, 512) || event.EventSeq <= 0 || !validIdentifier(event.EventType, 80) || event.OccurredAt.IsZero() || event.ConfigVersion <= 0 || event.AuthorizationVersion <= 0 {
		return requestError{status: http.StatusBadRequest, code: "InvalidEvent", message: "Event identity or authorization metadata is invalid."}
	}
	if event.ChatType != ChatGroup && event.ChatType != ChatPrivate {
		return requestError{status: http.StatusBadRequest, code: "InvalidChatType", message: "Only group and private chat events are supported."}
	}
	if event.EventType != "message" {
		return requestError{status: http.StatusBadRequest, code: "UnsupportedEvent", message: "Only message events are supported in v1."}
	}
	if len(event.Content) > 16*1024 {
		return requestError{status: http.StatusBadRequest, code: "EventTooLarge", message: "Event content is too large."}
	}
	if !event.Authorized {
		return requestError{status: http.StatusForbidden, code: "ReportingNotAuthorized", message: "The node did not authorize this event."}
	}
	contentHash := eventHash(event)
	var receipt EventReceipt
	var stored StoredEvent
	err := s.store.Mutate(func(state *PersistedState) error {
		for _, existing := range state.Events {
			if existing.NodeID != event.NodeID || existing.AccountID != event.AccountID || existing.ChatID != event.ChatID || existing.EventSeq != event.EventSeq {
				continue
			}
			if existing.ContentHash != contentHash {
				return requestError{status: http.StatusConflict, code: "EventIdempotencyConflict", message: "The event sequence is already bound to different content."}
			}
			receipt = EventReceipt{Accepted: true, Duplicate: true, EventID: existing.EventID}
			return nil
		}
		now := time.Now().UTC()
		stored = StoredEvent{MessageEvent: event, EventID: randomID(), ContentHash: contentHash, ReceivedAt: now}
		state.Events = append(state.Events, stored)
		if len(state.Events) > 10000 {
			state.Events = state.Events[len(state.Events)-10000:]
		}
		state.Audit = appendAudit(state.Audit, "node:"+nodeID, "event.accept", event.AccountID+":"+event.ChatID, correlationID, "success", now)
		receipt = EventReceipt{Accepted: true, EventID: stored.EventID}
		return nil
	})
	if err != nil {
		return err
	}
	if !receipt.Duplicate {
		go s.enqueueEventRuns(stored)
	}
	writeJSON(w, http.StatusAccepted, receipt)
	return nil
}

func (s *Server) listEvents(w http.ResponseWriter, r *http.Request) error {
	query := r.URL.Query()
	nodeID, accountID, chatID := query.Get("node_id"), query.Get("account_id"), query.Get("chat_id")
	limit := queryLimit(query.Get("limit"))
	var events []StoredEvent
	if err := s.store.Read(func(state PersistedState) error {
		for index := len(state.Events) - 1; index >= 0 && len(events) < limit; index-- {
			event := state.Events[index]
			if nodeID != "" && event.NodeID != nodeID || accountID != "" && event.AccountID != accountID || chatID != "" && event.ChatID != chatID {
				continue
			}
			events = append(events, event)
		}
		return nil
	}); err != nil {
		return err
	}
	writeJSON(w, http.StatusOK, map[string]any{"events": events})
	return nil
}

func (s *Server) auditRoute(w http.ResponseWriter, r *http.Request, _ string) error {
	if _, err := s.authenticateWeb(r); err != nil {
		return err
	}
	limit := queryLimit(r.URL.Query().Get("limit"))
	var audit []AuditEntry
	if err := s.store.Read(func(state PersistedState) error {
		for index := len(state.Audit) - 1; index >= 0 && len(audit) < limit; index-- {
			audit = append(audit, state.Audit[index])
		}
		return nil
	}); err != nil {
		return err
	}
	writeJSON(w, http.StatusOK, map[string]any{"audit": audit})
	return nil
}

func (s *Server) authenticateNode(r *http.Request) (string, error) {
	if !s.clientCertificateAllowed(r) {
		return "", requestError{status: http.StatusUnauthorized, code: "Unauthorized", message: "Node certificate authentication failed."}
	}
	token := bearerToken(r)
	if token == "" {
		return "", requestError{status: http.StatusUnauthorized, code: "Unauthorized", message: "Node authentication is required."}
	}
	provided := sha256.Sum256([]byte(token))
	for nodeID, expected := range s.nodeTokenHashes {
		if subtle.ConstantTimeCompare(provided[:], expected[:]) == 1 {
			return nodeID, nil
		}
	}
	return "", requestError{status: http.StatusUnauthorized, code: "Unauthorized", message: "Node authentication failed."}
}

func (s *Server) authenticateWeb(r *http.Request) (string, error) {
	token := bearerToken(r)
	if token == "" {
		return "", requestError{status: http.StatusUnauthorized, code: "Unauthorized", message: "Web authentication is required."}
	}
	now := time.Now().UTC()
	s.sessionMu.Lock()
	value, ok := s.sessions[token]
	if ok && !now.Before(value.Expires) {
		delete(s.sessions, token)
		ok = false
	}
	s.sessionMu.Unlock()
	if !ok {
		return "", requestError{status: http.StatusUnauthorized, code: "Unauthorized", message: "Web session is missing or expired."}
	}
	return value.Username, nil
}

func (s *Server) appendAudit(principal, action, resource, correlationID, outcome string) error {
	return s.store.Mutate(func(state *PersistedState) error {
		state.Audit = appendAudit(state.Audit, principal, action, resource, correlationID, outcome, time.Now().UTC())
		return nil
	})
}

func (s *Server) serveFrontend(w http.ResponseWriter, name string) {
	if name == "" || name == "." || strings.Contains(name, "..") {
		writeError(w, requestError{status: http.StatusNotFound, code: "NotFound", message: "Resource was not found."}, randomID())
		return
	}
	content, err := fs.ReadFile(webassets.Dist, name)
	if err != nil {
		writeError(w, requestError{status: http.StatusNotFound, code: "NotFound", message: "Resource was not found."}, randomID())
		return
	}
	contentType := mime.TypeByExtension(path.Ext(name))
	if contentType == "" {
		contentType = "application/octet-stream"
	}
	w.Header().Set("Content-Type", contentType+"; charset=utf-8")
	w.Header().Set("Content-Security-Policy", "default-src 'self'; script-src 'self'; style-src 'self'; connect-src 'self'; frame-ancestors 'none'; base-uri 'none'; form-action 'self'")
	w.WriteHeader(http.StatusOK)
	_, _ = w.Write(content)
}

func decodeJSON(r *http.Request, target any, maxBytes int64) error {
	defer r.Body.Close()
	decoder := json.NewDecoder(io.LimitReader(r.Body, maxBytes))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(target); err != nil {
		return requestError{status: http.StatusBadRequest, code: "InvalidRequest", message: "Request JSON is invalid."}
	}
	var extra any
	if decoder.Decode(&extra) != io.EOF {
		return requestError{status: http.StatusBadRequest, code: "InvalidRequest", message: "Request must contain one JSON value."}
	}
	return nil
}

func writeJSON(w http.ResponseWriter, status int, value any) {
	w.Header().Set("Content-Type", "application/json; charset=utf-8")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(value)
}

func writeError(w http.ResponseWriter, err error, correlationID string) {
	requestErr, ok := err.(requestError)
	if !ok {
		requestErr = requestError{status: http.StatusInternalServerError, code: "InternalError", message: "Request failed; use the correlation ID for diagnosis."}
	}
	writeJSON(w, requestErr.status, map[string]any{
		"error":          map[string]string{"code": requestErr.code, "message": requestErr.message},
		"correlation_id": correlationID,
	})
}

func pathParts(path string) []string {
	trimmed := strings.Trim(path, "/")
	if trimmed == "" {
		return nil
	}
	parts := strings.Split(trimmed, "/")
	for index, part := range parts {
		decoded, err := url.PathUnescape(part)
		if err == nil {
			parts[index] = decoded
		}
	}
	return parts
}

func bearerToken(r *http.Request) string {
	value := r.Header.Get("Authorization")
	if len(value) < 8 || !strings.EqualFold(value[:7], "Bearer ") {
		return ""
	}
	return strings.TrimSpace(value[7:])
}

func validIdentifier(value string, max int) bool {
	return value != "" && len(value) <= max && !strings.ContainsAny(value, "\r\n\x00")
}

func terminal(status TaskStatus) bool {
	switch status {
	case TaskSucceeded, TaskFailed, TaskCancelled, TaskExpired, TaskResultUnconfirmed:
		return true
	default:
		return false
	}
}

func leaseActive(task Task, now time.Time) bool {
	return task.LeaseOwner != "" && task.LeaseExpiresAt != nil && now.Before(*task.LeaseExpiresAt)
}

func markUnconfirmed(task *Task, reason string, now time.Time) {
	task.Status = TaskResultUnconfirmed
	task.StateVersion++
	task.UpdatedAt = now
	task.LeaseOwner = ""
	task.LeaseExpiresAt = nil
	task.Result = &TaskResult{TaskID: task.TaskID, AccountID: task.AccountID, LeaseGeneration: task.LeaseGeneration, Status: TaskResultUnconfirmed, ErrorCode: reason, Message: "The control plane could not prove that the previous executor stopped before the lease expired.", CorrelationID: task.LastCorrelationID}
}

func quarantineNodeTasks(state *PersistedState, nodeID string, now time.Time) {
	for taskID, value := range state.Tasks {
		if value.NodeID != nodeID || terminal(value.Status) {
			continue
		}
		task := value
		switch task.Status {
		case TaskPending, TaskWaitingForClient:
			if task.Status == TaskWaitingForClient && task.LeaseOwner == "" && task.LeaseExpiresAt == nil {
				continue
			}
			task.Status = TaskWaitingForClient
			task.StateVersion++
			task.UpdatedAt = now
			task.LeaseOwner = ""
			task.LeaseExpiresAt = nil
		case TaskAccepted, TaskRunning:
			markUnconfirmed(&task, "ClientDisconnected", now)
		default:
			continue
		}
		state.Tasks[taskID] = task
	}
}

func validCapabilities(capabilities []string) bool {
	for _, capability := range capabilities {
		if !validIdentifier(capability, 80) {
			return false
		}
	}
	return true
}

func validAccountSummaries(accounts []AccountSummary) bool {
	seen := make(map[string]struct{}, len(accounts))
	for _, account := range accounts {
		if !validIdentifier(account.AccountID, 200) || account.AllowedGroupCount < 0 || account.AllowedPrivateCount < 0 {
			return false
		}
		if _, exists := seen[account.AccountID]; exists {
			return false
		}
		seen[account.AccountID] = struct{}{}
	}
	return len(accounts) <= 100
}

func validNodeStatus(status NodeStatus) bool {
	switch status {
	case NodeRegistered, NodeOnline, NodeDegraded, NodeOffline, NodeSessionLocked, NodeWechatNotRunning, NodeWechatNotLogged:
		return true
	default:
		return false
	}
}

func nodeHasAccount(node Node, accountID string) bool {
	for _, account := range node.Accounts {
		if account.AccountID == accountID && account.Active && account.Verified {
			return true
		}
	}
	return false
}

func nodeAvailable(node Node, now time.Time, heartbeatTimeout time.Duration) bool {
	return node.Status != NodeOffline && node.LastHeartbeatAt != nil && now.Sub(*node.LastHeartbeatAt) <= heartbeatTimeout
}

func validSendTextPayload(payload jsonRaw) bool {
	if len(payload) == 0 || len(payload) > 64*1024 || !json.Valid(payload) {
		return false
	}
	var value struct {
		TargetID  string `json:"target_id"`
		Text      string `json:"text"`
		Confirmed bool   `json:"confirmed"`
	}
	if json.Unmarshal(payload, &value) != nil {
		return false
	}
	return validIdentifier(value.TargetID, 512) && len(value.Text) > 0 && len(value.Text) <= 4000 && value.Confirmed
}

func eventHash(event MessageEvent) string {
	value := struct {
		NodeID        string    `json:"node_id"`
		AccountID     string    `json:"account_id"`
		ChatID        string    `json:"chat_id"`
		ChatType      ChatType  `json:"chat_type"`
		EventSeq      int64     `json:"event_seq"`
		EventType     string    `json:"event_type"`
		OccurredAt    time.Time `json:"occurred_at"`
		Content       string    `json:"content"`
		ConfigVersion int64     `json:"config_version"`
	}{event.NodeID, event.AccountID, event.ChatID, event.ChatType, event.EventSeq, event.EventType, event.OccurredAt, event.Content, event.ConfigVersion}
	data, _ := json.Marshal(value)
	digest := sha256.Sum256(data)
	return hex.EncodeToString(digest[:])
}

func appendAudit(entries []AuditEntry, principal, action, resource, correlationID, outcome string, at time.Time) []AuditEntry {
	entries = append(entries, AuditEntry{ID: randomID(), At: at, Principal: principal, Action: action, Resource: resource, CorrelationID: correlationID, Outcome: outcome})
	if len(entries) > 10000 {
		entries = entries[len(entries)-10000:]
	}
	return entries
}

func queryLimit(raw string) int {
	if value, err := strconv.Atoi(raw); err == nil && value >= 1 && value <= 200 {
		return value
	}
	return 50
}

func queryWaitSeconds(raw string) int {
	if value, err := strconv.Atoi(raw); err == nil && value >= 1 && value <= 30 {
		return value
	}
	return 0
}

func firstNonEmpty(value, fallback string) string {
	if value != "" {
		return value
	}
	return fallback
}

func randomID() string {
	buffer := make([]byte, 16)
	if _, err := rand.Read(buffer); err != nil {
		return fmt.Sprintf("fallback-%d", time.Now().UnixNano())
	}
	return base64.RawURLEncoding.EncodeToString(buffer)
}