diff --git a/cmd/gateway.go b/cmd/gateway.go index 783dec93..a1d13ae9 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -2,6 +2,7 @@ package cmd import ( "context" + "database/sql" "fmt" "log/slog" "os" @@ -617,6 +618,10 @@ func runGateway() { // Create gateway server and wire enforcement server := gateway.NewServer(cfg, msgBus, agentRouter, sessStore, toolsReg) + server.SetVersion(Version) + if managedStores != nil { + server.SetDB(managedStores.DB) + } server.SetPolicyEngine(permPE) server.SetPairingService(pairingStore) @@ -940,7 +945,12 @@ func runGateway() { } // Register quota usage RPC (nil-safe — returns {enabled: false} in standalone mode). - methods.NewQuotaMethods(quotaChecker).Register(server.Router()) + // Pass DB so summary cards still work when quota is disabled (queries traces directly). + var quotaDB *sql.DB + if managedStores != nil { + quotaDB = managedStores.DB + } + methods.NewQuotaMethods(quotaChecker, quotaDB).Register(server.Router()) // Reload quota config on config changes via pub/sub. if quotaChecker != nil { diff --git a/internal/channels/quota.go b/internal/channels/quota.go index 1141c9b2..e07ca209 100644 --- a/internal/channels/quota.go +++ b/internal/channels/quota.go @@ -212,7 +212,7 @@ func (qc *QuotaChecker) Usage(ctx context.Context) QuotaUsageResult { } if !cfg.Enabled { // Still return today's summary even when quota is disabled - qc.queryTodaySummary(ctx, &result) + QueryTodaySummary(ctx, qc.db, &result) return result } @@ -237,7 +237,7 @@ func (qc *QuotaChecker) Usage(ctx context.Context) QuotaUsageResult { ) if err != nil { slog.Warn("quota.usage: failed to query user counts", "error", err) - qc.queryTodaySummary(ctx, &result) + QueryTodaySummary(ctx, qc.db, &result) return result } defer rows.Close() @@ -258,16 +258,17 @@ func (qc *QuotaChecker) Usage(ctx context.Context) QuotaUsageResult { }) } - qc.queryTodaySummary(ctx, &result) + QueryTodaySummary(ctx, qc.db, &result) return result } -// queryTodaySummary fills today's aggregate stats into the result. -func (qc *QuotaChecker) queryTodaySummary(ctx context.Context, result *QuotaUsageResult) { +// QueryTodaySummary fills today's aggregate stats into the result. +// Exported so QuotaMethods can call it directly when QuotaChecker is nil. +func QueryTodaySummary(ctx context.Context, db *sql.DB, result *QuotaUsageResult) { now := time.Now().UTC() startOfDay := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC) - err := qc.db.QueryRowContext(ctx, ` + err := db.QueryRowContext(ctx, ` SELECT COUNT(*), COALESCE(SUM(total_input_tokens), 0), diff --git a/internal/gateway/client.go b/internal/gateway/client.go index 3bf9cf7b..45bfcc6d 100644 --- a/internal/gateway/client.go +++ b/internal/gateway/client.go @@ -25,17 +25,22 @@ type Client struct { send chan []byte mu sync.Mutex + connectedAt time.Time // when the client connected + remoteAddr string // peer IP:port from websocket + // Browser pairing state pairingCode string // 8-char code if pending approval pairingPending bool // true while waiting for admin approval } -func NewClient(conn *websocket.Conn, server *Server) *Client { +func NewClient(conn *websocket.Conn, server *Server, remoteIP string) *Client { return &Client{ - id: uuid.NewString(), - conn: conn, - server: server, - send: make(chan []byte, 256), + id: uuid.NewString(), + conn: conn, + server: server, + send: make(chan []byte, 256), + connectedAt: time.Now(), + remoteAddr: remoteIP, } } @@ -178,6 +183,12 @@ func (c *Client) Role() permissions.Role { return c.role } // UserID returns the external user ID set during connect. func (c *Client) UserID() string { return c.userID } +// ConnectedAt returns when the client connected. +func (c *Client) ConnectedAt() time.Time { return c.connectedAt } + +// RemoteAddr returns the peer IP:port. +func (c *Client) RemoteAddr() string { return c.remoteAddr } + // Close shuts down the client connection. func (c *Client) Close() { close(c.send) diff --git a/internal/gateway/methods/quota_methods.go b/internal/gateway/methods/quota_methods.go index e9abff05..81c42b25 100644 --- a/internal/gateway/methods/quota_methods.go +++ b/internal/gateway/methods/quota_methods.go @@ -2,6 +2,7 @@ package methods import ( "context" + "database/sql" "github.com/nextlevelbuilder/goclaw/internal/channels" "github.com/nextlevelbuilder/goclaw/internal/gateway" @@ -10,12 +11,14 @@ import ( // QuotaMethods handles quota.usage — returns per-user quota consumption for the dashboard. // Nil-safe: returns {enabled: false} when quotaChecker is nil (standalone mode). +// When checker is nil but db is available, still queries today's summary from traces. type QuotaMethods struct { checker *channels.QuotaChecker + db *sql.DB } -func NewQuotaMethods(checker *channels.QuotaChecker) *QuotaMethods { - return &QuotaMethods{checker: checker} +func NewQuotaMethods(checker *channels.QuotaChecker, db *sql.DB) *QuotaMethods { + return &QuotaMethods{checker: checker, db: db} } func (m *QuotaMethods) Register(router *gateway.MethodRouter) { @@ -24,10 +27,15 @@ func (m *QuotaMethods) Register(router *gateway.MethodRouter) { func (m *QuotaMethods) handleUsage(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { if m.checker == nil { - client.SendResponse(protocol.NewOKResponse(req.ID, channels.QuotaUsageResult{ + result := channels.QuotaUsageResult{ Enabled: false, Entries: []channels.QuotaUsageEntry{}, - })) + } + // Still query today's summary from traces when DB is available + if m.db != nil { + channels.QueryTodaySummary(ctx, m.db, &result) + } + client.SendResponse(protocol.NewOKResponse(req.ID, result)) return } diff --git a/internal/gateway/router.go b/internal/gateway/router.go index 4e05458c..8b01da6d 100644 --- a/internal/gateway/router.go +++ b/internal/gateway/router.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "log/slog" + "time" "github.com/nextlevelbuilder/goclaw/internal/permissions" "github.com/nextlevelbuilder/goclaw/pkg/protocol" @@ -162,16 +163,86 @@ func (r *MethodRouter) sendConnectResponse(client *Client, reqID string) { } func (r *MethodRouter) handleHealth(ctx context.Context, client *Client, req *protocol.RequestFrame) { + s := r.server + uptimeMs := time.Since(s.startedAt).Milliseconds() + + mode := "standalone" + if s.cfg.Database.Mode == "managed" { + mode = "managed" + } + + // Database status (real ping) + dbStatus := "n/a" + if s.db != nil { + pingCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + defer cancel() + if err := s.db.PingContext(pingCtx); err != nil { + dbStatus = "error" + } else { + dbStatus = "ok" + } + } + + // Connected clients list + type clientInfo struct { + ID string `json:"id"` + RemoteAddr string `json:"remoteAddr"` + UserID string `json:"userId"` + Role string `json:"role"` + ConnectedAt string `json:"connectedAt"` + } + clients := s.ClientList() + clientList := make([]clientInfo, 0, len(clients)) + for _, c := range clients { + clientList = append(clientList, clientInfo{ + ID: c.ID(), + RemoteAddr: c.RemoteAddr(), + UserID: c.UserID(), + Role: string(c.Role()), + ConnectedAt: c.ConnectedAt().UTC().Format(time.RFC3339), + }) + } + + // Tool count + toolCount := 0 + if s.tools != nil { + toolCount = s.tools.Count() + } + client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{ - "status": "ok", + "status": "ok", + "version": s.version, + "uptime": uptimeMs, + "mode": mode, + "database": dbStatus, + "tools": toolCount, + "clients": clientList, + "currentId": client.ID(), })) } func (r *MethodRouter) handleStatus(ctx context.Context, client *Client, req *protocol.RequestFrame) { agents := r.server.agents.ListInfo() + + sessionCount := 0 + if r.server.sessions != nil { + sessionCount = len(r.server.sessions.List("")) + } + + // In managed mode, agents are lazily resolved — router only has loaded agents. + // Query the DB store for the real total count. + agentTotal := len(agents) + if r.server.agentStore != nil { + if dbAgents, err := r.server.agentStore.List(ctx, ""); err == nil && len(dbAgents) > agentTotal { + agentTotal = len(dbAgents) + } + } + client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{ - "agents": agents, - "clients": len(r.server.clients), + "agents": agents, + "agentTotal": agentTotal, + "clients": len(r.server.clients), + "sessions": sessionCount, })) } diff --git a/internal/gateway/server.go b/internal/gateway/server.go index b3a5fe6b..97234857 100644 --- a/internal/gateway/server.go +++ b/internal/gateway/server.go @@ -49,6 +49,10 @@ type Server struct { clients map[string]*Client mu sync.RWMutex + startedAt time.Time + version string + db interface{ PingContext(context.Context) error } // for health check DB ping + httpServer *http.Server mux *http.ServeMux } @@ -56,11 +60,12 @@ type Server struct { // NewServer creates a new gateway server. func NewServer(cfg *config.Config, eventPub bus.EventPublisher, agents *agent.Router, sess store.SessionStore, toolsReg ...*tools.Registry) *Server { s := &Server{ - cfg: cfg, - eventPub: eventPub, - agents: agents, - sessions: sess, - clients: make(map[string]*Client), + cfg: cfg, + eventPub: eventPub, + agents: agents, + sessions: sess, + clients: make(map[string]*Client), + startedAt: time.Now(), } s.upgrader = websocket.Upgrader{ @@ -222,7 +227,7 @@ func (s *Server) handleWebSocket(w http.ResponseWriter, r *http.Request) { return } - client := NewClient(conn, s) + client := NewClient(conn, s, clientIP(r)) s.registerClient(client) defer func() { @@ -240,6 +245,21 @@ func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { fmt.Fprintf(w, `{"status":"ok","protocol":%d}`, protocol.ProtocolVersion) } +// clientIP extracts the real client IP from the request, checking proxy headers first. +func clientIP(r *http.Request) string { + if ip := r.Header.Get("X-Real-IP"); ip != "" { + return ip + } + if fwd := r.Header.Get("X-Forwarded-For"); fwd != "" { + if i := strings.IndexByte(fwd, ','); i > 0 { + return strings.TrimSpace(fwd[:i]) + } + return strings.TrimSpace(fwd) + } + host, _, _ := net.SplitHostPort(r.RemoteAddr) + return host +} + // Router returns the method router for registering additional handlers. func (s *Server) Router() *MethodRouter { return s.router } @@ -283,6 +303,29 @@ func (s *Server) SetBuiltinToolsHandler(h *httpapi.BuiltinToolsHandler) { // SetAgentStore sets the agent store for context injection in tools_invoke. func (s *Server) SetAgentStore(as store.AgentStore) { s.agentStore = as } +// SetVersion sets the server version for health responses. +func (s *Server) SetVersion(v string) { s.version = v } + +// SetDB sets the database connection for health check pings. +func (s *Server) SetDB(db interface{ PingContext(context.Context) error }) { s.db = db } + +// StartedAt returns the server start time. +func (s *Server) StartedAt() time.Time { return s.startedAt } + +// Version returns the server version string. +func (s *Server) Version() string { return s.version } + +// ClientList returns a snapshot of all connected clients. +func (s *Server) ClientList() []*Client { + s.mu.RLock() + defer s.mu.RUnlock() + list := make([]*Client, 0, len(s.clients)) + for _, c := range s.clients { + list = append(list, c) + } + return list +} + // BroadcastEvent sends an event to all connected clients. func (s *Server) BroadcastEvent(event protocol.EventFrame) { s.mu.RLock() diff --git a/ui/web/src/pages/overview/overview-page.tsx b/ui/web/src/pages/overview/overview-page.tsx index 02db774f..c8fa41a6 100644 --- a/ui/web/src/pages/overview/overview-page.tsx +++ b/ui/web/src/pages/overview/overview-page.tsx @@ -1,4 +1,4 @@ -import { useEffect, useCallback } from "react"; +import { useEffect, useCallback, useState, useRef } from "react"; import { Activity, Bot, @@ -8,10 +8,18 @@ import { ArrowRight, Clock, Timer, + Monitor, + Database, + Wrench, + CheckCircle2, + XCircle, + Minus, + Users, } from "lucide-react"; import { Link } from "react-router"; import { PageHeader } from "@/components/shared/page-header"; import { StatusBadge } from "@/components/shared/status-badge"; +import { Badge } from "@/components/ui/badge"; import { Alert, AlertTitle, AlertDescription } from "@/components/ui/alert"; import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card"; import { useAuthStore } from "@/stores/use-auth-store"; @@ -29,9 +37,23 @@ import { // --- Types --- +interface ClientInfo { + id: string; + remoteAddr: string; + userId: string; + role: string; + connectedAt: string; +} + interface HealthPayload { status?: string; + version?: string; uptime?: number; + mode?: string; + database?: string; + tools?: number; + clients?: ClientInfo[]; + currentId?: string; } interface AgentInfo { @@ -42,6 +64,7 @@ interface AgentInfo { interface StatusPayload { agents?: AgentInfo[]; + agentTotal?: number; sessions?: number; clients?: number; } @@ -153,70 +176,153 @@ function QuotaCell({ usage }: { usage: QuotaUsage }) { ); } +function StatusDot({ ok }: { ok: boolean | undefined }) { + if (ok === undefined) + return ; + return ok ? ( + + ) : ( + + ); +} + +function useLiveUptime(serverUptimeMs: number | undefined) { + const [tick, setTick] = useState(0); + const baseRef = useRef<{ serverMs: number; localTs: number } | null>(null); + + useEffect(() => { + if (serverUptimeMs != null) { + baseRef.current = { serverMs: serverUptimeMs, localTs: Date.now() }; + } + }, [serverUptimeMs]); + + useEffect(() => { + const id = setInterval(() => setTick((t) => t + 1), 1000); + return () => clearInterval(id); + }, []); + + if (!baseRef.current) return undefined; + void tick; // used for re-render + return baseRef.current.serverMs + (Date.now() - baseRef.current.localTs); +} + function formatUptime(ms: number | undefined): string { - if (!ms) return "—"; + if (!ms) return "--"; const sec = Math.floor(ms / 1000); + const s = sec % 60; const min = Math.floor(sec / 60) % 60; const hr = Math.floor(sec / 3600) % 24; const day = Math.floor(sec / 86400); - if (day > 0) return `${day}d ${hr}h ${min}m`; - if (hr > 0) return `${hr}h ${min}m`; - return `${min}m`; + if (day > 0) return `${day}d ${hr}h ${min}m ${s}s`; + if (hr > 0) return `${hr}h ${min}m ${s}s`; + if (min > 0) return `${min}m ${s}s`; + return `${s}s`; +} + +function formatClientTime(iso: string): string { + try { + const d = new Date(iso); + const now = Date.now(); + const diffMs = now - d.getTime(); + if (diffMs < 0) return "just now"; + const sec = Math.floor(diffMs / 1000); + if (sec < 60) return `${sec}s ago`; + const min = Math.floor(sec / 60); + if (min < 60) return `${min}m ago`; + const hr = Math.floor(min / 60); + return `${hr}h ${min % 60}m ago`; + } catch { + return "--"; + } } // --- Page --- +const REFRESH_INTERVAL = 30_000; + export function OverviewPage() { const connected = useAuthStore((s) => s.connected); - const { call: fetchHealth, data: health } = useWsCall(Methods.HEALTH); - const { call: fetchStatus, data: status } = useWsCall(Methods.STATUS); - const { call: fetchQuota, data: quota } = useWsCall(Methods.QUOTA_USAGE); - const { call: fetchCron, data: cronData } = useWsCall(Methods.CRON_LIST); - const { call: fetchChannels, data: channelStatusData } = useWsCall(Methods.CHANNELS_STATUS); + const { call: fetchHealth, data: health } = + useWsCall(Methods.HEALTH); + const { call: fetchStatus, data: status } = + useWsCall(Methods.STATUS); + const { call: fetchQuota, data: quota } = + useWsCall(Methods.QUOTA_USAGE); + const { call: fetchCron, data: cronData } = + useWsCall(Methods.CRON_LIST); + const { call: fetchChannels, data: channelStatusData } = + useWsCall(Methods.CHANNELS_STATUS); const { providers, loading: providersLoading } = useProviders(); const { traces } = useTraces({ limit: 8 }); const hasNoProviders = !providersLoading && providers.length === 0; const hasNoEnabledProviders = - !providersLoading && providers.length > 0 && !providers.some((p) => p.enabled); + !providersLoading && + providers.length > 0 && + !providers.some((p) => p.enabled); + const fetchAll = useCallback(() => { + fetchHealth(); + fetchStatus(); + fetchQuota(); + fetchCron({ includeDisabled: true }); + fetchChannels(); + }, [fetchHealth, fetchStatus, fetchQuota, fetchCron, fetchChannels]); + + // Initial fetch + auto-refresh every 30s useEffect(() => { - if (connected) { - fetchHealth(); - fetchStatus(); - fetchQuota(); - fetchCron({ includeDisabled: true }); - fetchChannels(); - } - }, [connected, fetchHealth, fetchStatus, fetchQuota, fetchCron, fetchChannels]); + if (!connected) return; + fetchAll(); + const id = setInterval(fetchAll, REFRESH_INTERVAL); + return () => clearInterval(id); + }, [connected, fetchAll]); - const handleHealth = useCallback(() => { + // Health event listener + const handleHealthEvent = useCallback(() => { fetchHealth(); fetchStatus(); }, [fetchHealth, fetchStatus]); + useWsEvent(Events.HEALTH, handleHealthEvent); - useWsEvent(Events.HEALTH, handleHealth); + // Live uptime tick + const liveUptime = useLiveUptime(health?.uptime); - // Compute stats + // Computed const agents = status?.agents ?? []; const runningAgents = agents.filter((a) => a.isRunning).length; + const agentTotal = status?.agentTotal ?? agents.length; const channelEntries = channelStatusData?.channels - ? Object.values(channelStatusData.channels) + ? Object.entries(channelStatusData.channels) : []; - const channelsOnline = channelEntries.filter((c) => c.running).length; + const channelsOnline = channelEntries.filter(([, c]) => c.running).length; const cronJobs = cronData?.jobs ?? []; - const enabledProviders = providers.filter((p) => p.enabled).length; + const enabledProviders = providers.filter((p) => p.enabled); + const clientList = health?.clients ?? []; + const currentId = health?.currentId; return (
+ {/* Header */} +
+ {health?.mode && ( + + {health.mode} + + )} + {health?.version && ( + + v{health.version} + + )} + +
} /> @@ -249,7 +355,11 @@ export function OverviewPage() { icon={Activity} label="Requests Today" value={quota?.requestsToday ?? 0} - sub={quota?.uniqueUsersToday ? `${quota.uniqueUsersToday} users` : undefined} + sub={ + quota?.uniqueUsersToday + ? `${quota.uniqueUsersToday} users` + : undefined + } /> 0 ? `${runningAgents} / ${agents.length}` : "0"} - sub={agents.length > 0 ? "running" : undefined} + value={ + agentTotal > 0 + ? `${runningAgents} / ${agentTotal}` + : "0" + } + sub={agentTotal > 0 ? "running" : undefined} /> {quota.entries.map((entry) => ( - + - {entry.userId} + + {entry.userId} + @@ -323,6 +442,226 @@ export function OverviewPage() { )} + {/* Middle row: System + Cron + Connected Clients */} +
+ {/* System Health */} + + + + System + + + +
+
+
+ Uptime +
+
+ {formatUptime(liveUptime)} +
+
+ {health?.mode === "managed" && ( +
+
+ Database +
+
+ + {health.database === "ok" ? "Connected" : health.database} +
+
+ )} +
+
Providers
+
+ {enabledProviders.length > 0 ? ( + + + {enabledProviders.length} active + + ) : ( + + + none + + )} +
+
+ {channelEntries.length > 0 && ( +
+
Channels
+
+ {channelEntries.map(([name, ch]) => ( + + + {name} + + ))} +
+
+ )} + {(health?.tools ?? 0) > 0 && ( +
+
+ Tools +
+
{health!.tools} registered
+
+ )} +
+
Sessions
+
{status?.sessions ?? 0}
+
+
+ + + + {/* Cron Jobs */} + + + Cron Jobs + {cronJobs.length > 0 && ( + + Manage + + )} + + + {cronJobs.length === 0 ? ( +

+ No cron jobs configured +

+ ) : ( +
+ {cronJobs.slice(0, 5).map((job) => ( +
+
+ + + {job.name} + +
+ + {job.enabled && job.state.nextRunAtMs ? ( + <> + + {formatRelativeTime( + new Date(job.state.nextRunAtMs), + ).replace(" ago", "")} + + ) : !job.enabled ? ( + "disabled" + ) : ( + "--" + )} + +
+ ))} +
+ )} + + + + {/* Connected Clients */} + + + + Connected Clients + {clientList.length > 0 && ( + + {clientList.length} + + )} + + + + {clientList.length === 0 ? ( +

+ No clients connected +

+ ) : ( +
+ + + + + + + + + + + {clientList.map((c) => { + const isYou = c.id === currentId; + return ( + + + + + + + ); + })} + +
IPUserRoleConnected
+ {c.remoteAddr.replace(/:\d+$/, "")} + {isYou && ( + + you + + )} + + {c.userId || "--"} + + + {c.role} + + + {formatClientTime(c.connectedAt)} +
+
+ )} + + +
+ {/* Recent Requests */} {traces.length > 0 && ( @@ -344,8 +683,12 @@ export function OverviewPage() { Name User Channel - Tokens - Duration + + Tokens + + + Duration + Status @@ -356,14 +699,16 @@ export function OverviewPage() { {formatRelativeTime(t.created_at)} - {t.name || "—"} + {t.name || "--"} - {t.user_id || "—"} + {t.user_id || "--"} - {t.channel || "—"} + {t.channel || "--"} - {formatTokens(t.total_input_tokens + t.total_output_tokens)} + {formatTokens( + t.total_input_tokens + t.total_output_tokens, + )} {formatDuration(t.duration_ms)} @@ -390,94 +735,6 @@ export function OverviewPage() { )} - - {/* Bottom row: System + Cron */} -
- {/* System */} - - - System - - -
-
-
- Uptime -
-
{formatUptime(health?.uptime)}
-
-
-
Sessions
-
{status?.sessions ?? 0}
-
-
-
Connected Clients
-
{status?.clients ?? 0}
-
-
-
Providers
-
- {enabledProviders > 0 - ? `${enabledProviders} active` - : "none"} -
-
-
- - - - {/* Cron Jobs */} - - - Cron Jobs - {cronJobs.length > 0 && ( - - Manage - - )} - - - {cronJobs.length === 0 ? ( -

No cron jobs configured

- ) : ( -
- {cronJobs.slice(0, 5).map((job) => ( -
-
- - - {job.name} - -
- - {job.enabled && job.state.nextRunAtMs ? ( - <> - - {formatRelativeTime(new Date(job.state.nextRunAtMs)).replace(" ago", "")} - - ) : !job.enabled ? ( - "disabled" - ) : ( - "—" - )} - -
- ))} -
- )} - - -
); }