package services import ( "code.tczkiot.com/wlw/ai-agent/internal/models" "code.tczkiot.com/wlw/ai-agent/internal/pkg/dto" "code.tczkiot.com/wlw/ai-agent/internal/pkg/enums" "code.tczkiot.com/wlw/ai-agent/internal/pkg/errorsx" "code.tczkiot.com/wlw/ai-agent/internal/repositories" "encoding/json" "strings" "time" "code.tczkiot.com/wlw/ai-agent/internal/pkg/httpx/params" "github.com/mlogclub/simple/sqls" ) var ChannelMessageOutboxService = newChannelMessageOutboxService() func newChannelMessageOutboxService() *channelMessageOutboxService { return &channelMessageOutboxService{} } type channelMessageOutboxService struct { } func (s *channelMessageOutboxService) Get(id int64) *models.ChannelMessageOutbox { return repositories.ChannelMessageOutboxRepository.Get(sqls.DB(), id) } func (s *channelMessageOutboxService) Take(where ...interface{}) *models.ChannelMessageOutbox { return repositories.ChannelMessageOutboxRepository.Take(sqls.DB(), where...) } func (s *channelMessageOutboxService) Find(cnd *sqls.Cnd) []models.ChannelMessageOutbox { return repositories.ChannelMessageOutboxRepository.Find(sqls.DB(), cnd) } func (s *channelMessageOutboxService) FindOne(cnd *sqls.Cnd) *models.ChannelMessageOutbox { return repositories.ChannelMessageOutboxRepository.FindOne(sqls.DB(), cnd) } func (s *channelMessageOutboxService) FindPageByParams(params *params.QueryParams) (list []models.ChannelMessageOutbox, paging *sqls.Paging) { return repositories.ChannelMessageOutboxRepository.FindPageByParams(sqls.DB(), params) } func (s *channelMessageOutboxService) FindPageByCnd(cnd *sqls.Cnd) (list []models.ChannelMessageOutbox, paging *sqls.Paging) { return repositories.ChannelMessageOutboxRepository.FindPageByCnd(sqls.DB(), cnd) } func (s *channelMessageOutboxService) Count(cnd *sqls.Cnd) int64 { return repositories.ChannelMessageOutboxRepository.Count(sqls.DB(), cnd) } func (s *channelMessageOutboxService) Create(t *models.ChannelMessageOutbox) error { return repositories.ChannelMessageOutboxRepository.Create(sqls.DB(), t) } func (s *channelMessageOutboxService) Update(t *models.ChannelMessageOutbox) error { return repositories.ChannelMessageOutboxRepository.Update(sqls.DB(), t) } func (s *channelMessageOutboxService) Updates(id int64, columns map[string]interface{}) error { return repositories.ChannelMessageOutboxRepository.Updates(sqls.DB(), id, columns) } func (s *channelMessageOutboxService) UpdateColumn(id int64, name string, value interface{}) error { return repositories.ChannelMessageOutboxRepository.UpdateColumn(sqls.DB(), id, name, value) } func (s *channelMessageOutboxService) Delete(id int64) { repositories.ChannelMessageOutboxRepository.Delete(sqls.DB(), id) } // GetByMessageID retrieves the outbox entry by message ID and channel type. func (s *channelMessageOutboxService) GetByMessageID(channelType string, messageID int64) *models.ChannelMessageOutbox { return repositories.ChannelMessageOutboxRepository.Take(sqls.DB(), "channel_type = ? AND message_id = ?", channelType, messageID) } func (s *channelMessageOutboxService) EnqueueWxWorkKFMessage(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.ChannelTypeWxWorkKF { return nil } if message.SenderType != enums.IMSenderTypeAgent && message.SenderType != enums.IMSenderTypeAI { return nil } if message.MessageType != enums.IMMessageTypeText && message.MessageType != enums.IMMessageTypeHTML { return nil } if existing := s.GetByMessageID(enums.ChannelTypeWxWorkKF, message.ID); existing != nil { return nil } payload, err := json.Marshal(map[string]any{ "conversation_id": conversation.ID, "message_id": message.ID, "message_type": message.MessageType, "content": strings.TrimSpace(message.Content), "payload": strings.TrimSpace(message.Payload), "sender_id": message.SenderID, }) if err != nil { return err } now := time.Now() return s.Create(&models.ChannelMessageOutbox{ ChannelType: enums.ChannelTypeWxWorkKF, 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, }, }) } func (s *channelMessageOutboxService) ListPending(channelType string, limit int) []models.ChannelMessageOutbox { if limit <= 0 { limit = 20 } cnd := sqls.NewCnd(). Eq("channel_type", strings.TrimSpace(channelType)). In("send_status", []string{ string(enums.ChannelMessageOutboxStatusPending), string(enums.ChannelMessageOutboxStatusFailed), }). Asc("id"). Limit(limit) return s.Find(cnd) } func (s *channelMessageOutboxService) RetryWxWorkFailure(id int64, operator *dto.AuthPrincipal) error { item := s.Get(id) if item == nil || item.ChannelType != enums.ChannelTypeWxWorkKF { return errorsx.InvalidParam("outbox record does not exist") } if item.SendStatus != string(enums.ChannelMessageOutboxStatusFailed) && item.SendStatus != string(enums.ChannelMessageOutboxStatusIgnored) { return errorsx.InvalidParam("only failed or ignored outbox records can be retried") } now := time.Now() columns := map[string]interface{}{ "send_status": string(enums.ChannelMessageOutboxStatusPending), "next_retry_at": nil, "updated_at": now, } if operator != nil { columns["update_user_id"] = operator.UserID columns["update_user_name"] = operator.Username } return s.Updates(id, columns) } func (s *channelMessageOutboxService) IgnoreWxWorkFailure(id int64, operator *dto.AuthPrincipal) error { item := s.Get(id) if item == nil || item.ChannelType != enums.ChannelTypeWxWorkKF { return errorsx.InvalidParam("outbox record does not exist") } if item.SendStatus != string(enums.ChannelMessageOutboxStatusFailed) { return errorsx.InvalidParam("only failed outbox records can be ignored") } now := time.Now() columns := map[string]interface{}{ "send_status": string(enums.ChannelMessageOutboxStatusIgnored), "next_retry_at": nil, "updated_at": now, } if operator != nil { columns["update_user_id"] = operator.UserID columns["update_user_name"] = operator.Username } return s.Updates(id, columns) }