diff --git a/internal/im/dingtalk/adapter.go b/internal/im/dingtalk/adapter.go index 80844825..3540de09 100644 --- a/internal/im/dingtalk/adapter.go +++ b/internal/im/dingtalk/adapter.go @@ -41,8 +41,7 @@ var ( type Adapter struct { clientID string clientSecret string - cardTemplateID string // optional: enables AI card streaming when set - client *LongConnClient // nil for webhook mode + cardTemplateID string // optional: enables AI card streaming when set // accessToken cache tokenMu sync.RWMutex @@ -60,14 +59,15 @@ func NewWebhookAdapter(clientID, clientSecret, cardTemplateID string) *Adapter { } } -// NewAdapter creates a DingTalk adapter backed by a stream client. -func NewAdapter(client *LongConnClient, clientID, clientSecret, cardTemplateID string) *Adapter { +// NewAdapter creates a DingTalk adapter for stream (websocket) mode. +// The stream connection itself is managed separately by the supervisor; the +// adapter only sends replies (via sessionWebhook or OpenAPI). +func NewAdapter(clientID, clientSecret, cardTemplateID string) *Adapter { startStreamReaper() return &Adapter{ clientID: clientID, clientSecret: clientSecret, cardTemplateID: cardTemplateID, - client: client, } } @@ -457,7 +457,7 @@ const ( var ( streamsMu sync.Mutex - dStreams = map[string]*streamState{} + dStreams = map[string]*streamState{} startReaperOnce sync.Once reaperStopCh = make(chan struct{}) ) diff --git a/internal/im/dingtalk/factory.go b/internal/im/dingtalk/factory.go index e911aea4..f36bb4d3 100644 --- a/internal/im/dingtalk/factory.go +++ b/internal/im/dingtalk/factory.go @@ -5,7 +5,6 @@ import ( "fmt" "github.com/Tencent/WeKnora/internal/im" - "github.com/Tencent/WeKnora/internal/logger" ) // NewFactory returns an im.AdapterFactory for DingTalk channels. @@ -29,16 +28,20 @@ func NewFactory() im.AdapterFactory { return adapter, nil, nil case "websocket": - client := NewLongConnClient(clientID, clientSecret, msgHandler) - wsCtx, wsCancel := context.WithCancel(context.Background()) - go func() { - if err := client.Start(wsCtx); err != nil && wsCtx.Err() == nil { - logger.Errorf(context.Background(), "[IM] DingTalk stream stopped for channel %s: %v", channel.ID, err) - } - }() + go im.RunSupervised(wsCtx, im.SupervisorConfig{ + Name: fmt.Sprintf("DingTalk channel %s", channel.ID), + Connect: func(ctx context.Context) (func(), error) { + client := NewLongConnClient(clientID, clientSecret, msgHandler) + if err := client.Start(ctx); err != nil { + client.Close() + return nil, err + } + return client.Close, nil + }, + }) - adapter := NewAdapter(client, clientID, clientSecret, cardTemplateID) + adapter := NewAdapter(clientID, clientSecret, cardTemplateID) return adapter, wsCancel, nil default: diff --git a/internal/im/dingtalk/longconn.go b/internal/im/dingtalk/longconn.go index 0ef103a1..0931b912 100644 --- a/internal/im/dingtalk/longconn.go +++ b/internal/im/dingtalk/longconn.go @@ -4,8 +4,8 @@ import ( "context" "strings" - dtsdk "github.com/open-dingtalk/dingtalk-stream-sdk-go/client" "github.com/open-dingtalk/dingtalk-stream-sdk-go/chatbot" + dtsdk "github.com/open-dingtalk/dingtalk-stream-sdk-go/client" "github.com/Tencent/WeKnora/internal/im" "github.com/Tencent/WeKnora/internal/logger" @@ -43,21 +43,19 @@ func NewLongConnClient(clientID, clientSecret string, handler MessageHandler) *L return c } -// Start begins the stream connection. It blocks until ctx is cancelled. +// Start establishes the stream connection. The underlying SDK's Start is +// non-blocking: it dials the websocket, spawns its internal read/reconnect +// loops, and returns once the connection is established (or the attempt fails). func (c *LongConnClient) Start(ctx context.Context) error { logger.Infof(ctx, "[IM] DingTalk Stream connecting...") + return c.streamClient.Start(ctx) +} - errCh := make(chan error, 1) - go func() { - errCh <- c.streamClient.Start(ctx) - }() - - select { - case <-ctx.Done(): - return ctx.Err() - case err := <-errCh: - return err - } +// Close tears down the stream connection. AutoReconnect is disabled first so +// that closing the connection does not trigger the SDK's internal reconnect. +func (c *LongConnClient) Close() { + c.streamClient.AutoReconnect = false + c.streamClient.Close() } func (c *LongConnClient) onChatBotMessage(ctx context.Context, data *chatbot.BotCallbackDataModel) ([]byte, error) { diff --git a/internal/im/supervisor.go b/internal/im/supervisor.go new file mode 100644 index 00000000..4c920b60 --- /dev/null +++ b/internal/im/supervisor.go @@ -0,0 +1,97 @@ +package im + +import ( + "context" + "time" + + "github.com/Tencent/WeKnora/internal/logger" +) + +// defaultRecycleInterval is how often a supervised long connection is proactively +// torn down and rebuilt. Some IM SDKs (e.g. DingTalk, Feishu) delegate reconnection +// to internal logic that can silently enter a "zombie" state on long-running +// connections — the connection object stays alive but no longer receives messages. +// Periodically recreating the connection bounds the worst-case outage to this +// interval, on top of whatever auto-reconnect the SDK already provides. +const defaultRecycleInterval = 6 * time.Hour + +// defaultSupervisorRetryDelay is the backoff applied after a failed connect attempt. +const defaultSupervisorRetryDelay = 5 * time.Second + +// SupervisorConfig configures RunSupervised. +type SupervisorConfig struct { + // Name identifies the supervised connection in logs (e.g. "DingTalk channel xxx"). + Name string + + // MaxConnAge is the interval between proactive reconnections. + // Defaults to defaultRecycleInterval when <= 0. + MaxConnAge time.Duration + + // RetryDelay is the backoff after a failed connect attempt. + // Defaults to defaultSupervisorRetryDelay when <= 0. + RetryDelay time.Duration + + // Connect establishes a fresh connection and returns a stop function that + // tears it down cleanly (preventing any further internal reconnects). + // It may wrap either a blocking or a non-blocking SDK Start: the only + // contract is that it returns once the connection has been established + // (or the attempt has failed), along with a stop function. + Connect func(ctx context.Context) (stop func(), err error) +} + +// RunSupervised manages the lifecycle of a long-lived IM connection. +// +// It keeps a connection alive by (re)establishing it via cfg.Connect, then +// proactively recycling it every cfg.MaxConnAge. The underlying SDK is still +// expected to handle transient drops via its own auto-reconnect; the periodic +// recycle is a safety net that eliminates stuck/zombie connections that the SDK +// fails to recover on its own. +// +// RunSupervised blocks until ctx is cancelled. On cancellation it tears down the +// active connection via the stop function before returning, so callers can rely +// on cancelling ctx to fully stop the connection. +func RunSupervised(ctx context.Context, cfg SupervisorConfig) { + maxAge := cfg.MaxConnAge + if maxAge <= 0 { + maxAge = defaultRecycleInterval + } + retryDelay := cfg.RetryDelay + if retryDelay <= 0 { + retryDelay = defaultSupervisorRetryDelay + } + + for { + if ctx.Err() != nil { + return + } + + stop, err := cfg.Connect(ctx) + if err != nil { + if ctx.Err() != nil { + return + } + logger.Warnf(ctx, "[IM] %s connect failed: %v, retrying in %v", cfg.Name, err, retryDelay) + select { + case <-time.After(retryDelay): + continue + case <-ctx.Done(): + return + } + } + + logger.Infof(ctx, "[IM] %s connection established (recycle in %v)", cfg.Name, maxAge) + + select { + case <-time.After(maxAge): + logger.Infof(ctx, "[IM] %s periodic reconnect to refresh connection", cfg.Name) + if stop != nil { + stop() + } + case <-ctx.Done(): + if stop != nil { + stop() + } + return + } + } +}