marketing-shibata50/multica
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(activity): unified activity timeline with comment reply support

Browse source 
Replace the comment-only list with a Linear-style unified timeline that
interleaves field changes and comments chronologically.

Backend:
- activity_listeners.go: records field changes (status, assignee, description,
  task completed/failed) to activity_log table on domain events
- Timeline API: GET /api/issues/{id}/timeline merges activity_log + comments
  sorted by created_at
- Comment reply: parent_id column + handler support for threading

Frontend:
- Unified timeline replaces comment list: activity entries as compact muted
  lines, comments as Card components with reply threading
- Filter toggle (All / Comments / Activity)
- Reply UI: inline editor under comments with Cancel/Reply buttons
- Real-time sync for activity:created + comment events
- 10 new Go tests, all passing

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Naiyuan Qing 2026-03-28 21:53:08 +08:00
parent 3bb79564ed
commit e7fe6ea79b
21 changed files with 1307 additions and 132 deletions
Download patch file Download diff file Expand all files Collapse all files

207
server/cmd/server/activity_listeners.go Normal file
View file

@ -0,0 +1,207 @@
package main
import (
"context"
"encoding/json"
"log/slog"
"github.com/multica-ai/multica/server/internal/events"
"github.com/multica-ai/multica/server/internal/handler"
"github.com/multica-ai/multica/server/internal/util"
db "github.com/multica-ai/multica/server/pkg/db/generated"
"github.com/multica-ai/multica/server/pkg/protocol"
)
// registerActivityListeners wires up event bus listeners that record activity
// entries in the activity_log table. Each listener creates one or more activity
// records depending on what changed, then publishes an activity:created event
// for WS broadcasting.
func registerActivityListeners(bus *events.Bus, queries *db.Queries) {
ctx := context.Background()
// issue:created — record "created" activity
bus.Subscribe(protocol.EventIssueCreated, func(e events.Event) {
payload, ok := e.Payload.(map[string]any)
if !ok {
return
}
issue, ok := payload["issue"].(handler.IssueResponse)
if !ok {
return
}
activity, err := queries.CreateActivity(ctx, db.CreateActivityParams{
WorkspaceID: parseUUID(issue.WorkspaceID),
IssueID: parseUUID(issue.ID),
ActorType: util.StrToText(e.ActorType),
ActorID: parseUUID(e.ActorID),
Action: "created",
Details: []byte("{}"),
})
if err != nil {
slog.Error("activity: failed to record issue created",
"issue_id", issue.ID, "error", err)
return
}
publishActivityEvent(bus, e, activity)
})
// issue:updated — record specific changes as separate activities
bus.Subscribe(protocol.EventIssueUpdated, func(e events.Event) {
payload, ok := e.Payload.(map[string]any)
if !ok {
return
}
issue, ok := payload["issue"].(handler.IssueResponse)
if !ok {
return
}
statusChanged, _ := payload["status_changed"].(bool)
assigneeChanged, _ := payload["assignee_changed"].(bool)
descriptionChanged, _ := payload["description_changed"].(bool)
if statusChanged {
prevStatus, _ := payload["prev_status"].(string)
details, _ := json.Marshal(map[string]string{
"from": prevStatus,
"to": issue.Status,
})
activity, err := queries.CreateActivity(ctx, db.CreateActivityParams{
WorkspaceID: parseUUID(issue.WorkspaceID),
IssueID: parseUUID(issue.ID),
ActorType: util.StrToText(e.ActorType),
ActorID: parseUUID(e.ActorID),
Action: "status_changed",
Details: details,
})
if err != nil {
slog.Error("activity: failed to record status change",
"issue_id", issue.ID, "error", err)
} else {
publishActivityEvent(bus, e, activity)
}
}
if assigneeChanged {
prevAssigneeType, _ := payload["prev_assignee_type"].(*string)
prevAssigneeID, _ := payload["prev_assignee_id"].(*string)
detailsMap := map[string]string{}
if prevAssigneeType != nil {
detailsMap["from_type"] = *prevAssigneeType
}
if prevAssigneeID != nil {
detailsMap["from_id"] = *prevAssigneeID
}
if issue.AssigneeType != nil {
detailsMap["to_type"] = *issue.AssigneeType
}
if issue.AssigneeID != nil {
detailsMap["to_id"] = *issue.AssigneeID
}
details, _ := json.Marshal(detailsMap)
activity, err := queries.CreateActivity(ctx, db.CreateActivityParams{
WorkspaceID: parseUUID(issue.WorkspaceID),
IssueID: parseUUID(issue.ID),
ActorType: util.StrToText(e.ActorType),
ActorID: parseUUID(e.ActorID),
Action: "assignee_changed",
Details: details,
})
if err != nil {
slog.Error("activity: failed to record assignee change",
"issue_id", issue.ID, "error", err)
} else {
publishActivityEvent(bus, e, activity)
}
}
if descriptionChanged {
activity, err := queries.CreateActivity(ctx, db.CreateActivityParams{
WorkspaceID: parseUUID(issue.WorkspaceID),
IssueID: parseUUID(issue.ID),
ActorType: util.StrToText(e.ActorType),
ActorID: parseUUID(e.ActorID),
Action: "description_updated",
Details: []byte("{}"),
})
if err != nil {
slog.Error("activity: failed to record description change",
"issue_id", issue.ID, "error", err)
} else {
publishActivityEvent(bus, e, activity)
}
}
})
// task:completed — record "task_completed" activity
bus.Subscribe(protocol.EventTaskCompleted, func(e events.Event) {
handleTaskActivity(ctx, bus, queries, e, "task_completed")
})
// task:failed — record "task_failed" activity
bus.Subscribe(protocol.EventTaskFailed, func(e events.Event) {
handleTaskActivity(ctx, bus, queries, e, "task_failed")
})
}
// handleTaskActivity records an activity for task:completed or task:failed events.
func handleTaskActivity(ctx context.Context, bus *events.Bus, queries *db.Queries, e events.Event, action string) {
payload, ok := e.Payload.(map[string]any)
if !ok {
return
}
agentID, _ := payload["agent_id"].(string)
issueID, _ := payload["issue_id"].(string)
if issueID == "" {
return
}
// Look up issue to get workspace_id
issue, err := queries.GetIssue(ctx, parseUUID(issueID))
if err != nil {
slog.Error("activity: failed to get issue for task event",
"issue_id", issueID, "action", action, "error", err)
return
}
activity, err := queries.CreateActivity(ctx, db.CreateActivityParams{
WorkspaceID: issue.WorkspaceID,
IssueID: parseUUID(issueID),
ActorType: util.StrToText("agent"),
ActorID: parseUUID(agentID),
Action: action,
Details: []byte("{}"),
})
if err != nil {
slog.Error("activity: failed to record task activity",
"issue_id", issueID, "action", action, "error", err)
return
}
publishActivityEvent(bus, e, activity)
}
// publishActivityEvent sends an activity:created event for WS broadcasting.
func publishActivityEvent(bus *events.Bus, original events.Event, activity db.ActivityLog) {
bus.Publish(events.Event{
Type: protocol.EventActivityCreated,
WorkspaceID: original.WorkspaceID,
ActorType: original.ActorType,
ActorID: original.ActorID,
Payload: map[string]any{
"activity": map[string]any{
"id": util.UUIDToString(activity.ID),
"issue_id": util.UUIDToString(activity.IssueID),
"actor_type": util.TextToPtr(activity.ActorType),
"actor_id": util.UUIDToString(activity.ActorID),
"action": activity.Action,
"details": json.RawMessage(activity.Details),
"created_at": util.TimestampToString(activity.CreatedAt),
},
},
})
}

