package v1 import ( "fmt" "io" "net/http" "strconv" "strings" "github.com/gin-gonic/gin" "github.com/gochat/gochat/internal/llm" "github.com/gochat/gochat/internal/service" applogger "github.com/gochat/gochat/pkg/logger" "github.com/gochat/gochat/pkg/response" ) // CaptainTaskHandler handles Captain standalone AI task endpoints. // Reference: Chatwoot enterprise/app/controllers/api/v1/accounts/captain/tasks_controller.rb // These endpoints provide one-shot AI tasks (reply_suggestion, summarize, rewrite) // independent of the CopilotThread flow. type CaptainTaskHandler struct { svc *service.CaptainTaskService } // NewCaptainTaskHandler creates a new CaptainTaskHandler. func NewCaptainTaskHandler(svc *service.CaptainTaskService) *CaptainTaskHandler { return &CaptainTaskHandler{svc: svc} } // ReplySuggestion generates AI-powered reply suggestions for a conversation. // POST /api/v1/accounts/:account_id/captain/tasks/reply_suggestion func (h *CaptainTaskHandler) ReplySuggestion(c *gin.Context) { accountID, err := strconv.ParseUint(c.Param("id"), 10, 64) if err != nil { response.AbortWithStatusError(c, http.StatusBadRequest, response.ErrBadRequest, "invalid account_id") return } var req service.TaskReplySuggestionRequest if err := c.ShouldBindJSON(&req); err != nil { response.AbortWithStatusError(c, http.StatusBadRequest, response.ErrBadRequest, "invalid request body: "+err.Error()) return } result, err := h.svc.ReplySuggestion(c.Request.Context(), uint(accountID), &req) if err != nil { applogger.L().Errorf("ReplySuggestion: %v", err) renderCaptainTaskError(c, err) return } renderCaptainTaskPayload(c, result.Message, result.FollowUpContext) } // Summarize generates a concise summary of a conversation. // POST /api/v1/accounts/:account_id/captain/tasks/summarize func (h *CaptainTaskHandler) Summarize(c *gin.Context) { accountID, err := strconv.ParseUint(c.Param("id"), 10, 64) if err != nil { response.AbortWithStatusError(c, http.StatusBadRequest, response.ErrBadRequest, "invalid account_id") return } var req service.TaskSummarizeRequest if err := c.ShouldBindJSON(&req); err != nil { response.AbortWithStatusError(c, http.StatusBadRequest, response.ErrBadRequest, "invalid request body: "+err.Error()) return } result, err := h.svc.Summarize(c.Request.Context(), uint(accountID), &req) if err != nil { applogger.L().Errorf("Summarize: %v", err) renderCaptainTaskError(c, err) return } renderCaptainTaskPayload(c, result.Message, result.FollowUpContext) } // Rewrite rewrites a draft message to improve tone, clarity, or language. // POST /api/v1/accounts/:account_id/captain/tasks/rewrite func (h *CaptainTaskHandler) Rewrite(c *gin.Context) { accountID, err := strconv.ParseUint(c.Param("id"), 10, 64) if err != nil { response.AbortWithStatusError(c, http.StatusBadRequest, response.ErrBadRequest, "invalid account_id") return } var req service.TaskRewriteRequest if err := c.ShouldBindJSON(&req); err != nil { response.AbortWithStatusError(c, http.StatusBadRequest, response.ErrBadRequest, "invalid request body: "+err.Error()) return } result, err := h.svc.Rewrite(c.Request.Context(), uint(accountID), &req) if err != nil { applogger.L().Errorf("Rewrite: %v", err) renderCaptainTaskError(c, err) return } renderCaptainTaskPayload(c, result.Message, result.FollowUpContext) } // --- SSE Streaming Endpoints (M12) --- // These endpoints stream Captain task results via Server-Sent Events, // following the same SSE pattern as SSEStreamHandler.StreamCopilotMessage. // StreamReplySuggestion streams AI-powered reply suggestions via SSE. // POST /api/v1/accounts/:id/captain/tasks/reply_suggestion/stream func (h *CaptainTaskHandler) StreamReplySuggestion(c *gin.Context) { accountID, err := strconv.ParseUint(c.Param("id"), 10, 64) if err != nil { captainWriteSSEError(c, "invalid account_id") return } var req service.TaskReplySuggestionRequest if err := c.ShouldBindJSON(&req); err != nil { captainWriteSSEError(c, "invalid request body: "+err.Error()) return } captainSetSSEHeaders(c) err = h.svc.ReplySuggestionStream(c.Request.Context(), uint(accountID), &req, func(chunk llm.StreamChunk) error { if len(chunk.Choices) > 0 { content := chunk.Choices[0].Delta.Content if content != "" { escaped := captainEscapeJSONString(content) captainWriteSSEMessage(c, "message", fmt.Sprintf(`{"content": "%s", "done": false}`, escaped)) c.Writer.Flush() } finishReason := chunk.Choices[0].FinishReason if finishReason != "" && finishReason != "null" { captainWriteSSEMessage(c, "done", `{"done": true, "content": ""}`) c.Writer.Flush() return io.EOF } } return nil }) if err != nil && err != io.EOF { applogger.L().Errorf("StreamReplySuggestion: %v", err) captainWriteSSEMessage(c, "error", fmt.Sprintf(`{"error": "%s"}`, captainEscapeJSONString(err.Error()))) captainWriteSSEMessage(c, "done", `{"done": true}`) } } // StreamSummarize streams a conversation summary via SSE. // POST /api/v1/accounts/:id/captain/tasks/summarize/stream func (h *CaptainTaskHandler) StreamSummarize(c *gin.Context) { accountID, err := strconv.ParseUint(c.Param("id"), 10, 64) if err != nil { captainWriteSSEError(c, "invalid account_id") return } var req service.TaskSummarizeRequest if err := c.ShouldBindJSON(&req); err != nil { captainWriteSSEError(c, "invalid request body: "+err.Error()) return } captainSetSSEHeaders(c) err = h.svc.SummarizeStream(c.Request.Context(), uint(accountID), &req, func(chunk llm.StreamChunk) error { if len(chunk.Choices) > 0 { content := chunk.Choices[0].Delta.Content if content != "" { escaped := captainEscapeJSONString(content) captainWriteSSEMessage(c, "message", fmt.Sprintf(`{"content": "%s", "done": false}`, escaped)) c.Writer.Flush() } finishReason := chunk.Choices[0].FinishReason if finishReason != "" && finishReason != "null" { captainWriteSSEMessage(c, "done", `{"done": true, "content": ""}`) c.Writer.Flush() return io.EOF } } return nil }) if err != nil && err != io.EOF { applogger.L().Errorf("StreamSummarize: %v", err) captainWriteSSEMessage(c, "error", fmt.Sprintf(`{"error": "%s"}`, captainEscapeJSONString(err.Error()))) captainWriteSSEMessage(c, "done", `{"done": true}`) } } // StreamRewrite streams a rewritten message via SSE. // POST /api/v1/accounts/:id/captain/tasks/rewrite/stream func (h *CaptainTaskHandler) StreamRewrite(c *gin.Context) { accountID, err := strconv.ParseUint(c.Param("id"), 10, 64) if err != nil { captainWriteSSEError(c, "invalid account_id") return } var req service.TaskRewriteRequest if err := c.ShouldBindJSON(&req); err != nil { captainWriteSSEError(c, "invalid request body: "+err.Error()) return } captainSetSSEHeaders(c) err = h.svc.RewriteStream(c.Request.Context(), uint(accountID), &req, func(chunk llm.StreamChunk) error { if len(chunk.Choices) > 0 { content := chunk.Choices[0].Delta.Content if content != "" { escaped := captainEscapeJSONString(content) captainWriteSSEMessage(c, "message", fmt.Sprintf(`{"content": "%s", "done": false}`, escaped)) c.Writer.Flush() } finishReason := chunk.Choices[0].FinishReason if finishReason != "" && finishReason != "null" { captainWriteSSEMessage(c, "done", `{"done": true, "content": ""}`) c.Writer.Flush() return io.EOF } } return nil }) if err != nil && err != io.EOF { applogger.L().Errorf("StreamRewrite: %v", err) captainWriteSSEMessage(c, "error", fmt.Sprintf(`{"error": "%s"}`, captainEscapeJSONString(err.Error()))) captainWriteSSEMessage(c, "done", `{"done": true}`) } } // --- Captain SSE Helper Functions --- // These are Captain-specific SSE helpers to avoid naming conflicts // with SSEStreamHandler helpers in the same v1 package. // captainSetSSEHeaders sets standard SSE response headers. func captainSetSSEHeaders(c *gin.Context) { c.Header("Content-Type", "text/event-stream") c.Header("Cache-Control", "no-cache") c.Header("Connection", "keep-alive") c.Header("X-Accel-Buffering", "no") } // captainWriteSSEMessage writes a single SSE event to the Gin response writer. func captainWriteSSEMessage(c *gin.Context, event string, data string) { c.Writer.WriteString(fmt.Sprintf("event: %s\ndata: %s\n\n", event, data)) } // captainWriteSSEError writes an error SSE event followed by a done event. func captainWriteSSEError(c *gin.Context, errMsg string) { captainSetSSEHeaders(c) captainWriteSSEMessage(c, "error", fmt.Sprintf(`{"error": "%s"}`, captainEscapeJSONString(errMsg))) captainWriteSSEMessage(c, "done", `{"done": true}`) } // captainEscapeJSONString escapes special characters for safe JSON embedding. func captainEscapeJSONString(s string) string { var result strings.Builder for _, ch := range s { switch ch { case '"': result.WriteString(`\"`) case '\\': result.WriteString(`\\`) case '\n': result.WriteString(`\n`) case '\r': result.WriteString(`\r`) case '\t': result.WriteString(`\t`) default: result.WriteRune(ch) } } return result.String() } func renderCaptainTaskPayload(c *gin.Context, message string, followUpContext map[string]interface{}) { payload := gin.H{"message": message} if followUpContext != nil { payload["follow_up_context"] = followUpContext } c.JSON(http.StatusOK, payload) } func renderCaptainTaskError(c *gin.Context, err error) { status, message, ok := service.CaptainTaskErrorStatus(err) if !ok { status = http.StatusUnprocessableEntity message = err.Error() } c.JSON(status, gin.H{"error": message}) }