From 8b1febbe99ced2ff465c2a7d112c9c2246094288 Mon Sep 17 00:00:00 2001 From: wizardchen Date: Mon, 16 Mar 2026 02:40:42 +0800 Subject: [PATCH] feat: implement IM channel sessions and update message handling --- internal/im/service.go | 36 +++++++++++++---- .../versioned/000021_im_channel.down.sql | 3 ++ ...ssions.up.sql => 000021_im_channel.up.sql} | 39 +++++++++++++++++-- .../000021_im_channel_sessions.down.sql | 1 - .../versioned/000022_im_channels.down.sql | 7 ---- .../versioned/000022_im_channels.up.sql | 36 ----------------- ...own.sql => 000022_message_images.down.sql} | 0 ...es.up.sql => 000022_message_images.up.sql} | 0 8 files changed, 67 insertions(+), 55 deletions(-) create mode 100644 migrations/versioned/000021_im_channel.down.sql rename migrations/versioned/{000021_im_channel_sessions.up.sql => 000021_im_channel.up.sql} (50%) delete mode 100644 migrations/versioned/000021_im_channel_sessions.down.sql delete mode 100644 migrations/versioned/000022_im_channels.down.sql delete mode 100644 migrations/versioned/000022_im_channels.up.sql rename migrations/versioned/{000021_message_images.down.sql => 000022_message_images.down.sql} (100%) rename migrations/versioned/{000021_message_images.up.sql => 000022_message_images.up.sql} (100%) diff --git a/internal/im/service.go b/internal/im/service.go index a4d92451d..ff521c2d0 100644 --- a/internal/im/service.go +++ b/internal/im/service.go @@ -2,7 +2,7 @@ package im import ( "context" -"fmt" + "fmt" "strings" "sync" "time" @@ -65,6 +65,24 @@ type Service struct { stopCh chan struct{} } +func buildIMQARequest( + session *types.Session, + query string, + assistantMessageID string, + userMessageID string, + customAgent *types.CustomAgent, + kbIDs []string, +) *types.QARequest { + return &types.QARequest{ + Session: session, + Query: query, + AssistantMessageID: assistantMessageID, + CustomAgent: customAgent, + KnowledgeBaseIDs: kbIDs, + UserMessageID: userMessageID, + } +} + // NewService creates a new IM service. func NewService( db *gorm.DB, @@ -479,10 +497,11 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage, requestID := uuid.New().String() // Create user message - if _, err := s.messageService.CreateMessage(qaCtx, &types.Message{ + userMsg, err := s.messageService.CreateMessage(qaCtx, &types.Message{ SessionID: session.ID, Role: "user", Content: msg.Content, RequestID: requestID, CreatedAt: time.Now(), IsCompleted: true, - }); err != nil { + }) + if err != nil { return fmt.Errorf("create user message: %w", err) } @@ -498,10 +517,11 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage, // Run QA async go func() { var err error + req := buildIMQARequest(session, msg.Content, assistantMsg.ID, userMsg.ID, customAgent, kbIDs) if useAgent { - err = s.sessionService.AgentQA(qaCtx, session, msg.Content, assistantMsg.ID, "", eventBus, customAgent, kbIDs, nil) + err = s.sessionService.AgentQA(qaCtx, req, eventBus) } else { - err = s.sessionService.KnowledgeQA(qaCtx, session, msg.Content, kbIDs, nil, assistantMsg.ID, "", false, eventBus, customAgent, false) + err = s.sessionService.KnowledgeQA(qaCtx, req, eventBus) } if err != nil { logger.Errorf(ctx, "[IM] QA stream execution error: %v", err) @@ -644,7 +664,6 @@ func (s *Service) runQA(ctx context.Context, session *types.Session, query strin if err != nil { return "", fmt.Errorf("create user message: %w", err) } - _ = userMsg // Create a placeholder assistant message assistantMsg, err := s.messageService.CreateMessage(ctx, &types.Message{ @@ -661,10 +680,11 @@ func (s *Service) runQA(ctx context.Context, session *types.Session, query strin // Run QA async go func() { var err error + req := buildIMQARequest(session, query, assistantMsg.ID, userMsg.ID, customAgent, kbIDs) if useAgent { - err = s.sessionService.AgentQA(ctx, session, query, assistantMsg.ID, "", eventBus, customAgent, kbIDs, nil) + err = s.sessionService.AgentQA(ctx, req, eventBus) } else { - err = s.sessionService.KnowledgeQA(ctx, session, query, kbIDs, nil, assistantMsg.ID, "", false, eventBus, customAgent, false) + err = s.sessionService.KnowledgeQA(ctx, req, eventBus) } if err != nil { logger.Errorf(ctx, "[IM] QA execution error: %v", err) diff --git a/migrations/versioned/000021_im_channel.down.sql b/migrations/versioned/000021_im_channel.down.sql new file mode 100644 index 000000000..1f3288743 --- /dev/null +++ b/migrations/versioned/000021_im_channel.down.sql @@ -0,0 +1,3 @@ +ALTER TABLE im_channel_sessions DROP COLUMN IF EXISTS im_channel_id; +DROP TABLE IF EXISTS im_channels; +DROP TABLE IF EXISTS im_channel_sessions; diff --git a/migrations/versioned/000021_im_channel_sessions.up.sql b/migrations/versioned/000021_im_channel.up.sql similarity index 50% rename from migrations/versioned/000021_im_channel_sessions.up.sql rename to migrations/versioned/000021_im_channel.up.sql index 175d2d76e..66048012a 100644 --- a/migrations/versioned/000021_im_channel_sessions.up.sql +++ b/migrations/versioned/000021_im_channel.up.sql @@ -1,6 +1,6 @@ -- Migration: 000021_im_channel_sessions --- Description: Create IM channel-to-session mapping table -DO $$ BEGIN RAISE NOTICE '[Migration 000021] Creating table: im_channel_sessions'; END $$; +-- Description: Create IM channel session mapping and IM channel configuration tables +DO $$ BEGIN RAISE NOTICE '[Migration 000021] Creating IM channel integration tables'; END $$; CREATE TABLE IF NOT EXISTS im_channel_sessions ( id VARCHAR(36) PRIMARY KEY DEFAULT uuid_generate_v4(), @@ -41,4 +41,37 @@ COMMENT ON COLUMN im_channel_sessions.agent_id IS 'Custom agent ID used for this COMMENT ON COLUMN im_channel_sessions.status IS 'Channel status: active, paused, expired'; COMMENT ON COLUMN im_channel_sessions.metadata IS 'Platform-specific extra data (JSON)'; -DO $$ BEGIN RAISE NOTICE '[Migration 000021] im_channel_sessions setup completed successfully!'; END $$; +DO $$ BEGIN RAISE NOTICE '[Migration 000021] Creating table: im_channels'; END $$; + +CREATE TABLE IF NOT EXISTS im_channels ( + id VARCHAR(36) PRIMARY KEY DEFAULT uuid_generate_v4(), + tenant_id BIGINT NOT NULL, + agent_id VARCHAR(36) NOT NULL, + platform VARCHAR(20) NOT NULL, + name VARCHAR(255) NOT NULL DEFAULT '', + enabled BOOLEAN NOT NULL DEFAULT true, + mode VARCHAR(20) NOT NULL DEFAULT 'websocket', + output_mode VARCHAR(20) NOT NULL DEFAULT 'stream', + credentials JSONB NOT NULL DEFAULT '{}', + created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, + deleted_at TIMESTAMP WITH TIME ZONE +); + +CREATE INDEX IF NOT EXISTS idx_im_channels_tenant ON im_channels (tenant_id); +CREATE INDEX IF NOT EXISTS idx_im_channels_agent ON im_channels (agent_id); +CREATE INDEX IF NOT EXISTS idx_im_channels_deleted ON im_channels (deleted_at) WHERE deleted_at IS NOT NULL; + +COMMENT ON TABLE im_channels IS 'IM platform channel configurations bound to agents'; +COMMENT ON COLUMN im_channels.agent_id IS 'Agent ID this channel is bound to'; +COMMENT ON COLUMN im_channels.platform IS 'IM platform: wecom, feishu'; +COMMENT ON COLUMN im_channels.name IS 'User-defined channel name for identification'; +COMMENT ON COLUMN im_channels.mode IS 'Connection mode: webhook or websocket'; +COMMENT ON COLUMN im_channels.output_mode IS 'Output mode: stream (real-time) or full (wait for complete answer)'; +COMMENT ON COLUMN im_channels.credentials IS 'Platform credentials (JSONB): WeCom webhook={corp_id,agent_secret,token,encoding_aes_key,corp_agent_id}, WeCom ws={bot_id,bot_secret}, Feishu={app_id,app_secret,verification_token,encrypt_key}'; + +-- Add im_channel_id column to im_channel_sessions for linking +ALTER TABLE im_channel_sessions ADD COLUMN IF NOT EXISTS im_channel_id VARCHAR(36) DEFAULT ''; +CREATE INDEX IF NOT EXISTS idx_im_channel_sessions_channel ON im_channel_sessions (im_channel_id) WHERE im_channel_id != ''; + +DO $$ BEGIN RAISE NOTICE '[Migration 000021] IM channel integration setup completed successfully!'; END $$; diff --git a/migrations/versioned/000021_im_channel_sessions.down.sql b/migrations/versioned/000021_im_channel_sessions.down.sql deleted file mode 100644 index 2ca2bfb6b..000000000 --- a/migrations/versioned/000021_im_channel_sessions.down.sql +++ /dev/null @@ -1 +0,0 @@ -DROP TABLE IF EXISTS im_channel_sessions; diff --git a/migrations/versioned/000022_im_channels.down.sql b/migrations/versioned/000022_im_channels.down.sql deleted file mode 100644 index e51ebe38e..000000000 --- a/migrations/versioned/000022_im_channels.down.sql +++ /dev/null @@ -1,7 +0,0 @@ --- Rollback: 000022_im_channels -DO $$ BEGIN RAISE NOTICE '[Migration 000022] Rolling back: im_channels'; END $$; - -ALTER TABLE im_channel_sessions DROP COLUMN IF EXISTS im_channel_id; -DROP TABLE IF EXISTS im_channels; - -DO $$ BEGIN RAISE NOTICE '[Migration 000022] Rollback completed'; END $$; diff --git a/migrations/versioned/000022_im_channels.up.sql b/migrations/versioned/000022_im_channels.up.sql deleted file mode 100644 index 5b476e69b..000000000 --- a/migrations/versioned/000022_im_channels.up.sql +++ /dev/null @@ -1,36 +0,0 @@ --- Migration: 000022_im_channels --- Description: Create IM channels table for database-driven IM integration (replaces config.yaml IM settings) -DO $$ BEGIN RAISE NOTICE '[Migration 000022] Creating table: im_channels'; END $$; - -CREATE TABLE IF NOT EXISTS im_channels ( - id VARCHAR(36) PRIMARY KEY DEFAULT uuid_generate_v4(), - tenant_id BIGINT NOT NULL, - agent_id VARCHAR(36) NOT NULL, - platform VARCHAR(20) NOT NULL, - name VARCHAR(255) NOT NULL DEFAULT '', - enabled BOOLEAN NOT NULL DEFAULT true, - mode VARCHAR(20) NOT NULL DEFAULT 'websocket', - output_mode VARCHAR(20) NOT NULL DEFAULT 'stream', - credentials JSONB NOT NULL DEFAULT '{}', - created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, - updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, - deleted_at TIMESTAMP WITH TIME ZONE -); - -CREATE INDEX IF NOT EXISTS idx_im_channels_tenant ON im_channels (tenant_id); -CREATE INDEX IF NOT EXISTS idx_im_channels_agent ON im_channels (agent_id); -CREATE INDEX IF NOT EXISTS idx_im_channels_deleted ON im_channels (deleted_at) WHERE deleted_at IS NOT NULL; - -COMMENT ON TABLE im_channels IS 'IM platform channel configurations bound to agents'; -COMMENT ON COLUMN im_channels.agent_id IS 'Agent ID this channel is bound to'; -COMMENT ON COLUMN im_channels.platform IS 'IM platform: wecom, feishu'; -COMMENT ON COLUMN im_channels.name IS 'User-defined channel name for identification'; -COMMENT ON COLUMN im_channels.mode IS 'Connection mode: webhook or websocket'; -COMMENT ON COLUMN im_channels.output_mode IS 'Output mode: stream (real-time) or full (wait for complete answer)'; -COMMENT ON COLUMN im_channels.credentials IS 'Platform credentials (JSONB): WeCom webhook={corp_id,agent_secret,token,encoding_aes_key,corp_agent_id}, WeCom ws={bot_id,bot_secret}, Feishu={app_id,app_secret,verification_token,encrypt_key}'; - --- Add im_channel_id column to im_channel_sessions for linking -ALTER TABLE im_channel_sessions ADD COLUMN IF NOT EXISTS im_channel_id VARCHAR(36) DEFAULT ''; -CREATE INDEX IF NOT EXISTS idx_im_channel_sessions_channel ON im_channel_sessions (im_channel_id) WHERE im_channel_id != ''; - -DO $$ BEGIN RAISE NOTICE '[Migration 000022] im_channels setup completed successfully!'; END $$; diff --git a/migrations/versioned/000021_message_images.down.sql b/migrations/versioned/000022_message_images.down.sql similarity index 100% rename from migrations/versioned/000021_message_images.down.sql rename to migrations/versioned/000022_message_images.down.sql diff --git a/migrations/versioned/000021_message_images.up.sql b/migrations/versioned/000022_message_images.up.sql similarity index 100% rename from migrations/versioned/000021_message_images.up.sql rename to migrations/versioned/000022_message_images.up.sql