From 5249de53daf2e61e074fc65bf96b6329956e0ff8 Mon Sep 17 00:00:00 2001 From: Haruko386 Date: Fri, 31 Jul 2026 11:40:25 +0800 Subject: [PATCH] feat[Go]: add feishu chat bot for chat channel (#17561) ### Summary As title, related to #17520 --- go.mod | 2 + go.sum | 14 + internal/channels/bootstrap.go | 2 + internal/channels/feishu.go | 442 +++++++++++++++++++++++++++++++ internal/channels/feishu_test.go | 85 ++++++ 5 files changed, 545 insertions(+) create mode 100644 internal/channels/feishu_test.go diff --git a/go.mod b/go.mod index d399ffe451..deeb25ebb1 100644 --- a/go.mod +++ b/go.mod @@ -35,6 +35,7 @@ require ( github.com/iromli/go-itsdangerous v0.0.0-20220223194502-9c8bef8dac6a github.com/json-iterator/go v1.1.12 github.com/kaptinlin/jsonrepair v0.4.8 + github.com/larksuite/oapi-sdk-go/v3 v3.9.9 github.com/ledongthuc/pdf v0.0.0-20250511090121-5959a4027728 github.com/lib/pq v1.10.9 github.com/minio/minio-go/v7 v7.0.99 @@ -135,6 +136,7 @@ require ( github.com/go-playground/universal-translator v0.18.1 // indirect github.com/go-playground/validator/v10 v10.30.2 // indirect github.com/goccy/go-yaml v1.19.2 // indirect + github.com/gogo/protobuf v1.3.2 // indirect github.com/golang-sql/civil v0.0.0-20190719163853-cb61b32ac6fe // indirect github.com/golang-sql/sqlexp v0.1.0 // indirect github.com/google/go-cmp v0.7.0 // indirect diff --git a/go.sum b/go.sum index e82da0b254..29df731181 100644 --- a/go.sum +++ b/go.sum @@ -252,6 +252,8 @@ github.com/goccy/go-json v0.10.6/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PU github.com/goccy/go-yaml v1.19.2 h1:PmFC1S6h8ljIz6gMRBopkjP1TVT7xuwrButHID66PoM= github.com/goccy/go-yaml v1.19.2/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA= github.com/gofrs/uuid v3.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM= +github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= +github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= github.com/golang-jwt/jwt/v5 v5.3.0 h1:pv4AsKCKKZuqlgs5sUmn4x8UlGa0kEVt/puTpKx9vvo= github.com/golang-jwt/jwt/v5 v5.3.0/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE= github.com/golang-sql/civil v0.0.0-20190719163853-cb61b32ac6fe h1:lXe2qZdvpiX5WZkZR4hgp4KJVfY3nMkvmwbVkpv1rVY= @@ -301,6 +303,7 @@ github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORR github.com/gopherjs/gopherjs v0.0.0-20200217142428-fce0ec30dd00/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY= github.com/gopherjs/gopherjs v1.17.2 h1:fQnZVsXk8uxXIStYb0N4bGk7jeyTalG/wsZjQ25dO0g= github.com/gopherjs/gopherjs v1.17.2/go.mod h1:pRRIvn/QzFLrKfvEz3qUuEhtE/zLCWfreZ6J5gM2i+k= +github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4= @@ -324,6 +327,8 @@ github.com/juju/gnuflag v0.0.0-20171113085948-2ce1bb71843d/go.mod h1:2PavIy+JPci github.com/kaptinlin/jsonrepair v0.4.8 h1:9oaoEe/vaKgm8ko4TLjBLUEog6tBW6WUzZXLPL2yTCk= github.com/kaptinlin/jsonrepair v0.4.8/go.mod h1:eWRC42KDUT0MHkMplUN6necu59FQFqKOKe+86akpY3g= github.com/kardianos/osext v0.0.0-20190222173326-2bc1f35cddc0/go.mod h1:1NbS8ALrpOvjt0rHPNLyCIeMtbizbir8U//inJ+zuB8= +github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= +github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= github.com/kisielk/sqlstruct v0.0.0-20201105191214-5f3e10d3ab46/go.mod h1:yyMNCyc/Ib3bDTKd379tNMpB/7/H5TjM2Y9QJ5THLbE= github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao= github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= @@ -340,6 +345,8 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/larksuite/oapi-sdk-go/v3 v3.9.9 h1:qzVK5U1AuT/n0Z4LqCN2ATU6MlYp2Iipa7q+fVK5Nh8= +github.com/larksuite/oapi-sdk-go/v3 v3.9.9/go.mod h1:ZEplY+kwuIrj/nqw5uSCINNATcH3KdxSN7y+UxYY5fI= github.com/ledongthuc/pdf v0.0.0-20250511090121-5959a4027728 h1:QwWKgMY28TAXaDl+ExRDqGQltzXqN/xypdKP86niVn8= github.com/ledongthuc/pdf v0.0.0-20250511090121-5959a4027728/go.mod h1:1fEHWurg7pvf5SG6XNE5Q8UZmOwex51Mkx3SLhrW5B4= github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ= @@ -525,6 +532,7 @@ github.com/yfedoseev/pdf_oxide/go v0.3.67 h1:Fm1R/KtpmJPNbVmdT1fvYM/Yl41Uu2FdyT7 github.com/yfedoseev/pdf_oxide/go v0.3.67/go.mod h1:QbJ/nLbez0al2EnqEdEPIlGflFprWmiuUM4mo9rNNOI= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.1.30/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.4.13 h1:fVcFKWvrslecOb/tg+Cc05dkeYx540o0FuFt3nUVDoE= github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= @@ -599,6 +607,7 @@ golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTk golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU= golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc= golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= golang.org/x/mod v0.12.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= @@ -614,6 +623,7 @@ golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLL golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20200506145744-7e3656a0809f/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A= golang.org/x/net v0.0.0-20201010224723-4f7140c49acb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= +golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= golang.org/x/net v0.0.0-20210610132358-84b48f89b13b/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= @@ -638,6 +648,7 @@ golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y= @@ -708,6 +719,8 @@ golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3 golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20200509030707-2212a7e161a5/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= +golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= +golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU= golang.org/x/tools v0.13.0/go.mod h1:HvlwmtVNQAhOuCjW7xxvovg8wbNq7LwfXh/k7wXUl58= @@ -715,6 +728,7 @@ golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxb golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= google.golang.org/api v0.287.1 h1:LiyJx32VU3cwQfLchn/513qKhc25hq0pEANYJoWNnnI= diff --git a/internal/channels/bootstrap.go b/internal/channels/bootstrap.go index b003d87ec2..48c5a89364 100644 --- a/internal/channels/bootstrap.go +++ b/internal/channels/bootstrap.go @@ -277,6 +277,8 @@ func (r *Runtime) startChannel(ctx context.Context, accountID string, wanted des // 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 "feishu": + return newFeishuChannelFromConfig(accountID, wanted.credential) case "discord": return newDiscordChannelFromConfig(accountID, wanted.credential) case "qqbot": diff --git a/internal/channels/feishu.go b/internal/channels/feishu.go index a4ce67278c..ba23fe9ee2 100644 --- a/internal/channels/feishu.go +++ b/internal/channels/feishu.go @@ -15,3 +15,445 @@ // package channels + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log" + "strconv" + "strings" + "sync" + "time" + + lark "github.com/larksuite/oapi-sdk-go/v3" + larkcore "github.com/larksuite/oapi-sdk-go/v3/core" + "github.com/larksuite/oapi-sdk-go/v3/event/dispatcher" + larkim "github.com/larksuite/oapi-sdk-go/v3/service/im/v1" + larkws "github.com/larksuite/oapi-sdk-go/v3/ws" + + "ragflow/internal/channels/core" +) + +const ( + defaultFeishuDomain = "feishu" + defaultFeishuTimeout = 30 * time.Second + feishuMessageTTL = time.Hour + feishuQueueSize = 64 + feishuWorkerIdle = 5 * time.Minute +) + +type feishuAccount struct { + AccountID string + AppID string + AppSecret string + Domain string + Timeout time.Duration +} + +type feishuChannel struct { + account feishuAccount + + mu sync.Mutex + ctx context.Context + cancel context.CancelFunc + handler core.MessageHandler + rest *lark.Client + wsClient *larkws.Client + seen map[string]time.Time + workers map[string]*feishuWorker +} + +type feishuWorker struct { + queue chan core.IncomingMessage +} + +// newFeishuChannel creates a Feishu channel backed by the official Lark SDK. +func newFeishuChannel(account feishuAccount) *feishuChannel { + if account.Domain == "" { + account.Domain = defaultFeishuDomain + } + if account.Timeout <= 0 { + account.Timeout = defaultFeishuTimeout + } + ch := &feishuChannel{ + account: account, + ctx: context.Background(), + seen: map[string]time.Time{}, + workers: map[string]*feishuWorker{}, + } + ch.rest = lark.NewClient( + account.AppID, + account.AppSecret, + lark.WithOpenBaseUrl(feishuBaseURL(account.Domain)), + lark.WithLogLevel(larkcore.LogLevelInfo), + lark.WithReqTimeout(account.Timeout), + ) + ch.wsClient = larkws.NewClient( + account.AppID, + account.AppSecret, + larkws.WithDomain(feishuBaseURL(account.Domain)), + larkws.WithEventHandler(ch.eventDispatcher()), + larkws.WithLogLevel(larkcore.LogLevelInfo), + ) + return ch +} + +// newFeishuChannelFromConfig builds a Feishu channel from chat_channel.config.credential. +func newFeishuChannelFromConfig(accountID string, cfg map[string]any) (*feishuChannel, error) { + appID := firstString(cfg, "app_id", "appId") + appSecret := firstString(cfg, "app_secret", "appSecret") + if appID == "" || appSecret == "" { + return nil, fmt.Errorf("feishu account %q is missing app_id or app_secret", accountID) + } + + timeout := defaultFeishuTimeout + if raw, ok := cfg["timeout_secs"]; ok { + if parsed := feishuDurationSeconds(raw); parsed > 0 { + timeout = parsed + } + } + + return newFeishuChannel(feishuAccount{ + AccountID: accountID, + AppID: appID, + AppSecret: appSecret, + Domain: valueOrDefault(firstString(cfg, "domain"), defaultFeishuDomain), + Timeout: timeout, + }), nil +} + +// ChannelID returns the platform identifier used by the chat-channel runtime. +func (c *feishuChannel) ChannelID() string { + return "feishu" +} + +// AccountID returns the chat_channel.id bound to this Feishu runtime instance. +func (c *feishuChannel) AccountID() string { + return c.account.AccountID +} + +// SetMessageHandler installs the RAGFlow bridge invoked for inbound Feishu messages. +func (c *feishuChannel) SetMessageHandler(handler core.MessageHandler) { + c.mu.Lock() + defer c.mu.Unlock() + c.handler = handler +} + +// Start begins the Feishu WebSocket event client. +func (c *feishuChannel) Start(ctx context.Context) error { + c.mu.Lock() + if c.cancel != nil { + c.mu.Unlock() + return nil + } + + runCtx, cancel := context.WithCancel(ctx) + c.ctx = runCtx + c.cancel = cancel + c.mu.Unlock() + + go c.run(runCtx) + return nil +} + +// Stop cancels the Feishu WebSocket client and clears transient workers. +func (c *feishuChannel) Stop(ctx context.Context) error { + c.mu.Lock() + cancel := c.cancel + wsClient := c.wsClient + + c.cancel = nil + c.seen = map[string]time.Time{} + c.workers = map[string]*feishuWorker{} + c.mu.Unlock() + + if cancel != nil { + cancel() + } + + if wsClient != nil { + wsClient.Close() + } + return nil +} + +// Send posts an outgoing RAGFlow answer to Feishu. +func (c *feishuChannel) Send(ctx context.Context, msg core.OutgoingMessage) error { + if strings.TrimSpace(msg.ChatID) == "" { + return errors.New("chat_id is required") + } + if strings.TrimSpace(msg.Text) == "" { + return nil + } + + content, _ := json.Marshal(map[string]string{"text": msg.Text}) + contentText := string(content) + msgType := "text" + + if strings.TrimSpace(msg.ReplyToMessageID) != "" { + resp, err := c.rest.Im.Message.Reply(ctx, larkim.NewReplyMessageReqBuilder(). + MessageId(msg.ReplyToMessageID). + Body(larkim.NewReplyMessageReqBodyBuilder(). + MsgType(msgType). + Content(contentText). + Build()). + Build()) + if err != nil { + return err + } + return feishuResponseError("reply", resp.Code, resp.Msg) + } + + resp, err := c.rest.Im.Message.Create(ctx, larkim.NewCreateMessageReqBuilder(). + ReceiveIdType("chat_id"). + Body(larkim.NewCreateMessageReqBodyBuilder(). + ReceiveId(msg.ChatID). + MsgType(msgType). + Content(contentText). + Build()). + Build()) + if err != nil { + return err + } + return feishuResponseError("create", resp.Code, resp.Msg) +} + +// run keeps the SDK WebSocket client active until the channel stops. +func (c *feishuChannel) run(ctx context.Context) { + if c.wsClient == nil { + log.Printf("[feishu:%s] WebSocket client is not initialized", c.account.AccountID) + return + } + if err := c.wsClient.Start(ctx); err != nil && ctx.Err() == nil { + log.Printf("[feishu:%s] WebSocket client exited: %v", c.account.AccountID, err) + } +} + +// eventDispatcher builds the Feishu event handler used by the WebSocket client. +func (c *feishuChannel) eventDispatcher() *dispatcher.EventDispatcher { + return dispatcher.NewEventDispatcher("", "").OnP2MessageReceiveV1( + func(ctx context.Context, event *larkim.P2MessageReceiveV1) error { + incoming, ok := c.normalizeMessage(event) + if !ok { + return nil + } + c.enqueueIncoming(incoming) + return nil + }, + ) +} + +// normalizeMessage converts a Feishu SDK message event to the chat-channel message model. +func (c *feishuChannel) normalizeMessage(event *larkim.P2MessageReceiveV1) (core.IncomingMessage, bool) { + if event == nil || event.Event == nil || event.Event.Message == nil { + return core.IncomingMessage{}, false + } + + message := event.Event.Message + messageID := stringValue(message.MessageId) + if messageID == "" { + return core.IncomingMessage{}, false + } + + raw := map[string]any{} + if data, err := json.Marshal(event); err == nil { + _ = json.Unmarshal(data, &raw) + } + + return core.IncomingMessage{ + Channel: c.ChannelID(), + AccountID: c.account.AccountID, + ChatID: stringValue(message.ChatId), + ChatType: stringValue(message.ChatType), + MessageID: messageID, + SenderID: feishuSenderID(event.Event.Sender), + Text: feishuMessageText(message.Content), + Raw: raw, + }, true +} + +// enqueueIncoming schedules Feishu message handling and marks it seen after a successful handoff. +func (c *feishuChannel) enqueueIncoming(incoming core.IncomingMessage) bool { + now := time.Now() + var worker *feishuWorker + 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]*feishuWorker{} + } + worker = c.workers[incoming.ChatID] + if worker == nil { + worker = &feishuWorker{queue: make(chan core.IncomingMessage, feishuQueueSize)} + c.workers[incoming.ChatID] = worker + startWorker = true + } + + select { + case worker.queue <- incoming: + c.seen[incoming.MessageID] = now + c.mu.Unlock() + if startWorker { + go c.runWorker(c.ctx, incoming.ChatID, worker) + } + return true + case <-c.ctx.Done(): + c.mu.Unlock() + default: + queueFull = true + c.mu.Unlock() + } + + if startWorker { + go c.runWorker(c.ctx, incoming.ChatID, worker) + } + if queueFull { + log.Printf("[feishu:%s] dropping message %s for chat %s: queue is full", c.account.AccountID, incoming.MessageID, incoming.ChatID) + } + return false +} + +// runWorker processes one Feishu chat's inbound messages sequentially. +func (c *feishuChannel) runWorker(ctx context.Context, chatID string, worker *feishuWorker) { + idle := time.NewTimer(feishuWorkerIdle) + defer idle.Stop() + for { + select { + case <-ctx.Done(): + return + case msg := <-worker.queue: + if ctx.Err() != nil { + return + } + handlerCtx, cancel := context.WithTimeout(ctx, c.account.Timeout) + c.handleIncoming(handlerCtx, msg) + cancel() + resetTimer(idle, feishuWorkerIdle) + case <-idle.C: + if c.retireWorker(chatID, worker) { + return + } + resetTimer(idle, feishuWorkerIdle) + } + } +} + +// retireWorker removes an idle Feishu chat worker only while it is current and empty. +func (c *feishuChannel) retireWorker(chatID string, worker *feishuWorker) bool { + c.mu.Lock() + defer c.mu.Unlock() + current := c.workers[chatID] + if current != worker { + return true + } + if len(worker.queue) > 0 { + return false + } + delete(c.workers, chatID) + return true +} + +// handleIncoming invokes the installed RAGFlow bridge for one queued Feishu message. +func (c *feishuChannel) 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("[feishu:%s] message handler error: %v", c.account.AccountID, err) + } +} + +// pruneSeenLocked removes expired duplicate-tracking entries while c.mu is held. +func (c *feishuChannel) pruneSeenLocked(now time.Time) { + for key, ts := range c.seen { + if now.Sub(ts) > feishuMessageTTL { + delete(c.seen, key) + } + } +} + +// feishuMessageText extracts plain text from Feishu's JSON-encoded message content. +func feishuMessageText(content *string) string { + raw := stringValue(content) + if raw == "" { + return "" + } + var payload map[string]any + if err := json.Unmarshal([]byte(raw), &payload); err != nil { + return raw + } + if text, ok := payload["text"].(string); ok { + return text + } + return "" +} + +// feishuSenderID returns the sender open_id from a Feishu event sender. +func feishuSenderID(sender *larkim.EventSender) string { + if sender == nil || sender.SenderId == nil { + return "" + } + return stringValue(sender.SenderId.OpenId) +} + +// feishuBaseURL maps Feishu/Lark domain config to the SDK base URL. +func feishuBaseURL(domain string) string { + if strings.EqualFold(strings.TrimSpace(domain), "lark") { + return lark.LarkBaseUrl + } + return lark.FeishuBaseUrl +} + +// feishuResponseError returns an error when a Feishu SDK response has a non-zero code. +func feishuResponseError(action string, code int, message string) error { + if code == 0 { + return nil + } + return fmt.Errorf("feishu %s failed: code=%d msg=%s", action, code, message) +} + +// feishuDurationSeconds converts a config value measured in seconds to a duration. +func feishuDurationSeconds(value any) time.Duration { + switch v := value.(type) { + case int: + return time.Duration(v) * time.Second + case int64: + return time.Duration(v) * time.Second + case float64: + return time.Duration(v) * time.Second + case json.Number: + n, err := v.Int64() + if err != nil { + return 0 + } + return time.Duration(n) * time.Second + case string: + n, err := strconv.Atoi(strings.TrimSpace(v)) + if err != nil { + return 0 + } + return time.Duration(n) * time.Second + default: + return 0 + } +} + +// stringValue dereferences an optional string. +func stringValue(value *string) string { + if value == nil { + return "" + } + return *value +} diff --git a/internal/channels/feishu_test.go b/internal/channels/feishu_test.go new file mode 100644 index 0000000000..d0c4eec335 --- /dev/null +++ b/internal/channels/feishu_test.go @@ -0,0 +1,85 @@ +// +// 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 ( + "testing" + "time" + + "ragflow/internal/channels/core" +) + +func TestNewFeishuChannelFromConfigRequiresCredentials(t *testing.T) { + if _, err := newFeishuChannelFromConfig("account-1", map[string]any{"app_id": "app-1"}); err == nil { + t.Fatal("newFeishuChannelFromConfig succeeded without app_secret") + } +} + +func TestNewFeishuChannelFromConfigNormalizesConfig(t *testing.T) { + ch, err := newFeishuChannelFromConfig("account-1", map[string]any{ + "app_id": "app-1", + "app_secret": "secret-1", + "domain": "lark", + "timeout_secs": "7", + }) + if err != nil { + t.Fatalf("newFeishuChannelFromConfig returned error: %v", err) + } + if ch.account.AppID != "app-1" { + t.Fatalf("app id = %q, want app-1", ch.account.AppID) + } + if ch.account.AppSecret != "secret-1" { + t.Fatalf("app secret = %q, want secret-1", ch.account.AppSecret) + } + if ch.account.Domain != "lark" { + t.Fatalf("domain = %q, want lark", ch.account.Domain) + } + if ch.account.Timeout != 7*time.Second { + t.Fatalf("timeout = %s, want 7s", ch.account.Timeout) + } +} + +func TestFeishuMessageText(t *testing.T) { + raw := `{"text":"hello"}` + if got := feishuMessageText(&raw); got != "hello" { + t.Fatalf("feishuMessageText() = %q, want hello", got) + } + + plain := "hello" + if got := feishuMessageText(&plain); got != "hello" { + t.Fatalf("feishuMessageText() = %q, want raw fallback", got) + } +} + +func TestFeishuEnqueueDoesNotMarkDroppedMessageSeen(t *testing.T) { + ch := newFeishuChannel(feishuAccount{AccountID: "account-1", AppID: "app-1", AppSecret: "secret-1"}) + worker := &feishuWorker{queue: make(chan core.IncomingMessage, 1)} + worker.queue <- core.IncomingMessage{MessageID: "queued"} + ch.workers["chat-1"] = worker + + ok := ch.enqueueIncoming(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") + } +}