Cyrene_v2/internal/services/gateway/server.go

// Package gateway implements the Cyrene v2 HTTP gateway: the only socket the
// single binary opens. It serves the versioned REST API (/api/v1), the realtime
// WebSocket hub (/ws/chat) and the RBAC-protected admin API (/api/v1/admin/*)
// all on one port (:8080).
//
// It is deliberately lean. All backend capability flows through injected
// dependencies (construction-time dependency injection) and the in-process bus,
// never through cross-process IPC, so the package never opens a listener of its
// own beyond the gateway port.
//
// Client contract (DevDocs/02-development-constraints.md §九, §十一):
//   - REST responses are {data} on success and {error:{code,message}} on error.
//   - Authentication is JWT bearer (internal/auth); authorization is the
//     owner>admin>viewer RBAC hierarchy (internal/auth).
//   - Admin writes are audited (who/what/when/result).
//   - Realtime events follow DevDocs/design/07 §3 (WS protocol).
package gateway

import (
	"context"
	"fmt"
	"io"
	"io/fs"
	"log/slog"
	"net"
	"net/http"
	"strconv"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
	capreg "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/iot"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
	personareg "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/registry"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/task"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/services/automations"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/services/backthink"
)

// source is the log provenance tag for every record this package emits.
const source = "gateway"

// Authenticator abstracts the credential + RBAC service. *auth.JWTService is the
// production implementation; it is an interface so tests substitute a mock.
type Authenticator interface {
	// Sign issues a token for sub with the given role and lifetime.
	Sign(sub string, role auth.Role, ttl time.Duration) (string, error)
	// Verify validates a token and returns its claims.
	Verify(token string) (*auth.Claims, error)
	// Authorize returns an HTTP middleware enforcing at least one required role.
	Authorize(required ...auth.Role) func(http.Handler) http.Handler
}

// CapabilityStore abstracts the global capability registry used to enumerate
// plugins and channels in the admin API. *capability.Registry is the production
// implementation.
type CapabilityStore interface {
	// List returns every capability whose Kind matches, or all when kind is
	// empty.
	List(kind capreg.Kind) []capreg.Capability
	// SetPolicy replaces the reachability markers for the named capability.
	SetPolicy(name string, p capreg.ToolPolicy) error
}

// Chatter runs a single agent turn. *loop.Loop is the production implementation.
type Chatter interface {
	// Run drives one turn for sessionID and returns the final answer.
	Run(ctx context.Context, sessionID string, messages []model.Message) (*loop.Result, error)
}

// SessionStore lists known channel sessions. *session.SessionManager is the
// production implementation.
type SessionStore interface {
	List(ctx context.Context) ([]session.Session, error)
}

// PersonaStore lists registered characters (each bound one-to-one to a persona).
// *persona/registry.Registry is the production implementation. The gateway maps
// each Character to its wire summary; the interface deliberately returns the
// domain type so the wire shape stays out of the store contract.
type PersonaStore interface {
	List() []personareg.Character
}

// MemoryStore lists memory entries. *runtime/memory.Store is the production
// implementation. The gateway maps each MemoryItem to its wire summary.
type MemoryStore interface {
	List(ctx context.Context) ([]memory.MemoryItem, error)
}

// IoTStore lists IoT devices. *iot.Manager is the production implementation.
type IoTStore interface {
	List(ctx context.Context) ([]iot.Device, error)
}

// SubagentAuditor audits a sub-agent's conversation context for the subagent_ask
// WS route. *orchestration.Scheduler is the production implementation.
type SubagentAuditor interface {
	AuditContext(ctx context.Context, id string, mode orchestration.AuditMode, n int, query string) (*orchestration.AuditResult, error)
}

// TaskOutputStore queries a task's output for the task_output_query WS route.
// *task.Manager is the production implementation.
type TaskOutputStore interface {
	TaskOutput(ctx context.Context, id string) (task.OutputEvent, error)
}

// AutomationStore enumerates the automations and reminders for the admin list.
// *automations.Service is the production implementation. The gateway maps each
// record to its wire shape.
type AutomationStore interface {
	List() []automations.AutomationRecord
}

// UserStore enumerates the accounts and API keys visible to the admin panel.
// The production implementation reads the configured admin account plus the
// auth.APIKey table. The gateway maps each summary to its wire shape.
type UserStore interface {
	List() []auth.UserSummary
}

