feat: add phone frequency controls and modularize codebase
This commit is contained in:
hectorzhao
@@ -0,0 +1,281 @@
|
||||
package inbound
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Recovery only replays API-owned pending deliveries; Redis recovery locks and
|
||||
// presence snapshots prevent multiple Gateway instances from racing the replay.
|
||||
|
||||
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) 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 (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, ","))
|
||||
}
|
||||
Reference in New Issue
Block a user