lislgosms/gateway/internal/inbound/acknowledgement.go

package inbound

import (
	"fmt"
	cmpp "github.com/bigwhite/gocmpp"
	"log"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"
)

// The Submit barrier is intentionally colocated with ACK tracking: a queued
// receipt must never overtake the SubmitResp that establishes its Msg_Id.

const defaultDownstreamAckTimeout = 30 * time.Second

type downstreamAckTracker struct {
	deliveryID   string
	claimID      string
	connectionID string
	sequenceID   uint32
	messageID    uint64
	session      *downstreamSession
	timer        *time.Timer
}

var downstreamAckRegistry = struct {
	sync.Mutex
	items map[string]*downstreamAckTracker
}{items: make(map[string]*downstreamAckTracker)}

var downstreamSubmitBarrier = struct {
	sync.RWMutex
	byConn map[*cmpp.Conn]int
}{byConn: make(map[*cmpp.Conn]int)}

func beginDownstreamSubmitBarrier(conn *cmpp.Conn) func() {
	if conn == nil {
		return func() {}
	}
	downstreamSubmitBarrier.Lock()
	downstreamSubmitBarrier.byConn[conn]++
	downstreamSubmitBarrier.Unlock()
	var once sync.Once
	return func() {
		once.Do(func() {
			downstreamSubmitBarrier.Lock()
			if downstreamSubmitBarrier.byConn[conn] <= 1 {
				delete(downstreamSubmitBarrier.byConn, conn)
			} else {
				downstreamSubmitBarrier.byConn[conn]--
			}
			downstreamSubmitBarrier.Unlock()
		})
	}
}

func downstreamSubmitResponsePending(conn *cmpp.Conn) bool {
	if conn == nil {
		return false
	}
	downstreamSubmitBarrier.RLock()
	defer downstreamSubmitBarrier.RUnlock()
	return downstreamSubmitBarrier.byConn[conn] > 0
}

func downstreamAckKey(conn *cmpp.Conn, sequenceID uint32) string {
	return fmt.Sprintf("%p:%d", conn, sequenceID)
}

func registerDownstreamAck(session *downstreamSession, deliveryID string, claimID string, sequenceID uint32, messageID uint64, deadline time.Time) *downstreamAckTracker {
	if session == nil || session.conn == nil || strings.TrimSpace(deliveryID) == "" {
		return nil
	}
	tracker := &downstreamAckTracker{
		deliveryID: deliveryID, claimID: claimID, connectionID: session.connectionID,
		sequenceID: sequenceID, messageID: messageID, session: session,
	}
	key := downstreamAckKey(session.conn, sequenceID)
	downstreamAckRegistry.Lock()
	downstreamAckRegistry.items[key] = tracker
	tracker.timer = time.AfterFunc(time.Until(deadline), func() {
		timedOut := takeDownstreamAck(session.conn, sequenceID)
		if timedOut == nil || timedOut.session == nil || timedOut.session.deliveryReport == nil {
			return
		}
		timedOut.session.deliveryReport(downstreamDeliveryLifecycleEvent{
			Kind: "failed", DeliveryID: timedOut.deliveryID, ClaimID: timedOut.claimID, ConnectionID: timedOut.connectionID,
			SequenceID: timedOut.sequenceID, MessageID: timedOut.messageID, ObservedAt: time.Now().UTC(),
			FailureType: "ack_timeout", ErrorMessage: "CMPP_DELIVER_RESP timeout",
		})
	})
	downstreamAckRegistry.Unlock()
	return tracker
}

func takeDownstreamAck(conn *cmpp.Conn, sequenceID uint32) *downstreamAckTracker {
	key := downstreamAckKey(conn, sequenceID)
	downstreamAckRegistry.Lock()
	tracker := downstreamAckRegistry.items[key]
	delete(downstreamAckRegistry.items, key)
	downstreamAckRegistry.Unlock()
	if tracker != nil && tracker.timer != nil {
		tracker.timer.Stop()
	}
	return tracker
}

func removeDownstreamAck(tracker *downstreamAckTracker) {
	if tracker == nil || tracker.session == nil {
		return
	}
	_ = takeDownstreamAck(tracker.session.conn, tracker.sequenceID)
}

func handleDownstreamAcknowledgement(conn *cmpp.Conn, sequenceID uint32, messageID uint64, result uint32, logger *log.Logger) {
	tracker := takeDownstreamAck(conn, sequenceID)
	if tracker == nil {
		if logger != nil {
			logger.Printf("cmpp inbound event=deliver_ack_unmatched seq=%d message_id=%d result=%d", sequenceID, messageID, result)
		}
		if session := findSessionByConn(conn); session != nil && session.protocolLog != nil {
			session.protocolLog(protocolLogEvent{
				Protocol:         "cmpp",
				Direction:        "client_to_platform",
				EventType:        "deliver_resp",
				Status:           "failed",
				TenantID:         session.tenantID,
				ApplicationID:    session.applicationID,
				Account:          session.account,
				MessageID:        session.messageID,
				GatewayMessageID: strconv.FormatUint(messageID, 10),
				Phone:            session.phoneNumber,
				ResultCode:       strconv.FormatUint(uint64(result), 10),
				Detail:           map[string]any{"sequenceId": sequenceID, "unmatched": true},
			})
		}
		return
	}
	if tracker.messageID != messageID {
		if logger != nil {
			logger.Printf("cmpp inbound event=deliver_ack_message_mismatch delivery_id=%s seq=%d expected_message_id=%d actual_message_id=%d", tracker.deliveryID, sequenceID, tracker.messageID, messageID)
		}
		result = 1
	}
	if logger != nil {
		logger.Printf("cmpp inbound event=deliver_acknowledged delivery_id=%s connection_id=%s seq=%d message_id=%d result=%d", tracker.deliveryID, tracker.connectionID, sequenceID, messageID, result)
	}
	if tracker.session != nil && tracker.session.deliveryReport != nil {
		go tracker.session.deliveryReport(downstreamDeliveryLifecycleEvent{
			Kind: "acknowledged", DeliveryID: tracker.deliveryID, ClaimID: tracker.claimID, ConnectionID: tracker.connectionID,
			SequenceID: sequenceID, MessageID: messageID, Result: result, ObservedAt: time.Now().UTC(),
		})
	}
	if tracker.session != nil && tracker.session.protocolLog != nil {
		status := "success"
		if result != 0 {
			status = "failed"
		}
		tracker.session.protocolLog(protocolLogEvent{
			Protocol:         "cmpp",
			Direction:        "client_to_platform",
			EventType:        "deliver_resp",
			Status:           status,
			TenantID:         tracker.session.tenantID,
			ApplicationID:    tracker.session.applicationID,
			Account:          tracker.session.account,
			MessageID:        tracker.session.messageID,
			GatewayMessageID: strconv.FormatUint(messageID, 10),
			Phone:            tracker.session.phoneNumber,
			ResultCode:       strconv.FormatUint(uint64(result), 10),
			Detail:           map[string]any{"sequenceId": sequenceID, "deliveryId": tracker.deliveryID},
		})
	}
}

func downstreamAckTimeout() time.Duration {
	configured, err := strconv.Atoi(strings.TrimSpace(os.Getenv("CMPP_DOWNSTREAM_ACK_TIMEOUT_SECONDS")))
	if err != nil || configured <= 0 {
		return defaultDownstreamAckTimeout
	}
	if configured < 5 {
		configured = 5
	}
	return time.Duration(configured) * time.Second
}