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,omitempty"`
	PhoneNumbers []string                    `json:"phoneNumbers,omitempty"`
	Content      string                      `json:"content"`
	SrcID        string                      `json:"srcId,omitempty"`
	DestID       string                      `json:"destId,omitempty"`
	SequenceID   uint32                      `json:"sequenceId,omitempty"`
	RemoteIP     string                      `json:"remoteIp,omitempty"`
	LongMessage  *inboundLongMessageFragment `json:"longMessage,omitempty"`
}

type inboundLongMessageFragment struct {
	Reference int `json:"reference"`
	Total     int `json:"total"`
	Index     int `json:"index"`
	Format    int `json:"format"`
}

type submitResponseMessage struct {
	PhoneNumber string `json:"phoneNumber"`
	MessageID   string `json:"messageId"`
}

type submitResponse struct {
	Accepted      bool                    `json:"accepted"`
	Result        uint32                  `json:"result,omitempty"`
	TenantID      string                  `json:"tenantId,omitempty"`
	ApplicationID string                  `json:"applicationId,omitempty"`
	MessageID     string                  `json:"messageId"`
	Messages      []submitResponseMessage `json:"messages,omitempty"`
}

type protocolLogEvent struct {
	Protocol         string         `json:"protocol"`
	Direction        string         `json:"direction"`
	EventType        string         `json:"eventType"`
	Status           string         `json:"status"`
	TenantID         string         `json:"tenantId,omitempty"`
	ApplicationID    string         `json:"applicationId,omitempty"`
	Account          string         `json:"account,omitempty"`
	MessageID        string         `json:"messageId,omitempty"`
	GatewayMessageID string         `json:"gatewayMessageId,omitempty"`
	Phone            string         `json:"phone,omitempty"`
	ResultCode       string         `json:"resultCode,omitempty"`
	Detail           map[string]any `json:"detail,omitempty"`
}

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

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"`
	SubmitGroupMessageID string `json:"submitGroupMessageId,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"`
	Retryable     bool   `json:"retryable"`
	ReasonCode    string `json:"reasonCode,omitempty"`
	ErrorMessage  string `json:"errorMessage,omitempty"`
	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
	tenantID       string
	applicationID  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)
	protocolLog    func(protocolLogEvent)
}

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)}

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

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.ListenAndServeWithClose(addr, cmpp.V30, 30*time.Second, 3, s.LogWriter, s.handleConnectionClosed,
		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]
	}
	now := time.Now().UTC()
	session := &downstreamSession{
		account:        strings.TrimSpace(defaultString(auth.Account, account)),
		tenantID:       strings.TrimSpace(auth.TenantID),
		applicationID:  strings.TrimSpace(auth.ApplicationID),
		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,
		protocolLog:    s.emitProtocolLog,
	}
	if !rememberAccount(session, auth.MaxConnections) {
		logger.Printf("cmpp inbound auth failed account=%s remote=%s err=connection limit exceeded max=%d", account, packet.Conn.Conn.RemoteAddr(), auth.MaxConnections)
		setInboundConnectResponse(response.Packer, cmpp.ErrnoConnOthers, req.AuthSrc, "", req.Version)
		return false, cmpp.ConnRspStatusErrMap[cmpp.ErrnoConnOthers]
	}
	setInboundConnectResponse(response.Packer, 0, req.AuthSrc, auth.PasswordCipher, req.Version)
	go s.reportConnectionOrDisconnect(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)
		response.AfterSend = s.submitResponseProtocolLogger("", req.protocol, req.sequenceID, "", "", 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)
		response.AfterSend = s.submitResponseProtocolLogger(account, defaultString(session.protocol, req.protocol), req.sequenceID, "", "", 0, 9)
		return false, nil
	}
	phones := make([]string, len(req.destTerminalIDs))
	for index, destination := range req.destTerminalIDs {
		phones[index] = strings.TrimSpace(strings.TrimRight(destination, "\x00"))
	}
	phone := ""
	if len(phones) > 0 {
		phone = phones[0]
	}
	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, longMessage, err := decodeInboundSubmitContent(req)
	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)
		response.AfterSend = s.submitResponseProtocolLogger(account, clientProtocol, req.sequenceID, phone, "", 0, 9)
		return false, nil
	}
	contentHash := fmt.Sprintf("%x", md5.Sum([]byte(content)))
	startedAt := time.Now()
	releaseSubmitBarrier := beginDownstreamSubmitBarrier(packet.Conn)
	result, err := s.submit(remote, submitRequest{
		Account:      account,
		PhoneNumber:  phone,
		PhoneNumbers: phones,
		Content:      content,
		SrcID:        req.srcID,
		DestID:       phone,
		SequenceID:   req.sequenceID,
		RemoteIP:     remoteIP(remote),
		LongMessage:  longMessage,
	})
	if err != nil || !result.Accepted {
		reason := "api returned accepted=false"
		if err != nil {
			reason = err.Error()
		}
		responseResult := result.Result
		if responseResult == 0 {
			responseResult = 9
		}
		logger.Printf(
			"cmpp inbound event=submit_rejected protocol=%s packet_type=%s account=%s remote=%s seq=%d phone=%s result=%d stage=business duration_ms=%d content_chars=%d content_hash=%s reason=%q",
			clientProtocol, req.protocol, account, remote, req.sequenceID, phone, responseResult, time.Since(startedAt).Milliseconds(), len([]rune(content)), contentHash, reason,
		)
		setInboundSubmitResponse(response.Packer, 0, responseResult)
		protocolLogger := s.submitResponseProtocolLogger(account, clientProtocol, req.sequenceID, phone, result.MessageID, 0, responseResult)
		response.AfterSend = func(sendErr error) {
			releaseSubmitBarrier()
			protocolLogger(sendErr)
		}
		return false, nil
	}
	gatewayMsgID := messageIDFrom(result.MessageID, req.sequenceID)
	setInboundSubmitResponse(response.Packer, gatewayMsgID, 0)
	responseMessages := result.Messages
	if len(responseMessages) == 0 {
		responseMessages = []submitResponseMessage{{PhoneNumber: phone, MessageID: result.MessageID}}
	}
	for index, acceptedMessage := range responseMessages {
		acceptedPhone := strings.TrimSpace(acceptedMessage.PhoneNumber)
		if acceptedPhone == "" && index < len(phones) {
			acceptedPhone = phones[index]
		}
		rememberDownstream(downstreamSession{
			messageID:      acceptedMessage.MessageID,
			account:        account,
			tenantID:       result.TenantID,
			applicationID:  result.ApplicationID,
			enterpriseCode: session.enterpriseCode,
			protocol:       clientProtocol,
			srcID:          strings.TrimSpace(req.srcID),
			phoneNumber:    acceptedPhone,
			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,
			protocolLog:    session.protocolLog,
		})
	}
	if current := findSessionByConn(packet.Conn); current != nil && current.report != nil {
		go current.report(current, "submit", "")
	}
	response.AfterSend = func(sendErr error) {
		releaseSubmitBarrier()
		s.emitProtocolLog(protocolLogEvent{
			Protocol:         "cmpp",
			Direction:        "platform_to_client",
			EventType:        "submit_resp",
			Status:           protocolSendStatus(sendErr),
			TenantID:         result.TenantID,
			ApplicationID:    result.ApplicationID,
			Account:          account,
			MessageID:        result.MessageID,
			GatewayMessageID: fmt.Sprint(gatewayMsgID),
			Phone:            phone,
			ResultCode:       protocolSendResultCode(sendErr, 0),
			Detail:           protocolSubmitResponseDetail(req.sequenceID, sendErr),
		})
		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 dest_count=%d accepted_count=%d 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, len(phones), len(responseMessages), result.MessageID, gatewayMsgID, time.Since(startedAt).Milliseconds(), len([]rune(content)), contentHash,
	)
	return false, nil
}

