fix(lark): ingest inbound images and videos as chat attachments (MUL-4934)

server/cmd/server/router.go

@@ -391,8 +391,9 @@ func NewRouterWithOptions(pool *pgxpool.Pool, hub *realtime.Hub, bus *events.Bus
 					Credentials: installSvc,
 					Logger:      slog.Default(),
 				})
+				mediaResolver := lark.NewFeishuMediaResolver(larkClient, installSvc, store, slog.Default())
 				channelRouter.Register(channel.TypeFeishu, lark.NewFeishuResolverSet(
-					cs, feishuSession, auditLogger, resolverReplier, typingIndicator,
+					cs, feishuSession, auditLogger, resolverReplier, typingIndicator, mediaResolver,
 				))
 				slog.Info("lark inbound pipeline wired", "connector", connectorLabel)
 

server/internal/integrations/channel/engine/resolvers.go

@@ -93,6 +93,7 @@ type AppendParams struct {
 	SessionID      pgtype.UUID
 	Sender         pgtype.UUID
 	InstallationID pgtype.UUID
+	WorkspaceID    pgtype.UUID
 	Message        channel.InboundMessage
 	ClaimToken     pgtype.UUID
 }
@@ -163,6 +164,14 @@ type SessionBinder interface {
 	AppendMessage(ctx context.Context, p AppendParams) (AppendResult, error)
 }
 
