diff --git a/.env.example b/.env.example index 77623625..12afb717 100644 --- a/.env.example +++ b/.env.example @@ -43,3 +43,10 @@ QDRANT_GRPC_PORT=6334 # Webhook & Organization Sync # ORG_SYNC_SECRET=your-webhook-hmac-secret + +# Discord Bot Channel +# Deployment-wide bot token, used when a Discord channel does not carry its own. +# DISCORD_BOT_TOKEN=your-discord-bot-token +# DISCORD_CLIENT_ID=your-discord-application-id +# DISCORD_CLIENT_SECRET=your-discord-client-secret +# DISCORD_PUBLIC_KEY=your-discord-application-public-key diff --git a/internal/bootstrap/routes.go b/internal/bootstrap/routes.go index e35dff8c..1482e841 100644 --- a/internal/bootstrap/routes.go +++ b/internal/bootstrap/routes.go @@ -435,3 +435,8 @@ func registerThirdZaloRoutes(group *gin.RouterGroup) { group.POST("/webhook", third.ZaloPostWebhook) group.POST("/webhook/:channel_id", third.ZaloPostWebhook) } + +func registerThirdDiscordRoutes(group *gin.RouterGroup) { + group.POST("/webhook", third.DiscordPostWebhook) + group.POST("/webhook/:channel_id", third.DiscordPostWebhook) +} diff --git a/internal/bootstrap/server.go b/internal/bootstrap/server.go index d5fc669a..228f22e1 100644 --- a/internal/bootstrap/server.go +++ b/internal/bootstrap/server.go @@ -197,6 +197,7 @@ func addRouter(app *gin.Engine) { registerThirdWechatRoutes(thirdGroup.Group("/wechat")) registerThirdTelegramRoutes(thirdGroup.Group("/telegram")) registerThirdZaloRoutes(thirdGroup.Group("/zalo")) + registerThirdDiscordRoutes(thirdGroup.Group("/discord")) } type spaShellRewrite struct { diff --git a/internal/discord/client.go b/internal/discord/client.go new file mode 100644 index 00000000..4045ae1c --- /dev/null +++ b/internal/discord/client.go @@ -0,0 +1,139 @@ +package discord + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "time" +) + +const defaultBaseURL = "https://discord.com/api/v10" + +type Client struct { + botToken string + baseURL string + httpClient *http.Client +} + +func NewClient(botToken string) *Client { + return &Client{ + botToken: strings.TrimSpace(botToken), + baseURL: defaultBaseURL, + httpClient: &http.Client{Timeout: 15 * time.Second}, + } +} + +func (c *Client) SetBaseURL(url string) { + if strings.TrimSpace(url) != "" { + c.baseURL = strings.TrimRight(strings.TrimSpace(url), "/") + } +} + +func (c *Client) GetMe(ctx context.Context) (*User, error) { + var user User + if err := c.doRequest(ctx, http.MethodGet, "/users/@me", nil, &user); err != nil { + return nil, err + } + return &user, nil +} + +func (c *Client) CreateDMChannel(ctx context.Context, recipientID string) (*Channel, error) { + if strings.TrimSpace(recipientID) == "" { + return nil, fmt.Errorf("recipient_id is required") + } + req := CreateDMRequest{RecipientID: strings.TrimSpace(recipientID)} + var channel Channel + if err := c.doRequest(ctx, http.MethodPost, "/users/@me/channels", req, &channel); err != nil { + return nil, err + } + return &channel, nil +} + +func (c *Client) SendMessage(ctx context.Context, channelID string, content string) (*Message, error) { + channelID = strings.TrimSpace(channelID) + if channelID == "" { + return nil, fmt.Errorf("channel_id is required") + } + if strings.TrimSpace(content) == "" { + return nil, fmt.Errorf("content is required") + } + + req := SendMessageRequest{Content: content} + var msg Message + endpoint := fmt.Sprintf("/channels/%s/messages", channelID) + if err := c.doRequest(ctx, http.MethodPost, endpoint, req, &msg); err != nil { + return nil, err + } + return &msg, nil +} + +func (c *Client) SendEmbedMessage(ctx context.Context, channelID string, content string, embeds []Embed) (*Message, error) { + channelID = strings.TrimSpace(channelID) + if channelID == "" { + return nil, fmt.Errorf("channel_id is required") + } + + req := SendMessageRequest{ + Content: content, + Embeds: embeds, + } + var msg Message + endpoint := fmt.Sprintf("/channels/%s/messages", channelID) + if err := c.doRequest(ctx, http.MethodPost, endpoint, req, &msg); err != nil { + return nil, err + } + return &msg, nil +} + +func (c *Client) doRequest(ctx context.Context, method, path string, payload any, result any) error { + if c.botToken == "" { + return fmt.Errorf("discord bot token is required") + } + + endpoint := fmt.Sprintf("%s%s", c.baseURL, path) + + var bodyReader io.Reader + if payload != nil { + bodyBytes, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("marshal discord request failed: %w", err) + } + bodyReader = bytes.NewBuffer(bodyBytes) + } + + req, err := http.NewRequestWithContext(ctx, method, endpoint, bodyReader) + if err != nil { + return fmt.Errorf("create discord request failed: %w", err) + } + + req.Header.Set("Authorization", "Bot "+c.botToken) + if payload != nil { + req.Header.Set("Content-Type", "application/json") + } + + res, err := c.httpClient.Do(req) + if err != nil { + return fmt.Errorf("discord http request failed: %w", err) + } + defer res.Body.Close() + + bodyBytes, err := io.ReadAll(res.Body) + if err != nil { + return fmt.Errorf("read discord response failed: %w", err) + } + + if res.StatusCode < 200 || res.StatusCode >= 300 { + return fmt.Errorf("discord api error (%d): %s", res.StatusCode, string(bodyBytes)) + } + + if result != nil { + if err := json.Unmarshal(bodyBytes, result); err != nil { + return fmt.Errorf("unmarshal discord response failed: %w (body: %s)", err, string(bodyBytes)) + } + } + return nil +} diff --git a/internal/discord/client_test.go b/internal/discord/client_test.go new file mode 100644 index 00000000..de1b7c85 --- /dev/null +++ b/internal/discord/client_test.go @@ -0,0 +1,90 @@ +package discord + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" +) + +func TestDiscordSendMessage(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bot test_token" { + t.Errorf("expected Bot test_token, got %s", r.Header.Get("Authorization")) + } + if r.URL.Path != "/channels/789/messages" { + t.Errorf("expected path /channels/789/messages, got %s", r.URL.Path) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"id":"123456","channel_id":"789","content":"hello"}`)) + })) + defer server.Close() + + client := NewClient("test_token") + client.SetBaseURL(server.URL) + + resp, err := client.SendMessage(context.Background(), "789", "hello") + if err != nil { + t.Fatalf("SendMessage failed: %v", err) + } + if resp.ID != "123456" { + t.Errorf("expected ID 123456, got %s", resp.ID) + } +} + +func TestDiscordSendEmbedMessage(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bot test_token" { + t.Errorf("expected Bot test_token, got %s", r.Header.Get("Authorization")) + } + if r.URL.Path != "/channels/789/messages" { + t.Errorf("expected path /channels/789/messages, got %s", r.URL.Path) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"id":"embed_123","channel_id":"789","content":"Check image"}`)) + })) + defer server.Close() + + client := NewClient("test_token") + client.SetBaseURL(server.URL) + + embed := Embed{ + Title: "Screenshot", + Image: &EmbedMedia{URL: "https://example.com/img.png"}, + } + resp, err := client.SendEmbedMessage(context.Background(), "789", "Check image", []Embed{embed}) + if err != nil { + t.Fatalf("SendEmbedMessage failed: %v", err) + } + if resp.ID != "embed_123" { + t.Errorf("expected ID embed_123, got %s", resp.ID) + } +} + +func TestDiscordCreateDMChannel(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bot test_token" { + t.Errorf("expected Bot test_token, got %s", r.Header.Get("Authorization")) + } + if r.URL.Path != "/users/@me/channels" { + t.Errorf("expected path /users/@me/channels, got %s", r.URL.Path) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"id":"dm_chan_123","type":1}`)) + })) + defer server.Close() + + client := NewClient("test_token") + client.SetBaseURL(server.URL) + + resp, err := client.CreateDMChannel(context.Background(), "user_999") + if err != nil { + t.Fatalf("CreateDMChannel failed: %v", err) + } + if resp.ID != "dm_chan_123" { + t.Errorf("expected ID dm_chan_123, got %s", resp.ID) + } +} diff --git a/internal/discord/types.go b/internal/discord/types.go new file mode 100644 index 00000000..3363bebd --- /dev/null +++ b/internal/discord/types.go @@ -0,0 +1,80 @@ +package discord + +// User represents a Discord user. +type User struct { + ID string `json:"id"` + Username string `json:"username"` + Discriminator string `json:"discriminator,omitempty"` + GlobalName string `json:"global_name,omitempty"` + Avatar string `json:"avatar,omitempty"` + Bot bool `json:"bot,omitempty"` +} + +// Channel represents a Discord channel (Guild Text, DM, Thread, etc.). +type Channel struct { + ID string `json:"id"` + Type int `json:"type"` + GuildID string `json:"guild_id,omitempty"` + Name string `json:"name,omitempty"` +} + +// Attachment represents a file or image uploaded to Discord. +type Attachment struct { + ID string `json:"id"` + Filename string `json:"filename"` + URL string `json:"url"` + ProxyURL string `json:"proxy_url,omitempty"` + ContentType string `json:"content_type,omitempty"` + Size int64 `json:"size,omitempty"` +} + +// EmbedMedia represents an image/video/thumbnail inside an Embed. +type EmbedMedia struct { + URL string `json:"url"` +} + +// Embed represents a Discord rich embed object. +type Embed struct { + Title string `json:"title,omitempty"` + Description string `json:"description,omitempty"` + URL string `json:"url,omitempty"` + Color int `json:"color,omitempty"` + Image *EmbedMedia `json:"image,omitempty"` +} + +// Message represents a Discord message. +type Message struct { + ID string `json:"id"` + ChannelID string `json:"channel_id"` + GuildID string `json:"guild_id,omitempty"` + Author User `json:"author"` + Content string `json:"content"` + Timestamp string `json:"timestamp"` + Attachments []Attachment `json:"attachments,omitempty"` + Embeds []Embed `json:"embeds,omitempty"` +} + +// SendMessageRequest represents payload for Discord create message API. +type SendMessageRequest struct { + Content string `json:"content,omitempty"` + Embeds []Embed `json:"embeds,omitempty"` +} + +// CreateDMRequest represents payload for Discord create DM channel API. +type CreateDMRequest struct { + RecipientID string `json:"recipient_id"` +} + +// WebhookPayload represents an incoming message/event from Discord Gateway or Webhook. +type WebhookPayload struct { + ID string `json:"id,omitempty"` + Type int `json:"type,omitempty"` + GuildID string `json:"guild_id,omitempty"` + ChannelID string `json:"channel_id,omitempty"` + Author *User `json:"author,omitempty"` + Content string `json:"content,omitempty"` + Timestamp string `json:"timestamp,omitempty"` + Attachments []Attachment `json:"attachments,omitempty"` + Embeds []Embed `json:"embeds,omitempty"` + Message *Message `json:"message,omitempty"` +} diff --git a/internal/handlers/third/discord_handler.go b/internal/handlers/third/discord_handler.go new file mode 100644 index 00000000..e36ba526 --- /dev/null +++ b/internal/handlers/third/discord_handler.go @@ -0,0 +1,39 @@ +package third + +import ( + "bytes" + "io" + "net/http" + "strings" + + "agent-desk/internal/services" + + "github.com/gin-gonic/gin" +) + +// DiscordPostWebhook receives incoming Webhook events from Discord. +func DiscordPostWebhook(ctx *gin.Context) { + channelID := strings.TrimSpace(ctx.Param("channel_id")) + if channelID == "" { + channelID = strings.TrimSpace(ctx.Query("channel_id")) + } + + secretHeader := ctx.GetHeader("X-Discord-Secret-Token") + if secretHeader == "" { + secretHeader = ctx.GetHeader("X-Webhook-Secret") + } + + bodyBytes, err := io.ReadAll(ctx.Request.Body) + if err != nil { + ctx.JSON(http.StatusBadRequest, gin.H{"ok": false, "error": "failed to read body"}) + return + } + ctx.Request.Body = io.NopCloser(bytes.NewBuffer(bodyBytes)) + + if err := services.DiscordInboundService.HandleWebhook(ctx.Request.Context(), channelID, secretHeader, bodyBytes); err != nil { + ctx.JSON(http.StatusOK, gin.H{"ok": false, "error": err.Error()}) + return + } + + ctx.JSON(http.StatusOK, gin.H{"ok": true}) +} diff --git a/internal/handlers/third/discord_handler_test.go b/internal/handlers/third/discord_handler_test.go new file mode 100644 index 00000000..e3b8e4f5 --- /dev/null +++ b/internal/handlers/third/discord_handler_test.go @@ -0,0 +1,115 @@ +package third + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "agent-desk/internal/models" + "agent-desk/internal/pkg/dto" + "agent-desk/internal/pkg/dto/request" + "agent-desk/internal/pkg/enums" + "agent-desk/internal/repositories" + "agent-desk/internal/services" + + "github.com/gin-gonic/gin" + "github.com/mlogclub/simple/sqls" +) + +func TestDiscordPostWebhook_Handler(t *testing.T) { + gin.SetMode(gin.TestMode) + db := setupThirdHandlerTestDB(t) + + now := time.Now() + agent := &models.AIAgent{ + Name: "Discord Agent", + ServiceMode: enums.IMConversationServiceModeAIFirst, + PublishedRevisionID: 1, + WelcomeMessage: "Hello Discord!", + Status: enums.StatusOk, + AuditFields: models.AuditFields{CreatedAt: now, UpdatedAt: now}, + } + _ = db.Create(agent) + + discordConfig, _ := json.Marshal(dto.DiscordChannelConfig{ + GuildID: "guild_999", + BotToken: "test_bot_token", + WebhookSecret: "secret_discord_123", + WelcomeMessage: "Welcome!", + }) + + operator := &dto.AuthPrincipal{UserID: 1, Username: "admin"} + channel, err := services.ChannelService.CreateChannel(request.CreateChannelRequest{ + Name: "Discord Community", + ChannelType: enums.ChannelTypeDiscord, + AIAgentID: agent.ID, + AIAgentRolloutPercent: 100, + ConfigJSON: string(discordConfig), + Status: int(enums.StatusOk), + }, operator) + if err != nil { + t.Fatalf("CreateChannel failed: %v", err) + } + + router := gin.New() + router.POST("/api/third/discord/webhook/:channel_id", DiscordPostWebhook) + router.POST("/api/third/discord/webhook", DiscordPostWebhook) + + payload := []byte(`{ + "id": "msg_001", + "channel_id": "ch_777", + "guild_id": "guild_999", + "content": "Need help with setup", + "author": { + "id": "user_456", + "username": "gamer_one", + "global_name": "Gamer One", + "bot": false + } + }`) + + // 1. Invalid secret + req, _ := http.NewRequest(http.MethodPost, "/api/third/discord/webhook/"+channel.ChannelID, bytes.NewBuffer(payload)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Discord-Secret-Token", "wrong_secret") + + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("expected 200 OK wrapper, got: %d", rec.Code) + } + var resp map[string]any + _ = json.Unmarshal(rec.Body.Bytes(), &resp) + if resp["ok"] == true { + t.Fatalf("expected error for invalid secret token") + } + + // 2. Valid secret + req2, _ := http.NewRequest(http.MethodPost, "/api/third/discord/webhook/"+channel.ChannelID, bytes.NewBuffer(payload)) + req2.Header.Set("Content-Type", "application/json") + req2.Header.Set("X-Discord-Secret-Token", "secret_discord_123") + + rec2 := httptest.NewRecorder() + router.ServeHTTP(rec2, req2) + + if rec2.Code != http.StatusOK { + t.Fatalf("expected 200 OK, got %d", rec2.Code) + } + var resp2 map[string]any + _ = json.Unmarshal(rec2.Body.Bytes(), &resp2) + if resp2["ok"] != true { + t.Fatalf("expected ok: true, got: %+v", resp2) + } + + // Verify identity in DB + identity := repositories.CustomerIdentityRepository.FindOne(db, sqls.NewCnd(). + Eq("external_source", enums.ExternalSourceDiscord). + Eq("external_id", "user_456")) + if identity == nil { + t.Fatalf("expected customer identity for user_456") + } +} diff --git a/internal/pkg/config/config.go b/internal/pkg/config/config.go index 082de4dc..5925d94d 100644 --- a/internal/pkg/config/config.go +++ b/internal/pkg/config/config.go @@ -25,6 +25,7 @@ type Config struct { OIDC OIDCConfig `yaml:"oidc"` CustomerSession CustomerSessionConfig `yaml:"customerSession"` Webhook WebhookConfig `yaml:"webhook"` + Discord DiscordConfig `yaml:"discord"` } func (c Config) LanguageOrDefault() string { @@ -230,6 +231,16 @@ type WebhookConfig struct { DOSOrgSyncSecret string `yaml:"dosOrgSyncSecret"` } +// DiscordConfig holds deployment-wide Discord bot credentials. A channel may +// carry its own bot token, which takes precedence; these are the fallback for a +// single shared bot. +type DiscordConfig struct { + ClientID string `yaml:"clientId"` + ClientSecret string `yaml:"clientSecret"` + BotToken string `yaml:"botToken"` + PublicKey string `yaml:"publicKey"` +} + func Load(path string) (*Config, error) { loadDotEnv(path) @@ -312,6 +323,10 @@ func bindConfigDefaults(v *viper.Viper) { v.SetDefault("vectorDB.qdrant.host", "127.0.0.1") v.SetDefault("vectorDB.qdrant.grpcPort", 6334) v.SetDefault("mcp.enabled", true) + v.SetDefault("discord.clientId", "") + v.SetDefault("discord.clientSecret", "") + v.SetDefault("discord.botToken", "") + v.SetDefault("discord.publicKey", "") } func bindEnvironmentAliases(v *viper.Viper) { @@ -339,6 +354,10 @@ func bindEnvironmentAliases(v *viper.Viper) { _ = v.BindEnv("oidc.clientSecret", "AGENT_DESK_OIDC_CLIENTSECRET", "OIDC_CLIENT_SECRET", "CUSTOM_OAUTH_CLIENT_SECRET") _ = v.BindEnv("oidc.redirectUrl", "AGENT_DESK_OIDC_REDIRECTURL", "OIDC_REDIRECT_URL", "CUSTOM_OAUTH_REDIRECT_URI") _ = v.BindEnv("webhook.orgSyncSecret", "AGENT_DESK_WEBHOOK_ORGSYNCSECRET", "ORG_SYNC_SECRET", "WEBHOOK_SECRET") + _ = v.BindEnv("discord.clientId", "AGENT_DESK_DISCORD_CLIENTID", "DISCORD_CLIENT_ID") + _ = v.BindEnv("discord.clientSecret", "AGENT_DESK_DISCORD_CLIENTSECRET", "DISCORD_CLIENT_SECRET") + _ = v.BindEnv("discord.botToken", "AGENT_DESK_DISCORD_BOTTOKEN", "DISCORD_BOT_TOKEN") + _ = v.BindEnv("discord.publicKey", "AGENT_DESK_DISCORD_PUBLICKEY", "DISCORD_PUBLIC_KEY") } func normalizeLoadedConfig(cfg *Config) { diff --git a/internal/pkg/dto/dto.go b/internal/pkg/dto/dto.go index 276d03a1..5f456c03 100644 --- a/internal/pkg/dto/dto.go +++ b/internal/pkg/dto/dto.go @@ -49,3 +49,14 @@ type ZaloOAChannelConfig struct { WebhookSecret string `json:"webhookSecret,omitempty"` WelcomeMessage string `json:"welcomeMessage,omitempty"` } + +type DiscordChannelConfig struct { + GuildID string `json:"guildId,omitempty"` + GuildName string `json:"guildName,omitempty"` + ChannelScope string `json:"channelScope,omitempty"` // all | dm_only + BotToken string `json:"botToken,omitempty"` + ApplicationID string `json:"applicationId,omitempty"` + PublicKey string `json:"publicKey,omitempty"` + WebhookSecret string `json:"webhookSecret,omitempty"` + WelcomeMessage string `json:"welcomeMessage,omitempty"` +} diff --git a/internal/pkg/enums/external_identity.go b/internal/pkg/enums/external_identity.go index 8135fd56..e9fdf9c0 100644 --- a/internal/pkg/enums/external_identity.go +++ b/internal/pkg/enums/external_identity.go @@ -11,6 +11,7 @@ const ( ExternalSourceUser ExternalSource = "user" // 用户信息 ExternalSourceTelegram ExternalSource = "telegram" // Telegram ExternalSourceZaloOA ExternalSource = "zalo_oa" // Zalo OA + ExternalSourceDiscord ExternalSource = "discord" // Discord ) var externalSourceLabelMap = map[ExternalSource]string{ @@ -19,6 +20,7 @@ var externalSourceLabelMap = map[ExternalSource]string{ ExternalSourceUser: "用户", ExternalSourceTelegram: "Telegram", ExternalSourceZaloOA: "Zalo OA", + ExternalSourceDiscord: "Discord", } func GetExternalSourceLabel(v ExternalSource) string { diff --git a/internal/pkg/enums/wxwork_kf.go b/internal/pkg/enums/wxwork_kf.go index 825f7fa6..33d7fc9b 100644 --- a/internal/pkg/enums/wxwork_kf.go +++ b/internal/pkg/enums/wxwork_kf.go @@ -23,6 +23,7 @@ const ( ChannelTypeWxWorkKF = "wxwork_kf" ChannelTypeTelegram = "telegram" ChannelTypeZaloOA = "zalo_oa" + ChannelTypeDiscord = "discord" ) type WxWorkKFMessageSendStatus string diff --git a/internal/services/channel_message_outbox_service.go b/internal/services/channel_message_outbox_service.go index 12014241..8e40f10d 100644 --- a/internal/services/channel_message_outbox_service.go +++ b/internal/services/channel_message_outbox_service.go @@ -250,6 +250,68 @@ func (s *channelMessageOutboxService) EnqueueZaloOAMessage(conversation *models. return nil } +func (s *channelMessageOutboxService) EnqueueDiscordMessage(conversation *models.Conversation, message *models.Message) error { + if conversation == nil || message == nil { + return nil + } + channel := ChannelService.Get(conversation.ChannelID) + if channel == nil || channel.ChannelType != enums.ChannelTypeDiscord { + return nil + } + if message.SenderType != enums.IMSenderTypeAgent && message.SenderType != enums.IMSenderTypeAI { + return nil + } + if message.MessageType != enums.IMMessageTypeText && message.MessageType != enums.IMMessageTypeHTML && message.MessageType != enums.IMMessageTypeImage && message.MessageType != enums.IMMessageTypeAttachment { + return nil + } + if existing := s.GetByMessageID(enums.ChannelTypeDiscord, message.ID); existing != nil { + return nil + } + + payload, err := json.Marshal(map[string]any{ + "conversationId": conversation.ID, + "messageId": message.ID, + "messageType": message.MessageType, + "content": strings.TrimSpace(message.Content), + "payload": strings.TrimSpace(message.Payload), + "senderId": message.SenderID, + }) + if err != nil { + return err + } + + now := time.Now() + err = s.Create(&models.ChannelMessageOutbox{ + ChannelType: enums.ChannelTypeDiscord, + ConversationID: conversation.ID, + MessageID: message.ID, + Payload: string(payload), + SendStatus: string(enums.ChannelMessageOutboxStatusPending), + AuditFields: models.AuditFields{ + CreatedAt: now, + CreateUserID: message.UpdateUserID, + CreateUserName: message.UpdateUserName, + UpdatedAt: now, + UpdateUserID: message.UpdateUserID, + UpdateUserName: message.UpdateUserName, + }, + }) + if err != nil { + return err + } + + go func() { + defer func() { + if r := recover(); r != nil { + slog.Error("recovered from panic in discord outbound dispatch", "error", r) + } + }() + DiscordOutboundService.DispatchPendingOutbox() + }() + + return nil +} + func (s *channelMessageOutboxService) ListPending(channelType string, limit int) []models.ChannelMessageOutbox { if limit <= 0 { limit = 20 diff --git a/internal/services/channel_service.go b/internal/services/channel_service.go index 6a5399fc..92e2b640 100644 --- a/internal/services/channel_service.go +++ b/internal/services/channel_service.go @@ -333,6 +333,25 @@ func (s *channelService) ParseZaloOAChannelConfig(raw string) (*dto.ZaloOAChanne return cfg, nil } +func (s *channelService) ParseDiscordChannelConfig(raw string) (*dto.DiscordChannelConfig, error) { + raw = strings.TrimSpace(raw) + cfg := &dto.DiscordChannelConfig{} + if raw != "" { + if err := json.Unmarshal([]byte(raw), cfg); err != nil { + return nil, err + } + } + cfg.GuildID = strings.TrimSpace(cfg.GuildID) + cfg.GuildName = strings.TrimSpace(cfg.GuildName) + cfg.ChannelScope = strings.TrimSpace(cfg.ChannelScope) + cfg.BotToken = strings.TrimSpace(cfg.BotToken) + cfg.ApplicationID = strings.TrimSpace(cfg.ApplicationID) + cfg.PublicKey = strings.TrimSpace(cfg.PublicKey) + cfg.WebhookSecret = strings.TrimSpace(cfg.WebhookSecret) + cfg.WelcomeMessage = strings.TrimSpace(cfg.WelcomeMessage) + return cfg, nil +} + func (s *channelService) GetUserTokenSecret(channel *models.Channel) string { if channel == nil { return "" @@ -449,7 +468,7 @@ func (s *channelService) GetEnabledChannel(ctx *gin.Context) *models.Channel { func (s *channelService) buildChannelModel(id int64, req request.CreateChannelRequest) (*models.Channel, error) { channelType := strings.TrimSpace(req.ChannelType) - if channelType != enums.ChannelTypeWeb && channelType != enums.ChannelTypeWechatMP && channelType != enums.ChannelTypeWxWorkKF && channelType != enums.ChannelTypeTelegram && channelType != enums.ChannelTypeZaloOA { + if channelType != enums.ChannelTypeWeb && channelType != enums.ChannelTypeWechatMP && channelType != enums.ChannelTypeWxWorkKF && channelType != enums.ChannelTypeTelegram && channelType != enums.ChannelTypeZaloOA && channelType != enums.ChannelTypeDiscord { return nil, errorsx.InvalidParamI18n("error.e0250") } name := strings.TrimSpace(req.Name) @@ -596,6 +615,32 @@ func (s *channelService) buildChannelModel(id int64, req request.CreateChannelRe return nil, err } configJSON = string(configBytes) + case enums.ChannelTypeDiscord: + if channelID == "" { + channelID = strs.UUID() + } + if exists := s.Take("channel_id = ? AND status <> ? AND id <> ?", channelID, enums.StatusDeleted, id); exists != nil { + return nil, errorsx.InvalidParamI18n("error.e0248") + } + cfg, err := s.ParseDiscordChannelConfig(configJSON) + if err != nil { + return nil, errorsx.InvalidParam("invalid discord configuration") + } + // A channel may rely on the deployment-wide bot token instead of carrying + // its own, so the token is not required here the way Telegram's is. + if cfg.ChannelScope != "" && cfg.ChannelScope != "all" && cfg.ChannelScope != "dm_only" { + return nil, errorsx.InvalidParam("discord channelScope must be all or dm_only") + } + if cfg.WebhookSecret == "" { + if secret, err := generateUserTokenSecret(); err == nil { + cfg.WebhookSecret = secret + } + } + configBytes, err := json.Marshal(cfg) + if err != nil { + return nil, err + } + configJSON = string(configBytes) } return &models.Channel{ diff --git a/internal/services/cronx/cron.go b/internal/services/cronx/cron.go index 52ab00e2..e5c3af47 100644 --- a/internal/services/cronx/cron.go +++ b/internal/services/cronx/cron.go @@ -34,6 +34,10 @@ func Init() { if zaloCount > 0 { slog.Info("zalo oa outbox dispatched", "count", zaloCount) } + discordCount := services.DiscordOutboundService.DispatchPendingOutbox() + if discordCount > 0 { + slog.Info("discord outbox dispatched", "count", discordCount) + } }) c.Start() diff --git a/internal/services/discord_inbound_service.go b/internal/services/discord_inbound_service.go new file mode 100644 index 00000000..93f20eba --- /dev/null +++ b/internal/services/discord_inbound_service.go @@ -0,0 +1,176 @@ +package services + +import ( + "context" + "crypto/subtle" + "encoding/json" + "fmt" + "log/slog" + "strings" + + "agent-desk/internal/discord" + "agent-desk/internal/models" + "agent-desk/internal/pkg/enums" + "agent-desk/internal/pkg/errorsx" + "agent-desk/internal/pkg/openidentity" +) + +var DiscordInboundService = newDiscordInboundService() + +func newDiscordInboundService() *discordInboundService { + return &discordInboundService{} +} + +type discordInboundService struct{} + +// HandleWebhook processes an incoming webhook or gateway payload from Discord. +func (s *discordInboundService) HandleWebhook(ctx context.Context, channelID string, secretHeader string, rawPayload []byte) error { + channelID = strings.TrimSpace(channelID) + var channel *models.Channel + if channelID != "" { + channel = ChannelService.Take("channel_id = ? AND channel_type = ? AND status = ?", channelID, enums.ChannelTypeDiscord, enums.StatusOk) + } + if channel == nil { + channel = ChannelService.Take("channel_type = ? AND status = ?", enums.ChannelTypeDiscord, enums.StatusOk) + } + if channel == nil { + return errorsx.InvalidParam("discord channel not found or disabled") + } + + cfg, err := ChannelService.ParseDiscordChannelConfig(channel.ConfigJSON) + if err != nil || cfg == nil { + return errorsx.InvalidParam("discord channel config invalid") + } + + // Compared in constant time: a byte-wise != leaks how much of the prefix + // matched through response timing. + if cfg.WebhookSecret != "" && + subtle.ConstantTimeCompare([]byte(strings.TrimSpace(secretHeader)), []byte(cfg.WebhookSecret)) != 1 { + return errorsx.UnauthorizedI18n("error.auth.invalidSignature") + } + + var payload discord.WebhookPayload + if err := json.Unmarshal(rawPayload, &payload); err != nil { + return fmt.Errorf("unmarshal discord payload failed: %w", err) + } + + author := payload.Author + text := strings.TrimSpace(payload.Content) + msgID := payload.ID + targetChannelID := payload.ChannelID + guildID := payload.GuildID + attachments := payload.Attachments + embeds := payload.Embeds + + if payload.Message != nil { + if author == nil { + author = &payload.Message.Author + } + if text == "" { + text = strings.TrimSpace(payload.Message.Content) + } + if msgID == "" { + msgID = payload.Message.ID + } + if targetChannelID == "" { + targetChannelID = payload.Message.ChannelID + } + if guildID == "" { + guildID = payload.Message.GuildID + } + if len(attachments) == 0 && len(payload.Message.Attachments) > 0 { + attachments = payload.Message.Attachments + } + if len(embeds) == 0 && len(payload.Message.Embeds) > 0 { + embeds = payload.Message.Embeds + } + } + + if author == nil || author.Bot || strings.TrimSpace(author.ID) == "" { + return nil // Ignore bot messages or invalid authors + } + + // Honour the channel's guild scope. A bot can be invited to several servers, + // and without these checks GuildID and ChannelScope would be stored + // configuration that silently does nothing. + if cfg.GuildID != "" && guildID != cfg.GuildID { + slog.Debug("ignoring discord message from an out-of-scope guild", + "guild_id", guildID, + "channel", channel.ID, + ) + return nil + } + if cfg.ChannelScope == "dm_only" && guildID != "" { + slog.Debug("ignoring discord guild message, channel is dm_only", + "guild_id", guildID, + "channel", channel.ID, + ) + return nil + } + + if text == "" && len(attachments) > 0 { + firstAtt := attachments[0] + if firstAtt.Filename != "" { + text = fmt.Sprintf("[%s] %s", firstAtt.Filename, firstAtt.URL) + } else { + text = firstAtt.URL + } + } + + if text == "" && len(attachments) == 0 && len(embeds) == 0 { + return nil // Ignore empty messages + } + if text == "" && len(embeds) > 0 { + text = embeds[0].Description + if text == "" { + text = embeds[0].Title + } + } + + // 1. Resolve customer identity + externalID := author.ID + name := strings.TrimSpace(author.GlobalName) + if name == "" { + name = strings.TrimSpace(author.Username) + } + if name == "" { + name = fmt.Sprintf("Discord User %s", author.ID) + } + + externalUser := openidentity.ExternalUser{ + ExternalSource: enums.ExternalSourceDiscord, + ExternalID: externalID, + ExternalName: name, + } + + // 2. Create or match Conversation + conversation, err := ConversationService.Create(externalUser, channel.ID, channel.AIAgentID) + if err != nil { + return fmt.Errorf("create discord conversation failed: %w", err) + } + + // 3. Send message through MessageService + clientMsgID := fmt.Sprintf("discord_%s_%s", targetChannelID, msgID) + payloadMap := map[string]any{ + "discord_message_id": msgID, + "discord_channel_id": targetChannelID, + "discord_guild_id": guildID, + "discord_user_id": author.ID, + "discord_attachments": attachments, + } + payloadBytes, _ := json.Marshal(payloadMap) + + _, err = MessageService.SendCustomerMessage( + conversation.ID, + clientMsgID, + enums.IMMessageTypeText, + text, + string(payloadBytes), + externalUser, + ) + if err != nil { + return fmt.Errorf("send customer message failed: %w", err) + } + + return nil +} diff --git a/internal/services/discord_inbound_service_test.go b/internal/services/discord_inbound_service_test.go new file mode 100644 index 00000000..b6c31b24 --- /dev/null +++ b/internal/services/discord_inbound_service_test.go @@ -0,0 +1,294 @@ +package services + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "agent-desk/internal/models" + "agent-desk/internal/pkg/dto" + "agent-desk/internal/pkg/enums" + "agent-desk/internal/repositories" + + "github.com/glebarez/sqlite" + "github.com/mlogclub/simple/sqls" + "gorm.io/gorm" + "gorm.io/gorm/schema" +) + +func setupDiscordTestDB(t *testing.T) *gorm.DB { + t.Helper() + db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{ + NamingStrategy: schema.NamingStrategy{ + TablePrefix: "t_", + SingularTable: true, + }, + }) + if err != nil { + t.Fatalf("open sqlite db: %v", err) + } + if err := db.AutoMigrate( + &models.Channel{}, + &models.ChannelMessageOutbox{}, + &models.Customer{}, + &models.CustomerIdentity{}, + &models.CustomerContact{}, + &models.Conversation{}, + &models.ConversationParticipant{}, + &models.ConversationReadState{}, + &models.ConversationInterrupt{}, + &models.ConversationEventLog{}, + &models.Message{}, + &models.AIAgent{}, + &models.User{}, + &models.Role{}, + &models.UserRole{}, + &models.Permission{}, + &models.RolePermission{}, + &models.UserPermission{}, + ); err != nil { + t.Fatalf("migrate discord test tables: %v", err) + } + sqls.SetDB(db) + return db +} + +func TestDiscordInboundAndOutbound(t *testing.T) { + db := setupDiscordTestDB(t) + + mockServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"id":"out_msg_100","channel_id":"text_chan_1","content":"Agent reply"}`)) + })) + defer mockServer.Close() + + now := time.Now() + aiAgent := &models.AIAgent{ + Name: "Support AI", + Status: enums.StatusOk, + PublishedRevisionID: 1, + AuditFields: models.AuditFields{CreatedAt: now, UpdatedAt: now}, + } + if err := db.Create(aiAgent).Error; err != nil { + t.Fatalf("create ai agent: %v", err) + } + + discordConfig := dto.DiscordChannelConfig{ + GuildID: "guild_12345", + GuildName: "Test Guild", + BotToken: "discord_bot_token", + WebhookSecret: "test_secret", + } + cfgBytes, _ := json.Marshal(discordConfig) + + channel := &models.Channel{ + ChannelType: enums.ChannelTypeDiscord, + ChannelID: "discord_ch_1", + AIAgentID: aiAgent.ID, + AIAgentRolloutPercent: 100, + Name: "Community Support", + ConfigJSON: string(cfgBytes), + Status: enums.StatusOk, + AuditFields: models.AuditFields{CreatedAt: now, UpdatedAt: now}, + } + if err := db.Create(channel).Error; err != nil { + t.Fatalf("create discord channel: %v", err) + } + + payload := `{ + "id": "msg_999", + "channel_id": "text_chan_1", + "guild_id": "guild_12345", + "content": "", + "author": { + "id": "user_888", + "username": "gamer_joy", + "global_name": "Joy Le", + "bot": false + }, + "attachments": [ + { + "id": "att_1", + "filename": "screenshot.png", + "url": "https://cdn.discordapp.com/attachments/1/screenshot.png", + "content_type": "image/png", + "size": 10240 + } + ] + }` + + ctx := context.Background() + err := DiscordInboundService.HandleWebhook(ctx, channel.ChannelID, "test_secret", []byte(payload)) + if err != nil { + t.Fatalf("HandleWebhook failed: %v", err) + } + + // Verify customer identity + identity := repositories.CustomerIdentityRepository.FindOne(sqls.DB(), sqls.NewCnd(). + Eq("external_source", enums.ExternalSourceDiscord). + Eq("external_id", "user_888")) + if identity == nil { + t.Fatalf("expected customer identity to be created") + } + + // Verify conversation + conv := repositories.ConversationRepository.FindOne(sqls.DB(), sqls.NewCnd(). + Eq("customer_id", identity.CustomerID). + Eq("channel_id", channel.ID)) + if conv == nil { + t.Fatalf("expected conversation to be created") + } + + // Verify image message created from attachment + custMsg := repositories.MessageRepository.FindOne(sqls.DB(), sqls.NewCnd(). + Eq("conversation_id", conv.ID). + Eq("sender_type", enums.IMSenderTypeCustomer)) + if custMsg == nil { + t.Fatalf("expected customer message to be created") + } + + operator := &dto.AuthPrincipal{UserID: 1, Nickname: "Agent Joy"} + + // Test Outbound enqueue with Message + replyMsg, err := MessageService.SendAIMessage(conv.ID, aiAgent.ID, "ai_msg_1", enums.IMMessageTypeText, "Here is your response image: https://example.com/response_img.png", "", operator) + if err != nil { + t.Fatalf("MessageService.SendAIMessage failed: %v", err) + } + + outbox := ChannelMessageOutboxService.GetByMessageID(enums.ChannelTypeDiscord, replyMsg.ID) + if outbox == nil { + t.Fatalf("expected outbox entry for discord message") + } + if outbox.SendStatus != string(enums.ChannelMessageOutboxStatusPending) && outbox.SendStatus != string(enums.ChannelMessageOutboxStatusSending) && outbox.SendStatus != string(enums.ChannelMessageOutboxStatusFailed) { + t.Fatalf("unexpected outbox status: %s", outbox.SendStatus) + } +} + +func seedDiscordScopedChannel(t *testing.T, db *gorm.DB, channelID string, cfg dto.DiscordChannelConfig) *models.Channel { + t.Helper() + now := time.Now() + agent := &models.AIAgent{ + Name: "Support AI", + Status: enums.StatusOk, + PublishedRevisionID: 1, + AuditFields: models.AuditFields{CreatedAt: now, UpdatedAt: now}, + } + if err := db.Create(agent).Error; err != nil { + t.Fatalf("create ai agent: %v", err) + } + cfg.BotToken = "discord_bot_token" + cfg.WebhookSecret = "scope_secret" + cfgBytes, err := json.Marshal(cfg) + if err != nil { + t.Fatalf("marshal discord config: %v", err) + } + channel := &models.Channel{ + ChannelType: enums.ChannelTypeDiscord, + ChannelID: channelID, + AIAgentID: agent.ID, + AIAgentRolloutPercent: 100, + Name: "Discord " + channelID, + ConfigJSON: string(cfgBytes), + Status: enums.StatusOk, + AuditFields: models.AuditFields{CreatedAt: now, UpdatedAt: now}, + } + if err := db.Create(channel).Error; err != nil { + t.Fatalf("create discord channel: %v", err) + } + return channel +} + +func discordScopedPayload(t *testing.T, guildID, messageID string) []byte { + t.Helper() + body := map[string]any{ + "id": messageID, + "channel_id": "discord_text_chan", + "content": "hello from discord", + "author": map[string]any{"id": "user_scope", "username": "scoped_user", "bot": false}, + } + if guildID != "" { + body["guild_id"] = guildID + } + raw, err := json.Marshal(body) + if err != nil { + t.Fatalf("marshal payload: %v", err) + } + return raw +} + +// A bot can be invited to several servers, and GuildID and ChannelScope are +// stored on the channel. Messages outside that scope must not create a +// conversation, otherwise the stored scope is configuration that does nothing. +func TestDiscordInboundHonoursGuildScope(t *testing.T) { + db := setupDiscordTestDB(t) + + guildScoped := seedDiscordScopedChannel(t, db, "discord_guild_scoped", dto.DiscordChannelConfig{ + GuildID: "guild_in_scope", + GuildName: "In Scope", + }) + dmOnly := seedDiscordScopedChannel(t, db, "discord_dm_only", dto.DiscordChannelConfig{ + ChannelScope: "dm_only", + }) + unscoped := seedDiscordScopedChannel(t, db, "discord_unscoped", dto.DiscordChannelConfig{}) + + cases := []struct { + name string + channel *models.Channel + guildID string + wantStored bool + }{ + {"matching guild is accepted", guildScoped, "guild_in_scope", true}, + {"another guild is ignored", guildScoped, "guild_elsewhere", false}, + {"a dm is ignored by a guild scoped channel", guildScoped, "", false}, + {"a dm is accepted by a dm_only channel", dmOnly, "", true}, + {"a guild message is ignored by a dm_only channel", dmOnly, "guild_anywhere", false}, + {"an unscoped channel accepts any guild", unscoped, "guild_anywhere", true}, + {"an unscoped channel accepts a dm", unscoped, "", true}, + } + + for _, tc := range cases { + messageID := "scope_" + strings.ReplaceAll(tc.name, " ", "_") + payload := discordScopedPayload(t, tc.guildID, messageID) + + if err := DiscordInboundService.HandleWebhook(context.Background(), tc.channel.ChannelID, "scope_secret", payload); err != nil { + t.Fatalf("%s: HandleWebhook failed: %v", tc.name, err) + } + + var count int64 + if err := db.Table("t_message").Where("client_msg_id LIKE ?", "%"+messageID).Count(&count).Error; err != nil { + t.Fatalf("%s: count messages: %v", tc.name, err) + } + switch { + case tc.wantStored && count == 0: + t.Errorf("%s: expected the message to be stored", tc.name) + case !tc.wantStored && count != 0: + t.Errorf("%s: expected the message to be dropped, found %d", tc.name, count) + } + } +} + +// The webhook secret is compared in constant time, so a wrong secret of the same +// length must be rejected rather than accepted by a prefix match. +func TestDiscordInboundRejectsWrongWebhookSecret(t *testing.T) { + db := setupDiscordTestDB(t) + channel := seedDiscordScopedChannel(t, db, "discord_secret", dto.DiscordChannelConfig{}) + + payload := discordScopedPayload(t, "", "secret_msg_1") + err := DiscordInboundService.HandleWebhook(context.Background(), channel.ChannelID, "wrong_secret_value", payload) + if err == nil { + t.Fatalf("expected a wrong webhook secret to be rejected") + } + + var count int64 + if err := db.Table("t_message").Where("client_msg_id LIKE ?", "%secret_msg_1").Count(&count).Error; err != nil { + t.Fatalf("count messages: %v", err) + } + if count != 0 { + t.Fatalf("a rejected delivery stored %d messages", count) + } +} diff --git a/internal/services/discord_integration_test.go b/internal/services/discord_integration_test.go new file mode 100644 index 00000000..e470bd27 --- /dev/null +++ b/internal/services/discord_integration_test.go @@ -0,0 +1,169 @@ +package services + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "agent-desk/internal/models" + "agent-desk/internal/pkg/dto" + "agent-desk/internal/pkg/dto/request" + "agent-desk/internal/pkg/enums" + "agent-desk/internal/repositories" + + "github.com/glebarez/sqlite" + "github.com/mlogclub/simple/sqls" + "gorm.io/gorm" + "gorm.io/gorm/schema" +) + +func setupDiscordIntegrationTestDB(t *testing.T) *gorm.DB { + t.Helper() + db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{ + NamingStrategy: schema.NamingStrategy{ + TablePrefix: "t_", + SingularTable: true, + }, + }) + if err != nil { + t.Fatalf("open sqlite db: %v", err) + } + if err := db.AutoMigrate( + &models.Channel{}, + &models.ChannelMessageOutbox{}, + &models.Customer{}, + &models.CustomerIdentity{}, + &models.CustomerContact{}, + &models.Conversation{}, + &models.ConversationParticipant{}, + &models.ConversationReadState{}, + &models.ConversationInterrupt{}, + &models.ConversationEventLog{}, + &models.Message{}, + &models.AIAgent{}, + &models.AgentProfile{}, + &models.AgentTeam{}, + &models.User{}, + &models.Role{}, + &models.UserRole{}, + &models.Permission{}, + &models.RolePermission{}, + &models.UserPermission{}, + ); err != nil { + t.Fatalf("migrate discord integration test tables: %v", err) + } + sqls.SetDB(db) + return db +} + +func TestDiscordIntegrationFullFlow(t *testing.T) { + db := setupDiscordIntegrationTestDB(t) + + mockDiscordServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"id":"discord_msg_reply_999","channel_id":"ch_discord_general","content":"Cảm ơn bạn! Đội ngũ hỗ trợ sẽ kiểm tra ngay."}`)) + })) + defer mockDiscordServer.Close() + + now := time.Now() + // 1. Create AI Agent + agent := &models.AIAgent{ + Name: "Discord Support AI", + ServiceMode: enums.IMConversationServiceModeAIFirst, + PublishedRevisionID: 1, + WelcomeMessage: "Chào mừng đến với máy chủ Discord Crove Desk!", + Status: enums.StatusOk, + AuditFields: models.AuditFields{ + CreatedAt: now, + UpdatedAt: now, + }, + } + _ = db.Create(agent) + + // 2. Create Discord Channel + discordConfig, _ := json.Marshal(dto.DiscordChannelConfig{ + GuildID: "guild_987654321", + GuildName: "Crove Community Discord", + BotToken: "test-discord-bot-token-xyz", + WebhookSecret: "discord-secret-token-123", + WelcomeMessage: "Welcome to Discord Support!", + }) + + operator := &dto.AuthPrincipal{UserID: 1, Username: "admin"} + channel, err := ChannelService.CreateChannel(request.CreateChannelRequest{ + Name: "Crove Discord Support", + ChannelType: enums.ChannelTypeDiscord, + AIAgentID: agent.ID, + AIAgentRolloutPercent: 100, + ConfigJSON: string(discordConfig), + Status: int(enums.StatusOk), + }, operator) + if err != nil { + t.Fatalf("CreateChannel failed: %v", err) + } + + // 3. Simulate Inbound Discord Webhook / Gateway message from user + inboundPayload := []byte(`{ + "id": "msg_discord_user_001", + "channel_id": "ch_discord_general", + "guild_id": "guild_987654321", + "content": "Tôi muốn hỏi về cách cấu hình Custom Domain cho Email Channel trên Crove Desk", + "author": { + "id": "discord_uid_555", + "username": "gamer_joy", + "global_name": "Anh Le", + "bot": false + } + }`) + + ctx := context.Background() + err = DiscordInboundService.HandleWebhook(ctx, channel.ChannelID, "discord-secret-token-123", inboundPayload) + if err != nil { + t.Fatalf("DiscordInboundService.HandleWebhook failed: %v", err) + } + + // Verify Customer Identity + identity := repositories.CustomerIdentityRepository.FindOne(db, sqls.NewCnd(). + Eq("external_source", enums.ExternalSourceDiscord). + Eq("external_id", "discord_uid_555")) + if identity == nil { + t.Fatalf("expected customer identity for discord_uid_555") + } + + customer := repositories.CustomerRepository.Get(db, identity.CustomerID) + if customer == nil || customer.Name != "Anh Le" { + t.Fatalf("unexpected customer profile: %+v", customer) + } + + // Verify Conversation created + conv := repositories.ConversationRepository.FindOne(db, sqls.NewCnd().Eq("customer_id", customer.ID)) + if conv == nil || conv.ChannelID != channel.ID { + t.Fatalf("unexpected conversation: %+v", conv) + } + + // Verify Customer Message stored + msg := repositories.MessageRepository.FindOne(db, sqls.NewCnd(). + Eq("conversation_id", conv.ID). + Eq("sender_type", enums.IMSenderTypeCustomer)) + if msg == nil || msg.Content != "Tôi muốn hỏi về cách cấu hình Custom Domain cho Email Channel trên Crove Desk" { + t.Fatalf("unexpected stored customer message: %+v", msg) + } + + // 4. Simulate Agent / AI Reply and test Outbox Enqueue & Outbound Dispatch + replyMsg, err := MessageService.SendAIMessage(conv.ID, agent.ID, "ai_reply_001", enums.IMMessageTypeText, "Cảm ơn bạn! Đội ngũ hỗ trợ sẽ kiểm tra ngay.", "", operator) + if err != nil { + t.Fatalf("MessageService.SendAIMessage failed: %v", err) + } + + outbox := ChannelMessageOutboxService.GetByMessageID(enums.ChannelTypeDiscord, replyMsg.ID) + if outbox == nil { + t.Fatalf("expected discord outbox entry for AI message") + } + if outbox.ChannelType != enums.ChannelTypeDiscord { + t.Fatalf("expected outbox channel type 'discord', got '%s'", outbox.ChannelType) + } +} diff --git a/internal/services/discord_outbound_service.go b/internal/services/discord_outbound_service.go new file mode 100644 index 00000000..2024f8af --- /dev/null +++ b/internal/services/discord_outbound_service.go @@ -0,0 +1,231 @@ +package services + +import ( + "context" + "encoding/json" + "log/slog" + "strings" + "time" + + "agent-desk/internal/discord" + "agent-desk/internal/models" + "agent-desk/internal/pkg/config" + "agent-desk/internal/pkg/enums" + "agent-desk/internal/repositories" + "agent-desk/internal/services/storage" + "os" + + "github.com/mlogclub/simple/sqls" +) + +const ( + discordOutboxBatchSize = 20 + discordOutboxMaxRetry = 5 +) + +var DiscordOutboundService = newDiscordOutboundService() + +func newDiscordOutboundService() *discordOutboundService { + return &discordOutboundService{} +} + +type discordOutboundService struct{} + +func (s *discordOutboundService) DispatchPendingOutbox() int { + return s.doDispatchPendingOutbox(discordOutboxBatchSize) +} + +func (s *discordOutboundService) doDispatchPendingOutbox(limit int) int { + if limit <= 0 { + limit = discordOutboxBatchSize + } + items := ChannelMessageOutboxService.ListPending(enums.ChannelTypeDiscord, limit) + if len(items) == 0 { + return 0 + } + + successCount := 0 + for i := range items { + if err := s.processOutbox(items[i].ID); err != nil { + slog.Warn("process discord outbox failed", + "outbox_id", items[i].ID, + "error", err, + ) + continue + } + successCount++ + } + return successCount +} + +func (s *discordOutboundService) processOutbox(outboxID int64) error { + outbox := ChannelMessageOutboxService.Get(outboxID) + if outbox == nil { + return nil + } + if outbox.ChannelType != enums.ChannelTypeDiscord { + return nil + } + if outbox.SendStatus == string(enums.ChannelMessageOutboxStatusSent) { + return nil + } + if outbox.NextRetryAt != nil && outbox.NextRetryAt.After(time.Now()) { + return nil + } + + if err := ChannelMessageOutboxService.Updates(outbox.ID, map[string]any{ + "send_status": string(enums.ChannelMessageOutboxStatusSending), + "updated_at": time.Now(), + }); err != nil { + return err + } + + message := MessageService.Get(outbox.MessageID) + if message == nil { + return s.markOutboxFailed(outbox, "message not found") + } + conversation := ConversationService.Get(outbox.ConversationID) + if conversation == nil { + return s.markOutboxFailed(outbox, "conversation not found") + } + + channel := ChannelService.Get(conversation.ChannelID) + if channel == nil || channel.Status != enums.StatusOk { + return s.markOutboxFailed(outbox, "discord channel not found or disabled") + } + cfg, err := ChannelService.ParseDiscordChannelConfig(channel.ConfigJSON) + if err != nil { + return s.markOutboxFailed(outbox, "invalid discord channel config") + } + botToken := "" + if cfg != nil { + botToken = strings.TrimSpace(cfg.BotToken) + } + if botToken == "" { + botToken = strings.TrimSpace(config.Current().Discord.BotToken) + } + if botToken == "" { + botToken = strings.TrimSpace(os.Getenv("DISCORD_BOT_TOKEN")) + } + if botToken == "" { + return s.markOutboxFailed(outbox, "discord bot token not configured") + } + + // Resolve target Discord User ID and/or Channel ID + var discordUserID string + customerIdentity := repositories.CustomerIdentityRepository.FindOne(sqls.DB(), sqls.NewCnd(). + Eq("customer_id", conversation.CustomerID). + Eq("external_source", enums.ExternalSourceDiscord)) + if customerIdentity != nil { + discordUserID = strings.TrimSpace(customerIdentity.ExternalID) + } + + // Check if there is a discord_channel_id in last message payload + var targetChannelID string + lastCustomerMsg := repositories.MessageRepository.FindOne(sqls.DB(), sqls.NewCnd(). + Eq("conversation_id", conversation.ID). + Eq("sender_type", enums.IMSenderTypeCustomer). + Desc("id")) + if lastCustomerMsg != nil && lastCustomerMsg.Payload != "" { + var payloadMap map[string]any + if err := json.Unmarshal([]byte(lastCustomerMsg.Payload), &payloadMap); err == nil { + if chID, ok := payloadMap["discord_channel_id"].(string); ok && chID != "" { + targetChannelID = chID + } + } + } + + client := discord.NewClient(botToken) + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + if targetChannelID == "" { + if discordUserID == "" { + return s.markOutboxFailed(outbox, "unable to resolve discord target user or channel") + } + dmChannel, err := client.CreateDMChannel(ctx, discordUserID) + if err != nil { + return s.markOutboxFailed(outbox, "create discord dm channel failed: "+err.Error()) + } + targetChannelID = dmChannel.ID + } + + var sendErr error + if message.MessageType == enums.IMMessageTypeImage { + assetPayload, err := parseIMMessageAssetPayload(message.Payload) + var imageURL string + if err == nil && assetPayload != nil { + assetPayload = hydrateIMMessageAssetPayload(assetPayload) + if assetPayload.Provider != "" && assetPayload.StorageKey != "" { + if provider, err := storage.NewProvider(assetPayload.Provider); err == nil { + imageURL = provider.GetSignedURL(assetPayload.StorageKey) + } + } + } + if imageURL == "" && strings.HasPrefix(strings.TrimSpace(message.Content), "http") { + imageURL = strings.TrimSpace(message.Content) + } + + if imageURL != "" { + embed := discord.Embed{ + Title: "Image Attachment", + Image: &discord.EmbedMedia{URL: imageURL}, + } + _, sendErr = client.SendEmbedMessage(ctx, targetChannelID, message.Content, []discord.Embed{embed}) + } else { + _, sendErr = client.SendMessage(ctx, targetChannelID, message.Content) + } + } else if message.MessageType == enums.IMMessageTypeAttachment { + assetPayload, err := parseIMMessageAssetPayload(message.Payload) + var fileURL string + if err == nil && assetPayload != nil { + assetPayload = hydrateIMMessageAssetPayload(assetPayload) + if assetPayload.Provider != "" && assetPayload.StorageKey != "" { + if provider, err := storage.NewProvider(assetPayload.Provider); err == nil { + fileURL = provider.GetSignedURL(assetPayload.StorageKey) + } + } + } + textToSend := message.Content + if fileURL != "" { + if textToSend != "" { + textToSend += "\n" + fileURL + } else { + textToSend = fileURL + } + } + _, sendErr = client.SendMessage(ctx, targetChannelID, textToSend) + } else { + _, sendErr = client.SendMessage(ctx, targetChannelID, message.Content) + } + + if sendErr != nil { + return s.markOutboxFailed(outbox, sendErr.Error()) + } + + return ChannelMessageOutboxService.Updates(outbox.ID, map[string]any{ + "send_status": string(enums.ChannelMessageOutboxStatusSent), + "sent_at": time.Now(), + "updated_at": time.Now(), + }) +} + +func (s *discordOutboundService) markOutboxFailed(outbox *models.ChannelMessageOutbox, errMsg string) error { + if outbox == nil { + return nil + } + retryCount := outbox.RetryCount + 1 + status := string(enums.ChannelMessageOutboxStatusFailed) + if retryCount >= discordOutboxMaxRetry { + status = string(enums.ChannelMessageOutboxStatusIgnored) + } + nextRetryAt := time.Now().Add(time.Duration(retryCount*30) * time.Second) + + return ChannelMessageOutboxService.Updates(outbox.ID, map[string]any{ + "send_status": status, + "retry_count": retryCount, + "next_retry_at": &nextRetryAt, + "last_error": errMsg, + "updated_at": time.Now(), + }) +} diff --git a/internal/services/message_service.go b/internal/services/message_service.go index 48464472..b892b182 100644 --- a/internal/services/message_service.go +++ b/internal/services/message_service.go @@ -559,6 +559,15 @@ func (s *messageService) sendValidatedMessage(conversation *models.Conversation, ) } + // Discord 渠道消息入队，异步发送 + if enqueueErr := ChannelMessageOutboxService.EnqueueDiscordMessage(conversation, message); enqueueErr != nil { + slog.Error("enqueue discord outbox failed", + "conversation_id", conversation.ID, + "message_id", message.ID, + "error", enqueueErr, + ) + } + // 客户发送消息，触发AI回复 if senderType == enums.IMSenderTypeCustomer { if TriggerAIReplyAsyncHook != nil { diff --git a/web/app/(dashboard)/dashboard/channels/_components/edit.tsx b/web/app/(dashboard)/dashboard/channels/_components/edit.tsx index c5b96168..26ff48b7 100644 --- a/web/app/(dashboard)/dashboard/channels/_components/edit.tsx +++ b/web/app/(dashboard)/dashboard/channels/_components/edit.tsx @@ -73,6 +73,13 @@ type ZaloOAChannelConfig = { webhookSecret?: string } +type DiscordChannelConfig = { + guildId?: string + guildName?: string + botToken?: string + webhookSecret?: string +} + function getDefaultWebChannelConfig(t: Translate): Required { return { title: t("channel.defaultTitleWeb"), @@ -87,7 +94,7 @@ function getDefaultWebChannelConfig(t: Translate): Required { function createSchema(t: Translate) { return z .object({ - channelType: z.enum(["web", "wechat_mp", "wxwork_kf", "telegram", "zalo_oa"], t("channel.typeRequired")), + channelType: z.enum(["web", "wechat_mp", "wxwork_kf", "telegram", "zalo_oa", "discord"], t("channel.typeRequired")), aiAgentId: z.string().trim().regex(/^\d+$/, t("channel.agentRequired")), aiAgentRolloutPercent: z.coerce.number().int().min(1).max(100), name: z.string().trim().min(1, t("channel.nameRequired")), @@ -99,6 +106,9 @@ function createSchema(t: Translate) { zaloOaId: z.string().trim(), zaloAccessToken: z.string().trim(), zaloSecretKey: z.string().trim(), + discordGuildId: z.string().trim(), + discordGuildName: z.string().trim(), + discordBotToken: z.string().trim(), widgetTitle: z.string().trim(), widgetSubtitle: z.string().trim(), widgetThemeColor: z.string().trim(), @@ -133,7 +143,7 @@ function createSchema(t: Translate) { } type EditForm = { - channelType: "web" | "wechat_mp" | "wxwork_kf" | "telegram" | "zalo_oa" + channelType: "web" | "wechat_mp" | "wxwork_kf" | "telegram" | "zalo_oa" | "discord" aiAgentId: string aiAgentRolloutPercent: number name: string @@ -145,6 +155,9 @@ type EditForm = { zaloOaId: string zaloAccessToken: string zaloSecretKey: string + discordGuildId: string + discordGuildName: string + discordBotToken: string widgetTitle: string widgetSubtitle: string widgetThemeColor: string @@ -169,6 +182,9 @@ function createEmptyForm(t: Translate): EditForm { zaloOaId: "", zaloAccessToken: "", zaloSecretKey: "", + discordGuildId: "", + discordGuildName: "", + discordBotToken: "", widgetTitle: defaultWebChannelConfig.title, widgetSubtitle: defaultWebChannelConfig.subtitle, widgetThemeColor: defaultWebChannelConfig.themeColor, @@ -222,6 +238,21 @@ function parseZaloOAChannelConfig(configJson: string): ZaloOAChannelConfig { } } +function parseDiscordChannelConfig(configJson: string): DiscordChannelConfig { + if (!configJson.trim()) return {} + try { + const parsed = JSON.parse(configJson) as DiscordChannelConfig + return { + guildId: parsed.guildId?.trim() || "", + guildName: parsed.guildName?.trim() || "", + botToken: parsed.botToken?.trim() || "", + webhookSecret: parsed.webhookSecret?.trim() || "", + } + } catch { + return {} + } +} + function parseWebChannelConfig(configJson: string, t: Translate): Required { const defaultWebChannelConfig = getDefaultWebChannelConfig(t) if (!configJson.trim()) { @@ -276,6 +307,7 @@ function buildForm(item: AdminChannel | null, t: Translate): EditForm { const isWechatMP = item.channelType === "wechat_mp" const isTelegram = item.channelType === "telegram" const isZaloOA = item.channelType === "zalo_oa" + const isDiscord = item.channelType === "discord" const webConfig = parseWebChannelConfig(item.configJson, t) const wechatConfig = isWechatMP ? parseWechatMPChannelConfig(item.configJson, t) @@ -286,6 +318,9 @@ function buildForm(item: AdminChannel | null, t: Translate): EditForm { const zaloConfig = isZaloOA ? parseZaloOAChannelConfig(item.configJson) : null + const discordConfig = isDiscord + ? parseDiscordChannelConfig(item.configJson) + : null return { channelType: item.channelType === "wxwork_kf" @@ -294,20 +329,29 @@ function buildForm(item: AdminChannel | null, t: Translate): EditForm { ? "telegram" : item.channelType === "zalo_oa" ? "zalo_oa" - : item.channelType === "wechat_mp" - ? "wechat_mp" - : "web", + : item.channelType === "discord" + ? "discord" + : item.channelType === "wechat_mp" + ? "wechat_mp" + : "web", aiAgentId: item.aiAgentId > 0 ? String(item.aiAgentId) : "", aiAgentRolloutPercent: item.aiAgentRolloutPercent || 100, name: item.name, openKfId: parseOpenKfId(item.configJson), botToken: telegramConfig?.botToken ?? "", botUsername: telegramConfig?.botUsername ?? "", - webhookSecret: telegramConfig?.webhookSecret ?? zaloConfig?.webhookSecret ?? "", + webhookSecret: + telegramConfig?.webhookSecret ?? + zaloConfig?.webhookSecret ?? + discordConfig?.webhookSecret ?? + "", zaloAppId: zaloConfig?.appId ?? "", zaloOaId: zaloConfig?.oaId ?? "", zaloAccessToken: zaloConfig?.accessToken ?? "", zaloSecretKey: zaloConfig?.secretKey ?? "", + discordGuildId: discordConfig?.guildId ?? "", + discordGuildName: discordConfig?.guildName ?? "", + discordBotToken: discordConfig?.botToken ?? "", widgetTitle: wechatConfig?.title ?? webConfig.title, widgetSubtitle: wechatConfig?.subtitle ?? webConfig.subtitle, widgetThemeColor: wechatConfig?.themeColor ?? webConfig.themeColor, @@ -347,14 +391,21 @@ function buildPayload(form: EditForm, status: number, t: Translate): CreateAdmin secretKey: form.zaloSecretKey.trim(), webhookSecret: form.webhookSecret.trim(), }) - : channelType === "wechat_mp" - ? JSON.stringify(webLikeConfig) - : JSON.stringify({ - ...webLikeConfig, - position: form.widgetPosition || defaultWebChannelConfig.position, - width: form.widgetWidth.trim() || defaultWebChannelConfig.width, - userTokenSecret: form.userTokenSecret.trim(), + : channelType === "discord" + ? JSON.stringify({ + guildId: form.discordGuildId.trim(), + guildName: form.discordGuildName.trim(), + botToken: form.discordBotToken.trim(), + webhookSecret: form.webhookSecret.trim(), }) + : channelType === "wechat_mp" + ? JSON.stringify(webLikeConfig) + : JSON.stringify({ + ...webLikeConfig, + position: form.widgetPosition || defaultWebChannelConfig.position, + width: form.widgetWidth.trim() || defaultWebChannelConfig.width, + userTokenSecret: form.userTokenSecret.trim(), + }) return { channelType, aiAgentId: Number(form.aiAgentId), @@ -543,6 +594,7 @@ function ChannelFormBody({ const channelTypeOptions = [ { value: "web", label: t("channel.typeWeb") }, { value: "telegram", label: t("channel.typeTelegram") }, + { value: "discord", label: t("channel.typeDiscord") }, { value: "wechat_mp", label: t("channel.typeWechatMp") }, { value: "wxwork_kf", label: t("channel.typeWxworkKf") }, ] as const @@ -765,6 +817,57 @@ function ChannelFormBody({ ) : null} + {channelType === "discord" ? ( +
+
+ + {t("channel.discordGuildId")} + + + + + + + + {t("channel.discordGuildName")} + + + + + +
+ + + {t("channel.discordBotToken")} + + + + + + +
+
{t("channel.discordSetupTitle")}
+
{t("channel.discordSetupDescription")}
+
+ {t("channel.inboundWebhookUrl")}: /api/third/discord/webhook +
+
+
+ ) : null} + {channelType === "telegram" ? (
diff --git a/web/app/(dashboard)/dashboard/channels/page.tsx b/web/app/(dashboard)/dashboard/channels/page.tsx index aa8c06f7..4ecbc8de 100644 --- a/web/app/(dashboard)/dashboard/channels/page.tsx +++ b/web/app/(dashboard)/dashboard/channels/page.tsx @@ -2,6 +2,7 @@ import { Building2Icon, + Gamepad2Icon, MessagesSquareIcon, MessageSquareMoreIcon, SendIcon, @@ -39,6 +40,9 @@ function getChannelTypeLabel(channelType: string, t: (key: string) => string) { if (channelType === "zalo_oa") { return t("channel.typeZaloOa") } + if (channelType === "discord") { + return t("channel.typeDiscord") + } return t("channel.typeWeb") } @@ -62,6 +66,9 @@ function ChannelIcon({ channelType }: { channelType: string }) { if (channelType === "telegram" || channelType === "zalo_oa") { return } + if (channelType === "discord") { + return + } return } @@ -78,6 +85,7 @@ export default function DashboardChannelsPage() { { value: "all", label: t("channel.allTypes") }, { value: "web", label: t("channel.typeWeb") }, { value: "telegram", label: t("channel.typeTelegram") }, + { value: "discord", label: t("channel.typeDiscord") }, { value: "zalo_oa", label: t("channel.typeZaloOa") }, { value: "wechat_mp", label: t("channel.typeWechatMp") }, { value: "wxwork_kf", label: t("channel.typeWxworkKf") }, diff --git a/web/messages/en-US.json b/web/messages/en-US.json index c83bc000..c961bf00 100644 --- a/web/messages/en-US.json +++ b/web/messages/en-US.json @@ -624,6 +624,13 @@ "zaloAppId": "App ID", "zaloAutoConnectTitle": "Zalo Official Account Connection", "zaloAutoConnectDescription": "Connect your Zalo Official Account using the Access Token to automatically receive customer messages and dispatch replies.", + "typeDiscord": "Discord", + "discordGuildId": "Guild / Server ID", + "discordGuildName": "Guild / Server Name", + "discordBotToken": "Bot Token", + "discordSetupTitle": "Discord Bot Connection", + "discordSetupDescription": "Create a bot in the Discord Developer Portal, enable the Message Content privileged intent, invite it to your server, and point an interaction or bridge endpoint at the webhook URL below. Leave the bot token empty to use the deployment-wide DISCORD_BOT_TOKEN.", + "inboundWebhookUrl": "Inbound Webhook Endpoint", "loadFailed": "Could not load channels.", "created": "Channel created: {name}", "updated": "Channel updated: {name}", diff --git a/web/messages/zh-CN.json b/web/messages/zh-CN.json index 14e67ae3..5472359b 100644 --- a/web/messages/zh-CN.json +++ b/web/messages/zh-CN.json @@ -624,6 +624,13 @@ "zaloAppId": "App ID", "zaloAutoConnectTitle": "Zalo OA 渠道连接", "zaloAutoConnectDescription": "输入 Zalo OA 的 Access Token 即可自动双向同步客户会话与消息。", + "typeDiscord": "Discord", + "discordGuildId": "服务器 ID", + "discordGuildName": "服务器名称", + "discordBotToken": "Bot Token", + "discordSetupTitle": "Discord 机器人接入", + "discordSetupDescription": "在 Discord Developer Portal 创建机器人，开启 Message Content 特权意图，邀请机器人进入服务器，并将交互或转发端点指向下方的 Webhook 地址。Bot Token 留空则使用部署级 DISCORD_BOT_TOKEN。", + "inboundWebhookUrl": "入站 Webhook 地址", "loadFailed": "加载接入渠道失败", "created": "已创建接入渠道：{name}", "updated": "已更新接入渠道：{name}",