func (s Server) submitResponseProtocolLogger(
	account string,
	protocol string,
	sequenceID uint32,
	phone string,
	messageID string,
	gatewayMessageID uint64,
	result uint32,
) func(error) {
	return func(sendErr error) {
		s.emitProtocolLog(protocolLogEvent{
			Protocol:         "cmpp",
			Direction:        "platform_to_client",
			EventType:        "submit_resp",
			Status:           protocolSendStatus(sendErr),
			Account:          account,
			MessageID:        messageID,
			GatewayMessageID: fmt.Sprint(gatewayMessageID),
			Phone:            phone,
			ResultCode:       protocolSendResultCode(sendErr, result),
			Detail:           protocolSubmitResponseDetail(sequenceID, sendErr),
		})
	}
}

func protocolSendStatus(sendErr error) string {
	if sendErr != nil {
		return "failed"
	}
	return "success"
}

func protocolSendResultCode(sendErr error, result uint32) string {
	if sendErr != nil {
		return "SEND_FAILED"
	}
	return fmt.Sprint(result)
}

func protocolSubmitResponseDetail(sequenceID uint32, sendErr error) map[string]any {
	detail := map[string]any{"sequenceId": sequenceID}
	if sendErr != nil {
		detail["error"] = sendErr.Error()
	}
	return detail
}

