package message import ( "context" "database/sql" "errors" "fmt" "net/url" "strconv" "strings" "time" "github.com/tidbcloud/AgentTeam/internal/coredb" "github.com/tidbcloud/AgentTeam/internal/store/access" "github.com/tidbcloud/AgentTeam/internal/store/model" ) const ( ConversationKindChannel = "channel" ConversationKindThread = "thread" ConversationKindDM = "dm" ConversationKindTask = "task" ConversationKindServiceAccount = "service_account" ) const ( ConversationActionPin = "pin" ConversationActionUnpin = "unpin" ConversationActionMarkRead = "mark_read" ConversationActionMarkUnread = "mark_unread" ConversationActionMute = "mute" ConversationActionUnmute = "unmute" ConversationActionHide = "hide" ) type ConversationKindFilter struct { Kinds []string AllSubsystems bool } type ConversationBulkActionResult struct { UpdatedCount int64 ReadScopes []ConversationReadScope } type ConversationReadScope struct { ScopeID string Kind string LastReadSeq int64 LastActivityReadSeq int64 ManualUnreadGeneration int64 ManualUnreadGenerationKnown bool } func NormalizeConversationKindFilter(filter ConversationKindFilter) (ConversationKindFilter, error) { if len(filter.Kinds) == 0 { return ConversationKindFilter{}, nil } seen := make(map[string]struct{}, len(filter.Kinds)) ordered := make([]string, 0, len(filter.Kinds)) for _, raw := range filter.Kinds { kind := strings.TrimSpace(raw) switch kind { case "channel", "dm", ConversationKindTask, ConversationKindServiceAccount: case "serviceAccount": kind = ConversationKindServiceAccount case "thread", "replyThread": kind = ConversationKindThread case "": continue default: return ConversationKindFilter{}, badRequest("unsupported conversation kind filter") } if _, ok := seen[kind]; ok { continue } seen[kind] = struct{}{} ordered = append(ordered, kind) } if len(ordered) == 0 { return ConversationKindFilter{}, nil } return ConversationKindFilter{Kinds: ordered}, nil } type conversationKindSelection struct { channel bool dm bool thread bool task bool serviceAccount bool } func conversationKindSelectionFromFilter(filter ConversationKindFilter) conversationKindSelection { if len(filter.Kinds) == 0 { return conversationKindSelection{channel: true, dm: true, thread: true, task: true, serviceAccount: true} } var selection conversationKindSelection for _, kind := range filter.Kinds { switch kind { case ConversationKindChannel: selection.channel = true case ConversationKindDM: selection.dm = true case ConversationKindThread: selection.thread = true case ConversationKindTask: selection.task = true case ConversationKindServiceAccount: selection.serviceAccount = true } } return selection } func (s conversationKindSelection) includesVisibleThreadRefs() bool { return s.channel || s.dm || s.thread || s.serviceAccount } func (f ConversationKindFilter) subsystemWhereClause() string { if f.AllSubsystems { return "" } return "AND (s.type <> 'channel' OR cp.subsystem IN ('im', 'team', 'tutorial'))" } type ConversationItem struct { Kind string ScopeID string ParentScopeID string ParentScopeType string ParentTitle string ParentMessageID string ParentMessageShortID string ParentMessageAuthor string Title string Subtitle string AvatarURL string Subsystem string TeamID string TeamHandle string TeamDisplayName string PeerType string PeerID string Preview string LastAuthorName string LastMessageAt *time.Time LastActivityAt *time.Time PinnedAt *time.Time UnreadCount int64 ManualUnread bool // ManualUnreadGeneration is the committed causal version used only by the // owner-local unread runtime. It is intentionally omitted from public DTOs. ManualUnreadGeneration int64 ManualUnreadGenerationKnown bool LastReadSeq int64 LastActivityReadSeq int64 // HiddenThroughSeq and HiddenThroughActivitySeq are the committed Task hide // frontiers consumed only by the owner-local unread runtime. Task discussion // messages and Task Activity have independent scope-local sequences. HiddenThroughSeq int64 HiddenThroughActivitySeq int64 SubscriptionState string SidebarHidden bool Archived bool SortAt time.Time ReplyCount int64 LastReplyAt *time.Time Task *ConversationTaskMetadata } type ConversationTaskMetadata struct { ID string ScopeID string ChannelID string Number int64 Title string Status string ParentTaskRef string SourceThreadID string SourceThreadParentChannelID string SourceThreadParentMessageID string SourceThreadRef string AssigneeProfile *AuthorProfile Route string } type ConversationPage struct { Items []ConversationItem HasMore bool NextCursor ConversationCursor } func (store *Store) ListConversationsForUser(ctx context.Context, userID string, workspaceID string, limit int, cursor ConversationCursor) (ConversationPage, error) { return store.ListConversationsForUserFiltered(ctx, userID, workspaceID, limit, cursor, ConversationKindFilter{}) } func (store *Store) ListConversationsForUserFiltered(ctx context.Context, userID string, workspaceID string, limit int, cursor ConversationCursor, filter ConversationKindFilter) (ConversationPage, error) { var result ConversationPage err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error { var err error result, err = store.InTx(tx).ListConversationsForUserFiltered(ctx, userID, workspaceID, limit, cursor, filter) return err }) if err != nil { return ConversationPage{}, err } return result, nil } func (store *Store) MarkConversationsReadForUser(ctx context.Context, userID string, workspaceID string, filter ConversationKindFilter) (ConversationBulkActionResult, error) { var result ConversationBulkActionResult err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error { var err error result, err = store.InTx(tx).MarkConversationsReadForUser(ctx, userID, workspaceID, filter) return err }) if err != nil { return ConversationBulkActionResult{}, err } return result, nil } func (txs TxStore) ListConversationsForUser(ctx context.Context, userID string, workspaceID string, limit int, cursor ConversationCursor) (ConversationPage, error) { return txs.ListConversationsForUserFiltered(ctx, userID, workspaceID, limit, cursor, ConversationKindFilter{}) } func (txs TxStore) ListConversationsForUserFiltered(ctx context.Context, userID string, workspaceID string, limit int, cursor ConversationCursor, filter ConversationKindFilter) (ConversationPage, error) { if limit <= 0 { return ConversationPage{}, badRequest("limit must be positive") } membership, err := access.LoadHumanWorkspaceMember(ctx, txs.tx, strings.TrimSpace(userID), strings.TrimSpace(workspaceID)) if err != nil { return ConversationPage{}, mapAccessError(err) } normalizedCursor, err := normalizeConversationCursor(cursor) if err != nil { return ConversationPage{}, err } query, args, err := BuildListConversationsForUserScopedQueryFiltered(workspaceID, membership.Member.ID, userID, limit, normalizedCursor, filter) if err != nil { return ConversationPage{}, err } rows, err := txs.tx.Query(ctx, query, args...) if err != nil { return ConversationPage{}, fmt.Errorf("list conversations: %w", err) } defer rows.Close() items := make([]ConversationItem, 0, limit+1) for rows.Next() { item, err := scanConversationItem(rows) if err != nil { return ConversationPage{}, err } items = append(items, item) } if err := rows.Err(); err != nil { return ConversationPage{}, fmt.Errorf("iterate conversation rows: %w", err) } hasMore := len(items) > limit if hasMore { items = items[:limit] } nextCursor := ConversationCursor{} if hasMore && len(items) > 0 { last := items[len(items)-1] nextCursor, err = normalizeConversationCursor(ConversationCursor{PinnedAt: last.PinnedAt, SortAt: last.SortAt, ScopeID: last.ScopeID}) if err != nil { return ConversationPage{}, err } } return ConversationPage{Items: items, HasMore: hasMore, NextCursor: nextCursor}, nil } func (txs TxStore) MarkConversationsReadForUser(ctx context.Context, userID string, workspaceID string, filter ConversationKindFilter) (ConversationBulkActionResult, error) { membership, err := access.LoadHumanWorkspaceMember(ctx, txs.tx, strings.TrimSpace(userID), strings.TrimSpace(workspaceID)) if err != nil { return ConversationBulkActionResult{}, mapAccessError(err) } normalizedFilter, err := NormalizeConversationKindFilter(filter) if err != nil { return ConversationBulkActionResult{}, err } selection := conversationKindSelectionFromFilter(normalizedFilter) query, args, err := BuildVisibleConversationScopeIDsForUserScopedQueryFiltered(workspaceID, membership.Member.ID, userID, normalizedFilter) if err != nil { return ConversationBulkActionResult{}, err } rows, err := txs.tx.Query(ctx, query, args...) if err != nil { return ConversationBulkActionResult{}, fmt.Errorf("list conversations for bulk read: %w", err) } type visibleConversation struct { scopeID string kind string } visible := make([]visibleConversation, 0) for rows.Next() { var row visibleConversation if err := rows.Scan(&row.scopeID, &row.kind); err != nil { return ConversationBulkActionResult{}, err } visible = append(visible, row) } if err := rows.Err(); err != nil { rows.Close() return ConversationBulkActionResult{}, fmt.Errorf("iterate conversations for bulk read: %w", err) } rows.Close() if selection.task { taskRows, err := txs.tx.Query(ctx, ` SELECT idx.scope_id, 'task' FROM im_conversation_index idx JOIN scopes task_scope ON task_scope.workspace_id = idx.workspace_id AND task_scope.id = idx.scope_id AND task_scope.type = 'task' AND task_scope.status = 'active' JOIN scope_memberships self_sm ON self_sm.scope_id = task_scope.id AND self_sm.member_id = idx.member_id AND self_sm.removed_at IS NULL AND self_sm.archived_at IS NULL AND self_sm.effective_to_seq IS NULL WHERE idx.workspace_id = $1 AND idx.member_id = $2 AND idx.kind = 'task' AND idx.archived_at IS NULL ORDER BY idx.scope_id `, strings.TrimSpace(workspaceID), strings.TrimSpace(membership.Member.ID)) if err != nil { return ConversationBulkActionResult{}, fmt.Errorf("list task conversations for bulk read: %w", err) } for taskRows.Next() { var row visibleConversation if err := taskRows.Scan(&row.scopeID, &row.kind); err != nil { taskRows.Close() return ConversationBulkActionResult{}, err } visible = append(visible, row) } if err := taskRows.Err(); err != nil { taskRows.Close() return ConversationBulkActionResult{}, fmt.Errorf("iterate task conversations for bulk read: %w", err) } taskRows.Close() } now := time.Now().UTC() readScopes := make([]ConversationReadScope, 0, len(visible)) for _, row := range visible { latestSeq, latestGlobalSeq, err := loadLatestHumanVisibleConversationSeq(ctx, txs.tx, workspaceID, row.scopeID) if err != nil { return ConversationBulkActionResult{}, err } if err := WriteThreadReadStateByGlobalSeq(ctx, txs.tx, membership.Member.ID, row.scopeID, latestSeq, latestGlobalSeq, now); err != nil { return ConversationBulkActionResult{}, fmt.Errorf("mark conversation read: %w", err) } var latestActivitySeq int64 latestActivitySeq, err = loadLatestConversationActivitySeq(ctx, txs.tx, workspaceID, row.scopeID) if err != nil { return ConversationBulkActionResult{}, err } if err := WriteActivityReadStateByGlobalSeq(ctx, txs.tx, membership.Member.ID, row.scopeID, latestActivitySeq, now); err != nil { return ConversationBulkActionResult{}, fmt.Errorf("mark conversation activity read: %w", err) } if err := ClearConversationManualUnreadState(ctx, txs.tx, membership.Member.ID, row.scopeID, now); err != nil { return ConversationBulkActionResult{}, err } _, manualUnreadGeneration, err := loadConversationManualUnreadState(ctx, txs.tx, membership.Member.ID, row.scopeID) if err != nil { return ConversationBulkActionResult{}, err } readScopes = append(readScopes, ConversationReadScope{ ScopeID: row.scopeID, Kind: row.kind, LastReadSeq: latestSeq, LastActivityReadSeq: latestActivitySeq, ManualUnreadGeneration: manualUnreadGeneration, ManualUnreadGenerationKnown: true, }) } return ConversationBulkActionResult{UpdatedCount: int64(len(visible)), ReadScopes: readScopes}, nil } func (store *Store) LoadConversationForUser(ctx context.Context, userID string, workspaceID string, scopeID string, includeHidden bool) (ConversationItem, error) { var result ConversationItem err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error { var err error result, err = store.InTx(tx).LoadConversationForUser(ctx, userID, workspaceID, scopeID, includeHidden) return err }) if err != nil { return ConversationItem{}, err } return result, nil } // MaterializeTaskConversationForMember ensures the owner-local Inbox conversation // row for one member and a visible Task scope exists, idempotently. It is the // message-store face of the authoritative Task Watch materialization boundary used by // the Task notification self-state persistence path and by the IM conversation // policy projector fallback. func (store *Store) MaterializeTaskConversationForMember(ctx context.Context, workspaceID string, memberID string, taskScopeID string) error { if store == nil || store.db == nil { return fmt.Errorf("message store is not configured") } return coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error { return MaterializeTaskConversationForMemberTx(ctx, tx, workspaceID, memberID, taskScopeID) }) } func (txs TxStore) LoadConversationForUser(ctx context.Context, userID string, workspaceID string, scopeID string, includeHidden bool) (ConversationItem, error) { membership, err := access.LoadHumanWorkspaceMember(ctx, txs.tx, strings.TrimSpace(userID), strings.TrimSpace(workspaceID)) if err != nil { return ConversationItem{}, mapAccessError(err) } trimmedScopeID := strings.TrimSpace(scopeID) if trimmedScopeID == "" { return ConversationItem{}, badRequest("scopeId is required") } return txs.loadConversationForUser(ctx, userID, workspaceID, membership.Member.ID, trimmedScopeID, includeHidden) } func (store *Store) PerformConversationActionForUser(ctx context.Context, userID string, workspaceID string, conversationID string, action string) (ConversationItem, error) { var result ConversationItem err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error { var err error result, err = store.InTx(tx).PerformConversationActionForUser(ctx, userID, workspaceID, conversationID, action) return err }) if err != nil { return ConversationItem{}, err } return result, nil } func (txs TxStore) PerformConversationActionForUser(ctx context.Context, userID string, workspaceID string, conversationID string, action string) (ConversationItem, error) { membership, err := access.LoadHumanWorkspaceMember(ctx, txs.tx, strings.TrimSpace(userID), strings.TrimSpace(workspaceID)) if err != nil { return ConversationItem{}, mapAccessError(err) } expectedKind, scopeID, err := parseConversationActionID(conversationID) if err != nil { return ConversationItem{}, err } trimmedAction := strings.TrimSpace(action) current, err := txs.loadConversationForUser(ctx, userID, workspaceID, membership.Member.ID, scopeID, true) if err != nil { if trimmedAction != ConversationActionHide || !errors.Is(err, ErrNotFound) { return ConversationItem{}, err } current, err = txs.loadDeletedPeerConversationForHide(ctx, workspaceID, membership.Member.ID, scopeID) if err != nil { return ConversationItem{}, err } } if expectedKind != "" && current.Kind != expectedKind { return ConversationItem{}, ErrNotFound } now := time.Now().UTC() progress, err := txs.applyConversationAction( ctx, membership.Member.ID, workspaceID, scopeID, current.Kind, trimmedAction, now, ) if err != nil { return ConversationItem{}, err } item, err := txs.loadConversationForUser(ctx, userID, workspaceID, membership.Member.ID, scopeID, true) if err != nil { if trimmedAction == ConversationActionHide && errors.Is(err, ErrNotFound) { current.SidebarHidden = true item = current progress.applyTo(&item) return item, nil } return ConversationItem{}, err } progress.applyTo(&item) if trimmedAction == ConversationActionMarkRead || trimmedAction == ConversationActionMarkUnread { manualUnread, generation, err := loadConversationManualUnreadState(ctx, txs.tx, membership.Member.ID, scopeID) if err != nil { return ConversationItem{}, err } item.ManualUnread = manualUnread item.ManualUnreadGeneration = generation item.ManualUnreadGenerationKnown = true } return item, nil } type conversationActionProgress struct { lastReadSeq int64 lastActivityReadSeq int64 hiddenThroughSeq int64 hiddenThroughActivitySeq int64 } func (progress conversationActionProgress) applyTo(item *ConversationItem) { item.LastReadSeq = progress.lastReadSeq item.LastActivityReadSeq = progress.lastActivityReadSeq item.HiddenThroughSeq = progress.hiddenThroughSeq item.HiddenThroughActivitySeq = progress.hiddenThroughActivitySeq } func (txs TxStore) applyConversationAction( ctx context.Context, memberID string, workspaceID string, scopeID string, conversationKind string, action string, now time.Time, ) (conversationActionProgress, error) { var progress conversationActionProgress switch action { case ConversationActionPin: if err := upsertConversationPinnedState(ctx, txs.tx, memberID, scopeID, &now, now); err != nil { return progress, err } case ConversationActionUnpin: if err := upsertConversationPinnedState(ctx, txs.tx, memberID, scopeID, nil, now); err != nil { return progress, err } case ConversationActionMarkRead: latestSeq, latestGlobalSeq, err := loadLatestHumanVisibleConversationSeq(ctx, txs.tx, workspaceID, scopeID) if err != nil { return progress, err } progress.lastReadSeq = latestSeq if err := WriteThreadReadStateByGlobalSeq(ctx, txs.tx, memberID, scopeID, latestSeq, latestGlobalSeq, now); err != nil { return progress, fmt.Errorf("mark conversation read: %w", err) } progress.lastActivityReadSeq, err = loadLatestConversationActivitySeq(ctx, txs.tx, workspaceID, scopeID) if err != nil { return progress, err } if err := WriteActivityReadStateByGlobalSeq(ctx, txs.tx, memberID, scopeID, progress.lastActivityReadSeq, now); err != nil { return progress, fmt.Errorf("mark conversation activity read: %w", err) } if err := ClearConversationManualUnreadState(ctx, txs.tx, memberID, scopeID, now); err != nil { return progress, err } case ConversationActionMarkUnread: latestSeq, _, err := loadLatestHumanVisibleConversationSeq(ctx, txs.tx, workspaceID, scopeID) if err != nil { return progress, err } if err := upsertConversationManualUnreadState(ctx, txs.tx, memberID, scopeID, latestSeq, now); err != nil { return progress, err } case ConversationActionMute, ConversationActionUnmute: upsert := UpsertScopeNotificationState if conversationKind == ConversationKindTask { upsert = UpsertTaskWatchNotificationState } state := HumanScopeSubscriptionStateMute operation := "mute" if action == ConversationActionUnmute { state = HumanScopeSubscriptionStateActive operation = "unmute" } if err := upsert(ctx, txs.tx, memberID, scopeID, state, now); err != nil { return progress, fmt.Errorf("%s conversation: %w", operation, err) } case ConversationActionHide: latestSeq, latestGlobalSeq, latestActivitySeq, err := loadConversationLatestFrontiers(ctx, txs.tx, workspaceID, scopeID) if err != nil { return progress, err } progress.hiddenThroughSeq = latestSeq progress.hiddenThroughActivitySeq = latestActivitySeq if err := upsertConversationHiddenState(ctx, txs.tx, memberID, scopeID, latestSeq, latestActivitySeq, now); err != nil { return progress, err } // Hide dismisses all currently visible Task attention. Persist the same // two frontiers in the authoritative read cursors so a delayed outbox replay // or process restart cannot resurrect pre-hide unread work. if conversationKind == ConversationKindTask { if err := WriteThreadReadStateByGlobalSeq(ctx, txs.tx, memberID, scopeID, latestSeq, latestGlobalSeq, now); err != nil { return progress, fmt.Errorf("advance hidden Task message cursor: %w", err) } if err := WriteActivityReadStateByGlobalSeq(ctx, txs.tx, memberID, scopeID, latestActivitySeq, now); err != nil { return progress, fmt.Errorf("advance hidden Task Activity cursor: %w", err) } progress.lastReadSeq = latestSeq progress.lastActivityReadSeq = latestActivitySeq } default: return progress, badRequest("unsupported conversation action") } return progress, nil } func (txs TxStore) loadDeletedPeerConversationForHide(ctx context.Context, workspaceID string, memberID string, scopeID string) (ConversationItem, error) { row := txs.tx.QueryRow(ctx, ` SELECT s.id, s.type, CASE WHEN s.visible_last_message_at IS NOT NULL THEN s.visible_last_message_at WHEN s.latest_message_at IS NOT NULL THEN s.latest_message_at WHEN s.updated_at IS NOT NULL THEN s.updated_at ELSE s.created_at END AS sort_at, s.latest_message_at, cms.pinned_at, (cms.manual_unread_at IS NOT NULL) AS manual_unread, CASE WHEN sns.state IS NULL THEN '' ELSE sns.state END AS subscription_state, CASE WHEN cms.hidden_at IS NOT NULL AND CASE WHEN s.latest_seq IS NULL THEN 0 ELSE s.latest_seq END <= CASE WHEN cms.hidden_through_seq IS NULL THEN 0 ELSE cms.hidden_through_seq END THEN TRUE ELSE FALSE END AS sidebar_hidden, CASE WHEN s.type = 'service_account' THEN 'service' WHEN peer.type IS NULL THEN '' ELSE peer.type END AS peer_type, CASE WHEN s.type = 'service_account' THEN s.id WHEN peer.id IS NOT NULL THEN peer.id ELSE '' END AS peer_id, CASE WHEN NULLIF(peer.display_name, '') IS NOT NULL THEN NULLIF(peer.display_name, '') WHEN peer.id IS NOT NULL THEN peer.id ELSE s.id END AS title, '' AS subtitle, '' AS avatar_url FROM scopes s JOIN scope_memberships self_sm ON self_sm.scope_id = s.id AND self_sm.member_id = $2 AND self_sm.removed_at IS NULL AND self_sm.archived_at IS NULL AND self_sm.effective_to_seq IS NULL LEFT JOIN conversation_member_states cms ON cms.scope_id = s.id AND cms.member_id = $2 LEFT JOIN scope_notification_settings sns ON sns.scope_id = s.id AND sns.member_id = $2 LEFT JOIN scope_memberships peer_sm ON peer_sm.scope_id = s.id AND peer_sm.member_id <> $2 AND s.type = 'dm' LEFT JOIN members peer ON peer.id = peer_sm.member_id WHERE s.workspace_id = $1 AND s.id = $3 AND s.status = 'active' AND s.archived_at IS NULL AND ( s.type = 'dm' OR (s.type = 'service_account' AND self_sm.visibility = 'public') ) ORDER BY CASE WHEN peer_sm.removed_at IS NULL THEN 0 ELSE 1 END ASC, peer_sm.updated_at DESC LIMIT 1 `, strings.TrimSpace(workspaceID), strings.TrimSpace(memberID), strings.TrimSpace(scopeID)) var item ConversationItem var sortAt time.Time var lastMessageAt sql.NullTime var pinnedAt sql.NullTime if err := row.Scan( &item.ScopeID, &item.Kind, &sortAt, &lastMessageAt, &pinnedAt, &item.ManualUnread, &item.SubscriptionState, &item.SidebarHidden, &item.PeerType, &item.PeerID, &item.Title, &item.Subtitle, &item.AvatarURL, ); err != nil { if coredb.IsNotFound(err) { return ConversationItem{}, ErrNotFound } return ConversationItem{}, fmt.Errorf("load deleted peer conversation for hide: %w", err) } item.SortAt = sortAt.UTC() if lastMessageAt.Valid { value := lastMessageAt.Time.UTC() item.LastMessageAt = &value } lastActivityAt := item.SortAt item.LastActivityAt = &lastActivityAt if pinnedAt.Valid { value := pinnedAt.Time.UTC() item.PinnedAt = &value } if item.Kind == ConversationKindDM && strings.TrimSpace(item.PeerType) == "" { return ConversationItem{}, ErrNotFound } return item, nil } func parseConversationActionID(raw string) (string, string, error) { trimmed := strings.TrimSpace(raw) if trimmed == "" { return "", "", badRequest("conversationId is required") } for _, prefix := range []struct { prefix string kind string }{ {prefix: "channel:", kind: ConversationKindChannel}, {prefix: "thread:", kind: ConversationKindThread}, {prefix: "dm:", kind: ConversationKindDM}, {prefix: "task:", kind: ConversationKindTask}, {prefix: "serviceAccount:", kind: ConversationKindServiceAccount}, } { if strings.HasPrefix(trimmed, prefix.prefix) { scopeID := strings.TrimSpace(strings.TrimPrefix(trimmed, prefix.prefix)) if scopeID == "" { return "", "", badRequest("conversationId is required") } return prefix.kind, scopeID, nil } } return "", trimmed, nil } func upsertConversationPinnedState(ctx context.Context, execer CommandTagExecer, memberID string, scopeID string, pinnedAt *time.Time, updatedAt time.Time) error { _, err := execer.Exec(ctx, ` INSERT INTO conversation_member_states ( scope_id, member_id, pinned_at, manual_unread_at, manual_unread_seq, hidden_at, hidden_through_seq, created_at, updated_at ) VALUES ($1, $2, $3, NULL, NULL, NULL, NULL, $4, $4) ON DUPLICATE KEY UPDATE pinned_at = VALUES(pinned_at), updated_at = VALUES(updated_at) `, strings.TrimSpace(scopeID), strings.TrimSpace(memberID), pinnedAt, updatedAt.UTC()) if err != nil { return fmt.Errorf("upsert conversation pinned state: %w", err) } return nil } func upsertConversationManualUnreadState(ctx context.Context, execer CommandTagExecer, memberID string, scopeID string, manualUnreadSeq int64, updatedAt time.Time) error { _, err := execer.Exec(ctx, ` INSERT INTO conversation_member_states ( scope_id, member_id, pinned_at, manual_unread_at, manual_unread_seq, hidden_at, hidden_through_seq, unread_generation, created_at, updated_at ) VALUES ($1, $2, NULL, $4, $3, NULL, NULL, 1, $4, $4) ON DUPLICATE KEY UPDATE unread_generation = CASE WHEN manual_unread_at IS NULL OR manual_unread_seq IS NULL OR manual_unread_seq <> VALUES(manual_unread_seq) THEN unread_generation + 1 ELSE unread_generation END, updated_at = CASE WHEN manual_unread_at IS NULL OR manual_unread_seq IS NULL OR manual_unread_seq <> VALUES(manual_unread_seq) THEN VALUES(updated_at) ELSE updated_at END, manual_unread_at = CASE WHEN manual_unread_at IS NULL OR manual_unread_seq IS NULL OR manual_unread_seq <> VALUES(manual_unread_seq) THEN VALUES(manual_unread_at) ELSE manual_unread_at END, manual_unread_seq = CASE WHEN manual_unread_at IS NULL OR manual_unread_seq IS NULL OR manual_unread_seq <> VALUES(manual_unread_seq) THEN VALUES(manual_unread_seq) ELSE manual_unread_seq END `, strings.TrimSpace(scopeID), strings.TrimSpace(memberID), manualUnreadSeq, updatedAt.UTC()) if err != nil { return fmt.Errorf("upsert conversation manual unread state: %w", err) } return nil } // SetConversationManualUnreadState records the durable manual unread marker for // a member/scope pair after an explicit mark-unread action. func SetConversationManualUnreadState(ctx context.Context, execer CommandTagExecer, memberID string, scopeID string, manualUnreadSeq int64, updatedAt time.Time) error { return upsertConversationManualUnreadState(ctx, execer, memberID, scopeID, manualUnreadSeq, updatedAt) } func ClearConversationManualUnreadState(ctx context.Context, execer CommandTagExecer, memberID string, scopeID string, updatedAt time.Time) error { _, err := execer.Exec(ctx, ` UPDATE conversation_member_states SET manual_unread_at = NULL, manual_unread_seq = NULL, unread_generation = unread_generation + 1, updated_at = $3 WHERE member_id = $1 AND scope_id = $2 AND ( manual_unread_at IS NOT NULL OR manual_unread_seq IS NOT NULL ) `, strings.TrimSpace(memberID), strings.TrimSpace(scopeID), updatedAt.UTC()) if err != nil { return fmt.Errorf("clear conversation manual unread state: %w", err) } return nil } func loadConversationManualUnreadState(ctx context.Context, queryer coredb.QueryRower, memberID string, scopeID string) (bool, int64, error) { var manualUnread bool var generation int64 if err := queryer.QueryRow(ctx, ` SELECT (manual_unread_at IS NOT NULL), unread_generation FROM conversation_member_states WHERE member_id = $1 AND scope_id = $2 `, strings.TrimSpace(memberID), strings.TrimSpace(scopeID)).Scan(&manualUnread, &generation); err != nil { if coredb.IsNotFound(err) { return false, 0, nil } return false, 0, fmt.Errorf("load conversation manual unread state: %w", err) } if generation < 0 { return false, 0, fmt.Errorf("invalid conversation manual unread generation") } return manualUnread, generation, nil } func upsertConversationHiddenState( ctx context.Context, execer CommandTagExecer, memberID string, scopeID string, hiddenThroughSeq int64, hiddenThroughActivitySeq int64, updatedAt time.Time, ) error { _, err := execer.Exec(ctx, ` INSERT INTO conversation_member_states ( scope_id, member_id, pinned_at, manual_unread_at, manual_unread_seq, hidden_at, hidden_through_seq, hidden_through_activity_seq, created_at, updated_at ) VALUES ($1, $2, NULL, NULL, NULL, $5, $3, $4, $5, $5) ON DUPLICATE KEY UPDATE hidden_at = VALUES(hidden_at), hidden_through_seq = VALUES(hidden_through_seq), hidden_through_activity_seq = VALUES(hidden_through_activity_seq), updated_at = VALUES(updated_at) `, strings.TrimSpace(scopeID), strings.TrimSpace(memberID), hiddenThroughSeq, hiddenThroughActivitySeq, updatedAt.UTC(), ) if err != nil { return fmt.Errorf("upsert conversation hidden state: %w", err) } return nil } func loadConversationLatestFrontiers(ctx context.Context, queryer coredb.QueryRower, workspaceID string, scopeID string) (int64, int64, int64, error) { row := queryer.QueryRow(ctx, ` SELECT latest_seq, latest_global_seq, latest_activity_seq FROM scopes WHERE workspace_id = $1 AND id = $2 AND status = 'active' `, strings.TrimSpace(workspaceID), strings.TrimSpace(scopeID)) var latestSeq int64 var latestGlobalSeq int64 var latestActivitySeq int64 if err := row.Scan(&latestSeq, &latestGlobalSeq, &latestActivitySeq); err != nil { if coredb.IsNotFound(err) { return 0, 0, 0, ErrNotFound } return 0, 0, 0, fmt.Errorf("load conversation latest frontiers: %w", err) } if latestSeq < 0 || latestGlobalSeq < 0 || latestActivitySeq < 0 { return 0, 0, 0, fmt.Errorf("invalid conversation latest frontiers") } return latestSeq, latestGlobalSeq, latestActivitySeq, nil } func loadLatestConversationActivitySeq(ctx context.Context, queryer coredb.QueryRower, workspaceID string, scopeID string) (int64, error) { row := queryer.QueryRow(ctx, ` SELECT seq FROM scope_activity_events WHERE workspace_id = $1 AND scope_id = $2 AND suppressed_at IS NULL ORDER BY seq DESC LIMIT 1 `, strings.TrimSpace(workspaceID), strings.TrimSpace(scopeID)) var latestGlobalSeq int64 if err := row.Scan(&latestGlobalSeq); err != nil { if coredb.IsNotFound(err) { return 0, nil } return 0, fmt.Errorf("load latest conversation activity seq: %w", err) } if latestGlobalSeq < 0 { return 0, fmt.Errorf("invalid latest conversation activity seq") } return latestGlobalSeq, nil } func loadLatestHumanVisibleConversationSeq(ctx context.Context, queryer coredb.QueryRower, workspaceID string, scopeID string) (int64, int64, error) { row := queryer.QueryRow(ctx, fmt.Sprintf(` SELECT seq, workspace_global_seq FROM scope_messages WHERE workspace_id = $1 AND scope_id = $2 AND %s ORDER BY seq DESC LIMIT 1 `, HumanVisibleMessageTypeSQL("scope_messages")), strings.TrimSpace(workspaceID), strings.TrimSpace(scopeID)) var latestSeq int64 var latestGlobalSeq int64 if err := row.Scan(&latestSeq, &latestGlobalSeq); err != nil { if coredb.IsNotFound(err) { return 0, 0, nil } return 0, 0, fmt.Errorf("load latest visible conversation seq: %w", err) } if latestSeq < 0 || latestGlobalSeq < 0 { return 0, 0, fmt.Errorf("invalid latest visible conversation seq") } return latestSeq, latestGlobalSeq, nil } func (txs TxStore) loadConversationForUser(ctx context.Context, userID string, workspaceID string, memberID string, scopeID string, includeHidden bool) (ConversationItem, error) { taskItem, err := txs.loadTaskConversationForMember(ctx, workspaceID, memberID, scopeID, includeHidden) if err == nil { return taskItem, nil } if !errors.Is(err, ErrNotFound) { return ConversationItem{}, err } query, args, err := BuildConversationForUserScopedQuery(workspaceID, memberID, userID, scopeID, includeHidden) if err != nil { return ConversationItem{}, err } rows, err := txs.tx.Query(ctx, query, args...) if err != nil { return ConversationItem{}, fmt.Errorf("load conversation: %w", err) } defer rows.Close() if !rows.Next() { if err := rows.Err(); err != nil { return ConversationItem{}, fmt.Errorf("iterate conversation row: %w", err) } return ConversationItem{}, ErrNotFound } item, err := scanConversationItem(rows) if err != nil { return ConversationItem{}, err } if rows.Next() { return ConversationItem{}, fmt.Errorf("load conversation returned multiple rows") } if err := rows.Err(); err != nil { return ConversationItem{}, fmt.Errorf("iterate conversation row: %w", err) } return item, nil } func (txs TxStore) loadTaskConversationForMember(ctx context.Context, workspaceID string, memberID string, scopeID string, includeHidden bool) (ConversationItem, error) { row := txs.tx.QueryRow(ctx, taskConversationForMemberQuery(), strings.TrimSpace(workspaceID), strings.TrimSpace(memberID), strings.TrimSpace(scopeID), includeHidden) var task ConversationTaskMetadata var sortAt time.Time var pinnedAt sql.NullTime var sidebarHidden bool var projectionRaw []byte var subscriptionState string var assigneeMemberType string var assigneeMemberID string var assigneeUserID string var assigneeUserName string var assigneeUserDisplayName string var assigneeUserAvatarURL string var assigneeAgentID string var assigneeAgentName string var assigneeAgentAvatarURL string var assigneeMemberDisplayName string if err := row.Scan( &task.ID, &task.ScopeID, &task.ChannelID, &task.Number, &task.Title, &task.Status, &task.ParentTaskRef, &task.SourceThreadID, &task.SourceThreadParentChannelID, &task.SourceThreadParentMessageID, &task.SourceThreadRef, &assigneeMemberType, &assigneeMemberID, &assigneeUserID, &assigneeUserName, &assigneeUserDisplayName, &assigneeUserAvatarURL, &assigneeAgentID, &assigneeAgentName, &assigneeAgentAvatarURL, &assigneeMemberDisplayName, &sortAt, &pinnedAt, &sidebarHidden, &projectionRaw, &subscriptionState, ); err != nil { if coredb.IsNotFound(err) { return ConversationItem{}, ErrNotFound } return ConversationItem{}, fmt.Errorf("load task conversation: %w", err) } projection, err := model.UnmarshalTaskConversationProjection(projectionRaw) if err != nil { return ConversationItem{}, err } task.AssigneeProfile = taskConversationAssigneeProfile( assigneeMemberType, assigneeMemberID, assigneeUserID, assigneeUserName, assigneeUserDisplayName, assigneeUserAvatarURL, assigneeAgentID, assigneeAgentName, assigneeAgentAvatarURL, assigneeMemberDisplayName, ) task.Route = "/channels/" + url.PathEscape(strings.TrimSpace(task.ChannelID)) + "/tasks/" + strconv.FormatInt(task.Number, 10) title := fmt.Sprintf("#t%d", task.Number) if taskTitle := strings.TrimSpace(task.Title); taskTitle != "" { title += " " + taskTitle } sortAt = sortAt.UTC() lastActivityAt := sortAt item := ConversationItem{ Kind: ConversationKindTask, ScopeID: strings.TrimSpace(task.ScopeID), Title: title, Preview: strings.TrimSpace(projection.Preview), LastAuthorName: strings.TrimSpace(projection.LastAuthorName), LastActivityAt: &lastActivityAt, SubscriptionState: HumanScopeSubscriptionStateMute, SortAt: sortAt, Task: &task, } if strings.TrimSpace(subscriptionState) == HumanScopeSubscriptionStateActive { item.SubscriptionState = HumanScopeSubscriptionStateActive } if pinnedAt.Valid { value := pinnedAt.Time.UTC() item.PinnedAt = &value } if sidebarHidden { item.Archived = true item.SidebarHidden = true } return item, nil } // taskConversationAssigneeProfile derives the current Task responsibility // profile from authoritative membership rows instead of the event-time Inbox // projection. The projection intentionally keeps preview/sort facts frozen, // but assignee identity must reflect a later reassignment for every member who // already has the aggregated Task row, including the mutation actor. func taskConversationAssigneeProfile( memberType string, memberID string, userID string, userName string, userDisplayName string, userAvatarURL string, agentID string, agentName string, agentAvatarURL string, memberDisplayName string, ) *AuthorProfile { memberType = strings.TrimSpace(memberType) memberID = strings.TrimSpace(memberID) if memberType == "" || memberID == "" { return nil } profile := AuthorProfile{ ActorType: memberType, MemberID: memberID, } switch memberType { case AuthorActorTypeHuman: profile.PrincipalID = strings.TrimSpace(userID) profile.TargetID = firstNonEmptyConversationString(userName, memberID) profile.DisplayName = firstNonEmptyConversationString(userDisplayName, memberDisplayName, userName, memberID) profile.AvatarURL = strings.TrimSpace(userAvatarURL) case AuthorActorTypeAgent: profile.PrincipalID = strings.TrimSpace(agentID) profile.TargetID = firstNonEmptyConversationString(agentName, memberID) profile.DisplayName = firstNonEmptyConversationString(memberDisplayName, agentName, memberID) profile.AvatarURL = strings.TrimSpace(agentAvatarURL) default: return nil } profile = profile.Normalize() if !profile.Valid() { return nil } return &profile } func firstNonEmptyConversationString(values ...string) string { for _, value := range values { if value = strings.TrimSpace(value); value != "" { return value } } return "" } func scanConversationItem(rows coredb.Rows) (ConversationItem, error) { var item ConversationItem var sortAt time.Time if err := rows.Scan( &item.ScopeID, &item.Kind, &item.ParentScopeID, &item.ParentScopeType, &item.ParentTitle, &item.ParentMessageID, &item.ParentMessageShortID, &item.ParentMessageAuthor, &item.Title, &item.Subtitle, &item.AvatarURL, &item.Subsystem, &item.TeamID, &item.TeamHandle, &item.TeamDisplayName, &item.PeerType, &item.PeerID, &item.Preview, &item.LastAuthorName, &item.LastMessageAt, &sortAt, &item.PinnedAt, &item.UnreadCount, &item.ManualUnread, &item.SubscriptionState, &item.SidebarHidden, &item.Archived, &item.ReplyCount, &item.LastReplyAt, ); err != nil { return ConversationItem{}, fmt.Errorf("scan conversation row: %w", err) } item.SortAt = sortAt.UTC() lastActivityAt := item.SortAt item.LastActivityAt = &lastActivityAt if item.PinnedAt != nil { normalizedPinnedAt := item.PinnedAt.UTC() item.PinnedAt = &normalizedPinnedAt } return item, nil } func BuildListConversationsForUserScopedQuery(workspaceID string, memberID string, userID string, limit int, cursor ConversationCursor) (string, []any, error) { return BuildListConversationsForUserScopedQueryFiltered(workspaceID, memberID, userID, limit, cursor, ConversationKindFilter{}) } func BuildListConversationsForUserScopedQueryFiltered(workspaceID string, memberID string, userID string, limit int, cursor ConversationCursor, filter ConversationKindFilter) (string, []any, error) { if limit <= 0 { return "", nil, fmt.Errorf("invalid limit") } normalizedFilter, err := NormalizeConversationKindFilter(filter) if err != nil { return "", nil, err } args := conversationBaseArgs(workspaceID, memberID, userID) kindClause, args := appendConversationKindFilterClause(args, normalizedFilter, "kind") kindSelection := conversationKindSelectionFromFilter(normalizedFilter) cursorClause := "" if !cursor.IsZero() { var err error cursorClause, args, err = appendConversationCursorClauses(args, cursor) if err != nil { return "", nil, err } } limitArg := len(args) + 1 args = append(args, limit+1) visibleConversationRefsCTE := buildBoundedVisibleConversationRefsCTE(kindSelection, cursorClause, filter.subsystemWhereClause(), limitArg) pageCTE := fmt.Sprintf(` visible_conversations AS ( SELECT base_conversations.* FROM base_conversations WHERE sidebar_hidden = FALSE%s ), page_conversations AS ( SELECT visible_conversations.* FROM visible_conversations %s ORDER BY pinned_rank ASC, sort_at DESC, scope_id ASC LIMIT $%d )`, kindClause, cursorClause, limitArg) query := buildConversationRowsFromVisibleConversationRefsQuery(visibleConversationRefsCTE, pageCTE, filter.subsystemWhereClause()) return query, args, nil } func BuildVisibleConversationScopeIDsForUserScopedQueryFiltered(workspaceID string, memberID string, userID string, filter ConversationKindFilter) (string, []any, error) { normalizedFilter, err := NormalizeConversationKindFilter(filter) if err != nil { return "", nil, err } args := conversationBaseArgs(workspaceID, memberID, userID) kindClause, args := appendConversationKindFilterClause(args, normalizedFilter, "kind") kindSelection := conversationKindSelectionFromFilter(normalizedFilter) query := buildVisibleConversationScopeIDsForUserScopedQuery(kindSelection, kindClause, filter.subsystemWhereClause()) return query, args, nil } func BuildVisibleConversationScopeIDsForMemberPageQuery(workspaceID string, memberID string, userID string, afterScopeID string, limit int) (string, []any, error) { if limit <= 0 { return "", nil, badRequest("limit must be positive") } query, args, err := BuildVisibleConversationScopeIDsForUserScopedQueryFiltered( workspaceID, memberID, userID, ConversationKindFilter{}, ) if err != nil { return "", nil, err } args = append(args, strings.TrimSpace(afterScopeID), limit) return fmt.Sprintf(` SELECT visible_inbox_scopes.scope_id FROM ( %s ) visible_inbox_scopes WHERE visible_inbox_scopes.scope_id > $%d ORDER BY visible_inbox_scopes.scope_id LIMIT $%d `, strings.TrimSpace(query), len(args)-1, len(args)), args, nil } func BuildConversationForUserScopedQuery(workspaceID string, memberID string, userID string, scopeID string, includeHidden bool) (string, []any, error) { trimmedScopeID := strings.TrimSpace(scopeID) if trimmedScopeID == "" { return "", nil, badRequest("conversationId is required") } args := conversationBaseArgs(workspaceID, memberID, userID) args[3] = trimmedScopeID scopeIDArg := len(args) + 1 includeHiddenArg := scopeIDArg + 1 args = append(args, trimmedScopeID, includeHidden) pageCTE := fmt.Sprintf(` page_conversations AS ( SELECT base_conversations.* FROM base_conversations WHERE scope_id = $%d AND ($%d OR sidebar_hidden = FALSE) LIMIT 1 )`, scopeIDArg, includeHiddenArg) query := buildConversationRowsForUserScopedQuery(pageCTE, trimmedScopeID, "AND (s.type <> 'channel' OR cp.subsystem IN ('im', 'team', 'tutorial'))") return query, args, nil } func conversationBaseArgs(workspaceID string, memberID string, userID string) []any { return []any{strings.TrimSpace(workspaceID), strings.TrimSpace(memberID), strings.TrimSpace(userID), ""} } func appendConversationCursorClauses(args []any, cursor ConversationCursor) (string, []any, error) { normalizedCursor, err := normalizeConversationCursor(cursor) if err != nil { return "", nil, err } if normalizedCursor.PinnedAt != nil { sortAtArg := len(args) + 1 scopeIDArg := sortAtArg + 1 args = append(args, normalizedCursor.SortAt, normalizedCursor.ScopeID) pageCursorClause := fmt.Sprintf(` WHERE ( pinned_rank > 0 OR ( pinned_rank = 0 AND (sort_at < $%d OR (sort_at = $%d AND scope_id > $%d)) ) )`, sortAtArg, sortAtArg, scopeIDArg) return pageCursorClause, args, nil } sortAtArg := len(args) + 1 scopeIDArg := sortAtArg + 1 args = append(args, normalizedCursor.SortAt, normalizedCursor.ScopeID) pageCursorClause := fmt.Sprintf(` WHERE ( pinned_rank = 1 AND (sort_at < $%d OR (sort_at = $%d AND scope_id > $%d)) )`, sortAtArg, sortAtArg, scopeIDArg) return pageCursorClause, args, nil } func appendConversationKindFilterClause(args []any, filter ConversationKindFilter, column string) (string, []any) { if len(filter.Kinds) == 0 { return "", args } placeholders := make([]string, 0, len(filter.Kinds)) for _, kind := range filter.Kinds { args = append(args, kind) placeholders = append(placeholders, fmt.Sprintf("$%d", len(args))) } return fmt.Sprintf("\n AND %s IN (%s)", column, strings.Join(placeholders, ", ")), args } func buildBoundedVisibleConversationRefsCTE(selection conversationKindSelection, cursorClause string, channelSubsystemClause string, limitArg int) string { ctes := make([]string, 0, 4) if selection.channel { ctes = append(ctes, buildBoundedChannelConversationRefsCTE(cursorClause, channelSubsystemClause, limitArg)) } if selection.dm { ctes = append(ctes, buildBoundedDMConversationRefsCTE(cursorClause, limitArg)) } if selection.thread { ctes = append(ctes, buildBoundedThreadConversationRefsCTE(cursorClause, limitArg)) } if selection.serviceAccount { ctes = append(ctes, buildBoundedServiceAccountConversationRefsCTE(cursorClause, limitArg)) } refSelects := make([]string, 0, 4) if selection.channel { refSelects = append(refSelects, "SELECT scope_id FROM channel_conversation_refs") } if selection.dm { refSelects = append(refSelects, "SELECT scope_id FROM dm_conversation_refs") } if selection.thread { refSelects = append(refSelects, "SELECT scope_id FROM thread_conversation_refs") } if selection.serviceAccount { refSelects = append(refSelects, "SELECT scope_id FROM service_account_conversation_refs") } ctes = append(ctes, buildVisibleConversationRefsCTE(refSelects)) return strings.Join(ctes, ",\n") } func buildVisibleConversationRefsCTE(refSelects []string) string { if len(refSelects) == 0 { return `visible_conversation_refs AS ( SELECT '' AS scope_id WHERE 1 = 0 )` } return fmt.Sprintf(`visible_conversation_refs AS ( %s )`, strings.Join(refSelects, "\n\n UNION\n\n ")) } func buildBoundedChannelConversationRefsCTE(cursorClause string, channelSubsystemClause string, limitArg int) string { return fmt.Sprintf(`channel_conversation_candidate_rows AS ( SELECT s.id AS scope_id, CASE WHEN cms.pinned_at IS NULL THEN 1 ELSE 0 END AS pinned_rank, cms.pinned_at, CASE WHEN s.visible_last_message_at IS NOT NULL THEN s.visible_last_message_at WHEN s.latest_message_at IS NOT NULL THEN s.latest_message_at WHEN s.updated_at IS NOT NULL THEN s.updated_at ELSE s.created_at END AS sort_at FROM scope_memberships sm JOIN scopes s ON s.id = sm.scope_id AND s.workspace_id = $1 AND s.type = 'channel' AND s.status = 'active' AND s.archived_at IS NULL JOIN scope_channel_profiles cp ON cp.scope_id = s.id AND cp.workspace_id = $1 LEFT JOIN conversation_member_states cms ON cms.scope_id = s.id AND cms.member_id = $2 WHERE sm.member_id = $2 AND sm.removed_at IS NULL AND sm.archived_at IS NULL AND sm.effective_to_seq IS NULL AND sm.visibility = 'public' AND NOT ( cms.hidden_at IS NOT NULL AND CASE WHEN s.latest_seq IS NULL THEN 0 ELSE s.latest_seq END <= CASE WHEN cms.hidden_through_seq IS NULL THEN 0 ELSE cms.hidden_through_seq END ) %s ), channel_conversation_refs AS ( SELECT scope_id FROM channel_conversation_candidate_rows %s ORDER BY pinned_rank ASC, sort_at DESC, scope_id ASC LIMIT $%d )`, channelSubsystemClause, cursorClause, limitArg) } func buildBoundedDMConversationRefsCTE(cursorClause string, limitArg int) string { return fmt.Sprintf(`dm_conversation_candidate_rows AS ( SELECT s.id AS scope_id, CASE WHEN cms.pinned_at IS NULL THEN 1 ELSE 0 END AS pinned_rank, cms.pinned_at, CASE WHEN s.visible_last_message_at IS NOT NULL THEN s.visible_last_message_at WHEN s.latest_message_at IS NOT NULL THEN s.latest_message_at WHEN s.updated_at IS NOT NULL THEN s.updated_at ELSE s.created_at END AS sort_at FROM scope_memberships self_sm JOIN scopes s ON s.id = self_sm.scope_id AND s.workspace_id = $1 AND s.type = 'dm' AND s.status = 'active' AND s.archived_at IS NULL JOIN scope_memberships peer_sm ON peer_sm.scope_id = s.id AND peer_sm.member_id <> $2 AND peer_sm.removed_at IS NULL AND peer_sm.archived_at IS NULL AND peer_sm.effective_to_seq IS NULL JOIN members peer ON peer.id = peer_sm.member_id LEFT JOIN users peer_user ON peer.type = 'human' AND peer_user.id = peer.user_id LEFT JOIN agents peer_agent ON peer.type = 'agent' AND peer_agent.member_id = peer.id LEFT JOIN service_members peer_service ON peer.type = 'service' AND peer_service.workspace_id = s.workspace_id AND peer_service.member_id = peer.id LEFT JOIN conversation_member_states cms ON cms.scope_id = s.id AND cms.member_id = $2 LEFT JOIN dm_view_states dvs ON dvs.dm_thread_id = s.id AND dvs.member_id = $2 WHERE self_sm.member_id = $2 AND self_sm.removed_at IS NULL AND self_sm.archived_at IS NULL AND self_sm.effective_to_seq IS NULL AND ( (peer.type = 'human' AND peer_user.id IS NOT NULL AND peer_user.id <> '') OR (peer.type = 'agent' AND peer_agent.id IS NOT NULL AND peer_agent.id <> '') OR (peer.type = 'service' AND peer_service.id IS NOT NULL AND peer_service.id <> '') ) AND NOT ( cms.hidden_at IS NOT NULL AND CASE WHEN s.latest_seq IS NULL THEN 0 ELSE s.latest_seq END <= CASE WHEN cms.hidden_through_seq IS NULL THEN 0 ELSE cms.hidden_through_seq END ) AND NOT ( CASE WHEN dvs.hidden_through_seq IS NULL THEN 0 ELSE dvs.hidden_through_seq END > 0 AND CASE WHEN s.latest_seq IS NULL THEN 0 ELSE s.latest_seq END <= dvs.hidden_through_seq ) AND NOT EXISTS ( SELECT 1 FROM agents a JOIN machines mac ON mac.workspace_id = a.workspace_id AND mac.id = a.machine_id WHERE a.workspace_id = $1 AND a.member_id = peer_sm.member_id AND mac.created_by_user_id <> $3 ) ), dm_conversation_refs AS ( SELECT scope_id FROM dm_conversation_candidate_rows %s ORDER BY pinned_rank ASC, sort_at DESC, scope_id ASC LIMIT $%d )`, cursorClause, limitArg) } func buildBoundedThreadConversationRefsCTE(cursorClause string, limitArg int) string { return fmt.Sprintf(`thread_conversation_candidate_rows AS ( SELECT s.id AS scope_id, CASE WHEN cms.pinned_at IS NULL THEN 1 ELSE 0 END AS pinned_rank, cms.pinned_at, CASE WHEN s.visible_last_message_at IS NOT NULL THEN s.visible_last_message_at WHEN s.latest_message_at IS NOT NULL THEN s.latest_message_at WHEN s.updated_at IS NOT NULL THEN s.updated_at ELSE s.created_at END AS sort_at FROM scope_notification_settings sns JOIN scopes s ON s.id = sns.scope_id AND s.workspace_id = $1 AND s.type = 'thread' AND s.status = 'active' AND s.archived_at IS NULL JOIN scopes parent_scope ON parent_scope.id = s.parent_scope_id AND parent_scope.workspace_id = $1 AND parent_scope.status = 'active' AND parent_scope.archived_at IS NULL LEFT JOIN conversation_member_states cms ON cms.scope_id = s.id AND cms.member_id = $2 WHERE sns.member_id = $2 AND sns.state IN ('active', 'quiet', 'mute') AND ( (parent_scope.type = 'channel' AND EXISTS ( SELECT 1 FROM scope_memberships parent_sm WHERE parent_sm.scope_id = parent_scope.id AND parent_sm.member_id = $2 AND parent_sm.removed_at IS NULL AND parent_sm.archived_at IS NULL AND parent_sm.effective_to_seq IS NULL AND parent_sm.visibility = 'public' )) OR (parent_scope.type = 'dm' AND ( EXISTS ( SELECT 1 FROM scope_memberships child_sm WHERE child_sm.scope_id = s.id AND child_sm.member_id = $2 AND child_sm.removed_at IS NULL AND child_sm.archived_at IS NULL AND child_sm.effective_to_seq IS NULL ) OR EXISTS ( SELECT 1 FROM scope_memberships parent_sm WHERE parent_sm.scope_id = parent_scope.id AND parent_sm.member_id = $2 AND parent_sm.removed_at IS NULL AND parent_sm.archived_at IS NULL AND parent_sm.effective_to_seq IS NULL ) )) ) AND NOT ( cms.hidden_at IS NOT NULL AND CASE WHEN s.latest_seq IS NULL THEN 0 ELSE s.latest_seq END <= CASE WHEN cms.hidden_through_seq IS NULL THEN 0 ELSE cms.hidden_through_seq END ) AND (parent_scope.type <> 'dm' OR NOT EXISTS ( SELECT 1 FROM scope_memberships parent_peer JOIN agents a ON a.workspace_id = $1 AND a.member_id = parent_peer.member_id JOIN machines mac ON mac.workspace_id = a.workspace_id AND mac.id = a.machine_id WHERE parent_peer.scope_id = parent_scope.id AND parent_peer.member_id <> $2 AND parent_peer.removed_at IS NULL AND parent_peer.archived_at IS NULL AND parent_peer.effective_to_seq IS NULL AND mac.created_by_user_id <> $3 )) ), thread_conversation_refs AS ( SELECT scope_id FROM thread_conversation_candidate_rows %s ORDER BY pinned_rank ASC, sort_at DESC, scope_id ASC LIMIT $%d )`, cursorClause, limitArg) } func buildBoundedServiceAccountConversationRefsCTE(cursorClause string, limitArg int) string { return fmt.Sprintf(`service_account_conversation_candidate_rows AS ( SELECT s.id AS scope_id, CASE WHEN cms.pinned_at IS NULL THEN 1 ELSE 0 END AS pinned_rank, cms.pinned_at, CASE WHEN s.visible_last_message_at IS NOT NULL THEN s.visible_last_message_at WHEN s.latest_message_at IS NOT NULL THEN s.latest_message_at WHEN s.updated_at IS NOT NULL THEN s.updated_at ELSE s.created_at END AS sort_at FROM scope_memberships sm JOIN scopes s ON s.id = sm.scope_id AND s.workspace_id = $1 AND s.type = 'service_account' AND s.status = 'active' AND s.archived_at IS NULL LEFT JOIN conversation_member_states cms ON cms.scope_id = s.id AND cms.member_id = $2 WHERE sm.member_id = $2 AND sm.removed_at IS NULL AND sm.archived_at IS NULL AND sm.effective_to_seq IS NULL AND sm.visibility = 'public' AND NOT ( cms.hidden_at IS NOT NULL AND CASE WHEN s.latest_seq IS NULL THEN 0 ELSE s.latest_seq END <= CASE WHEN cms.hidden_through_seq IS NULL THEN 0 ELSE cms.hidden_through_seq END ) ), service_account_conversation_refs AS ( SELECT scope_id FROM service_account_conversation_candidate_rows %s ORDER BY pinned_rank ASC, sort_at DESC, scope_id ASC LIMIT $%d )`, cursorClause, limitArg) } func buildConversationRowsForUserScopedQuery(pageConversationsCTE string, requestedScopeID string, subsystemWhereClause string) string { visibleThreadRefsCTE := BuildVisibleThreadRefsSQLCTEForRequest("visible_thread_refs", 1, 2, 3, 4, requestedScopeID) visibleConversationRefsCTE := fmt.Sprintf(`%s, visible_conversation_refs AS ( SELECT channel_id AS scope_id FROM visible_thread_refs )`, visibleThreadRefsCTE) return buildConversationRowsFromVisibleConversationRefsQuery(visibleConversationRefsCTE, pageConversationsCTE, subsystemWhereClause) } func buildConversationRowsFromVisibleConversationRefsQuery(visibleConversationRefsCTE string, pageConversationsCTE string, subsystemWhereClause string) string { return fmt.Sprintf(` WITH %s, base_conversations AS ( SELECT s.id AS scope_id, s.type AS kind, CASE WHEN parent_scope.id IS NULL THEN '' ELSE parent_scope.id END AS parent_scope_id, CASE WHEN parent_cp.subsystem = 'kb' THEN 'kb' ELSE CASE WHEN parent_scope.type IS NULL THEN '' ELSE parent_scope.type END END AS parent_scope_type, CASE WHEN parent_scope.type = 'channel' THEN CASE WHEN NULLIF(parent_cp.display_name, '') IS NOT NULL THEN NULLIF(parent_cp.display_name, '') WHEN NULLIF(parent_cp.slug, '') IS NOT NULL THEN NULLIF(parent_cp.slug, '') ELSE parent_scope.id END WHEN parent_peer.type = 'agent' THEN CASE WHEN NULLIF(parent_peer.display_name, '') IS NOT NULL THEN NULLIF(parent_peer.display_name, '') WHEN NULLIF(parent_peer_agent.name, '') IS NOT NULL THEN NULLIF(parent_peer_agent.name, '') ELSE parent_peer.id END WHEN parent_peer.type = 'human' THEN CASE WHEN NULLIF(parent_peer_user.display_name, '') IS NOT NULL THEN NULLIF(parent_peer_user.display_name, '') WHEN NULLIF(parent_peer.display_name, '') IS NOT NULL THEN NULLIF(parent_peer.display_name, '') WHEN parent_peer_user.name IS NOT NULL THEN parent_peer_user.name ELSE parent_peer.id END WHEN parent_peer.type = 'service' THEN CASE WHEN NULLIF(parent_peer_service.display_name, '') IS NOT NULL THEN NULLIF(parent_peer_service.display_name, '') WHEN NULLIF(parent_peer.display_name, '') IS NOT NULL THEN NULLIF(parent_peer.display_name, '') WHEN NULLIF(parent_peer_service_definition.name, '') IS NOT NULL THEN NULLIF(parent_peer_service_definition.name, '') WHEN NULLIF(parent_peer_service.handle, '') IS NOT NULL THEN NULLIF(parent_peer_service.handle, '') ELSE parent_peer.id END ELSE '' END AS parent_title, CASE WHEN s.root_message_id IS NULL THEN '' ELSE s.root_message_id END AS parent_message_id, CASE WHEN s.type = 'thread' THEN CASE WHEN s.root_message_short_id IS NULL THEN '' ELSE s.root_message_short_id END ELSE '' END AS parent_message_short_id, CASE WHEN root_message.content IS NULL THEN '' ELSE root_message.content END AS root_message_preview, CASE WHEN root_message.author_type = 'agent' THEN CASE WHEN NULLIF(root_author.display_name, '') IS NOT NULL THEN NULLIF(root_author.display_name, '') WHEN NULLIF(root_author_agent.name, '') IS NOT NULL THEN NULLIF(root_author_agent.name, '') WHEN root_author.id IS NOT NULL THEN root_author.id ELSE '' END WHEN root_message.author_type = 'human' THEN CASE WHEN NULLIF(root_author_user.display_name, '') IS NOT NULL THEN NULLIF(root_author_user.display_name, '') WHEN NULLIF(root_author.display_name, '') IS NOT NULL THEN NULLIF(root_author.display_name, '') WHEN root_author_user.name IS NOT NULL THEN root_author_user.name WHEN root_author.id IS NOT NULL THEN root_author.id ELSE '' END WHEN root_message.author_type = 'service' THEN CASE WHEN NULLIF(root_author.display_name, '') IS NOT NULL THEN NULLIF(root_author.display_name, '') WHEN root_author.id IS NOT NULL THEN root_author.id ELSE '' END ELSE '' END AS parent_message_author, CASE WHEN s.type = 'channel' THEN CASE WHEN NULLIF(cp.display_name, '') IS NOT NULL THEN NULLIF(cp.display_name, '') WHEN NULLIF(cp.slug, '') IS NOT NULL THEN NULLIF(cp.slug, '') ELSE s.id END WHEN s.type = 'service_account' THEN CASE WHEN NULLIF(service_account.name, '') IS NOT NULL THEN NULLIF(service_account.name, '') WHEN NULLIF(service_account_member.display_name, '') IS NOT NULL THEN NULLIF(service_account_member.display_name, '') ELSE s.id END WHEN s.type = 'dm' AND dm_peer.type = 'agent' THEN CASE WHEN NULLIF(dm_peer.display_name, '') IS NOT NULL THEN NULLIF(dm_peer.display_name, '') WHEN NULLIF(dm_peer_agent.name, '') IS NOT NULL THEN NULLIF(dm_peer_agent.name, '') ELSE dm_peer.id END WHEN s.type = 'dm' AND dm_peer.type = 'human' THEN CASE WHEN NULLIF(dm_peer_user.display_name, '') IS NOT NULL THEN NULLIF(dm_peer_user.display_name, '') WHEN NULLIF(dm_peer.display_name, '') IS NOT NULL THEN NULLIF(dm_peer.display_name, '') WHEN dm_peer_user.name IS NOT NULL THEN dm_peer_user.name ELSE dm_peer.id END WHEN s.type = 'dm' AND dm_peer.type = 'service' THEN CASE WHEN NULLIF(dm_peer_service.display_name, '') IS NOT NULL THEN NULLIF(dm_peer_service.display_name, '') WHEN NULLIF(dm_peer.display_name, '') IS NOT NULL THEN NULLIF(dm_peer.display_name, '') WHEN NULLIF(dm_peer_service_definition.name, '') IS NOT NULL THEN NULLIF(dm_peer_service_definition.name, '') WHEN NULLIF(dm_peer_service.handle, '') IS NOT NULL THEN NULLIF(dm_peer_service.handle, '') ELSE dm_peer.id END WHEN s.type = 'thread' THEN CASE WHEN NULLIF(root_message.content, '') IS NOT NULL THEN NULLIF(root_message.content, '') WHEN NULLIF(parent_cp.display_name, '') IS NOT NULL THEN NULLIF(parent_cp.display_name, '') WHEN NULLIF(parent_cp.slug, '') IS NOT NULL THEN NULLIF(parent_cp.slug, '') WHEN parent_scope.id IS NOT NULL THEN parent_scope.id ELSE s.id END ELSE s.id END AS title, CASE WHEN s.type = 'channel' THEN CASE WHEN NULLIF(cp.slug, '') IS NULL THEN '' ELSE NULLIF(cp.slug, '') END WHEN s.type = 'service_account' THEN '' WHEN s.type = 'dm' AND dm_peer.type = 'service' THEN CASE WHEN NULLIF(dm_peer_service.handle, '') IS NOT NULL THEN NULLIF(dm_peer_service.handle, '') ELSE dm_peer.id END WHEN s.type = 'thread' THEN CASE WHEN parent_scope.type = 'channel' THEN CASE WHEN NULLIF(parent_cp.display_name, '') IS NOT NULL THEN NULLIF(parent_cp.display_name, '') WHEN NULLIF(parent_cp.slug, '') IS NOT NULL THEN NULLIF(parent_cp.slug, '') ELSE parent_scope.id END WHEN parent_peer.type = 'agent' THEN CASE WHEN NULLIF(parent_peer.display_name, '') IS NOT NULL THEN NULLIF(parent_peer.display_name, '') WHEN NULLIF(parent_peer_agent.name, '') IS NOT NULL THEN NULLIF(parent_peer_agent.name, '') ELSE parent_peer.id END WHEN parent_peer.type = 'human' THEN CASE WHEN NULLIF(parent_peer_user.display_name, '') IS NOT NULL THEN NULLIF(parent_peer_user.display_name, '') WHEN NULLIF(parent_peer.display_name, '') IS NOT NULL THEN NULLIF(parent_peer.display_name, '') WHEN parent_peer_user.name IS NOT NULL THEN parent_peer_user.name ELSE parent_peer.id END WHEN parent_peer.type = 'service' THEN CASE WHEN NULLIF(parent_peer_service.handle, '') IS NOT NULL THEN NULLIF(parent_peer_service.handle, '') ELSE parent_peer.id END ELSE '' END ELSE '' END AS subtitle, CASE WHEN s.type = 'dm' AND dm_peer.type = 'agent' THEN CASE WHEN dm_peer_agent.avatar_url IS NULL THEN '' ELSE dm_peer_agent.avatar_url END WHEN s.type = 'dm' AND dm_peer.type = 'human' THEN CASE WHEN dm_peer_user.avatar_url IS NULL THEN '' ELSE dm_peer_user.avatar_url END WHEN s.type = 'dm' AND dm_peer.type = 'service' THEN CASE WHEN NULLIF(dm_peer_service.avatar_url, '') IS NOT NULL THEN dm_peer_service.avatar_url WHEN dm_peer_service_definition.avatar_url IS NOT NULL THEN dm_peer_service_definition.avatar_url ELSE '' END WHEN s.type = 'service_account' THEN CASE WHEN service_account_member.avatar_url IS NULL THEN '' ELSE service_account_member.avatar_url END ELSE '' END AS avatar_url, CASE WHEN s.type = 'channel' THEN CASE WHEN cp.subsystem IS NULL OR cp.subsystem = '' THEN 'im' ELSE cp.subsystem END ELSE '' END AS subsystem, CASE WHEN s.type = 'channel' THEN CASE WHEN channel_team.id IS NULL THEN '' ELSE channel_team.id END ELSE '' END AS team_id, CASE WHEN s.type = 'channel' THEN CASE WHEN channel_team.handle IS NULL THEN '' ELSE channel_team.handle END ELSE '' END AS team_handle, CASE WHEN s.type = 'channel' THEN CASE WHEN channel_team.display_name IS NULL THEN '' ELSE channel_team.display_name END ELSE '' END AS team_display_name, CASE WHEN s.type = 'dm' AND dm_peer.type = 'human' THEN 'human' WHEN s.type = 'dm' AND dm_peer.type = 'agent' THEN 'agent' WHEN s.type = 'dm' AND dm_peer.type = 'service' THEN 'service' WHEN s.type = 'service_account' AND service_account.id IS NOT NULL THEN 'service' ELSE '' END AS peer_type, CASE WHEN s.type = 'dm' AND dm_peer.type = 'human' THEN CASE WHEN dm_peer_user.id IS NULL THEN '' ELSE dm_peer_user.id END WHEN s.type = 'dm' AND dm_peer.type = 'agent' THEN CASE WHEN dm_peer_agent.id IS NULL THEN '' ELSE dm_peer_agent.id END WHEN s.type = 'dm' AND dm_peer.type = 'service' THEN CASE WHEN dm_peer_service.service_installation_id IS NULL THEN dm_peer.id ELSE dm_peer_service.service_installation_id END WHEN s.type = 'service_account' THEN CASE WHEN service_account.principal_id IS NULL THEN '' ELSE service_account.principal_id END ELSE '' END AS peer_id, s.latest_message_at, s.visible_message_count, s.visible_last_message_at, CASE WHEN s.visible_last_message_at IS NOT NULL THEN s.visible_last_message_at WHEN s.latest_message_at IS NOT NULL THEN s.latest_message_at WHEN s.updated_at IS NOT NULL THEN s.updated_at ELSE s.created_at END AS sort_at, cms.pinned_at, CASE WHEN cms.pinned_at IS NULL THEN 1 ELSE 0 END AS pinned_rank, (cms.manual_unread_at IS NOT NULL) AS manual_unread, CASE WHEN sns.state IS NULL THEN '' ELSE sns.state END AS subscription_state, CASE WHEN cms.hidden_at IS NOT NULL AND CASE WHEN s.latest_seq IS NULL THEN 0 ELSE s.latest_seq END <= CASE WHEN cms.hidden_through_seq IS NULL THEN 0 ELSE cms.hidden_through_seq END THEN TRUE WHEN s.type = 'dm' AND CASE WHEN dvs.hidden_through_seq IS NULL THEN 0 ELSE dvs.hidden_through_seq END > 0 AND CASE WHEN s.latest_seq IS NULL THEN 0 ELSE s.latest_seq END <= dvs.hidden_through_seq THEN TRUE ELSE FALSE END AS sidebar_hidden, FALSE AS archived, CASE WHEN s.type = 'thread' THEN CASE WHEN s.latest_seq IS NULL THEN 0 ELSE s.latest_seq END ELSE 0 END AS reply_count, CASE WHEN s.type = 'thread' THEN s.latest_message_at ELSE NULL END AS last_reply_at FROM visible_conversation_refs JOIN scopes s ON s.id = visible_conversation_refs.scope_id AND s.workspace_id = $1 AND s.status = 'active' AND s.archived_at IS NULL AND s.type IN ('channel', 'dm', 'thread', 'service_account') LEFT JOIN service_accounts service_account ON service_account.message_scope_id = s.id AND service_account.workspace_id = $1 AND s.type = 'service_account' LEFT JOIN service_members service_account_member ON service_account_member.member_id = service_account.author_member_id AND service_account_member.workspace_id = service_account.workspace_id AND service_account_member.service_kind = 'service_account' LEFT JOIN scope_notification_settings sns ON sns.scope_id = s.id AND sns.member_id = $2 LEFT JOIN conversation_member_states cms ON cms.scope_id = s.id AND cms.member_id = $2 LEFT JOIN dm_view_states dvs ON dvs.dm_thread_id = s.id AND dvs.member_id = $2 LEFT JOIN scope_channel_profiles cp ON cp.scope_id = s.id AND cp.workspace_id = $1 AND s.type = 'channel' LEFT JOIN workspace_teams channel_team ON channel_team.workspace_id = s.workspace_id AND channel_team.channel_id = s.id AND channel_team.status = 'active' AND s.type = 'channel' LEFT JOIN scope_memberships dm_peer_sm ON dm_peer_sm.scope_id = s.id AND dm_peer_sm.member_id <> $2 AND dm_peer_sm.removed_at IS NULL AND dm_peer_sm.archived_at IS NULL AND dm_peer_sm.effective_to_seq IS NULL AND s.type = 'dm' LEFT JOIN members dm_peer ON dm_peer.id = dm_peer_sm.member_id LEFT JOIN users dm_peer_user ON dm_peer.type = 'human' AND dm_peer_user.id = dm_peer.user_id LEFT JOIN agents dm_peer_agent ON dm_peer.type = 'agent' AND dm_peer_agent.member_id = dm_peer.id LEFT JOIN service_members dm_peer_service ON dm_peer.type = 'service' AND dm_peer_service.workspace_id = s.workspace_id AND dm_peer_service.member_id = dm_peer.id LEFT JOIN service_definitions dm_peer_service_definition ON dm_peer_service_definition.id = dm_peer_service.service_definition_id AND dm_peer_service_definition.service_kind = dm_peer_service.service_kind LEFT JOIN scopes parent_scope ON parent_scope.id = s.parent_scope_id AND parent_scope.workspace_id = $1 AND s.type = 'thread' LEFT JOIN scope_channel_profiles parent_cp ON parent_cp.scope_id = parent_scope.id AND parent_cp.workspace_id = $1 LEFT JOIN scope_memberships parent_peer_sm ON parent_peer_sm.scope_id = parent_scope.id AND parent_peer_sm.member_id <> $2 AND parent_peer_sm.removed_at IS NULL AND parent_peer_sm.archived_at IS NULL AND parent_peer_sm.effective_to_seq IS NULL AND parent_scope.type = 'dm' LEFT JOIN members parent_peer ON parent_peer.id = parent_peer_sm.member_id LEFT JOIN users parent_peer_user ON parent_peer.type = 'human' AND parent_peer_user.id = parent_peer.user_id LEFT JOIN agents parent_peer_agent ON parent_peer.type = 'agent' AND parent_peer_agent.member_id = parent_peer.id LEFT JOIN service_members parent_peer_service ON parent_peer.type = 'service' AND parent_peer_service.workspace_id = parent_scope.workspace_id AND parent_peer_service.member_id = parent_peer.id LEFT JOIN service_definitions parent_peer_service_definition ON parent_peer_service_definition.id = parent_peer_service.service_definition_id AND parent_peer_service_definition.service_kind = parent_peer_service.service_kind LEFT JOIN scope_messages root_message ON root_message.id = s.root_message_id AND root_message.scope_id = s.parent_scope_id AND s.type = 'thread' LEFT JOIN members root_author ON root_author.id = root_message.author_member_id LEFT JOIN users root_author_user ON root_message.author_type = 'human' AND root_author_user.id = root_author.user_id LEFT JOIN agents root_author_agent ON root_message.author_type = 'agent' AND root_author_agent.member_id = root_author.id WHERE ( s.type NOT IN ('dm', 'service_account') OR (s.type = 'service_account' AND service_account.id IS NOT NULL) OR (dm_peer.type = 'human' AND dm_peer_user.id IS NOT NULL AND dm_peer_user.id <> '') OR (dm_peer.type = 'agent' AND dm_peer_agent.id IS NOT NULL AND dm_peer_agent.id <> '') OR (dm_peer.type = 'service' AND dm_peer_service.id IS NOT NULL AND dm_peer_service.id <> '') ) %s AND ( s.type <> 'thread' OR sns.state IN ('active', 'quiet', 'mute') ) ),%s, latest_visible_messages_ranked AS ( SELECT m.*, ROW_NUMBER() OVER (PARTITION BY pc.scope_id ORDER BY m.seq DESC) AS rn FROM page_conversations pc JOIN scope_messages m ON m.scope_id = pc.scope_id AND m.workspace_id = $1 AND m.created_at = pc.visible_last_message_at AND %s WHERE pc.visible_message_count > 0 ), latest_visible_messages AS ( SELECT scope_id, content, author_type, author_member_id FROM latest_visible_messages_ranked WHERE rn = 1 ) SELECT pc.scope_id, pc.kind, pc.parent_scope_id, pc.parent_scope_type, pc.parent_title, pc.parent_message_id, pc.parent_message_short_id, pc.parent_message_author, pc.title, pc.subtitle, pc.avatar_url, pc.subsystem, pc.team_id, pc.team_handle, pc.team_display_name, pc.peer_type, pc.peer_id, CASE WHEN NULLIF(lm.content, '') IS NOT NULL THEN NULLIF(lm.content, '') WHEN NULLIF(pc.root_message_preview, '') IS NOT NULL THEN NULLIF(pc.root_message_preview, '') ELSE '' END AS preview, CASE WHEN lm.author_type = 'agent' THEN CASE WHEN NULLIF(author.display_name, '') IS NOT NULL THEN NULLIF(author.display_name, '') WHEN NULLIF(author_agent.name, '') IS NOT NULL THEN NULLIF(author_agent.name, '') WHEN author.id IS NOT NULL THEN author.id ELSE '' END WHEN lm.author_type = 'human' THEN CASE WHEN NULLIF(author_user.display_name, '') IS NOT NULL THEN NULLIF(author_user.display_name, '') WHEN NULLIF(author.display_name, '') IS NOT NULL THEN NULLIF(author.display_name, '') WHEN author_user.name IS NOT NULL THEN author_user.name WHEN author.id IS NOT NULL THEN author.id ELSE '' END WHEN lm.author_type = 'service' THEN CASE WHEN NULLIF(author.display_name, '') IS NOT NULL THEN NULLIF(author.display_name, '') WHEN author.id IS NOT NULL THEN author.id ELSE '' END ELSE '' END AS last_author_name, pc.latest_message_at, pc.sort_at, pc.pinned_at, 0 AS unread_count, pc.manual_unread, pc.subscription_state, pc.sidebar_hidden, pc.archived, pc.reply_count, pc.last_reply_at FROM page_conversations pc LEFT JOIN latest_visible_messages lm ON lm.scope_id = pc.scope_id LEFT JOIN members author ON author.id = lm.author_member_id LEFT JOIN users author_user ON lm.author_type = 'human' AND author_user.id = author.user_id LEFT JOIN agents author_agent ON lm.author_type = 'agent' AND author_agent.member_id = author.id ORDER BY pc.pinned_rank ASC, pc.sort_at DESC, pc.scope_id ASC `, visibleConversationRefsCTE, subsystemWhereClause, pageConversationsCTE, HumanVisibleMessageTypeSQL("m")) } func buildVisibleConversationScopeIDsForUserScopedQuery(selection conversationKindSelection, kindClause string, subsystemWhereClause string) string { ctes := make([]string, 0, 2) refSelects := make([]string, 0, 1) if selection.includesVisibleThreadRefs() { ctes = append(ctes, buildVisibleThreadRefsSQLCTEForConversationKinds("visible_thread_refs", 1, 2, 3, 4, selection)) refSelects = append(refSelects, `SELECT channel_id AS scope_id FROM visible_thread_refs`) } ctes = append(ctes, buildVisibleConversationRefsCTE(refSelects)) return fmt.Sprintf(` WITH %s, base_conversations AS ( SELECT s.id AS scope_id, s.type AS kind, CASE WHEN cms.hidden_at IS NOT NULL AND CASE WHEN s.latest_seq IS NULL THEN 0 ELSE s.latest_seq END <= CASE WHEN cms.hidden_through_seq IS NULL THEN 0 ELSE cms.hidden_through_seq END THEN TRUE WHEN s.type = 'dm' AND CASE WHEN dvs.hidden_through_seq IS NULL THEN 0 ELSE dvs.hidden_through_seq END > 0 AND CASE WHEN s.latest_seq IS NULL THEN 0 ELSE s.latest_seq END <= dvs.hidden_through_seq THEN TRUE ELSE FALSE END AS sidebar_hidden FROM visible_conversation_refs JOIN scopes s ON s.id = visible_conversation_refs.scope_id AND s.workspace_id = $1 AND s.status = 'active' AND s.archived_at IS NULL AND s.type IN ('channel', 'dm', 'thread', 'service_account') LEFT JOIN scope_notification_settings sns ON sns.scope_id = s.id AND sns.member_id = $2 LEFT JOIN conversation_member_states cms ON cms.scope_id = s.id AND cms.member_id = $2 LEFT JOIN dm_view_states dvs ON dvs.dm_thread_id = s.id AND dvs.member_id = $2 LEFT JOIN scope_channel_profiles cp ON cp.scope_id = s.id AND cp.workspace_id = $1 AND s.type = 'channel' LEFT JOIN scope_memberships dm_peer_sm ON dm_peer_sm.scope_id = s.id AND dm_peer_sm.member_id <> $2 AND dm_peer_sm.removed_at IS NULL AND dm_peer_sm.archived_at IS NULL AND dm_peer_sm.effective_to_seq IS NULL AND s.type = 'dm' LEFT JOIN members dm_peer ON dm_peer.id = dm_peer_sm.member_id LEFT JOIN users dm_peer_user ON dm_peer.type = 'human' AND dm_peer_user.id = dm_peer.user_id LEFT JOIN agents dm_peer_agent ON dm_peer.type = 'agent' AND dm_peer_agent.member_id = dm_peer.id LEFT JOIN service_members dm_peer_service ON dm_peer.type = 'service' AND dm_peer_service.workspace_id = s.workspace_id AND dm_peer_service.member_id = dm_peer.id WHERE ( s.type NOT IN ('dm', 'service_account') OR s.type = 'service_account' OR (dm_peer.type = 'human' AND dm_peer_user.id IS NOT NULL AND dm_peer_user.id <> '') OR (dm_peer.type = 'agent' AND dm_peer_agent.id IS NOT NULL AND dm_peer_agent.id <> '') OR (dm_peer.type = 'service' AND dm_peer_service.id IS NOT NULL AND dm_peer_service.id <> '') ) %s AND ( s.type <> 'thread' OR sns.state IN ('active', 'quiet', 'mute') ) ) SELECT scope_id, kind FROM base_conversations WHERE sidebar_hidden = FALSE%s `, strings.Join(ctes, ",\n"), subsystemWhereClause, kindClause) }