lislgosms/gateway/internal/inbound/server.go

package inbound

import (
	"bytes"
	"context"
	"crypto/md5"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"

	cmpp "github.com/bigwhite/gocmpp"
	cmpputils "github.com/bigwhite/gocmpp/utils"
)

const defaultHTTPTimeout = 10 * time.Second
const defaultDownstreamAckTimeout = 30 * time.Second

type Server struct {
	Addr                 string
	APIBaseURL           string
	HTTPClient           *http.Client
	LogWriter            io.Writer
	PendingFlushInterval time.Duration
	PresenceStore        PresenceStore
	RecoveryStore        RecoveryStore
	GatewayInstanceID    string
}

type authRequest struct {
	Account    string `json:"account"`
	AuthSource string `json:"authSource"`
	Timestamp  uint32 `json:"timestamp"`
	RemoteIP   string `json:"remoteIp,omitempty"`
}

type submitRequest struct {
	Account     string `json:"account"`
	PhoneNumber string `json:"phoneNumber"`
	Content     string `json:"content"`
	SrcID       string `json:"srcId,omitempty"`
	DestID      string `json:"destId,omitempty"`
	SequenceID  uint32 `json:"sequenceId,omitempty"`
	RemoteIP    string `json:"remoteIp,omitempty"`
}

type submitResponse struct {
	Accepted  bool   `json:"accepted"`
	MessageID string `json:"messageId"`
}

type authResponse struct {
	PasswordCipher string `json:"passwordCipher"`
	ApplicationID  string `json:"applicationId"`
	TenantID       string `json:"tenantId"`
	Account        string `json:"account"`
	EnterpriseCode string `json:"enterpriseCode"`
}

type DownstreamReceipt struct {
	DeliveryID       string `json:"deliveryId,omitempty"`
	Account          string `json:"account,omitempty"`
	ApplicationID    string `json:"applicationId,omitempty"`
	MessageID        string `json:"messageId"`
	GatewayMessageID string `json:"gatewayMessageId,omitempty"`
	PhoneNumber      string `json:"phoneNumber,omitempty"`
	ReceiptStatus    string `json:"receiptStatus"`
	RawStatus        string `json:"rawStatus,omitempty"`
	ErrorCode        string `json:"errorCode,omitempty"`
	SubmitSequenceID uint32 `json:"submitSequenceId,omitempty"`
	DeliveredAt      string `json:"deliveredAt,omitempty"`
}

type DownstreamUplink struct {
	DeliveryID    string `json:"deliveryId,omitempty"`
	Account       string `json:"account,omitempty"`
	ApplicationID string `json:"applicationId,omitempty"`
	MessageID     string `json:"messageId,omitempty"`
	PhoneNumber   string `json:"phoneNumber"`
	DestID        string `json:"destId"`
	Content       string `json:"content"`
	ReceivedAt    string `json:"receivedAt,omitempty"`
}

type DownstreamSendResult struct {
	Sent          bool   `json:"sent"`
	ConnectionID  string `json:"connectionId,omitempty"`
	SequenceID    string `json:"sequenceId,omitempty"`
	MessageID     string `json:"messageId,omitempty"`
	SentAt        string `json:"sentAt,omitempty"`
	AckDeadlineAt string `json:"ackDeadlineAt,omitempty"`
}

type downstreamDeliveryLifecycleEvent struct {
	Kind          string
	DeliveryID    string
	ConnectionID  string
	SequenceID    uint32
	MessageID     uint64
	Result        uint32
	ObservedAt    time.Time
	AckDeadlineAt time.Time
	FailureType   string
	ErrorMessage  string
}

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

type downstreamConnectionEvent struct {
	Account      string `json:"account"`
	ConnectionID string `json:"connectionId"`
	Status       string `json:"status"`
	RemoteIP     string `json:"remoteIp,omitempty"`
	Protocol     string `json:"protocol,omitempty"`
	ConnectedAt  string `json:"connectedAt,omitempty"`
	ObservedAt   string `json:"observedAt,omitempty"`
	ErrorMessage string `json:"errorMessage,omitempty"`
}

type downstreamSession struct {
	messageID      string
	account        string
	enterpriseCode string
	protocol       string
	srcID          string
	phoneNumber    string
	gatewayMsgID   uint64
	remoteIP       string
	connectedAt    time.Time
	connectionID   string
	conn           *cmpp.Conn
	mu             *sync.Mutex
	presence       PresenceStore
	instanceID     string
	report         func(*downstreamSession, string, string)
	deliveryReport func(downstreamDeliveryLifecycleEvent)
}

var downstreamRegistry = struct {
	sync.RWMutex
	byMessageID map[string]*downstreamSession
	byAccount   map[string]*downstreamSession
	byConn      map[*cmpp.Conn]*downstreamSession
}{
	byMessageID: make(map[string]*downstreamSession),
	byAccount:   make(map[string]*downstreamSession),
	byConn:      make(map[*cmpp.Conn]*downstreamSession),
}

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

func (s Server) ListenAndServe() error {
	addr := s.Addr
	if addr == "" {
		addr = ":17890"
	}
	s.logRecoveryCandidates(log.Default())
	go s.recoverPendingCandidates(log.Default())
	go s.runPendingFlusher(log.Default())
	return cmpp.ListenAndServe(addr, cmpp.V30, 30*time.Second, 3, s.LogWriter,
		cmpp.HandlerFunc(s.handleLogin),
		cmpp.HandlerFunc(s.handleSubmit),
		cmpp.HandlerFunc(s.handleActivity),
	)
}

