diff --git a/server/cmd/server/notification_listeners.go b/server/cmd/server/notification_listeners.go index e94c95f471..0c15c38f0f 100644 --- a/server/cmd/server/notification_listeners.go +++ b/server/cmd/server/notification_listeners.go @@ -485,7 +485,7 @@ func notifyDirect( }) if err != nil { slog.Error("direct notification creation failed", - "recipient_id", recipientID, "type", notifType, "error", err) + "issue_id", issueID, "recipient_id", recipientID, "type", notifType, "error", err) return } @@ -634,8 +634,8 @@ func registerNotificationListeners(bus *events.Bus, queries *db.Queries) { // Track who already got notified to avoid duplicates skip := map[string]bool{e.ActorID: true} - // Direct notification to assignee - if issue.AssigneeType != nil && issue.AssigneeID != nil { + // Direct notification to assignees that own an inbox. + if issue.AssigneeType != nil && issue.AssigneeID != nil && isAssignmentRecipientType(*issue.AssigneeType) { skip[*issue.AssigneeID] = true notifyDirect(ctx, queries, bus, *issue.AssigneeType, *issue.AssigneeID, @@ -697,8 +697,8 @@ func registerNotificationListeners(bus *events.Bus, queries *db.Queries) { } assigneeDetails, _ := json.Marshal(detailsMap) - // Direct: notify new assignee about assignment - if issue.AssigneeType != nil && issue.AssigneeID != nil { + // Direct: notify new assignee about assignment when it owns an inbox. + if issue.AssigneeType != nil && issue.AssigneeID != nil && isAssignmentRecipientType(*issue.AssigneeType) { notifyDirect(ctx, queries, bus, *issue.AssigneeType, *issue.AssigneeID, e.WorkspaceID, e, issue.ID, issue.Status, @@ -709,7 +709,9 @@ func registerNotificationListeners(bus *events.Bus, queries *db.Queries) { ) } - // Direct: notify old assignee about unassignment + // Direct: notify only a previous member assignee about unassignment. + // This is intentionally narrower than isAssignmentRecipientType: agents + // do not receive unassigned notifications. if prevAssigneeType != nil && prevAssigneeID != nil && *prevAssigneeType == "member" { notifyDirect(ctx, queries, bus, "member", *prevAssigneeID, diff --git a/server/cmd/server/squad_assignee_listeners_test.go b/server/cmd/server/squad_assignee_listeners_test.go new file mode 100644 index 0000000000..4ec2ee09f8 --- /dev/null +++ b/server/cmd/server/squad_assignee_listeners_test.go @@ -0,0 +1,122 @@ +package main + +import ( + "bytes" + "context" + "log/slog" + "strings" + "testing" + + "github.com/multica-ai/multica/server/internal/events" + "github.com/multica-ai/multica/server/internal/handler" + db "github.com/multica-ai/multica/server/pkg/db/generated" + "github.com/multica-ai/multica/server/pkg/protocol" +) + +func createAssignmentListenerTestSquad(t *testing.T) string { + t.Helper() + + ctx := context.Background() + var leaderID string + if err := testPool.QueryRow(ctx, ` + SELECT id::text FROM agent + WHERE workspace_id = $1 + ORDER BY created_at ASC + LIMIT 1 + `, testWorkspaceID).Scan(&leaderID); err != nil { + t.Fatalf("load squad leader: %v", err) + } + + var squadID string + if err := testPool.QueryRow(ctx, ` + INSERT INTO squad (workspace_id, name, description, leader_id, creator_id) + VALUES ($1, 'Squad assignee listener test', '', $2, $3) + RETURNING id + `, testWorkspaceID, leaderID, testUserID).Scan(&squadID); err != nil { + t.Fatalf("create squad: %v", err) + } + t.Cleanup(func() { + if _, err := testPool.Exec(context.Background(), `DELETE FROM squad WHERE id = $1`, squadID); err != nil { + t.Errorf("cleanup squad: %v", err) + } + }) + return squadID +} + +// A squad is a routing object, not a subscriber or inbox recipient. Both +// issue event paths must stop before attempting writes that the database +// constrains to member/agent identities. The log assertion is load-bearing: +// checking only for zero squad rows would also pass in the broken version +// because PostgreSQL rejects those rows before the listeners log and return. +func TestSquadAssigneeListenersSkipUnsupportedRecipientWrites(t *testing.T) { + queries := db.New(testPool) + bus := events.New() + registerSubscriberListeners(bus, testPool) + registerNotificationListeners(bus, queries) + + squadID := createAssignmentListenerTestSquad(t) + issueID := createTestIssue(t, testWorkspaceID, testUserID) + t.Cleanup(func() { + cleanupInboxForIssue(t, issueID) + cleanupTestIssue(t, issueID) + }) + + var logs bytes.Buffer + previousLogger := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelError}))) + defer slog.SetDefault(previousLogger) + + assigneeType := "squad" + issue := handler.IssueResponse{ + ID: issueID, + WorkspaceID: testWorkspaceID, + Title: "squad-assigned issue", + Status: "todo", + Priority: "medium", + CreatorType: "member", + CreatorID: testUserID, + AssigneeType: &assigneeType, + AssigneeID: &squadID, + } + + bus.Publish(events.Event{ + Type: protocol.EventIssueCreated, + WorkspaceID: testWorkspaceID, + ActorType: "member", + ActorID: testUserID, + Payload: map[string]any{"issue": issue}, + }) + bus.Publish(events.Event{ + Type: protocol.EventIssueUpdated, + WorkspaceID: testWorkspaceID, + ActorType: "member", + ActorID: testUserID, + Payload: map[string]any{ + "issue": issue, + "assignee_changed": true, + }, + }) + + // slog's default logger is process-global. Scope the captured records to + // this test's unique issue so unrelated background errors cannot fail it. + var issueLogLines []string + for _, line := range strings.Split(logs.String(), "\n") { + if strings.Contains(line, "issue_id="+issueID) { + issueLogLines = append(issueLogLines, line) + } + } + issueLogs := strings.Join(issueLogLines, "\n") + for _, unexpected := range []string{ + "failed to add issue subscriber", + "direct notification creation failed", + "SQLSTATE 23514", + } { + if strings.Contains(issueLogs, unexpected) { + t.Fatalf("squad assignment attempted an unsupported recipient write (%q):\n%s", unexpected, issueLogs) + } + } + + if count := subscriberCount(t, queries, issueID); count != 1 { + t.Fatalf("subscriber count = %d, want creator only", count) + } +} diff --git a/server/cmd/server/subscriber_listeners.go b/server/cmd/server/subscriber_listeners.go index c898a10839..afaf056823 100644 --- a/server/cmd/server/subscriber_listeners.go +++ b/server/cmd/server/subscriber_listeners.go @@ -13,6 +13,13 @@ import ( "github.com/multica-ai/multica/server/pkg/protocol" ) +// isAssignmentRecipientType reports whether an assignee can own a subscriber +// or inbox row. Squads are routing objects whose work runs through the leader; +// they are not user identities and have no inbox to consume. +func isAssignmentRecipientType(assigneeType string) bool { + return assigneeType == "member" || assigneeType == "agent" +} + // registerSubscriberListeners wires up event bus listeners that auto-subscribe // relevant users to issues. This ensures creators, assignees, and commenters // are automatically tracked as issue subscribers. @@ -38,8 +45,9 @@ func registerSubscriberListeners(bus *events.Bus, pool *pgxpool.Pool) { // Subscribe the creator addSubscriber(bus, queries, e.WorkspaceID, issue.ID, issue.CreatorType, issue.CreatorID, "creator") - // Subscribe the assignee if exists and different from creator + // Subscribe the assignee if it is a direct recipient and differs from the creator. if issue.AssigneeType != nil && issue.AssigneeID != nil && + isAssignmentRecipientType(*issue.AssigneeType) && !(*issue.AssigneeType == issue.CreatorType && *issue.AssigneeID == issue.CreatorID) { addSubscriber(bus, queries, e.WorkspaceID, issue.ID, *issue.AssigneeType, *issue.AssigneeID, "assignee") } @@ -72,7 +80,7 @@ func registerSubscriberListeners(bus *events.Bus, pool *pgxpool.Pool) { // Subscribe new assignee if assignee changed if assigneeChanged, _ := payload["assignee_changed"].(bool); assigneeChanged { - if issue.AssigneeType != nil && issue.AssigneeID != nil { + if issue.AssigneeType != nil && issue.AssigneeID != nil && isAssignmentRecipientType(*issue.AssigneeType) { addSubscriber(bus, queries, e.WorkspaceID, issue.ID, *issue.AssigneeType, *issue.AssigneeID, "assignee") } }