feat: add report material workflows and gateway safeguards

gateway/cmd/gateway/main.go

@@ -9,6 +9,7 @@ import (
 	"cmpp-platform/gateway/internal/control"
 	"cmpp-platform/gateway/internal/health"
 	"cmpp-platform/gateway/internal/inbound"
+	"cmpp-platform/gateway/internal/ratelimit"
 	"cmpp-platform/gateway/internal/submitworker"
 	"cmpp-platform/gateway/internal/upstream"
 )
@@ -24,6 +25,10 @@ func main() {
 	}
 	apiBaseURL := os.Getenv("API_BASE_URL")
 	upstreamManager := &upstream.Manager{APIBaseURL: apiBaseURL}
+	channelLimiter, err := ratelimit.New(os.Getenv("REDIS_URL"))
+	if err != nil {
+		log.Fatalf("gateway channel rate limiter init failed: %v", err)
+	}
 	presenceStore, err := inbound.NewRedisPresenceStore(os.Getenv("REDIS_URL"))
 	if err != nil {
 		log.Printf("gateway downstream presence store init failed: %v", err)
@@ -69,6 +74,7 @@ func main() {
 	control.Register(mux, control.Server{
 		APIBaseURL: apiBaseURL,
 		Upstream:   upstreamManager,
+		Limiter:    channelLimiter,
 		RecoveryCandidates: func(ctx context.Context) ([]inbound.DownstreamPresence, error) {
 			return inbound.ListRecoveryCandidates(ctx, presenceStore)
 		},

gateway/internal/control/server.go

@@ -9,10 +9,12 @@ import (
 
 	"cmpp-platform/gateway/internal/inbound"
 	"cmpp-platform/gateway/internal/queue"
+	"cmpp-platform/gateway/internal/ratelimit"
 	"cmpp-platform/gateway/internal/upstream"
 )
 
 type ConnectFunc func(context.Context, ConnectChannelCommand) (ConnectionStateCallback, error)
+type SubmitFunc func(context.Context, queue.SubmitCommand) (queue.SubmitResult, error)
 
 type ConnectChannelCommand struct {
 	SchemaVersion      string        `json:"schemaVersion"`
@@ -54,7 +56,9 @@ type Server struct {
 	APIBaseURL         string
 	HTTPClient         *http.Client
 	Connect            ConnectFunc
+	Submit             SubmitFunc
 	Upstream           *upstream.Manager
+	Limiter            ratelimit.Limiter
 	RecoveryCandidates func(context.Context) ([]inbound.DownstreamPresence, error)
 	RecoveryStatuses   func(context.Context) ([]inbound.DownstreamRecoveryStatus, error)
 }
@@ -74,6 +78,9 @@ func Register(mux *http.ServeMux, server Server) {
 	if server.Connect == nil {
 		server.Connect = server.connectChannel
 	}
+	if server.Submit == nil {
+		server.Submit = server.Upstream.Submit
+	}
 	mux.HandleFunc("/connections/connect", server.handleConnectChannel)
 	mux.HandleFunc("/upstream/submit", server.handleUpstreamSubmit)
 	mux.HandleFunc("/downstream/receipt", server.handleDownstreamReceipt)
@@ -98,6 +105,12 @@ func (s Server) handleConnectChannel(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
+	if configurer, ok := s.Limiter.(ratelimit.Configurer); ok {
+		if err := configurer.Configure(r.Context(), command.ChannelID, command.Channel.RateLimitPerSecond); err != nil {
+			http.Error(w, fmt.Sprintf("failed to configure gateway channel rate limit: %v", err), http.StatusServiceUnavailable)
+			return
+		}
+	}
 
 	status, err := s.Connect(r.Context(), command)
 	if err != nil {
@@ -119,7 +132,13 @@ func (s Server) handleUpstreamSubmit(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, fmt.Sprintf("invalid submit command: %v", err), http.StatusBadRequest)
 		return
 	}
-	result, err := s.Upstream.Submit(r.Context(), command)
+	if s.Limiter != nil {
+		if _, err := s.Limiter.Wait(r.Context(), command.ChannelID, command.Route.RateLimitPerSecond); err != nil {
+			http.Error(w, fmt.Sprintf("gateway channel rate limit unavailable: %v", err), http.StatusServiceUnavailable)
+			return
+		}
+	}
+	result, err := s.Submit(r.Context(), command)
 	if err != nil {
 		w.WriteHeader(http.StatusBadGateway)
 		_ = json.NewEncoder(w).Encode(result)

gateway/internal/control/server_test.go

@@ -10,10 +10,57 @@ import (
 	"time"
 
 	"cmpp-platform/gateway/internal/inbound"
+	"cmpp-platform/gateway/internal/queue"
 )
 
+type controlRecordingLimiter struct {
+	channelID           string
+	rate                int
+	configuredChannelID string
+	configuredRate      int
+}
+
+func (l *controlRecordingLimiter) Wait(_ context.Context, channelID string, rate int) (time.Duration, error) {
+	l.channelID = channelID
+	l.rate = rate
+	return 0, nil
+}
+
+func (l *controlRecordingLimiter) Configure(_ context.Context, channelID string, rate int) error {
+	l.configuredChannelID = channelID
+	l.configuredRate = rate
+	return nil
+}
+
+func TestUpstreamSubmitUsesGatewayChannelLimiter(t *testing.T) {
+	limiter := &controlRecordingLimiter{}
+	handler := handlerWithServer(Server{
+		Limiter: limiter,
+		Submit: func(_ context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) {
+			return queue.SubmitResult{Envelope: command.Envelope, SubmitStatus: "accepted"}, nil
+		},
+	})
+	resp := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodPost, "/upstream/submit", strings.NewReader(`{
+		"schemaVersion":"v1","messageType":"SubmitCommand","messageId":"msg-1","channelId":"channel-1",
+		"submitId":"submit-1","tenantId":"tenant-1","applicationId":"app-1","phoneNumber":"13800138000","content":"hello",
+		"route":{"channelCode":"CMPP-A","cmppAccountCode":"sp","rateLimitPerSecond":100},
+		"cmpp":{"serviceId":"SMS","srcId":"10690000","registeredDelivery":1,"msgFmt":8},
+		"upstream":{"gatewayHost":"127.0.0.1","gatewayPort":17890,"account":"sp","passwordCipher":"secret","cmppVersion":"3.0"},
+		"retry":{"attempt":0,"maxAttempts":1}
+	}`))
+	handler.ServeHTTP(resp, req)
+	if resp.Code != http.StatusOK {
+		t.Fatalf("unexpected response status: %d body=%s", resp.Code, resp.Body.String())
+	}
+	if limiter.channelID != "channel-1" || limiter.rate != 100 {
+		t.Fatalf("unexpected limiter call: %+v", limiter)
+	}
+}
+
 func TestConnectChannelCallbacksConnectedState(t *testing.T) {
-	handler := handlerWithConnect(func(context.Context, ConnectChannelCommand) (ConnectionStateCallback, error) {
+	limiter := &controlRecordingLimiter{}
+	handler := handlerWithServer(Server{Limiter: limiter, Connect: func(context.Context, ConnectChannelCommand) (ConnectionStateCallback, error) {
 		return ConnectionStateCallback{
 			ChannelID:          "channel-1",
 			ConnectionID:       "channel-1:primary",
@@ -23,7 +70,7 @@ func TestConnectChannelCallbacksConnectedState(t *testing.T) {
 			LastConnectedAt:    time.Now().UTC().Format(time.RFC3339Nano),
 			LastHeartbeatAt:    time.Now().UTC().Format(time.RFC3339Nano),
 		}, nil
-	})
+	}})
 
 	resp := httptest.NewRecorder()
 	req := httptest.NewRequest(http.MethodPost, "/connections/connect", strings.NewReader(validConnectCommand()))
@@ -45,6 +92,9 @@ func TestConnectChannelCallbacksConnectedState(t *testing.T) {
 	if callback.LastConnectedAt == "" || callback.LastHeartbeatAt == "" {
 		t.Fatalf("expected connection timestamps: %+v", callback)
 	}
+	if limiter.configuredChannelID != "channel-1" || limiter.configuredRate != 100 {
+		t.Fatalf("unexpected configured limiter: %+v", limiter)
+	}
 }
 
 func TestConnectChannelCallbacksFailedState(t *testing.T) {

gateway/internal/ratelimit/limiter.go

@@ -0,0 +1,133 @@
+package ratelimit
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+const (
+	defaultRatePerSecond = 100
+	defaultKeyPrefix     = "rate:gateway:channel:"
+)
+
+// Limiter reserves one distributed, per-channel submit slot and waits until it
+// becomes available. The reservation is stored in Redis so multiple Gateway
+// instances share one supplier TPS budget.
+type Limiter interface {
+	Wait(context.Context, string, int) (time.Duration, error)
+}
+
+// Configurer stores the authoritative channel limit received with the channel
+// connection command. Wait always applies the lower of this value and the
+// message value, so a producer cannot raise the supplier TPS ceiling.
+type Configurer interface {
+	Configure(context.Context, string, int) error
+}
+
+type RedisLimiter struct {
+	Redis     *redis.Client
+	KeyPrefix string
+}
+
+var reserveScript = redis.NewScript(`
+local now = redis.call('TIME')
+local now_us = (tonumber(now[1]) * 1000000) + tonumber(now[2])
+local requested_rate = tonumber(ARGV[1])
+local configured_rate = tonumber(redis.call('GET', KEYS[2]))
+local effective_rate = requested_rate
+if configured_rate and configured_rate > 0 and configured_rate < effective_rate then
+  effective_rate = configured_rate
+end
+local interval_us = math.ceil(1000000 / effective_rate)
+local next_us = tonumber(redis.call('GET', KEYS[1])) or now_us
+if next_us < now_us then
+  next_us = now_us
+end
+local delay_us = next_us - now_us
+local reserved_until_us = next_us + interval_us
+local ttl_ms = math.ceil((reserved_until_us - now_us) / 1000) + 1000
+redis.call('PSETEX', KEYS[1], ttl_ms, reserved_until_us)
+return delay_us
+`)
+
+func New(redisURL string) (*RedisLimiter, error) {
+	if strings.TrimSpace(redisURL) == "" {
+		redisURL = "redis://127.0.0.1:6379"
+	}
+	options, err := redis.ParseURL(redisURL)
+	if err != nil {
+		return nil, err
+	}
+	return NewWithClient(redis.NewClient(options)), nil
+}
+
+func NewWithClient(client *redis.Client) *RedisLimiter {
+	return &RedisLimiter{Redis: client, KeyPrefix: defaultKeyPrefix}
+}
+
+func (l *RedisLimiter) Configure(ctx context.Context, channelID string, ratePerSecond int) error {
+	if l == nil || l.Redis == nil {
+		return fmt.Errorf("gateway rate limiter Redis client is required")
+	}
+	channelID = strings.TrimSpace(channelID)
+	if channelID == "" {
+		return fmt.Errorf("gateway rate limiter channelId is required")
+	}
+	if ratePerSecond <= 0 {
+		ratePerSecond = defaultRatePerSecond
+	}
+	if err := l.Redis.Set(ctx, l.configKey(channelID), ratePerSecond, 0).Err(); err != nil {
+		return fmt.Errorf("configure gateway channel rate: %w", err)
+	}
+	return nil
+}
+
+func (l *RedisLimiter) Wait(ctx context.Context, channelID string, ratePerSecond int) (time.Duration, error) {
+	if l == nil || l.Redis == nil {
+		return 0, fmt.Errorf("gateway rate limiter Redis client is required")
+	}
+	channelID = strings.TrimSpace(channelID)
+	if channelID == "" {
+		return 0, fmt.Errorf("gateway rate limiter channelId is required")
+	}
+	if ratePerSecond <= 0 {
+		ratePerSecond = defaultRatePerSecond
+	}
+	delayMicros, err := reserveScript.Run(
+		ctx,
+		l.Redis,
+		[]string{l.key(channelID), l.configKey(channelID)},
+		ratePerSecond,
+	).Int64()
+	if err != nil {
+		return 0, fmt.Errorf("reserve gateway channel rate slot: %w", err)
+	}
+	delay := time.Duration(delayMicros) * time.Microsecond
+	if delay <= 0 {
+		return 0, nil
+	}
+	timer := time.NewTimer(delay)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return delay, ctx.Err()
+	case <-timer.C:
+		return delay, nil
+	}
+}
+
+func (l *RedisLimiter) configKey(channelID string) string {
+	return l.key("config:" + channelID)
+}
+
+func (l *RedisLimiter) key(channelID string) string {
+	prefix := l.KeyPrefix
+	if prefix == "" {
+		prefix = defaultKeyPrefix
+	}
+	return prefix + channelID
+}

gateway/internal/ratelimit/limiter_test.go

@@ -0,0 +1,68 @@
+package ratelimit
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/redis/go-redis/v9"
+)
+
+func TestRedisLimiterSharesOneChannelBudget(t *testing.T) {
+	mr := miniredis.RunT(t)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	limiter := NewWithClient(client)
+
+	first, err := limiter.Wait(context.Background(), "channel-a", 20)
+	if err != nil {
+		t.Fatalf("first wait: %v", err)
+	}
+	second, err := limiter.Wait(context.Background(), "channel-a", 20)
+	if err != nil {
+		t.Fatalf("second wait: %v", err)
+	}
+	if first != 0 {
+		t.Fatalf("first delay = %v, want 0", first)
+	}
+	if second < 40*time.Millisecond {
+		t.Fatalf("second delay = %v, want a shared channel delay", second)
+	}
+}
+
+func TestRedisLimiterUsesIndependentChannelBudgets(t *testing.T) {
+	mr := miniredis.RunT(t)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	limiter := NewWithClient(client)
+
+	if _, err := limiter.Wait(context.Background(), "channel-a", 1); err != nil {
+		t.Fatalf("channel a wait: %v", err)
+	}
+	delay, err := limiter.Wait(context.Background(), "channel-b", 1)
+	if err != nil {
+		t.Fatalf("channel b wait: %v", err)
+	}
+	if delay != 0 {
+		t.Fatalf("channel b delay = %v, want 0", delay)
+	}
+}
+
+func TestRedisLimiterConfiguredRateIsAuthoritativeUpperBound(t *testing.T) {
+	mr := miniredis.RunT(t)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	limiter := NewWithClient(client)
+
+	if err := limiter.Configure(context.Background(), "channel-a", 10); err != nil {
+		t.Fatalf("configure: %v", err)
+	}
+	if _, err := limiter.Wait(context.Background(), "channel-a", 1000); err != nil {
+		t.Fatalf("first wait: %v", err)
+	}
+	delay, err := limiter.Wait(context.Background(), "channel-a", 1000)
+	if err != nil {
+		t.Fatalf("second wait: %v", err)
+	}
+	if delay < 90*time.Millisecond {
+		t.Fatalf("delay = %v, want configured 10 TPS ceiling", delay)
+	}
+}

gateway/internal/submitworker/worker.go

@@ -9,9 +9,11 @@ import (
 	"log"
 	"net/http"
 	"strings"
+	"sync"
 	"time"
 
 	"cmpp-platform/gateway/internal/queue"
+	"cmpp-platform/gateway/internal/ratelimit"
 	"cmpp-platform/gateway/internal/upstream"
 
 	"github.com/redis/go-redis/v9"
@@ -28,6 +30,7 @@ const (
 type Worker struct {
 	Redis            *redis.Client
 	Upstream         *upstream.Manager
+	Limiter          ratelimit.Limiter
 	Submit           func(context.Context, queue.SubmitCommand) (queue.SubmitResult, error)
 	ReportDeadLetter func(context.Context, DeadLetterEvent) error
 	Stream           string
@@ -64,7 +67,7 @@ func New(redisURL string, manager *upstream.Manager) (*Worker, error) {
 	if err != nil {
 		return nil, err
 	}
-	return &Worker{Redis: client, Upstream: manager}, nil
+	return &Worker{Redis: client, Upstream: manager, Limiter: ratelimit.NewWithClient(client)}, nil
 }
 
 func (w *Worker) Run(ctx context.Context) error {
@@ -163,12 +166,18 @@ func (w *Worker) recoverPending(ctx context.Context) error {
 }
 
 func (w *Worker) processMessages(ctx context.Context, messages []redis.XMessage) error {
+	var group sync.WaitGroup
 	for _, message := range messages {
-		if err := w.processMessage(ctx, message); err != nil {
-			w.logf("gateway submit worker message %s failed: %v", message.ID, err)
-			continue
-		}
+		message := message
+		group.Add(1)
+		go func() {
+			defer group.Done()
+			if err := w.processMessage(ctx, message); err != nil {
+				w.logf("gateway submit worker message %s failed: %v", message.ID, err)
+			}
+		}()
 	}
+	group.Wait()
 	return nil
 }
 
@@ -178,6 +187,9 @@ func (w *Worker) processMessage(ctx context.Context, message redis.XMessage) err
 		return w.deadLetterMalformedMessage(ctx, message, err)
 	}
 	if err := w.handleCommand(ctx, command); err != nil {
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
 		attempts, attemptsErr := w.incrementFailureAttempt(ctx, message.ID)
 		if attemptsErr != nil {
 			w.logf("gateway submit worker increment failure %s failed: %v", message.ID, attemptsErr)
@@ -194,6 +206,11 @@ func (w *Worker) processMessage(ctx context.Context, message redis.XMessage) err
 }
 
 func (w *Worker) handleCommand(ctx context.Context, command queue.SubmitCommand) error {
+	if w.Limiter != nil {
+		if _, err := w.Limiter.Wait(ctx, command.ChannelID, command.Route.RateLimitPerSecond); err != nil {
+			return err
+		}
+	}
 	submit := w.Submit
 	if submit == nil {
 		if w.Upstream == nil {

gateway/internal/submitworker/worker_test.go

@@ -3,12 +3,26 @@ package submitworker
 import (
 	"context"
 	"testing"
+	"time"
 
 	"cmpp-platform/gateway/internal/queue"
 	"github.com/alicebob/miniredis/v2"
 	"github.com/redis/go-redis/v9"
 )
 
+type recordingLimiter struct {
+	channelID string
+	rate      int
+	called    bool
+}
+
+func (l *recordingLimiter) Wait(_ context.Context, channelID string, rate int) (time.Duration, error) {
+	l.called = true
+	l.channelID = channelID
+	l.rate = rate
+	return 0, nil
+}
+
 func TestCommandFromStreamValuesParsesSubmitCommand(t *testing.T) {
 	command, err := CommandFromStreamValues(map[string]interface{}{
 		"messageType": "SubmitCommand",
@@ -70,20 +84,22 @@ func TestCommandFromStreamValuesRejectsMissingData(t *testing.T) {
 
 func TestHandleMessageUsesInjectedSubmit(t *testing.T) {
 	var got queue.SubmitCommand
+	limiter := &recordingLimiter{}
 	worker := &Worker{
+		Limiter: limiter,
 		Submit: func(_ context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) {
 			got = command
 			return queue.SubmitResult{SubmitStatus: "accepted"}, nil
 		},
 	}
 	if err := worker.handleCommand(context.Background(), queue.SubmitCommand{
-		Envelope:      queue.Envelope{MessageID: "msg-worker-0002"},
+		Envelope:      queue.Envelope{MessageID: "msg-worker-0002", ChannelID: "channel-1"},
 		SubmitID:      "submit-2",
 		PhoneNumber:   "13800138000",
 		Content:       "hello",
 		Upstream:      queue.UpstreamConfig{GatewayHost: "127.0.0.1", GatewayPort: 17890, Account: "account-a", PasswordCipher: "secret", CMPPVersion: "3.0"},
 		CMPP:          queue.CMPP{ServiceID: "SMS", SrcID: "10690000", RegisteredDelivery: 1, MsgFmt: 8},
-		Route:         queue.Route{ChannelCode: "CMPP-A"},
+		Route:         queue.Route{ChannelCode: "CMPP-A", RateLimitPerSecond: 320},
 		Retry:         queue.Retry{Attempt: 0, MaxAttempts: 1},
 		ApplicationID: "app-1",
 		TenantID:      "tenant-1",
@@ -93,6 +109,9 @@ func TestHandleMessageUsesInjectedSubmit(t *testing.T) {
 	if got.MessageID != "msg-worker-0002" || got.SubmitID != "submit-2" {
 		t.Fatalf("unexpected command: %+v", got)
 	}
+	if !limiter.called || limiter.channelID != "channel-1" || limiter.rate != 320 {
+		t.Fatalf("unexpected limiter call: %+v", limiter)
+	}
 }
 
 func TestMinIdleDefaultsToThirtySeconds(t *testing.T) {
@@ -102,6 +121,52 @@ func TestMinIdleDefaultsToThirtySeconds(t *testing.T) {
 	}
 }
 
+func TestProcessMessagesDoesNotLetOneChannelBlockAnother(t *testing.T) {
+	mr := miniredis.RunT(t)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	startedA := make(chan struct{})
+	startedB := make(chan struct{})
+	releaseA := make(chan struct{})
+	worker := &Worker{
+		Redis: client,
+		Submit: func(_ context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) {
+			switch command.ChannelID {
+			case "channel-a":
+				close(startedA)
+				<-releaseA
+			case "channel-b":
+				close(startedB)
+			}
+			return queue.SubmitResult{SubmitStatus: "accepted"}, nil
+		},
+	}
+	messages := []redis.XMessage{
+		{ID: "1-0", Values: submitCommandValues("message-a", "channel-a")},
+		{ID: "2-0", Values: submitCommandValues("message-b", "channel-b")},
+	}
+	done := make(chan struct{})
+	go func() {
+		_ = worker.processMessages(context.Background(), messages)
+		close(done)
+	}()
+	select {
+	case <-startedA:
+	case <-time.After(time.Second):
+		t.Fatal("channel-a did not start")
+	}
+	select {
+	case <-startedB:
+	case <-time.After(200 * time.Millisecond):
+		t.Fatal("channel-b was blocked by channel-a")
+	}
+	close(releaseA)
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("message batch did not complete")
+	}
+}
+
 func TestProcessMessageDeadLettersAfterMaxFailures(t *testing.T) {
 	mr := miniredis.RunT(t)
 	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
@@ -178,3 +243,17 @@ func TestProcessMessageDeadLettersAfterMaxFailures(t *testing.T) {
 		t.Fatalf("failure attempt key was not cleared")
 	}
 }
+
+func submitCommandValues(messageID string, channelID string) map[string]interface{} {
+	return map[string]interface{}{
+		"data": `{
+			"schemaVersion":"v1","messageType":"SubmitCommand","traceId":"trace-1",
+			"messageId":"` + messageID + `","channelId":"` + channelID + `","submitId":"submit-1",
+			"tenantId":"tenant-1","applicationId":"app-1","phoneNumber":"13800138000","content":"hello",
+			"route":{"channelCode":"CMPP-A","rateLimitPerSecond":100},
+			"cmpp":{"serviceId":"SMS","srcId":"10690000","registeredDelivery":1,"msgFmt":8},
+			"upstream":{"gatewayHost":"127.0.0.1","gatewayPort":17890,"account":"sp","passwordCipher":"secret","cmppVersion":"3.0"},
+			"retry":{"attempt":0,"maxAttempts":1}
+		}`,
+	}
+}