func (s Server) handleLogin(response *cmpp.Response, packet *cmpp.Packet, logger *log.Logger) (bool, error) {
	req, ok := packet.Packer.(*cmpp.CmppConnReqPkt)
	if !ok {
		return true, nil
	}
	account := strings.TrimRight(req.SrcAddr, "\x00")
	if account == "" {
		setInboundConnectResponse(response.Packer, cmpp.ErrnoConnInvalidSrcAddr, req.AuthSrc, "", req.Version)
		return false, cmpp.ConnRspStatusErrMap[cmpp.ErrnoConnInvalidSrcAddr]
	}
	if req.Version != cmpp.V20 && req.Version != cmpp.V21 && req.Version != cmpp.V30 {
		setInboundConnectResponse(response.Packer, cmpp.ErrnoConnVerTooHigh, req.AuthSrc, "", cmpp.V30)
		return false, cmpp.ConnRspStatusErrMap[cmpp.ErrnoConnVerTooHigh]
	}
	auth, err := s.authenticate(packet.Conn.Conn.RemoteAddr(), account, req.AuthSrc, req.Timestamp)
	if err != nil {
		logger.Printf("cmpp inbound auth failed account=%s remote=%s err=%v", account, packet.Conn.Conn.RemoteAddr(), err)
		setInboundConnectResponse(response.Packer, cmpp.ErrnoConnAuthFailed, req.AuthSrc, "", req.Version)
		return false, cmpp.ConnRspStatusErrMap[cmpp.ErrnoConnAuthFailed]
	}
	setInboundConnectResponse(response.Packer, 0, req.AuthSrc, auth.PasswordCipher, req.Version)
	now := time.Now().UTC()
	session := downstreamSession{
		account:        strings.TrimSpace(defaultString(auth.Account, account)),
		enterpriseCode: strings.TrimSpace(auth.EnterpriseCode),
		protocol:       cmppVersionName(req.Version),
		srcID:          strings.TrimSpace(auth.Account),
		remoteIP:       remoteIP(packet.Conn.Conn.RemoteAddr()),
		connectedAt:    now,
		connectionID:   fmt.Sprintf("%s-%d", s.gatewayInstanceID(), now.UnixNano()),
		conn:           packet.Conn,
		mu:             &sync.Mutex{},
		presence:       s.PresenceStore,
		instanceID:     s.gatewayInstanceID(),
		report:         s.reportConnection,
		deliveryReport: s.reportDownstreamDelivery,
	}
	rememberAccount(session)
	go s.reportConnection(&session, "connected", "")
	response.AfterSend = func(sendErr error) {
		if sendErr == nil {
			go s.flushPending(defaultString(auth.Account, account), logger)
		}
	}
	logger.Printf(
		"cmpp inbound event=login_accepted protocol=%s requested_version=0x%02x response_version=0x%02x account=%s remote=%s",
		cmppVersionName(req.Version), uint8(req.Version), uint8(req.Version), account, packet.Conn.Conn.RemoteAddr(),
	)
	return false, nil
}

