diff --git a/cmd/gateway.go b/cmd/gateway.go index 8cc54660..324440ce 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -588,7 +588,7 @@ func runGateway() { registerConfigChannels(cfg, channelMgr, msgBus, pgStores, instanceLoader, audioMgr) // Register channels/instances/links/teams RPC methods - chInstancesM := wireChannelRPCMethods(server, pgStores, channelMgr, agentRouter, msgBus, workspace) + chInstancesM := wireChannelRPCMethods(server, pgStores, channelMgr, instanceLoader, agentRouter, msgBus, workspace) // Bitrix24 orphan-bot cleaner. Fires from channel_instances delete handler // when the channel is no longer loaded in the Manager (typical scenario: diff --git a/cmd/gateway_channels_setup.go b/cmd/gateway_channels_setup.go index 3e902060..eb81e188 100644 --- a/cmd/gateway_channels_setup.go +++ b/cmd/gateway_channels_setup.go @@ -148,7 +148,7 @@ func registerConfigChannels(cfg *config.Config, channelMgr *channels.Manager, ms // Returns the channel-instances methods handler so the caller can register // per-channel-type orphan cleaners (e.g. Bitrix24 imbot.unregister) after // per-channel dependencies (portal store, encryption key) are in scope. -func wireChannelRPCMethods(server *gateway.Server, pgStores *store.Stores, channelMgr *channels.Manager, agentRouter *agent.Router, msgBus *bus.MessageBus, dataDir string) *methods.ChannelInstancesMethods { +func wireChannelRPCMethods(server *gateway.Server, pgStores *store.Stores, channelMgr *channels.Manager, instanceLoader *channels.InstanceLoader, agentRouter *agent.Router, msgBus *bus.MessageBus, dataDir string) *methods.ChannelInstancesMethods { // Register channels RPC methods (after channelMgr is initialized with all channels) methods.NewChannelsMethods(channelMgr).Register(server.Router()) @@ -159,7 +159,7 @@ func wireChannelRPCMethods(server *gateway.Server, pgStores *store.Stores, chann chInstancesM.Register(server.Router()) zalomethods.NewQRMethods(pgStores.ChannelInstances, msgBus).Register(server.Router()) zalomethods.NewContactsMethods(pgStores.ChannelInstances).Register(server.Router()) - whatsapp.NewQRMethods(pgStores.ChannelInstances, channelMgr).Register(server.Router()) + whatsapp.NewQRMethods(pgStores.ChannelInstances, channelMgr, instanceLoader).Register(server.Router()) } // Register agent links WS RPC methods @@ -196,6 +196,19 @@ func wireChannelEventSubscribers( if !ok || payload.Kind != bus.CacheKindChannelInstances { return } + if payload.Key != "" { + id, err := uuid.Parse(payload.Key) + if err != nil { + slog.Warn("channel instance targeted reload skipped: invalid id", "id", payload.Key, "error", err) + return + } + go func() { + if err := instanceLoader.LoadInstanceByID(store.WithCrossTenant(context.Background()), id); err != nil { + slog.Warn("channel instance targeted reload failed", "id", payload.Key, "error", err) + } + }() + return + } go instanceLoader.Reload(context.Background()) }) } @@ -277,4 +290,3 @@ func wireChannelEventSubscribers( }) } } - diff --git a/internal/channels/instance_loader.go b/internal/channels/instance_loader.go index 7f845d80..e795cb7a 100644 --- a/internal/channels/instance_loader.go +++ b/internal/channels/instance_loader.go @@ -157,6 +157,37 @@ func (l *InstanceLoader) Reload(ctx context.Context) { slog.Info("channel instances reloaded", "count", registered) } +// LoadInstanceByID loads or refreshes one enabled channel instance without +// stopping unrelated channels. Used by create-time cache invalidation and QR +// auth flows that need the just-created channel to become available quickly. +func (l *InstanceLoader) LoadInstanceByID(ctx context.Context, id uuid.UUID) error { + l.mu.Lock() + defer l.mu.Unlock() + + inst, err := l.store.Get(ctx, id) + if err != nil { + return err + } + + if _, ok := l.loaded[inst.Name]; ok { + if _, exists := l.manager.GetChannel(inst.Name); exists { + return nil + } + delete(l.loaded, inst.Name) + } + + if !inst.Enabled { + return nil + } + + if err := l.loadInstance(ctx, *inst, true); err != nil { + return err + } + + slog.Info("channel instance loaded by id", "id", id, "name", inst.Name, "type", inst.ChannelType) + return nil +} + // Stop stops all managed channels. func (l *InstanceLoader) Stop(ctx context.Context) { l.mu.Lock() diff --git a/internal/channels/instance_loader_timeout_test.go b/internal/channels/instance_loader_timeout_test.go index bdbdfae9..8b520a59 100644 --- a/internal/channels/instance_loader_timeout_test.go +++ b/internal/channels/instance_loader_timeout_test.go @@ -3,10 +3,13 @@ package channels import ( "context" "encoding/json" + "errors" "sync/atomic" "testing" "time" + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/store" ) @@ -85,6 +88,59 @@ func withShortReloadTimeout(t *testing.T, d time.Duration) { t.Cleanup(func() { reloadStartTimeout = orig }) } +type singleInstanceStore struct { + store.ChannelInstanceStore + inst store.ChannelInstanceData +} + +func (s *singleInstanceStore) Get(context.Context, uuid.UUID) (*store.ChannelInstanceData, error) { + return &s.inst, nil +} + +type scopedSingleInstanceStore struct { + singleInstanceStore +} + +func (s *scopedSingleInstanceStore) Get(ctx context.Context, id uuid.UUID) (*store.ChannelInstanceData, error) { + if !store.IsCrossTenant(ctx) && store.TenantIDFromContext(ctx) == uuid.Nil { + return nil, errors.New("missing tenant scope") + } + return &s.inst, nil +} + +func TestLoadInstanceByIDLoadsTargetWithoutReloadingExisting(t *testing.T) { + msgBus := bus.New() + mgr := NewManager(msgBus) + targetID := uuid.New() + + loader := NewInstanceLoader(&scopedSingleInstanceStore{ + singleInstanceStore: singleInstanceStore{inst: store.ChannelInstanceData{ + BaseModel: store.BaseModel{ID: targetID}, + Name: "telegram-new", + ChannelType: TypeTelegram, + Enabled: true, + }}, + }, nil, mgr, msgBus, nil) + + existing := newTimeoutTestChannel("telegram-existing", TypeTelegram, false) + mgr.RegisterChannel("telegram-existing", existing) + + loader.RegisterFactory(TypeTelegram, func(name string, _ json.RawMessage, _ json.RawMessage, _ *bus.MessageBus, _ store.PairingStore) (Channel, error) { + return newTimeoutTestChannel(name, TypeTelegram, false), nil + }) + + if err := loader.LoadInstanceByID(store.WithCrossTenant(context.Background()), targetID); err != nil { + t.Fatalf("LoadInstanceByID returned error: %v", err) + } + + if _, ok := mgr.GetChannel("telegram-new"); !ok { + t.Fatal("expected target channel to be registered") + } + if _, ok := mgr.GetChannel("telegram-existing"); !ok { + t.Fatal("targeted load should not unregister unrelated channels") + } +} + // TestLoadInstance_HungStartDoesNotBlock verifies that a Start() that never // returns is abandoned after reloadStartTimeout so Reload() can proceed to // other channels instead of deadlocking on the loader mutex. diff --git a/internal/channels/whatsapp/qr_methods.go b/internal/channels/whatsapp/qr_methods.go index 8ca4d262..a3917945 100644 --- a/internal/channels/whatsapp/qr_methods.go +++ b/internal/channels/whatsapp/qr_methods.go @@ -28,11 +28,16 @@ type cancelEntry struct { type QRMethods struct { instanceStore store.ChannelInstanceStore manager *channels.Manager + loader *channels.InstanceLoader activeSessions sync.Map // instanceID (string) -> *cancelEntry } -func NewQRMethods(instanceStore store.ChannelInstanceStore, manager *channels.Manager) *QRMethods { - return &QRMethods{instanceStore: instanceStore, manager: manager} +func NewQRMethods(instanceStore store.ChannelInstanceStore, manager *channels.Manager, loader ...*channels.InstanceLoader) *QRMethods { + m := &QRMethods{instanceStore: instanceStore, manager: manager} + if len(loader) > 0 { + m.loader = loader[0] + } + return m } func (m *QRMethods) Register(router *gateway.MethodRouter) { @@ -73,15 +78,24 @@ func (m *QRMethods) handleQRStart(ctx context.Context, client *gateway.Client, r // ACK immediately — QR/done events arrive asynchronously. client.SendResponse(goclawprotocol.NewOKResponse(req.ID, map[string]any{"status": "started"})) - go m.runQRSession(qrCtx, entry, client, params.InstanceID, inst.Name, params.ForceReauth) + go m.runQRSession(qrCtx, entry, client, instID, params.InstanceID, inst.Name, params.ForceReauth) } func (m *QRMethods) runQRSession(ctx context.Context, entry *cancelEntry, - client *gateway.Client, instanceIDStr, channelName string, forceReauth bool) { + client *gateway.Client, instanceID uuid.UUID, instanceIDStr, channelName string, forceReauth bool) { defer entry.cancel() defer m.activeSessions.CompareAndDelete(instanceIDStr, entry) + if m.loader != nil { + // Channel Start stores its context for long-lived WhatsApp work; keep tenant values + // from the request without tying the channel lifetime to this QR session timeout. + loadCtx := context.WithoutCancel(ctx) + if err := m.loader.LoadInstanceByID(loadCtx, instanceID); err != nil { + slog.Warn("whatsapp QR: targeted channel load failed", "instance", instanceIDStr, "channel", channelName, "error", err) + } + } + // Wait for channel to appear in manager — instance creation triggers an async // reload, so the channel may not be registered yet when the wizard fires QR start. var wa *Channel diff --git a/internal/gateway/methods/channel_instances.go b/internal/gateway/methods/channel_instances.go index e64a776c..5058f0d7 100644 --- a/internal/gateway/methods/channel_instances.go +++ b/internal/gateway/methods/channel_instances.go @@ -70,13 +70,13 @@ func (m *ChannelInstancesMethods) Register(router *gateway.MethodRouter) { router.Register(protocol.MethodChannelInstancesDelete, m.handleDelete) } -func (m *ChannelInstancesMethods) emitCacheInvalidate() { +func (m *ChannelInstancesMethods) emitCacheInvalidate(key string) { if m.msgBus == nil { return } m.msgBus.Broadcast(bus.Event{ Name: protocol.EventCacheInvalidate, - Payload: bus.CacheInvalidatePayload{Kind: bus.CacheKindChannelInstances}, + Payload: bus.CacheInvalidatePayload{Kind: bus.CacheKindChannelInstances, Key: key}, }) } @@ -179,7 +179,7 @@ func (m *ChannelInstancesMethods) handleCreate(ctx context.Context, client *gate return } - m.emitCacheInvalidate() + m.emitCacheInvalidate(inst.ID.String()) emitAudit(m.eventBus, client, "channel_instance.created", "channel_instance", inst.ID.String()) client.SendResponse(protocol.NewOKResponse(req.ID, maskInstance(*inst))) } @@ -222,7 +222,7 @@ func (m *ChannelInstancesMethods) handleUpdate(ctx context.Context, client *gate return } - m.emitCacheInvalidate() + m.emitCacheInvalidate("") emitAudit(m.eventBus, client, "channel_instance.updated", "channel_instance", id.String()) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{"status": "updated"})) } @@ -279,7 +279,7 @@ func (m *ChannelInstancesMethods) handleDelete(ctx context.Context, client *gate return } - m.emitCacheInvalidate() + m.emitCacheInvalidate("") emitAudit(m.eventBus, client, "channel_instance.deleted", "channel_instance", id.String()) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{"status": "deleted"})) } diff --git a/internal/http/channel_instances.go b/internal/http/channel_instances.go index ac88145a..4ab471bf 100644 --- a/internal/http/channel_instances.go +++ b/internal/http/channel_instances.go @@ -119,13 +119,13 @@ func (h *ChannelInstancesHandler) adminAuth(next http.HandlerFunc) http.HandlerF return requireAuth(permissions.RoleAdmin, next) } -func (h *ChannelInstancesHandler) emitCacheInvalidate() { +func (h *ChannelInstancesHandler) emitCacheInvalidate(key string) { if h.msgBus == nil { return } h.msgBus.Broadcast(bus.Event{ Name: protocol.EventCacheInvalidate, - Payload: bus.CacheInvalidatePayload{Kind: bus.CacheKindChannelInstances}, + Payload: bus.CacheInvalidatePayload{Kind: bus.CacheKindChannelInstances, Key: key}, }) } @@ -228,7 +228,7 @@ func (h *ChannelInstancesHandler) handleCreate(w http.ResponseWriter, r *http.Re return } - h.emitCacheInvalidate() + h.emitCacheInvalidate(inst.ID.String()) emitAudit(h.msgBus, r, "channel_instance.created", "channel_instance", inst.ID.String()) writeJSON(w, http.StatusCreated, maskInstanceHTTP(*inst)) } @@ -273,7 +273,7 @@ func (h *ChannelInstancesHandler) handleUpdate(w http.ResponseWriter, r *http.Re return } - h.emitCacheInvalidate() + h.emitCacheInvalidate("") emitAudit(h.msgBus, r, "channel_instance.updated", "channel_instance", id.String()) writeJSON(w, http.StatusOK, map[string]string{"status": "updated"}) } @@ -338,7 +338,7 @@ func (h *ChannelInstancesHandler) handleDelete(w http.ResponseWriter, r *http.Re return } - h.emitCacheInvalidate() + h.emitCacheInvalidate("") emitAudit(h.msgBus, r, "channel_instance.deleted", "channel_instance", id.String()) writeJSON(w, http.StatusOK, map[string]string{"status": "deleted"}) }