diff --git a/.gitignore b/.gitignore index b8d904f..591b8c9 100644 --- a/.gitignore +++ b/.gitignore @@ -16,7 +16,8 @@ data/ web/node_modules/ web/build/ web/.svelte-kit/ -internal/web/dist/ +internal/web/dist/* +!internal/web/dist/index.html # IDE .idea/ diff --git a/Makefile b/Makefile index 422ef70..b62ffe7 100644 --- a/Makefile +++ b/Makefile @@ -18,7 +18,8 @@ dev: web: cd web && npm install && npm run build - @echo "Svelte SPA built to internal/web/dist/" + rm -rf internal/web/dist + cp -r web/build internal/web/dist clean: rm -rf $(BUILD_DIR) diff --git a/cmd/synapbus/main.go b/cmd/synapbus/main.go index 6f5f9b2..496e2cb 100644 --- a/cmd/synapbus/main.go +++ b/cmd/synapbus/main.go @@ -29,6 +29,7 @@ import ( "github.com/smart-mcp-proxy/synapbus/internal/search/embedding" "github.com/smart-mcp-proxy/synapbus/internal/storage" "github.com/smart-mcp-proxy/synapbus/internal/trace" + "github.com/smart-mcp-proxy/synapbus/internal/web" ) var ( @@ -339,10 +340,26 @@ func runServe(cmd *cobra.Command, args []string) error { // MCP SSE endpoint r.Mount("/mcp", mcpSrv.SSEHandler()) - // Mount API routes (traces, export, stats, metrics, attachments) - apiRouter := api.NewRouter(traceStore, metrics, attachmentService) + // Create SSE hub for real-time events + sseHub := api.NewSSEHub() + + // Mount API routes (traces, export, stats, metrics, attachments, messages, agents, channels, SSE) + sessionMiddleware := api.SessionToOwnerMiddleware(userStore, sessionStore) + apiRouter := api.NewRouterWithConfig(api.RouterConfig{ + TraceStore: traceStore, + Metrics: metrics, + AttachmentService: attachmentService, + MsgService: msgService, + AgentService: agentService, + ChannelService: channelService, + SSEHub: sseHub, + SessionMiddleware: sessionMiddleware, + }) r.Mount("/", apiRouter) + // Serve embedded Web UI SPA (catch-all for non-API routes) + r.NotFound(web.NewSPAHandler().ServeHTTP) + // Start HTTP server addr := fmt.Sprintf(":%d", port) srv := &http.Server{ diff --git a/internal/agents/service.go b/internal/agents/service.go index 0a8679d..3b9332f 100644 --- a/internal/agents/service.go +++ b/internal/agents/service.go @@ -209,6 +209,52 @@ func (s *AgentService) ListAgents(ctx context.Context, ownerID int64) ([]*Agent, return s.store.ListAgentsByOwner(ctx, ownerID) } +// RevokeKey generates a new API key for an agent. Only the owner can do this. +// Returns the agent and the new raw API key (shown once). +func (s *AgentService) RevokeKey(ctx context.Context, name string, ownerID int64) (*Agent, string, error) { + agent, err := s.store.GetAgentByName(ctx, name) + if err != nil { + if err == sql.ErrNoRows { + return nil, "", fmt.Errorf("agent not found: %s", name) + } + return nil, "", err + } + + if agent.OwnerID != ownerID { + return nil, "", fmt.Errorf("only the agent's owner can revoke its API key") + } + + // Generate new API key + apiKey, err := generateAPIKey() + if err != nil { + return nil, "", fmt.Errorf("generate API key: %w", err) + } + + hash, err := bcrypt.GenerateFromPassword([]byte(apiKey), bcrypt.DefaultCost) + if err != nil { + return nil, "", fmt.Errorf("hash API key: %w", err) + } + + agent.APIKeyHash = string(hash) + if err := s.store.UpdateAgent(ctx, agent); err != nil { + return nil, "", fmt.Errorf("update agent: %w", err) + } + + s.logger.Info("agent API key revoked", + "name", name, + "owner_id", ownerID, + ) + + if s.tracer != nil { + s.tracer.Record(ctx, name, "revoke_api_key", map[string]any{ + "agent_id": agent.ID, + "owner_id": ownerID, + }) + } + + return agent, apiKey, nil +} + // generateAPIKey creates a cryptographically random API key (32 bytes, hex encoded). func generateAPIKey() (string, error) { b := make([]byte, 32) diff --git a/internal/api/agents_handler.go b/internal/api/agents_handler.go new file mode 100644 index 0000000..cc7fe4b --- /dev/null +++ b/internal/api/agents_handler.go @@ -0,0 +1,166 @@ +package api + +import ( + "encoding/json" + "log/slog" + "net/http" + + "github.com/go-chi/chi/v5" + + "github.com/smart-mcp-proxy/synapbus/internal/agents" + "github.com/smart-mcp-proxy/synapbus/internal/trace" +) + +// AgentsHandler handles REST API requests for agents. +type AgentsHandler struct { + agentService *agents.AgentService + traceStore trace.TraceStore + logger *slog.Logger +} + +// NewAgentsHandler creates a new agents handler. +func NewAgentsHandler(agentService *agents.AgentService, traceStore trace.TraceStore) *AgentsHandler { + return &AgentsHandler{ + agentService: agentService, + traceStore: traceStore, + logger: slog.Default().With("component", "api.agents"), + } +} + +// ListAgents handles GET /api/agents. +func (h *AgentsHandler) ListAgents(w http.ResponseWriter, r *http.Request) { + ownerID, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + agentList, err := h.agentService.ListAgents(r.Context(), ownerID) + if err != nil { + h.logger.Error("list agents failed", "error", err) + writeJSON(w, http.StatusInternalServerError, errorBody("server_error", "Failed to list agents")) + return + } + + writeJSON(w, http.StatusOK, map[string]any{"agents": agentList}) +} + +// GetAgent handles GET /api/agents/{name}. +func (h *AgentsHandler) GetAgent(w http.ResponseWriter, r *http.Request) { + ownerID, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + name := chi.URLParam(r, "name") + agent, err := h.agentService.GetAgent(r.Context(), name) + if err != nil { + writeJSON(w, http.StatusNotFound, errorBody("not_found", "Agent not found")) + return + } + + if agent.OwnerID != ownerID { + writeJSON(w, http.StatusForbidden, errorBody("forbidden", "You do not have access to this agent")) + return + } + + // Get recent traces for this agent + var traces []trace.Trace + if h.traceStore != nil { + filter := trace.TraceFilter{ + AgentName: name, + PageSize: 20, + Page: 1, + } + traces, _, _ = h.traceStore.Query(r.Context(), filter) + } + if traces == nil { + traces = []trace.Trace{} + } + + writeJSON(w, http.StatusOK, map[string]any{ + "agent": agent, + "traces": traces, + }) +} + +// RegisterAgent handles POST /api/agents. +func (h *AgentsHandler) RegisterAgent(w http.ResponseWriter, r *http.Request) { + ownerID, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + var req struct { + Name string `json:"name"` + DisplayName string `json:"display_name"` + Type string `json:"type"` + Capabilities json.RawMessage `json:"capabilities,omitempty"` + } + + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_request", "Invalid JSON body")) + return + } + + if req.Name == "" { + writeJSON(w, http.StatusBadRequest, errorBody("validation_error", "Agent name is required")) + return + } + + agent, apiKey, err := h.agentService.Register(r.Context(), req.Name, req.DisplayName, req.Type, req.Capabilities, ownerID) + if err != nil { + h.logger.Error("register agent failed", "error", err) + writeJSON(w, http.StatusBadRequest, errorBody("register_failed", err.Error())) + return + } + + writeJSON(w, http.StatusCreated, map[string]any{ + "agent": agent, + "api_key": apiKey, + }) +} + +// DeleteAgent handles DELETE /api/agents/{name}. +func (h *AgentsHandler) DeleteAgent(w http.ResponseWriter, r *http.Request) { + ownerID, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + name := chi.URLParam(r, "name") + + if err := h.agentService.Deregister(r.Context(), name, ownerID); err != nil { + h.logger.Error("deregister agent failed", "error", err) + writeJSON(w, http.StatusBadRequest, errorBody("deregister_failed", err.Error())) + return + } + + writeJSON(w, http.StatusOK, map[string]string{"status": "deregistered"}) +} + +// RevokeKey handles POST /api/agents/{name}/revoke-key. +func (h *AgentsHandler) RevokeKey(w http.ResponseWriter, r *http.Request) { + ownerID, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + name := chi.URLParam(r, "name") + + agent, newKey, err := h.agentService.RevokeKey(r.Context(), name, ownerID) + if err != nil { + h.logger.Error("revoke key failed", "error", err) + writeJSON(w, http.StatusBadRequest, errorBody("revoke_failed", err.Error())) + return + } + + writeJSON(w, http.StatusOK, map[string]any{ + "agent": agent, + "api_key": newKey, + }) +} diff --git a/internal/api/channels_handler.go b/internal/api/channels_handler.go new file mode 100644 index 0000000..259f07f --- /dev/null +++ b/internal/api/channels_handler.go @@ -0,0 +1,248 @@ +package api + +import ( + "encoding/json" + "log/slog" + "net/http" + + "github.com/go-chi/chi/v5" + + "github.com/smart-mcp-proxy/synapbus/internal/agents" + "github.com/smart-mcp-proxy/synapbus/internal/channels" + "github.com/smart-mcp-proxy/synapbus/internal/messaging" +) + +// ChannelsHandler handles REST API requests for channels. +type ChannelsHandler struct { + channelService *channels.Service + agentService *agents.AgentService + msgService *messaging.MessagingService + logger *slog.Logger +} + +// NewChannelsHandler creates a new channels handler. +func NewChannelsHandler(channelService *channels.Service, agentService *agents.AgentService, msgService *messaging.MessagingService) *ChannelsHandler { + return &ChannelsHandler{ + channelService: channelService, + agentService: agentService, + msgService: msgService, + logger: slog.Default().With("component", "api.channels"), + } +} + +// ListChannels handles GET /api/channels. +func (h *ChannelsHandler) ListChannels(w http.ResponseWriter, r *http.Request) { + ownerID, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + // Get the first owned agent to list channels visible to it + ownedAgents, err := h.agentService.ListAgents(r.Context(), ownerID) + if err != nil { + h.logger.Error("list agents failed", "error", err) + writeJSON(w, http.StatusInternalServerError, errorBody("server_error", "Failed to list agents")) + return + } + + var allChannels []*channels.ChannelWithCount + seen := make(map[int64]bool) + + for _, agent := range ownedAgents { + chs, err := h.channelService.ListChannels(r.Context(), agent.Name) + if err != nil { + continue + } + for _, ch := range chs { + if !seen[ch.ID] { + seen[ch.ID] = true + allChannels = append(allChannels, ch) + } + } + } + + if allChannels == nil { + allChannels = []*channels.ChannelWithCount{} + } + + writeJSON(w, http.StatusOK, map[string]any{"channels": allChannels}) +} + +// GetChannel handles GET /api/channels/{name}. +func (h *ChannelsHandler) GetChannel(w http.ResponseWriter, r *http.Request) { + _, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + name := chi.URLParam(r, "name") + ch, err := h.channelService.GetChannelByName(r.Context(), name) + if err != nil { + writeJSON(w, http.StatusNotFound, errorBody("not_found", "Channel not found")) + return + } + + members, err := h.channelService.GetMembers(r.Context(), ch.ID) + if err != nil { + members = []*channels.Membership{} + } + + writeJSON(w, http.StatusOK, map[string]any{ + "channel": ch, + "members": members, + }) +} + +// CreateChannel handles POST /api/channels. +func (h *ChannelsHandler) CreateChannel(w http.ResponseWriter, r *http.Request) { + ownerID, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + var req struct { + Name string `json:"name"` + Description string `json:"description"` + Topic string `json:"topic"` + Type string `json:"type"` + IsPrivate bool `json:"is_private"` + } + + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_request", "Invalid JSON body")) + return + } + + if req.Name == "" { + writeJSON(w, http.StatusBadRequest, errorBody("validation_error", "Channel name is required")) + return + } + + // Use the first owned agent as the creator + ownedAgents, err := h.agentService.ListAgents(r.Context(), ownerID) + if err != nil || len(ownedAgents) == 0 { + writeJSON(w, http.StatusBadRequest, errorBody("no_agents", "No agents registered. Register an agent first.")) + return + } + + createReq := channels.CreateChannelRequest{ + Name: req.Name, + Description: req.Description, + Topic: req.Topic, + Type: req.Type, + IsPrivate: req.IsPrivate, + CreatedBy: ownedAgents[0].Name, + } + + ch, err := h.channelService.CreateChannel(r.Context(), createReq) + if err != nil { + h.logger.Error("create channel failed", "error", err) + writeJSON(w, http.StatusBadRequest, errorBody("create_failed", err.Error())) + return + } + + writeJSON(w, http.StatusCreated, ch) +} + +// JoinChannel handles POST /api/channels/{name}/join. +func (h *ChannelsHandler) JoinChannel(w http.ResponseWriter, r *http.Request) { + ownerID, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + name := chi.URLParam(r, "name") + ch, err := h.channelService.GetChannelByName(r.Context(), name) + if err != nil { + writeJSON(w, http.StatusNotFound, errorBody("not_found", "Channel not found")) + return + } + + var req struct { + Agent string `json:"agent"` + } + _ = json.NewDecoder(r.Body).Decode(&req) + + // If no agent specified, use the first owned agent + agentName := req.Agent + if agentName == "" { + ownedAgents, err := h.agentService.ListAgents(r.Context(), ownerID) + if err != nil || len(ownedAgents) == 0 { + writeJSON(w, http.StatusBadRequest, errorBody("no_agents", "No agents registered")) + return + } + agentName = ownedAgents[0].Name + } + + // Verify the agent belongs to this user + agent, err := h.agentService.GetAgent(r.Context(), agentName) + if err != nil { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_agent", "Agent not found")) + return + } + if agent.OwnerID != ownerID { + writeJSON(w, http.StatusForbidden, errorBody("forbidden", "You do not own this agent")) + return + } + + if err := h.channelService.JoinChannel(r.Context(), ch.ID, agentName); err != nil { + h.logger.Error("join channel failed", "error", err) + writeJSON(w, http.StatusBadRequest, errorBody("join_failed", err.Error())) + return + } + + writeJSON(w, http.StatusOK, map[string]string{"status": "joined"}) +} + +// LeaveChannel handles POST /api/channels/{name}/leave. +func (h *ChannelsHandler) LeaveChannel(w http.ResponseWriter, r *http.Request) { + ownerID, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + name := chi.URLParam(r, "name") + ch, err := h.channelService.GetChannelByName(r.Context(), name) + if err != nil { + writeJSON(w, http.StatusNotFound, errorBody("not_found", "Channel not found")) + return + } + + var req struct { + Agent string `json:"agent"` + } + _ = json.NewDecoder(r.Body).Decode(&req) + + agentName := req.Agent + if agentName == "" { + ownedAgents, err := h.agentService.ListAgents(r.Context(), ownerID) + if err != nil || len(ownedAgents) == 0 { + writeJSON(w, http.StatusBadRequest, errorBody("no_agents", "No agents registered")) + return + } + agentName = ownedAgents[0].Name + } + + agent, err := h.agentService.GetAgent(r.Context(), agentName) + if err != nil { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_agent", "Agent not found")) + return + } + if agent.OwnerID != ownerID { + writeJSON(w, http.StatusForbidden, errorBody("forbidden", "You do not own this agent")) + return + } + + if err := h.channelService.LeaveChannel(r.Context(), ch.ID, agentName); err != nil { + h.logger.Error("leave channel failed", "error", err) + writeJSON(w, http.StatusBadRequest, errorBody("leave_failed", err.Error())) + return + } + + writeJSON(w, http.StatusOK, map[string]string{"status": "left"}) +} diff --git a/internal/api/helpers.go b/internal/api/helpers.go new file mode 100644 index 0000000..5807858 --- /dev/null +++ b/internal/api/helpers.go @@ -0,0 +1,32 @@ +package api + +import ( + "encoding/json" + "net/http" +) + +// writeJSON writes a JSON response with the given status code. +func writeJSON(w http.ResponseWriter, status int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + json.NewEncoder(w).Encode(v) +} + +// errorBody creates a standard error response body. +func errorBody(code, message string) map[string]string { + return map[string]string{ + "error": code, + "message": message, + } +} + +// truncateStr truncates a string to the given length, appending "..." if truncated. +func truncateStr(s string, maxLen int) string { + if len(s) <= maxLen { + return s + } + if maxLen <= 3 { + return s[:maxLen] + } + return s[:maxLen-3] + "..." +} diff --git a/internal/api/messages_handler.go b/internal/api/messages_handler.go new file mode 100644 index 0000000..28e0994 --- /dev/null +++ b/internal/api/messages_handler.go @@ -0,0 +1,404 @@ +package api + +import ( + "encoding/json" + "log/slog" + "net/http" + "strconv" + "time" + + "github.com/go-chi/chi/v5" + + "github.com/smart-mcp-proxy/synapbus/internal/agents" + "github.com/smart-mcp-proxy/synapbus/internal/messaging" +) + +// MessagesHandler handles REST API requests for messages. +type MessagesHandler struct { + msgService *messaging.MessagingService + agentService *agents.AgentService + logger *slog.Logger +} + +// NewMessagesHandler creates a new messages handler. +func NewMessagesHandler(msgService *messaging.MessagingService, agentService *agents.AgentService) *MessagesHandler { + return &MessagesHandler{ + msgService: msgService, + agentService: agentService, + logger: slog.Default().With("component", "api.messages"), + } +} + +// ListMessages handles GET /api/messages. +func (h *MessagesHandler) ListMessages(w http.ResponseWriter, r *http.Request) { + ownerID, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + ownedAgents, err := h.agentService.ListAgents(r.Context(), ownerID) + if err != nil { + h.logger.Error("list agents failed", "error", err) + writeJSON(w, http.StatusInternalServerError, errorBody("server_error", "Failed to list agents")) + return + } + + if len(ownedAgents) == 0 { + writeJSON(w, http.StatusOK, map[string]any{"messages": []*messaging.Message{}, "total": 0}) + return + } + + limit, _ := strconv.Atoi(r.URL.Query().Get("limit")) + if limit <= 0 { + limit = 50 + } + status := r.URL.Query().Get("status") + agentFilter := r.URL.Query().Get("agent") + + var allMessages []*messaging.Message + for _, agent := range ownedAgents { + if agentFilter != "" && agent.Name != agentFilter { + continue + } + opts := messaging.ReadOptions{ + Limit: limit, + IncludeRead: true, + Status: status, + } + msgs, err := h.msgService.ReadInbox(r.Context(), agent.Name, opts) + if err != nil { + h.logger.Error("read inbox failed", "agent", agent.Name, "error", err) + continue + } + allMessages = append(allMessages, msgs...) + } + + if allMessages == nil { + allMessages = []*messaging.Message{} + } + + sortMessagesByTime(allMessages) + if len(allMessages) > limit { + allMessages = allMessages[:limit] + } + + writeJSON(w, http.StatusOK, map[string]any{ + "messages": allMessages, + "total": len(allMessages), + }) +} + +// GetMessage handles GET /api/messages/{id}. +func (h *MessagesHandler) GetMessage(w http.ResponseWriter, r *http.Request) { + ownerID, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64) + if err != nil { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_id", "Invalid message ID")) + return + } + + msg, err := h.msgService.GetMessageByID(r.Context(), id) + if err != nil { + writeJSON(w, http.StatusNotFound, errorBody("not_found", "Message not found")) + return + } + + if !h.isAgentOwnedBy(r, msg.FromAgent, ownerID) && !h.isAgentOwnedBy(r, msg.ToAgent, ownerID) { + writeJSON(w, http.StatusForbidden, errorBody("forbidden", "You do not have access to this message")) + return + } + + writeJSON(w, http.StatusOK, msg) +} + +// ListConversations handles GET /api/conversations. +func (h *MessagesHandler) ListConversations(w http.ResponseWriter, r *http.Request) { + ownerID, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + ownedAgents, err := h.agentService.ListAgents(r.Context(), ownerID) + if err != nil { + h.logger.Error("list agents failed", "error", err) + writeJSON(w, http.StatusInternalServerError, errorBody("server_error", "Failed to list agents")) + return + } + + if len(ownedAgents) == 0 { + writeJSON(w, http.StatusOK, map[string]any{"conversations": []any{}}) + return + } + + type convSummary struct { + ID int64 `json:"id"` + Subject string `json:"subject"` + LastMessage string `json:"last_message"` + LastAgent string `json:"last_agent"` + LastTime string `json:"last_time"` + MsgCount int `json:"message_count"` + } + + convMap := make(map[int64]*convSummary) + for _, agent := range ownedAgents { + opts := messaging.ReadOptions{ + Limit: 100, + IncludeRead: true, + } + msgs, err := h.msgService.ReadInbox(r.Context(), agent.Name, opts) + if err != nil { + continue + } + for _, msg := range msgs { + existing, exists := convMap[msg.ConversationID] + if !exists { + convMap[msg.ConversationID] = &convSummary{ + ID: msg.ConversationID, + LastMessage: truncateStr(msg.Body, 100), + LastAgent: msg.FromAgent, + LastTime: msg.CreatedAt.Format(time.RFC3339), + MsgCount: 1, + } + } else { + existing.MsgCount++ + lt, _ := time.Parse(time.RFC3339, existing.LastTime) + if msg.CreatedAt.After(lt) { + existing.LastMessage = truncateStr(msg.Body, 100) + existing.LastAgent = msg.FromAgent + existing.LastTime = msg.CreatedAt.Format(time.RFC3339) + } + } + } + } + + conversations := make([]*convSummary, 0, len(convMap)) + for _, c := range convMap { + conv, _, err := h.msgService.GetConversation(r.Context(), c.ID) + if err == nil { + c.Subject = conv.Subject + } + conversations = append(conversations, c) + } + + writeJSON(w, http.StatusOK, map[string]any{"conversations": conversations}) +} + +// GetConversation handles GET /api/conversations/{id}. +func (h *MessagesHandler) GetConversation(w http.ResponseWriter, r *http.Request) { + ownerID, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64) + if err != nil { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_id", "Invalid conversation ID")) + return + } + + conv, messages, err := h.msgService.GetConversation(r.Context(), id) + if err != nil { + writeJSON(w, http.StatusNotFound, errorBody("not_found", "Conversation not found")) + return + } + + hasAccess := false + for _, msg := range messages { + if h.isAgentOwnedBy(r, msg.FromAgent, ownerID) || h.isAgentOwnedBy(r, msg.ToAgent, ownerID) { + hasAccess = true + break + } + } + if !hasAccess && len(messages) > 0 { + writeJSON(w, http.StatusForbidden, errorBody("forbidden", "You do not have access to this conversation")) + return + } + + writeJSON(w, http.StatusOK, map[string]any{ + "conversation": conv, + "messages": messages, + }) +} + +// SendMessage handles POST /api/messages. +func (h *MessagesHandler) SendMessage(w http.ResponseWriter, r *http.Request) { + ownerID, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + var req struct { + From string `json:"from"` + To string `json:"to"` + Body string `json:"body"` + Priority int `json:"priority"` + ChannelID *int64 `json:"channel_id,omitempty"` + Subject string `json:"subject,omitempty"` + } + + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_request", "Invalid JSON body")) + return + } + + if req.Body == "" { + writeJSON(w, http.StatusBadRequest, errorBody("validation_error", "Message body is required")) + return + } + + if req.From == "" { + ownedAgents, err := h.agentService.ListAgents(r.Context(), ownerID) + if err != nil || len(ownedAgents) == 0 { + writeJSON(w, http.StatusBadRequest, errorBody("no_agents", "No agents registered. Register an agent first.")) + return + } + req.From = ownedAgents[0].Name + } + + if !h.isAgentOwnedBy(r, req.From, ownerID) { + writeJSON(w, http.StatusForbidden, errorBody("forbidden", "You do not own agent: "+req.From)) + return + } + + opts := messaging.SendOptions{ + Priority: req.Priority, + ChannelID: req.ChannelID, + Subject: req.Subject, + } + + msg, err := h.msgService.SendMessage(r.Context(), req.From, req.To, req.Body, opts) + if err != nil { + h.logger.Error("send message failed", "error", err) + writeJSON(w, http.StatusBadRequest, errorBody("send_failed", err.Error())) + return + } + + writeJSON(w, http.StatusCreated, msg) +} + +// MarkDone handles POST /api/messages/{id}/done. +func (h *MessagesHandler) MarkDone(w http.ResponseWriter, r *http.Request) { + ownerID, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64) + if err != nil { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_id", "Invalid message ID")) + return + } + + msg, err := h.msgService.GetMessageByID(r.Context(), id) + if err != nil { + writeJSON(w, http.StatusNotFound, errorBody("not_found", "Message not found")) + return + } + + agentToCheck := msg.ToAgent + if msg.ClaimedBy != "" { + agentToCheck = msg.ClaimedBy + } + if !h.isAgentOwnedBy(r, agentToCheck, ownerID) { + writeJSON(w, http.StatusForbidden, errorBody("forbidden", "You do not have access to this message")) + return + } + + if msg.Status == messaging.StatusPending { + _, _ = h.msgService.ClaimMessages(r.Context(), msg.ToAgent, 1) + msg, err = h.msgService.GetMessageByID(r.Context(), id) + if err != nil { + writeJSON(w, http.StatusNotFound, errorBody("not_found", "Message not found")) + return + } + } + + if msg.Status != messaging.StatusProcessing { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_status", "Message is not in processing status")) + return + } + + if err := h.msgService.MarkDone(r.Context(), id, msg.ClaimedBy); err != nil { + h.logger.Error("mark done failed", "error", err) + writeJSON(w, http.StatusInternalServerError, errorBody("server_error", err.Error())) + return + } + + writeJSON(w, http.StatusOK, map[string]string{"status": "done"}) +} + +// SearchMessages handles GET /api/messages/search. +func (h *MessagesHandler) SearchMessages(w http.ResponseWriter, r *http.Request) { + ownerID, ok := OwnerIDFromContext(r.Context()) + if !ok { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required")) + return + } + + query := r.URL.Query().Get("q") + if query == "" { + writeJSON(w, http.StatusBadRequest, errorBody("missing_query", "Search query 'q' is required")) + return + } + + ownedAgents, err := h.agentService.ListAgents(r.Context(), ownerID) + if err != nil { + h.logger.Error("list agents failed", "error", err) + writeJSON(w, http.StatusInternalServerError, errorBody("server_error", "Failed to list agents")) + return + } + + limit, _ := strconv.Atoi(r.URL.Query().Get("limit")) + if limit <= 0 { + limit = 20 + } + + var allMessages []*messaging.Message + for _, agent := range ownedAgents { + opts := messaging.SearchOptions{Limit: limit} + msgs, err := h.msgService.SearchMessages(r.Context(), agent.Name, query, opts) + if err != nil { + continue + } + allMessages = append(allMessages, msgs...) + } + + if allMessages == nil { + allMessages = []*messaging.Message{} + } + + writeJSON(w, http.StatusOK, map[string]any{ + "messages": allMessages, + "query": query, + "total": len(allMessages), + }) +} + +func (h *MessagesHandler) isAgentOwnedBy(r *http.Request, agentName string, ownerID int64) bool { + if agentName == "" { + return false + } + agent, err := h.agentService.GetAgent(r.Context(), agentName) + if err != nil { + return false + } + return agent.OwnerID == ownerID +} + +func sortMessagesByTime(msgs []*messaging.Message) { + for i := 1; i < len(msgs); i++ { + for j := i; j > 0 && msgs[j].CreatedAt.After(msgs[j-1].CreatedAt); j-- { + msgs[j], msgs[j-1] = msgs[j-1], msgs[j] + } + } +} diff --git a/internal/api/router.go b/internal/api/router.go index c2a33cc..a498b4c 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -5,25 +5,56 @@ import ( "github.com/go-chi/chi/v5" + "github.com/smart-mcp-proxy/synapbus/internal/agents" "github.com/smart-mcp-proxy/synapbus/internal/attachments" + "github.com/smart-mcp-proxy/synapbus/internal/channels" + "github.com/smart-mcp-proxy/synapbus/internal/messaging" "github.com/smart-mcp-proxy/synapbus/internal/trace" ) +// RouterConfig holds optional services for the API router. +// Fields may be nil if the corresponding feature is not configured. +type RouterConfig struct { + TraceStore trace.TraceStore + Metrics *trace.Metrics + AttachmentService *attachments.Service + MsgService *messaging.MessagingService + AgentService *agents.AgentService + ChannelService *channels.Service + SSEHub *SSEHub + SessionMiddleware func(http.Handler) http.Handler +} + // NewRouter creates a chi router with all API routes configured. // metricsInstance may be nil if metrics are disabled. // attachmentService may be nil if attachments are not configured. func NewRouter(traceStore trace.TraceStore, metricsInstance *trace.Metrics, attachmentService *attachments.Service) chi.Router { + return NewRouterWithConfig(RouterConfig{ + TraceStore: traceStore, + Metrics: metricsInstance, + AttachmentService: attachmentService, + }) +} + +// NewRouterWithConfig creates a chi router using the full configuration. +func NewRouterWithConfig(cfg RouterConfig) chi.Router { r := chi.NewRouter() // Global middleware r.Use(RequestIDMiddleware) r.Use(LoggingMiddleware) - tracesHandler := NewTracesHandler(traceStore) + tracesHandler := NewTracesHandler(cfg.TraceStore) - // Authenticated API routes + // Determine which auth middleware to use for API routes + authMiddleware := OwnerAuthMiddleware + if cfg.SessionMiddleware != nil { + authMiddleware = cfg.SessionMiddleware + } + + // Authenticated API routes (traces) r.Group(func(r chi.Router) { - r.Use(OwnerAuthMiddleware) + r.Use(authMiddleware) r.Get("/api/traces", tracesHandler.ListTraces) r.Get("/api/traces/export", tracesHandler.ExportTraces) @@ -31,21 +62,71 @@ func NewRouter(traceStore trace.TraceStore, metricsInstance *trace.Metrics, atta }) // Attachment API routes (for Web UI) - if attachmentService != nil { - attachmentsHandler := NewAttachmentsHandler(attachmentService) + if cfg.AttachmentService != nil { + attachmentsHandler := NewAttachmentsHandler(cfg.AttachmentService) r.Get("/api/attachments/{hash}", attachmentsHandler.Download) r.Get("/api/attachments/{hash}/meta", attachmentsHandler.Metadata) r.Group(func(r chi.Router) { - r.Use(OwnerAuthMiddleware) + r.Use(authMiddleware) r.Post("/api/attachments", attachmentsHandler.Upload) }) } + // Web UI API routes (messages, agents, channels, SSE) + if cfg.MsgService != nil && cfg.AgentService != nil { + messagesHandler := NewMessagesHandler(cfg.MsgService, cfg.AgentService) + agentsHandler := NewAgentsHandler(cfg.AgentService, cfg.TraceStore) + + r.Group(func(r chi.Router) { + r.Use(authMiddleware) + + // Messages + r.Get("/api/messages", messagesHandler.ListMessages) + r.Get("/api/messages/search", messagesHandler.SearchMessages) + r.Get("/api/messages/{id}", messagesHandler.GetMessage) + r.Post("/api/messages", messagesHandler.SendMessage) + r.Post("/api/messages/{id}/done", messagesHandler.MarkDone) + + // Conversations + r.Get("/api/conversations", messagesHandler.ListConversations) + r.Get("/api/conversations/{id}", messagesHandler.GetConversation) + + // Agents + r.Get("/api/agents", agentsHandler.ListAgents) + r.Get("/api/agents/{name}", agentsHandler.GetAgent) + r.Post("/api/agents", agentsHandler.RegisterAgent) + r.Delete("/api/agents/{name}", agentsHandler.DeleteAgent) + r.Post("/api/agents/{name}/revoke-key", agentsHandler.RevokeKey) + }) + + // Channels + if cfg.ChannelService != nil { + channelsHandler := NewChannelsHandler(cfg.ChannelService, cfg.AgentService, cfg.MsgService) + r.Group(func(r chi.Router) { + r.Use(authMiddleware) + + r.Get("/api/channels", channelsHandler.ListChannels) + r.Get("/api/channels/{name}", channelsHandler.GetChannel) + r.Post("/api/channels", channelsHandler.CreateChannel) + r.Post("/api/channels/{name}/join", channelsHandler.JoinChannel) + r.Post("/api/channels/{name}/leave", channelsHandler.LeaveChannel) + }) + } + + // SSE events + if cfg.SSEHub != nil { + r.Group(func(r chi.Router) { + r.Use(authMiddleware) + r.Get("/api/events", cfg.SSEHub.HandleEvents) + }) + } + } + // Metrics endpoint (unauthenticated, only registered when enabled) - if metricsInstance != nil { + if cfg.Metrics != nil { r.Get("/metrics", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8") - metricsInstance.WritePrometheus(w) + cfg.Metrics.WritePrometheus(w) }) } diff --git a/internal/api/session_middleware.go b/internal/api/session_middleware.go new file mode 100644 index 0000000..de17a40 --- /dev/null +++ b/internal/api/session_middleware.go @@ -0,0 +1,25 @@ +package api + +import ( + "net/http" + + "github.com/smart-mcp-proxy/synapbus/internal/auth" +) + +// SessionToOwnerMiddleware wraps RequireSession and extracts the owner ID +// into the API context so that API handlers can use OwnerIDFromContext. +func SessionToOwnerMiddleware(userStore auth.UserStore, sessionStore auth.SessionStore) func(http.Handler) http.Handler { + sessionMW := auth.RequireSession(userStore, sessionStore) + + return func(next http.Handler) http.Handler { + return sessionMW(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + user, ok := auth.UserFromContext(r.Context()) + if !ok { + http.Error(w, `{"error":"unauthorized","message":"Authentication required"}`, http.StatusUnauthorized) + return + } + ctx := ContextWithOwnerID(r.Context(), user.ID) + next.ServeHTTP(w, r.WithContext(ctx)) + })) + } +} diff --git a/internal/api/sse_handler.go b/internal/api/sse_handler.go new file mode 100644 index 0000000..870460c --- /dev/null +++ b/internal/api/sse_handler.go @@ -0,0 +1,154 @@ +package api + +import ( + "encoding/json" + "fmt" + "log/slog" + "net/http" + "sync" + "time" +) + +// SSEEvent represents a server-sent event. +type SSEEvent struct { + Type string `json:"type"` + Data any `json:"data"` +} + +// SSEHub manages SSE client connections and event broadcasting. +type SSEHub struct { + mu sync.RWMutex + clients map[int64]map[chan SSEEvent]struct{} // ownerID -> set of channels + nextID int64 + logger *slog.Logger +} + +// NewSSEHub creates a new SSE hub. +func NewSSEHub() *SSEHub { + return &SSEHub{ + clients: make(map[int64]map[chan SSEEvent]struct{}), + logger: slog.Default().With("component", "api.sse"), + } +} + +// Broadcast sends an event to all clients for the given owner. +func (h *SSEHub) Broadcast(ownerID int64, event SSEEvent) { + h.mu.RLock() + defer h.mu.RUnlock() + + clientSet, ok := h.clients[ownerID] + if !ok { + return + } + + for ch := range clientSet { + select { + case ch <- event: + default: + // Client channel full, skip + } + } +} + +// BroadcastAll sends an event to all connected clients. +func (h *SSEHub) BroadcastAll(event SSEEvent) { + h.mu.RLock() + defer h.mu.RUnlock() + + for _, clientSet := range h.clients { + for ch := range clientSet { + select { + case ch <- event: + default: + } + } + } +} + +func (h *SSEHub) addClient(ownerID int64) chan SSEEvent { + h.mu.Lock() + defer h.mu.Unlock() + + ch := make(chan SSEEvent, 32) + if _, ok := h.clients[ownerID]; !ok { + h.clients[ownerID] = make(map[chan SSEEvent]struct{}) + } + h.clients[ownerID][ch] = struct{}{} + + h.logger.Info("SSE client connected", "owner_id", ownerID) + return ch +} + +func (h *SSEHub) removeClient(ownerID int64, ch chan SSEEvent) { + h.mu.Lock() + defer h.mu.Unlock() + + if clientSet, ok := h.clients[ownerID]; ok { + delete(clientSet, ch) + if len(clientSet) == 0 { + delete(h.clients, ownerID) + } + } + close(ch) + + h.logger.Info("SSE client disconnected", "owner_id", ownerID) +} + +// HandleEvents handles GET /api/events — the SSE endpoint. +func (h *SSEHub) HandleEvents(w http.ResponseWriter, r *http.Request) { + ownerID, ok := OwnerIDFromContext(r.Context()) + if !ok { + http.Error(w, `{"error":"unauthorized"}`, http.StatusUnauthorized) + return + } + + flusher, ok := w.(http.Flusher) + if !ok { + http.Error(w, "Streaming not supported", http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") + + ch := h.addClient(ownerID) + defer h.removeClient(ownerID, ch) + + // Send initial connected event + writeSSE(w, flusher, "connected", map[string]any{ + "owner_id": ownerID, + "timestamp": time.Now().Format(time.RFC3339), + }) + + // Heartbeat ticker + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + + ctx := r.Context() + for { + select { + case <-ctx.Done(): + return + case event, ok := <-ch: + if !ok { + return + } + writeSSE(w, flusher, event.Type, event.Data) + case <-ticker.C: + writeSSE(w, flusher, "heartbeat", map[string]any{ + "timestamp": time.Now().Format(time.RFC3339), + }) + } + } +} + +func writeSSE(w http.ResponseWriter, flusher http.Flusher, eventType string, data any) { + jsonData, err := json.Marshal(data) + if err != nil { + return + } + fmt.Fprintf(w, "event: %s\ndata: %s\n\n", eventType, string(jsonData)) + flusher.Flush() +} diff --git a/internal/messaging/service.go b/internal/messaging/service.go index 5b09078..3b31949 100644 --- a/internal/messaging/service.go +++ b/internal/messaging/service.go @@ -314,6 +314,18 @@ func (s *MessagingService) SearchMessages(ctx context.Context, agentName, query return messages, nil } +// GetMessageByID returns a single message by its ID. +func (s *MessagingService) GetMessageByID(ctx context.Context, id int64) (*Message, error) { + msg, err := s.store.GetMessageByID(ctx, id) + if err != nil { + if err == sql.ErrNoRows { + return nil, fmt.Errorf("message not found: %d", id) + } + return nil, fmt.Errorf("get message: %w", err) + } + return msg, nil +} + // GetConversation returns a conversation and its messages. func (s *MessagingService) GetConversation(ctx context.Context, id int64) (*Conversation, []*Message, error) { conv, err := s.store.GetConversation(ctx, id) diff --git a/internal/web/dist/index.html b/internal/web/dist/index.html new file mode 100644 index 0000000..341b8f9 --- /dev/null +++ b/internal/web/dist/index.html @@ -0,0 +1,20 @@ + + + + + +SynapBus + + + +
+

SynapBus

+

Web UI placeholder. Run make web to build the Svelte SPA.

+
+ + diff --git a/internal/web/embed.go b/internal/web/embed.go new file mode 100644 index 0000000..d3786fa --- /dev/null +++ b/internal/web/embed.go @@ -0,0 +1,7 @@ +// Package web embeds the built Svelte SPA for serving from the Go binary. +package web + +import "embed" + +//go:embed dist/* +var DistFS embed.FS diff --git a/internal/web/handler.go b/internal/web/handler.go new file mode 100644 index 0000000..2ef488f --- /dev/null +++ b/internal/web/handler.go @@ -0,0 +1,43 @@ +package web + +import ( + "io/fs" + "net/http" + "strings" +) + +// NewSPAHandler returns an http.Handler that serves the embedded SPA. +// Static files are served directly; all other routes serve index.html +// to support client-side routing. +func NewSPAHandler() http.Handler { + subFS, err := fs.Sub(DistFS, "dist") + if err != nil { + // Should never happen — if dist/ is missing the binary won't compile. + panic("web: embedded dist/ directory not found: " + err.Error()) + } + + fileServer := http.FileServer(http.FS(subFS)) + + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + path := r.URL.Path + + // Strip leading slash for fs.Open + fsPath := strings.TrimPrefix(path, "/") + if fsPath == "" { + fsPath = "index.html" + } + + // Try to open the file in the embedded FS + f, err := subFS.Open(fsPath) + if err == nil { + f.Close() + // File exists — serve it directly + fileServer.ServeHTTP(w, r) + return + } + + // File doesn't exist — serve index.html for SPA routing + r.URL.Path = "/" + fileServer.ServeHTTP(w, r) + }) +} diff --git a/web/package.json b/web/package.json new file mode 100644 index 0000000..a2c1594 --- /dev/null +++ b/web/package.json @@ -0,0 +1,22 @@ +{ + "name": "synapbus-web", + "version": "0.1.0", + "private": true, + "scripts": { + "dev": "vite dev", + "build": "vite build", + "preview": "vite preview" + }, + "devDependencies": { + "@sveltejs/adapter-static": "^3.0.0", + "@sveltejs/kit": "^2.0.0", + "@sveltejs/vite-plugin-svelte": "^4.0.0", + "autoprefixer": "^10.4.0", + "postcss": "^8.4.0", + "svelte": "^5.0.0", + "tailwindcss": "^3.4.0", + "typescript": "^5.0.0", + "vite": "^6.0.0" + }, + "type": "module" +} diff --git a/web/postcss.config.js b/web/postcss.config.js new file mode 100644 index 0000000..0f77216 --- /dev/null +++ b/web/postcss.config.js @@ -0,0 +1,6 @@ +export default { + plugins: { + tailwindcss: {}, + autoprefixer: {} + } +}; diff --git a/web/src/app.css b/web/src/app.css new file mode 100644 index 0000000..561a385 --- /dev/null +++ b/web/src/app.css @@ -0,0 +1,45 @@ +@tailwind base; +@tailwind components; +@tailwind utilities; + +@layer base { + body { + @apply bg-white text-gray-900 dark:bg-gray-900 dark:text-gray-100; + } +} + +@layer components { + .btn-primary { + @apply bg-primary-600 hover:bg-primary-700 text-white px-4 py-2 rounded-lg font-medium transition-colors disabled:opacity-50 disabled:cursor-not-allowed; + } + .btn-secondary { + @apply bg-gray-200 hover:bg-gray-300 dark:bg-gray-700 dark:hover:bg-gray-600 text-gray-900 dark:text-gray-100 px-4 py-2 rounded-lg font-medium transition-colors; + } + .btn-danger { + @apply bg-red-600 hover:bg-red-700 text-white px-4 py-2 rounded-lg font-medium transition-colors disabled:opacity-50; + } + .input { + @apply w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-lg bg-white dark:bg-gray-800 text-gray-900 dark:text-gray-100 focus:ring-2 focus:ring-primary-500 focus:border-transparent outline-none transition-colors; + } + .card { + @apply bg-white dark:bg-gray-800 rounded-lg shadow-sm border border-gray-200 dark:border-gray-700; + } + .badge { + @apply inline-flex items-center px-2.5 py-0.5 rounded-full text-xs font-medium; + } + .badge-pending { + @apply badge bg-yellow-100 text-yellow-800 dark:bg-yellow-900 dark:text-yellow-200; + } + .badge-processing { + @apply badge bg-blue-100 text-blue-800 dark:bg-blue-900 dark:text-blue-200; + } + .badge-done { + @apply badge bg-green-100 text-green-800 dark:bg-green-900 dark:text-green-200; + } + .badge-failed { + @apply badge bg-red-100 text-red-800 dark:bg-red-900 dark:text-red-200; + } + .skeleton { + @apply animate-pulse bg-gray-200 dark:bg-gray-700 rounded; + } +} diff --git a/web/src/app.html b/web/src/app.html new file mode 100644 index 0000000..9d7862b --- /dev/null +++ b/web/src/app.html @@ -0,0 +1,20 @@ + + + + + + + SynapBus + + %sveltekit.head% + + +
%sveltekit.body%
+ + diff --git a/web/src/lib/api/client.ts b/web/src/lib/api/client.ts new file mode 100644 index 0000000..bf3e411 --- /dev/null +++ b/web/src/lib/api/client.ts @@ -0,0 +1,104 @@ +/** API client for SynapBus REST API */ + +class ApiError extends Error { + status: number; + code: string; + + constructor(status: number, code: string, message: string) { + super(message); + this.status = status; + this.code = code; + this.name = 'ApiError'; + } +} + +async function request(method: string, path: string, body?: unknown): Promise { + const opts: RequestInit = { + method, + headers: { 'Content-Type': 'application/json' }, + credentials: 'same-origin' + }; + if (body) { + opts.body = JSON.stringify(body); + } + + const res = await fetch(path, opts); + + if (res.status === 401) { + // Redirect to login on auth failure + if (typeof window !== 'undefined' && !window.location.pathname.startsWith('/login')) { + window.location.href = `/login?return=${encodeURIComponent(window.location.pathname)}`; + } + throw new ApiError(401, 'unauthorized', 'Session expired'); + } + + if (!res.ok) { + const err = await res.json().catch(() => ({ error: 'unknown', message: res.statusText })); + throw new ApiError(res.status, err.error || 'unknown', err.message || err.error_description || res.statusText); + } + + if (res.status === 204) return undefined as T; + return res.json(); +} + +// Auth +export const auth = { + login: (username: string, password: string) => + request<{ id: number; username: string; display_name: string }>('POST', '/auth/login', { username, password }), + logout: () => request('POST', '/auth/logout'), + me: () => request<{ id: number; username: string; display_name: string; role: string }>('GET', '/auth/me'), + changePassword: (current_password: string, new_password: string) => + request<{ status: string }>('PUT', '/auth/password', { current_password, new_password }) +}; + +// Messages +export const messages = { + list: (params?: { limit?: number; status?: string; agent?: string }) => { + const qs = new URLSearchParams(); + if (params?.limit) qs.set('limit', String(params.limit)); + if (params?.status) qs.set('status', params.status); + if (params?.agent) qs.set('agent', params.agent); + const q = qs.toString(); + return request<{ messages: any[]; total: number }>('GET', `/api/messages${q ? '?' + q : ''}`); + }, + get: (id: number) => request('GET', `/api/messages/${id}`), + send: (body: { from?: string; to: string; body: string; priority?: number; subject?: string; channel_id?: number }) => + request('POST', '/api/messages', body), + markDone: (id: number) => request<{ status: string }>('POST', `/api/messages/${id}/done`), + search: (q: string, limit?: number) => { + const qs = new URLSearchParams({ q }); + if (limit) qs.set('limit', String(limit)); + return request<{ messages: any[]; query: string; total: number }>('GET', `/api/messages/search?${qs}`); + } +}; + +// Conversations +export const conversations = { + list: () => request<{ conversations: any[] }>('GET', '/api/conversations'), + get: (id: number) => request<{ conversation: any; messages: any[] }>('GET', `/api/conversations/${id}`) +}; + +// Agents +export const agents = { + list: () => request<{ agents: any[] }>('GET', '/api/agents'), + get: (name: string) => request<{ agent: any; traces: any[] }>('GET', `/api/agents/${encodeURIComponent(name)}`), + register: (body: { name: string; display_name?: string; type?: string; capabilities?: object }) => + request<{ agent: any; api_key: string }>('POST', '/api/agents', body), + delete: (name: string) => request<{ status: string }>('DELETE', `/api/agents/${encodeURIComponent(name)}`), + revokeKey: (name: string) => + request<{ agent: any; api_key: string }>('POST', `/api/agents/${encodeURIComponent(name)}/revoke-key`) +}; + +// Channels +export const channels = { + list: () => request<{ channels: any[] }>('GET', '/api/channels'), + get: (name: string) => request<{ channel: any; members: any[] }>('GET', `/api/channels/${encodeURIComponent(name)}`), + create: (body: { name: string; description?: string; topic?: string; type?: string; is_private?: boolean }) => + request('POST', '/api/channels', body), + join: (name: string, agent?: string) => + request<{ status: string }>('POST', `/api/channels/${encodeURIComponent(name)}/join`, agent ? { agent } : {}), + leave: (name: string, agent?: string) => + request<{ status: string }>('POST', `/api/channels/${encodeURIComponent(name)}/leave`, agent ? { agent } : {}) +}; + +export { ApiError }; diff --git a/web/src/lib/api/sse.ts b/web/src/lib/api/sse.ts new file mode 100644 index 0000000..d29d3c1 --- /dev/null +++ b/web/src/lib/api/sse.ts @@ -0,0 +1,74 @@ +/** SSE client with auto-reconnect and exponential backoff */ + +type SSEHandler = (event: { type: string; data: any }) => void; + +export class SSEClient { + private url: string; + private eventSource: EventSource | null = null; + private handlers: SSEHandler[] = []; + private reconnectDelay = 1000; + private maxReconnectDelay = 30000; + private shouldReconnect = true; + + constructor(url: string = '/api/events') { + this.url = url; + } + + connect() { + if (this.eventSource) { + this.eventSource.close(); + } + + this.eventSource = new EventSource(this.url, { withCredentials: true }); + + this.eventSource.onopen = () => { + this.reconnectDelay = 1000; // Reset backoff on successful connection + }; + + // Listen for typed events + const eventTypes = ['connected', 'new_message', 'message_updated', 'agent_connected', 'agent_disconnected', 'heartbeat']; + for (const type of eventTypes) { + this.eventSource.addEventListener(type, (e: MessageEvent) => { + try { + const data = JSON.parse(e.data); + this.emit({ type, data }); + } catch { + // Ignore parse errors + } + }); + } + + this.eventSource.onerror = () => { + this.eventSource?.close(); + this.eventSource = null; + + if (this.shouldReconnect) { + setTimeout(() => this.connect(), this.reconnectDelay); + this.reconnectDelay = Math.min(this.reconnectDelay * 2, this.maxReconnectDelay); + } + }; + } + + onEvent(handler: SSEHandler) { + this.handlers.push(handler); + return () => { + this.handlers = this.handlers.filter((h) => h !== handler); + }; + } + + private emit(event: { type: string; data: any }) { + for (const handler of this.handlers) { + handler(event); + } + } + + disconnect() { + this.shouldReconnect = false; + this.eventSource?.close(); + this.eventSource = null; + } + + get connected(): boolean { + return this.eventSource?.readyState === EventSource.OPEN; + } +} diff --git a/web/src/lib/components/AgentCard.svelte b/web/src/lib/components/AgentCard.svelte new file mode 100644 index 0000000..e096bc1 --- /dev/null +++ b/web/src/lib/components/AgentCard.svelte @@ -0,0 +1,31 @@ + + + +
+
+

{agent.display_name || agent.name}

+ {#if agent.display_name} +

@{agent.name}

+ {/if} +
+
+ + {agent.type} + + + {agent.status} + +
+
+ diff --git a/web/src/lib/components/ComposeForm.svelte b/web/src/lib/components/ComposeForm.svelte new file mode 100644 index 0000000..ae756c6 --- /dev/null +++ b/web/src/lib/components/ComposeForm.svelte @@ -0,0 +1,84 @@ + + +
+

New Message

+ + {#if error} +
+ {error} +
+ {/if} + +
+ + + + {#if showAdvanced} + +
+ + + {priority} +
+ {/if} + +
+ + +
+
+
diff --git a/web/src/lib/components/Header.svelte b/web/src/lib/components/Header.svelte new file mode 100644 index 0000000..6ac7f4e --- /dev/null +++ b/web/src/lib/components/Header.svelte @@ -0,0 +1,93 @@ + + +
+ + + + +
+
+ + + + +
+
+ +
+ + + + +
+ + + {#if showUserMenu} +
+ (showUserMenu = false)}> + Settings + + +
+ {/if} +
+
+
diff --git a/web/src/lib/components/MessageList.svelte b/web/src/lib/components/MessageList.svelte new file mode 100644 index 0000000..d1f3e3d --- /dev/null +++ b/web/src/lib/components/MessageList.svelte @@ -0,0 +1,82 @@ + + +{#if messages.length === 0} +
+ + + +

No messages yet

+
+{:else} +
+ {#each messages as msg (msg.id)} +
+
+
+
+ {msg.from_agent} + {#if msg.to_agent} + + + + {msg.to_agent} + {/if} + {msg.status} + {#if msg.priority >= 8} + P{msg.priority} + {:else if msg.priority >= 5} + P{msg.priority} + {/if} +
+

{truncate(msg.body, 200)}

+
+
+ {formatTime(msg.created_at)} + {#if showConversationLink} + + View thread + + {/if} +
+
+
+ {/each} +
+{/if} diff --git a/web/src/lib/components/Sidebar.svelte b/web/src/lib/components/Sidebar.svelte new file mode 100644 index 0000000..be1941f --- /dev/null +++ b/web/src/lib/components/Sidebar.svelte @@ -0,0 +1,47 @@ + + + +{#if open} +
(open = false)}>
+{/if} + + diff --git a/web/src/lib/components/TraceViewer.svelte b/web/src/lib/components/TraceViewer.svelte new file mode 100644 index 0000000..760c44a --- /dev/null +++ b/web/src/lib/components/TraceViewer.svelte @@ -0,0 +1,56 @@ + + +{#if traces.length === 0} +
+

No activity traces

+
+{:else} +
+ {#each traces as trace (trace.id)} +
+ + {#if expandedId === trace.id} +
+ 
{JSON.stringify(trace.details, null, 2)}
+ {#if trace.error} +

Error: {trace.error}

+ {/if} +
+ {/if} +
+ {/each} +
+{/if} diff --git a/web/src/lib/stores/auth.ts b/web/src/lib/stores/auth.ts new file mode 100644 index 0000000..20ba36b --- /dev/null +++ b/web/src/lib/stores/auth.ts @@ -0,0 +1,36 @@ +import { writable } from 'svelte/store'; +import { auth as authApi } from '$lib/api/client'; + +export type User = { + id: number; + username: string; + display_name: string; + role: string; +} | null; + +export const user = writable(null); +export const loading = writable(true); + +export async function checkAuth(): Promise { + try { + loading.set(true); + const u = await authApi.me(); + user.set(u); + return true; + } catch { + user.set(null); + return false; + } finally { + loading.set(false); + } +} + +export async function login(username: string, password: string): Promise { + const u = await authApi.login(username, password); + user.set({ ...u, role: 'user' }); +} + +export async function logout(): Promise { + await authApi.logout(); + user.set(null); +} diff --git a/web/src/lib/stores/theme.ts b/web/src/lib/stores/theme.ts new file mode 100644 index 0000000..0322d86 --- /dev/null +++ b/web/src/lib/stores/theme.ts @@ -0,0 +1,27 @@ +import { writable } from 'svelte/store'; +import { browser } from '$app/environment'; + +function getInitialTheme(): 'light' | 'dark' { + if (!browser) return 'light'; + const stored = localStorage.getItem('synapbus-theme'); + if (stored === 'dark') return 'dark'; + if (stored === 'light') return 'light'; + return window.matchMedia('(prefers-color-scheme: dark)').matches ? 'dark' : 'light'; +} + +export const theme = writable<'light' | 'dark'>(getInitialTheme()); + +export function toggleTheme() { + theme.update((t) => { + const next = t === 'light' ? 'dark' : 'light'; + if (browser) { + localStorage.setItem('synapbus-theme', next); + if (next === 'dark') { + document.documentElement.classList.add('dark'); + } else { + document.documentElement.classList.remove('dark'); + } + } + return next; + }); +} diff --git a/web/src/routes/+layout.svelte b/web/src/routes/+layout.svelte new file mode 100644 index 0000000..bcb9d7c --- /dev/null +++ b/web/src/routes/+layout.svelte @@ -0,0 +1,59 @@ + + +{#if $loading} +
+
+
+{:else if isLoginPage} + {@render children()} +{:else if $user} +
+ +
+
(sidebarOpen = !sidebarOpen)} /> +
+ {@render children()} +
+
+
+{/if} diff --git a/web/src/routes/+layout.ts b/web/src/routes/+layout.ts new file mode 100644 index 0000000..83addb7 --- /dev/null +++ b/web/src/routes/+layout.ts @@ -0,0 +1,2 @@ +export const ssr = false; +export const prerender = false; diff --git a/web/src/routes/+page.svelte b/web/src/routes/+page.svelte new file mode 100644 index 0000000..e42d4bc --- /dev/null +++ b/web/src/routes/+page.svelte @@ -0,0 +1,123 @@ + + +
+
+

Dashboard

+ +
+ + {#if showCompose} +
+ { showCompose = false; loadData(); }} /> +
+ {/if} + + +
+
+

Messages

+

{loadingData ? '-' : recentMessages.length}

+
+
+

Conversations

+

{loadingData ? '-' : recentConversations.length}

+
+
+

Agents

+

{loadingData ? '-' : agentCount}

+
+
+ + + {#if recentConversations.length > 0} +
+
+

Recent Conversations

+
+
+ {#each recentConversations.slice(0, 5) as conv} + +
+
+

+ {conv.subject || 'Untitled conversation'} +

+

+ {conv.last_agent}: {conv.last_message} +

+
+
+ {conv.message_count} msgs +
+
+ + {/each} +
+ {#if recentConversations.length > 5} +
+ View all conversations +
+ {/if} +
+ {/if} + + +
+
+

Recent Messages

+
+ {#if loadingData} +
+ {#each Array(3) as _} +
+
+
+
+ {/each} +
+ {:else} + + {/if} +
+ + {#if !loadingData && recentMessages.length === 0 && agentCount === 0} +
+

Welcome to SynapBus

+

No conversations yet. Register an agent to get started.

+ Register an Agent +
+ {/if} +
diff --git a/web/src/routes/agents/+page.svelte b/web/src/routes/agents/+page.svelte new file mode 100644 index 0000000..5573ba7 --- /dev/null +++ b/web/src/routes/agents/+page.svelte @@ -0,0 +1,114 @@ + + +
+
+

Agents

+ +
+ + {#if showRegister} +
+

Register New Agent

+ {#if registerError} +
{registerError}
+ {/if} + {#if newApiKey} +
+

Agent registered! Save this API key - it will not be shown again:

+ {newApiKey} +
+ {/if} +
+ + + + +
+
+ {/if} + + {#if loadingData} +
+ {#each Array(4) as _} +
+
+
+
+ {/each} +
+ {:else if agentList.length === 0} +
+

No agents registered yet.

+
+ {:else} +
+ {#each agentList as agent (agent.id)} + + {/each} +
+ {/if} +
diff --git a/web/src/routes/agents/[name]/+page.svelte b/web/src/routes/agents/[name]/+page.svelte new file mode 100644 index 0000000..2ffa02c --- /dev/null +++ b/web/src/routes/agents/[name]/+page.svelte @@ -0,0 +1,153 @@ + + +
+ + ← Back to agents + + + {#if loadingData} +
+
+
+
+
+ {:else if agent} + +
+
+
+
+

{agent.display_name || agent.name}

+ {#if agent.display_name} +

@{agent.name}

+ {/if} +
+
+ + {agent.type} + + + {agent.status} + +
+
+
+ +
+
+
+

Created

+

{new Date(agent.created_at).toLocaleString()}

+
+
+

Last Updated

+

{new Date(agent.updated_at).toLocaleString()}

+
+
+ + {#if agent.capabilities && JSON.stringify(agent.capabilities) !== '{}'} +
+

Capabilities

+ 
{JSON.stringify(agent.capabilities, null, 2)}
+
+ {/if} +
+ + +
+ {#if revokeKey} +
+

New API key generated. Save it now - it will not be shown again:

+ {revokeKey} +
+ {/if} + +
+ + {#if !confirmDelete} + + {:else} + + + {/if} +
+
+
+ + +
+
+

Activity Traces

+
+ +
+ {:else} +
+

Agent not found

+
+ {/if} +
diff --git a/web/src/routes/channels/+page.svelte b/web/src/routes/channels/+page.svelte new file mode 100644 index 0000000..9b9ec7f --- /dev/null +++ b/web/src/routes/channels/+page.svelte @@ -0,0 +1,122 @@ + + +
+
+

Channels

+ +
+ + {#if showCreate} +
+ {#if createError} +
{createError}
+ {/if} +
+ + + + +
+
+ {/if} + +
+ {#if loadingData} +
+ {#each Array(3) as _} +
+ {/each} +
+ {:else if channelList.length === 0} +
+

No channels yet. Create one to get started.

+
+ {:else} +
+ {#each channelList as ch} + +
+
+
+ #{ch.name} + {#if ch.is_private} + + + + {/if} +
+ {#if ch.description} +

{ch.description}

+ {/if} +
+ + {ch.member_count} members + +
+ + {/each} +
+ {/if} +
+
diff --git a/web/src/routes/channels/[name]/+page.svelte b/web/src/routes/channels/[name]/+page.svelte new file mode 100644 index 0000000..2793c7c --- /dev/null +++ b/web/src/routes/channels/[name]/+page.svelte @@ -0,0 +1,126 @@ + + +
+ + ← Back to channels + + + {#if loadingData} +
+
+
+
+
+ {:else if channel} +
+
+
+
+
+

#{channel.name}

+ {#if channel.is_private} + Private + {/if} + {channel.type} +
+ {#if channel.description} +

{channel.description}

+ {/if} + {#if channel.topic} +

Topic: {channel.topic}

+ {/if} +
+
+ + +
+
+ {#if joinError} +
{joinError}
+ {/if} +
+
+ + +
+
+

Members ({members.length})

+
+ {#if members.length === 0} +
No members
+ {:else} +
+ {#each members as member} +
+ {member.agent_name} + + {member.role} + +
+ {/each} +
+ {/if} +
+ {:else} +
+

Channel not found

+
+ {/if} +
diff --git a/web/src/routes/conversations/+page.svelte b/web/src/routes/conversations/+page.svelte new file mode 100644 index 0000000..0f8db78 --- /dev/null +++ b/web/src/routes/conversations/+page.svelte @@ -0,0 +1,101 @@ + + +
+

Conversations

+ + {#if searchResults !== null} +
+
+

+ Search Results ({searchResults.length}) +

+
+ {#if searchResults.length === 0} +
+ No messages found for '{$page.url.searchParams.get('q')}' +
+ {:else} +
+ {#each searchResults as msg} + +
+ {msg.from_agent} + {new Date(msg.created_at).toLocaleString()} +
+

{msg.body.slice(0, 200)}

+ + {/each} +
+ {/if} +
+ {/if} + +
+
+

All Conversations

+
+ {#if loadingData} +
+ {#each Array(5) as _} +
+
+
+
+ {/each} +
+ {:else if convList.length === 0} +
+

No conversations yet.

+
+ {:else} +
+ {#each convList as conv} + +
+
+

+ {conv.subject || 'Untitled'} +

+

+ {conv.last_agent}: {conv.last_message} +

+
+ + {conv.message_count} + +
+ + {/each} +
+ {/if} +
+
diff --git a/web/src/routes/conversations/[id]/+page.svelte b/web/src/routes/conversations/[id]/+page.svelte new file mode 100644 index 0000000..21bb73e --- /dev/null +++ b/web/src/routes/conversations/[id]/+page.svelte @@ -0,0 +1,130 @@ + + +
+ + ← Back to conversations + + + {#if loadingData} +
+
+
+ {#each Array(3) as _} +
+ {/each} +
+
+ {:else if conversation} +
+
+

+ {conversation.subject || 'Conversation #' + conversation.id} +

+

+ Started by {conversation.created_by} · {messagesList.length} messages +

+
+ + +
+ {#each messagesList as msg (msg.id)} +
+
+
+ {msg.from_agent.charAt(0).toUpperCase()} +
+ {msg.from_agent} + {#if msg.to_agent} + → {msg.to_agent} + {/if} + {msg.status} + + {new Date(msg.created_at).toLocaleString()} + +
+
{msg.body}
+
+ {/each} +
+ + +
+ {#if error} +
{error}
+ {/if} +
+ + +
+
+
+ {:else} +
+

Conversation not found

+
+ {/if} +
diff --git a/web/src/routes/login/+page.svelte b/web/src/routes/login/+page.svelte new file mode 100644 index 0000000..aba2db1 --- /dev/null +++ b/web/src/routes/login/+page.svelte @@ -0,0 +1,76 @@ + + +
+
+
+ + + +

SynapBus

+

Agent-to-agent messaging

+
+ +
+

Sign In

+ + {#if error} +
+ {error} +
+ {/if} + +
+
+ + +
+
+ + +
+ +
+
+
+
diff --git a/web/src/routes/settings/+page.svelte b/web/src/routes/settings/+page.svelte new file mode 100644 index 0000000..336ccfe --- /dev/null +++ b/web/src/routes/settings/+page.svelte @@ -0,0 +1,118 @@ + + +
+

Settings

+ + +
+
+

Appearance

+
+
+
+
+

Dark Mode

+

Toggle between light and dark theme

+
+ +
+
+
+ + +
+
+

Account

+
+
+
+
+

Username

+

{$user?.username ?? '-'}

+
+
+

Display Name

+

{$user?.display_name ?? '-'}

+
+
+
+
+ + +
+
+

Change Password

+
+
+ {#if passwordError} +
{passwordError}
+ {/if} + {#if passwordSuccess} +
{passwordSuccess}
+ {/if} +
+
+ + +
+
+ + +
+
+ + +
+ +
+
+
+
diff --git a/web/static/favicon.svg b/web/static/favicon.svg new file mode 100644 index 0000000..d96f844 --- /dev/null +++ b/web/static/favicon.svg @@ -0,0 +1,3 @@ + + + diff --git a/web/svelte.config.js b/web/svelte.config.js new file mode 100644 index 0000000..c84c060 --- /dev/null +++ b/web/svelte.config.js @@ -0,0 +1,18 @@ +import adapter from '@sveltejs/adapter-static'; + +/** @type {import('@sveltejs/kit').Config} */ +const config = { + kit: { + adapter: adapter({ + pages: 'build', + assets: 'build', + fallback: 'index.html', + precompress: false + }), + paths: { + base: '' + } + } +}; + +export default config; diff --git a/web/tailwind.config.js b/web/tailwind.config.js new file mode 100644 index 0000000..2735e6a --- /dev/null +++ b/web/tailwind.config.js @@ -0,0 +1,24 @@ +/** @type {import('tailwindcss').Config} */ +export default { + content: ['./src/**/*.{html,js,svelte,ts}'], + darkMode: 'class', + theme: { + extend: { + colors: { + primary: { + 50: '#eff6ff', + 100: '#dbeafe', + 200: '#bfdbfe', + 300: '#93c5fd', + 400: '#60a5fa', + 500: '#3b82f6', + 600: '#2563eb', + 700: '#1d4ed8', + 800: '#1e40af', + 900: '#1e3a8a' + } + } + } + }, + plugins: [] +}; diff --git a/web/tsconfig.json b/web/tsconfig.json new file mode 100644 index 0000000..a8f10c8 --- /dev/null +++ b/web/tsconfig.json @@ -0,0 +1,14 @@ +{ + "extends": "./.svelte-kit/tsconfig.json", + "compilerOptions": { + "allowJs": true, + "checkJs": true, + "esModuleInterop": true, + "forceConsistentCasingInFileNames": true, + "resolveJsonModule": true, + "skipLibCheck": true, + "sourceMap": true, + "strict": true, + "moduleResolution": "bundler" + } +} diff --git a/web/vite.config.ts b/web/vite.config.ts new file mode 100644 index 0000000..00889a0 --- /dev/null +++ b/web/vite.config.ts @@ -0,0 +1,13 @@ +import { sveltekit } from '@sveltejs/kit/vite'; +import { defineConfig } from 'vite'; + +export default defineConfig({ + plugins: [sveltekit()], + server: { + proxy: { + '/api': 'http://localhost:8080', + '/auth': 'http://localhost:8080', + '/health': 'http://localhost:8080' + } + } +});