func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logger *log.Logger) (bool, error) {
	req, ok := normalizeInboundSubmit(packet.Packer)
	if !ok {
		return true, nil
	}
	session := findSessionByConn(packet.Conn)
	if session == nil || strings.TrimSpace(session.account) == "" {
		logger.Printf(
			"cmpp inbound event=submit_rejected protocol=%s packet_type=%s remote=%s seq=%d result=9 stage=session reason=%q",
			req.protocol, req.protocol, packet.Conn.Conn.RemoteAddr(), req.sequenceID, "authenticated connection session not found",
		)
		setInboundSubmitResponse(response.Packer, 0, 9)
		return false, nil
	}
	account := session.account
	enterpriseCode := strings.TrimRight(req.msgSrc, "\x00")
	if session.enterpriseCode != "" && enterpriseCode != session.enterpriseCode {
		logger.Printf(
			"cmpp inbound event=submit_rejected protocol=%s packet_type=%s account=%s enterprise_code=%s remote=%s seq=%d result=9 stage=protocol reason=%q",
			defaultString(session.protocol, req.protocol), req.protocol, account, enterpriseCode, packet.Conn.Conn.RemoteAddr(), req.sequenceID,
			fmt.Sprintf("enterprise code mismatch: expected %s", session.enterpriseCode),
		)
		setInboundSubmitResponse(response.Packer, 0, 9)
		return false, nil
	}
	phone := ""
	if len(req.destTerminalIDs) > 0 {
		phone = strings.TrimRight(req.destTerminalIDs[0], "\x00")
	}
	remote := packet.Conn.Conn.RemoteAddr()
	clientProtocol := defaultString(session.protocol, req.protocol)
	logger.Printf(
		"cmpp inbound event=submit_received protocol=%s packet_type=%s account=%s enterprise_code=%s remote=%s seq=%d phone=%s src_id=%s msg_fmt=%d pk=%d/%d dest_count=%d content_bytes=%d",
		clientProtocol, req.protocol, account, enterpriseCode, remote, req.sequenceID, phone, strings.TrimSpace(req.srcID), req.msgFmt,
		req.pkNumber, req.pkTotal, len(req.destTerminalIDs), len(req.msgContent),
	)
	content, err := decodeContent(req.msgFmt, req.msgContent)
	if err != nil {
		logger.Printf(
			"cmpp inbound event=submit_rejected protocol=%s packet_type=%s account=%s remote=%s seq=%d phone=%s result=9 stage=decode reason=%q",
			clientProtocol, req.protocol, account, remote, req.sequenceID, phone, err,
		)
		setInboundSubmitResponse(response.Packer, 0, 9)
		return false, nil
	}
	contentHash := fmt.Sprintf("%x", md5.Sum([]byte(content)))
	startedAt := time.Now()
	result, err := s.submit(remote, submitRequest{
		Account:     account,
		PhoneNumber: phone,
		Content:     content,
		SrcID:       req.srcID,
		DestID:      phone,
		SequenceID:  req.sequenceID,
		RemoteIP:    remoteIP(remote),
	})
	if err != nil || !result.Accepted {
		reason := "api returned accepted=false"
		if err != nil {
			reason = err.Error()
		}
		logger.Printf(
			"cmpp inbound event=submit_rejected protocol=%s packet_type=%s account=%s remote=%s seq=%d phone=%s result=9 stage=business duration_ms=%d content_chars=%d content_hash=%s reason=%q",
			clientProtocol, req.protocol, account, remote, req.sequenceID, phone, time.Since(startedAt).Milliseconds(), len([]rune(content)), contentHash, reason,
		)
		setInboundSubmitResponse(response.Packer, 0, 9)
		return false, nil
	}
	gatewayMsgID := messageIDFrom(result.MessageID, req.sequenceID)
	setInboundSubmitResponse(response.Packer, gatewayMsgID, 0)
	rememberDownstream(downstreamSession{
		messageID:      result.MessageID,
		account:        account,
		enterpriseCode: session.enterpriseCode,
		protocol:       clientProtocol,
		srcID:          strings.TrimSpace(req.srcID),
		phoneNumber:    phone,
		gatewayMsgID:   gatewayMsgID,
		remoteIP:       remoteIP(remote),
		connectedAt:    time.Now().UTC(),
		connectionID:   session.connectionID,
		conn:           packet.Conn,
		mu:             &sync.Mutex{},
		presence:       s.PresenceStore,
		instanceID:     s.gatewayInstanceID(),
		report:         session.report,
		deliveryReport: session.deliveryReport,
	})
	if current := findSessionByConn(packet.Conn); current != nil && current.report != nil {
		go current.report(current, "submit", "")
	}
	response.AfterSend = func(sendErr error) {
		if sendErr != nil {
			return
		}
		go func() {
			if _, err := s.flushPending(account, logger); err != nil {
				logger.Printf("cmpp inbound event=post_submit_pending_flush_failed account=%s message_id=%s error=%q", account, result.MessageID, err.Error())
			}
		}()
	}
	logger.Printf(
		"cmpp inbound event=submit_accepted protocol=%s packet_type=%s account=%s remote=%s seq=%d phone=%s result=0 message_id=%s gateway_message_id=%d duration_ms=%d content_chars=%d content_hash=%s",
		clientProtocol, req.protocol, account, remote, req.sequenceID, phone, result.MessageID, gatewayMsgID, time.Since(startedAt).Milliseconds(), len([]rune(content)), contentHash,
	)
	return false, nil
}

func (s Server) handleActivity(_ *cmpp.Response, packet *cmpp.Packet, logger *log.Logger) (bool, error) {
	session := findSessionByConn(packet.Conn)
	if session == nil {
		return true, nil
	}
	switch response := packet.Packer.(type) {
	case *cmpp.CmppActiveTestReqPkt, *cmpp.CmppActiveTestRspPkt:
		if session.report != nil {
			go session.report(session, "heartbeat", "")
		}
	case *cmpp.Cmpp2DeliverRspPkt:
		handleDownstreamAcknowledgement(packet.Conn, response.SeqId, response.MsgId, uint32(response.Result), logger)
	case *cmpp.Cmpp3DeliverRspPkt:
		handleDownstreamAcknowledgement(packet.Conn, response.SeqId, response.MsgId, response.Result, logger)
	}
	return true, nil
}

func (s Server) reportConnection(session *downstreamSession, status string, errorMessage string) {
	if session == nil || strings.TrimSpace(session.account) == "" || strings.TrimSpace(session.connectionID) == "" {
		return
	}
	event := downstreamConnectionEvent{
		Account: session.account, ConnectionID: session.connectionID, Status: status,
		RemoteIP: session.remoteIP, Protocol: session.protocol,
		ConnectedAt: formatRFC3339Nano(session.connectedAt), ObservedAt: formatRFC3339Nano(time.Now()),
		ErrorMessage: errorMessage,
	}
	if err := s.post(context.Background(), "/gateway/events/inbound/connection", event, nil); err != nil {
		log.Printf("cmpp inbound connection state callback failed account=%s connection_id=%s status=%s err=%v", session.account, session.connectionID, status, err)
	}
}

func (s Server) reportDownstreamDelivery(event downstreamDeliveryLifecycleEvent) {
	if strings.TrimSpace(event.DeliveryID) == "" {
		return
	}
	payload := map[string]any{
		"id": event.DeliveryID, "connectionId": event.ConnectionID,
		"sequenceId": strconv.FormatUint(uint64(event.SequenceID), 10),
		"messageId":  strconv.FormatUint(event.MessageID, 10),
	}
	switch event.Kind {
	case "sent":
		payload["sentAt"] = formatRFC3339Nano(event.ObservedAt)
		payload["ackDeadlineAt"] = formatRFC3339Nano(event.AckDeadlineAt)
		_ = s.post(context.Background(), "/gateway/events/downstream/sent", payload, nil)
	case "acknowledged":
		payload["result"] = event.Result
		payload["acknowledgedAt"] = formatRFC3339Nano(event.ObservedAt)
		_ = s.post(context.Background(), "/gateway/events/downstream/acknowledged", payload, nil)
	case "failed":
		payload["failureType"] = event.FailureType
		payload["errorMessage"] = event.ErrorMessage
		_ = s.post(context.Background(), "/gateway/events/downstream/failed", payload, nil)
	}
}