func (s Server) emitProtocolLog(event protocolLogEvent) {
	go func() {
		ctx, cancel := context.WithTimeout(context.Background(), defaultHTTPTimeout)
		defer cancel()
		if err := s.post(ctx, "/gateway/events/protocol-log", event, nil); err != nil {
			log.Printf("cmpp inbound protocol_event direction=%s event=%s status=telemetry_failed account=%s message_id=%s error=%q", event.Direction, event.EventType, event.Account, event.MessageID, err)
		}
	}()
}

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 s.reportConnectionOrDisconnect(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) {
	_ = s.reportConnectionChecked(session, status, errorMessage)
}

func (s Server) reportConnectionChecked(session *downstreamSession, status string, errorMessage string) error {
	if session == nil || strings.TrimSpace(session.account) == "" || strings.TrimSpace(session.connectionID) == "" {
		return nil
	}
	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)
		return err
	}
	return nil
}

func (s Server) reportConnectionOrDisconnect(session *downstreamSession, status string, errorMessage string) {
	if err := s.reportConnectionChecked(session, status, errorMessage); err != nil && status != "disconnected" {
		_ = s.reportConnectionChecked(session, "disconnected", err.Error())
		forgetDownstream(session)
		if session.conn != nil {
			session.conn.Close()
		}
	}
}

