diff --git a/server/cmd/server/router.go b/server/cmd/server/router.go index c48bda71968..7932d561f2e 100644 --- a/server/cmd/server/router.go +++ b/server/cmd/server/router.go @@ -386,6 +386,22 @@ func NewRouterWithOptions(pool *pgxpool.Pool, hub *realtime.Hub, bus *events.Bus // (inbound pipeline seams) is all it takes to add the platform // to the engine — no engine edit. connector, connectorLabel := buildLarkConnector(installSvc, larkClient) + var inboundMedia *lark.InboundMediaService + if downloader, ok := larkClient.(lark.MessageResourceDownloader); ok { + mediaSvc, mediaErr := lark.NewInboundMediaService( + downloader, + installSvc, + h.Storage, + lark.InboundMediaServiceConfig{Logger: slog.Default()}, + ) + if mediaErr != nil { + slog.Error("lark inbound media init failed", "error", mediaErr) + } else { + inboundMedia = mediaSvc + } + } else { + slog.Error("lark inbound media disabled: API client cannot download message resources") + } lark.RegisterFeishu(channelRegistry, lark.FeishuChannelDeps{ Connector: connector, APIClient: larkClient, @@ -393,7 +409,7 @@ func NewRouterWithOptions(pool *pgxpool.Pool, hub *realtime.Hub, bus *events.Bus Logger: slog.Default(), }) channelRouter.Register(channel.TypeFeishu, lark.NewFeishuResolverSet( - cs, feishuSession, auditLogger, resolverReplier, typingIndicator, + cs, feishuSession, auditLogger, resolverReplier, typingIndicator, inboundMedia, )) slog.Info("lark inbound pipeline wired", "connector", connectorLabel) diff --git a/server/internal/integrations/channel/engine/resolvers.go b/server/internal/integrations/channel/engine/resolvers.go index 80b1d76a6cc..3136ed6815a 100644 --- a/server/internal/integrations/channel/engine/resolvers.go +++ b/server/internal/integrations/channel/engine/resolvers.go @@ -92,6 +92,7 @@ type EnsureSessionParams struct { type AppendParams struct { SessionID pgtype.UUID Sender pgtype.UUID + Installation ResolvedInstallation InstallationID pgtype.UUID Message channel.InboundMessage ClaimToken pgtype.UUID diff --git a/server/internal/integrations/channel/engine/router.go b/server/internal/integrations/channel/engine/router.go index 1335ea54875..f013e05a530 100644 --- a/server/internal/integrations/channel/engine/router.go +++ b/server/internal/integrations/channel/engine/router.go @@ -263,6 +263,7 @@ func (r *Router) processClaimed(ctx context.Context, set ResolverSet, msg channe appendRes, err := set.Session.AppendMessage(ctx, AppendParams{ SessionID: sessionID, Sender: identity.UserID, + Installation: inst, InstallationID: inst.ID, Message: msg, ClaimToken: claimToken, diff --git a/server/internal/integrations/channel/engine/session.go b/server/internal/integrations/channel/engine/session.go index cba3e0a835b..e5ae381f13d 100644 --- a/server/internal/integrations/channel/engine/session.go +++ b/server/internal/integrations/channel/engine/session.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" + "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgtype" @@ -42,6 +43,8 @@ type SessionQueries interface { CreateChatSession(ctx context.Context, arg db.CreateChatSessionParams) (db.ChatSession, error) CreateChannelChatSessionBinding(ctx context.Context, arg db.CreateChannelChatSessionBindingParams) (db.ChannelChatSessionBinding, error) CreateChatMessage(ctx context.Context, arg db.CreateChatMessageParams) (db.ChatMessage, error) + CreateAttachment(ctx context.Context, arg db.CreateAttachmentParams) (db.Attachment, error) + LinkAttachmentsToChatMessage(ctx context.Context, arg db.LinkAttachmentsToChatMessageParams) ([]pgtype.UUID, error) TouchChatSession(ctx context.Context, id pgtype.UUID) error GetMostRecentUserChatMessage(ctx context.Context, chatSessionID pgtype.UUID) (db.ChatMessage, error) UpdateChannelChatSessionBindingReplyTarget(ctx context.Context, arg db.UpdateChannelChatSessionBindingReplyTargetParams) error @@ -71,6 +74,12 @@ func (a dbSessionQueries) CreateChannelChatSessionBinding(ctx context.Context, a func (a dbSessionQueries) CreateChatMessage(ctx context.Context, arg db.CreateChatMessageParams) (db.ChatMessage, error) { return a.q.CreateChatMessage(ctx, arg) } +func (a dbSessionQueries) CreateAttachment(ctx context.Context, arg db.CreateAttachmentParams) (db.Attachment, error) { + return a.q.CreateAttachment(ctx, arg) +} +func (a dbSessionQueries) LinkAttachmentsToChatMessage(ctx context.Context, arg db.LinkAttachmentsToChatMessageParams) ([]pgtype.UUID, error) { + return a.q.LinkAttachmentsToChatMessage(ctx, arg) +} func (a dbSessionQueries) TouchChatSession(ctx context.Context, id pgtype.UUID) error { return a.q.TouchChatSession(ctx, id) } @@ -243,12 +252,14 @@ func (s *ChatSession) createSessionAndBinding(ctx context.Context, in EnsureSess type AppendInput struct { SessionID pgtype.UUID Sender pgtype.UUID + WorkspaceID pgtype.UUID InstallationID pgtype.UUID Body string CommandText string MessageID string ThreadID string ClaimToken pgtype.UUID + MediaRefs []channel.MediaRef } // AppendUserMessage writes the user message into the chat_session (touching it @@ -257,6 +268,9 @@ type AppendInput struct { // ErrClaimLost when a concurrent reclaim rotated the dedup token mid-flight, in // which case the whole transaction rolls back (no chat_message lands). func (s *ChatSession) AppendUserMessage(ctx context.Context, in AppendInput) (AppendResult, error) { + if len(in.MediaRefs) > 0 && !in.WorkspaceID.Valid { + return AppendResult{}, errors.New("append user message: workspace_id is required for media") + } tx, err := s.tx.Begin(ctx) if err != nil { return AppendResult{}, fmt.Errorf("begin tx: %w", err) @@ -280,13 +294,53 @@ func (s *ChatSession) AppendUserMessage(ctx context.Context, in AppendInput) (Ap } } - if _, err := qtx.CreateChatMessage(ctx, db.CreateChatMessageParams{ + message, err := qtx.CreateChatMessage(ctx, db.CreateChatMessageParams{ ChatSessionID: in.SessionID, Role: "user", Content: in.Body, - }); err != nil { + }) + if err != nil { return AppendResult{}, fmt.Errorf("create chat message: %w", err) } + + if len(in.MediaRefs) > 0 { + attachmentIDs := make([]pgtype.UUID, 0, len(in.MediaRefs)) + for _, media := range in.MediaRefs { + id, err := uuid.NewV7() + if err != nil { + return AppendResult{}, fmt.Errorf("create attachment id: %w", err) + } + attachmentID := pgtype.UUID{Bytes: id, Valid: true} + if _, err := qtx.CreateAttachment(ctx, db.CreateAttachmentParams{ + ID: attachmentID, + WorkspaceID: in.WorkspaceID, + ChatSessionID: in.SessionID, + UploaderType: "member", + UploaderID: in.Sender, + Filename: media.Filename, + Url: media.URL, + ContentType: media.MimeType, + SizeBytes: media.SizeBytes, + }); err != nil { + return AppendResult{}, fmt.Errorf("create chat attachment: %w", err) + } + attachmentIDs = append(attachmentIDs, attachmentID) + } + bound, err := qtx.LinkAttachmentsToChatMessage(ctx, db.LinkAttachmentsToChatMessageParams{ + ChatMessageID: message.ID, + ChatSessionID: in.SessionID, + WorkspaceID: in.WorkspaceID, + UploaderType: "member", + UploaderID: in.Sender, + AttachmentIds: attachmentIDs, + }) + if err != nil { + return AppendResult{}, fmt.Errorf("link chat attachments: %w", err) + } + if len(bound) != len(attachmentIDs) { + return AppendResult{}, fmt.Errorf("link chat attachments: linked %d of %d", len(bound), len(attachmentIDs)) + } + } if err := qtx.TouchChatSession(ctx, in.SessionID); err != nil { return AppendResult{}, fmt.Errorf("touch chat session: %w", err) } diff --git a/server/internal/integrations/channel/engine/session_test.go b/server/internal/integrations/channel/engine/session_test.go index 3b9dd8e5658..23ccbb1dc61 100644 --- a/server/internal/integrations/channel/engine/session_test.go +++ b/server/internal/integrations/channel/engine/session_test.go @@ -37,8 +37,11 @@ func (fakeTxStarter) Begin(context.Context) (pgx.Tx, error) { return fakeTx{}, n type fakeSessionQueries struct { bindings map[string]pgtype.UUID nextSession byte + nextMessage byte createdSessions int messages []string + attachments []db.CreateAttachmentParams + linked []pgtype.UUID touched int replyTargets int lockedWorkspace int // count of LockWorkspaceForChatSessionCreate calls @@ -89,7 +92,18 @@ func (f *fakeSessionQueries) CreateChannelChatSessionBinding(_ context.Context, func (f *fakeSessionQueries) CreateChatMessage(_ context.Context, arg db.CreateChatMessageParams) (db.ChatMessage, error) { f.messages = append(f.messages, arg.Content) - return db.ChatMessage{}, nil + f.nextMessage++ + return db.ChatMessage{ID: uid(f.nextMessage + 100)}, nil +} + +func (f *fakeSessionQueries) CreateAttachment(_ context.Context, arg db.CreateAttachmentParams) (db.Attachment, error) { + f.attachments = append(f.attachments, arg) + return db.Attachment{ID: arg.ID}, nil +} + +func (f *fakeSessionQueries) LinkAttachmentsToChatMessage(_ context.Context, arg db.LinkAttachmentsToChatMessageParams) ([]pgtype.UUID, error) { + f.linked = append(f.linked, arg.AttachmentIds...) + return arg.AttachmentIds, nil } func (f *fakeSessionQueries) TouchChatSession(context.Context, pgtype.UUID) error { @@ -238,6 +252,43 @@ func TestAppendUserMessage_PlainText(t *testing.T) { } } +func TestAppendUserMessage_CreatesAndLinksInboundMedia(t *testing.T) { + f := newFake() + s := newTestSession(f) + _, err := s.AppendUserMessage(context.Background(), AppendInput{ + SessionID: uid(1), + Sender: uid(7), + WorkspaceID: uid(8), + Body: "[Image]", + MediaRefs: []channel.MediaRef{{ + Type: channel.MsgTypeImage, + StorageKey: "workspaces/ws/channels/feishu/image.png", + URL: "https://storage.example/image.png", + Filename: "feishu-image.png", + MimeType: "image/png", + SizeBytes: 4, + }}, + }) + if err != nil { + t.Fatalf("AppendUserMessage: %v", err) + } + if len(f.attachments) != 1 { + t.Fatalf("created attachments = %d, want 1", len(f.attachments)) + } + got := f.attachments[0] + if got.WorkspaceID != uid(8) || got.ChatSessionID != uid(1) || + got.UploaderType != "member" || got.UploaderID != uid(7) { + t.Errorf("attachment ownership mapping wrong: %+v", got) + } + if got.Filename != "feishu-image.png" || got.ContentType != "image/png" || + got.SizeBytes != 4 || got.Url != "https://storage.example/image.png" { + t.Errorf("attachment metadata mapping wrong: %+v", got) + } + if len(f.linked) != 1 || f.linked[0] != got.ID { + t.Errorf("linked attachment ids = %v, want [%v]", f.linked, got.ID) + } +} + func TestAppendUserMessage_NoReplyTargetWithoutMessageID(t *testing.T) { f := newFake() s := newTestSession(f) diff --git a/server/internal/integrations/channel/message.go b/server/internal/integrations/channel/message.go index 6b688a67872..e78831f057b 100644 --- a/server/internal/integrations/channel/message.go +++ b/server/internal/integrations/channel/message.go @@ -86,6 +86,10 @@ type MediaRef struct { Type MsgType // StorageKey locates the persisted object in Multica object storage. StorageKey string + // URL is the storage backend's durable object URL. It is persisted on the + // attachment row; clients still download through the authenticated + // attachment endpoint rather than consuming this URL directly. + URL string // Filename is the original display name, when the platform supplies // one. Filename string diff --git a/server/internal/integrations/lark/client.go b/server/internal/integrations/lark/client.go index 6b9d6964dd1..b706ddf7f00 100644 --- a/server/internal/integrations/lark/client.go +++ b/server/internal/integrations/lark/client.go @@ -124,6 +124,22 @@ type APIClient interface { DeleteMessageReaction(ctx context.Context, p DeleteReactionParams) error } +// MessageResourceDownloader is the binary-resource slice of the Lark API. +// Kept separate from APIClient so outbound-only fakes do not need to implement +// an inbound attachment method they never exercise. +type MessageResourceDownloader interface { + DownloadMessageResource(ctx context.Context, creds InstallationCredentials, messageID, fileKey, resourceType string) (MessageResource, error) +} + +// MessageResource is one binary object downloaded from a Lark message. +// ResourceType is supplied by the caller ("image" or "file"); the response +// headers provide the remaining metadata. +type MessageResource struct { + Data []byte + Filename string + ContentType string +} + // ListMessagesParams selects a bounded, recent window of messages in a // single Lark chat for the group-context prefetch. Only the fields the // enricher needs today are exposed (ChatID, ThreadID, PageSize, EndTime); diff --git a/server/internal/integrations/lark/content_flatten.go b/server/internal/integrations/lark/content_flatten.go index 85e1c6dab4d..91c0e972219 100644 --- a/server/internal/integrations/lark/content_flatten.go +++ b/server/internal/integrations/lark/content_flatten.go @@ -20,13 +20,12 @@ import ( // REST item) carry the mentions array differently — only the caller // knows which one applies — so flattening stays mention-agnostic. // -// Non-text media types render as a stable bracketed placeholder so the -// agent sees that *something* was attached without us downloading the -// binary. Attachment ingestion is explicitly out of scope (tracked as a -// separate attachment-pipeline issue), and merge_forward is intercepted -// by the enricher before it reaches here (expanding it needs an HTTP -// round-trip); the inline placeholder is only a fallback for a forward -// nested inside another forward. +// Non-text media types render as a stable bracketed placeholder. Top-level +// inbound images and image spans in inbound posts are downloaded later by +// InboundMediaService; quoted/forwarded media and the other media kinds remain +// placeholders. merge_forward is intercepted by the enricher before it reaches +// here because expansion needs an HTTP round-trip; the inline placeholder is +// only a fallback for a forward nested inside another forward. func flattenContent(msgType, rawContent string) string { switch msgType { case "text": @@ -80,6 +79,7 @@ type larkPostSpan struct { Href string `json:"href"` UserID string `json:"user_id"` UserName string `json:"user_name"` + ImageKey string `json:"image_key"` } // flattenPostContent flattens a received `post` body.content into plain diff --git a/server/internal/integrations/lark/feishu_resolvers.go b/server/internal/integrations/lark/feishu_resolvers.go index 297fe26425c..b9262a3173e 100644 --- a/server/internal/integrations/lark/feishu_resolvers.go +++ b/server/internal/integrations/lark/feishu_resolvers.go @@ -44,12 +44,12 @@ func larkMsgFromRaw(msg channel.InboundMessage) (InboundMessage, error) { // shared session service, audit logger, and (optional) outbound replier + // typing indicator. Feishu is just another consumer of the channel-agnostic // engine.ChatSession — there is no Feishu-specific session implementation. -func NewFeishuResolverSet(store *ChannelStore, session *engine.ChatSession, audit AuditLogger, replier OutcomeReplier, typing *TypingIndicatorManager) engine.ResolverSet { +func NewFeishuResolverSet(store *ChannelStore, session *engine.ChatSession, audit AuditLogger, replier OutcomeReplier, typing *TypingIndicatorManager, media *InboundMediaService) engine.ResolverSet { set := engine.ResolverSet{ Installation: &feishuInstallationResolver{store: store}, Identity: &feishuIdentityResolver{store: store}, Dedup: &feishuDeduper{store: store}, - Session: &feishuSessionBinder{session: session}, + Session: &feishuSessionBinder{session: session, media: media}, Audit: &feishuAuditor{audit: audit}, OriginType: originFeishuChat, } @@ -159,7 +159,10 @@ type chatSession interface { AppendUserMessage(ctx context.Context, in engine.AppendInput) (engine.AppendResult, error) } -type feishuSessionBinder struct{ session chatSession } +type feishuSessionBinder struct { + session chatSession + media *InboundMediaService +} // larkBindingConfig is the opaque outbound routing persisted on the chat // binding's config when the binding key is a composite (Lark topic): the real @@ -206,15 +209,36 @@ func (r *feishuSessionBinder) AppendMessage(ctx context.Context, p engine.Append if err != nil { return engine.AppendResult{}, err } + mediaRefs := p.Message.MediaRefs + if r.media != nil && (lm.MessageType == "image" || lm.MessageType == "post") { + inst, ok := p.Installation.Platform.(Installation) + if !ok { + return engine.AppendResult{}, errors.New("lark: resolved installation platform has unexpected type") + } + refs, err := r.media.IngestMessageImages( + ctx, + inst, + p.Installation.WorkspaceID, + p.Message.MessageID, + lm.MessageType, + lm.RawContent, + ) + if err != nil { + return engine.AppendResult{}, err + } + mediaRefs = append(mediaRefs, refs...) + } return r.session.AppendUserMessage(ctx, engine.AppendInput{ SessionID: p.SessionID, Sender: p.Sender, + WorkspaceID: p.Installation.WorkspaceID, InstallationID: p.InstallationID, Body: p.Message.Text, CommandText: lm.CommandBody, MessageID: p.Message.MessageID, ThreadID: p.Message.Source.ThreadID, ClaimToken: p.ClaimToken, + MediaRefs: mediaRefs, }) } diff --git a/server/internal/integrations/lark/feishu_resolvers_test.go b/server/internal/integrations/lark/feishu_resolvers_test.go index 9b376a9cc0b..fcf3043119c 100644 --- a/server/internal/integrations/lark/feishu_resolvers_test.go +++ b/server/internal/integrations/lark/feishu_resolvers_test.go @@ -150,3 +150,100 @@ func TestFeishuSessionBinder_AppendUsesUnenrichedCommandBody(t *testing.T) { t.Errorf("append mapping wrong: %+v", got) } } + +func TestFeishuSessionBinder_IngestsImageBeforeAppend(t *testing.T) { + f := &fakeChatSession{} + downloader := &fakeMessageResourceDownloader{resource: MessageResource{ + Data: []byte("\x89PNG\r\n\x1a\nimage-body"), + ContentType: "image/png", + }} + media, err := NewInboundMediaService( + downloader, + fakeInboundMediaCredentials{}, + &fakeInboundMediaStorage{}, + InboundMediaServiceConfig{}, + ) + if err != nil { + t.Fatalf("NewInboundMediaService: %v", err) + } + b := &feishuSessionBinder{session: f, media: media} + inst := Installation{ID: binderUUID(2), WorkspaceID: binderUUID(3), AppID: "cli_1", Region: "feishu"} + raw, _ := json.Marshal(InboundMessage{ + MessageType: "image", + RawContent: `{"image_key":"img_1"}`, + }) + if _, err := b.AppendMessage(context.Background(), engine.AppendParams{ + SessionID: binderUUID(1), + Sender: binderUUID(7), + InstallationID: inst.ID, + Installation: engine.ResolvedInstallation{ + ID: inst.ID, + WorkspaceID: inst.WorkspaceID, + Platform: inst, + }, + Message: channel.InboundMessage{ + MessageID: "om_1", + Type: channel.MsgTypeImage, + Text: "[Image]", + Raw: raw, + }, + }); err != nil { + t.Fatalf("AppendMessage: %v", err) + } + if len(f.appendIn.MediaRefs) != 1 { + t.Fatalf("MediaRefs = %v, want one image", f.appendIn.MediaRefs) + } + if f.appendIn.WorkspaceID != binderUUID(3) || f.appendIn.MediaRefs[0].MimeType != "image/png" { + t.Errorf("append media mapping wrong: %+v", f.appendIn) + } +} + +func TestFeishuSessionBinder_IngestsPostImageBeforeAppend(t *testing.T) { + f := &fakeChatSession{} + downloader := &fakeMessageResourceDownloader{resource: MessageResource{ + Data: []byte("\x89PNG\r\n\x1a\nimage-body"), + ContentType: "image/png", + }} + media, err := NewInboundMediaService( + downloader, + fakeInboundMediaCredentials{}, + &fakeInboundMediaStorage{}, + InboundMediaServiceConfig{}, + ) + if err != nil { + t.Fatalf("NewInboundMediaService: %v", err) + } + b := &feishuSessionBinder{session: f, media: media} + inst := Installation{ID: binderUUID(2), WorkspaceID: binderUUID(3), AppID: "cli_1", Region: "feishu"} + raw, _ := json.Marshal(InboundMessage{ + MessageType: "post", + RawContent: `{"title":"","content":[[{"tag":"img","image_key":"img_post_1"},{"tag":"text","text":"图里有什么内容"}]]}`, + }) + if _, err := b.AppendMessage(context.Background(), engine.AppendParams{ + SessionID: binderUUID(1), + Sender: binderUUID(7), + InstallationID: inst.ID, + Installation: engine.ResolvedInstallation{ + ID: inst.ID, + WorkspaceID: inst.WorkspaceID, + Platform: inst, + }, + Message: channel.InboundMessage{ + MessageID: "om_post_1", + Type: channel.MsgTypeText, + Text: "[Image] 图里有什么内容", + Raw: raw, + }, + }); err != nil { + t.Fatalf("AppendMessage: %v", err) + } + if !downloader.called || downloader.msgID != "om_post_1" || downloader.key != "img_post_1" { + t.Errorf("download call = %+v", downloader) + } + if len(f.appendIn.MediaRefs) != 1 { + t.Fatalf("MediaRefs = %v, want one image", f.appendIn.MediaRefs) + } + if f.appendIn.MediaRefs[0].MimeType != "image/png" { + t.Errorf("append media mapping wrong: %+v", f.appendIn) + } +} diff --git a/server/internal/integrations/lark/feishu_types.go b/server/internal/integrations/lark/feishu_types.go index 179733420d2..d30c23ee7f7 100644 --- a/server/internal/integrations/lark/feishu_types.go +++ b/server/internal/integrations/lark/feishu_types.go @@ -34,6 +34,12 @@ type InboundMessage struct { // expand while the core stays msg_type-agnostic and only reads Body. MessageType string + // RawContent is Lark's JSON-encoded message.content string. Text is + // flattened into Body, while media consumers parse this field to obtain + // resource keys such as image_key without leaking platform-specific + // fields into channel.InboundMessage. + RawContent string + // CreateTime is the trigger message's creation time (epoch milliseconds). // The enricher anchors the group recent-context window to it; the typing // indicator uses it to skip stale reactions. diff --git a/server/internal/integrations/lark/http_client.go b/server/internal/integrations/lark/http_client.go index cd374ad227c..793cbbb83c5 100644 --- a/server/internal/integrations/lark/http_client.go +++ b/server/internal/integrations/lark/http_client.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "log/slog" + "mime" "net/http" "net/url" "strconv" @@ -596,6 +597,98 @@ func (c *httpAPIClient) GetMessage(ctx context.Context, creds InstallationCreden // silently gets the max rather than a 400 from Lark. const larkListMessagesMaxPageSize = 50 +// larkMessageResourceMaxBytes matches Lark's documented maximum for +// GET /im/v1/messages/{message_id}/resources/{file_key} and Multica's own +// attachment upload limit. The extra byte read in DownloadMessageResource +// detects an oversized or misbehaving upstream response without buffering it +// unboundedly. +const larkMessageResourceMaxBytes = 100 << 20 + +// DownloadMessageResource fetches a binary image/file attached to a message. +// Lark requires the message_id and resource key to match and the `type` query +// parameter to be either image or file. +func (c *httpAPIClient) DownloadMessageResource(ctx context.Context, creds InstallationCredentials, messageID, fileKey, resourceType string) (MessageResource, error) { + if messageID == "" { + return MessageResource{}, errors.New("lark http client: missing message_id") + } + if fileKey == "" { + return MessageResource{}, errors.New("lark http client: missing file_key") + } + if resourceType != "image" && resourceType != "file" { + return MessageResource{}, fmt.Errorf("lark http client: unsupported message resource type %q", resourceType) + } + token, err := c.tenantAccessToken(ctx, creds) + if err != nil { + return MessageResource{}, err + } + q := url.Values{} + q.Set("type", resourceType) + resourcePath := "/open-apis/im/v1/messages/" + url.PathEscape(messageID) + + "/resources/" + url.PathEscape(fileKey) + "?" + q.Encode() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.resolveBaseURL(creds)+resourcePath, nil) + if err != nil { + return MessageResource{}, fmt.Errorf("lark http client: download message resource: new request: %w", err) + } + req.Header.Set("Authorization", "Bearer "+token) + resp, err := c.cfg.HTTPClient.Do(req) + if err != nil { + return MessageResource{}, fmt.Errorf("lark http client: download message resource: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + raw, readErr := io.ReadAll(io.LimitReader(resp.Body, 513)) + if readErr != nil { + return MessageResource{}, fmt.Errorf("lark http client: download message resource: http %d", resp.StatusCode) + } + var apiErr struct { + Code int `json:"code"` + Msg string `json:"msg"` + } + if json.Unmarshal(raw, &apiErr) == nil && apiErr.Code != 0 { + if isTokenError(apiErr.Code) { + c.invalidateToken(creds.AppID) + } + return MessageResource{}, &APIError{Op: "download message resource", Code: apiErr.Code, Msg: apiErr.Msg} + } + return MessageResource{}, fmt.Errorf("lark http client: download message resource: http %d: %s", + resp.StatusCode, truncate(string(raw), 512)) + } + + data, err := io.ReadAll(io.LimitReader(resp.Body, larkMessageResourceMaxBytes+1)) + if err != nil { + return MessageResource{}, fmt.Errorf("lark http client: download message resource: read body: %w", err) + } + if len(data) > larkMessageResourceMaxBytes { + return MessageResource{}, fmt.Errorf("lark http client: download message resource exceeds %d bytes", larkMessageResourceMaxBytes) + } + + contentType := strings.TrimSpace(resp.Header.Get("Content-Type")) + if mediaType, _, parseErr := mime.ParseMediaType(contentType); parseErr == nil { + contentType = mediaType + } + if contentType == "application/json" { + var apiErr struct { + Code int `json:"code"` + Msg string `json:"msg"` + } + if json.Unmarshal(data, &apiErr) == nil && apiErr.Code != 0 { + if isTokenError(apiErr.Code) { + c.invalidateToken(creds.AppID) + } + return MessageResource{}, &APIError{Op: "download message resource", Code: apiErr.Code, Msg: apiErr.Msg} + } + } + + filename := "" + if disposition := resp.Header.Get("Content-Disposition"); disposition != "" { + if _, params, parseErr := mime.ParseMediaType(disposition); parseErr == nil { + filename = params["filename"] + } + } + return MessageResource{Data: data, Filename: filename, ContentType: contentType}, nil +} + // ListChatMessages retrieves a bounded, recent window of messages via // GET /open-apis/im/v1/messages. Where GetMessage fetches a single message // by id, this lists a conversation; it backs the enricher's group-context diff --git a/server/internal/integrations/lark/http_client_resource_test.go b/server/internal/integrations/lark/http_client_resource_test.go new file mode 100644 index 00000000000..508f377b21b --- /dev/null +++ b/server/internal/integrations/lark/http_client_resource_test.go @@ -0,0 +1,52 @@ +package lark + +import ( + "bytes" + "context" + "net/http" + "testing" +) + +func TestHTTPAPIClient_DownloadMessageResource(t *testing.T) { + f := newLarkFake(t) + f.stubToken("token-resource", 3600) + wantData := []byte("\x89PNG\r\n\x1a\nimage-body") + f.mux.HandleFunc("/open-apis/im/v1/messages/om_1/resources/img_1", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + t.Errorf("method = %s, want GET", r.Method) + } + if got := r.URL.Query().Get("type"); got != "image" { + t.Errorf("type query = %q, want image", got) + } + if got := r.Header.Get("Authorization"); got != "Bearer token-resource" { + t.Errorf("authorization = %q", got) + } + w.Header().Set("Content-Type", "image/png") + w.Header().Set("Content-Disposition", `attachment; filename="screen.png"`) + _, _ = w.Write(wantData) + }) + + client := NewHTTPAPIClient(HTTPClientConfig{BaseURL: f.URL()}) + downloader, ok := client.(MessageResourceDownloader) + if !ok { + t.Fatalf("client %T does not implement MessageResourceDownloader", client) + } + got, err := downloader.DownloadMessageResource(context.Background(), testCreds(), "om_1", "img_1", "image") + if err != nil { + t.Fatalf("DownloadMessageResource: %v", err) + } + if !bytes.Equal(got.Data, wantData) { + t.Errorf("data = %q, want %q", got.Data, wantData) + } + if got.ContentType != "image/png" || got.Filename != "screen.png" { + t.Errorf("metadata = %+v", got) + } +} + +func TestHTTPAPIClient_DownloadMessageResourceRejectsUnsupportedType(t *testing.T) { + client := NewHTTPAPIClient(HTTPClientConfig{}) + downloader := client.(MessageResourceDownloader) + if _, err := downloader.DownloadMessageResource(context.Background(), testCreds(), "om_1", "img_1", "sticker"); err == nil { + t.Fatal("expected unsupported resource type error") + } +} diff --git a/server/internal/integrations/lark/inbound_media.go b/server/internal/integrations/lark/inbound_media.go new file mode 100644 index 00000000000..8a98fbe8945 --- /dev/null +++ b/server/internal/integrations/lark/inbound_media.go @@ -0,0 +1,287 @@ +package lark + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "log/slog" + "mime" + "net/http" + "path" + "strings" + "time" + + "github.com/jackc/pgx/v5/pgtype" + + "github.com/multica-ai/multica/server/internal/integrations/channel" + "github.com/multica-ai/multica/server/internal/storage" +) + +const defaultInboundMediaTimeout = 2 * time.Second + +// InboundMediaService downloads a user-sent Feishu resource and persists it +// through Multica's configured object storage. Database attachment rows are +// created later, inside the chat-message transaction, from the returned +// channel.MediaRef. +type InboundMediaService struct { + downloader MessageResourceDownloader + credentials CredentialsResolver + storage storage.Storage + timeout time.Duration + logger *slog.Logger +} + +type InboundMediaServiceConfig struct { + Timeout time.Duration + Logger *slog.Logger +} + +func NewInboundMediaService(downloader MessageResourceDownloader, credentials CredentialsResolver, objectStorage storage.Storage, cfg InboundMediaServiceConfig) (*InboundMediaService, error) { + if downloader == nil { + return nil, errors.New("lark inbound media: downloader is required") + } + if credentials == nil { + return nil, errors.New("lark inbound media: credentials resolver is required") + } + if objectStorage == nil { + return nil, errors.New("lark inbound media: storage is required") + } + if cfg.Timeout <= 0 { + cfg.Timeout = defaultInboundMediaTimeout + } + if cfg.Logger == nil { + cfg.Logger = slog.Default() + } + return &InboundMediaService{ + downloader: downloader, + credentials: credentials, + storage: objectStorage, + timeout: cfg.Timeout, + logger: cfg.Logger, + }, nil +} + +// IngestImage ingests the single image carried by a top-level image message. +// It remains as the narrow helper used by callers and tests that already know +// the message kind. +func (s *InboundMediaService) IngestImage(ctx context.Context, inst Installation, workspaceID pgtype.UUID, messageID, rawContent string) (channel.MediaRef, error) { + refs, err := s.IngestMessageImages(ctx, inst, workspaceID, messageID, "image", rawContent) + if err != nil { + return channel.MediaRef{}, err + } + if len(refs) != 1 { + return channel.MediaRef{}, fmt.Errorf("lark inbound media: expected one image, got %d", len(refs)) + } + return refs[0], nil +} + +// IngestMessageImages resolves image keys from either a top-level image or +// image spans embedded in a rich-text post, downloads each resource, and +// uploads it under a deterministic object key. Redelivery overwrites the same +// objects rather than leaking a new orphan for every retry. +func (s *InboundMediaService) IngestMessageImages(ctx context.Context, inst Installation, workspaceID pgtype.UUID, messageID, messageType, rawContent string) ([]channel.MediaRef, error) { + if !workspaceID.Valid { + return nil, errors.New("lark inbound media: workspace_id is required") + } + if messageID == "" { + return nil, errors.New("lark inbound media: message_id is required") + } + + imageKeys, err := inboundImageKeys(messageType, rawContent) + if err != nil { + return nil, err + } + if len(imageKeys) == 0 { + return nil, nil + } + + secret, err := s.credentials.DecryptAppSecret(inst) + if err != nil { + return nil, fmt.Errorf("lark inbound media: decrypt app_secret: %w", err) + } + creds := InstallationCredentials{ + AppID: inst.AppID, + AppSecret: secret, + Region: RegionOrDefault(inst.Region), + } + if inst.TenantKey.Valid { + creds.TenantKey = inst.TenantKey.String + } + + refs := make([]channel.MediaRef, 0, len(imageKeys)) + for i, imageKey := range imageKeys { + ref, err := s.ingestImageKey(ctx, inst, workspaceID, creds, messageID, imageKey, i, len(imageKeys)) + if err != nil { + return nil, err + } + refs = append(refs, ref) + } + return refs, nil +} + +func inboundImageKeys(messageType, rawContent string) ([]string, error) { + switch messageType { + case "image": + var content struct { + ImageKey string `json:"image_key"` + } + if err := json.Unmarshal([]byte(rawContent), &content); err != nil { + return nil, fmt.Errorf("lark inbound media: decode image content: %w", err) + } + if content.ImageKey == "" { + return nil, errors.New("lark inbound media: image_key is required") + } + return []string{content.ImageKey}, nil + case "post": + var doc larkPostContent + if err := json.Unmarshal([]byte(rawContent), &doc); err != nil { + return nil, fmt.Errorf("lark inbound media: decode post content: %w", err) + } + seen := make(map[string]struct{}) + keys := make([]string, 0) + for _, paragraph := range doc.Content { + for _, span := range paragraph { + if span.Tag != "img" || span.ImageKey == "" { + continue + } + if _, ok := seen[span.ImageKey]; ok { + continue + } + seen[span.ImageKey] = struct{}{} + keys = append(keys, span.ImageKey) + } + } + return keys, nil + default: + return nil, nil + } +} + +func (s *InboundMediaService) ingestImageKey(ctx context.Context, inst Installation, workspaceID pgtype.UUID, creds InstallationCredentials, messageID, imageKey string, index, total int) (channel.MediaRef, error) { + mediaCtx, cancel := context.WithTimeout(ctx, s.timeout) + defer cancel() + resource, err := s.downloader.DownloadMessageResource(mediaCtx, creds, messageID, imageKey, "image") + if err != nil { + return channel.MediaRef{}, fmt.Errorf("lark inbound media: download image: %w", err) + } + if len(resource.Data) == 0 { + return channel.MediaRef{}, errors.New("lark inbound media: downloaded image is empty") + } + + contentType, err := normalizeInboundImageContentType(resource.ContentType, resource.Data) + if err != nil { + return channel.MediaRef{}, err + } + ext := inboundImageExtension(contentType) + filename := sanitizeInboundFilename(resource.Filename) + if filename == "" { + filename = "feishu-image" + if total > 1 { + filename += fmt.Sprintf("-%d", index+1) + } + filename += ext + } else if path.Ext(filename) == "" && ext != "" { + filename += ext + } + + sum := sha256.Sum256([]byte(uuidString(inst.ID) + "\x00" + messageID + "\x00" + imageKey)) + key := "workspaces/" + uuidString(workspaceID) + "/channels/feishu/" + + hex.EncodeToString(sum[:16]) + ext + objectURL, err := s.storage.Upload(mediaCtx, key, resource.Data, contentType, filename) + if err != nil { + return channel.MediaRef{}, fmt.Errorf("lark inbound media: store image: %w", err) + } + s.logger.Debug("lark inbound media: image persisted", + "message_id", messageID, + "workspace_id", uuidString(workspaceID), + "size_bytes", len(resource.Data), + "content_type", contentType, + ) + return channel.MediaRef{ + Type: channel.MsgTypeImage, + StorageKey: key, + URL: objectURL, + Filename: filename, + MimeType: contentType, + SizeBytes: int64(len(resource.Data)), + }, nil +} + +func normalizeInboundImageContentType(headerValue string, data []byte) (string, error) { + contentType := strings.TrimSpace(headerValue) + if mediaType, _, err := mime.ParseMediaType(contentType); err == nil { + contentType = mediaType + } + contentType = strings.ToLower(contentType) + + detected := strings.ToLower(http.DetectContentType(data)) + if mediaType, _, err := mime.ParseMediaType(detected); err == nil { + detected = mediaType + } + if isAllowedInboundImageContentType(detected) { + return detected, nil + } + if detected != "application/octet-stream" { + return "", fmt.Errorf("lark inbound media: downloaded resource is not an image: %s", detected) + } + if isAllowedInboundImageContentType(contentType) { + return contentType, nil + } + return "", fmt.Errorf("lark inbound media: unsupported image content type %q", contentType) +} + +func isAllowedInboundImageContentType(contentType string) bool { + switch contentType { + case "image/jpeg", + "image/png", + "image/gif", + "image/webp", + "image/bmp", + "image/tiff", + "image/x-icon", + "image/vnd.microsoft.icon": + return true + default: + return false + } +} + +func inboundImageExtension(contentType string) string { + switch contentType { + case "image/jpeg": + return ".jpg" + case "image/png": + return ".png" + case "image/gif": + return ".gif" + case "image/webp": + return ".webp" + case "image/tiff": + return ".tiff" + case "image/bmp": + return ".bmp" + case "image/x-icon", "image/vnd.microsoft.icon": + return ".ico" + } + exts, err := mime.ExtensionsByType(contentType) + if err != nil || len(exts) == 0 { + return "" + } + ext := strings.ToLower(exts[0]) + if !strings.HasPrefix(ext, ".") || len(ext) > 10 { + return "" + } + return ext +} + +func sanitizeInboundFilename(filename string) string { + filename = strings.TrimSpace(strings.ReplaceAll(filename, "\\", "/")) + filename = path.Base(filename) + if filename == "." || filename == "/" || filename == "" { + return "" + } + return filename +} diff --git a/server/internal/integrations/lark/inbound_media_test.go b/server/internal/integrations/lark/inbound_media_test.go new file mode 100644 index 00000000000..44047302ce4 --- /dev/null +++ b/server/internal/integrations/lark/inbound_media_test.go @@ -0,0 +1,149 @@ +package lark + +import ( + "bytes" + "context" + "testing" + + "github.com/multica-ai/multica/server/internal/storage" +) + +type fakeMessageResourceDownloader struct { + resource MessageResource + called bool + msgID string + key string + kind string +} + +func (f *fakeMessageResourceDownloader) DownloadMessageResource(_ context.Context, _ InstallationCredentials, messageID, fileKey, resourceType string) (MessageResource, error) { + f.called = true + f.msgID = messageID + f.key = fileKey + f.kind = resourceType + return f.resource, nil +} + +type fakeInboundMediaCredentials struct{} + +func (fakeInboundMediaCredentials) DecryptAppSecret(Installation) (string, error) { + return "secret", nil +} + +type fakeInboundMediaStorage struct { + storage.Storage + key string + data []byte + contentType string + filename string +} + +func (f *fakeInboundMediaStorage) Upload(_ context.Context, key string, data []byte, contentType, filename string) (string, error) { + f.key = key + f.data = append([]byte(nil), data...) + f.contentType = contentType + f.filename = filename + return "https://storage.example/" + key, nil +} + +func TestInboundMediaService_IngestImage(t *testing.T) { + png := []byte("\x89PNG\r\n\x1a\nimage-body") + downloader := &fakeMessageResourceDownloader{resource: MessageResource{ + Data: png, + ContentType: "application/octet-stream", + }} + objectStorage := &fakeInboundMediaStorage{} + service, err := NewInboundMediaService( + downloader, + fakeInboundMediaCredentials{}, + objectStorage, + InboundMediaServiceConfig{}, + ) + if err != nil { + t.Fatalf("NewInboundMediaService: %v", err) + } + inst := Installation{ID: binderUUID(1), AppID: "cli_1", Region: "feishu"} + ref, err := service.IngestImage(context.Background(), inst, binderUUID(2), "om_1", `{"image_key":"img_1"}`) + if err != nil { + t.Fatalf("IngestImage: %v", err) + } + if !downloader.called || downloader.msgID != "om_1" || downloader.key != "img_1" || downloader.kind != "image" { + t.Errorf("download call = %+v", downloader) + } + if !bytes.Equal(objectStorage.data, png) { + t.Errorf("uploaded data = %q", objectStorage.data) + } + if objectStorage.contentType != "image/png" || objectStorage.filename != "feishu-image.png" { + t.Errorf("upload metadata = content_type %q filename %q", objectStorage.contentType, objectStorage.filename) + } + if ref.MimeType != "image/png" || ref.Filename != "feishu-image.png" || + ref.SizeBytes != int64(len(png)) || ref.URL == "" || ref.StorageKey != objectStorage.key { + t.Errorf("media ref = %+v", ref) + } +} + +func TestInboundImageKeys_PostExtractsAndDeduplicatesImages(t *testing.T) { + keys, err := inboundImageKeys("post", `{ + "title": "", + "content": [ + [ + {"tag":"img","image_key":"img_1"}, + {"tag":"text","text":"图里有什么内容"} + ], + [ + {"tag":"img","image_key":"img_2"}, + {"tag":"img","image_key":"img_1"} + ] + ] + }`) + if err != nil { + t.Fatalf("inboundImageKeys: %v", err) + } + if len(keys) != 2 || keys[0] != "img_1" || keys[1] != "img_2" { + t.Fatalf("keys = %v, want [img_1 img_2]", keys) + } +} + +func TestInboundMediaService_RejectsMissingImageKey(t *testing.T) { + downloader := &fakeMessageResourceDownloader{} + service, err := NewInboundMediaService( + downloader, + fakeInboundMediaCredentials{}, + &fakeInboundMediaStorage{}, + InboundMediaServiceConfig{}, + ) + if err != nil { + t.Fatalf("NewInboundMediaService: %v", err) + } + if _, err := service.IngestImage(context.Background(), Installation{}, binderUUID(2), "om_1", `{}`); err == nil { + t.Fatal("expected missing image_key error") + } + if downloader.called { + t.Fatal("downloader must not run for malformed image content") + } +} + +func TestInboundMediaService_RejectsSpoofedImageContentType(t *testing.T) { + downloader := &fakeMessageResourceDownloader{resource: MessageResource{ + Data: []byte(""), + ContentType: "image/png", + }} + objectStorage := &fakeInboundMediaStorage{} + service, err := NewInboundMediaService( + downloader, + fakeInboundMediaCredentials{}, + objectStorage, + InboundMediaServiceConfig{}, + ) + if err != nil { + t.Fatalf("NewInboundMediaService: %v", err) + } + inst := Installation{ID: binderUUID(1), AppID: "cli_1", Region: "feishu"} + _, err = service.IngestImage(context.Background(), inst, binderUUID(2), "om_1", `{"image_key":"img_1"}`) + if err == nil { + t.Fatal("expected spoofed image content type error") + } + if objectStorage.key != "" { + t.Fatal("spoofed image must not be uploaded") + } +} diff --git a/server/internal/integrations/lark/ws_frame_decoder.go b/server/internal/integrations/lark/ws_frame_decoder.go index f09dc2b3417..d852ad9b04b 100644 --- a/server/internal/integrations/lark/ws_frame_decoder.go +++ b/server/internal/integrations/lark/ws_frame_decoder.go @@ -73,6 +73,7 @@ func (d *LarkJSONFrameDecoder) Decode(payload []byte, inst Installation) (Inboun MessageID: evt.Message.MessageID, SenderOpenID: OpenID(evt.Sender.SenderID.OpenID), MessageType: evt.Message.MessageType, + RawContent: evt.Message.Content, CreateTime: evt.Message.CreateTime, // parent_id / root_id are populated by Lark only in reply // scenarios. The enricher keys quoted-reply expansion off @@ -91,14 +92,13 @@ func (d *LarkJSONFrameDecoder) Decode(payload []byte, inst Installation) (Inboun botUnionID = inst.BotUnionID.String } - // text + post are flattened synchronously here (no external calls — - // the decoder must stay fast and dependency-free). merge_forward - // leaves Body empty: it needs an HTTP round-trip to expand and is - // handled downstream by the enricher, which keys off MessageType. - // Other types (image, file, …) also leave Body empty in this MVP; - // attachment ingestion is a separate issue. + // Text, post, and image are flattened synchronously here (no external + // calls — the decoder must stay fast and dependency-free). Image keeps a + // visible placeholder while its binary resource is downloaded later by + // the Feishu session binder. merge_forward leaves Body empty because it + // needs an HTTP round-trip to expand and is handled by the enricher. switch evt.Message.MessageType { - case "text", "post": + case "text", "post", "image": msg.Body = resolveMentions(flattenContent(evt.Message.MessageType, evt.Message.Content), evt.Message.Mentions, inst.BotOpenID, botUnionID) } diff --git a/server/internal/integrations/lark/ws_frame_decoder_test.go b/server/internal/integrations/lark/ws_frame_decoder_test.go index dcc51e54aae..d661f040865 100644 --- a/server/internal/integrations/lark/ws_frame_decoder_test.go +++ b/server/internal/integrations/lark/ws_frame_decoder_test.go @@ -429,7 +429,7 @@ func TestLarkJSONFrameDecoderMessageContentEmptyOnInvalidContentJSON(t *testing. } } -func TestLarkJSONFrameDecoderNonTextMessageHasEmptyBody(t *testing.T) { +func TestLarkJSONFrameDecoderImageMessageKeepsResourceAndPlaceholder(t *testing.T) { t.Parallel() raw := []byte(`{ "type":"event_callback", @@ -443,12 +443,15 @@ func TestLarkJSONFrameDecoderNonTextMessageHasEmptyBody(t *testing.T) { if err != nil || !ok { t.Fatalf("ok=%v err=%v", ok, err) } - if msg.Body != "" { - t.Errorf("Body = %q; non-text messages should have empty body in MVP", msg.Body) + if msg.Body != "[Image]" { + t.Errorf("Body = %q; image messages should keep a visible placeholder", msg.Body) } if msg.MessageID == "" { t.Error("MessageID should still be populated for non-text events") } + if msg.RawContent != `{"image_key":"img1"}` { + t.Errorf("RawContent = %q; want the image_key envelope", msg.RawContent) + } } // TestLarkJSONFrameDecoderPostMessageFlattened verifies that a rich-text diff --git a/server/internal/integrations/slack/resolvers.go b/server/internal/integrations/slack/resolvers.go index fe3f95ec04a..c23b6a96c1f 100644 --- a/server/internal/integrations/slack/resolvers.go +++ b/server/internal/integrations/slack/resolvers.go @@ -337,6 +337,7 @@ func (r *sessionBinder) AppendMessage(ctx context.Context, p engine.AppendParams return r.session.AppendUserMessage(ctx, engine.AppendInput{ SessionID: p.SessionID, Sender: p.Sender, + WorkspaceID: p.Installation.WorkspaceID, InstallationID: p.InstallationID, Body: p.Message.Text, // Slack text is not enriched, so the command source is the body itself.