type inboundSubmitPacket struct {
	protocol        string
	pkTotal         uint8
	pkNumber        uint8
	msgFmt          uint8
	msgSrc          string
	srcID           string
	destTerminalIDs []string
	msgContent      string
	sequenceID      uint32
}

func normalizeInboundSubmit(packet any) (inboundSubmitPacket, bool) {
	switch req := packet.(type) {
	case *cmpp.Cmpp2SubmitReqPkt:
		return inboundSubmitPacket{
			protocol: "cmpp20", pkTotal: req.PkTotal, pkNumber: req.PkNumber, msgFmt: req.MsgFmt,
			msgSrc: req.MsgSrc, srcID: req.SrcId, destTerminalIDs: req.DestTerminalId,
			msgContent: req.MsgContent, sequenceID: req.SeqId,
		}, true
	case *cmpp.Cmpp3SubmitReqPkt:
		return inboundSubmitPacket{
			protocol: "cmpp30", pkTotal: req.PkTotal, pkNumber: req.PkNumber, msgFmt: req.MsgFmt,
			msgSrc: req.MsgSrc, srcID: req.SrcId, destTerminalIDs: req.DestTerminalId,
			msgContent: req.MsgContent, sequenceID: req.SeqId,
		}, true
	default:
		return inboundSubmitPacket{}, false
	}
}

func setInboundSubmitResponse(packet any, messageID uint64, result uint32) {
	switch resp := packet.(type) {
	case *cmpp.Cmpp2SubmitRspPkt:
		resp.MsgId = messageID
		resp.Result = uint8(result)
	case *cmpp.Cmpp3SubmitRspPkt:
		resp.MsgId = messageID
		resp.Result = result
	}
}

func setInboundConnectResponse(packet any, status uint8, authSource string, secret string, version cmpp.Type) {
	switch resp := packet.(type) {
	case *cmpp.Cmpp2ConnRspPkt:
		resp.Status = status
		resp.AuthSrc = authSource
		resp.Secret = secret
		resp.Version = version
	case *cmpp.Cmpp3ConnRspPkt:
		resp.Status = uint32(status)
		resp.AuthSrc = authSource
		resp.Secret = secret
		resp.Version = version
	}
}

func findSessionByConn(conn *cmpp.Conn) *downstreamSession {
	downstreamRegistry.RLock()
	defer downstreamRegistry.RUnlock()
	return downstreamRegistry.byConn[conn]
}

func cmppVersionName(version cmpp.Type) string {
	switch version {
	case cmpp.V20:
		return "cmpp20"
	case cmpp.V21:
		return "cmpp21"
	case cmpp.V30:
		return "cmpp30"
	default:
		return fmt.Sprintf("unknown_0x%02x", uint8(version))
	}
}

func (s Server) authenticate(remote net.Addr, account string, authSource string, timestamp uint32) (authResponse, error) {
	payload := authRequest{
		Account:    account,
		AuthSource: base64.StdEncoding.EncodeToString([]byte(authSource)),
		Timestamp:  timestamp,
		RemoteIP:   remoteIP(remote),
	}
	var result authResponse
	err := s.post(context.Background(), "/gateway/events/inbound/authenticate", payload, &result)
	return result, err
}

func (s Server) submit(remote net.Addr, payload submitRequest) (submitResponse, error) {
	payload.RemoteIP = remoteIP(remote)
	var result submitResponse
	err := s.post(context.Background(), "/gateway/events/inbound/submit", payload, &result)
	return result, err
}

type pendingDeliveryRequest struct {
	Account string `json:"account"`
	Limit   int    `json:"limit,omitempty"`
}

type pendingDelivery struct {
	ID           string          `json:"id"`
	DeliveryType string          `json:"deliveryType"`
	Payload      json.RawMessage `json:"payload"`
	CreatedAt    time.Time       `json:"createdAt"`
}

type pendingFlushResult struct {
	Account        string
	Deliveries     int
	DeliveredCount int
	FailedCount    int
	WaitingCount   int
	LastError      string
}

func (s Server) flushPending(account string, logger *log.Logger) (pendingFlushResult, error) {
	result := pendingFlushResult{Account: account}
	if account == "" {
		return result, nil
	}
	var deliveries []pendingDelivery
	if err := s.post(context.Background(), "/gateway/events/downstream/pending", pendingDeliveryRequest{Account: account, Limit: 100}, &deliveries); err != nil {
		logger.Printf("cmpp inbound pending delivery fetch failed account=%s err=%v", account, err)
		result.LastError = err.Error()
		return result, err
	}
	result.Deliveries = len(deliveries)
	for _, delivery := range deliveries {
		sendResult, err := s.pushPendingDelivery(account, delivery)
		if err != nil {
			result.FailedCount++
			result.LastError = err.Error()
			_ = s.post(context.Background(), "/gateway/events/downstream/failed", map[string]string{
				"id":           delivery.ID,
				"errorMessage": err.Error(),
				"failureType":  "send_failed",
			}, nil)
			continue
		}
		if sendResult.Sent {
			result.DeliveredCount++
			continue
		}
		result.WaitingCount++
	}
	return result, nil
}