func (s Server) handleConnectionClosed(conn *cmpp.Conn) {
	downstreamSubmitBarrier.Lock()
	delete(downstreamSubmitBarrier.byConn, conn)
	downstreamSubmitBarrier.Unlock()
	session := findSessionByConn(conn)
	if session == nil {
		return
	}
	forgetDownstream(session)
	_ = s.reportConnectionChecked(session, "disconnected", "CMPP client connection closed")
}

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
	tpUdhi          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, tpUdhi: req.TpUdhi, 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, tpUdhi: req.TpUdhi, 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]any{
				"id": delivery.ID, "errorMessage": err.Error(), "failureType": "send_failed",
				"connectionId": sendResult.ConnectionID, "sequenceId": sendResult.SequenceID,
				"messageId": sendResult.MessageID, "sentAt": sendResult.SentAt,
			}, nil)
			continue
		}
		if sendResult.Sent {
			result.DeliveredCount++
			continue
		}
		if sendResult.ReasonCode == "SUBMIT_RESPONSE_PENDING" {
			result.WaitingCount++
			continue
		}
		errorMessage := defaultString(sendResult.ErrorMessage, "gateway did not complete downstream delivery")
		failureType := "unrecoverable"
		if sendResult.Retryable {
			failureType = "send_failed"
			result.WaitingCount++
		} else {
			result.FailedCount++
		}
		result.LastError = errorMessage
		_ = s.post(context.Background(), "/gateway/events/downstream/failed", map[string]any{
			"id": delivery.ID, "errorMessage": errorMessageWithCode(errorMessage, sendResult.ReasonCode),
			"failureType": failureType, "connectionId": sendResult.ConnectionID,
			"sequenceId": sendResult.SequenceID, "messageId": sendResult.MessageID, "sentAt": sendResult.SentAt,
		}, nil)
	}
	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 decodeInboundSubmitContent(req inboundSubmitPacket) (string, *inboundLongMessageFragment, error) {
	raw := []byte(req.msgContent)
	if req.tpUdhi == 0 && req.pkTotal <= 1 {
		content, err := decodeContent(req.msgFmt, req.msgContent)
		return content, nil, err
	}
	if len(raw) == 0 {
		return "", nil, errors.New("UDH message content is empty")
	}

	headerLength := int(raw[0]) + 1
	if headerLength > len(raw) {
		return "", nil, fmt.Errorf("UDH length %d exceeds message content length %d", headerLength, len(raw))
	}

	var reference, total, index int
	switch {
	case len(raw) >= 6 && raw[0] == 0x05 && raw[1] == 0x00 && raw[2] == 0x03:
		reference = int(raw[3])
		total = int(raw[4])
		index = int(raw[5])
	case len(raw) >= 7 && raw[0] == 0x06 && raw[1] == 0x08 && raw[2] == 0x04:
		reference = int(raw[3])<<8 | int(raw[4])
		total = int(raw[5])
		index = int(raw[6])
	default:
		if req.pkTotal > 1 {
			return "", nil, errors.New("concatenated CMPP submit is missing a supported 8-bit or 16-bit UDH")
		}
		content, err := decodeContent(req.msgFmt, string(raw[headerLength:]))
		return content, nil, err
	}
	if total < 2 || index < 1 || index > total {
		return "", nil, fmt.Errorf("invalid concatenated UDH total/index %d/%d", index, total)
	}
	if req.pkTotal > 0 && int(req.pkTotal) != total {
		return "", nil, fmt.Errorf("PkTotal %d does not match UDH total %d", req.pkTotal, total)
	}
	if req.pkNumber > 0 && int(req.pkNumber) != index {
		return "", nil, fmt.Errorf("PkNumber %d does not match UDH index %d", req.pkNumber, index)
	}

	content, err := decodeContent(req.msgFmt, string(raw[headerLength:]))
	if err != nil {
		return "", nil, err
	}
	return content, &inboundLongMessageFragment{
		Reference: reference,
		Total:     total,
		Index:     index,
		Format:    int(req.msgFmt),
	}, 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()
	if existing := downstreamRegistry.byMessageID[session.messageID]; existing != nil && existing.conn == session.conn {
		// A downstream long message returns one SUBMIT_RESP per fragment but is
		// persisted as one platform message. Keep the first fragment Msg_Id so
		// online delivery and restart recovery (which persists the first
		// Sequence_Id) address the same client-side message.
		session.gatewayMsgID = existing.gatewayMsgID
	}
	downstreamRegistry.byMessageID[session.messageID] = &session
	downstreamRegistry.byConn[session.conn] = &session
	if session.account != "" {
		downstreamRegistry.byAccount[session.account] = &session
	}
	downstreamRegistry.Unlock()
}

func rememberAccount(session *downstreamSession, maxConnections int) bool {
	if session == nil || session.account == "" || session.conn == nil {
		return false
	}
	if maxConnections <= 0 {
		maxConnections = 1
	}
	session.touchPresence("connected", false, false)
	downstreamRegistry.Lock()
	defer downstreamRegistry.Unlock()
	active := 0
	for _, current := range downstreamRegistry.byConn {
		if current != nil && current.account == session.account {
			active++
		}
	}
	if active >= maxConnections {
		return false
	}
	downstreamRegistry.byAccount[session.account] = session
	downstreamRegistry.byConn[session.conn] = session
	return true
}

