package handlers import ( "context" "database/sql" "encoding/json" "fmt" "log" "math" "net/http" "time" "github.com/gin-gonic/gin" capspkg "chat-switchboard/capabilities" "chat-switchboard/crypto" "chat-switchboard/database" "chat-switchboard/events" "chat-switchboard/models" "chat-switchboard/providers" "chat-switchboard/storage" "chat-switchboard/store" "chat-switchboard/tools" "chat-switchboard/treepath" ) // ── Request / Response types ──────────────── type createMessageRequest struct { Role string `json:"role" binding:"required,oneof=user assistant system"` Content string `json:"content" binding:"required"` Model string `json:"model,omitempty"` } type messageResponse struct { ID string `json:"id"` ChannelID string `json:"channel_id"` Role string `json:"role"` Content string `json:"content"` Model *string `json:"model"` TokensUsed *int `json:"tokens_used"` ParentID *string `json:"parent_id,omitempty"` SiblingCount int `json:"sibling_count"` SiblingIndex int `json:"sibling_index"` ParticipantType *string `json:"participant_type,omitempty"` ParticipantID *string `json:"participant_id,omitempty"` SenderName *string `json:"sender_name,omitempty"` SenderAvatar *string `json:"sender_avatar,omitempty"` CreatedAt string `json:"created_at"` } type editRequest struct { Content string `json:"content" binding:"required"` } type regenerateRequest struct { Model string `json:"model,omitempty"` PersonaID string `json:"persona_id,omitempty"` ProviderConfigID string `json:"provider_config_id,omitempty"` MaxTokens int `json:"max_tokens,omitempty"` Temperature *float64 `json:"temperature,omitempty"` DisabledTools []string `json:"disabled_tools,omitempty"` } type cursorRequest struct { ActiveLeafID string `json:"active_leaf_id" binding:"required"` } // MessageHandler holds dependencies for message endpoints. type MessageHandler struct { vault *crypto.KeyResolver stores store.Stores hub *events.Hub objStore storage.ObjectStore } // NewMessageHandler creates a new message handler. func NewMessageHandler(vault *crypto.KeyResolver, stores store.Stores, hub *events.Hub, objStore storage.ObjectStore) *MessageHandler { return &MessageHandler{vault: vault, stores: stores, hub: hub, objStore: objStore} } // ── List Messages (flat, all branches) ────── // GET /channels/:id/messages func (h *MessageHandler) ListMessages(c *gin.Context) { page, perPage, offset := parsePagination(c) userID := getUserID(c) channelID := c.Param("id") // v0.24.3: Session participants are pre-validated by AuthOrSession middleware if isSessionAuth(c) { if !sessionCanAccessChannel(c, channelID) { c.JSON(http.StatusForbidden, gin.H{"error": "session not authorized for this channel"}) return } } else if !userCanAccessChannel(c, h.stores, channelID, userID) { return } ctx := c.Request.Context() msgs, total, err := h.stores.Messages.ListWithSenderInfo(ctx, channelID, perPage, offset) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list messages"}) return } messages := make([]messageResponse, 0, len(msgs)) for _, m := range msgs { messages = append(messages, messageResponse{ ID: m.ID, ChannelID: m.ChannelID, Role: m.Role, Content: m.Content, Model: m.Model, TokensUsed: m.TokensUsed, ParentID: m.ParentID, SiblingIndex: m.SiblingIndex, ParticipantType: m.ParticipantType, ParticipantID: m.ParticipantID, SenderName: m.SenderName, SenderAvatar: m.SenderAvatar, CreatedAt: m.CreatedAt, }) } // Enrich with sibling counts for i := range messages { messages[i].SiblingCount = getSiblingCount(channelID, messages[i].ParentID) } c.JSON(http.StatusOK, paginatedResponse{ Data: messages, Page: page, PerPage: perPage, Total: total, TotalPages: int(math.Ceil(float64(total) / float64(perPage))), }) } // ── Active Path ───────────────────────────── // GET /channels/:id/path func (h *MessageHandler) GetActivePath(c *gin.Context) { userID := getUserID(c) channelID := c.Param("id") if !userCanAccessChannel(c, h.stores, channelID, userID) { return } path, err := getActivePath(channelID, userID) if err != nil { log.Printf("GetActivePath error: %v", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to load path"}) return } c.JSON(http.StatusOK, gin.H{"data": path}) } // ── Create Message (manual) ───────────────── // POST /channels/:id/messages func (h *MessageHandler) CreateMessage(c *gin.Context) { var req createMessageRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } userID := getUserID(c) channelID := c.Param("id") // v0.24.3: Session participants are pre-validated by AuthOrSession middleware if isSessionAuth(c) { if !sessionCanAccessChannel(c, channelID) { c.JSON(http.StatusForbidden, gin.H{"error": "session not authorized for this channel"}) return } } else if !userCanAccessChannel(c, h.stores, channelID, userID) { return } // Use cursor for parent, compute sibling_index effectiveUserID := userID if isSessionAuth(c) { effectiveUserID = c.GetString("session_id") } parentID, _ := getActiveLeaf(channelID, effectiveUserID) siblingIdx := nextSiblingIndex(channelID, parentID) participantType := "user" participantID := userID if isSessionAuth(c) { participantType = "session" participantID = c.GetString("session_id") } if req.Role == "assistant" { participantType = "model" if req.Model != "" { participantID = req.Model } } msg := &models.Message{ ChannelID: channelID, Role: req.Role, Content: req.Content, Model: req.Model, ParentID: parentID, SiblingIndex: siblingIdx, ParticipantType: participantType, ParticipantID: participantID, ToolCalls: models.JSONMap{}, Metadata: models.JSONMap{}, } if err := h.stores.Messages.CreateWithCursor(c.Request.Context(), msg, userID); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create message"}) return } // Broadcast via WS to channel participants if h.hub != nil && msg.Role == "user" { broadcastUserMessage(c.Request.Context(), h.hub, channelID, msg.ID, msg.Content, userID) } resp := messageResponse{ ID: msg.ID, ChannelID: msg.ChannelID, Role: msg.Role, Content: msg.Content, SiblingIndex: msg.SiblingIndex, CreatedAt: msg.CreatedAt.Format("2006-01-02T15:04:05Z"), } if msg.Model != "" { resp.Model = &msg.Model } resp.ParentID = msg.ParentID resp.SiblingCount = getSiblingCount(channelID, msg.ParentID) c.JSON(http.StatusCreated, resp) } // ── Edit Message (create sibling) ─────────── // POST /channels/:id/messages/:msgId/edit func (h *MessageHandler) EditMessage(c *gin.Context) { var req editRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } userID := getUserID(c) channelID := c.Param("id") messageID := c.Param("msgId") if !userCanAccessChannel(c, h.stores, channelID, userID) { return } ctx := c.Request.Context() // Load target — must exist, belong to channel, be a user message targetParentID, targetRole, err := h.stores.Messages.GetParentAndRole(ctx, messageID, channelID) if err == sql.ErrNoRows { c.JSON(http.StatusNotFound, gin.H{"error": "message not found"}) return } if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to load message"}) return } if targetRole != "user" { c.JSON(http.StatusBadRequest, gin.H{"error": "can only edit user messages"}) return } // Create sibling: same parent_id as the target siblingIdx := nextSiblingIndex(channelID, targetParentID) msg := &models.Message{ ChannelID: channelID, Role: "user", Content: req.Content, ParentID: targetParentID, SiblingIndex: siblingIdx, ParticipantType: "user", ParticipantID: userID, ToolCalls: models.JSONMap{}, Metadata: models.JSONMap{}, } if err := h.stores.Messages.CreateWithCursor(ctx, msg, userID); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create edit"}) return } resp := messageResponse{ ID: msg.ID, ChannelID: msg.ChannelID, Role: msg.Role, Content: msg.Content, ParentID: msg.ParentID, SiblingIndex: msg.SiblingIndex, SiblingCount: getSiblingCount(channelID, msg.ParentID), CreatedAt: msg.CreatedAt.Format("2006-01-02T15:04:05Z"), } c.JSON(http.StatusCreated, resp) } // ── Regenerate / Complete ────────────────────── // POST /channels/:id/messages/:msgId/regenerate func (h *MessageHandler) Regenerate(c *gin.Context) { var req regenerateRequest _ = c.ShouldBindJSON(&req) // body is optional userID := getUserID(c) channelID := c.Param("id") messageID := c.Param("msgId") if !userCanAccessChannel(c, h.stores, channelID, userID) { return } ctx := c.Request.Context() // Load target message targetParentID, targetRole, err := h.stores.Messages.GetParentAndRole(ctx, messageID, channelID) if err == sql.ErrNoRows { c.JSON(http.StatusNotFound, gin.H{"error": "message not found"}) return } if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to load message"}) return } if targetRole != "assistant" && targetRole != "user" { c.JSON(http.StatusBadRequest, gin.H{"error": "can only regenerate user or assistant messages"}) return } // Determine context path and new message's parent based on target role var contextPath []PathMessage var newParentID *string if targetRole == "assistant" { if targetParentID == nil { c.JSON(http.StatusBadRequest, gin.H{"error": "cannot regenerate root message"}) return } contextPath, err = getPathToLeaf(channelID, *targetParentID) newParentID = targetParentID } else { // user message — respond to it contextPath, err = getPathToLeaf(channelID, messageID) newParentID = &messageID } if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to build context"}) return } // ── Resolve model + provider ── comp := NewCompletionHandler(h.vault, h.stores, h.hub, h.objStore, nil) var personaSystemPrompt string var personaID string model := req.Model providerConfigID := req.ProviderConfigID temperature := req.Temperature maxTokens := req.MaxTokens if req.PersonaID != "" { persona := ResolvePersona(h.stores, req.PersonaID, userID) if persona == nil { c.JSON(http.StatusBadRequest, gin.H{"error": "persona not found"}) return } personaID = persona.ID if model == "" { model = persona.BaseModelID } if providerConfigID == "" && persona.ProviderConfigID != nil { providerConfigID = *persona.ProviderConfigID } if temperature == nil && persona.Temperature != nil { temperature = persona.Temperature } if maxTokens == 0 && persona.MaxTokens != nil { maxTokens = *persona.MaxTokens } if persona.SystemPrompt != "" { personaSystemPrompt = persona.SystemPrompt } } // Fallback: channel's stored model if model == "" { channelModel, _ := h.stores.Channels.GetDefaultModel(ctx, channelID) if channelModel != nil { model = *channelModel } } providerCfg, providerID, model, configID, providerScope, err := comp.resolveConfig(userID, channelID, completionRequest{ Model: model, ProviderConfigID: providerConfigID, }) if err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } provider, err := providers.Get(providerID) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } // Build LLM message array llmMessages := make([]providers.Message, 0, len(contextPath)+1) if personaSystemPrompt != "" { llmMessages = append(llmMessages, providers.Message{Role: "system", Content: personaSystemPrompt}) } else { systemPrompt, _ := h.stores.Channels.GetSystemPrompt(ctx, channelID) if systemPrompt != nil && *systemPrompt != "" { llmMessages = append(llmMessages, providers.Message{Role: "system", Content: *systemPrompt}) } } for _, m := range contextPath { if m.Role == "system" { continue } llmMessages = append(llmMessages, providers.Message{Role: m.Role, Content: m.Content}) } provReq := providers.CompletionRequest{ Model: model, Messages: llmMessages, } caps := comp.getModelCapabilities(c, model, configID) if maxTokens > 0 { provReq.MaxTokens = maxTokens } else { provReq.MaxTokens = capspkg.ResolveMaxOutput(model, caps) } if temperature != nil { provReq.Temperature = temperature } // Attach tool definitions (same as normal completion) workspaceID, _ := h.stores.Channels.ResolveWorkspaceID(ctx, channelID) // Query channel metadata for tool context chType, chTeamID, _ := h.stores.Channels.GetTypeAndTeamID(ctx, channelID) msgTeamID := "" if chTeamID != nil { msgTeamID = *chTeamID } hasBrowserTools := h.hub != nil && h.hub.IsConnected(userID) if caps.ToolCalling && (tools.HasTools() || hasBrowserTools) { tctx := tools.ToolContext{ ChannelType: chType, WorkspaceID: workspaceID, TeamID: msgTeamID, PersonaID: personaID, IsVisitor: isSessionAuth(c), } provReq.Tools = comp.buildToolDefs(ctx, userID, hasBrowserTools, req.DisabledTools, tctx, personaID) } // ── Stream the response ── if hooks := providers.GetHooks(providerID); hooks != nil { hooks.PreRequest(providerCfg, &provReq) } result := streamWithToolLoop(c, provider, providerCfg, &provReq, model, providerID, userID, channelID, personaID, workspaceID, configID, msgTeamID, h.hub, comp.health, comp.runner, comp.buildExtToolMap(c.Request.Context(), userID)) // Persist as sibling (regen) with tool activity if result.Content != "" { siblingIdx := nextSiblingIndex(channelID, newParentID) var toolCalls models.JSONMap if len(result.ToolActivity) > 0 { b, _ := json.Marshal(result.ToolActivity) _ = json.Unmarshal(b, &toolCalls) } if toolCalls == nil { toolCalls = models.JSONMap{} } msg := &models.Message{ ChannelID: channelID, Role: "assistant", Content: result.Content, Model: model, ToolCalls: toolCalls, Metadata: models.JSONMap{}, ParentID: newParentID, SiblingIndex: siblingIdx, ParticipantType: "model", ParticipantID: model, } if err := h.stores.Messages.CreateWithCursor(ctx, msg, userID); err != nil { log.Printf("Failed to persist regenerated message: %v", err) } else { flusher, _ := c.Writer.(http.Flusher) msgJSON, _ := json.Marshal(gin.H{ "id": msg.ID, "parent_id": newParentID, "sibling_index": siblingIdx, "sibling_count": getSiblingCount(channelID, newParentID), }) fmt.Fprintf(c.Writer, "data: {\"switchboard_meta\":%s}\n\n", msgJSON) if flusher != nil { flusher.Flush() } } } // Log usage for regeneration comp.logUsage(c, channelID, userID, configID, providerScope, model, result.InputTokens, result.OutputTokens, result.CacheCreationTokens, result.CacheReadTokens) } // ── Switch Branch (update cursor) ─────────── // PUT /channels/:id/cursor func (h *MessageHandler) UpdateCursor(c *gin.Context) { var req cursorRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } userID := getUserID(c) channelID := c.Param("id") if !userCanAccessChannel(c, h.stores, channelID, userID) { return } ctx := c.Request.Context() // Verify the target message belongs to this channel and is not deleted. // GetParentAndRole does: WHERE id=$1 AND channel_id=$2 AND deleted_at IS NULL _, _, err := h.stores.Messages.GetParentAndRole(ctx, req.ActiveLeafID, channelID) if err == sql.ErrNoRows { c.JSON(http.StatusNotFound, gin.H{"error": "message not found"}) return } if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to verify message"}) return } // Walk down to the leaf from the target leafID, err := findLeafFromMessage(req.ActiveLeafID) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to find leaf"}) return } if err := updateCursor(channelID, userID, leafID); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to update cursor"}) return } // Return the new active path path, err := getPathToLeaf(channelID, leafID) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to load path"}) return } c.JSON(http.StatusOK, gin.H{"data": path, "active_leaf_id": leafID}) } // ── List Siblings ─────────────────────────── // GET /channels/:id/messages/:msgId/siblings func (h *MessageHandler) ListSiblings(c *gin.Context) { userID := getUserID(c) channelID := c.Param("id") messageID := c.Param("msgId") if !userCanAccessChannel(c, h.stores, channelID, userID) { return } siblings, currentIdx, err := getSiblings(messageID) if err != nil { c.JSON(http.StatusNotFound, gin.H{"error": "message not found"}) return } c.JSON(http.StatusOK, gin.H{ "siblings": siblings, "current_index": currentIdx, "total": len(siblings), }) } // ── Ownership / Access Check ───────────────── // userCanAccessChannel verifies channel ownership or participation, // writing an error response if denied. Returns true if access is allowed. func userCanAccessChannel(c *gin.Context, stores store.Stores, channelID, userID string) bool { ok, err := stores.Channels.UserCanAccess(c.Request.Context(), channelID, userID) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to verify channel access"}) return false } if !ok { c.JSON(http.StatusForbidden, gin.H{"error": "not your channel"}) return false } return true } // userOwnsChannel is the legacy wrapper used by other handlers (completion.go). // Delegates to userCanAccessChannel using the treepath global stores. func userOwnsChannel(c *gin.Context, channelID, userID string) bool { ok, err := treepath.Stores.Channels.UserCanAccess(c.Request.Context(), channelID, userID) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to verify channel access"}) return false } if !ok { c.JSON(http.StatusForbidden, gin.H{"error": "not your channel"}) return false } return true } // broadcastUserMessage publishes a message.created event to all user // participants in the channel via WebSocket. func broadcastUserMessage(ctx context.Context, hub *events.Hub, channelID, msgID, content, senderID string) { // Look up sender display name var displayName, username string _ = database.DB.QueryRowContext(ctx, database.Q(` SELECT COALESCE(display_name, ''), COALESCE(username, '') FROM users WHERE id = $1 `), senderID).Scan(&displayName, &username) payload, _ := json.Marshal(map[string]any{ "id": msgID, "channel_id": channelID, "role": "user", "content": content, "user_id": senderID, "display_name": displayName, "username": username, "created_at": time.Now().UTC().Format("2006-01-02T15:04:05Z"), }) evt := events.Event{ Label: "message.created", Payload: payload, Ts: time.Now().UnixMilli(), } // Send to all user participants in the channel rows, err := database.DB.QueryContext(ctx, database.Q(` SELECT participant_id FROM channel_participants WHERE channel_id = $1 AND participant_type = 'user' `), channelID) if err != nil { // Fallback: at least send to the sender hub.PublishToUser(senderID, evt) return } defer rows.Close() for rows.Next() { var pid string if rows.Scan(&pid) == nil { hub.PublishToUser(pid, evt) } } } // broadcastAssistantMessage publishes a message.created event for an assistant // message to all user participants in the channel (except the requesting user, // who already received the response via SSE). func broadcastAssistantMessage(ctx context.Context, hub *events.Hub, channelID, msgID, content, model, senderUserID, personaID string) { // Look up persona display name if available var displayName, avatar string if personaID != "" { _ = database.DB.QueryRowContext(ctx, database.Q(` SELECT COALESCE(name, ''), COALESCE(avatar, '') FROM personas WHERE id = $1 `), personaID).Scan(&displayName, &avatar) } if displayName == "" { displayName = model } payload, _ := json.Marshal(map[string]any{ "id": msgID, "channel_id": channelID, "role": "assistant", "content": content, "model": model, "participant_type": "persona", "participant_id": personaID, "display_name": displayName, "avatar": avatar, "created_at": time.Now().UTC().Format("2006-01-02T15:04:05Z"), }) evt := events.Event{ Label: "message.created", Payload: payload, Ts: time.Now().UnixMilli(), } // Send to all user participants except the sender (who got SSE) rows, err := database.DB.QueryContext(ctx, database.Q(` SELECT participant_id FROM channel_participants WHERE channel_id = $1 AND participant_type = 'user' `), channelID) if err != nil { return } defer rows.Close() for rows.Next() { var pid string if rows.Scan(&pid) == nil && pid != senderUserID { hub.PublishToUser(pid, evt) } } } // ── Delete Message ────────────────────────────────────────── // DeleteMessage soft-deletes a message. // DELETE /channels/:id/messages/:msgId func (h *MessageHandler) DeleteMessage(c *gin.Context) { userID := getUserID(c) channelID := c.Param("id") msgID := c.Param("msgId") if !userCanAccessChannel(c, h.stores, channelID, userID) { return } // Verify message belongs to this channel and is not already deleted var exists bool _ = database.DB.QueryRowContext(c.Request.Context(), database.Q(` SELECT EXISTS(SELECT 1 FROM messages WHERE id = $1 AND channel_id = $2 AND deleted_at IS NULL) `), msgID, channelID).Scan(&exists) if !exists { c.JSON(http.StatusNotFound, gin.H{"error": "message not found"}) return } if err := h.stores.Messages.Delete(c.Request.Context(), msgID); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete message"}) return } // Broadcast to channel participants (exclude sender — they already removed it) broadcastMessageDeleted(c.Request.Context(), h.hub, channelID, msgID, userID) c.JSON(http.StatusOK, gin.H{"ok": true}) } // broadcastMessageDeleted publishes a message.deleted event to all user // participants in the channel except the sender. func broadcastMessageDeleted(ctx context.Context, hub *events.Hub, channelID, msgID, senderID string) { payload, _ := json.Marshal(map[string]any{ "id": msgID, "channel_id": channelID, }) evt := events.Event{ Label: "message.deleted", Payload: payload, Ts: time.Now().UnixMilli(), } rows, err := database.DB.QueryContext(ctx, database.Q(` SELECT participant_id FROM channel_participants WHERE channel_id = $1 AND participant_type = 'user' `), channelID) if err != nil { return } defer rows.Close() for rows.Next() { var pid string if rows.Scan(&pid) == nil && pid != senderID { hub.PublishToUser(pid, evt) } } }