func (s Server) pushPendingDelivery(account string, delivery pendingDelivery) (DownstreamSendResult, error) {
	switch delivery.DeliveryType {
	case "receipt":
		var event DownstreamReceipt
		if err := json.Unmarshal(delivery.Payload, &event); err != nil {
			return DownstreamSendResult{}, err
		}
		event.DeliveryID = delivery.ID
		event.Account = defaultString(event.Account, account)
		allowRecovery := !delivery.CreatedAt.IsZero() && time.Since(delivery.CreatedAt) >= 5*time.Second
		return pushReceiptWithResult(event, allowRecovery)
	case "uplink":
		var event DownstreamUplink
		if err := json.Unmarshal(delivery.Payload, &event); err != nil {
			return DownstreamSendResult{}, err
		}
		event.DeliveryID = delivery.ID
		event.Account = defaultString(event.Account, account)
		return PushUplinkWithResult(event)
	default:
		return DownstreamSendResult{}, fmt.Errorf("unsupported downstream delivery type %q", delivery.DeliveryType)
	}
}

func (s Server) post(ctx context.Context, path string, payload any, result any) error {
	client := s.HTTPClient
	if client == nil {
		client = &http.Client{Timeout: defaultHTTPTimeout}
	}
	body, err := json.Marshal(payload)
	if err != nil {
		return err
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(apiBaseURL(s.APIBaseURL), "/")+path, bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	resp, err := client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	responseBody, err := io.ReadAll(io.LimitReader(resp.Body, 64*1024))
	if err != nil {
		return fmt.Errorf("read api response: %w", err)
	}
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		detail := strings.TrimSpace(string(responseBody))
		if detail == "" {
			return fmt.Errorf("api returned %s", resp.Status)
		}
		return fmt.Errorf("api returned %s: %s", resp.Status, detail)
	}
	if result != nil {
		if len(responseBody) == 0 {
			return io.EOF
		}
		return json.Unmarshal(responseBody, result)
	}
	return nil
}

func decodeContent(format uint8, content string) (string, error) {
	switch format {
	case 8:
		return cmpputils.Ucs2ToUtf8(content)
	case 15:
		return cmpputils.GB18030ToUtf8(content)
	default:
		return content, nil
	}
}

func apiBaseURL(value string) string {
	if value == "" {
		return "http://127.0.0.1:3000/api"
	}
	return value
}

func remoteIP(addr net.Addr) string {
	if tcp, ok := addr.(*net.TCPAddr); ok {
		return tcp.IP.String()
	}
	host, _, err := net.SplitHostPort(addr.String())
	if err == nil {
		return host
	}
	return addr.String()
}

func messageIDFrom(value string, seq uint32) uint64 {
	hash := md5.Sum([]byte(value))
	result := uint64(seq)
	for _, item := range hash[:6] {
		result = (result << 8) + uint64(item)
	}
	if result == 0 {
		return uint64(time.Now().UnixNano())
	}
	return result
}

func rememberDownstream(session downstreamSession) {
	if session.messageID == "" || session.conn == nil {
		return
	}
	session.touchPresence("connected", true, false)
	downstreamRegistry.Lock()
	downstreamRegistry.byMessageID[session.messageID] = &session
	downstreamRegistry.byConn[session.conn] = &session
	if session.account != "" {
		downstreamRegistry.byAccount[session.account] = &session
	}
	downstreamRegistry.Unlock()
}

func rememberAccount(session downstreamSession) {
	if session.account == "" || session.conn == nil {
		return
	}
	session.touchPresence("connected", false, false)
	downstreamRegistry.Lock()
	downstreamRegistry.byAccount[session.account] = &session
	downstreamRegistry.byConn[session.conn] = &session
	downstreamRegistry.Unlock()
}

func forgetDownstream(session *downstreamSession) {
	if session == nil {
		return
	}
	downstreamRegistry.Lock()
	if session.messageID != "" {
		if current := downstreamRegistry.byMessageID[session.messageID]; current == session {
			delete(downstreamRegistry.byMessageID, session.messageID)
		}
	}
	if session.account != "" {
		if current := downstreamRegistry.byAccount[session.account]; current == session {
			delete(downstreamRegistry.byAccount, session.account)
		}
	}
	if current := downstreamRegistry.byConn[session.conn]; current == session {
		delete(downstreamRegistry.byConn, session.conn)
	}
	downstreamRegistry.Unlock()
	_ = session.removePresence()
}

func (s Server) runPendingFlusher(logger *log.Logger) {
	ticker := time.NewTicker(s.pendingFlushInterval())
	defer ticker.Stop()
	for range ticker.C {
		s.flushOnlineAccounts(logger)
		s.recoverPendingCandidates(logger)
	}
}

func (s Server) flushOnlineAccounts(logger *log.Logger) {
	for _, account := range onlineAccounts() {
		_, _ = s.flushPending(account, logger)
	}
}

