diff --git a/internal/store/heartbeat_store.go b/internal/store/heartbeat_store.go index ac3f3fa5..5c03ff02 100644 --- a/internal/store/heartbeat_store.go +++ b/internal/store/heartbeat_store.go @@ -116,8 +116,8 @@ type HeartbeatStore interface { InsertLog(ctx context.Context, log *HeartbeatRunLog) error ListLogs(ctx context.Context, agentID uuid.UUID, limit, offset int) ([]HeartbeatRunLog, int, error) - // Delivery targets — distinct (channel, chatID) from session history for an agent. - ListDeliveryTargets(ctx context.Context, agentID uuid.UUID) ([]DeliveryTarget, error) + // Delivery targets — known (channel, chatID) pairs from channel_contacts for a tenant. + ListDeliveryTargets(ctx context.Context, tenantID uuid.UUID) ([]DeliveryTarget, error) // Events SetOnEvent(fn func(HeartbeatEvent)) diff --git a/internal/store/pg/heartbeat.go b/internal/store/pg/heartbeat.go index 387dff46..9e3fc53d 100644 --- a/internal/store/pg/heartbeat.go +++ b/internal/store/pg/heartbeat.go @@ -274,39 +274,24 @@ func (s *PGHeartbeatStore) ListLogs(ctx context.Context, agentID uuid.UUID, limi return logs, total, nil } -// ListDeliveryTargets returns distinct (channel, chatID, title, kind) pairs from sessions for an agent. -// Uses idx_sessions_agent (btree on agent_id) for fast lookup. -// Session key format: agent:{key}:{channel}:{kind}:{chatId}[:topic:N] -// Extracts full chatId including topic/thread suffix via array_to_string([5:]). -// JOIN on base chatId (split_part segment 5) for contact display name resolution. -func (s *PGHeartbeatStore) ListDeliveryTargets(ctx context.Context, agentID uuid.UUID) ([]store.DeliveryTarget, error) { +// ListDeliveryTargets returns known delivery targets (channel, chatID, title, kind) from channel_contacts. +// Queries contacts with contact_type IN ('group','topic') for the given tenant. +// For topic contacts, chatID is built as senderID + ":topic:" + threadID. +func (s *PGHeartbeatStore) ListDeliveryTargets(ctx context.Context, tenantID uuid.UUID) ([]store.DeliveryTarget, error) { rows, err := s.db.QueryContext(ctx, - `SELECT DISTINCT ON (s.channel, chat_id) - s.channel, - array_to_string((string_to_array(s.session_key::text, ':'))[5:], ':') AS chat_id, - COALESCE( - s.metadata->>'chat_title', - cc.display_name, - CASE WHEN cc.username != '' THEN '@' || cc.username ELSE '' END, - '' - ) AS title, - CASE - WHEN s.session_key LIKE '%:group:%' THEN 'group' - WHEN s.session_key LIKE '%:direct:%' THEN 'dm' - ELSE 'other' - END AS kind - FROM sessions s - LEFT JOIN channel_contacts cc - ON cc.sender_id = split_part(s.session_key, ':', 5) - AND cc.channel_type = s.channel - WHERE s.agent_id = $1 - AND s.channel IS NOT NULL AND s.channel != '' - AND s.session_key NOT LIKE '%:heartbeat%' - AND s.session_key NOT LIKE '%:cron%' - AND s.session_key NOT LIKE '%:subagent%' - AND s.session_key NOT LIKE '%:team:%' - ORDER BY s.channel, chat_id, title`, - agentID, + `SELECT cc.sender_id, + cc.thread_id, + cc.thread_type, + cc.channel_instance AS channel, + COALESCE(cc.display_name, cc.sender_id) AS title, + CASE WHEN cc.contact_type = 'topic' THEN 'topic' + WHEN cc.peer_kind = 'group' THEN 'group' + ELSE 'dm' END AS kind + FROM channel_contacts cc + WHERE cc.tenant_id = $1 + AND cc.contact_type IN ('group', 'topic') + ORDER BY cc.channel_instance, cc.display_name`, + tenantID, ) if err != nil { return nil, err @@ -316,15 +301,29 @@ func (s *PGHeartbeatStore) ListDeliveryTargets(ctx context.Context, agentID uuid var targets []store.DeliveryTarget seen := make(map[string]bool) for rows.Next() { - var t store.DeliveryTarget - if err := rows.Scan(&t.Channel, &t.ChatID, &t.Title, &t.Kind); err != nil { + var senderID, title, kind string + var threadID, threadType, channel *string + if err := rows.Scan(&senderID, &threadID, &threadType, &channel, &title, &kind); err != nil { return nil, err } - // Deduplicate by channel+chatID (sessions can have multiple rows for same target). - key := t.Channel + ":" + t.ChatID - if t.ChatID != "" && !seen[key] { + if channel == nil { + continue + } + chatID := senderID + displayTitle := title + if threadID != nil && *threadID != "" { + chatID = senderID + ":topic:" + *threadID + displayTitle = title + " > topic:" + *threadID + } + key := *channel + ":" + chatID + if !seen[key] { seen[key] = true - targets = append(targets, t) + targets = append(targets, store.DeliveryTarget{ + Channel: *channel, + ChatID: chatID, + Title: displayTitle, + Kind: kind, + }) } } if err := rows.Err(); err != nil { diff --git a/internal/store/sqlitestore/heartbeat.go b/internal/store/sqlitestore/heartbeat.go index d9c4b0f8..19f204aa 100644 --- a/internal/store/sqlitestore/heartbeat.go +++ b/internal/store/sqlitestore/heartbeat.go @@ -6,7 +6,6 @@ import ( "context" "database/sql" "encoding/json" - "strings" "sync" "time" @@ -281,29 +280,24 @@ func (s *SQLiteHeartbeatStore) ListLogs(ctx context.Context, agentID uuid.UUID, return logs, total, nil } -// ListDeliveryTargets returns distinct (channel, chatID, title, kind) pairs from sessions. -// SQLite adaptation: replaces DISTINCT ON + split_part with GROUP BY + substr/instr + Go-side dedup. -// Session key format: agent:{key}:{channel}:{kind}:{chatId} -func (s *SQLiteHeartbeatStore) ListDeliveryTargets(ctx context.Context, agentID uuid.UUID) ([]store.DeliveryTarget, error) { +// ListDeliveryTargets returns known delivery targets (channel, chatID, title, kind) from channel_contacts. +// Queries contacts with contact_type IN ('group','topic') for the given tenant. +// For topic contacts, chatID is built as senderID + ":topic:" + threadID. +func (s *SQLiteHeartbeatStore) ListDeliveryTargets(ctx context.Context, tenantID uuid.UUID) ([]store.DeliveryTarget, error) { rows, err := s.db.QueryContext(ctx, - `SELECT - s.channel, - substr(s.session_key, instr(s.session_key||':', ':') + 1) AS raw_tail, - json_extract(s.metadata, '$.chat_title') AS chat_title, - s.session_key, - cc.display_name, - cc.username - FROM sessions s - LEFT JOIN channel_contacts cc - ON cc.channel_type = s.channel - WHERE s.agent_id = ? - AND s.channel IS NOT NULL AND s.channel != '' - AND s.session_key NOT LIKE '%:heartbeat%' - AND s.session_key NOT LIKE '%:cron%' - AND s.session_key NOT LIKE '%:subagent%' - AND s.session_key NOT LIKE '%:team:%' - GROUP BY s.channel, s.session_key`, - agentID, + `SELECT cc.sender_id, + cc.thread_id, + cc.thread_type, + cc.channel_instance AS channel, + COALESCE(cc.display_name, cc.sender_id) AS title, + CASE WHEN cc.contact_type = 'topic' THEN 'topic' + WHEN cc.peer_kind = 'group' THEN 'group' + ELSE 'dm' END AS kind + FROM channel_contacts cc + WHERE cc.tenant_id = ? + AND cc.contact_type IN ('group', 'topic') + ORDER BY cc.channel_instance, cc.display_name`, + tenantID, ) if err != nil { return nil, err @@ -313,47 +307,30 @@ func (s *SQLiteHeartbeatStore) ListDeliveryTargets(ctx context.Context, agentID seen := make(map[string]bool) var targets []store.DeliveryTarget for rows.Next() { - var channel, rawTail, sessionKey string - var chatTitle, displayName, username *string - if err := rows.Scan(&channel, &rawTail, &chatTitle, &sessionKey, &displayName, &username); err != nil { + var senderID, title, kind string + var threadID, threadType, channel *string + if err := rows.Scan(&senderID, &threadID, &threadType, &channel, &title, &kind); err != nil { return nil, err } - - // Parse chatId: everything after 4th colon (includes :topic:N suffix for forum groups). - chatID := extractSessionKeyTail(sessionKey, 5) - if chatID == "" { + if channel == nil { continue } - key := channel + ":" + chatID - if seen[key] { - continue + chatID := senderID + displayTitle := title + if threadID != nil && *threadID != "" { + chatID = senderID + ":topic:" + *threadID + displayTitle = title + " > topic:" + *threadID } - seen[key] = true - - title := "" - switch { - case chatTitle != nil && *chatTitle != "": - title = *chatTitle - case displayName != nil && *displayName != "": - title = *displayName - case username != nil && *username != "": - title = "@" + *username + key := *channel + ":" + chatID + if !seen[key] { + seen[key] = true + targets = append(targets, store.DeliveryTarget{ + Channel: *channel, + ChatID: chatID, + Title: displayTitle, + Kind: kind, + }) } - - kind := "other" - switch { - case strings.Contains(sessionKey, ":group:"): - kind = "group" - case strings.Contains(sessionKey, ":direct:"): - kind = "dm" - } - - targets = append(targets, store.DeliveryTarget{ - Channel: channel, - ChatID: chatID, - Title: title, - Kind: kind, - }) } if err := rows.Err(); err != nil { return nil, err @@ -361,22 +338,3 @@ func (s *SQLiteHeartbeatStore) ListDeliveryTargets(ctx context.Context, agentID return targets, nil } -// extractSessionKeyPart returns the n-th colon-delimited segment (1-indexed). -func extractSessionKeyPart(key string, n int) string { - parts := strings.SplitN(key, ":", n+1) - if len(parts) < n { - return "" - } - return parts[n-1] -} - -// extractSessionKeyTail returns everything from the n-th colon-delimited segment onwards. -// For session key "agent:default:telegram:group:-100123:topic:42" with n=5, -// returns "-100123:topic:42" (includes topic/thread suffix). -func extractSessionKeyTail(key string, n int) string { - parts := strings.SplitN(key, ":", n) - if len(parts) < n { - return "" - } - return parts[n-1] -}