// BackthinkStore enumerates the persisted background-think records for the
// M9/M10 monitor panel. *backthink.Backthink is the production implementation.
type BackthinkStore interface {
	List(ctx context.Context, limit int) ([]backthink.Record, error)
}

// Deps carries the gateway's injected dependencies. Every field is nil-tolerant
// except Auth, which is required (authentication backs login and every RBAC
// decision). Registry/Loop are satisfied by the concrete *capability.Registry and
// *loop.Loop; Sessions/Personas/Memories are satisfied by the concrete session
// manager, persona registry and memory store, and the handlers return [] only
// when a dependency is left nil.
type Deps struct {
	// Log is the base logger. Nil uses a discard logger.
	Log *slog.Logger
	// Auth is the JWT/RBAC service. Required.
	Auth Authenticator
	// Bus is the in-process event bus the WS hub subscribes to. Optional.
	Bus bus.Bus
	// Registry enumerates capabilities/plugins/channels. Optional.
	Registry CapabilityStore
	// Loop runs chat turns. Optional (chat returns an error when unset).
	Loop Chatter
	// Session lists channel sessions. Optional.
	Session SessionStore
	// Personas lists personas. Optional.
	Personas PersonaStore
	// Memories lists memory entries. Optional.
	Memories MemoryStore
	// IoT lists IoT devices. Optional.
	IoT IoTStore
	// Scheduler audits sub-agent context for the subagent_ask WS route.
	// Optional (subagent_ask returns a clear error when unset).
	Scheduler SubagentAuditor
	// Tasks queries task output for the task_output_query WS route.
	// Optional (task_output_query returns a clear error when unset).
	Tasks TaskOutputStore
	// Automations lists automations and reminders. Optional.
	Automations AutomationStore
	// Users lists accounts and API keys. Optional (falls back to the configured
	// single admin account).
	Users UserStore
	// Backthink lists background-think records. Optional.
	Backthink BackthinkStore
	// Streaming marks the injected Loop as streaming-capable (its OnStream is
	// wired to publish ws.stream_chunk). When set the hub skips its own single
	// final stream_chunk emission, since deltas already stream through OnStream.
	Streaming bool
	// Frontend is the SPA filesystem served for all non-API, non-WS routes.
	// When nil the gateway serves no static assets (API and WS only). The
	// embedded dist (internal/webui) or an external dist directory is used.
	Frontend fs.FS
}

// Server is the gateway. It owns a ServeMux aggregating REST + WS + admin and a
// WebSocket hub. It is safe for concurrent use once constructed.
type Server struct {
	cfg         *config.Config
	auth        Authenticator
	bus         bus.Bus
	caps        CapabilityStore
	runner      Chatter
	sessions    SessionStore
	personas    PersonaStore
	memories    MemoryStore
	iot         IoTStore
	auditor     SubagentAuditor
	taskOutputs TaskOutputStore
	automations AutomationStore
	users       UserStore
	backthink   BackthinkStore
	voice       VoiceService
	streaming   bool
	frontend    fs.FS
	log         *slog.Logger
	hub         *hub
}

// New validates cfg and deps and returns a ready Server. Auth is required; the
// WebSocket hub subscribes to deps.Bus immediately so events published before
// ListenAndServe are still relayed to clients.
func New(cfg *config.Config, deps Deps) (*Server, error) {
	if cfg == nil {
		return nil, fmt.Errorf("gateway: nil config")
	}
	if deps.Auth == nil {
		return nil, fmt.Errorf("gateway: auth dependency required")
	}
	s := &Server{
		cfg:         cfg,
		auth:        deps.Auth,
		bus:         deps.Bus,
		caps:        deps.Registry,
		runner:      deps.Loop,
		sessions:    deps.Session,
		personas:    deps.Personas,
		memories:    deps.Memories,
		iot:         deps.IoT,
		auditor:     deps.Scheduler,
		taskOutputs: deps.Tasks,
		automations: deps.Automations,
		users:       deps.Users,
		backthink:   deps.Backthink,
		streaming:   deps.Streaming,
		frontend:    deps.Frontend,
		log:         sourcedLog(deps.Log),
	}
	s.hub = newHub(s)
	s.hub.start()
	return s, nil
}