295
server/cmd/server/activity_listeners_test.go Normal file
View file

@ -0,0 +1,295 @@
package main
import (
"context"
"encoding/json"
"testing"
"github.com/multica-ai/multica/server/internal/events"
"github.com/multica-ai/multica/server/internal/handler"
"github.com/multica-ai/multica/server/internal/util"
db "github.com/multica-ai/multica/server/pkg/db/generated"
"github.com/multica-ai/multica/server/pkg/protocol"
)
// listActivitiesForIssue is a test helper that fetches all activity_log records for an issue.
func listActivitiesForIssue(t *testing.T, queries *db.Queries, issueID string) []db.ActivityLog {
t.Helper()
activities, err := queries.ListActivities(context.Background(), db.ListActivitiesParams{
IssueID: util.ParseUUID(issueID),
Limit: 100,
Offset: 0,
})
if err != nil {
t.Fatalf("ListActivities: %v", err)
}
return activities
}
func cleanupActivities(t *testing.T, issueID string) {
t.Helper()
testPool.Exec(context.Background(), `DELETE FROM activity_log WHERE issue_id = $1`, issueID)
}
func TestActivityIssueCreated(t *testing.T) {
queries := db.New(testPool)
bus := events.New()
registerActivityListeners(bus, queries)
issueID := createTestIssue(t, testWorkspaceID, testUserID)
t.Cleanup(func() {
cleanupActivities(t, issueID)
cleanupTestIssue(t, issueID)
})
bus.Publish(events.Event{
Type: protocol.EventIssueCreated,
WorkspaceID: testWorkspaceID,
ActorType: "member",
ActorID: testUserID,
Payload: map[string]any{
"issue": handler.IssueResponse{
ID: issueID,
WorkspaceID: testWorkspaceID,
Title: "activity test issue",
Status: "todo",
Priority: "medium",
CreatorType: "member",
CreatorID: testUserID,
},
},
})
activities := listActivitiesForIssue(t, queries, issueID)
if len(activities) != 1 {
t.Fatalf("expected 1 activity, got %d", len(activities))
}
if activities[0].Action != "created" {
t.Fatalf("expected action 'created', got %q", activities[0].Action)
}
if util.UUIDToString(activities[0].ActorID) != testUserID {
t.Fatalf("expected actor_id %s, got %s", testUserID, util.UUIDToString(activities[0].ActorID))
}
}
func TestActivityIssueUpdated_StatusChanged(t *testing.T) {
queries := db.New(testPool)
bus := events.New()
registerActivityListeners(bus, queries)
issueID := createTestIssue(t, testWorkspaceID, testUserID)
t.Cleanup(func() {
cleanupActivities(t, issueID)
cleanupTestIssue(t, issueID)
})
bus.Publish(events.Event{
Type: protocol.EventIssueUpdated,
WorkspaceID: testWorkspaceID,
ActorType: "member",
ActorID: testUserID,
Payload: map[string]any{
"issue": handler.IssueResponse{
ID: issueID,
WorkspaceID: testWorkspaceID,
Title: "activity test issue",
Status: "in_progress",
Priority: "medium",
CreatorType: "member",
CreatorID: testUserID,
},
"status_changed": true,
"prev_status": "todo",
},
})
activities := listActivitiesForIssue(t, queries, issueID)
if len(activities) != 1 {
t.Fatalf("expected 1 activity, got %d", len(activities))
}
if activities[0].Action != "status_changed" {
t.Fatalf("expected action 'status_changed', got %q", activities[0].Action)
}
var details map[string]string
if err := json.Unmarshal(activities[0].Details, &details); err != nil {
t.Fatalf("failed to unmarshal details: %v", err)
}
if details["from"] != "todo" {
t.Fatalf("expected from 'todo', got %q", details["from"])
}
if details["to"] != "in_progress" {
t.Fatalf("expected to 'in_progress', got %q", details["to"])
}
}
func TestActivityIssueUpdated_AssigneeChanged(t *testing.T) {
queries := db.New(testPool)
bus := events.New()
registerActivityListeners(bus, queries)
assigneeEmail := "activity-assignee-test@multica.ai"
assigneeID := createTestUser(t, assigneeEmail)
t.Cleanup(func() { cleanupTestUser(t, assigneeEmail) })
issueID := createTestIssue(t, testWorkspaceID, testUserID)
t.Cleanup(func() {
cleanupActivities(t, issueID)
cleanupTestIssue(t, issueID)
})
assigneeType := "member"
bus.Publish(events.Event{
Type: protocol.EventIssueUpdated,
WorkspaceID: testWorkspaceID,
ActorType: "member",
ActorID: testUserID,
Payload: map[string]any{
"issue": handler.IssueResponse{
ID: issueID,
WorkspaceID: testWorkspaceID,
Title: "activity test issue",
Status: "todo",
Priority: "medium",
CreatorType: "member",
CreatorID: testUserID,
AssigneeType: &assigneeType,
AssigneeID: &assigneeID,
},
"assignee_changed": true,
"prev_assignee_type": (*string)(nil),
"prev_assignee_id": (*string)(nil),
},
})
activities := listActivitiesForIssue(t, queries, issueID)
if len(activities) != 1 {
t.Fatalf("expected 1 activity, got %d", len(activities))
}
if activities[0].Action != "assignee_changed" {
t.Fatalf("expected action 'assignee_changed', got %q", activities[0].Action)
}
var details map[string]string
if err := json.Unmarshal(activities[0].Details, &details); err != nil {
t.Fatalf("failed to unmarshal details: %v", err)
}
if details["to_type"] != "member" {
t.Fatalf("expected to_type 'member', got %q", details["to_type"])
}
if details["to_id"] != assigneeID {
t.Fatalf("expected to_id %q, got %q", assigneeID, details["to_id"])
}
}
func TestActivityIssueUpdated_NoChangeFlags(t *testing.T) {
queries := db.New(testPool)
bus := events.New()
registerActivityListeners(bus, queries)
issueID := createTestIssue(t, testWorkspaceID, testUserID)
t.Cleanup(func() {
cleanupActivities(t, issueID)
cleanupTestIssue(t, issueID)
})
// Publish issue:updated with no change flags set
bus.Publish(events.Event{
Type: protocol.EventIssueUpdated,
WorkspaceID: testWorkspaceID,
ActorType: "member",
ActorID: testUserID,
Payload: map[string]any{
"issue": handler.IssueResponse{
ID: issueID,
WorkspaceID: testWorkspaceID,
Title: "activity test issue",
Status: "todo",
Priority: "medium",
CreatorType: "member",
CreatorID: testUserID,
},
"assignee_changed": false,
"status_changed": false,
"description_changed": false,
},
})
activities := listActivitiesForIssue(t, queries, issueID)
if len(activities) != 0 {
t.Fatalf("expected 0 activities when no change flags, got %d", len(activities))
}
}
func TestActivityTaskCompleted(t *testing.T) {
queries := db.New(testPool)
bus := events.New()
registerActivityListeners(bus, queries)
issueID := createTestIssue(t, testWorkspaceID, testUserID)
t.Cleanup(func() {
cleanupActivities(t, issueID)
cleanupTestIssue(t, issueID)
})
agentID := testUserID // reuse as a stand-in for agent ID
bus.Publish(events.Event{
Type: protocol.EventTaskCompleted,
WorkspaceID: testWorkspaceID,
ActorType: "system",
ActorID: "",
Payload: map[string]any{
"task_id": "00000000-0000-0000-0000-000000000001",
"agent_id": agentID,
"issue_id": issueID,
"status": "completed",
},
})
activities := listActivitiesForIssue(t, queries, issueID)
if len(activities) != 1 {
t.Fatalf("expected 1 activity, got %d", len(activities))
}
if activities[0].Action != "task_completed" {
t.Fatalf("expected action 'task_completed', got %q", activities[0].Action)
}
if util.UUIDToString(activities[0].ActorID) != agentID {
t.Fatalf("expected actor_id %s, got %s", agentID, util.UUIDToString(activities[0].ActorID))
}
}
func TestActivityTaskFailed(t *testing.T) {
queries := db.New(testPool)
bus := events.New()
registerActivityListeners(bus, queries)
issueID := createTestIssue(t, testWorkspaceID, testUserID)
t.Cleanup(func() {
cleanupActivities(t, issueID)
cleanupTestIssue(t, issueID)
})
agentID := testUserID
bus.Publish(events.Event{
Type: protocol.EventTaskFailed,
WorkspaceID: testWorkspaceID,
ActorType: "system",
ActorID: "",
Payload: map[string]any{
"task_id": "00000000-0000-0000-0000-000000000002",
"agent_id": agentID,
"issue_id": issueID,
"status": "failed",
},
})
activities := listActivitiesForIssue(t, queries, issueID)
if len(activities) != 1 {
t.Fatalf("expected 1 activity, got %d", len(activities))
}
if activities[0].Action != "task_failed" {
t.Fatalf("expected action 'task_failed', got %q", activities[0].Action)
}
}

