mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-14 14:17:40 +00:00
fix(channels): load WhatsApp QR channel on create
Squash merge PR #98 in priority order after #97. Existing PR checks were green; branch workflow status will be tracked.
This commit is contained in:
1 parent
361f2abbf5
commit
3569f8d681
7 files changed
+131
-18
No files matched your search
+1
-1
@@ -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:
|
||||
|
||||
@@ -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(
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"}))
|
||||
}
|
||||
|
||||
@@ -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"})
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user