func (s Server) recoverPendingCandidates(logger *log.Logger) {
	candidates, err := ListRecoveryCandidates(context.Background(), s.PresenceStore)
	if err != nil {
		logger.Printf("cmpp inbound recovery candidate refresh failed err=%v", err)
		return
	}
	for _, candidate := range candidates {
		account := strings.TrimSpace(candidate.Account)
		if account == "" {
			continue
		}
		activeRecovery := DownstreamRecoveryStatus{
			Account:           account,
			GatewayInstanceID: s.gatewayInstanceID(),
		}
		if s.RecoveryStore != nil {
			decision, recoveryErr := s.RecoveryStore.StartAccountRecovery(context.Background(), account, s.gatewayInstanceID())
			if recoveryErr != nil {
				logger.Printf("cmpp inbound recovery start failed account=%s err=%v", account, recoveryErr)
				continue
			}
			if !decision.Allowed {
				logger.Printf("cmpp inbound recovery skipped account=%s reason=%s", account, decision.SkipReason)
				decision.Status.Account = account
				decision.Status.GatewayInstanceID = s.gatewayInstanceID()
				decision.Status.State = defaultString(decision.Status.State, "failed")
				decision.Status.LastSkipReason = defaultString(decision.Status.LastSkipReason, decision.SkipReason)
				if decision.Status.FailureCategory == "" {
					decision.Status.FailureCategory = recoveryFailureCategory(decision.Status.State, "", decision.Status.LastSkipReason)
				}
				s.syncRecoveryStatus(logger, account, decision.Status)
				continue
			}
			activeRecovery = decision.Status
			activeRecovery.Account = account
			activeRecovery.GatewayInstanceID = s.gatewayInstanceID()
		}
		result, flushErr := s.flushPending(account, logger)
		if s.RecoveryStore != nil {
			status := DownstreamRecoveryStatus{
				Account:           account,
				GatewayInstanceID: s.gatewayInstanceID(),
				LockToken:         activeRecovery.LockToken,
				LockOwner:         defaultString(activeRecovery.LockOwner, s.gatewayInstanceID()),
				LockAcquiredAt:    activeRecovery.LockAcquiredAt,
				LockExpiresAt:     activeRecovery.LockExpiresAt,
				LastAttemptAt:     activeRecovery.LastAttemptAt,
			}
			switch {
			case flushErr != nil:
				status.State = "failed"
				status.LastError = flushErr.Error()
				status.FailureCategory = recoveryFailureCategory(status.State, status.LastError, status.LastSkipReason)
			case result.WaitingCount > 0 && result.DeliveredCount == 0 && result.FailedCount == 0:
				status.State = "waiting_connection"
				status.LastError = "downstream client is not connected"
				status.FailureCategory = recoveryFailureCategory(status.State, status.LastError, status.LastSkipReason)
			case result.FailedCount > 0 && result.DeliveredCount > 0:
				status.State = "partial"
				status.LastError = result.LastError
				status.FailureCategory = recoveryFailureCategory(status.State, status.LastError, status.LastSkipReason)
			default:
				status.State = "success"
				status.LastError = ""
				status.FailureCategory = ""
			}
			if err := s.RecoveryStore.CompleteAccountRecovery(context.Background(), status); err != nil {
				logger.Printf("cmpp inbound recovery completion failed account=%s err=%v", account, err)
				if err == ErrRecoveryLockLost {
					status.State = "failed"
					status.LastSkipReason = "lock_lost"
					status.FailureCategory = "lock_lost"
					s.syncRecoveryStatus(logger, account, status)
				}
			} else if persisted, err := s.RecoveryStore.GetAccountRecoveryStatus(context.Background(), account); err != nil {
				logger.Printf("cmpp inbound recovery status fetch failed account=%s err=%v", account, err)
			} else {
				s.syncRecoveryStatus(logger, account, persisted)
			}
		}
	}
}

func (s Server) syncRecoveryStatus(logger *log.Logger, account string, status DownstreamRecoveryStatus) {
	if err := s.post(context.Background(), "/gateway/events/downstream/recovery-status", map[string]any{
		"account":           status.Account,
		"gatewayInstanceId": status.GatewayInstanceID,
		"state":             status.State,
		"lockOwner":         status.LockOwner,
		"lockExpiresAt":     formatRFC3339Nano(status.LockExpiresAt),
		"lastAttemptAt":     formatRFC3339Nano(status.LastAttemptAt),
		"lastSuccessAt":     formatRFC3339Nano(status.LastSuccessAt),
		"lastFailureAt":     formatRFC3339Nano(status.LastFailureAt),
		"nextRetryAt":       formatRFC3339Nano(status.NextRetryAt),
		"attemptCount":      status.AttemptCount,
		"failureCategory":   status.FailureCategory,
		"lastError":         status.LastError,
		"lastSkipReason":    status.LastSkipReason,
	}, nil); err != nil {
		logger.Printf("cmpp inbound recovery status sync failed account=%s err=%v", account, err)
	}
}

func recoveryFailureCategory(state string, lastError string, lastSkipReason string) string {
	if state == "success" || state == "running" {
		return ""
	}
	if lastSkipReason == "backoff" {
		return "backoff"
	}
	if lastSkipReason == "locked" {
		return "lock_contended"
	}
	if lastSkipReason == "lock_lost" {
		return "lock_lost"
	}
	if state == "waiting_connection" {
		return "client_disconnected"
	}
	if state == "partial" {
		return "partial_delivery_failed"
	}
	if state == "failed" && strings.TrimSpace(lastError) != "" {
		return "flush_failed"
	}
	return "unknown"
}

func onlineAccounts() []string {
	downstreamRegistry.RLock()
	defer downstreamRegistry.RUnlock()
	accounts := make([]string, 0, len(downstreamRegistry.byAccount))
	for account := range downstreamRegistry.byAccount {
		if strings.TrimSpace(account) != "" {
			accounts = append(accounts, account)
		}
	}
	return accounts
}