+// MediaResolver resolves platform media after a message has passed routing,
+// group-addressing, identity, and session checks, but before AppendMessage
+// persists the user message. Implementations are best-effort: failures should
+// leave the message text intact and omit the failed MediaRefs.
+type MediaResolver interface {
+	ResolveMedia(ctx context.Context, inst ResolvedInstallation, sender ResolvedIdentity, sessionID pgtype.UUID, msg channel.InboundMessage) channel.InboundMessage
+}
+
 // Auditor records a dropped inbound event (no message body — drop-audit
 // policy). instID may be the zero UUID for installation-less events.
 type Auditor interface {
@@ -200,6 +209,7 @@ type ResolverSet struct {
 	Identity     IdentityResolver
 	Dedup        Deduper
 	Session      SessionBinder
+	Media        MediaResolver
 	Audit        Auditor
 	Replier      OutboundReplier
 	Typing       TypingNotifier

server/internal/integrations/channel/engine/router.go

@@ -39,6 +39,7 @@ type Router struct {
 	batcher *pendingBatcher
 
 	replyTimeout time.Duration
+	mediaTimeout time.Duration
 	replyWg      sync.WaitGroup
 
 	logger *slog.Logger
@@ -53,6 +54,10 @@ type RouterConfig struct {
 	// call. It runs off the connector ACK path, so it must stay strictly
 	// under the platform ACK deadline (Lark: 3s). Defaults to 2.5s.
 	ReplyTimeout time.Duration
+	// MediaTimeout caps best-effort media resolution on the connector ACK
+	// path. On timeout, the original message is appended without MediaRefs.
+	// Defaults to 2.5s.
+	MediaTimeout time.Duration
 	Logger       *slog.Logger
 }
 
@@ -64,6 +69,9 @@ func NewRouter(issues IssueCreator, tasks TaskEnqueuer, reader SessionReader, cf
 	if cfg.ReplyTimeout == 0 {
 		cfg.ReplyTimeout = 2500 * time.Millisecond
 	}
+	if cfg.MediaTimeout == 0 {
+		cfg.MediaTimeout = 2500 * time.Millisecond
+	}
 	if cfg.Logger == nil {
 		cfg.Logger = slog.Default()
 	}
@@ -73,6 +81,7 @@ func NewRouter(issues IssueCreator, tasks TaskEnqueuer, reader SessionReader, cf
 		tasks:        tasks,
 		reader:       reader,
 		replyTimeout: cfg.ReplyTimeout,
+		mediaTimeout: cfg.MediaTimeout,
 		logger:       cfg.Logger,
 		pendingFresh: make(map[string]bool),
 	}
@@ -258,12 +267,16 @@ func (r *Router) processClaimed(ctx context.Context, set ResolverSet, msg channe
 		// Single tx; an error rolled it back, nothing landed. Release.
 		return Result{}, finalizeRelease, fmt.Errorf("ensure chat session: %w", err)
 	}
+	if set.Media != nil {
+		msg = r.resolveMedia(ctx, set, inst, identity, sessionID, msg)
+	}
 
 	// 6. Append message + in-tx dedup Mark — the durable transition point.
 	appendRes, err := set.Session.AppendMessage(ctx, AppendParams{
 		SessionID:      sessionID,
 		Sender:         identity.UserID,
 		InstallationID: inst.ID,
+		WorkspaceID:    inst.WorkspaceID,
 		Message:        msg,
 		ClaimToken:     claimToken,
 	})
@@ -315,6 +328,29 @@ func (r *Router) processClaimed(ctx context.Context, set ResolverSet, msg channe
 	return res, postAppendFinalize, nil
 }
 
+func (r *Router) resolveMedia(ctx context.Context, set ResolverSet, inst ResolvedInstallation, identity ResolvedIdentity, sessionID pgtype.UUID, msg channel.InboundMessage) channel.InboundMessage {
+	if r.mediaTimeout <= 0 {
+		return set.Media.ResolveMedia(ctx, inst, identity, sessionID, msg)
+	}
+	mctx, cancel := context.WithTimeout(ctx, r.mediaTimeout)
+	defer cancel()
+	done := make(chan channel.InboundMessage, 1)
+	go func() {
+		done <- set.Media.ResolveMedia(mctx, inst, identity, sessionID, msg)
+	}()
+	select {
+	case resolved := <-done:
+		return resolved
+	case <-mctx.Done():
+		r.logger.Warn("channel router: media resolution timed out",
+			"channel_type", string(msg.Source.ChannelType),
+			"event_id", msg.EventID,
+			"message_id", msg.MessageID,
+			"timeout", r.mediaTimeout.String())
+		return msg
+	}
+}
+
 // scheduleRun hands the per-session run trigger to the debouncer (or fires it
 // inline when batching is disabled).
 func (r *Router) scheduleRun(set ResolverSet, inst ResolvedInstallation, msg channel.InboundMessage, sessionID, initiatorUserID pgtype.UUID) {

server/internal/integrations/channel/engine/router_test.go

@@ -140,6 +140,38 @@ func (f *fakeTyping) OnSettled(_ context.Context, _ pgtype.UUID) {
 func (f *fakeTyping) calls() int        { f.mu.Lock(); defer f.mu.Unlock(); return f.count }
 func (f *fakeTyping) settledCalls() int { f.mu.Lock(); defer f.mu.Unlock(); return f.settled }
 
+type fakeMedia struct {
+	mu            sync.Mutex
+	count         int
+	waitForCancel bool
+}
+
+func (f *fakeMedia) ResolveMedia(ctx context.Context, _ ResolvedInstallation, _ ResolvedIdentity, _ pgtype.UUID, msg channel.InboundMessage) channel.InboundMessage {
+	f.mu.Lock()
+	f.count++
+	waitForCancel := f.waitForCancel
+	f.mu.Unlock()
+	if waitForCancel {
+		<-ctx.Done()
+		return msg
+	}
+	msg.MediaRefs = append(msg.MediaRefs, channel.MediaRef{
+		Type:       channel.MsgTypeImage,
+		StorageKey: "workspaces/ws/lark/image",
+		StorageURL: "https://cdn.example.test/image",
+		Filename:   "image.png",
+		MimeType:   "image/png",
+		SizeBytes:  3,
+	})
+	return msg
+}
+
+func (f *fakeMedia) calls() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.count
+}
+
 type fakeIssues struct {
 	called bool
 	params service.IssueCreateParams
@@ -221,6 +253,7 @@ type harness struct {
 	audit   *fakeAuditor
 	replier *fakeReplier
 	typing  *fakeTyping
+	media   *fakeMedia
 	issues  *fakeIssues
 	tasks   *fakeTasks
 	reader  *fakeReader
@@ -236,6 +269,7 @@ func newHarness(t *testing.T) *harness {
 		audit:   &fakeAuditor{},
 		replier: &fakeReplier{},
 		typing:  &fakeTyping{},
+		media:   &fakeMedia{},
 		issues:  &fakeIssues{},
 		tasks:   &fakeTasks{},
 		reader:  &fakeReader{ws: db.Workspace{IssuePrefix: "MUL"}},
@@ -249,6 +283,7 @@ func newHarness(t *testing.T) *harness {
 		Audit:        h.audit,
 		Replier:      h.replier,
 		Typing:       h.typing,
+		Media:        h.media,
 		OriginType:   "lark_chat",
 	})
 	return h
@@ -286,6 +321,9 @@ func TestRouter_RevokedInstallation_Drops(t *testing.T) {
 	if r, _ := h.audit.last(); r != DropReasonRevokedInstallation {
 		t.Fatalf("expected revoked_installation, got %q", r)
 	}
+	if h.media.calls() != 0 {
+		t.Fatal("revoked installation must not resolve media")
+	}
 }
 
 func TestRouter_Duplicate_Drops(t *testing.T) {
@@ -297,6 +335,9 @@ func TestRouter_Duplicate_Drops(t *testing.T) {
 	if r, _ := h.audit.last(); r != DropReasonDuplicate {
 		t.Fatalf("expected duplicate, got %q", r)
 	}
+	if h.media.calls() != 0 {
+		t.Fatal("duplicate message must not resolve media")
+	}
 }
 
 func TestRouter_GroupNotAddressed_Drops(t *testing.T) {
@@ -313,6 +354,9 @@ func TestRouter_GroupNotAddressed_Drops(t *testing.T) {
 	if h.dedup.marks() != 1 {
 		t.Fatalf("group-filter drop must finalize Mark (1), got %d", h.dedup.marks())
 	}
+	if h.media.calls() != 0 {
+		t.Fatal("unaddressed group message must not resolve media")
+	}
 }
 
 func TestRouter_UnboundSender_NeedsBinding(t *testing.T) {
@@ -327,6 +371,9 @@ func TestRouter_UnboundSender_NeedsBinding(t *testing.T) {
 	if h.dedup.marks() != 1 {
 		t.Fatalf("unbound drop must finalize Mark, got %d", h.dedup.marks())
 	}
+	if h.media.calls() != 0 {
+		t.Fatal("unbound sender must not resolve media")
+	}
 	if !waitFor(time.Second, func() bool {
 		for _, r := range h.replier.calls() {
 			if r.Outcome == OutcomeNeedsBinding && r.Sender == "ou_user_a" {
@@ -348,6 +395,9 @@ func TestRouter_NonMember_Drops(t *testing.T) {
 	if r, _ := h.audit.last(); r != DropReasonNonWorkspaceMember {
 		t.Fatalf("expected non_workspace_member, got %q", r)
 	}
+	if h.media.calls() != 0 {
+		t.Fatal("non-member sender must not resolve media")
+	}
 }
 
 func TestRouter_EnsureSessionError_Releases(t *testing.T) {
@@ -362,6 +412,25 @@ func TestRouter_EnsureSessionError_Releases(t *testing.T) {
 	}
 }
 
+func TestRouter_AppendErrorReleasesClaimAndAllowsMediaRetry(t *testing.T) {
+	h := newHarness(t)
+	h.binder.appendErr = errors.New("db down")
+	if err := h.router.Handle(context.Background(), p2pMessage(t)); err == nil {
+		t.Fatal("append error must surface to the caller")
+	}
+	if h.dedup.releases() != 1 || h.media.calls() != 1 {
+		t.Fatalf("failed attempt: releases=%d media_calls=%d, want 1 each", h.dedup.releases(), h.media.calls())
+	}
+
+	h.binder.appendErr = nil
+	if err := h.router.Handle(context.Background(), p2pMessage(t)); err != nil {
+		t.Fatalf("retry: %v", err)
+	}
+	if h.media.calls() != 2 {
+		t.Fatalf("retry media calls = %d, want 2 total", h.media.calls())
+	}
+}
+
 func TestRouter_Ingested_InTxMark_FinalizeNone(t *testing.T) {
 	h := newHarness(t)
 	h.reader.session = db.ChatSession{}
@@ -381,6 +450,45 @@ func TestRouter_Ingested_InTxMark_FinalizeNone(t *testing.T) {
 	if !waitFor(time.Second, func() bool { return h.typing.calls() == 1 }) {
 		t.Fatalf("ingest must show the typing indicator")
 	}
+	if h.media.calls() != 1 {
+		t.Fatalf("ingested message resolved media %d times, want 1", h.media.calls())
+	}
+	if len(h.binder.lastAppend.Message.MediaRefs) != 1 {
+		t.Fatalf("resolved media not passed to append: %+v", h.binder.lastAppend.Message.MediaRefs)
+	}
+}
+
+func TestRouter_MediaResolverTimeoutAppendsOriginalMessage(t *testing.T) {
+	h := newHarness(t)
+	h.router = NewRouter(h.issues, h.tasks, h.reader, RouterConfig{MediaTimeout: 10 * time.Millisecond, Logger: discardLogger()})
+	h.media.waitForCancel = true
+	h.router.Register(channel.TypeFeishu, ResolverSet{
+		Installation: h.inst,
+		Identity:     h.ident,
+		Dedup:        h.dedup,
+		Session:      h.binder,
+		Audit:        h.audit,
+		Replier:      h.replier,
+		Typing:       h.typing,
+		Media:        h.media,
+		OriginType:   "lark_chat",
+	})
+
+	if err := h.router.Handle(context.Background(), p2pMessage(t)); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if h.media.calls() != 1 {
+		t.Fatalf("media resolver calls = %d, want 1", h.media.calls())
+	}
+	if len(h.binder.lastAppend.Message.MediaRefs) != 0 {
+		t.Fatalf("timed-out media refs must not block or attach: %+v", h.binder.lastAppend.Message.MediaRefs)
+	}
+	if !h.tasks.wasCalled() {
+		t.Fatalf("message should still be ingested and trigger a chat run")
+	}
+	if h.dedup.releases() != 0 {
+		t.Fatalf("media timeout must not release a durably appended message, got %d", h.dedup.releases())
+	}
 }
 
 func TestRouter_ClaimLost_Drops(t *testing.T) {