mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-11 03:13:24 +00:00
minhdang03
1d68a21f65
channel_pending_messages is a buffer, not an archive. Rows were deleted outright on two paths: the bot being mentioned hands the buffer to the agent and clears the key, and LLM compaction replaces old rows with a summary. For group capture that buffer held the only copy of the raw text, so a single mention or compaction pass destroyed days of messages nothing had read yet. Copy every row into channel_message_archive inside the same transaction as the delete, tagged with the reason (consumed, compacted, stale). Archived rows keep their original id, so a replayed delete is a no-op. Add ListArchivedByKey so consumers that need full history read the archive instead of the buffer.
94 lines
4.4 KiB
Go
94 lines
4.4 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
)
|
|
|
|
// PendingMessage represents a buffered group chat message (or LLM-generated summary)
|
|
// stored in channel_pending_messages table.
|
|
type PendingMessage struct {
|
|
ID uuid.UUID `json:"id" db:"id"`
|
|
ChannelName string `json:"channel_name" db:"channel_name"`
|
|
HistoryKey string `json:"history_key" db:"history_key"`
|
|
ParentHistoryKey string `json:"parent_history_key,omitempty" db:"parent_history_key"`
|
|
Sender string `json:"sender" db:"sender"`
|
|
SenderID string `json:"sender_id" db:"sender_id"`
|
|
Body string `json:"body" db:"body"`
|
|
PlatformMsgID string `json:"platform_msg_id" db:"platform_msg_id"`
|
|
IsSummary bool `json:"is_summary" db:"is_summary"`
|
|
CreatedAt time.Time `json:"created_at" db:"created_at"`
|
|
UpdatedAt time.Time `json:"updated_at" db:"updated_at"`
|
|
}
|
|
|
|
// Archive reasons recorded on channel_message_archive rows.
|
|
const (
|
|
// ArchiveReasonConsumed marks messages dropped because the buffer was handed
|
|
// to the agent (bot mentioned) or cleared by an operator.
|
|
ArchiveReasonConsumed = "consumed"
|
|
// ArchiveReasonCompacted marks messages replaced by an LLM summary row.
|
|
ArchiveReasonCompacted = "compacted"
|
|
// ArchiveReasonStale marks messages dropped by TTL cleanup.
|
|
ArchiveReasonStale = "stale"
|
|
)
|
|
|
|
// ArchivedMessage is a pending message preserved after it left the live buffer.
|
|
type ArchivedMessage struct {
|
|
PendingMessage
|
|
ArchivedAt time.Time `json:"archived_at" db:"archived_at"`
|
|
ArchiveReason string `json:"archive_reason" db:"archive_reason"`
|
|
}
|
|
|
|
// PendingMessageGroup is a summary row for the grouped overview page.
|
|
type PendingMessageGroup struct {
|
|
ChannelName string `json:"channel_name" db:"channel_name"`
|
|
HistoryKey string `json:"history_key" db:"history_key"`
|
|
ParentHistoryKey string `json:"parent_history_key,omitempty" db:"parent_history_key"`
|
|
GroupTitle string `json:"group_title,omitempty" db:"group_title"`
|
|
ParentGroupTitle string `json:"parent_group_title,omitempty" db:"-"`
|
|
MessageCount int `json:"message_count" db:"message_count"`
|
|
HasSummary bool `json:"has_summary" db:"has_summary"`
|
|
LastActivity time.Time `json:"last_activity" db:"last_activity"`
|
|
}
|
|
|
|
// PendingMessageStore persists group chat messages for context when bot is mentioned.
|
|
type PendingMessageStore interface {
|
|
// AppendBatch inserts multiple pending messages in a single query.
|
|
AppendBatch(ctx context.Context, msgs []PendingMessage) error
|
|
|
|
// ListByKey returns all pending messages for a channel+historyKey, ordered by created_at ASC.
|
|
ListByKey(ctx context.Context, channelName, historyKey string) ([]PendingMessage, error)
|
|
|
|
// DeleteByKey removes all pending messages for a channel+historyKey.
|
|
// Removed rows are copied to the archive first.
|
|
DeleteByKey(ctx context.Context, channelName, historyKey string) error
|
|
|
|
// Compact atomically deletes old messages (by IDs) and inserts a summary row.
|
|
// Deleted rows are copied to the archive in the same transaction.
|
|
Compact(ctx context.Context, deleteIDs []uuid.UUID, summary *PendingMessage) error
|
|
|
|
// DeleteStale removes messages older than the given duration for inactive groups.
|
|
// Removed rows are copied to the archive first.
|
|
DeleteStale(ctx context.Context, olderThan time.Duration) (int64, error)
|
|
|
|
// ListArchivedByKey returns archived messages for a channel+historyKey created at
|
|
// or after `since`, ordered by created_at ASC. A zero `since` returns everything;
|
|
// limit <= 0 means no limit.
|
|
ListArchivedByKey(ctx context.Context, channelName, historyKey string, since time.Time, limit int) ([]ArchivedMessage, error)
|
|
|
|
// ListGroups returns all distinct channel+historyKey groups with message counts.
|
|
ListGroups(ctx context.Context) ([]PendingMessageGroup, error)
|
|
|
|
// 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)
|
|
}
|