func PushReceipt(event DownstreamReceipt) (bool, error) {
	result, err := PushReceiptWithResult(event)
	return result.Sent, err
}

func PushReceiptWithResult(event DownstreamReceipt) (DownstreamSendResult, error) {
	return pushReceiptWithResult(event, false)
}

func pushReceiptWithResult(event DownstreamReceipt, allowRecovery bool) (DownstreamSendResult, error) {
	session := findReceiptSession(event.MessageID, event.Account)
	if session == nil && allowRecovery {
		session = recoverReceiptSession(event)
	}
	if session == nil {
		return DownstreamSendResult{}, nil
	}
	stat := strings.TrimSpace(event.RawStatus)
	if stat == "" {
		stat = cmppReceiptStatus(event.ReceiptStatus)
	}
	when := time.Now()
	if event.DeliveredAt != "" {
		if parsed, err := time.Parse(time.RFC3339Nano, event.DeliveredAt); err == nil {
			when = parsed
		}
	}
	receipt := &cmpp.CmppReceiptPkt{
		MsgId:          session.gatewayMsgID,
		Stat:           stat,
		SubmitTime:     when.Format("0601021504"),
		DoneTime:       when.Format("0601021504"),
		DestTerminalId: defaultString(event.PhoneNumber, session.phoneNumber),
		SmscSequence:   uint32(time.Now().UnixNano() & 0xffffffff),
	}
	receiptBytes, err := receipt.Pack()
	if err != nil {
		return DownstreamSendResult{}, err
	}
	deliver := downstreamDeliverPacket(session, session.gatewayMsgID, session.srcID, defaultString(event.PhoneNumber, session.phoneNumber), 0, 1, string(receiptBytes))
	return sendDownstream(session, deliver, event.DeliveryID)
}

func findReceiptSession(messageID string, account string) *downstreamSession {
	downstreamRegistry.RLock()
	defer downstreamRegistry.RUnlock()
	if messageID != "" {
		return downstreamRegistry.byMessageID[messageID]
	}
	if account != "" {
		return downstreamRegistry.byAccount[account]
	}
	return nil
}

func recoverReceiptSession(event DownstreamReceipt) *downstreamSession {
	if event.MessageID == "" || event.SubmitSequenceID == 0 || event.Account == "" {
		return nil
	}
	downstreamRegistry.RLock()
	accountSession := downstreamRegistry.byAccount[event.Account]
	downstreamRegistry.RUnlock()
	if accountSession == nil || accountSession.conn == nil {
		return nil
	}
	recovered := *accountSession
	recovered.messageID = event.MessageID
	recovered.gatewayMsgID = messageIDFrom(event.MessageID, event.SubmitSequenceID)
	return &recovered
}

func PushUplink(event DownstreamUplink) (bool, error) {
	result, err := PushUplinkWithResult(event)
	return result.Sent, err
}

func PushUplinkWithResult(event DownstreamUplink) (DownstreamSendResult, error) {
	session := findSession(event.MessageID, event.Account)
	if session == nil {
		return DownstreamSendResult{}, nil
	}
	content, err := cmpputils.Utf8ToUcs2(event.Content)
	if err != nil {
		return DownstreamSendResult{}, err
	}
	deliver := downstreamDeliverPacket(
		session,
		messageIDFrom(defaultString(event.MessageID, event.Account), uint32(time.Now().UnixNano())),
		defaultString(event.DestID, session.srcID),
		event.PhoneNumber,
		8,
		0,
		content,
	)
	return sendDownstream(session, deliver, event.DeliveryID)
}

func downstreamDeliverPacket(session *downstreamSession, messageID uint64, destID string, sourceTerminalID string, msgFmt uint8, registerDelivery uint8, content string) cmpp.Packer {
	if session != nil && (session.protocol == "cmpp20" || session.protocol == "cmpp21") {
		return &cmpp.Cmpp2DeliverReqPkt{
			MsgId: messageID, DestId: destID, ServiceId: "cmpp", MsgFmt: msgFmt,
			SrcTerminalId: sourceTerminalID, RegisterDelivery: registerDelivery,
			MsgLength: uint8(len(content)), MsgContent: content,
		}
	}
	return &cmpp.Cmpp3DeliverReqPkt{
		MsgId: messageID, DestId: destID, ServiceId: "cmpp", MsgFmt: msgFmt,
		SrcTerminalId: sourceTerminalID, RegisterDelivery: registerDelivery,
		MsgLength: uint8(len(content)), MsgContent: content,
	}
}

func findSession(messageID string, account string) *downstreamSession {
	downstreamRegistry.RLock()
	defer downstreamRegistry.RUnlock()
	if messageID != "" {
		if session := downstreamRegistry.byMessageID[messageID]; session != nil {
			return session
		}
	}
	if account != "" {
		return downstreamRegistry.byAccount[account]
	}
	return nil
}