// Handler returns the fully assembled http.Handler: the ServeMux routing
// REST + WS + admin, wrapped in recovery, CORS and request logging.
func (s *Server) Handler() http.Handler {
	mux := http.NewServeMux()

	// Public, no auth.
	mux.HandleFunc("GET /api/v1/health", s.handleHealth)
	mux.HandleFunc("POST /api/v1/auth/login", s.handleLogin)

	// User side, authenticated. Chat is a write (owner/admin); reads accept any
	// authenticated role so a viewer can still list personas/sessions.
	mux.Handle("POST /api/v1/chat", s.auth.Authorize(auth.RoleOwner, auth.RoleAdmin)(http.HandlerFunc(s.handleChat)))
	readAuth := s.auth.Authorize()
	mux.Handle("GET /api/v1/sessions", readAuth(http.HandlerFunc(s.handleListSessions)))
	mux.Handle("GET /api/v1/personas", readAuth(http.HandlerFunc(s.handleListPersonas)))

	// Realtime.
	mux.HandleFunc("/ws/chat", s.handleWS)

	// Admin: reads require owner or admin; the audited policy write requires owner.
	adminAuth := s.auth.Authorize(auth.RoleOwner, auth.RoleAdmin)
	mux.Handle("GET /api/v1/admin/personas", adminAuth(http.HandlerFunc(s.adminListPersonas)))
	mux.Handle("GET /api/v1/admin/sessions", adminAuth(http.HandlerFunc(s.adminListSessions)))
	mux.Handle("GET /api/v1/admin/memories", adminAuth(http.HandlerFunc(s.adminListMemories)))
	mux.Handle("GET /api/v1/admin/plugins", adminAuth(http.HandlerFunc(s.adminListPlugins)))
	mux.Handle("GET /api/v1/admin/channels", adminAuth(http.HandlerFunc(s.adminListChannels)))
	mux.Handle("GET /api/v1/admin/iot", adminAuth(http.HandlerFunc(s.adminListIoT)))
	mux.Handle("GET /api/v1/admin/llm", adminAuth(http.HandlerFunc(s.adminGetLlm)))
	mux.Handle("GET /api/v1/admin/automations", adminAuth(http.HandlerFunc(s.adminListAutomations)))
	mux.Handle("GET /api/v1/admin/users", adminAuth(http.HandlerFunc(s.adminListUsers)))
	mux.Handle("GET /api/v1/admin/backthink", adminAuth(http.HandlerFunc(s.adminListBackthink)))
	mux.Handle("POST /api/v1/admin/plugins/policy", s.auth.Authorize(auth.RoleOwner)(http.HandlerFunc(s.adminSetPolicy)))

	var h http.Handler = mux
	if s.frontend != nil {
		h = s.spa(h)
	}
	h = s.recover(h)
	h = s.cors(h)
	h = s.logging(h)
	return h
}

// Start launches the gateway in a goroutine, returning a channel that carries the
// ListenAndServe result. The server runs until ctx is canceled.
func (s *Server) Start(ctx context.Context) (<-chan error, error) {
	errCh := make(chan error, 1)
	go func() { errCh <- s.ListenAndServe(ctx) }()
	return errCh, nil
}

// ListenAndServe binds the configured host:port (default :8080) and blocks until
// ctx is canceled, then gracefully drains in-flight requests.
func (s *Server) ListenAndServe(ctx context.Context) error {
	if ctx == nil {
		ctx = context.Background()
	}
	addr := net.JoinHostPort(s.cfg.Server.Host, strconv.Itoa(s.cfg.Server.Port))
	httpSrv := &http.Server{Addr: addr, Handler: s.Handler()}

	done := make(chan struct{})
	go func() {
		<-ctx.Done()
		s.log.Info("gateway shutting down", "addr", addr)
		s.hub.close()
		shCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
		defer cancel()
		_ = httpSrv.Shutdown(shCtx)
		close(done)
	}()

	s.log.Info("gateway listening", "addr", addr)
	err := httpSrv.ListenAndServe()
	if err != nil && err != http.ErrServerClosed {
		return fmt.Errorf("gateway: listen: %w", err)
	}
	<-done
	return nil
}

// Close frees the hub's bus subscription and closes live connections.
func (s *Server) Close() { s.hub.close() }

// sourcedLog returns a logger tagged with source "gateway", substituting a
// discard logger for nil so Server can always log safely.
func sourcedLog(l *slog.Logger) *slog.Logger {
	if l == nil {
		l = slog.New(slog.NewTextHandler(io.Discard, nil))
	}
	return logger.WithSource(l, source)
}