func forgetDownstream(session *downstreamSession) {
	if session == nil {
		return
	}
	downstreamRegistry.Lock()
	for messageID, current := range downstreamRegistry.byMessageID {
		if current != nil && current.conn == session.conn {
			delete(downstreamRegistry.byMessageID, 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)
			case result.FailedCount > 0:
				status.State = "failed"
				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, true)
}

func pushReceiptWithResult(event DownstreamReceipt, allowRecovery bool) (DownstreamSendResult, error) {
	session := findReceiptSession(event.MessageID, event.Account)
	if session == nil && allowRecovery {
		session = recoverReceiptSession(event)
	}
	if session == nil {
		if event.SubmitSequenceID == 0 {
			return DownstreamSendResult{
				Retryable:    false,
				ReasonCode:   "MISSING_SUBMIT_SEQUENCE_ID",
				ErrorMessage: "历史回执缺少原 Submit Sequence_Id，无法重建 Msg_Id，系统已终止重投",
			}, nil
		}
		if strings.TrimSpace(event.MessageID) == "" || strings.TrimSpace(event.Account) == "" {
			return DownstreamSendResult{
				Retryable:    false,
				ReasonCode:   "INVALID_RECEIPT_PAYLOAD",
				ErrorMessage: "状态回执缺少平台消息 ID 或客户账号，系统已终止重投",
			}, nil
		}
		return DownstreamSendResult{
			Retryable:    true,
			ReasonCode:   "CLIENT_DISCONNECTED",
			ErrorMessage: "下游客户端当前未连接，等待自动重试",
		}, nil
	}
	if downstreamSubmitResponsePending(session.conn) {
		return DownstreamSendResult{
			Retryable:    true,
			ReasonCode:   "SUBMIT_RESPONSE_PENDING",
			ErrorMessage: "客户 SubmitResp 尚未完成写出，回执已保留并等待响应后投递",
		}, 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 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 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(defaultString(event.SubmitGroupMessageID, 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{
			Retryable:    true,
			ReasonCode:   "CLIENT_DISCONNECTED",
			ErrorMessage: "下游客户端当前未连接，等待自动重试",
		}, 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 errorMessageWithCode(message string, code string) string {
	message = strings.TrimSpace(message)
	code = strings.TrimSpace(code)
	if message == "" {
		message = "gateway did not complete downstream delivery"
	}
	if code == "" {
		return message
	}
	return fmt.Sprintf("%s (%s)", message, code)
}

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())
	result := DownstreamSendResult{
		ConnectionID:  session.connectionID,
		SequenceID:    strconv.FormatUint(uint64(sequenceID), 10),
		MessageID:     strconv.FormatUint(messageID, 10),
		SentAt:        formatRFC3339Nano(sentAt),
		AckDeadlineAt: formatRFC3339Nano(ackDeadlineAt),
	}
	tracker := registerDownstreamAck(session, deliveryID, sequenceID, messageID, ackDeadlineAt)
	if err := session.conn.SendPkt(deliver, sequenceID); err != nil {
		removeDownstreamAck(tracker)
		session.recordDownstreamProtocol(deliver, deliveryID, sequenceID, messageID, "failed", "SEND_FAILED", err)
		if session.report != nil {
			go session.report(session, "disconnected", err.Error())
		}
		forgetDownstream(session)
		result.Retryable = true
		result.ReasonCode = "SEND_FAILED"
		result.ErrorMessage = err.Error()
		return result, nil
	}
	session.recordDownstreamProtocol(deliver, deliveryID, sequenceID, messageID, "success", "", nil)
	result.Sent = true
	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 (session *downstreamSession) recordDownstreamProtocol(
	deliver cmpp.Packer,
	deliveryID string,
	sequenceID uint32,
	messageID uint64,
	status string,
	resultCode string,
	sendErr error,
) {
	if session == nil || session.protocolLog == nil {
		return
	}
	eventType, phone := downstreamDeliverMetadata(deliver)
	detail := map[string]any{"sequenceId": sequenceID, "deliveryId": deliveryID}
	if sendErr != nil {
		detail["error"] = sendErr.Error()
	}
	session.protocolLog(protocolLogEvent{
		Protocol:         "cmpp",
		Direction:        "platform_to_client",
		EventType:        eventType,
		Status:           status,
		TenantID:         session.tenantID,
		ApplicationID:    session.applicationID,
		Account:          session.account,
		MessageID:        session.messageID,
		GatewayMessageID: strconv.FormatUint(messageID, 10),
		Phone:            defaultString(phone, session.phoneNumber),
		ResultCode:       resultCode,
		Detail:           detail,
	})
}

func downstreamDeliverMetadata(deliver cmpp.Packer) (string, string) {
	switch packet := deliver.(type) {
	case *cmpp.Cmpp2DeliverReqPkt:
		if packet.RegisterDelivery == 1 {
			return "deliver_receipt", packet.SrcTerminalId
		}
		return "deliver_uplink", packet.SrcTerminalId
	case *cmpp.Cmpp3DeliverReqPkt:
		if packet.RegisterDelivery == 1 {
			return "deliver_receipt", packet.SrcTerminalId
		}
		return "deliver_uplink", packet.SrcTerminalId
	default:
		return "deliver", ""
	}
}

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)
		}
		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, 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
}

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))
}