//go:build sqlite || sqliteonly package sqlitestore import ( "context" "database/sql" "fmt" "strings" "time" "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/store" ) // SQLitePendingMessageStore implements store.PendingMessageStore backed by SQLite. type SQLitePendingMessageStore struct { db *sql.DB } func NewSQLitePendingMessageStore(db *sql.DB) *SQLitePendingMessageStore { return &SQLitePendingMessageStore{db: db} } func (s *SQLitePendingMessageStore) AppendBatch(ctx context.Context, msgs []store.PendingMessage) error { if len(msgs) == 0 { return nil } const cols = 12 placeholders := make([]string, len(msgs)) args := make([]any, 0, len(msgs)*cols) now := time.Now() tid := tenantIDForInsert(ctx) for i := range msgs { if msgs[i].ID == uuid.Nil { msgs[i].ID = uuid.Must(uuid.NewV7()) } placeholders[i] = "(?,?,?,?,?,?,?,?,?,?,?,?)" args = append(args, msgs[i].ID, msgs[i].ChannelName, msgs[i].HistoryKey, msgs[i].ParentHistoryKey, msgs[i].Sender, msgs[i].SenderID, msgs[i].Body, msgs[i].PlatformMsgID, msgs[i].IsSummary, now, now, tid) } _, err := s.db.ExecContext(ctx, `INSERT INTO channel_pending_messages (id, channel_name, history_key, parent_history_key, sender, sender_id, body, platform_msg_id, is_summary, created_at, updated_at, tenant_id) VALUES `+strings.Join(placeholders, ","), args..., ) return err } func (s *SQLitePendingMessageStore) ListByKey(ctx context.Context, channelName, historyKey string) ([]store.PendingMessage, error) { tClause, tArgs, err := scopeClause(ctx) if err != nil { return nil, err } args := append([]any{channelName, historyKey}, tArgs...) rows, err := s.db.QueryContext(ctx, `SELECT id, channel_name, history_key, parent_history_key, sender, sender_id, body, platform_msg_id, is_summary, created_at, updated_at FROM channel_pending_messages WHERE channel_name = ? AND history_key = ?`+tClause+` ORDER BY created_at ASC, id ASC`, args..., ) if err != nil { return nil, err } defer rows.Close() var result []store.PendingMessage for rows.Next() { var m store.PendingMessage createdAt, updatedAt := scanTimePair() if err := rows.Scan(&m.ID, &m.ChannelName, &m.HistoryKey, &m.ParentHistoryKey, &m.Sender, &m.SenderID, &m.Body, &m.PlatformMsgID, &m.IsSummary, createdAt, updatedAt); err != nil { return nil, err } m.CreatedAt = createdAt.Time m.UpdatedAt = updatedAt.Time result = append(result, m) } 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...) 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 { if len(deleteIDs) == 0 { return nil } tx, err := s.db.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("begin compact tx: %w", err) } defer tx.Rollback() placeholders := make([]string, len(deleteIDs)) args := make([]any, len(deleteIDs)) for i, id := range deleteIDs { placeholders[i] = "?" args[i] = id } 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) } affected, _ := res.RowsAffected() if affected == 0 { return nil } if summary.ID == uuid.Nil { summary.ID = uuid.Must(uuid.NewV7()) } now := time.Now() _, err = tx.ExecContext(ctx, `INSERT INTO channel_pending_messages (id, channel_name, history_key, parent_history_key, sender, sender_id, body, platform_msg_id, is_summary, created_at, updated_at, tenant_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, summary.ID, summary.ChannelName, summary.HistoryKey, summary.ParentHistoryKey, summary.Sender, summary.SenderID, summary.Body, summary.PlatformMsgID, true, now, now, tenantIDForInsert(ctx), ) if err != nil { return fmt.Errorf("compact insert summary: %w", err) } return tx.Commit() } func (s *SQLitePendingMessageStore) DeleteStale(ctx context.Context, olderThan time.Duration) (int64, error) { 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 } 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) { tClause, tArgs, err := scopeClause(ctx) if err != nil { return nil, err } // SQLite: BOOL_OR → MAX(is_summary), EXISTS subquery logic preserved q := `SELECT channel_name, history_key, MAX(parent_history_key) AS parent_history_key, COUNT(*) AS message_count, MAX(is_summary) AND NOT EXISTS ( SELECT 1 FROM channel_pending_messages n WHERE n.channel_name = m.channel_name AND n.history_key = m.history_key AND NOT n.is_summary AND n.created_at > ( SELECT MAX(s.created_at) FROM channel_pending_messages s WHERE s.channel_name = m.channel_name AND s.history_key = m.history_key AND s.is_summary ) ) AS has_summary, MAX(created_at) AS last_activity FROM channel_pending_messages m WHERE 1=1` + tClause + ` GROUP BY channel_name, history_key ORDER BY last_activity DESC` rows, err := s.db.QueryContext(ctx, q, tArgs...) if err != nil { return nil, err } defer rows.Close() var result []store.PendingMessageGroup for rows.Next() { var g store.PendingMessageGroup if err := rows.Scan(&g.ChannelName, &g.HistoryKey, &g.ParentHistoryKey, &g.MessageCount, &g.HasSummary, &g.LastActivity); err != nil { return nil, err } result = append(result, g) } return result, rows.Err() } func (s *SQLitePendingMessageStore) CountAll(ctx context.Context) (int64, error) { tClause, tArgs, err := scopeClause(ctx) if err != nil { return 0, err } var count int64 err = s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM channel_pending_messages WHERE 1=1`+tClause, tArgs..., ).Scan(&count) return count, err } func (s *SQLitePendingMessageStore) CountByKey(ctx context.Context, channelName, historyKey string) (int, error) { tClause, tArgs, err := scopeClause(ctx) if err != nil { return 0, err } args := append([]any{channelName, historyKey}, tArgs...) var count int err = s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM channel_pending_messages WHERE channel_name = ? AND history_key = ?`+tClause, args..., ).Scan(&count) return count, err } func (s *SQLitePendingMessageStore) ResolveGroupTitles(ctx context.Context, groups []store.PendingMessageGroup) (map[string]string, error) { if len(groups) == 0 { return nil, nil } result, missing := s.resolveGroupTitlesFromContacts(ctx, groups) if len(missing) == 0 { return result, nil } // Build OR conditions using LIKE with ? placeholders conditions := make([]string, 0, len(missing)) args := make([]any, 0, len(missing)*2) for _, g := range missing { conditions = append(conditions, "(session_key LIKE '%:' || ? || ':group:' || ? || '%')") args = append(args, g.ChannelName, g.HistoryKey) } if !store.IsCrossTenant(ctx) { tid := store.TenantIDFromContext(ctx) if tid == uuid.Nil { tid = store.MasterTenantID } args = append(args, tid) } tenantFilter := "" if !store.IsCrossTenant(ctx) { tenantFilter = " AND tenant_id = ?" } rows, err := s.db.QueryContext(ctx, "SELECT session_key, json_extract(metadata, '$.chat_title')"+ " FROM sessions"+ " WHERE json_extract(metadata, '$.chat_title') != ''"+ " AND ("+strings.Join(conditions, " OR ")+")"+tenantFilter, args..., ) if err != nil { return nil, err } defer rows.Close() for rows.Next() { var sessionKey, title string if err := rows.Scan(&sessionKey, &title); err != nil { return nil, err } for _, g := range missing { pattern := ":" + g.ChannelName + ":group:" + g.HistoryKey if strings.Contains(sessionKey, pattern) { mapKey := g.ChannelName + ":" + g.HistoryKey if _, exists := result[mapKey]; !exists { result[mapKey] = title } break } } } return result, rows.Err() } func (s *SQLitePendingMessageStore) resolveGroupTitlesFromContacts(ctx context.Context, groups []store.PendingMessageGroup) (map[string]string, []store.PendingMessageGroup) { result := make(map[string]string, len(groups)) unique := uniquePendingTitleGroups(groups) if len(unique) == 0 { return result, nil } conditions := make([]string, 0, len(unique)) args := make([]any, 0, len(unique)*2+1) for _, g := range unique { conditions = append(conditions, "(channel_instance = ? AND sender_id = ?)") args = append(args, g.ChannelName, g.HistoryKey) } tenantFilter := "" if !store.IsCrossTenant(ctx) { tid := store.TenantIDFromContext(ctx) if tid == uuid.Nil { tid = store.MasterTenantID } tenantFilter = " AND tenant_id = ?" args = append(args, tid.String()) } rows, err := s.db.QueryContext(ctx, `SELECT channel_instance, sender_id, COALESCE(NULLIF(json_extract(metadata, '$.display_title'), ''), display_name) FROM channel_contacts WHERE display_name IS NOT NULL AND display_name <> '' AND contact_type IN ('group', 'topic') AND (`+strings.Join(conditions, " OR ")+`)`+tenantFilter, args..., ) if err != nil { return result, unique } defer rows.Close() for rows.Next() { var channelName, senderID, title string if err := rows.Scan(&channelName, &senderID, &title); err != nil { return result, unique } result[channelName+":"+senderID] = title } if err := rows.Err(); err != nil { return result, unique } missing := make([]store.PendingMessageGroup, 0, len(unique)-len(result)) for _, g := range unique { if result[g.ChannelName+":"+g.HistoryKey] == "" { missing = append(missing, g) } } return result, missing } func uniquePendingTitleGroups(groups []store.PendingMessageGroup) []store.PendingMessageGroup { out := make([]store.PendingMessageGroup, 0, len(groups)) seen := make(map[string]struct{}, len(groups)) for _, g := range groups { if g.ChannelName == "" || g.HistoryKey == "" { continue } key := g.ChannelName + ":" + g.HistoryKey if _, ok := seen[key]; ok { continue } seen[key] = struct{}{} out = append(out, g) } return out }