feat: task comments SSR fetch

This commit is contained in:
Carlos Florêncio
2026-01-14 20:52:22 +00:00
parent 5813bd8f14
commit e965524590
12 changed files with 184 additions and 37 deletions
+9 -1
View File
@@ -379,6 +379,7 @@ func main() {
if worktreeMgr != nil {
taskController.SetWorktreeLookup(worktreeMgr) // Enable worktree info in task responses
}
commentController := taskcontroller.NewCommentController(taskSvc)
repositoryController := taskcontroller.NewRepositoryController(taskSvc)
executorController := taskcontroller.NewExecutorController(taskSvc)
environmentController := taskcontroller.NewEnvironmentController(taskSvc)
@@ -658,7 +659,14 @@ func main() {
taskhandlers.RegisterRepositoryRoutes(router, gateway.Dispatcher, repositoryController, log)
taskhandlers.RegisterExecutorRoutes(router, gateway.Dispatcher, executorController, log)
taskhandlers.RegisterEnvironmentRoutes(router, gateway.Dispatcher, environmentController, log)
taskhandlers.RegisterCommentRoutes(router, gateway.Dispatcher, taskSvc, &orchestratorAdapter{svc: orchestratorSvc}, log)
taskhandlers.RegisterCommentRoutes(
router,
gateway.Dispatcher,
commentController,
taskController,
&orchestratorAdapter{svc: orchestratorSvc},
log,
)
log.Info("Registered Task Service handlers (HTTP + WebSocket)")
agentsettingshandlers.RegisterRoutes(router, agentSettingsController, gateway.Hub, log)
@@ -36,6 +36,7 @@ type Client struct {
subscriptions map[string]bool // Task IDs this client is subscribed to
userSubscriptions map[string]bool // User IDs this client is subscribed to
mu sync.RWMutex
closed bool
logger *logger.Logger
}
@@ -262,11 +263,22 @@ func (c *Client) sendMessage(msg *ws.Message) {
c.logger.Error("Failed to marshal message", zap.Error(err))
return
}
c.sendBytes(data)
}
func (c *Client) sendBytes(data []byte) bool {
c.mu.Lock()
defer c.mu.Unlock()
if c.closed {
return false
}
select {
case c.send <- data:
return true
default:
c.logger.Warn("Client send buffer full")
return false
}
}
@@ -280,6 +292,16 @@ func (c *Client) sendError(id, action, code, message string, details map[string]
c.sendMessage(msg)
}
func (c *Client) closeSend() {
c.mu.Lock()
defer c.mu.Unlock()
if c.closed {
return
}
c.closed = true
close(c.send)
}
// WritePump pumps messages from the hub to the WebSocket connection
func (c *Client) WritePump() {
ticker := time.NewTicker(pingPeriod)
+13 -17
View File
@@ -87,7 +87,7 @@ func (h *Hub) closeAllClients() {
defer h.mu.Unlock()
for client := range h.clients {
close(client.send)
client.closeSend()
delete(h.clients, client)
}
h.taskSubscribers = make(map[string]map[*Client]bool)
@@ -100,7 +100,7 @@ func (h *Hub) removeClient(client *Client) {
if _, ok := h.clients[client]; ok {
delete(h.clients, client)
close(client.send)
client.closeSend()
// Remove from all task subscriptions
for taskID := range client.subscriptions {
@@ -137,11 +137,7 @@ func (h *Hub) broadcastMessage(msg *ws.Message) {
// For now, broadcast to all clients
// TODO: Add topic-based routing for task-specific notifications
for client := range h.clients {
select {
case client.send <- data:
default:
// Client buffer full, will be cleaned up by write pump
}
client.sendBytes(data)
}
}
@@ -179,14 +175,14 @@ func (h *Hub) BroadcastToTask(taskID string, msg *ws.Message) {
zap.Int("subscriber_count", numClients))
for client := range clients {
select {
case client.send <- data:
if client.sendBytes(data) {
h.logger.Debug("Sent message to client",
zap.String("client_id", client.ID),
zap.String("action", msg.Action))
default:
h.logger.Warn("Client buffer full, dropping message",
zap.String("client_id", client.ID))
} else {
h.logger.Warn("Client send buffer full, dropping message",
zap.String("client_id", client.ID),
zap.String("action", msg.Action))
}
}
}
@@ -210,14 +206,14 @@ func (h *Hub) BroadcastToUser(userID string, msg *ws.Message) {
zap.Int("subscriber_count", numClients))
for client := range clients {
select {
case client.send <- data:
if client.sendBytes(data) {
h.logger.Debug("Sent message to client",
zap.String("client_id", client.ID),
zap.String("action", msg.Action))
default:
h.logger.Warn("Client buffer full, dropping message",
zap.String("client_id", client.ID))
} else {
h.logger.Warn("Client send buffer full, dropping message",
zap.String("client_id", client.ID),
zap.String("action", msg.Action))
}
}
}
@@ -0,0 +1,50 @@
package controller
import (
"context"
"github.com/kandev/kandev/internal/task/dto"
"github.com/kandev/kandev/internal/task/service"
v1 "github.com/kandev/kandev/pkg/api/v1"
)
type CommentController struct {
service *service.Service
}
func NewCommentController(svc *service.Service) *CommentController {
return &CommentController{service: svc}
}
func (c *CommentController) ListComments(ctx context.Context, req dto.ListCommentsRequest) (dto.ListCommentsResponse, error) {
comments, err := c.service.ListComments(ctx, req.TaskID)
if err != nil {
return dto.ListCommentsResponse{}, err
}
result := make([]*v1.Comment, 0, len(comments))
for _, comment := range comments {
result = append(result, comment.ToAPI())
}
return dto.ListCommentsResponse{
Comments: result,
Total: len(result),
}, nil
}
func (c *CommentController) CreateComment(ctx context.Context, req dto.CreateCommentRequest) (*v1.Comment, error) {
comment, err := c.service.CreateComment(ctx, &service.CreateCommentRequest{
TaskID: req.TaskID,
Content: req.Content,
AuthorType: req.AuthorType,
AuthorID: req.AuthorID,
Type: req.Type,
RequestsInput: req.RequestsInput,
Metadata: req.Metadata,
ACPSessionID: req.ACPSessionID,
AgentSessionID: req.AgentSessionID,
})
if err != nil {
return nil, err
}
return comment.ToAPI(), nil
}
+5
View File
@@ -119,6 +119,11 @@ type BoardSnapshotDTO struct {
Tasks []TaskDTO `json:"tasks"`
}
type ListCommentsResponse struct {
Comments []*v1.Comment `json:"comments"`
Total int `json:"total"`
}
type ListBoardsResponse struct {
Boards []BoardDTO `json:"boards"`
Total int `json:"total"`
@@ -260,6 +260,22 @@ type DeleteTaskRequest struct {
ID string
}
type ListCommentsRequest struct {
TaskID string
}
type CreateCommentRequest struct {
TaskID string
Content string
AuthorType string
AuthorID string
Type string
RequestsInput bool
Metadata map[string]interface{}
ACPSessionID string
AgentSessionID string
}
type MoveTaskRequest struct {
ID string
BoardID string
@@ -2,12 +2,14 @@ package handlers
import (
"context"
"net/http"
"github.com/gin-gonic/gin"
"go.uber.org/zap"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/task/service"
"github.com/kandev/kandev/internal/task/controller"
"github.com/kandev/kandev/internal/task/dto"
v1 "github.com/kandev/kandev/pkg/api/v1"
ws "github.com/kandev/kandev/pkg/websocket"
)
@@ -25,31 +27,54 @@ type OrchestratorService interface {
// CommentHandlers handles WebSocket requests for comments
type CommentHandlers struct {
svc *service.Service
commentController *controller.CommentController
taskController *controller.TaskController
orchestrator OrchestratorService
logger *logger.Logger
}
// NewCommentHandlers creates a new CommentHandlers instance
func NewCommentHandlers(svc *service.Service, orchestrator OrchestratorService, log *logger.Logger) *CommentHandlers {
func NewCommentHandlers(commentCtrl *controller.CommentController, taskCtrl *controller.TaskController, orchestrator OrchestratorService, log *logger.Logger) *CommentHandlers {
return &CommentHandlers{
svc: svc,
commentController: commentCtrl,
taskController: taskCtrl,
orchestrator: orchestrator,
logger: log.WithFields(zap.String("component", "task-comment-handlers")),
}
}
// RegisterCommentRoutes registers comment WebSocket handlers
func RegisterCommentRoutes(router *gin.Engine, dispatcher *ws.Dispatcher, svc *service.Service, orchestrator OrchestratorService, log *logger.Logger) {
handlers := NewCommentHandlers(svc, orchestrator, log)
func RegisterCommentRoutes(router *gin.Engine, dispatcher *ws.Dispatcher, commentCtrl *controller.CommentController, taskCtrl *controller.TaskController, orchestrator OrchestratorService, log *logger.Logger) {
handlers := NewCommentHandlers(commentCtrl, taskCtrl, orchestrator, log)
handlers.registerHTTP(router)
handlers.registerWS(dispatcher)
}
func (h *CommentHandlers) registerHTTP(router *gin.Engine) {
api := router.Group("/api/v1")
api.GET("/tasks/:id/comments", h.httpListComments)
}
func (h *CommentHandlers) registerWS(dispatcher *ws.Dispatcher) {
dispatcher.RegisterFunc(ws.ActionCommentAdd, h.wsAddComment)
dispatcher.RegisterFunc(ws.ActionCommentList, h.wsListComments)
}
func (h *CommentHandlers) httpListComments(c *gin.Context) {
taskID := c.Param("id")
if taskID == "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "task id is required"})
return
}
resp, err := h.commentController.ListComments(c.Request.Context(), dto.ListCommentsRequest{TaskID: taskID})
if err != nil {
h.logger.Error("failed to list comments", zap.Error(err))
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list comments"})
return
}
c.JSON(http.StatusOK, resp)
}
// WS handlers
type wsAddCommentRequest struct {
@@ -71,7 +96,7 @@ func (h *CommentHandlers) wsAddComment(ctx context.Context, msg *ws.Message) (*w
}
// Get the current task state to determine if we need to transition
task, err := h.svc.GetTask(ctx, req.TaskID)
task, err := h.taskController.GetTask(ctx, dto.GetTaskRequest{ID: req.TaskID})
if err != nil {
h.logger.Error("failed to get task", zap.String("task_id", req.TaskID), zap.Error(err))
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to get task", nil)
@@ -82,7 +107,8 @@ func (h *CommentHandlers) wsAddComment(ctx context.Context, msg *ws.Message) (*w
// REVIEW → (user sends follow-up) → IN_PROGRESS
// The transition back to REVIEW happens in handleAgentReady when the agent finishes
if task.State == v1.TaskStateReview {
if _, err := h.svc.UpdateTaskState(ctx, req.TaskID, v1.TaskStateInProgress); err != nil {
nextState := v1.TaskStateInProgress
if _, err := h.taskController.UpdateTask(ctx, dto.UpdateTaskRequest{ID: req.TaskID, State: &nextState}); err != nil {
h.logger.Error("failed to transition task from REVIEW to IN_PROGRESS",
zap.String("task_id", req.TaskID),
zap.Error(err))
@@ -93,7 +119,7 @@ func (h *CommentHandlers) wsAddComment(ctx context.Context, msg *ws.Message) (*w
}
}
comment, err := h.svc.CreateComment(ctx, &service.CreateCommentRequest{
comment, err := h.commentController.CreateComment(ctx, dto.CreateCommentRequest{
TaskID: req.TaskID,
Content: req.Content,
AuthorType: "user",
@@ -132,12 +158,11 @@ func (h *CommentHandlers) wsListComments(ctx context.Context, msg *ws.Message) (
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil)
}
comments, err := h.svc.ListComments(ctx, req.TaskID)
resp, err := h.commentController.ListComments(ctx, dto.ListCommentsRequest{TaskID: req.TaskID})
if err != nil {
h.logger.Error("failed to list comments", zap.Error(err))
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to list comments", nil)
}
return ws.NewResponse(msg.ID, msg.Action, comments)
return ws.NewResponse(msg.ID, msg.Action, resp)
}
+8 -3
View File
@@ -46,6 +46,7 @@ const DEFAULT_HORIZONTAL_LAYOUT: [number, number] = [75, 25];
export default function TaskPage({ task: initialTask }: TaskPageClientProps) {
const store = useAppStoreApi();
const commentsState = useAppStore((state) => state.comments);
const [isMounted, setIsMounted] = useState(false);
const [horizontalLayout, setHorizontalLayout] = useState<[number, number]>(
getLocalStorage('task-layout-horizontal', DEFAULT_HORIZONTAL_LAYOUT)
@@ -118,6 +119,10 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) {
// Clear git status when switching tasks to avoid showing stale data
store.getState().clearGitStatus();
if (commentsState.taskId === task.id) {
return;
}
const fetchComments = async () => {
const client = getWebSocketClient();
if (!client) return;
@@ -126,9 +131,9 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) {
store.getState().setCommentsLoading(true);
try {
const response = await client.request<Comment[]>('comment.list', { task_id: task.id }, 10000);
const response = await client.request<{ comments: Comment[] }>('comment.list', { task_id: task.id }, 10000);
console.log('[API] comment.list response:', JSON.stringify(response, null, 2));
store.getState().setComments(task.id, response);
store.getState().setComments(task.id, response.comments ?? []);
} catch (error) {
console.error('Failed to fetch comments:', error);
store.getState().setComments(task.id, []);
@@ -138,7 +143,7 @@ export default function TaskPage({ task: initialTask }: TaskPageClientProps) {
};
fetchComments();
}, [task?.id, store]);
}, [commentsState.taskId, task?.id, store]);
// Subscribe to task for real-time updates
useEffect(() => {
+4 -2
View File
@@ -1,5 +1,5 @@
import { StateHydrator } from '@/components/state-hydrator';
import { fetchTask } from '@/lib/http';
import { fetchTask, listTaskComments } from '@/lib/http';
import type { Task } from '@/lib/types/http';
import { taskToState } from '@/lib/ssr/mapper';
import TaskPageClient from './page-client';
@@ -7,11 +7,13 @@ import TaskPageClient from './page-client';
export default async function TaskPage({ params }: { params: Promise<{ id: string }> }) {
let initialState: ReturnType<typeof taskToState> | null = null;
let task: Task | null = null;
let comments: Awaited<ReturnType<typeof listTaskComments>> | null = null;
try {
const { id } = await params;
task = await fetchTask(id, { cache: 'no-store' });
initialState = taskToState(task);
comments = await listTaskComments(id, { cache: 'no-store' });
initialState = taskToState(task, comments.comments);
} catch {
initialState = null;
task = null;
+6
View File
@@ -8,6 +8,8 @@ import type {
ListExecutorsResponse,
ListRepositoriesResponse,
ListWorkspacesResponse,
Comment,
ListCommentsResponse,
RepositoryBranchesResponse,
Task,
UserSettingsResponse,
@@ -160,6 +162,10 @@ export async function fetchTask(taskId: string, options?: ApiRequestOptions) {
return fetchJson<Task>(`/api/v1/tasks/${taskId}`, options);
}
export async function listTaskComments(taskId: string, options?: ApiRequestOptions) {
return fetchJson<ListCommentsResponse>(`/api/v1/tasks/${taskId}/comments`, options);
}
export async function listExecutors(options?: ApiRequestOptions): Promise<ListExecutorsResponse> {
return fetchJson<ListExecutorsResponse>('/api/v1/executors', options);
}
+9 -2
View File
@@ -1,5 +1,5 @@
import type { AppState, KanbanState } from '@/lib/state/store';
import type { BoardSnapshot, Task } from '@/lib/types/http';
import type { BoardSnapshot, Comment, Task } from '@/lib/types/http';
type KanbanTask = KanbanState['tasks'][number];
@@ -34,10 +34,17 @@ export function snapshotToState(snapshot: BoardSnapshot): Partial<AppState> {
};
}
export function taskToState(task: Task): Partial<AppState> {
export function taskToState(task: Task, comments?: Comment[]): Partial<AppState> {
return {
tasks: {
activeTaskId: task.id,
},
comments: comments
? {
taskId: task.id,
items: comments,
isLoading: false,
}
: undefined,
};
}
+5
View File
@@ -212,6 +212,11 @@ export type ListEnvironmentsResponse = {
total: number;
};
export type ListCommentsResponse = {
comments: Comment[];
total: number;
};
export type CommentAuthorType = 'user' | 'agent';
export type CommentType = 'message' | 'content' | 'tool_call' | 'progress' | 'error' | 'status';