//go:build sqlite || sqliteonly package sqlitestore import ( "context" "database/sql" "encoding/json" "sync" "time" "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/store" ) // SQLiteHeartbeatStore implements store.HeartbeatStore backed by SQLite. type SQLiteHeartbeatStore struct { db *sql.DB mu sync.Mutex onEvent func(store.HeartbeatEvent) dueCache []store.AgentHeartbeat cacheLoaded bool cacheTime time.Time cacheTTL time.Duration } const defaultHeartbeatCacheTTL = 30 * time.Second func NewSQLiteHeartbeatStore(db *sql.DB) *SQLiteHeartbeatStore { return &SQLiteHeartbeatStore{db: db, cacheTTL: defaultHeartbeatCacheTTL} } func (s *SQLiteHeartbeatStore) SetOnEvent(fn func(store.HeartbeatEvent)) { s.mu.Lock() defer s.mu.Unlock() s.onEvent = fn } func (s *SQLiteHeartbeatStore) emitEvent(event store.HeartbeatEvent) { s.mu.Lock() fn := s.onEvent s.mu.Unlock() if fn != nil { fn(event) } } func (s *SQLiteHeartbeatStore) InvalidateCache() { s.mu.Lock() defer s.mu.Unlock() s.cacheLoaded = false } const heartbeatSelectCols = `id, agent_id, enabled, interval_sec, prompt, provider_id, model, isolated_session, light_context, ack_max_chars, max_retries, active_hours_start, active_hours_end, timezone, channel, chat_id, next_run_at, last_run_at, last_status, last_error, run_count, suppress_count, metadata, created_at, updated_at` func scanHeartbeat(row interface { Scan(...any) error }) (*store.AgentHeartbeat, error) { var hb store.AgentHeartbeat var metadata []byte var nextRunAt, lastRunAt nullSqliteTime createdAt, updatedAt := scanTimePair() err := row.Scan( &hb.ID, &hb.AgentID, &hb.Enabled, &hb.IntervalSec, &hb.Prompt, &hb.ProviderID, &hb.Model, &hb.IsolatedSession, &hb.LightContext, &hb.AckMaxChars, &hb.MaxRetries, &hb.ActiveHoursStart, &hb.ActiveHoursEnd, &hb.Timezone, &hb.Channel, &hb.ChatID, &nextRunAt, &lastRunAt, &hb.LastStatus, &hb.LastError, &hb.RunCount, &hb.SuppressCount, &metadata, createdAt, updatedAt, ) if err != nil { return nil, err } hb.CreatedAt = createdAt.Time hb.UpdatedAt = updatedAt.Time if nextRunAt.Valid { hb.NextRunAt = &nextRunAt.Time } if lastRunAt.Valid { hb.LastRunAt = &lastRunAt.Time } hb.Metadata = metadata return &hb, nil } func (s *SQLiteHeartbeatStore) Get(ctx context.Context, agentID uuid.UUID) (*store.AgentHeartbeat, error) { row := s.db.QueryRowContext(ctx, `SELECT `+heartbeatSelectCols+` FROM agent_heartbeats WHERE agent_id = ?`, agentID, ) return scanHeartbeat(row) } func (s *SQLiteHeartbeatStore) Upsert(ctx context.Context, hb *store.AgentHeartbeat) error { meta := hb.Metadata if meta == nil { meta = json.RawMessage("{}") } now := time.Now().UTC() createdAt, updatedAt := scanTimePair() err := s.db.QueryRowContext(ctx, `INSERT INTO agent_heartbeats (agent_id, enabled, interval_sec, prompt, provider_id, model, isolated_session, light_context, ack_max_chars, max_retries, active_hours_start, active_hours_end, timezone, channel, chat_id, next_run_at, metadata, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT (agent_id) DO UPDATE SET enabled = excluded.enabled, interval_sec = excluded.interval_sec, prompt = excluded.prompt, provider_id = excluded.provider_id, model = excluded.model, isolated_session = excluded.isolated_session, light_context = excluded.light_context, ack_max_chars = excluded.ack_max_chars, max_retries = excluded.max_retries, active_hours_start = excluded.active_hours_start, active_hours_end = excluded.active_hours_end, timezone = excluded.timezone, channel = excluded.channel, chat_id = excluded.chat_id, next_run_at = excluded.next_run_at, metadata = excluded.metadata, updated_at = excluded.updated_at RETURNING id, created_at, updated_at`, hb.AgentID, hb.Enabled, hb.IntervalSec, hb.Prompt, hb.ProviderID, hb.Model, hb.IsolatedSession, hb.LightContext, hb.AckMaxChars, hb.MaxRetries, hb.ActiveHoursStart, hb.ActiveHoursEnd, hb.Timezone, hb.Channel, hb.ChatID, hb.NextRunAt, meta, now, now, ).Scan(&hb.ID, createdAt, updatedAt) if err != nil { return err } hb.CreatedAt = createdAt.Time hb.UpdatedAt = updatedAt.Time s.InvalidateCache() return nil } func (s *SQLiteHeartbeatStore) ListDue(ctx context.Context, now time.Time) ([]store.AgentHeartbeat, error) { s.mu.Lock() if s.cacheLoaded && time.Since(s.cacheTime) < s.cacheTTL { cached := s.dueCache s.mu.Unlock() var due []store.AgentHeartbeat for _, hb := range cached { if hb.NextRunAt != nil && !hb.NextRunAt.After(now) { due = append(due, hb) } } return due, nil } s.mu.Unlock() rows, err := s.db.QueryContext(ctx, `SELECT `+heartbeatSelectCols+` FROM agent_heartbeats WHERE enabled = 1 AND next_run_at IS NOT NULL`) if err != nil { return nil, err } defer rows.Close() var all []store.AgentHeartbeat for rows.Next() { hb, err := scanHeartbeat(rows) if err != nil { return nil, err } all = append(all, *hb) } if err := rows.Err(); err != nil { return nil, err } s.mu.Lock() s.dueCache = all s.cacheLoaded = true s.cacheTime = time.Now() s.mu.Unlock() var due []store.AgentHeartbeat for _, hb := range all { if hb.NextRunAt != nil && !hb.NextRunAt.After(now) { due = append(due, hb) } } return due, nil } func (s *SQLiteHeartbeatStore) UpdateState(ctx context.Context, id uuid.UUID, state store.HeartbeatState) error { _, err := s.db.ExecContext(ctx, `UPDATE agent_heartbeats SET next_run_at = ?, last_run_at = ?, last_status = ?, last_error = ?, run_count = ?, suppress_count = ?, updated_at = ? WHERE id = ?`, state.NextRunAt, state.LastRunAt, state.LastStatus, state.LastError, state.RunCount, state.SuppressCount, time.Now().UTC(), id, ) if err == nil { s.InvalidateCache() } return err } func (s *SQLiteHeartbeatStore) Delete(ctx context.Context, agentID uuid.UUID) error { _, err := s.db.ExecContext(ctx, `DELETE FROM agent_heartbeats WHERE agent_id = ?`, agentID) if err == nil { s.InvalidateCache() } return err } func (s *SQLiteHeartbeatStore) InsertLog(ctx context.Context, log *store.HeartbeatRunLog) error { meta := log.Metadata if meta == nil { meta = json.RawMessage("{}") } _, err := s.db.ExecContext(ctx, `INSERT INTO heartbeat_run_logs (heartbeat_id, agent_id, status, summary, error, duration_ms, input_tokens, output_tokens, skip_reason, metadata, ran_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)`, log.HeartbeatID, log.AgentID, log.Status, log.Summary, log.Error, log.DurationMS, log.InputTokens, log.OutputTokens, log.SkipReason, meta, log.RanAt, ) return err } func (s *SQLiteHeartbeatStore) ListLogs(ctx context.Context, agentID uuid.UUID, limit, offset int) ([]store.HeartbeatRunLog, int, error) { if limit <= 0 { limit = 20 } var total int if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM heartbeat_run_logs WHERE agent_id = ?`, agentID, ).Scan(&total); err != nil { return nil, 0, err } rows, err := s.db.QueryContext(ctx, `SELECT id, heartbeat_id, agent_id, status, summary, error, duration_ms, input_tokens, output_tokens, skip_reason, metadata, ran_at, created_at FROM heartbeat_run_logs WHERE agent_id = ? ORDER BY ran_at DESC LIMIT ? OFFSET ?`, agentID, limit, offset, ) if err != nil { return nil, 0, err } defer rows.Close() var logs []store.HeartbeatRunLog for rows.Next() { var l store.HeartbeatRunLog var metadata []byte var ranAt, createdAt sqliteTime if err := rows.Scan( &l.ID, &l.HeartbeatID, &l.AgentID, &l.Status, &l.Summary, &l.Error, &l.DurationMS, &l.InputTokens, &l.OutputTokens, &l.SkipReason, &metadata, &ranAt, &createdAt, ); err != nil { return nil, 0, err } l.RanAt = ranAt.Time l.CreatedAt = createdAt.Time l.Metadata = metadata logs = append(logs, l) } if err := rows.Err(); err != nil { return nil, 0, err } return logs, total, nil } // ListDeliveryTargets returns known delivery targets (channel, chatID, title, kind) from channel_contacts. // Queries contacts with contact_type IN ('group','topic','user') 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 cc.sender_id, cc.thread_id, cc.thread_type, cc.channel_instance AS channel, COALESCE(NULLIF(json_extract(cc.metadata, '$.display_title'), ''), 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', 'user') ORDER BY cc.channel_instance, cc.display_name`, tenantID, ) if err != nil { return nil, err } defer rows.Close() seen := make(map[string]bool) var targets []store.DeliveryTarget for rows.Next() { 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 } 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, store.DeliveryTarget{ Channel: *channel, ChatID: chatID, Title: displayTitle, Kind: kind, }) } } if err := rows.Err(); err != nil { return nil, err } return targets, nil }