1
server/cmd/server/listeners.go
View file

@ -37,6 +37,7 @@ func registerListeners(bus *events.Bus, hub *realtime.Hub) {
protocol.EventMemberRemoved,
protocol.EventSubscriberAdded,
protocol.EventSubscriberRemoved,
protocol.EventActivityCreated,
}
for _, et := range allEvents {

1
server/cmd/server/main.go
View file

@ -55,6 +55,7 @@ func main() {
// The notification listener queries the subscriber table to determine recipients,
// so subscribers must be written first within the same synchronous event dispatch.
registerSubscriberListeners(bus, queries)
registerActivityListeners(bus, queries)
registerNotificationListeners(bus, queries)
r := NewRouter(pool, hub, bus)

1
server/cmd/server/router.go
View file

@ -117,6 +117,7 @@ func NewRouter(pool *pgxpool.Pool, hub *realtime.Hub, bus *events.Bus) chi.Route
r.Delete("/", h.DeleteIssue)
r.Post("/comments", h.CreateComment)
r.Get("/comments", h.ListComments)
r.Get("/timeline", h.ListTimeline)
r.Get("/subscribers", h.ListIssueSubscribers)
r.Post("/subscribe", h.SubscribeToIssue)
r.Post("/unsubscribe", h.UnsubscribeFromIssue)