package handlers import ( "database/sql" "encoding/json" "fmt" "log" "math" "net/http" "github.com/gin-gonic/gin" capspkg "git.gobha.me/xcaliber/chat-switchboard/capabilities" "git.gobha.me/xcaliber/chat-switchboard/crypto" "git.gobha.me/xcaliber/chat-switchboard/database" "git.gobha.me/xcaliber/chat-switchboard/events" "git.gobha.me/xcaliber/chat-switchboard/providers" "git.gobha.me/xcaliber/chat-switchboard/storage" "git.gobha.me/xcaliber/chat-switchboard/store" "git.gobha.me/xcaliber/chat-switchboard/tools" ) // ── 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 // // Returns all live messages across all branches (useful for admin/debug). // Frontend should prefer /channels/:id/path for rendering. 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 !userOwnsChannel(c, channelID, userID) { return } var total int err := database.DB.QueryRow( database.Q(`SELECT COUNT(*) FROM messages WHERE channel_id = $1 AND deleted_at IS NULL`), channelID, ).Scan(&total) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to count messages"}) return } rows, err := database.DB.Query(database.Q(` SELECT m.id, m.channel_id, m.role, m.content, m.model, m.tokens_used, m.parent_id, m.sibling_index, m.participant_type, m.participant_id, CASE WHEN m.participant_type = 'user' THEN COALESCE(u.display_name, u.username) WHEN m.participant_type = 'persona' THEN p.name ELSE NULL END AS sender_name, CASE WHEN m.participant_type = 'user' THEN u.avatar_url WHEN m.participant_type = 'persona' THEN p.avatar ELSE NULL END AS sender_avatar, m.created_at FROM messages m LEFT JOIN users u ON m.participant_type = 'user' AND m.participant_id = u.id::text LEFT JOIN personas p ON m.participant_type = 'persona' AND m.participant_id = p.id::text WHERE m.channel_id = $1 AND m.deleted_at IS NULL ORDER BY m.created_at ASC LIMIT $2 OFFSET $3 `), channelID, perPage, offset) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list messages"}) return } defer rows.Close() messages := make([]messageResponse, 0) for rows.Next() { var msg messageResponse err := rows.Scan( &msg.ID, &msg.ChannelID, &msg.Role, &msg.Content, &msg.Model, &msg.TokensUsed, &msg.ParentID, &msg.SiblingIndex, &msg.ParticipantType, &msg.ParticipantID, &msg.SenderName, &msg.SenderAvatar, &msg.CreatedAt, ) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to scan message"}) return } messages = append(messages, msg) } rows.Close() // release connection before sibling queries // Enrich with sibling counts (requires DB access, must happen after rows are closed) 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 // // Returns the active branch from root → leaf with sibling metadata // at each node. This is what the frontend renders. func (h *MessageHandler) GetActivePath(c *gin.Context) { userID := getUserID(c) channelID := c.Param("id") if !userOwnsChannel(c, 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{"path": 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 !userOwnsChannel(c, channelID, userID) { return } // Use cursor for parent, compute sibling_index // Sessions use the same cursor logic (empty userID = channel-level latest) 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 } } var msg messageResponse if database.IsSQLite() { newID := store.NewID() _, err := database.DB.Exec(` INSERT INTO messages (id, channel_id, role, content, model, parent_id, participant_type, participant_id, sibling_index) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) `, newID, channelID, req.Role, req.Content, req.Model, parentID, participantType, participantID, siblingIdx) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create message"}) return } err = database.DB.QueryRow(` SELECT id, channel_id, role, content, model, tokens_used, parent_id, sibling_index, created_at FROM messages WHERE id = ?`, newID).Scan( &msg.ID, &msg.ChannelID, &msg.Role, &msg.Content, &msg.Model, &msg.TokensUsed, &msg.ParentID, &msg.SiblingIndex, &msg.CreatedAt, ) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to read created message"}) return } } else { err := database.DB.QueryRow(` INSERT INTO messages (channel_id, role, content, model, parent_id, participant_type, participant_id, sibling_index) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING id, channel_id, role, content, model, tokens_used, parent_id, sibling_index, created_at `, channelID, req.Role, req.Content, req.Model, parentID, participantType, participantID, siblingIdx, ).Scan( &msg.ID, &msg.ChannelID, &msg.Role, &msg.Content, &msg.Model, &msg.TokensUsed, &msg.ParentID, &msg.SiblingIndex, &msg.CreatedAt, ) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create message"}) return } } msg.SiblingCount = getSiblingCount(channelID, msg.ParentID) _ = updateCursor(channelID, userID, msg.ID) _, _ = database.DB.Exec(database.Q(`UPDATE channels SET updated_at = NOW() WHERE id = $1`), channelID) c.JSON(http.StatusCreated, msg) } // ── Edit Message (create sibling) ─────────── // POST /channels/:id/messages/:msgId/edit // // Creates a new user message as a sibling of the target (same parent_id). // Does NOT auto-trigger completion — the frontend sends a separate request. 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 !userOwnsChannel(c, channelID, userID) { return } // Load target — must exist, belong to channel, be a user message var targetParentID *string var targetRole string err := database.DB.QueryRow(database.Q(` SELECT parent_id, role FROM messages WHERE id = $1 AND channel_id = $2 AND deleted_at IS NULL `), messageID, channelID).Scan(&targetParentID, &targetRole) 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) var msg messageResponse if database.IsSQLite() { newID := store.NewID() _, err = database.DB.Exec(` INSERT INTO messages (id, channel_id, role, content, parent_id, participant_type, participant_id, sibling_index) VALUES (?, ?, 'user', ?, ?, 'user', ?, ?) `, newID, channelID, req.Content, targetParentID, userID, siblingIdx) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create edit"}) return } err = database.DB.QueryRow(` SELECT id, channel_id, role, content, model, tokens_used, parent_id, sibling_index, created_at FROM messages WHERE id = ?`, newID).Scan( &msg.ID, &msg.ChannelID, &msg.Role, &msg.Content, &msg.Model, &msg.TokensUsed, &msg.ParentID, &msg.SiblingIndex, &msg.CreatedAt, ) } else { err = database.DB.QueryRow(` INSERT INTO messages (channel_id, role, content, parent_id, participant_type, participant_id, sibling_index) VALUES ($1, 'user', $2, $3, 'user', $4, $5) RETURNING id, channel_id, role, content, model, tokens_used, parent_id, sibling_index, created_at `, channelID, req.Content, targetParentID, userID, siblingIdx, ).Scan( &msg.ID, &msg.ChannelID, &msg.Role, &msg.Content, &msg.Model, &msg.TokensUsed, &msg.ParentID, &msg.SiblingIndex, &msg.CreatedAt, ) } if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create edit"}) return } msg.SiblingCount = getSiblingCount(channelID, msg.ParentID) // Cursor now points to the new sibling (it's a leaf — no children yet) _ = updateCursor(channelID, userID, msg.ID) _, _ = database.DB.Exec(database.Q(`UPDATE channels SET updated_at = NOW() WHERE id = $1`), channelID) c.JSON(http.StatusCreated, msg) } // ── Regenerate / Complete ────────────────────── // POST /channels/:id/messages/:msgId/regenerate // // Two modes depending on target role: // - assistant message → creates a NEW assistant sibling (same parent_id). // Context is root → target's parent. ("Give me a different response.") // - user message → creates a child assistant response. // Context is root → target. ("Respond to this message.") // // The second mode is used after editing: the edit endpoint creates a // sibling user message, then the frontend calls regenerate on it to // get an assistant response without duplicating the user message. 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 !userOwnsChannel(c, channelID, userID) { return } // Load target message var targetParentID *string var targetRole string err := database.DB.QueryRow(database.Q(` SELECT parent_id, role FROM messages WHERE id = $1 AND channel_id = $2 AND deleted_at IS NULL `), messageID, channelID).Scan(&targetParentID, &targetRole) 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: // // assistant target: context = root → target's parent (exclude target) // new parent = target's parent (sibling of target) // // user target: context = root → target (include target) // new parent = target itself (child of target) 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 == "" { var channelModel *string _ = database.DB.QueryRow(database.Q(`SELECT model FROM channels WHERE id = $1`), channelID).Scan(&channelModel) 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 { var systemPrompt *string _ = database.DB.QueryRow(database.Q(`SELECT system_prompt FROM channels WHERE id = $1`), channelID).Scan(&systemPrompt) 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(c.Request.Context(), channelID) hasBrowserTools := h.hub != nil && h.hub.IsConnected(userID) if caps.ToolCalling && (tools.HasTools() || hasBrowserTools) { // v0.25.0: Build ToolContext with channel metadata var chType string var chTeamID *string _ = database.DB.QueryRowContext(c.Request.Context(), database.Q(` SELECT COALESCE(type, 'direct'), team_id FROM channels WHERE id = $1 `), channelID).Scan(&chType, &chTeamID) msgTeamID := "" if chTeamID != nil { msgTeamID = *chTeamID } tctx := tools.ToolContext{ ChannelType: chType, WorkspaceID: workspaceID, TeamID: msgTeamID, PersonaID: personaID, IsVisitor: isSessionAuth(c), } provReq.Tools = comp.buildToolDefs(c.Request.Context(), userID, hasBrowserTools, req.DisabledTools, tctx, personaID) } // ── Stream the response (shared loop handles tools, reasoning, SSE) ── // Apply provider-specific request hooks (v0.22.1) if hooks := providers.GetHooks(providerID); hooks != nil { hooks.PreRequest(providerCfg, &provReq) } result := streamWithToolLoop(c, provider, providerCfg, &provReq, model, providerID, userID, channelID, personaID, workspaceID, configID, h.hub, comp.health) // Persist as sibling (regen) with tool activity if result.Content != "" { siblingIdx := nextSiblingIndex(channelID, newParentID) // Match persistMessage pattern: nil interface{} → SQL NULL, // non-nil string → valid JSONB. A nil json.RawMessage ([]byte) // is sent by pq as empty bytes, not NULL, causing "invalid input // syntax for type json". var tcVal interface{} if len(result.ToolActivity) > 0 { b, _ := json.Marshal(result.ToolActivity) tcVal = string(b) } var newID string if database.IsSQLite() { newID = store.NewID() _, err := database.DB.Exec(` INSERT INTO messages (id, channel_id, role, content, model, tool_calls, parent_id, participant_type, participant_id, sibling_index) VALUES (?, ?, 'assistant', ?, ?, ?, ?, 'model', ?, ?) `, newID, channelID, result.Content, model, tcVal, newParentID, model, siblingIdx) if err != nil { log.Printf("Failed to persist regenerated message: %v", err) newID = "" } } else { err := database.DB.QueryRow(` INSERT INTO messages (channel_id, role, content, model, tool_calls, parent_id, participant_type, participant_id, sibling_index) VALUES ($1, 'assistant', $2, $3, $4, $5, 'model', $6, $7) RETURNING id `, channelID, result.Content, model, tcVal, newParentID, model, siblingIdx).Scan(&newID) if err != nil { log.Printf("Failed to persist regenerated message: %v", err) newID = "" } } if newID != "" { _ = updateCursor(channelID, userID, newID) flusher, _ := c.Writer.(http.Flusher) msgJSON, _ := json.Marshal(gin.H{ "id": newID, "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() } } _, _ = database.DB.Exec(database.Q(`UPDATE channels SET updated_at = NOW() WHERE id = $1`), channelID) } // 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 // // Navigates to a different branch. If the target message has children, // follows the first-child chain to find the leaf. 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 !userOwnsChannel(c, channelID, userID) { return } // Verify the target message belongs to this channel var msgChannelID string err := database.DB.QueryRow(database.Q(` SELECT channel_id FROM messages WHERE id = $1 AND deleted_at IS NULL `), req.ActiveLeafID).Scan(&msgChannelID) 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 } if msgChannelID != channelID { c.JSON(http.StatusForbidden, gin.H{"error": "message does not belong to this channel"}) 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{"path": path, "active_leaf_id": leafID}) } // ── List Siblings ─────────────────────────── // GET /channels/:id/messages/:msgId/siblings // // Returns all siblings of a message (messages sharing the same parent_id). // Used by the branch navigator to populate the arrow navigation. func (h *MessageHandler) ListSiblings(c *gin.Context) { userID := getUserID(c) channelID := c.Param("id") messageID := c.Param("msgId") if !userOwnsChannel(c, 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 Check ───────────────────────── func userOwnsChannel(c *gin.Context, channelID, userID string) bool { var ownerID string err := database.DB.QueryRow( database.Q(`SELECT user_id FROM channels WHERE id = $1`), channelID, ).Scan(&ownerID) if err == sql.ErrNoRows { c.JSON(http.StatusNotFound, gin.H{"error": "channel not found"}) return false } if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to verify channel ownership"}) return false } if ownerID == userID { return true } // v0.23.2: Also allow if user is a participant (multi-user DMs, channels) var exists bool _ = database.DB.QueryRow( database.Q(`SELECT EXISTS(SELECT 1 FROM channel_participants WHERE channel_id = $1 AND participant_type = 'user' AND participant_id = $2)`), channelID, userID, ).Scan(&exists) if exists { return true } c.JSON(http.StatusForbidden, gin.H{"error": "not your channel"}) return false }