From 6b29ea056c47295f9efe1d0aa812bdb79a3dd7da Mon Sep 17 00:00:00 2001 From: viettranx Date: Mon, 23 Mar 2026 17:28:49 +0700 Subject: [PATCH] fix(channels): propagate tenant_id to pending history for compaction/sweep Channel factories create PendingHistory before InstanceLoader calls SetTenantID on the channel, so tenantID was always uuid.Nil causing "tenant_id required" errors on compaction sweep and auto-compact. Fix: add SetTenantID to PendingHistory and propagate it from InstanceLoader after channel creation. Also fix HTTP compact endpoint using context.Background() instead of preserving request tenant context. --- internal/channels/discord/discord.go | 4 ++++ internal/channels/feishu/feishu.go | 5 +++++ internal/channels/history.go | 7 +++++++ internal/channels/instance_loader.go | 5 +++++ internal/channels/slack/channel.go | 4 ++++ internal/channels/telegram/channel.go | 4 ++++ internal/channels/zalo/personal/channel.go | 5 +++++ internal/http/pending_messages.go | 3 ++- 8 files changed, 36 insertions(+), 1 deletion(-) diff --git a/internal/channels/discord/discord.go b/internal/channels/discord/discord.go index 89f90dbd..d995cf56 100644 --- a/internal/channels/discord/discord.go +++ b/internal/channels/discord/discord.go @@ -8,6 +8,7 @@ import ( "time" "github.com/bwmarrin/discordgo" + "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/channels" @@ -110,6 +111,9 @@ func (c *Channel) SetPendingCompaction(cfg *channels.CompactionConfig) { c.groupHistory.SetCompactionConfig(cfg) } +// SetPendingHistoryTenantID propagates tenant_id to the pending history for DB operations. +func (c *Channel) SetPendingHistoryTenantID(id uuid.UUID) { c.groupHistory.SetTenantID(id) } + // Stop closes the Discord gateway connection. func (c *Channel) Stop(_ context.Context) error { c.groupHistory.StopFlusher() diff --git a/internal/channels/feishu/feishu.go b/internal/channels/feishu/feishu.go index 7db0c75d..fd67abff 100644 --- a/internal/channels/feishu/feishu.go +++ b/internal/channels/feishu/feishu.go @@ -15,6 +15,8 @@ import ( "sync" "time" + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/channels" "github.com/nextlevelbuilder/goclaw/internal/config" @@ -127,6 +129,9 @@ func (c *Channel) SetPendingCompaction(cfg *channels.CompactionConfig) { c.groupHistory.SetCompactionConfig(cfg) } +// SetPendingHistoryTenantID propagates tenant_id to the pending history for DB operations. +func (c *Channel) SetPendingHistoryTenantID(id uuid.UUID) { c.groupHistory.SetTenantID(id) } + // Stop shuts down the Feishu channel. func (c *Channel) Stop(_ context.Context) error { c.groupHistory.StopFlusher() diff --git a/internal/channels/history.go b/internal/channels/history.go index 4360945f..0e503435 100644 --- a/internal/channels/history.go +++ b/internal/channels/history.go @@ -116,6 +116,13 @@ func MakeHistory(channelName string, s store.PendingMessageStore, tenantID uuid. return NewPendingHistory() } +// SetTenantID updates the tenant scope for DB operations. +// Called by InstanceLoader after channel creation to fix initialization order +// (factory captures uuid.Nil because SetTenantID on BaseChannel hasn't been called yet). +func (ph *PendingHistory) SetTenantID(id uuid.UUID) { + ph.tenantID = id +} + // tenantCtx returns a context with the tenant ID set for DB operations. func (ph *PendingHistory) tenantCtx() context.Context { ctx := context.Background() diff --git a/internal/channels/instance_loader.go b/internal/channels/instance_loader.go index 9022d9db..86f16240 100644 --- a/internal/channels/instance_loader.go +++ b/internal/channels/instance_loader.go @@ -240,6 +240,11 @@ func (l *InstanceLoader) loadInstance(ctx context.Context, inst store.ChannelIns if base, ok := ch.(interface{ SetTenantID(uuid.UUID) }); ok { base.SetTenantID(inst.TenantID) } + // Propagate tenant_id to pending history for compaction/sweep DB operations. + // Factory creates PendingHistory before SetTenantID is called, so tenantID is uuid.Nil at construction. + if ph, ok := ch.(interface{ SetPendingHistoryTenantID(uuid.UUID) }); ok { + ph.SetPendingHistoryTenantID(inst.TenantID) + } // Wire pending message auto-compaction. // Priority: config provider/model > agent's provider/model > fallback. diff --git a/internal/channels/slack/channel.go b/internal/channels/slack/channel.go index 4c6954f3..61538cae 100644 --- a/internal/channels/slack/channel.go +++ b/internal/channels/slack/channel.go @@ -8,6 +8,7 @@ import ( "sync" "time" + "github.com/google/uuid" slackapi "github.com/slack-go/slack" "github.com/slack-go/slack/socketmode" @@ -273,6 +274,9 @@ func (c *Channel) SetPendingCompaction(cfg *channels.CompactionConfig) { c.groupHistory.SetCompactionConfig(cfg) } +// SetPendingHistoryTenantID propagates tenant_id to the pending history for DB operations. +func (c *Channel) SetPendingHistoryTenantID(id uuid.UUID) { c.groupHistory.SetTenantID(id) } + // Stop gracefully shuts down the Slack channel. func (c *Channel) Stop(_ context.Context) error { c.groupHistory.StopFlusher() diff --git a/internal/channels/telegram/channel.go b/internal/channels/telegram/channel.go index 8cbbeeaf..16c3dd13 100644 --- a/internal/channels/telegram/channel.go +++ b/internal/channels/telegram/channel.go @@ -11,6 +11,7 @@ import ( "sync" "time" + "github.com/google/uuid" "github.com/mymmrac/telego" "github.com/nextlevelbuilder/goclaw/internal/bus" @@ -277,6 +278,9 @@ func (c *Channel) SetPendingCompaction(cfg *channels.CompactionConfig) { c.groupHistory.SetCompactionConfig(cfg) } +// SetPendingHistoryTenantID propagates tenant_id to the pending history for DB operations. +func (c *Channel) SetPendingHistoryTenantID(id uuid.UUID) { c.groupHistory.SetTenantID(id) } + // Stop shuts down the Telegram bot by cancelling the long polling context // and waiting for the polling goroutine to exit. func (c *Channel) Stop(_ context.Context) error { diff --git a/internal/channels/zalo/personal/channel.go b/internal/channels/zalo/personal/channel.go index 07dd74ed..664b8046 100644 --- a/internal/channels/zalo/personal/channel.go +++ b/internal/channels/zalo/personal/channel.go @@ -6,6 +6,8 @@ import ( "log/slog" "sync" + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/channels" "github.com/nextlevelbuilder/goclaw/internal/channels/typing" @@ -128,6 +130,9 @@ func (c *Channel) SetPendingCompaction(cfg *channels.CompactionConfig) { c.groupHistory.SetCompactionConfig(cfg) } +// SetPendingHistoryTenantID propagates tenant_id to the pending history for DB operations. +func (c *Channel) SetPendingHistoryTenantID(id uuid.UUID) { c.groupHistory.SetTenantID(id) } + // Stop gracefully shuts down the Zalo Personal channel. func (c *Channel) Stop(_ context.Context) error { c.groupHistory.StopFlusher() diff --git a/internal/http/pending_messages.go b/internal/http/pending_messages.go index bb3b3e93..4f612de7 100644 --- a/internal/http/pending_messages.go +++ b/internal/http/pending_messages.go @@ -146,8 +146,9 @@ func (h *PendingMessagesHandler) handleCompact(w http.ResponseWriter, r *http.Re if keepRecent <= 0 { keepRecent = 15 } + tenantID := store.TenantIDFromContext(r.Context()) go func() { - ctx, cancel := context.WithTimeout(context.Background(), 180*time.Second) + ctx, cancel := context.WithTimeout(store.WithTenantID(context.Background(), tenantID), 180*time.Second) defer cancel() remaining, err := channels.CompactGroup(ctx, h.store, req.ChannelName, req.HistoryKey, provider, model, keepRecent, h.maxTokens) if err != nil {