diff --git a/api/channels/whatsapp/gateway-node/README.md b/api/channels/whatsapp/gateway-node/README.md index 94adcda720..78417f67af 100644 --- a/api/channels/whatsapp/gateway-node/README.md +++ b/api/channels/whatsapp/gateway-node/README.md @@ -20,6 +20,7 @@ npm start ## API +- `GET /health` - `POST /whatsapp/:sessionKey/start` - `GET /whatsapp/:sessionKey/status` - `GET /whatsapp/:sessionKey/events/ws?after=` (WebSocket) diff --git a/api/channels/whatsapp/gateway-node/package-lock.json b/api/channels/whatsapp/gateway-node/package-lock.json index eac4c7ebc5..161dfcfd67 100644 --- a/api/channels/whatsapp/gateway-node/package-lock.json +++ b/api/channels/whatsapp/gateway-node/package-lock.json @@ -199,9 +199,6 @@ "cpu": [ "arm" ], - "libc": [ - "glibc" - ], "license": "LGPL-3.0-or-later", "optional": true, "os": [ @@ -219,9 +216,6 @@ "cpu": [ "arm64" ], - "libc": [ - "glibc" - ], "license": "LGPL-3.0-or-later", "optional": true, "os": [ @@ -239,9 +233,6 @@ "cpu": [ "ppc64" ], - "libc": [ - "glibc" - ], "license": "LGPL-3.0-or-later", "optional": true, "os": [ @@ -259,9 +250,6 @@ "cpu": [ "riscv64" ], - "libc": [ - "glibc" - ], "license": "LGPL-3.0-or-later", "optional": true, "os": [ @@ -279,9 +267,6 @@ "cpu": [ "s390x" ], - "libc": [ - "glibc" - ], "license": "LGPL-3.0-or-later", "optional": true, "os": [ @@ -299,9 +284,6 @@ "cpu": [ "x64" ], - "libc": [ - "glibc" - ], "license": "LGPL-3.0-or-later", "optional": true, "os": [ @@ -319,9 +301,6 @@ "cpu": [ "arm64" ], - "libc": [ - "musl" - ], "license": "LGPL-3.0-or-later", "optional": true, "os": [ @@ -339,9 +318,6 @@ "cpu": [ "x64" ], - "libc": [ - "musl" - ], "license": "LGPL-3.0-or-later", "optional": true, "os": [ @@ -359,9 +335,6 @@ "cpu": [ "arm" ], - "libc": [ - "glibc" - ], "license": "Apache-2.0", "optional": true, "os": [ @@ -385,9 +358,6 @@ "cpu": [ "arm64" ], - "libc": [ - "glibc" - ], "license": "Apache-2.0", "optional": true, "os": [ @@ -411,9 +381,6 @@ "cpu": [ "ppc64" ], - "libc": [ - "glibc" - ], "license": "Apache-2.0", "optional": true, "os": [ @@ -437,9 +404,6 @@ "cpu": [ "riscv64" ], - "libc": [ - "glibc" - ], "license": "Apache-2.0", "optional": true, "os": [ @@ -463,9 +427,6 @@ "cpu": [ "s390x" ], - "libc": [ - "glibc" - ], "license": "Apache-2.0", "optional": true, "os": [ @@ -489,9 +450,6 @@ "cpu": [ "x64" ], - "libc": [ - "glibc" - ], "license": "Apache-2.0", "optional": true, "os": [ @@ -515,9 +473,6 @@ "cpu": [ "arm64" ], - "libc": [ - "musl" - ], "license": "Apache-2.0", "optional": true, "os": [ @@ -541,9 +496,6 @@ "cpu": [ "x64" ], - "libc": [ - "musl" - ], "license": "Apache-2.0", "optional": true, "os": [ diff --git a/cmd/ragflow_server.go b/cmd/ragflow_server.go index 944c63ed16..46f6300716 100644 --- a/cmd/ragflow_server.go +++ b/cmd/ragflow_server.go @@ -27,6 +27,7 @@ import ( "ragflow/internal/agent/audio" "ragflow/internal/agent/canvas" agenttool "ragflow/internal/agent/tool" + "ragflow/internal/channels" "ragflow/internal/handler" ingestion "ragflow/internal/ingestion/service" "ragflow/internal/mcp" @@ -846,6 +847,8 @@ func startServer(ctx context.Context, config *server.Config) { // Setup routes r.Setup(ginEngine) + channels.Start(ctx) + // Create HTTP server with timeouts to prevent slow clients from blocking shutdown addr := fmt.Sprintf(":%d", config.APIServer.Port) srv := &http.Server{ diff --git a/internal/channels/bootstrap.go b/internal/channels/bootstrap.go new file mode 100644 index 0000000000..3da5899bd7 --- /dev/null +++ b/internal/channels/bootstrap.go @@ -0,0 +1,325 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package channels + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "log" + "sync" + "time" + + "ragflow/internal/channels/core" + "ragflow/internal/dao" + "ragflow/internal/entity" + "ragflow/internal/service" +) + +const ( + reconcileInterval = 10 * time.Second + initialStartRetryDelay = 10 * time.Second + maxStartRetryDelay = 5 * time.Minute +) + +type runningChannel struct { + channel core.Channel + fp string +} + +type failedChannel struct { + fp string + attempts int + nextRetryAt time.Time +} + +type Runtime struct { + mu sync.Mutex + running map[string]runningChannel + failed map[string]failedChannel +} + +// NewRuntime creates an empty chat-channel runtime reconciler. +func NewRuntime() *Runtime { + return &Runtime{ + running: map[string]runningChannel{}, + failed: map[string]failedChannel{}, + } +} + +// Start launches the chat-channel runtime reconciler in the background. +func Start(ctx context.Context) *Runtime { + rt := NewRuntime() + service.SetChatChannelRuntimeProvider(whatsappRuntimeSnapshotMap) + go rt.Run(ctx) + return rt +} + +// Run reconciles configured chat channels until the context is cancelled. +func (r *Runtime) Run(ctx context.Context) { + ticker := time.NewTicker(reconcileInterval) + defer ticker.Stop() + defer r.stopAll(context.Background()) + + for { + if err := r.Reconcile(ctx); err != nil { + log.Printf("chat channel reconcile failed: %v", err) + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } +} + +// Reconcile starts, stops, and restarts channel instances to match the database state. +func (r *Runtime) Reconcile(ctx context.Context) error { + desired, err := desiredChannels(ctx) + if err != nil { + return err + } + + var toStop []core.Channel + r.mu.Lock() + for accountID, entry := range r.running { + wanted, ok := desired[accountID] + if !ok || wanted.fp != entry.fp { + delete(r.running, accountID) + toStop = append(toStop, entry.channel) + } + } + for accountID, failure := range r.failed { + wanted, ok := desired[accountID] + if !ok || wanted.fp != failure.fp { + delete(r.failed, accountID) + } + } + r.mu.Unlock() + + for _, ch := range toStop { + stopChannel(context.Background(), ch) + } + + activeWhatsApp := false + for _, wanted := range desired { + if wanted.channel == "whatsapp" { + activeWhatsApp = true + break + } + } + if err := syncWhatsAppGateway(ctx, activeWhatsApp); err != nil && activeWhatsApp { + log.Printf("failed to sync WhatsApp gateway: %v", err) + } + + now := time.Now() + for accountID, wanted := range desired { + r.mu.Lock() + _, isRunning := r.running[accountID] + failure, failed := r.failed[accountID] + retryPending := failed && failure.fp == wanted.fp && now.Before(failure.nextRetryAt) + r.mu.Unlock() + if isRunning || retryPending { + continue + } + if err := r.startChannel(ctx, accountID, wanted); err != nil { + log.Printf("failed to start chat channel %s (%s): %v", accountID, wanted.channel, err) + r.recordStartFailure(accountID, wanted.fp, now) + continue + } + r.clearStartFailure(accountID) + } + return nil +} + +// recordStartFailure saves the next retry window for a failed channel start. +func (r *Runtime) recordStartFailure(accountID string, fp string, now time.Time) { + r.mu.Lock() + defer r.mu.Unlock() + + attempts := 1 + if failure, ok := r.failed[accountID]; ok && failure.fp == fp { + attempts = failure.attempts + 1 + } + r.failed[accountID] = failedChannel{ + fp: fp, + attempts: attempts, + nextRetryAt: now.Add(startRetryDelay(attempts)), + } +} + +// clearStartFailure removes a stale failed-start record after a successful start. +func (r *Runtime) clearStartFailure(accountID string) { + r.mu.Lock() + defer r.mu.Unlock() + delete(r.failed, accountID) +} + +// startRetryDelay returns the bounded exponential backoff for a start attempt. +func startRetryDelay(attempts int) time.Duration { + if attempts < 1 { + attempts = 1 + } + delay := initialStartRetryDelay + for i := 1; i < attempts; i++ { + delay *= 2 + if delay >= maxStartRetryDelay { + return maxStartRetryDelay + } + } + return delay +} + +type desiredChannel struct { + channel string + credential map[string]any + fp string +} + +// desiredChannels loads enabled chat-channel rows and reduces them to runtime configuration. +func desiredChannels(ctx context.Context) (map[string]desiredChannel, error) { + rows, err := dao.NewChatChannel().ListActive(ctx, dao.DB) + if err != nil { + return nil, err + } + out := make(map[string]desiredChannel, len(rows)) + for _, row := range rows { + credential := credentialFromConfig(row.Config) + out[row.ID] = desiredChannel{ + channel: row.Channel, + credential: credential, + fp: fingerprint(row.Channel, credential), + } + } + return out, nil +} + +// credentialFromConfig extracts the platform credential block from chat_channel.config. +func credentialFromConfig(config entity.JSONMap) map[string]any { + if config == nil { + return map[string]any{} + } + if raw, ok := config["credential"].(map[string]any); ok { + return raw + } + if raw, ok := config["credential"].(entity.JSONMap); ok { + return raw + } + return map[string]any{} +} + +// fingerprint returns a stable hash for the configuration that requires a channel restart. +func fingerprint(channel string, credential map[string]any) string { + payload, _ := json.Marshal(map[string]any{ + "channel": channel, + "credential": credential, + }) + sum := sha256.Sum256(payload) + return hex.EncodeToString(sum[:]) +} + +// startChannel builds a platform channel, attaches the RAG bridge, and starts it. +func (r *Runtime) startChannel(ctx context.Context, accountID string, wanted desiredChannel) error { + ch, err := buildChannel(accountID, wanted) + if err != nil { + return err + } + if ch == nil { + return nil + } + + channelService := service.NewChatChannelService() + ch.SetMessageHandler(func(ctx context.Context, msg core.IncomingMessage) error { + answer, err := channelService.HandleIncomingMessage(ctx, service.ChatChannelIncomingMessage{ + Channel: msg.Channel, + AccountID: msg.AccountID, + ChatID: msg.ChatID, + ChatType: msg.ChatType, + MessageID: msg.MessageID, + SenderID: msg.SenderID, + Text: msg.Text, + }) + if err != nil || answer == "" { + return err + } + return ch.Send(ctx, core.OutgoingMessage{ + ChatID: msg.ChatID, + Text: answer, + ReplyToMessageID: msg.MessageID, + }) + }) + if err = ch.Start(ctx); err != nil { + return err + } + r.mu.Lock() + r.running[accountID] = runningChannel{channel: ch, fp: wanted.fp} + r.mu.Unlock() + log.Printf("started chat channel %s:%s", ch.ChannelID(), accountID) + return nil +} + +// buildChannel constructs the platform-specific channel implementation for one chat_channel row. +func buildChannel(accountID string, wanted desiredChannel) (core.Channel, error) { + switch wanted.channel { + case "whatsapp": + return newWhatsAppChannelFromConfig(accountID, wanted.credential) + default: + return nil, fmt.Errorf("unknown channel: %s", wanted.channel) + } +} + +// whatsappRuntimeSnapshotMap returns the API payload for a live WhatsApp runtime snapshot. +func whatsappRuntimeSnapshotMap(accountID string) (map[string]any, bool) { + snapshot, ok := getWhatsAppRuntimeSnapshot(accountID) + if !ok { + return nil, false + } + return map[string]any{ + "account_id": snapshot.AccountID, + "session_key": snapshot.SessionKey, + "status": snapshot.Status, + "connected_at": snapshot.ConnectedAt, + "qr_updated_at": snapshot.QRUpdatedAt, + "qr_data_url": snapshot.QRDataURL, + "last_error": snapshot.LastError, + "session_id": snapshot.SessionID, + "last_snapshot_at": snapshot.LastSnapshotAt, + "gateway_base_url": snapshot.GatewayBaseURL, + "event_cursor": snapshot.EventCursor, + }, true +} + +// stopAll stops every running channel and shuts down shared gateway processes. +func (r *Runtime) stopAll(ctx context.Context) { + r.mu.Lock() + running := r.running + r.running = map[string]runningChannel{} + r.mu.Unlock() + for _, entry := range running { + stopChannel(ctx, entry.channel) + } + _ = syncWhatsAppGateway(ctx, false) +} + +// stopChannel stops one platform channel and logs any shutdown error. +func stopChannel(ctx context.Context, ch core.Channel) { + if err := ch.Stop(ctx); err != nil { + log.Printf("failed to stop chat channel %s:%s: %v", ch.ChannelID(), ch.AccountID(), err) + } +} diff --git a/internal/channels/bootstrap_test.go b/internal/channels/bootstrap_test.go new file mode 100644 index 0000000000..773bdb763f --- /dev/null +++ b/internal/channels/bootstrap_test.go @@ -0,0 +1,179 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package channels + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "ragflow/internal/channels/core" +) + +func TestStartRetryDelay(t *testing.T) { + tests := []struct { + name string + attempts int + want time.Duration + }{ + {name: "zero attempt uses first delay", attempts: 0, want: initialStartRetryDelay}, + {name: "first attempt", attempts: 1, want: initialStartRetryDelay}, + {name: "second attempt doubles", attempts: 2, want: 2 * initialStartRetryDelay}, + {name: "fifth attempt", attempts: 5, want: 16 * initialStartRetryDelay}, + {name: "bounded", attempts: 20, want: maxStartRetryDelay}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := startRetryDelay(tt.attempts); got != tt.want { + t.Fatalf("startRetryDelay(%d) = %s, want %s", tt.attempts, got, tt.want) + } + }) + } +} + +func TestRecordStartFailure(t *testing.T) { + rt := NewRuntime() + now := time.Unix(100, 0) + + rt.recordStartFailure("account-1", "fp-1", now) + first := rt.failed["account-1"] + if first.attempts != 1 { + t.Fatalf("attempts after first failure = %d, want 1", first.attempts) + } + if !first.nextRetryAt.Equal(now.Add(initialStartRetryDelay)) { + t.Fatalf("nextRetryAt after first failure = %s, want %s", first.nextRetryAt, now.Add(initialStartRetryDelay)) + } + + rt.recordStartFailure("account-1", "fp-1", now) + second := rt.failed["account-1"] + if second.attempts != 2 { + t.Fatalf("attempts after second failure = %d, want 2", second.attempts) + } + if !second.nextRetryAt.Equal(now.Add(2 * initialStartRetryDelay)) { + t.Fatalf("nextRetryAt after second failure = %s, want %s", second.nextRetryAt, now.Add(2*initialStartRetryDelay)) + } + + rt.recordStartFailure("account-1", "fp-2", now) + changed := rt.failed["account-1"] + if changed.attempts != 1 { + t.Fatalf("attempts after fingerprint change = %d, want 1", changed.attempts) + } +} + +func TestClearStartFailure(t *testing.T) { + rt := NewRuntime() + rt.recordStartFailure("account-1", "fp-1", time.Unix(100, 0)) + + rt.clearStartFailure("account-1") + + if _, ok := rt.failed["account-1"]; ok { + t.Fatal("clearStartFailure left failed entry behind") + } +} + +func TestGatewayWorkdirDefaultContainsNodeEntrypoint(t *testing.T) { + t.Setenv("WHATSAPP_GATEWAY_WORKDIR", "") + + entry := filepath.Join(gatewayWorkdir(), "index.js") + if _, err := os.Stat(entry); err != nil { + t.Fatalf("default gateway entry %s is not available: %v", entry, err) + } +} + +func TestGatewayEnabledDefaultsToUserManaged(t *testing.T) { + t.Setenv("WHATSAPP_GATEWAY_ENABLED", "") + + if gatewayEnabled() { + t.Fatal("gatewayEnabled() = true, want false by default") + } +} + +func TestGatewayEnabledCanBeExplicitlyEnabled(t *testing.T) { + t.Setenv("WHATSAPP_GATEWAY_ENABLED", "true") + + if !gatewayEnabled() { + t.Fatal("gatewayEnabled() = false, want true") + } +} + +func TestGatewayEnabledRejectsInvalidValue(t *testing.T) { + t.Setenv("WHATSAPP_GATEWAY_ENABLED", "maybe") + + if gatewayEnabled() { + t.Fatal("gatewayEnabled() = true for invalid value, want false") + } +} + +func TestWhatsAppEnqueueDoesNotMarkDroppedMessageSeen(t *testing.T) { + ch := newWhatsAppChannel(whatsappAccount{AccountID: "account-1"}) + worker := &whatsappChatWorker{queue: make(chan core.IncomingMessage, 1)} + worker.queue <- core.IncomingMessage{MessageID: "queued"} + ch.workers["chat-1"] = worker + + ok := ch.enqueueIncoming(context.Background(), core.IncomingMessage{ + ChatID: "chat-1", + MessageID: "dropped", + }) + + if ok { + t.Fatal("enqueueIncoming succeeded for a full queue") + } + if _, seen := ch.seen["dropped"]; seen { + t.Fatal("dropped message was marked seen") + } +} + +func TestWhatsAppEnqueueMarksSeenAfterSuccessfulHandoff(t *testing.T) { + ch := newWhatsAppChannel(whatsappAccount{AccountID: "account-1"}) + ch.workers["chat-1"] = &whatsappChatWorker{queue: make(chan core.IncomingMessage, 1)} + msg := core.IncomingMessage{ChatID: "chat-1", MessageID: "message-1"} + + if ok := ch.enqueueIncoming(context.Background(), msg); !ok { + t.Fatal("enqueueIncoming failed for an empty queue") + } + if _, seen := ch.seen[msg.MessageID]; !seen { + t.Fatal("successfully handed off message was not marked seen") + } + if ok := ch.enqueueIncoming(context.Background(), msg); ok { + t.Fatal("duplicate message was enqueued") + } +} + +func TestWhatsAppRetireChatWorkerRequiresCurrentEmptyWorker(t *testing.T) { + ch := newWhatsAppChannel(whatsappAccount{AccountID: "account-1"}) + worker := &whatsappChatWorker{queue: make(chan core.IncomingMessage, 1)} + worker.queue <- core.IncomingMessage{ChatID: "chat-1", MessageID: "message-1"} + ch.workers["chat-1"] = worker + + if ch.retireChatWorker("chat-1", worker) { + t.Fatal("retireChatWorker retired a non-empty worker") + } + if ch.workers["chat-1"] != worker { + t.Fatal("non-empty worker was removed from ownership map") + } + + <-worker.queue + if !ch.retireChatWorker("chat-1", worker) { + t.Fatal("retireChatWorker did not retire an empty current worker") + } + if _, ok := ch.workers["chat-1"]; ok { + t.Fatal("empty retired worker remains in ownership map") + } +} diff --git a/internal/channels/core/channel.go b/internal/channels/core/channel.go new file mode 100644 index 0000000000..14a630e620 --- /dev/null +++ b/internal/channels/core/channel.go @@ -0,0 +1,47 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package core + +import "context" + +type IncomingMessage struct { + Channel string + AccountID string + ChatID string + ChatType string + MessageID string + SenderID string + Text string + Raw map[string]any +} + +type OutgoingMessage struct { + ChatID string + Text string + ReplyToMessageID string +} + +type MessageHandler func(ctx context.Context, msg IncomingMessage) error + +type Channel interface { + ChannelID() string // ChannelID return channel's name + AccountID() string // AccountID return channel's channelID + SetMessageHandler(MessageHandler) // SetMessageHandler set channel's Message Handler + Start(ctx context.Context) error // Start the channel + Stop(ctx context.Context) error // Stop cancels channel event + Send(ctx context.Context, msg OutgoingMessage) error // Send posts an outgoing RAGFlow answer to the channel +} diff --git a/internal/channels/dingtalk.go b/internal/channels/dingtalk.go new file mode 100644 index 0000000000..a4ce67278c --- /dev/null +++ b/internal/channels/dingtalk.go @@ -0,0 +1,17 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package channels diff --git a/internal/channels/discord.go b/internal/channels/discord.go new file mode 100644 index 0000000000..a4ce67278c --- /dev/null +++ b/internal/channels/discord.go @@ -0,0 +1,17 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package channels diff --git a/internal/channels/feishu.go b/internal/channels/feishu.go new file mode 100644 index 0000000000..a4ce67278c --- /dev/null +++ b/internal/channels/feishu.go @@ -0,0 +1,17 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package channels diff --git a/internal/channels/line.go b/internal/channels/line.go new file mode 100644 index 0000000000..a4ce67278c --- /dev/null +++ b/internal/channels/line.go @@ -0,0 +1,17 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package channels diff --git a/internal/channels/qqbot.go b/internal/channels/qqbot.go new file mode 100644 index 0000000000..a4ce67278c --- /dev/null +++ b/internal/channels/qqbot.go @@ -0,0 +1,17 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package channels diff --git a/internal/channels/telegram.go b/internal/channels/telegram.go new file mode 100644 index 0000000000..a4ce67278c --- /dev/null +++ b/internal/channels/telegram.go @@ -0,0 +1,17 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package channels diff --git a/internal/channels/wecom.go b/internal/channels/wecom.go new file mode 100644 index 0000000000..a4ce67278c --- /dev/null +++ b/internal/channels/wecom.go @@ -0,0 +1,17 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package channels diff --git a/internal/channels/whatsapp.go b/internal/channels/whatsapp.go new file mode 100644 index 0000000000..f87c87788a --- /dev/null +++ b/internal/channels/whatsapp.go @@ -0,0 +1,693 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package channels + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log" + "net/http" + "net/url" + "strconv" + "strings" + "sync" + "time" + + "github.com/gorilla/websocket" + + "ragflow/internal/channels/core" +) + +const ( + defaultGatewayBaseURL = "http://127.0.0.1:3005" + defaultTimeout = 30 * time.Second + startTimeout = 2 * time.Minute + wsHandshakeTimeout = 2 * time.Minute + wsReadTimeout = 2 * time.Minute + wsPingInterval = 30 * time.Second + defaultReconnect = 3 * time.Second + messageTTL = time.Hour + messageQueueSize = 64 + chatWorkerIdle = 5 * time.Minute +) + +type whatsappAccount struct { + AccountID string + GatewayBaseURL string + GatewayToken string + SessionKey string + Timeout time.Duration +} + +type whatsappRuntimeSnapshot struct { + AccountID string `json:"account_id"` + SessionKey string `json:"session_key"` + Status string `json:"status"` + ConnectedAt *float64 `json:"connected_at"` + QRUpdatedAt *float64 `json:"qr_updated_at"` + QRDataURL *string `json:"qr_data_url"` + LastError *string `json:"last_error"` + SessionID *string `json:"session_id"` + LastSnapshotAt *float64 `json:"last_snapshot_at"` + GatewayBaseURL *string `json:"gateway_base_url"` + EventCursor int64 `json:"event_cursor"` +} + +type whatsappChannel struct { + account whatsappAccount + + mu sync.Mutex + cancel context.CancelFunc + handler core.MessageHandler + client *http.Client + status string + lastErr string + qrDataURL string + qrUpdatedAt float64 + connectedAt float64 + lastSnapshotAt float64 + sessionID string + eventCursor int64 + seen map[string]time.Time + workers map[string]*whatsappChatWorker +} + +type whatsappChatWorker struct { + queue chan core.IncomingMessage +} + +var liveWhatsAppChannels sync.Map + +// newWhatsAppChannel creates a WhatsApp channel instance with default runtime settings filled in. +func newWhatsAppChannel(account whatsappAccount) *whatsappChannel { + if account.GatewayBaseURL == "" { + account.GatewayBaseURL = defaultGatewayBaseURL + } + if account.SessionKey == "" { + account.SessionKey = account.AccountID + } + if account.Timeout <= 0 { + account.Timeout = defaultTimeout + } + return &whatsappChannel{ + account: account, + client: &http.Client{Timeout: account.Timeout}, + status: "stopped", + seen: map[string]time.Time{}, + workers: map[string]*whatsappChatWorker{}, + } +} + +// newWhatsAppChannelFromConfig builds a WhatsApp channel from chat_channel.config.credential. +func newWhatsAppChannelFromConfig(accountID string, cfg map[string]any) (*whatsappChannel, error) { + timeout := defaultTimeout + if raw, ok := cfg["timeout_secs"]; ok { + switch v := raw.(type) { + case float64: + timeout = time.Duration(v) * time.Second + case int: + timeout = time.Duration(v) * time.Second + case string: + if n, err := strconv.Atoi(v); err == nil { + timeout = time.Duration(n) * time.Second + } + } + } + baseURL := firstString(cfg, "gateway_base_url", "gateway_url", "control_url") + token := firstString(cfg, "gateway_token", "token") + sessionKey := firstString(cfg, "session_key", "session_id") + if sessionKey == "" { + sessionKey = accountID + } + return newWhatsAppChannel(whatsappAccount{ + AccountID: accountID, + GatewayBaseURL: baseURL, + GatewayToken: token, + SessionKey: sessionKey, + Timeout: timeout, + }), nil +} + +// getWhatsAppRuntimeSnapshot returns the in-memory runtime snapshot for a live WhatsApp channel. +func getWhatsAppRuntimeSnapshot(accountID string) (*whatsappRuntimeSnapshot, bool) { + raw, ok := liveWhatsAppChannels.Load(accountID) + if !ok { + return nil, false + } + return raw.(*whatsappChannel).Snapshot(), true +} + +// ChannelID returns the platform identifier used by the chat-channel runtime. +func (c *whatsappChannel) ChannelID() string { + return "whatsapp" +} + +// AccountID returns the chat_channel.id bound to this WhatsApp runtime instance. +func (c *whatsappChannel) AccountID() string { + return c.account.AccountID +} + +// SetMessageHandler installs the RAGFlow bridge invoked for inbound WhatsApp messages. +func (c *whatsappChannel) SetMessageHandler(handler core.MessageHandler) { + c.mu.Lock() + defer c.mu.Unlock() + c.handler = handler +} + +// Start begins the WhatsApp gateway session and event websocket loop. +func (c *whatsappChannel) Start(ctx context.Context) error { + c.mu.Lock() + if c.cancel != nil { + c.mu.Unlock() + return nil + } + runCtx, cancel := context.WithCancel(ctx) + c.cancel = cancel + c.status = "connecting" + c.lastErr = "" + c.mu.Unlock() + + liveWhatsAppChannels.Store(c.account.AccountID, c) + go c.run(runCtx) + return nil +} + +// Stop cancels the WhatsApp event loop and asks the gateway to stop the session. +func (c *whatsappChannel) Stop(ctx context.Context) error { + c.mu.Lock() + cancel := c.cancel + c.cancel = nil + c.status = "stopped" + c.lastErr = "" + c.qrDataURL = "" + c.qrUpdatedAt = 0 + c.connectedAt = 0 + c.lastSnapshotAt = 0 + c.sessionID = "" + c.eventCursor = 0 + c.seen = map[string]time.Time{} + c.workers = map[string]*whatsappChatWorker{} + c.mu.Unlock() + + if cancel != nil { + cancel() + } + liveWhatsAppChannels.Delete(c.account.AccountID) + _ = c.requestJSON(ctx, http.MethodPost, fmt.Sprintf("whatsapp/%s/stop", url.PathEscape(c.sessionKey())), map[string]any{ + "account_id": c.account.AccountID, + "session_key": c.sessionKey(), + }, nil) + return nil +} + +// Send posts an outgoing RAGFlow answer to the WhatsApp gateway. +func (c *whatsappChannel) Send(ctx context.Context, msg core.OutgoingMessage) error { + if strings.TrimSpace(msg.ChatID) == "" { + return errors.New("chat_id is required") + } + payload := map[string]any{ + "account_id": c.account.AccountID, + "session_key": c.sessionKey(), + "chat_id": msg.ChatID, + "text": msg.Text, + "reply_to_message_id": nil, + } + if msg.ReplyToMessageID != "" { + payload["reply_to_message_id"] = msg.ReplyToMessageID + } + return c.requestJSON(ctx, http.MethodPost, fmt.Sprintf("whatsapp/%s/send", url.PathEscape(c.sessionKey())), payload, nil) +} + +// Snapshot copies the current WhatsApp runtime state for the runtime API. +func (c *whatsappChannel) Snapshot() *whatsappRuntimeSnapshot { + c.mu.Lock() + defer c.mu.Unlock() + baseURL := strings.TrimRight(c.account.GatewayBaseURL, "/") + return &whatsappRuntimeSnapshot{ + AccountID: c.account.AccountID, + SessionKey: c.sessionKey(), + Status: c.status, + ConnectedAt: floatPtr(c.connectedAt), + QRUpdatedAt: floatPtr(c.qrUpdatedAt), + QRDataURL: stringPtr(c.qrDataURL), + LastError: stringPtr(c.lastErr), + SessionID: stringPtr(c.sessionID), + LastSnapshotAt: floatPtr(c.lastSnapshotAt), + GatewayBaseURL: stringPtr(baseURL), + EventCursor: c.eventCursor, + } +} + +// run starts the gateway session and reconnects the websocket event stream when needed. +func (c *whatsappChannel) run(ctx context.Context) { + if err := c.startSession(ctx); err != nil { + if ctx.Err() == nil { + c.setError(err) + } + return + } + + for ctx.Err() == nil { + if err := c.runEvents(ctx); err != nil && ctx.Err() == nil { + c.setError(err) + log.Printf("[whatsapp:%s] event loop error: %v", c.account.AccountID, err) + time.Sleep(defaultReconnect) + } + } +} + +// startSession asks the gateway to start the session and accepts an already-created session after timeout. +func (c *whatsappChannel) startSession(ctx context.Context) error { + err := c.requestJSONWithTimeout(ctx, startTimeout, http.MethodPost, fmt.Sprintf("whatsapp/%s/start", url.PathEscape(c.sessionKey())), map[string]any{ + "account_id": c.account.AccountID, + "session_key": c.sessionKey(), + "gateway_base_url": strings.TrimRight(c.account.GatewayBaseURL, "/"), + }, nil) + if err == nil { + return c.waitForStatus(ctx) + } + if statusErr := c.waitForStatus(ctx); statusErr == nil { + log.Printf("[whatsapp:%s] start request returned error after gateway created the session: %v", c.account.AccountID, err) + return nil + } + return err +} + +// runEvents reads gateway websocket messages until the connection closes or the context ends. +func (c *whatsappChannel) runEvents(ctx context.Context) error { + wsURL, err := c.eventsURL() + if err != nil { + return err + } + header := http.Header{} + if auth := c.authHeader(); auth != "" { + header.Set("Authorization", auth) + } + dialer := *websocket.DefaultDialer + dialer.HandshakeTimeout = wsHandshakeTimeout + conn, _, err := dialer.DialContext(ctx, wsURL, header) + if err != nil { + return err + } + defer conn.Close() + + go func() { + <-ctx.Done() + _ = conn.Close() + }() + _ = conn.SetReadDeadline(time.Now().Add(wsReadTimeout)) + conn.SetPongHandler(func(string) error { + return conn.SetReadDeadline(time.Now().Add(wsReadTimeout)) + }) + pingDone := make(chan struct{}) + defer close(pingDone) + go func() { + ticker := time.NewTicker(wsPingInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-pingDone: + return + case <-ticker.C: + deadline := time.Now().Add(10 * time.Second) + if err := conn.WriteControl(websocket.PingMessage, nil, deadline); err != nil { + _ = conn.Close() + return + } + } + } + }() + + for ctx.Err() == nil { + _, payload, err := conn.ReadMessage() + if err != nil { + return err + } + c.handleWSPayload(ctx, payload) + } + return ctx.Err() +} + +// waitForStatus waits briefly until the gateway exposes the started session status. +func (c *whatsappChannel) waitForStatus(ctx context.Context) error { + deadline := time.Now().Add(startTimeout) + for ctx.Err() == nil { + var response map[string]any + if err := c.requestJSON(ctx, http.MethodGet, fmt.Sprintf("whatsapp/%s/status", url.PathEscape(c.sessionKey())), nil, &response); err == nil { + if data, ok := response["data"].(map[string]any); ok { + c.applySnapshot(data) + } + return nil + } + if time.Now().After(deadline) { + return fmt.Errorf("WhatsApp session status did not become available within %s", startTimeout) + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(defaultReconnect): + } + } + return ctx.Err() +} + +// handleWSPayload dispatches a gateway websocket payload to snapshot or event handling. +func (c *whatsappChannel) handleWSPayload(ctx context.Context, payload []byte) { + var obj map[string]any + if err := json.Unmarshal(payload, &obj); err != nil { + return + } + data, _ := obj["data"].(map[string]any) + switch strings.TrimSpace(fmt.Sprint(obj["type"])) { + case "snapshot": + c.applySnapshot(data) + case "event": + c.handleEvent(ctx, data) + } +} + +// handleEvent converts a WhatsApp message event and queues it for ordered processing. +func (c *whatsappChannel) handleEvent(ctx context.Context, item map[string]any) { + if fmt.Sprint(item["kind"]) != "message" { + return + } + messageID := strings.TrimSpace(fmt.Sprint(item["message_id"])) + if messageID == "" { + return + } + incoming := core.IncomingMessage{ + Channel: c.ChannelID(), + AccountID: c.account.AccountID, + ChatID: fmt.Sprint(item["chat_id"]), + ChatType: valueOrDefault(item["chat_type"], "p2p"), + MessageID: messageID, + SenderID: fmt.Sprint(item["sender_id"]), + Text: fmt.Sprint(item["text"]), + Raw: item, + } + _ = c.enqueueIncoming(ctx, incoming) +} + +// enqueueIncoming schedules message handling and marks the message seen after a successful handoff. +func (c *whatsappChannel) enqueueIncoming(ctx context.Context, incoming core.IncomingMessage) bool { + if ctx.Err() != nil { + return false + } + + now := time.Now() + var worker *whatsappChatWorker + startWorker := false + queueFull := false + + c.mu.Lock() + c.pruneSeenLocked(now) + if _, ok := c.seen[incoming.MessageID]; ok { + c.mu.Unlock() + return false + } + if c.workers == nil { + c.workers = map[string]*whatsappChatWorker{} + } + worker = c.workers[incoming.ChatID] + if worker == nil { + worker = &whatsappChatWorker{queue: make(chan core.IncomingMessage, messageQueueSize)} + c.workers[incoming.ChatID] = worker + startWorker = true + } + select { + case worker.queue <- incoming: + c.seen[incoming.MessageID] = now + c.mu.Unlock() + if startWorker { + go c.runChatWorker(ctx, incoming.ChatID, worker) + } + return true + case <-ctx.Done(): + c.mu.Unlock() + default: + queueFull = true + c.mu.Unlock() + } + if startWorker { + go c.runChatWorker(ctx, incoming.ChatID, worker) + } + if queueFull { + log.Printf("[whatsapp:%s] dropping message %s for chat %s: queue is full", c.account.AccountID, incoming.MessageID, incoming.ChatID) + } + return false +} + +// runChatWorker processes one chat's inbound messages sequentially. +func (c *whatsappChannel) runChatWorker(ctx context.Context, chatID string, worker *whatsappChatWorker) { + idle := time.NewTimer(chatWorkerIdle) + defer idle.Stop() + for { + select { + case <-ctx.Done(): + return + case msg := <-worker.queue: + if ctx.Err() != nil { + return + } + c.handleIncoming(ctx, msg) + resetTimer(idle, chatWorkerIdle) + case <-idle.C: + if c.retireChatWorker(chatID, worker) { + return + } + resetTimer(idle, chatWorkerIdle) + } + } +} + +// retireChatWorker removes an idle worker only while it is still current and empty. +func (c *whatsappChannel) retireChatWorker(chatID string, worker *whatsappChatWorker) bool { + c.mu.Lock() + defer c.mu.Unlock() + current := c.workers[chatID] + if current != worker { + return true + } + if len(worker.queue) > 0 { + return false + } + if current == worker { + delete(c.workers, chatID) + } + return true +} + +// handleIncoming invokes the installed RAGFlow bridge for one queued message. +func (c *whatsappChannel) handleIncoming(ctx context.Context, incoming core.IncomingMessage) { + c.mu.Lock() + handler := c.handler + c.mu.Unlock() + if handler == nil { + return + } + if err := handler(ctx, incoming); err != nil { + log.Printf("[whatsapp:%s] message handler error: %v", c.account.AccountID, err) + } +} + +// pruneSeenLocked removes expired duplicate-tracking entries while c.mu is held. +func (c *whatsappChannel) pruneSeenLocked(now time.Time) { + for key, ts := range c.seen { + if now.Sub(ts) > messageTTL { + delete(c.seen, key) + } + } +} + +// applySnapshot updates the cached connection and QR-code state from the gateway. +func (c *whatsappChannel) applySnapshot(snapshot map[string]any) { + c.mu.Lock() + defer c.mu.Unlock() + c.lastSnapshotAt = float64(time.Now().Unix()) + c.status = valueOrDefault(snapshot["status"], "connecting") + c.lastErr = valueOrDefault(snapshot["last_error"], "") + c.qrDataURL = valueOrDefault(snapshot["qr_data_url"], "") + c.qrUpdatedAt = numeric(snapshot["qr_updated_at"]) + c.connectedAt = numeric(snapshot["connected_at"]) + c.sessionID = valueOrDefault(snapshot["session_id"], c.sessionID) + if cursor := int64(numeric(snapshot["event_cursor"])); cursor > c.eventCursor { + c.eventCursor = cursor + } + if c.status == "connected" { + c.lastErr = "" + } +} + +// setError records a runtime failure in the snapshot exposed to the API. +func (c *whatsappChannel) setError(err error) { + c.mu.Lock() + defer c.mu.Unlock() + c.status = "error" + c.lastErr = err.Error() + c.lastSnapshotAt = float64(time.Now().Unix()) +} + +// requestJSON sends an authenticated JSON request to the WhatsApp gateway. +func (c *whatsappChannel) requestJSON(ctx context.Context, method, path string, body map[string]any, out any) error { + return c.requestJSONWithClient(ctx, c.client, method, path, body, out) +} + +// requestJSONWithTimeout sends a JSON request with a one-off timeout override. +func (c *whatsappChannel) requestJSONWithTimeout(ctx context.Context, timeout time.Duration, method, path string, body map[string]any, out any) error { + client := *c.client + client.Timeout = timeout + return c.requestJSONWithClient(ctx, &client, method, path, body, out) +} + +// requestJSONWithClient sends an authenticated JSON request using the provided HTTP client. +func (c *whatsappChannel) requestJSONWithClient(ctx context.Context, client *http.Client, method, path string, body map[string]any, out any) error { + payload, _ := json.Marshal(body) + req, err := http.NewRequestWithContext(ctx, method, strings.TrimRight(c.account.GatewayBaseURL, "/")+"/"+strings.TrimLeft(path, "/"), bytes.NewReader(payload)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + if auth := c.authHeader(); auth != "" { + req.Header.Set("Authorization", auth) + } + resp, err := client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1024*1024)) + if resp.StatusCode >= 400 { + return fmt.Errorf("status: %d, response: %s", resp.StatusCode, string(respBody)) + } + if out != nil && len(bytes.TrimSpace(respBody)) > 0 { + return json.Unmarshal(respBody, out) + } + return nil +} + +// sessionKey returns the configured WhatsApp gateway session key. +func (c *whatsappChannel) sessionKey() string { + if strings.TrimSpace(c.account.SessionKey) == "" { + return "default" + } + return c.account.SessionKey +} + +// authHeader formats the optional gateway token as an Authorization header. +func (c *whatsappChannel) authHeader() string { + token := strings.TrimSpace(c.account.GatewayToken) + if token == "" { + return "" + } + if strings.HasPrefix(strings.ToLower(token), "bearer ") { + return token + } + return "Bearer " + token +} + +// eventsURL builds the websocket URL used to consume gateway events after the cursor. +func (c *whatsappChannel) eventsURL() (string, error) { + base, err := url.Parse(strings.TrimRight(c.account.GatewayBaseURL, "/")) + if err != nil { + return "", err + } + switch base.Scheme { + case "http": + base.Scheme = "ws" + case "https": + base.Scheme = "wss" + } + base.Path = strings.TrimRight(base.Path, "/") + "/whatsapp/" + url.PathEscape(c.sessionKey()) + "/events/ws" + q := base.Query() + c.mu.Lock() + q.Set("after", strconv.FormatInt(c.eventCursor, 10)) + c.mu.Unlock() + base.RawQuery = q.Encode() + return base.String(), nil +} + +// firstString returns the first non-empty string value among several config keys. +func firstString(cfg map[string]any, keys ...string) string { + for _, key := range keys { + if value := strings.TrimSpace(fmt.Sprint(cfg[key])); value != "" && value != "" { + return value + } + } + return "" +} + +// valueOrDefault converts a snapshot value to string with an empty-value fallback. +func valueOrDefault(value any, fallback string) string { + text := strings.TrimSpace(fmt.Sprint(value)) + if text == "" || text == "" { + return fallback + } + return text +} + +// numeric converts gateway numeric fields from common JSON-decoded Go types. +func numeric(value any) float64 { + switch v := value.(type) { + case float64: + return v + case int: + return float64(v) + case int64: + return float64(v) + case json.Number: + n, _ := v.Float64() + return n + case string: + n, _ := strconv.ParseFloat(v, 64) + return n + default: + return 0 + } +} + +// stringPtr converts an empty string to nil for Python-compatible JSON null fields. +func stringPtr(value string) *string { + if value == "" { + return nil + } + return &value +} + +// floatPtr converts zero timestamps to nil for Python-compatible JSON null fields. +func floatPtr(value float64) *float64 { + if value == 0 { + return nil + } + return &value +} + +// resetTimer restarts a timer after draining any pending tick. +func resetTimer(timer *time.Timer, delay time.Duration) { + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + timer.Reset(delay) +} diff --git a/internal/channels/whatsapp_gateway.go b/internal/channels/whatsapp_gateway.go new file mode 100644 index 0000000000..a359051191 --- /dev/null +++ b/internal/channels/whatsapp_gateway.go @@ -0,0 +1,255 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package channels + +import ( + "context" + "errors" + "fmt" + "log" + "net/http" + "os" + "os/exec" + "path/filepath" + "runtime" + "strconv" + "strings" + "sync" + "time" +) + +const ( + gatewayStartTimeout = 30 * time.Second + gatewayProbeEvery = 500 * time.Millisecond + gatewayStopTimeout = 10 * time.Second +) + +type gatewayProcess struct { + cmd *exec.Cmd + done chan struct{} +} + +type gatewayRuntime struct { + mu sync.Mutex + process *gatewayProcess +} + +var gateway gatewayRuntime + +// syncWhatsAppGateway starts or stops the shared WhatsApp gateway process when management is enabled. +func syncWhatsAppGateway(ctx context.Context, enabled bool) error { + if !enabled { + return gateway.stop() + } + if !gatewayEnabled() { + return nil + } + return gateway.start(ctx) +} + +// gatewayEnabled reads the environment switch controlling gateway process management. +func gatewayEnabled() bool { + raw := strings.TrimSpace(os.Getenv("WHATSAPP_GATEWAY_ENABLED")) + if raw == "" { + return false + } + v, err := strconv.ParseBool(raw) + return err == nil && v +} + +// gatewayCommand resolves the command and working directory used to start the gateway. +func gatewayCommand() ([]string, string) { + if raw := strings.TrimSpace(os.Getenv("WHATSAPP_GATEWAY_COMMAND")); raw != "" { + return strings.Fields(raw), gatewayWorkdir() + } + + entry := filepath.Join(gatewayWorkdir(), "index.js") + if _, err := os.Stat(entry); err != nil { + return nil, gatewayWorkdir() + } + return []string{"node", entry}, gatewayWorkdir() +} + +// gatewayWorkdir resolves the WhatsApp gateway-node working directory. +func gatewayWorkdir() string { + if raw := strings.TrimSpace(os.Getenv("WHATSAPP_GATEWAY_WORKDIR")); raw != "" { + return raw + } + _, file, _, ok := runtime.Caller(0) + if !ok { + return "." + } + return filepath.Clean(filepath.Join(filepath.Dir(file), "../../api/channels/whatsapp/gateway-node")) +} + +// start launches the gateway process if it is not already running. +func (r *gatewayRuntime) start(ctx context.Context) error { + r.mu.Lock() + if r.process != nil { + process := r.process + select { + case <-process.done: + r.process = nil + default: + r.mu.Unlock() + if err := waitForGateway(ctx, gatewayStartTimeout); err != nil { + r.clearProcess(process) + _ = stopGatewayProcess(process) + return err + } + return nil + } + } + if gatewayReachable(ctx) { + r.mu.Unlock() + return nil + } + + argv, cwd := gatewayCommand() + if len(argv) == 0 { + r.mu.Unlock() + return errors.New("WhatsApp gateway command is not configured") + } + + cmd := exec.Command(argv[0], argv[1:]...) + cmd.Dir = cwd + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + cmd.Env = os.Environ() + if err := cmd.Start(); err != nil { + r.mu.Unlock() + return err + } + process := &gatewayProcess{cmd: cmd, done: make(chan struct{})} + r.process = process + r.mu.Unlock() + + go func() { + err := cmd.Wait() + close(process.done) + + r.mu.Lock() + owned := r.process == process + if r.process == process { + r.process = nil + } + r.mu.Unlock() + if err != nil && owned { + log.Printf("whatsapp gateway exited: %v", err) + } + }() + if err := waitForGateway(ctx, gatewayStartTimeout); err != nil { + r.clearProcess(process) + _ = stopGatewayProcess(process) + return err + } + return nil +} + +// stop terminates the gateway process if this runtime started one. +func (r *gatewayRuntime) stop() error { + r.mu.Lock() + process := r.process + r.process = nil + r.mu.Unlock() + return stopGatewayProcess(process) +} + +// clearProcess removes process from the runtime if it is still the current managed process. +func (r *gatewayRuntime) clearProcess(process *gatewayProcess) { + r.mu.Lock() + defer r.mu.Unlock() + if r.process == process { + r.process = nil + } +} + +// stopGatewayProcess terminates one owned gateway process and waits for it to exit. +func stopGatewayProcess(process *gatewayProcess) error { + if process == nil || process.cmd == nil || process.cmd.Process == nil { + return nil + } + select { + case <-process.done: + return nil + default: + } + + if err := process.cmd.Process.Signal(os.Interrupt); err != nil { + _ = process.cmd.Process.Kill() + return err + } + select { + case <-process.done: + return nil + case <-time.After(gatewayStopTimeout): + _ = process.cmd.Process.Kill() + } + select { + case <-process.done: + case <-time.After(2 * time.Second): + } + return nil +} + +// gatewayReachable reports whether an externally managed gateway is already listening. +func gatewayReachable(ctx context.Context) bool { + probeCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + defer cancel() + + req, err := http.NewRequestWithContext(probeCtx, http.MethodGet, gatewayHealthURL(), nil) + if err != nil { + return false + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + return false + } + defer resp.Body.Close() + return resp.StatusCode >= 200 && resp.StatusCode < 300 +} + +// waitForGateway blocks until the gateway health endpoint is ready or times out. +func waitForGateway(ctx context.Context, timeout time.Duration) error { + deadline := time.Now().Add(timeout) + for { + if gatewayReachable(ctx) { + return nil + } + if time.Now().After(deadline) { + return fmt.Errorf("WhatsApp gateway did not become ready at %s within %s", gatewayHealthURL(), timeout) + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(gatewayProbeEvery): + } + } +} + +// gatewayHealthURL returns the health endpoint for the default managed gateway address. +func gatewayHealthURL() string { + host := strings.TrimSpace(os.Getenv("WHATSAPP_GATEWAY_HOST")) + if host == "" { + host = "127.0.0.1" + } + port := strings.TrimSpace(os.Getenv("WHATSAPP_GATEWAY_PORT")) + if port == "" { + port = "3005" + } + return fmt.Sprintf("http://%s:%s/health", host, port) +} diff --git a/internal/dao/chat_channel.go b/internal/dao/chat_channel.go index 4c67e9e04d..f845413bce 100644 --- a/internal/dao/chat_channel.go +++ b/internal/dao/chat_channel.go @@ -58,3 +58,10 @@ func (dao *ChatChannelDAO) ListByTenantID(ctx context.Context, db *gorm.DB, tena return results, err } + +// ListActive returns all enabled chat-channel rows for the runtime reconciler. +func (dao *ChatChannelDAO) ListActive(ctx context.Context, db *gorm.DB) ([]*entity.ChatChannel, error) { + rows := make([]*entity.ChatChannel, 0) + err := db.WithContext(ctx).Where("status = ?", 1).Find(&rows).Error + return rows, err +} diff --git a/internal/handler/chat_channel.go b/internal/handler/chat_channel.go index d7ee024476..9ccaf220bf 100644 --- a/internal/handler/chat_channel.go +++ b/internal/handler/chat_channel.go @@ -30,6 +30,7 @@ type ChatChannelService interface { List(ctx context.Context, tenantID string) ([]*entity.ChatChannelListResponse, error) GetChatChannel(ctx context.Context, userID, channelID string) (*entity.ChatChannel, common.ErrorCode, error) UpdateChatChannel(ctx context.Context, userID, channelID string, req map[string]interface{}) (*entity.ChatChannel, common.ErrorCode, error) + GetChatChannelRuntime(ctx context.Context, userID, channelID string) (map[string]any, common.ErrorCode, error) DeleteChatChannel(ctx context.Context, userID, channelID string) (bool, common.ErrorCode, error) } @@ -49,7 +50,7 @@ func NewChatChannel() *ChatChannelHandler { type CreateChatChannelRequest struct { Name string `json:"name" binding:"required"` Channel string `json:"channel" binding:"required"` - Config entity.JSONMap `json:"config" binding:"required"` + Config entity.JSONMap `json:"config"` ChatID *string `json:"chat_id"` } @@ -66,6 +67,9 @@ func (h *ChatChannelHandler) CreateChatChannel(c *gin.Context) { common.ResponseWithCodeData(c, common.CodeDataError, nil, "Invalid request: "+err.Error()) return } + if req.Config == nil { + req.Config = entity.JSONMap{} + } ctx := c.Request.Context() @@ -167,6 +171,35 @@ func (h *ChatChannelHandler) UpdateChatChannel(c *gin.Context) { common.SuccessWithData(c, result, "success") } +// GetChatChannelRuntime returns live runtime metadata for a running chat channel. +func (h *ChatChannelHandler) GetChatChannelRuntime(c *gin.Context) { + user, errorCode, errorMessage := GetUser(c) + if errorCode != common.CodeSuccess { + common.ErrorWithCode(c, errorCode, errorMessage) + return + } + + userID := strings.TrimSpace(user.ID) + if userID == "" { + common.ResponseWithCodeData(c, common.CodeArgumentError, nil, "user_id is required") + return + } + + channelID := strings.TrimSpace(c.Param("channel_id")) + if channelID == "" { + common.ResponseWithCodeData(c, common.CodeArgumentError, nil, "channel_id is required") + return + } + + ctx := c.Request.Context() + result, code, err := h.chatChannelService.GetChatChannelRuntime(ctx, userID, channelID) + if code != common.CodeSuccess || err != nil { + writeChatChannelError(c, code, chatChannelErrMsg(code, err)) + return + } + common.SuccessWithData(c, result, "success") +} + // DeleteChatChannel handles DELETE /chat-channels/:channel_id. func (h *ChatChannelHandler) DeleteChatChannel(c *gin.Context) { user, errorCode, errorMessage := GetUser(c) diff --git a/internal/handler/chat_channel_test.go b/internal/handler/chat_channel_test.go index 7cb98d4abc..f130db65dc 100644 --- a/internal/handler/chat_channel_test.go +++ b/internal/handler/chat_channel_test.go @@ -16,11 +16,12 @@ import ( ) type fakeChatChannelService struct { - createFn func(tenantID, name, channelType string, config entity.JSONMap, chatID *string) (*entity.ChatChannel, error) - listFn func(tenantID string) ([]*entity.ChatChannelListResponse, error) - getFn func(userID, channelID string) (*entity.ChatChannel, common.ErrorCode, error) - updateFn func(userID, channelID string, req map[string]interface{}) (*entity.ChatChannel, common.ErrorCode, error) - deleteFn func(userID, channelID string) (bool, common.ErrorCode, error) + createFn func(tenantID, name, channelType string, config entity.JSONMap, chatID *string) (*entity.ChatChannel, error) + listFn func(tenantID string) ([]*entity.ChatChannelListResponse, error) + getFn func(userID, channelID string) (*entity.ChatChannel, common.ErrorCode, error) + updateFn func(userID, channelID string, req map[string]interface{}) (*entity.ChatChannel, common.ErrorCode, error) + runtimeFn func(userID, channelID string) (map[string]any, common.ErrorCode, error) + deleteFn func(userID, channelID string) (bool, common.ErrorCode, error) } func (f fakeChatChannelService) CreateChatChannel(ctx context.Context, tenantID, name, channelType string, config entity.JSONMap, chatID *string) (*entity.ChatChannel, error) { @@ -51,6 +52,13 @@ func (f fakeChatChannelService) UpdateChatChannel(ctx context.Context, userID, c return f.updateFn(userID, channelID, req) } +func (f fakeChatChannelService) GetChatChannelRuntime(ctx context.Context, userID, channelID string) (map[string]any, common.ErrorCode, error) { + if f.runtimeFn == nil { + return nil, common.CodeServerError, errors.New("unexpected GetChatChannelRuntime call") + } + return f.runtimeFn(userID, channelID) +} + func (f fakeChatChannelService) DeleteChatChannel(ctx context.Context, userID, channelID string) (bool, common.ErrorCode, error) { if f.deleteFn == nil { return false, common.CodeServerError, errors.New("unexpected DeleteChatChannel call") @@ -327,3 +335,44 @@ func TestChatChannelHandlerDeleteChatChannelSuccess(t *testing.T) { t.Fatalf("payload=%v", payload) } } + +func TestChatChannelHandlerGetRuntimeSuccess(t *testing.T) { + gin.SetMode(gin.TestMode) + + var gotUserID, gotChannelID string + h := &ChatChannelHandler{ + chatChannelService: fakeChatChannelService{ + runtimeFn: func(userID, channelID string) (map[string]any, common.ErrorCode, error) { + gotUserID = userID + gotChannelID = channelID + return map[string]any{"status": "waiting"}, common.CodeSuccess, nil + }, + }, + } + + router := gin.New() + router.GET("/api/v1/chat-channels/:channel_id/runtime", func(c *gin.Context) { + c.Set("user", &entity.User{ID: "tenant-1"}) + h.GetChatChannelRuntime(c) + }) + + resp := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/api/v1/chat-channels/cc-1/runtime", nil) + router.ServeHTTP(resp, req) + + if gotUserID != "tenant-1" || gotChannelID != "cc-1" { + t.Fatalf("userID=%q channelID=%q", gotUserID, gotChannelID) + } + + var payload map[string]interface{} + if err := json.Unmarshal(resp.Body.Bytes(), &payload); err != nil { + t.Fatalf("unmarshal response: %v", err) + } + if payload["code"] != float64(common.CodeSuccess) { + t.Fatalf("payload=%v", payload) + } + data, _ := payload["data"].(map[string]interface{}) + if data["status"] != "waiting" { + t.Fatalf("payload=%v", payload) + } +} diff --git a/internal/router/router.go b/internal/router/router.go index 408de9cbca..6d1d4f4b68 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -691,13 +691,14 @@ func (r *Router) Setup(engine *gin.Engine) { } // Chat Channel - chanChannel := v1.Group("/chat-channels") + chatChannel := v1.Group("/chat-channels") { - chanChannel.POST("", r.chatChannelHandler.CreateChatChannel) - chanChannel.GET("", r.chatChannelHandler.ListChatChannel) - chanChannel.GET("/:channel_id", r.chatChannelHandler.GetChatChannel) - chanChannel.PATCH("/:channel_id", r.chatChannelHandler.UpdateChatChannel) - chanChannel.DELETE("/:channel_id", r.chatChannelHandler.DeleteChatChannel) + chatChannel.POST("", r.chatChannelHandler.CreateChatChannel) + chatChannel.GET("", r.chatChannelHandler.ListChatChannel) + chatChannel.GET("/:channel_id", r.chatChannelHandler.GetChatChannel) + chatChannel.PATCH("/:channel_id", r.chatChannelHandler.UpdateChatChannel) + chatChannel.DELETE("/:channel_id", r.chatChannelHandler.DeleteChatChannel) + chatChannel.GET("/:channel_id/runtime", r.chatChannelHandler.GetChatChannelRuntime) } // Langfuse tracing keys diff --git a/internal/service/chat_channel.go b/internal/service/chat_channel.go index b672f3f29a..586e9e6aef 100644 --- a/internal/service/chat_channel.go +++ b/internal/service/chat_channel.go @@ -18,6 +18,10 @@ import ( "context" "errors" "fmt" + "log" + "strings" + "sync" + "ragflow/internal/utility" "ragflow/internal/common" @@ -29,6 +33,7 @@ type ChatChannelService struct { chatChannelDAO *dao.ChatChannelDAO chatDAO *dao.ChatDAO userTenantDAO *dao.UserTenantDAO + chatSessionSvc *ChatSessionService } func NewChatChannelService() *ChatChannelService { @@ -36,9 +41,42 @@ func NewChatChannelService() *ChatChannelService { chatChannelDAO: dao.NewChatChannel(), chatDAO: dao.NewChatDAO(), userTenantDAO: dao.NewUserTenantDAO(), + chatSessionSvc: NewChatSessionService(), } } +type ChatChannelIncomingMessage struct { + Channel string + AccountID string + ChatID string + ChatType string + MessageID string + SenderID string + Text string +} + +// ChatChannelRuntimeProvider reads runtime metadata for one live chat-channel instance. +type ChatChannelRuntimeProvider func(channelID string) (map[string]any, bool) + +var ( + chatChannelRuntimeProviderMu sync.RWMutex + chatChannelRuntimeProvider ChatChannelRuntimeProvider +) + +// SetChatChannelRuntimeProvider installs the runtime snapshot reader used by the chat-channel API. +func SetChatChannelRuntimeProvider(provider ChatChannelRuntimeProvider) { + chatChannelRuntimeProviderMu.Lock() + defer chatChannelRuntimeProviderMu.Unlock() + chatChannelRuntimeProvider = provider +} + +// getChatChannelRuntimeProvider returns the current runtime snapshot reader. +func getChatChannelRuntimeProvider() ChatChannelRuntimeProvider { + chatChannelRuntimeProviderMu.RLock() + defer chatChannelRuntimeProviderMu.RUnlock() + return chatChannelRuntimeProvider +} + func (s *ChatChannelService) Insert(ctx context.Context, channel *entity.ChatChannel) error { if channel == nil { return errors.New("channel is nil") @@ -216,6 +254,88 @@ func (s *ChatChannelService) UpdateChatChannel(ctx context.Context, userID, chan return updated, common.CodeSuccess, nil } +// GetChatChannelRuntime returns the live runtime metadata for a WhatsApp chat channel. +func (s *ChatChannelService) GetChatChannelRuntime(ctx context.Context, userID, channelID string) (map[string]any, common.ErrorCode, error) { + conn, ok, err := s.accessible(ctx, userID, channelID) + if err != nil { + return nil, common.CodeServerError, err + } + if !ok { + return nil, common.CodeAuthenticationError, errors.New("no authorization") + } + if conn == nil { + return nil, common.CodeDataError, errors.New("can't find this chat channel") + } + + if conn.Channel != "whatsapp" { + return nil, common.CodeDataError, errors.New("runtime snapshot is only available for WhatsApp") + } + + if provider := getChatChannelRuntimeProvider(); provider != nil { + if snapshot, ok := provider(channelID); ok { + return snapshot, common.CodeSuccess, nil + } + } + return waitingRuntimeSnapshotMap(channelID), common.CodeSuccess, nil +} + +// HandleIncomingMessage routes one external channel message through the bound RAGFlow dialog. +func (s *ChatChannelService) HandleIncomingMessage(ctx context.Context, msg ChatChannelIncomingMessage) (string, error) { + if strings.TrimSpace(msg.Text) == "" { + return "", nil + } + + channel, err := s.chatChannelDAO.GetByIDOnly(ctx, dao.DB, msg.AccountID) + if err != nil { + if dao.IsNotFoundErr(err) { + return "", nil + } + return "", err + } + if channel.ChatID == nil || strings.TrimSpace(*channel.ChatID) == "" { + return "", nil + } + + dialog, err := s.chatDAO.GetByID(ctx, dao.DB, *channel.ChatID) + if err != nil { + if dao.IsNotFoundErr(err) { + return "", nil + } + return "", err + } + + session, err := s.chatSessionSvc.GetOrCreateForChannel(ctx, dialog.ID, channel.ID, msg.ChatID) + if err != nil { + return "", err + } + + result, err := s.chatSessionSvc.ChatCompletions( + ctx, + dialog.TenantID, + dialog.ID, + session.ID, + []map[string]interface{}{{"role": "user", "content": msg.Text}}, + "", + nil, + "", + nil, + map[string]interface{}{"quote": false}, + false, + false, + false, + nil, + ) + if err != nil { + log.Printf("chat channel %s completion failed: %v", channel.ID, err) + return "Sorry, I couldn't process your message right now.", nil + } + if result == nil { + return "", nil + } + answer, _ := result["answer"].(string) + return answer, nil +} + func (s *ChatChannelService) DeleteChatChannel(ctx context.Context, userID, channelID string) (bool, common.ErrorCode, error) { channel, ok, err := s.accessible(ctx, userID, channelID) if err != nil { @@ -233,3 +353,20 @@ func (s *ChatChannelService) DeleteChatChannel(ctx context.Context, userID, chan } return true, common.CodeSuccess, nil } + +// waitingRuntimeSnapshotMap returns the API fallback payload before a runtime instance starts. +func waitingRuntimeSnapshotMap(channelID string) map[string]any { + return map[string]any{ + "account_id": channelID, + "session_key": channelID, + "status": "waiting", + "connected_at": nil, + "qr_updated_at": nil, + "qr_data_url": nil, + "last_error": nil, + "session_id": nil, + "last_snapshot_at": nil, + "gateway_base_url": nil, + "event_cursor": int64(0), + } +} diff --git a/internal/service/chat_session.go b/internal/service/chat_session.go index fe1f8c8c6d..a4f92d9bae 100644 --- a/internal/service/chat_session.go +++ b/internal/service/chat_session.go @@ -18,6 +18,7 @@ package service import ( "context" + "crypto/sha256" "encoding/json" "errors" "fmt" @@ -1827,6 +1828,43 @@ func (s *ChatSessionService) createSessionForCompletion(ctx context.Context, cha return session, nil } +// GetOrCreateForChannel finds or creates the deterministic conversation for one external chat. +func (s *ChatSessionService) GetOrCreateForChannel(ctx context.Context, dialogID, channelID, chatID string) (*entity.ChatSession, error) { + sessionID := channelSessionID(dialogID, channelID, chatID) + session, err := s.chatSessionDAO.GetByID(ctx, dao.DB, sessionID) + if err == nil { + return session, nil + } + if !dao.IsNotFoundErr(err) { + return nil, err + } + + messagesJSON, _ := json.Marshal([]map[string]interface{}{}) + referenceJSON, _ := json.Marshal([]interface{}{}) + name := fmt.Sprintf("channel:%s:%s", channelID, chatID) + session = &entity.ChatSession{ + ID: sessionID, + DialogID: dialogID, + Name: &name, + Message: messagesJSON, + Reference: referenceJSON, + } + if err = s.chatSessionDAO.Create(ctx, dao.DB, session); err != nil { + session, rereadErr := s.chatSessionDAO.GetByID(ctx, dao.DB, sessionID) + if rereadErr == nil { + return session, nil + } + return nil, err + } + return session, nil +} + +// channelSessionID derives the stable SHA-256 conversation ID for a channel chat. +func channelSessionID(dialogID, channelID, chatID string) string { + sum := sha256.Sum256([]byte(dialogID + ":" + channelID + ":" + chatID)) + return fmt.Sprintf("%x", sum[:])[:32] +} + // appendSessionMessage appends the last user message to the session's message history. func (s *ChatSessionService) appendSessionMessage(session *entity.ChatSession, requestMsg []map[string]interface{}) *entity.ChatSession { msgs := parseMessages(session.Message)