func sendDownstream(session *downstreamSession, deliver cmpp.Packer, deliveryID string) (DownstreamSendResult, error) {
	session.mu.Lock()
	defer session.mu.Unlock()
	messageID := downstreamDeliverMessageID(deliver)
	if messageID == 0 {
		return DownstreamSendResult{}, errors.New("refusing downstream CMPP_DELIVER with Msg_Id=0")
	}
	sequenceID := <-session.conn.SeqId
	sentAt := time.Now().UTC()
	ackDeadlineAt := sentAt.Add(downstreamAckTimeout())
	tracker := registerDownstreamAck(session, deliveryID, sequenceID, messageID, ackDeadlineAt)
	if err := session.conn.SendPkt(deliver, sequenceID); err != nil {
		removeDownstreamAck(tracker)
		if session.report != nil {
			go session.report(session, "disconnected", err.Error())
		}
		forgetDownstream(session)
		return DownstreamSendResult{}, err
	}
	result := DownstreamSendResult{
		Sent: true, ConnectionID: session.connectionID,
		SequenceID: strconv.FormatUint(uint64(sequenceID), 10), MessageID: strconv.FormatUint(messageID, 10),
		SentAt: formatRFC3339Nano(sentAt), AckDeadlineAt: formatRFC3339Nano(ackDeadlineAt),
	}
	if deliveryID != "" && session.deliveryReport != nil {
		go session.deliveryReport(downstreamDeliveryLifecycleEvent{
			Kind: "sent", DeliveryID: deliveryID, ConnectionID: session.connectionID,
			SequenceID: sequenceID, MessageID: messageID, ObservedAt: sentAt, AckDeadlineAt: ackDeadlineAt,
		})
	}
	session.touchPresence("connected", false, true)
	if session.report != nil {
		go session.report(session, "deliver", "")
	}
	return result, nil
}

func downstreamDeliverMessageID(deliver cmpp.Packer) uint64 {
	switch packet := deliver.(type) {
	case *cmpp.Cmpp2DeliverReqPkt:
		return packet.MsgId
	case *cmpp.Cmpp3DeliverReqPkt:
		return packet.MsgId
	default:
		return 0
	}
}

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

func registerDownstreamAck(session *downstreamSession, deliveryID string, sequenceID uint32, messageID uint64, deadline time.Time) *downstreamAckTracker {
	if session == nil || session.conn == nil || strings.TrimSpace(deliveryID) == "" {
		return nil
	}
	tracker := &downstreamAckTracker{
		deliveryID: deliveryID, connectionID: session.connectionID,
		sequenceID: sequenceID, messageID: messageID, session: session,
	}
	key := downstreamAckKey(session.conn, sequenceID)
	downstreamAckRegistry.Lock()
	downstreamAckRegistry.items[key] = tracker
	downstreamAckRegistry.Unlock()
	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, ConnectionID: timedOut.connectionID,
			SequenceID: timedOut.sequenceID, MessageID: timedOut.messageID, ObservedAt: time.Now().UTC(),
			FailureType: "ack_timeout", ErrorMessage: "CMPP_DELIVER_RESP timeout",
		})
	})
	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)
		}
		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, ConnectionID: tracker.connectionID,
			SequenceID: sequenceID, MessageID: messageID, Result: result, ObservedAt: time.Now().UTC(),
		})
	}
}

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
}

func (s Server) pendingFlushInterval() time.Duration {
	if s.PendingFlushInterval > 0 {
		return s.PendingFlushInterval
	}
	return time.Minute
}

func (s Server) logRecoveryCandidates(logger *log.Logger) {
	candidates, err := ListRecoveryCandidates(context.Background(), s.PresenceStore)
	if err != nil {
		logger.Printf("cmpp inbound recovery candidates load failed err=%v", err)
		return
	}
	if len(candidates) == 0 {
		logger.Printf("cmpp inbound recovery candidates loaded count=0")
		return
	}
	accounts := make([]string, 0, len(candidates))
	for _, item := range candidates {
		if strings.TrimSpace(item.Account) != "" {
			accounts = append(accounts, item.Account)
		}
	}
	logger.Printf("cmpp inbound recovery candidates loaded count=%d accounts=%s", len(candidates), strings.Join(accounts, ","))
}

func cmppReceiptStatus(status string) string {
	switch strings.ToLower(strings.TrimSpace(status)) {
	case "delivered":
		return "DELIVRD"
	case "unknown":
		return "UNKNOWN"
	default:
		return "UNDELIV"
	}
}

func defaultString(value string, fallback string) string {
	if value == "" {
		return fallback
	}
	return value
}

func formatRFC3339Nano(value time.Time) string {
	if value.IsZero() {
		return ""
	}
	return value.UTC().Format(time.RFC3339Nano)
}

func (s Server) gatewayInstanceID() string {
	if strings.TrimSpace(s.GatewayInstanceID) != "" {
		return strings.TrimSpace(s.GatewayInstanceID)
	}
	return "gateway-1"
}

func (session downstreamSession) touchPresence(state string, includeSubmit bool, includeDeliver bool) {
	if session.presence == nil || strings.TrimSpace(session.account) == "" {
		return
	}
	now := time.Now().UTC()
	snapshot := DownstreamPresence{
		Account:           strings.TrimSpace(session.account),
		SrcID:             strings.TrimSpace(session.srcID),
		RemoteIP:          strings.TrimSpace(session.remoteIP),
		GatewayInstanceID: strings.TrimSpace(session.instanceID),
		State:             defaultString(strings.TrimSpace(state), "connected"),
		ConnectedAt:       session.connectedAt,
		UpdatedAt:         now,
	}
	if includeSubmit {
		snapshot.LastSubmitAt = now
	}
	if includeDeliver {
		snapshot.LastDeliverAt = now
	}
	_ = session.presence.TouchAccount(context.Background(), snapshot)
}

func (session downstreamSession) removePresence() error {
	if session.presence == nil || strings.TrimSpace(session.account) == "" {
		return nil
	}
	return session.presence.RemoveAccount(context.Background(), strings.TrimSpace(session.account))
}