From 1d68a21f654907851a55ee23954aa9b600458cce Mon Sep 17 00:00:00 2001 From: minhdang03 Date: Sun, 16 Aug 2026 02:48:15 +1000 Subject: [PATCH] fix(channels): archive group messages before the pending buffer drops them (#1514) 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. --- docs/06-store-data-model.md | 20 +- internal/channelmemory/service_test.go | 4 + .../channels/discord/contact_refresh_test.go | 4 + .../http/channel_memory_extraction_test.go | 4 + internal/store/pending_message_store.go | 26 +++ internal/store/pg/pending_message_store.go | 114 +++++++-- .../pending_message_archive_test.go | 218 ++++++++++++++++++ .../store/sqlitestore/pending_messages.go | 132 +++++++++-- internal/store/sqlitestore/schema.go | 23 +- internal/store/sqlitestore/schema.sql | 26 +++ internal/upgrade/version.go | 2 +- .../000097_channel_message_archive.down.sql | 3 + .../000097_channel_message_archive.up.sql | 35 +++ .../pending_message_archive_test.go | 204 ++++++++++++++++ 14 files changed, 778 insertions(+), 37 deletions(-) create mode 100644 internal/store/sqlitestore/pending_message_archive_test.go create mode 100644 migrations/000097_channel_message_archive.down.sql create mode 100644 migrations/000097_channel_message_archive.up.sql create mode 100644 tests/integration/pending_message_archive_test.go diff --git a/docs/06-store-data-model.md b/docs/06-store-data-model.md index 2c030ed7..dc72d64e 100644 --- a/docs/06-store-data-model.md +++ b/docs/06-store-data-model.md @@ -419,13 +419,27 @@ Offline message queue for group chats. Buffers messages when the bot is not acti |--------|---------| | `AppendBatch(msgs)` | Insert multiple messages in one query | | `ListByKey(channelName, historyKey)` | Retrieve buffered messages for a group | -| `DeleteByKey(channelName, historyKey)` | Clear messages after processing | -| `Compact(deleteIDs, summary)` | Atomically delete old messages + insert summary | -| `DeleteStale(olderThan)` | Prune messages older than duration | +| `DeleteByKey(channelName, historyKey)` | Clear messages after processing (archives first) | +| `Compact(deleteIDs, summary)` | Atomically archive + delete old messages, insert summary | +| `DeleteStale(olderThan)` | Prune messages older than duration (archives first) | +| `ListArchivedByKey(channelName, historyKey, since, limit)` | Replay archived messages for a group | | `ListGroups()` | Return distinct channel+key groups with counts | | `CountAll()` | Total pending messages across all groups | | `ResolveGroupTitles(groups)` | Look up chat titles from session metadata | +**`channel_pending_messages` is a buffer, not an archive.** Rows leave it on two +paths: the bot being mentioned hands the buffer to the agent and clears the key, +and compaction replaces old rows with an LLM summary. Both are deletes, and for +group capture the buffer is the only place the raw text is stored — a mention or +a compaction pass used to destroy days of messages that nothing had read yet. + +Every row is therefore copied into `channel_message_archive` inside the same +transaction as the delete, tagged with `archive_reason` (`consumed`, `compacted`, +`stale`). Archived rows keep their original `id`, so a replayed delete is a +no-op. Reads go through `ListArchivedByKey`, which is tenant-scoped like every +other store read. Anything that needs the full group history — digest pipelines, +exports, audits — must read the archive, not the buffer. + ### KnowledgeGraphStore Entity-relationship graph storage for AI inference and knowledge extraction. Supports graph traversal, confidence pruning, and bulk ingestion. diff --git a/internal/channelmemory/service_test.go b/internal/channelmemory/service_test.go index f787281f..7786411d 100644 --- a/internal/channelmemory/service_test.go +++ b/internal/channelmemory/service_test.go @@ -140,6 +140,10 @@ func (f *fakePendingStore) DeleteStale(context.Context, time.Duration) (int64, e return 0, nil } +func (f *fakePendingStore) ListArchivedByKey(context.Context, string, string, time.Time, int) ([]store.ArchivedMessage, error) { + return nil, nil +} + func (f *fakePendingStore) ListGroups(context.Context) ([]store.PendingMessageGroup, error) { return f.groups, nil } diff --git a/internal/channels/discord/contact_refresh_test.go b/internal/channels/discord/contact_refresh_test.go index 569a9923..bc69a459 100644 --- a/internal/channels/discord/contact_refresh_test.go +++ b/internal/channels/discord/contact_refresh_test.go @@ -40,6 +40,10 @@ func (s *pendingGroupsStore) Compact(context.Context, []uuid.UUID, *store.Pendin func (s *pendingGroupsStore) DeleteStale(context.Context, time.Duration) (int64, error) { return 0, nil } + +func (s *pendingGroupsStore) ListArchivedByKey(context.Context, string, string, time.Time, int) ([]store.ArchivedMessage, error) { + return nil, nil +} func (s *pendingGroupsStore) ListGroups(context.Context) ([]store.PendingMessageGroup, error) { return s.groups, nil } diff --git a/internal/http/channel_memory_extraction_test.go b/internal/http/channel_memory_extraction_test.go index ab43c580..d28aa3bd 100644 --- a/internal/http/channel_memory_extraction_test.go +++ b/internal/http/channel_memory_extraction_test.go @@ -41,6 +41,10 @@ func (s *memoryExtractionPendingStore) DeleteStale(context.Context, time.Duratio return 0, nil } +func (s *memoryExtractionPendingStore) ListArchivedByKey(context.Context, string, string, time.Time, int) ([]store.ArchivedMessage, error) { + return nil, nil +} + func (s *memoryExtractionPendingStore) ListGroups(context.Context) ([]store.PendingMessageGroup, error) { return s.groups, nil } diff --git a/internal/store/pending_message_store.go b/internal/store/pending_message_store.go index eeecc7d8..57d737b1 100644 --- a/internal/store/pending_message_store.go +++ b/internal/store/pending_message_store.go @@ -23,6 +23,24 @@ type PendingMessage struct { 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"` @@ -44,14 +62,22 @@ type PendingMessageStore interface { 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) diff --git a/internal/store/pg/pending_message_store.go b/internal/store/pg/pending_message_store.go index 55c7fc9e..06adcf96 100644 --- a/internal/store/pg/pending_message_store.go +++ b/internal/store/pg/pending_message_store.go @@ -69,16 +69,56 @@ func (s *PGPendingMessageStore) ListByKey(ctx context.Context, channelName, hist return result, err } +// archivedColumns are copied verbatim from the buffer into channel_message_archive. +const archivedColumns = `id, channel_name, history_key, parent_history_key, sender, sender_id, + body, platform_msg_id, is_summary, created_at, updated_at, tenant_id` + +// archivedReadColumns omits tenant_id: reads are already tenant-scoped by the +// WHERE clause and ArchivedMessage carries no tenant field. +const archivedReadColumns = `id, channel_name, history_key, parent_history_key, sender, sender_id, + body, platform_msg_id, is_summary, created_at, updated_at, archived_at, archive_reason` + +// archivePending copies every buffer row matching where into the archive. Callers +// pass the same where/args they are about to DELETE with, so the archive cannot +// miss a row the delete removes. Replaying is a no-op: archived rows keep their +// original id. +func archivePending(ctx context.Context, tx *sql.Tx, where string, args []any, reasonParam int, reason string) error { + insertArgs := make([]any, 0, len(args)+1) + insertArgs = append(insertArgs, args...) + insertArgs = append(insertArgs, reason) + if _, err := tx.ExecContext(ctx, + `INSERT INTO channel_message_archive (`+archivedColumns+`, archived_at, archive_reason) + SELECT `+archivedColumns+fmt.Sprintf(", NOW(), $%d", reasonParam)+` + FROM channel_pending_messages `+where+` + ON CONFLICT (id) DO NOTHING`, + insertArgs..., + ); err != nil { + return fmt.Errorf("archive pending: %w", err) + } + return nil +} + func (s *PGPendingMessageStore) DeleteByKey(ctx context.Context, channelName, historyKey string) error { - tClause, tArgs, _, err := scopeClause(ctx, 3) + tClause, tArgs, nextParam, err := scopeClause(ctx, 3) if err != nil { return err } - _, err = s.db.ExecContext(ctx, - `DELETE FROM channel_pending_messages WHERE channel_name = $1 AND history_key = $2`+tClause, - append([]any{channelName, historyKey}, tArgs...)..., - ) - return err + args := append([]any{channelName, historyKey}, tArgs...) + where := `WHERE channel_name = $1 AND history_key = $2` + tClause + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("begin delete tx: %w", err) + } + defer tx.Rollback() + + if err := archivePending(ctx, tx, where, args, nextParam, store.ArchiveReasonConsumed); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `DELETE FROM channel_pending_messages `+where, args...); err != nil { + return fmt.Errorf("delete pending: %w", err) + } + return tx.Commit() } func (s *PGPendingMessageStore) Compact(ctx context.Context, deleteIDs []uuid.UUID, summary *store.PendingMessage) error { @@ -100,10 +140,12 @@ func (s *PGPendingMessageStore) Compact(ctx context.Context, deleteIDs []uuid.UU args[i] = id } - res, err := tx.ExecContext(ctx, - fmt.Sprintf("DELETE FROM channel_pending_messages WHERE id IN (%s)", strings.Join(placeholders, ",")), - args..., - ) + where := fmt.Sprintf("WHERE id IN (%s)", strings.Join(placeholders, ",")) + if err := archivePending(ctx, tx, where, args, len(deleteIDs)+1, store.ArchiveReasonCompacted); err != nil { + return err + } + + res, err := tx.ExecContext(ctx, "DELETE FROM channel_pending_messages "+where, args...) if err != nil { return fmt.Errorf("compact delete: %w", err) } @@ -133,15 +175,55 @@ func (s *PGPendingMessageStore) Compact(ctx context.Context, deleteIDs []uuid.UU func (s *PGPendingMessageStore) DeleteStale(ctx context.Context, olderThan time.Duration) (int64, error) { cutoff := time.Now().Add(-olderThan) - tid := tenantIDForInsert(ctx) - result, err := s.db.ExecContext(ctx, - `DELETE FROM channel_pending_messages WHERE updated_at < $1 AND tenant_id = $2`, - cutoff, tid, - ) + args := []any{cutoff, tenantIDForInsert(ctx)} + where := `WHERE updated_at < $1 AND tenant_id = $2` + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return 0, fmt.Errorf("begin delete stale tx: %w", err) + } + defer tx.Rollback() + + if err := archivePending(ctx, tx, where, args, len(args)+1, store.ArchiveReasonStale); err != nil { + return 0, err + } + result, err := tx.ExecContext(ctx, `DELETE FROM channel_pending_messages `+where, args...) if err != nil { return 0, err } - return result.RowsAffected() + affected, err := result.RowsAffected() + if err != nil { + return 0, err + } + if err := tx.Commit(); err != nil { + return 0, err + } + return affected, nil +} + +func (s *PGPendingMessageStore) ListArchivedByKey(ctx context.Context, channelName, historyKey string, since time.Time, limit int) ([]store.ArchivedMessage, error) { + tClause, tArgs, nextParam, err := scopeClause(ctx, 3) + if err != nil { + return nil, err + } + args := append([]any{channelName, historyKey}, tArgs...) + query := `SELECT ` + archivedReadColumns + ` + FROM channel_message_archive + WHERE channel_name = $1 AND history_key = $2` + tClause + if !since.IsZero() { + query += fmt.Sprintf(" AND created_at >= $%d", nextParam) + args = append(args, since) + nextParam++ + } + query += ` ORDER BY created_at ASC, id ASC` + if limit > 0 { + query += fmt.Sprintf(" LIMIT $%d", nextParam) + args = append(args, limit) + } + + var result []store.ArchivedMessage + err = pkgSqlxDB.SelectContext(ctx, &result, query, args...) + return result, err } func (s *PGPendingMessageStore) ListGroups(ctx context.Context) ([]store.PendingMessageGroup, error) { diff --git a/internal/store/sqlitestore/pending_message_archive_test.go b/internal/store/sqlitestore/pending_message_archive_test.go new file mode 100644 index 00000000..bc920500 --- /dev/null +++ b/internal/store/sqlitestore/pending_message_archive_test.go @@ -0,0 +1,218 @@ +//go:build sqlite || sqliteonly + +package sqlitestore + +import ( + "context" + "database/sql" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// newPendingFixture returns a store plus a tenant-scoped context with n group +// messages already buffered, oldest first. +func newPendingFixture(t *testing.T, slug string, n int) (*SQLitePendingMessageStore, context.Context, *sql.DB, []store.PendingMessage) { + t.Helper() + db := openTestDB(t) + if err := EnsureSchema(db); err != nil { + t.Fatalf("EnsureSchema: %v", err) + } + tenantID := uuid.New() + if _, err := db.Exec(`INSERT INTO tenants (id, name, slug, status) VALUES (?, 'T', ?, 'active')`, tenantID.String(), slug); err != nil { + t.Fatalf("insert tenant: %v", err) + } + ctx := store.WithTenantID(context.Background(), tenantID) + + base := time.Now().Add(-time.Duration(n) * time.Minute).UTC().Truncate(time.Second) + msgs := make([]store.PendingMessage, n) + for i := range msgs { + msgs[i] = store.PendingMessage{ + ID: uuid.Must(uuid.NewV7()), + ChannelName: "zalo-main", + HistoryKey: "group-1", + Sender: "Diệu Hương", + SenderID: "uid-1", + Body: "raw message " + uuid.NewString(), + PlatformMsgID: uuid.NewString(), + CreatedAt: base.Add(time.Duration(i) * time.Minute), + } + } + pending := NewSQLitePendingMessageStore(db) + if err := pending.AppendBatch(ctx, msgs); err != nil { + t.Fatalf("AppendBatch: %v", err) + } + return pending, ctx, db, msgs +} + +func archivedBodies(t *testing.T, archived []store.ArchivedMessage) map[string]string { + t.Helper() + out := make(map[string]string, len(archived)) + for _, a := range archived { + out[a.Body] = a.ArchiveReason + } + return out +} + +// Clearing the buffer after the bot is mentioned must not destroy the raw +// messages: they are the only stored copy of the group conversation. +func TestDeleteByKeyArchivesMessages(t *testing.T) { + pending, ctx, _, msgs := newPendingFixture(t, "t-archive-clear", 3) + + if err := pending.DeleteByKey(ctx, "zalo-main", "group-1"); err != nil { + t.Fatalf("DeleteByKey: %v", err) + } + + live, err := pending.ListByKey(ctx, "zalo-main", "group-1") + if err != nil { + t.Fatalf("ListByKey: %v", err) + } + if len(live) != 0 { + t.Fatalf("buffer has %d rows after clear, want 0", len(live)) + } + + archived, err := pending.ListArchivedByKey(ctx, "zalo-main", "group-1", time.Time{}, 0) + if err != nil { + t.Fatalf("ListArchivedByKey: %v", err) + } + if len(archived) != len(msgs) { + t.Fatalf("archived %d rows, want %d", len(archived), len(msgs)) + } + got := archivedBodies(t, archived) + for _, m := range msgs { + reason, ok := got[m.Body] + if !ok { + t.Fatalf("message %q missing from archive", m.Body) + } + if reason != store.ArchiveReasonConsumed { + t.Fatalf("archive reason = %q, want %q", reason, store.ArchiveReasonConsumed) + } + } + // Chronological order is what a replay depends on. + for i := 1; i < len(archived); i++ { + if archived[i].CreatedAt.Before(archived[i-1].CreatedAt) { + t.Fatalf("archive not ordered by created_at at index %d", i) + } + } +} + +// Compaction replaces old messages with an LLM summary. The summary is lossy, +// so the raw rows it replaces must survive in the archive. +func TestCompactArchivesReplacedMessages(t *testing.T) { + pending, ctx, _, msgs := newPendingFixture(t, "t-archive-compact", 5) + + deleteIDs := []uuid.UUID{msgs[0].ID, msgs[1].ID, msgs[2].ID} + summary := &store.PendingMessage{ + ChannelName: "zalo-main", + HistoryKey: "group-1", + Sender: "[summary]", + Body: "Concise summary of the first three messages.", + IsSummary: true, + } + if err := pending.Compact(ctx, deleteIDs, summary); err != nil { + t.Fatalf("Compact: %v", err) + } + + live, err := pending.ListByKey(ctx, "zalo-main", "group-1") + if err != nil { + t.Fatalf("ListByKey: %v", err) + } + if len(live) != 3 { // 2 kept raw + 1 summary + t.Fatalf("buffer has %d rows after compaction, want 3", len(live)) + } + + archived, err := pending.ListArchivedByKey(ctx, "zalo-main", "group-1", time.Time{}, 0) + if err != nil { + t.Fatalf("ListArchivedByKey: %v", err) + } + if len(archived) != 3 { + t.Fatalf("archived %d rows, want 3", len(archived)) + } + got := archivedBodies(t, archived) + for _, m := range msgs[:3] { + reason, ok := got[m.Body] + if !ok { + t.Fatalf("compacted message %q missing from archive", m.Body) + } + if reason != store.ArchiveReasonCompacted { + t.Fatalf("archive reason = %q, want %q", reason, store.ArchiveReasonCompacted) + } + } + for _, m := range msgs[3:] { + if _, ok := got[m.Body]; ok { + t.Fatalf("kept message %q must not be archived yet", m.Body) + } + } +} + +// A group that is compacted and later cleared must end up with every raw +// message archived exactly once — archived rows keep their original id. +func TestArchiveIsIdempotentAcrossCompactThenClear(t *testing.T) { + pending, ctx, db, msgs := newPendingFixture(t, "t-archive-idempotent", 4) + + summary := &store.PendingMessage{ + ChannelName: "zalo-main", + HistoryKey: "group-1", + Sender: "[summary]", + Body: "Summary.", + IsSummary: true, + } + if err := pending.Compact(ctx, []uuid.UUID{msgs[0].ID, msgs[1].ID}, summary); err != nil { + t.Fatalf("Compact: %v", err) + } + if err := pending.DeleteByKey(ctx, "zalo-main", "group-1"); err != nil { + t.Fatalf("DeleteByKey: %v", err) + } + // Replaying the clear must not duplicate rows. + if err := pending.DeleteByKey(ctx, "zalo-main", "group-1"); err != nil { + t.Fatalf("DeleteByKey replay: %v", err) + } + + archived, err := pending.ListArchivedByKey(ctx, "zalo-main", "group-1", time.Time{}, 0) + if err != nil { + t.Fatalf("ListArchivedByKey: %v", err) + } + // 4 raw + the summary row the clear removed. + if len(archived) != 5 { + t.Fatalf("archived %d rows, want 5", len(archived)) + } + got := archivedBodies(t, archived) + for _, m := range msgs { + if _, ok := got[m.Body]; !ok { + t.Fatalf("message %q missing from archive", m.Body) + } + } + + var distinct int + if err := db.QueryRow(`SELECT COUNT(DISTINCT id) FROM channel_message_archive`).Scan(&distinct); err != nil { + t.Fatalf("count distinct: %v", err) + } + if distinct != len(archived) { + t.Fatalf("archive has %d rows but %d distinct ids", len(archived), distinct) + } +} + +// The archive is tenant-scoped like every other read. +func TestListArchivedByKeyIsTenantScoped(t *testing.T) { + pending, ctx, db, _ := newPendingFixture(t, "t-archive-tenant", 2) + if err := pending.DeleteByKey(ctx, "zalo-main", "group-1"); err != nil { + t.Fatalf("DeleteByKey: %v", err) + } + + otherTenant := uuid.New() + if _, err := db.Exec(`INSERT INTO tenants (id, name, slug, status) VALUES (?, 'T2', 't-archive-tenant-2', 'active')`, otherTenant.String()); err != nil { + t.Fatalf("insert other tenant: %v", err) + } + otherCtx := store.WithTenantID(context.Background(), otherTenant) + + archived, err := pending.ListArchivedByKey(otherCtx, "zalo-main", "group-1", time.Time{}, 0) + if err != nil { + t.Fatalf("ListArchivedByKey: %v", err) + } + if len(archived) != 0 { + t.Fatalf("other tenant sees %d archived rows, want 0", len(archived)) + } +} diff --git a/internal/store/sqlitestore/pending_messages.go b/internal/store/sqlitestore/pending_messages.go index 498ad824..1a99c695 100644 --- a/internal/store/sqlitestore/pending_messages.go +++ b/internal/store/sqlitestore/pending_messages.go @@ -83,17 +83,58 @@ func (s *SQLitePendingMessageStore) ListByKey(ctx context.Context, channelName, return result, rows.Err() } +// archivedColumns are copied verbatim from the buffer into channel_message_archive. +const archivedColumns = `id, channel_name, history_key, parent_history_key, sender, sender_id, + body, platform_msg_id, is_summary, created_at, updated_at, tenant_id` + +// archivedReadColumns omits tenant_id: reads are already tenant-scoped by the +// WHERE clause and ArchivedMessage carries no tenant field. +const archivedReadColumns = `id, channel_name, history_key, parent_history_key, sender, sender_id, + body, platform_msg_id, is_summary, created_at, updated_at, archived_at, archive_reason` + +// archivePending copies every buffer row matching where into the archive. Callers +// pass the same where/args they are about to DELETE with, so the archive cannot +// miss a row the delete removes. Replaying is a no-op: archived rows keep their +// original id. +func archivePending(ctx context.Context, tx *sql.Tx, where string, args []any, reason string) error { + // SQLite binds `?` by position in the statement text. The reason placeholder + // sits in the SELECT list, ahead of every placeholder in where, so it must be + // bound first. + insertArgs := make([]any, 0, len(args)+1) + insertArgs = append(insertArgs, reason) + insertArgs = append(insertArgs, args...) + if _, err := tx.ExecContext(ctx, + `INSERT OR IGNORE INTO channel_message_archive (`+archivedColumns+`, archive_reason) + SELECT `+archivedColumns+`, ? + FROM channel_pending_messages `+where, + insertArgs..., + ); err != nil { + return fmt.Errorf("archive pending: %w", err) + } + return nil +} + func (s *SQLitePendingMessageStore) DeleteByKey(ctx context.Context, channelName, historyKey string) error { tClause, tArgs, err := scopeClause(ctx) if err != nil { return err } args := append([]any{channelName, historyKey}, tArgs...) - _, err = s.db.ExecContext(ctx, - `DELETE FROM channel_pending_messages WHERE channel_name = ? AND history_key = ?`+tClause, - args..., - ) - return err + where := `WHERE channel_name = ? AND history_key = ?` + tClause + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("begin delete tx: %w", err) + } + defer tx.Rollback() + + if err := archivePending(ctx, tx, where, args, store.ArchiveReasonConsumed); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `DELETE FROM channel_pending_messages `+where, args...); err != nil { + return fmt.Errorf("delete pending: %w", err) + } + return tx.Commit() } func (s *SQLitePendingMessageStore) Compact(ctx context.Context, deleteIDs []uuid.UUID, summary *store.PendingMessage) error { @@ -114,10 +155,12 @@ func (s *SQLitePendingMessageStore) Compact(ctx context.Context, deleteIDs []uui args[i] = id } - res, err := tx.ExecContext(ctx, - fmt.Sprintf("DELETE FROM channel_pending_messages WHERE id IN (%s)", strings.Join(placeholders, ",")), - args..., - ) + where := fmt.Sprintf("WHERE id IN (%s)", strings.Join(placeholders, ",")) + if err := archivePending(ctx, tx, where, args, store.ArchiveReasonCompacted); err != nil { + return err + } + + res, err := tx.ExecContext(ctx, "DELETE FROM channel_pending_messages "+where, args...) if err != nil { return fmt.Errorf("compact delete: %w", err) } @@ -144,16 +187,73 @@ func (s *SQLitePendingMessageStore) Compact(ctx context.Context, deleteIDs []uui } func (s *SQLitePendingMessageStore) DeleteStale(ctx context.Context, olderThan time.Duration) (int64, error) { - cutoff := time.Now().Add(-olderThan) - tid := tenantIDForInsert(ctx) - result, err := s.db.ExecContext(ctx, - `DELETE FROM channel_pending_messages WHERE updated_at < ? AND tenant_id = ?`, - cutoff, tid, - ) + args := []any{time.Now().Add(-olderThan), tenantIDForInsert(ctx)} + where := `WHERE updated_at < ? AND tenant_id = ?` + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return 0, fmt.Errorf("begin delete stale tx: %w", err) + } + defer tx.Rollback() + + if err := archivePending(ctx, tx, where, args, store.ArchiveReasonStale); err != nil { + return 0, err + } + result, err := tx.ExecContext(ctx, `DELETE FROM channel_pending_messages `+where, args...) if err != nil { return 0, err } - return result.RowsAffected() + affected, err := result.RowsAffected() + if err != nil { + return 0, err + } + if err := tx.Commit(); err != nil { + return 0, err + } + return affected, nil +} + +func (s *SQLitePendingMessageStore) ListArchivedByKey(ctx context.Context, channelName, historyKey string, since time.Time, limit int) ([]store.ArchivedMessage, error) { + tClause, tArgs, err := scopeClause(ctx) + if err != nil { + return nil, err + } + args := append([]any{channelName, historyKey}, tArgs...) + query := `SELECT ` + archivedReadColumns + ` + FROM channel_message_archive + WHERE channel_name = ? AND history_key = ?` + tClause + if !since.IsZero() { + query += ` AND created_at >= ?` + args = append(args, since) + } + query += ` ORDER BY created_at ASC, id ASC` + if limit > 0 { + query += ` LIMIT ?` + args = append(args, limit) + } + + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + var result []store.ArchivedMessage + for rows.Next() { + var m store.ArchivedMessage + createdAt, updatedAt := scanTimePair() + archivedAt := &sqliteTime{} + if err := rows.Scan(&m.ID, &m.ChannelName, &m.HistoryKey, &m.ParentHistoryKey, &m.Sender, &m.SenderID, + &m.Body, &m.PlatformMsgID, &m.IsSummary, createdAt, updatedAt, + archivedAt, &m.ArchiveReason); err != nil { + return nil, err + } + m.CreatedAt = createdAt.Time + m.UpdatedAt = updatedAt.Time + m.ArchivedAt = archivedAt.Time + result = append(result, m) + } + return result, rows.Err() } func (s *SQLitePendingMessageStore) ListGroups(ctx context.Context) ([]store.PendingMessageGroup, error) { diff --git a/internal/store/sqlitestore/schema.go b/internal/store/sqlitestore/schema.go index ef946e56..17f76e4d 100644 --- a/internal/store/sqlitestore/schema.go +++ b/internal/store/sqlitestore/schema.go @@ -16,7 +16,7 @@ var schemaSQL string // SchemaVersion is the current SQLite schema version. // Bump this when adding new migration steps below. -const SchemaVersion = 59 +const SchemaVersion = 60 // migrations maps version → SQL to apply when upgrading FROM that version. // schema.sql always represents the LATEST full schema (for fresh DBs). @@ -95,6 +95,27 @@ BEGIN END;` var migrations = map[int]string{ + // Version 59 → 60: keep an append-only copy of group capture. Pending rows are + // deleted when the buffer is handed to the agent and when compaction replaces + // them with a summary; before this table those deletes destroyed the only copy. + 59: `CREATE TABLE IF NOT EXISTS channel_message_archive ( + id TEXT NOT NULL PRIMARY KEY, + channel_name VARCHAR(100) NOT NULL, + history_key VARCHAR(200) NOT NULL, + parent_history_key VARCHAR(200) NOT NULL DEFAULT '', + sender VARCHAR(255) NOT NULL, + sender_id VARCHAR(255) NOT NULL DEFAULT '', + body TEXT NOT NULL, + platform_msg_id VARCHAR(100) NOT NULL DEFAULT '', + is_summary BOOLEAN NOT NULL DEFAULT 0, + tenant_id TEXT NOT NULL REFERENCES tenants(id), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + archived_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + archive_reason VARCHAR(20) NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_channel_message_archive_lookup ON channel_message_archive(tenant_id, channel_name, history_key, created_at); +CREATE INDEX IF NOT EXISTS idx_channel_message_archive_archived_at ON channel_message_archive(tenant_id, archived_at);`, // Version 58 → 59: scope persisted subagent tasks by immutable root-agent UUID. // Metadata is authoritative; key fallback is allowed only for one matching // agent that predates the task. Unmatched rows remain inaccessible. diff --git a/internal/store/sqlitestore/schema.sql b/internal/store/sqlitestore/schema.sql index 92539dee..540b04a3 100644 --- a/internal/store/sqlitestore/schema.sql +++ b/internal/store/sqlitestore/schema.sql @@ -1188,6 +1188,32 @@ CREATE INDEX IF NOT EXISTS idx_channel_pending_messages_lookup ON channel_pendin CREATE INDEX IF NOT EXISTS idx_channel_pending_messages_parent ON channel_pending_messages(channel_name, parent_history_key) WHERE parent_history_key <> ''; CREATE INDEX IF NOT EXISTS idx_channel_pending_messages_tenant ON channel_pending_messages(tenant_id); +-- ============================================================ +-- Table: channel_message_archive +-- ============================================================ +-- Append-only copy of every pending message taken before it leaves the buffer. +-- Rows keep their original id so a replayed delete cannot duplicate them. + +CREATE TABLE IF NOT EXISTS channel_message_archive ( + id TEXT NOT NULL PRIMARY KEY, + channel_name VARCHAR(100) NOT NULL, + history_key VARCHAR(200) NOT NULL, + parent_history_key VARCHAR(200) NOT NULL DEFAULT '', + sender VARCHAR(255) NOT NULL, + sender_id VARCHAR(255) NOT NULL DEFAULT '', + body TEXT NOT NULL, + platform_msg_id VARCHAR(100) NOT NULL DEFAULT '', + is_summary BOOLEAN NOT NULL DEFAULT 0, + tenant_id TEXT NOT NULL REFERENCES tenants(id), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + archived_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + archive_reason VARCHAR(20) NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_channel_message_archive_lookup ON channel_message_archive(tenant_id, channel_name, history_key, created_at); +CREATE INDEX IF NOT EXISTS idx_channel_message_archive_archived_at ON channel_message_archive(tenant_id, archived_at); + -- ============================================================ -- Table: channel_memory_extraction_runs -- ============================================================ diff --git a/internal/upgrade/version.go b/internal/upgrade/version.go index bc2504b5..f143a0fd 100644 --- a/internal/upgrade/version.go +++ b/internal/upgrade/version.go @@ -2,4 +2,4 @@ package upgrade // RequiredSchemaVersion is the schema migration version this binary requires. // Bump this whenever adding a new SQL migration file. -const RequiredSchemaVersion uint = 96 +const RequiredSchemaVersion uint = 97 diff --git a/migrations/000097_channel_message_archive.down.sql b/migrations/000097_channel_message_archive.down.sql new file mode 100644 index 00000000..250a2a11 --- /dev/null +++ b/migrations/000097_channel_message_archive.down.sql @@ -0,0 +1,3 @@ +DROP INDEX IF EXISTS idx_channel_message_archive_archived_at; +DROP INDEX IF EXISTS idx_channel_message_archive_lookup; +DROP TABLE IF EXISTS channel_message_archive; diff --git a/migrations/000097_channel_message_archive.up.sql b/migrations/000097_channel_message_archive.up.sql new file mode 100644 index 00000000..8c00564d --- /dev/null +++ b/migrations/000097_channel_message_archive.up.sql @@ -0,0 +1,35 @@ +-- Append-only archive of group capture. +-- +-- channel_pending_messages is a live buffer: rows leave it when the bot is +-- mentioned (the buffer is handed to the agent, then dropped) and when LLM +-- compaction replaces old rows with a summary. Both paths deleted the only +-- stored copy of the raw group messages, so anything that had not already been +-- read out of the buffer was unrecoverable. +-- +-- Every row is copied here before it is deleted from the buffer. Rows keep +-- their original id so replaying a delete cannot duplicate them. + +CREATE TABLE channel_message_archive ( + id UUID PRIMARY KEY, + channel_name VARCHAR(100) NOT NULL, + history_key VARCHAR(200) NOT NULL, + parent_history_key VARCHAR(200) NOT NULL DEFAULT '', + sender VARCHAR(255) NOT NULL, + sender_id VARCHAR(255) NOT NULL DEFAULT '', + body TEXT NOT NULL, + platform_msg_id VARCHAR(100) NOT NULL DEFAULT '', + is_summary BOOLEAN NOT NULL DEFAULT false, + created_at TIMESTAMPTZ NOT NULL, + updated_at TIMESTAMPTZ NOT NULL, + tenant_id UUID NOT NULL REFERENCES tenants(id), + archived_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + archive_reason VARCHAR(20) NOT NULL +); + +-- Primary read path: replay one group in chronological order. +CREATE INDEX idx_channel_message_archive_lookup + ON channel_message_archive (tenant_id, channel_name, history_key, created_at); + +-- Retention sweeps and "what was archived recently" queries. +CREATE INDEX idx_channel_message_archive_archived_at + ON channel_message_archive (tenant_id, archived_at); diff --git a/tests/integration/pending_message_archive_test.go b/tests/integration/pending_message_archive_test.go new file mode 100644 index 00000000..855ca005 --- /dev/null +++ b/tests/integration/pending_message_archive_test.go @@ -0,0 +1,204 @@ +//go:build integration + +package integration + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/internal/store/pg" +) + +// seedPendingGroup buffers n group messages for a fresh tenant and returns the +// store, a tenant-scoped context, and the messages in chronological order. +func seedPendingGroup(t *testing.T, historyKey string, n int) (*pg.PGPendingMessageStore, context.Context, []store.PendingMessage) { + t.Helper() + db := testDB(t) + tenantID, _ := seedTenantAgent(t, db) + ctx := store.WithTenantID(context.Background(), tenantID) + + base := time.Now().Add(-time.Duration(n) * time.Minute).UTC().Truncate(time.Millisecond) + msgs := make([]store.PendingMessage, n) + for i := range msgs { + msgs[i] = store.PendingMessage{ + ID: uuid.Must(uuid.NewV7()), + ChannelName: "dangzalo", + HistoryKey: historyKey, + Sender: "Diệu Hương", + SenderID: "uid-1", + Body: "raw message " + uuid.NewString(), + PlatformMsgID: uuid.NewString(), + CreatedAt: base.Add(time.Duration(i) * time.Minute), + } + } + + s := pg.NewPGPendingMessageStore(db) + if err := s.AppendBatch(ctx, msgs); err != nil { + t.Fatalf("AppendBatch: %v", err) + } + // AppendBatch stamps server time on every row, so a batch shares one + // created_at. Spread them to exercise ordering and the `since` filter. + for _, m := range msgs { + if _, err := db.Exec( + `UPDATE channel_pending_messages SET created_at = $1 WHERE id = $2`, m.CreatedAt, m.ID, + ); err != nil { + t.Fatalf("spread created_at: %v", err) + } + } + t.Cleanup(func() { + db.Exec("DELETE FROM channel_message_archive WHERE tenant_id = $1", tenantID) + db.Exec("DELETE FROM channel_pending_messages WHERE tenant_id = $1", tenantID) + }) + return s, ctx, msgs +} + +// Handing the buffer to the agent (bot mentioned) must preserve the raw group +// messages: the buffer used to be their only stored copy. +func TestPGDeleteByKeyArchivesMessages(t *testing.T) { + historyKey := "group-" + uuid.NewString() + s, ctx, msgs := seedPendingGroup(t, historyKey, 3) + + if err := s.DeleteByKey(ctx, "dangzalo", historyKey); err != nil { + t.Fatalf("DeleteByKey: %v", err) + } + + live, err := s.ListByKey(ctx, "dangzalo", historyKey) + if err != nil { + t.Fatalf("ListByKey: %v", err) + } + if len(live) != 0 { + t.Fatalf("buffer has %d rows after clear, want 0", len(live)) + } + + archived, err := s.ListArchivedByKey(ctx, "dangzalo", historyKey, time.Time{}, 0) + if err != nil { + t.Fatalf("ListArchivedByKey: %v", err) + } + if len(archived) != len(msgs) { + t.Fatalf("archived %d rows, want %d", len(archived), len(msgs)) + } + bodies := make(map[string]string, len(archived)) + for _, a := range archived { + bodies[a.Body] = a.ArchiveReason + } + for _, m := range msgs { + reason, ok := bodies[m.Body] + if !ok { + t.Fatalf("message %q missing from archive", m.Body) + } + if reason != store.ArchiveReasonConsumed { + t.Fatalf("archive reason = %q, want %q", reason, store.ArchiveReasonConsumed) + } + } +} + +// Compaction summarizes old messages with an LLM and drops the originals. The +// summary is lossy, so the raw rows must land in the archive. +func TestPGCompactArchivesReplacedMessages(t *testing.T) { + historyKey := "group-" + uuid.NewString() + s, ctx, msgs := seedPendingGroup(t, historyKey, 5) + + summary := &store.PendingMessage{ + ChannelName: "dangzalo", + HistoryKey: historyKey, + Sender: "[summary]", + Body: "Concise summary of the first three messages.", + IsSummary: true, + } + if err := s.Compact(ctx, []uuid.UUID{msgs[0].ID, msgs[1].ID, msgs[2].ID}, summary); err != nil { + t.Fatalf("Compact: %v", err) + } + + live, err := s.ListByKey(ctx, "dangzalo", historyKey) + if err != nil { + t.Fatalf("ListByKey: %v", err) + } + if len(live) != 3 { // 2 kept raw + 1 summary + t.Fatalf("buffer has %d rows after compaction, want 3", len(live)) + } + + archived, err := s.ListArchivedByKey(ctx, "dangzalo", historyKey, time.Time{}, 0) + if err != nil { + t.Fatalf("ListArchivedByKey: %v", err) + } + if len(archived) != 3 { + t.Fatalf("archived %d rows, want 3", len(archived)) + } + for _, a := range archived { + if a.ArchiveReason != store.ArchiveReasonCompacted { + t.Fatalf("archive reason = %q, want %q", a.ArchiveReason, store.ArchiveReasonCompacted) + } + } + // `since` must filter on the original capture time, not the archive time. + recent, err := s.ListArchivedByKey(ctx, "dangzalo", historyKey, msgs[2].CreatedAt, 0) + if err != nil { + t.Fatalf("ListArchivedByKey(since): %v", err) + } + if len(recent) != 1 || recent[0].Body != msgs[2].Body { + t.Fatalf("since filter returned %d rows, want only the newest compacted message", len(recent)) + } +} + +// Compact then clear must archive every raw row exactly once. +func TestPGArchiveIsIdempotentAcrossCompactThenClear(t *testing.T) { + historyKey := "group-" + uuid.NewString() + s, ctx, msgs := seedPendingGroup(t, historyKey, 4) + + summary := &store.PendingMessage{ + ChannelName: "dangzalo", + HistoryKey: historyKey, + Sender: "[summary]", + Body: "Summary.", + IsSummary: true, + } + if err := s.Compact(ctx, []uuid.UUID{msgs[0].ID, msgs[1].ID}, summary); err != nil { + t.Fatalf("Compact: %v", err) + } + if err := s.DeleteByKey(ctx, "dangzalo", historyKey); err != nil { + t.Fatalf("DeleteByKey: %v", err) + } + if err := s.DeleteByKey(ctx, "dangzalo", historyKey); err != nil { + t.Fatalf("DeleteByKey replay: %v", err) + } + + archived, err := s.ListArchivedByKey(ctx, "dangzalo", historyKey, time.Time{}, 0) + if err != nil { + t.Fatalf("ListArchivedByKey: %v", err) + } + // 4 raw + the summary row the clear removed. + if len(archived) != 5 { + t.Fatalf("archived %d rows, want 5", len(archived)) + } + seen := make(map[uuid.UUID]bool, len(archived)) + for _, a := range archived { + if seen[a.ID] { + t.Fatalf("archive contains duplicate id %s", a.ID) + } + seen[a.ID] = true + } +} + +// Archived rows stay inside their tenant. +func TestPGListArchivedByKeyIsTenantScoped(t *testing.T) { + historyKey := "group-" + uuid.NewString() + s, ctx, _ := seedPendingGroup(t, historyKey, 2) + if err := s.DeleteByKey(ctx, "dangzalo", historyKey); err != nil { + t.Fatalf("DeleteByKey: %v", err) + } + + db := testDB(t) + otherTenant, _ := seedTenantAgent(t, db) + otherCtx := store.WithTenantID(context.Background(), otherTenant) + + archived, err := s.ListArchivedByKey(otherCtx, "dangzalo", historyKey, time.Time{}, 0) + if err != nil { + t.Fatalf("ListArchivedByKey: %v", err) + } + if len(archived) != 0 { + t.Fatalf("other tenant sees %d archived rows, want 0", len(archived)) + } +}