package notificationimp import ( "context" "fmt" "regexp" "strconv" "strings" "sync" "time" "github.com/tech/sendico/notification/internal/server/notificationimp/telegram" "github.com/tech/sendico/pkg/merrors" msg "github.com/tech/sendico/pkg/messaging" confirmations "github.com/tech/sendico/pkg/messaging/notifications/confirmations" "github.com/tech/sendico/pkg/mlogger" "github.com/tech/sendico/pkg/model" "github.com/tech/sendico/pkg/mservice" paymenttypes "github.com/tech/sendico/pkg/payments/types" "go.uber.org/zap" ) const ( defaultConfirmationTimeout = 120 * time.Second ) type confirmationManager struct { logger mlogger.Logger tg telegram.Client sender string outbox msg.Producer mu sync.Mutex pendingByMessage map[string]*confirmationState pendingByRequest map[string]*confirmationState } type confirmationState struct { request model.ConfirmationRequest requestMessageID string targetChatID string callbackSubject string clarified bool timer *time.Timer } func newConfirmationManager(logger mlogger.Logger, tg telegram.Client, outbox msg.Producer) *confirmationManager { if logger != nil { logger = logger.Named("confirmations") } return &confirmationManager{ logger: logger, tg: tg, outbox: outbox, sender: string(mservice.Notifications), pendingByMessage: map[string]*confirmationState{}, pendingByRequest: map[string]*confirmationState{}, } } func (m *confirmationManager) logDebug(message string, fields ...zap.Field) { if m == nil || m.logger == nil { return } m.logger.Debug(message, fields...) } func (m *confirmationManager) logInfo(message string, fields ...zap.Field) { if m == nil || m.logger == nil { return } m.logger.Info(message, fields...) } func (m *confirmationManager) logWarn(message string, fields ...zap.Field) { if m == nil || m.logger == nil { return } m.logger.Warn(message, fields...) } func (m *confirmationManager) Stop() { if m == nil { return } m.logInfo("Stopping confirmation manager") m.mu.Lock() defer m.mu.Unlock() pending := len(m.pendingByMessage) m.logDebug("Stopping pending confirmation timers", zap.Int("pending_count", pending)) for _, state := range m.pendingByMessage { if state.timer != nil { if !state.timer.Stop() { m.logDebug("Confirmation timer already fired while stopping", zap.String("request_id", state.request.RequestID), zap.String("message_id", state.requestMessageID)) } } } m.pendingByMessage = map[string]*confirmationState{} m.pendingByRequest = map[string]*confirmationState{} m.logInfo("Confirmation manager stopped", zap.Int("pending_cleared", pending)) } func (m *confirmationManager) HandleRequest(ctx context.Context, request *model.ConfirmationRequest) error { if m == nil { return merrors.Internal("confirmation manager is nil") } m.logDebug("Handling confirmation request", zap.Bool("request_nil", request == nil)) if request == nil { m.logWarn("Confirmation request rejected: request is nil") return merrors.InvalidArgument("confirmation request is nil", "request") } if m.tg == nil { m.logWarn("Confirmation request rejected: telegram client is not configured", zap.String("request_id", strings.TrimSpace(request.RequestID))) return merrors.InvalidArgument("telegram client is not configured", "telegram") } req := normalizeConfirmationRequest(*request) m.logDebug("Confirmation request normalized", zap.String("request_id", req.RequestID), zap.String("target_chat_id", req.TargetChatID), zap.String("source_service", req.SourceService), zap.String("rail", req.Rail), zap.Int("accepted_users", len(req.AcceptedUserIDs)), zap.Int32("timeout_seconds", req.TimeoutSeconds)) if req.RequestID == "" { m.logWarn("Confirmation request rejected: request_id is required", zap.String("target_chat_id", req.TargetChatID)) return merrors.InvalidArgument("confirmation request_id is required", "request_id") } if req.TargetChatID == "" { m.logWarn("Confirmation request rejected: target_chat_id is required", zap.String("request_id", req.RequestID)) return merrors.InvalidArgument("confirmation target_chat_id is required", "target_chat_id") } if req.RequestedMoney == nil || strings.TrimSpace(req.RequestedMoney.Amount) == "" || strings.TrimSpace(req.RequestedMoney.Currency) == "" { m.logWarn("Confirmation request rejected: requested_money is required", zap.String("request_id", req.RequestID)) return merrors.InvalidArgument("confirmation requested_money is required", "requested_money") } if req.SourceService == "" { m.logWarn("Confirmation request rejected: source_service is required", zap.String("request_id", req.RequestID)) return merrors.InvalidArgument("confirmation source_service is required", "source_service") } m.mu.Lock() pendingBefore := len(m.pendingByMessage) if _, ok := m.pendingByRequest[req.RequestID]; ok { m.mu.Unlock() m.logInfo("Confirmation request already pending", zap.String("request_id", req.RequestID), zap.Int("pending_count", pendingBefore)) return nil } m.mu.Unlock() m.logDebug("Confirmation request accepted for processing", zap.String("request_id", req.RequestID), zap.String("target_chat_id", req.TargetChatID), zap.Int("pending_count_before", pendingBefore)) message := confirmationPrompt(&req) m.logDebug("Sending confirmation request to Telegram", zap.String("request_id", req.RequestID), zap.String("target_chat_id", req.TargetChatID), zap.Int("prompt_length", len(message))) sent, err := m.tg.SendText(ctx, req.TargetChatID, message, "") if err != nil { m.logWarn("Failed to send confirmation request to Telegram", zap.Error(err), zap.String("request_id", req.RequestID), zap.String("target_chat_id", req.TargetChatID)) return err } if sent == nil || strings.TrimSpace(sent.MessageID) == "" { m.logWarn("Confirmation request send succeeded without message_id", zap.String("request_id", req.RequestID)) return merrors.Internal("telegram confirmation message id is missing") } m.logDebug("Confirmation request sent to Telegram", zap.String("request_id", req.RequestID), zap.String("message_id", strings.TrimSpace(sent.MessageID)), zap.String("target_chat_id", req.TargetChatID)) state := &confirmationState{ request: req, requestMessageID: strings.TrimSpace(sent.MessageID), targetChatID: strings.TrimSpace(req.TargetChatID), callbackSubject: confirmationCallbackSubject(req.SourceService, req.Rail), } timeout := time.Duration(req.TimeoutSeconds) * time.Second if timeout <= 0 { m.logDebug("Confirmation timeout not provided, using default timeout", zap.String("request_id", req.RequestID), zap.Duration("timeout", defaultConfirmationTimeout)) timeout = defaultConfirmationTimeout } m.logDebug("Scheduling confirmation timeout", zap.String("request_id", req.RequestID), zap.String("message_id", state.requestMessageID), zap.Duration("timeout", timeout)) state.timer = time.AfterFunc(timeout, func() { m.handleTimeout(state.requestMessageID) }) m.mu.Lock() m.pendingByMessage[state.requestMessageID] = state m.pendingByRequest[req.RequestID] = state pendingAfter := len(m.pendingByMessage) m.mu.Unlock() m.logInfo("Confirmation request sent", zap.String("request_id", req.RequestID), zap.String("message_id", state.requestMessageID), zap.String("callback_subject", state.callbackSubject), zap.Int("pending_count", pendingAfter)) return nil } func (m *confirmationManager) HandleUpdate(ctx context.Context, update *telegram.Update) { if m == nil { return } if update == nil { m.logDebug("Telegram update ignored: update is nil") return } if update.Message == nil { m.logDebug("Telegram update ignored: message is nil") return } message := update.Message if message.ReplyToMessage == nil { m.logDebug("Telegram update ignored: message is not a reply", zap.Int64("message_id", message.MessageID)) return } replyToID := strconv.FormatInt(message.ReplyToMessage.MessageID, 10) m.logDebug("Telegram reply received", zap.String("reply_to_message_id", replyToID)) state := m.lookupByMessageID(replyToID) if state == nil { m.logDebug("Telegram reply ignored: no pending confirmation for message", zap.String("reply_to_message_id", replyToID)) return } m.logDebug("Telegram reply matched pending confirmation", zap.String("request_id", state.request.RequestID), zap.String("reply_to_message_id", replyToID)) chatID := strconv.FormatInt(message.Chat.ID, 10) if chatID != state.targetChatID { m.logDebug("Telegram reply ignored: chat mismatch", zap.String("request_id", state.request.RequestID), zap.String("expected_chat_id", state.targetChatID), zap.String("chat_id", chatID)) return } rawReply := message.ToModel() if !state.isUserAllowed(message.From) { userID := "" if message.From != nil { userID = strconv.FormatInt(message.From.ID, 10) } m.logWarn("Telegram reply rejected: unauthorized user", zap.String("request_id", state.request.RequestID), zap.String("user_id", userID), zap.String("chat_id", chatID)) m.publishResult(state, &model.ConfirmationResult{ RequestID: state.request.RequestID, Status: model.ConfirmationStatusRejected, ParseError: "unauthorized_user", RawReply: rawReply, }) m.sendNotice(ctx, state, rawReply, "Only approved users can confirm this payment.") m.removeState(state.requestMessageID) return } m.logDebug("Telegram reply accepted from authorized user", zap.String("request_id", state.request.RequestID)) money, reason, err := parseConfirmationReply(message.Text) if err != nil { m.logInfo("Telegram reply requires clarification", zap.String("request_id", state.request.RequestID), zap.String("reason", reason), zap.Error(err)) m.mu.Lock() state.clarified = true m.mu.Unlock() m.sendNotice(ctx, state, rawReply, clarificationMessage(reason)) return } m.mu.Lock() clarified := state.clarified m.mu.Unlock() status := model.ConfirmationStatusConfirmed if clarified { status = model.ConfirmationStatusClarified } m.logInfo("Telegram confirmation parsed", zap.String("request_id", state.request.RequestID), zap.String("status", string(status)), zap.String("amount", money.Amount), zap.String("currency", money.Currency)) m.publishResult(state, &model.ConfirmationResult{ RequestID: state.request.RequestID, Money: money, RawReply: rawReply, Status: status, }) m.removeState(state.requestMessageID) } func (m *confirmationManager) lookupByMessageID(messageID string) *confirmationState { if m == nil { return nil } messageID = strings.TrimSpace(messageID) if messageID == "" { m.logDebug("Pending confirmation lookup skipped: message_id is empty") return nil } m.mu.Lock() state := m.pendingByMessage[messageID] pendingCount := len(m.pendingByMessage) m.mu.Unlock() if state == nil { m.logDebug("Pending confirmation not found", zap.String("message_id", messageID), zap.Int("pending_count", pendingCount)) return nil } m.logDebug("Pending confirmation found", zap.String("message_id", messageID), zap.String("request_id", state.request.RequestID), zap.Int("pending_count", pendingCount)) return state } func (m *confirmationManager) handleTimeout(messageID string) { if m == nil { return } messageID = strings.TrimSpace(messageID) m.logInfo("Confirmation timeout triggered", zap.String("message_id", messageID)) state := m.lookupByMessageID(messageID) if state == nil { m.logDebug("Confirmation timeout ignored: state not found", zap.String("message_id", messageID)) return } m.logInfo("Publishing timeout confirmation result", zap.String("request_id", state.request.RequestID), zap.String("message_id", messageID)) m.publishResult(state, &model.ConfirmationResult{ RequestID: state.request.RequestID, Status: model.ConfirmationStatusTimeout, }) m.removeState(messageID) } func (m *confirmationManager) removeState(messageID string) { if m == nil { return } messageID = strings.TrimSpace(messageID) if messageID == "" { m.logDebug("State removal skipped: message_id is empty") return } m.mu.Lock() state := m.pendingByMessage[messageID] if state != nil && state.timer != nil { if !state.timer.Stop() { m.logDebug("Confirmation timer already fired before state removal", zap.String("message_id", messageID), zap.String("request_id", state.request.RequestID)) } } delete(m.pendingByMessage, messageID) if state != nil { delete(m.pendingByRequest, state.request.RequestID) } pendingCount := len(m.pendingByMessage) m.mu.Unlock() if state == nil { m.logDebug("State removal skipped: no state for message", zap.String("message_id", messageID), zap.Int("pending_count", pendingCount)) return } m.logInfo("Confirmation state removed", zap.String("message_id", messageID), zap.String("request_id", state.request.RequestID), zap.Int("pending_count", pendingCount)) } func (m *confirmationManager) publishResult(state *confirmationState, result *model.ConfirmationResult) { if m == nil || state == nil || result == nil { m.logDebug("Confirmation result publish skipped: missing context", zap.Bool("state_nil", state == nil), zap.Bool("result_nil", result == nil)) return } m.logDebug("Publishing confirmation result", zap.String("request_id", state.request.RequestID), zap.String("status", string(result.Status)), zap.String("callback_subject", state.callbackSubject)) if m.outbox == nil { m.logWarn("Confirmation result skipped: producer not configured", zap.String("request_id", state.request.RequestID), zap.String("callback_subject", state.callbackSubject)) return } env := confirmations.ConfirmationResult(m.sender, result, state.request.SourceService, state.request.Rail) m.logDebug("Confirmation result envelope prepared", zap.String("request_id", state.request.RequestID), zap.String("sender", m.sender), zap.String("source_service", state.request.SourceService), zap.String("rail", state.request.Rail)) if err := m.outbox.SendMessage(env); err != nil { m.logWarn("Failed to publish confirmation result", zap.Error(err), zap.String("request_id", state.request.RequestID), zap.String("callback_subject", state.callbackSubject)) return } m.logInfo("Confirmation result published", zap.String("request_id", state.request.RequestID), zap.String("status", string(result.Status)), zap.String("callback_subject", state.callbackSubject)) } func (m *confirmationManager) sendNotice(ctx context.Context, state *confirmationState, reply *model.TelegramMessage, text string) { if m == nil { return } if m.tg == nil { m.logWarn("Clarification notice skipped: telegram client is not configured") return } if state == nil { m.logDebug("Clarification notice skipped: state is nil") return } replyID := "" if reply != nil { replyID = reply.MessageID } m.logDebug("Sending clarification notice", zap.String("request_id", state.request.RequestID), zap.String("target_chat_id", state.targetChatID), zap.String("reply_to_message_id", replyID), zap.Int("text_length", len(text))) if _, err := m.tg.SendText(ctx, state.targetChatID, text, replyID); err != nil { m.logWarn("Failed to send clarification notice", zap.Error(err), zap.String("request_id", state.request.RequestID)) return } m.logInfo("Clarification notice sent", zap.String("request_id", state.request.RequestID), zap.String("target_chat_id", state.targetChatID)) } func (s *confirmationState) isUserAllowed(user *telegram.User) bool { if s == nil { return false } allowed := s.request.AcceptedUserIDs if len(allowed) == 0 { return true } if user == nil { return false } userID := strconv.FormatInt(user.ID, 10) for _, id := range allowed { if id == userID { return true } } return false } func confirmationCallbackSubject(sourceService, rail string) string { sourceService = strings.ToLower(strings.TrimSpace(sourceService)) if sourceService == "" { sourceService = "unknown" } rail = strings.ToLower(strings.TrimSpace(rail)) if rail == "" { rail = "default" } return "confirmations." + sourceService + "." + rail } func normalizeConfirmationRequest(request model.ConfirmationRequest) model.ConfirmationRequest { request.RequestID = strings.TrimSpace(request.RequestID) request.TargetChatID = strings.TrimSpace(request.TargetChatID) request.PaymentIntentID = strings.TrimSpace(request.PaymentIntentID) request.QuoteRef = strings.TrimSpace(request.QuoteRef) request.SourceService = strings.TrimSpace(request.SourceService) request.Rail = strings.TrimSpace(request.Rail) request.AcceptedUserIDs = normalizeStringList(request.AcceptedUserIDs) if request.RequestedMoney != nil { request.RequestedMoney.Amount = strings.TrimSpace(request.RequestedMoney.Amount) request.RequestedMoney.Currency = strings.TrimSpace(request.RequestedMoney.Currency) } return request } func normalizeStringList(values []string) []string { if len(values) == 0 { return nil } result := make([]string, 0, len(values)) seen := map[string]struct{}{} for _, value := range values { value = strings.TrimSpace(value) if value == "" { continue } if _, ok := seen[value]; ok { continue } seen[value] = struct{}{} result = append(result, value) } if len(result) == 0 { return nil } return result } var amountPattern = regexp.MustCompile(`^[0-9]+(\.[0-9]+)?$`) var currencyPattern = regexp.MustCompile(`^[A-Za-z]{3,10}$`) func parseConfirmationReply(text string) (*paymenttypes.Money, string, error) { text = strings.TrimSpace(text) if text == "" { return nil, "empty", merrors.InvalidArgument("empty reply") } parts := strings.Fields(text) if len(parts) < 2 { if len(parts) == 1 && amountPattern.MatchString(parts[0]) { return nil, "missing_currency", merrors.InvalidArgument("currency is required") } return nil, "missing_amount", merrors.InvalidArgument("amount is required") } if len(parts) > 2 { return nil, "format", merrors.InvalidArgument("reply format is invalid") } amount := parts[0] currency := parts[1] if !amountPattern.MatchString(amount) { return nil, "invalid_amount", merrors.InvalidArgument("amount format is invalid") } if !currencyPattern.MatchString(currency) { return nil, "invalid_currency", merrors.InvalidArgument("currency format is invalid") } return &paymenttypes.Money{ Amount: amount, Currency: strings.ToUpper(currency), }, "", nil } func confirmationPrompt(req *model.ConfirmationRequest) string { var builder strings.Builder builder.WriteString("Payment confirmation required\n") if req.QuoteRef != "" { builder.WriteString("Quote ref: ") builder.WriteString(req.QuoteRef) builder.WriteString("\n") } if req.RequestedMoney != nil { amountFloat, err := strconv.ParseFloat(req.RequestedMoney.Amount, 64) if err != nil { amountFloat = 0 } amount := fmt.Sprintf("%.2f", amountFloat) builder.WriteString(fmt.Sprintf( "\n*Requested: %s %s*\n\n", amount, req.RequestedMoney.Currency, )) } builder.WriteString("Reply with \" \" (e.g., 12.34 USD).") return builder.String() } func clarificationMessage(reason string) string { switch reason { case "missing_currency": return "Currency code is required. Reply with \" \" (e.g., 12.34 USD)." case "missing_amount": return "Amount is required. Reply with \" \" (e.g., 12.34 USD)." case "invalid_amount": return "Amount must be a decimal number. Reply with \" \" (e.g., 12.34 USD)." case "invalid_currency": return "Currency must be a code like USD or EUR. Reply with \" \"." default: return "Reply with \" \" (e.g., 12.34 USD)." } }