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.
This commit is contained in:
viettranx committed 2026-03-23 17:28:49 +07:00
1 parent 59a52f1e2f
commit 6b29ea056c
8 files changed
+36 -1

No files matched your search

+4
View File
@@ -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()
+5
View File
@@ -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()
+7
View File
@@ -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()
+5
View File
@@ -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.
+4
View File
@@ -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()
+4
View File
@@ -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 {
@@ -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()
+2 -1
View File
@@ -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 {