diff --git a/internal/channels/history_compaction.go b/internal/channels/history_compaction.go index a5812177..a0b172d8 100644 --- a/internal/channels/history_compaction.go +++ b/internal/channels/history_compaction.go @@ -24,6 +24,7 @@ type CompactionConfig struct { // MaybeCompact checks if compaction is needed for a history key and triggers it in background. // Called from Record() after appending. Thread-safe via sync.Map compaction guard. +// Uses DB count (not RAM count) to correctly detect threshold after server restarts. func (ph *PendingHistory) MaybeCompact(historyKey string, currentCount int, cfg *CompactionConfig) { if ph.store == nil || cfg == nil || cfg.Provider == nil { return @@ -32,9 +33,18 @@ func (ph *PendingHistory) MaybeCompact(historyKey string, currentCount int, cfg if threshold <= 0 { threshold = DefaultGroupHistoryLimit } + + // RAM count may be stale after restart (LoadFromDB doesn't warm full history). + // If RAM says below threshold, ask DB for the real count. if currentCount <= threshold { - return + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + dbCount, err := ph.store.CountByKey(ctx, ph.channelName, historyKey) + if err != nil || dbCount <= threshold { + return + } } + // Guard: only one compaction per key at a time if _, loaded := ph.compacting.LoadOrStore(historyKey, true); loaded { return diff --git a/internal/store/pending_message_store.go b/internal/store/pending_message_store.go index 3ede6329..81bffb6a 100644 --- a/internal/store/pending_message_store.go +++ b/internal/store/pending_message_store.go @@ -55,6 +55,9 @@ type PendingMessageStore interface { // CountAll returns the total number of pending messages across all groups. CountAll(ctx context.Context) (int64, error) + // CountByKey returns the number of pending messages for a specific channel+historyKey. + CountByKey(ctx context.Context, channelName, historyKey string) (int, error) + // ResolveGroupTitles looks up chat_title from session metadata for each group. // Returns a map of "channel_name:history_key" → title. Used only by the UI layer. ResolveGroupTitles(ctx context.Context, groups []PendingMessageGroup) (map[string]string, error) diff --git a/internal/store/pg/pending_message_store.go b/internal/store/pg/pending_message_store.go index 17889fa0..4736a480 100644 --- a/internal/store/pg/pending_message_store.go +++ b/internal/store/pg/pending_message_store.go @@ -191,6 +191,15 @@ func (s *PGPendingMessageStore) CountAll(ctx context.Context) (int64, error) { return count, err } +func (s *PGPendingMessageStore) CountByKey(ctx context.Context, channelName, historyKey string) (int, error) { + var count int + err := s.db.QueryRowContext(ctx, + `SELECT COUNT(*) FROM channel_pending_messages WHERE channel_name = $1 AND history_key = $2`, + channelName, historyKey, + ).Scan(&count) + return count, err +} + func (s *PGPendingMessageStore) ResolveGroupTitles(ctx context.Context, groups []store.PendingMessageGroup) (map[string]string, error) { if len(groups) == 0 { return nil, nil