refactor(heartbeat): simplify and optimize heartbeat store operations

Clean up heartbeat query logic and improve query efficiency across
PostgreSQL and SQLite implementations.
This commit is contained in:
viettranx committed 2026-04-02 08:23:11 +07:00
1 parent e682c53f73
commit ab0e2051a6
3 files changed
+74 -117

No files matched your search

+2 -2
View File
@@ -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))
+37 -38
View File
@@ -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 {
+35 -77
View File
@@ -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]
}