fix: recover gateway callbacks and client send flows

gateway/internal/resultoutbox/outbox.go

@@ -232,9 +232,12 @@ func EventFromStreamValues(values map[string]interface{}) (Event, error) {
 	if err := json.Unmarshal([]byte(data), &event); err != nil {
 		return Event{}, err
 	}
-	if event.SchemaVersion != queue.SchemaVersion || event.EventID == "" || event.MessageID == "" {
+	if event.SchemaVersion != queue.SchemaVersion || event.EventID == "" || event.ChannelID == "" {
 		return Event{}, fmt.Errorf("invalid result Outbox envelope")
 	}
+	if (event.EventType == "submit_result" || event.EventType == "submit_segment_result") && event.MessageID == "" {
+		return Event{}, fmt.Errorf("submit result Outbox messageId is required")
+	}
 	if event.Path != "/gateway/events/submit-result" && event.Path != "/gateway/events/submit-segment-result" && event.Path != "/gateway/events/receipt/intake" && event.Path != "/gateway/events/uplink" && event.Path != "/gateway/events/dead-letter" {
 		return Event{}, fmt.Errorf("unsupported result Outbox path %q", event.Path)
 	}

gateway/internal/resultoutbox/outbox_test.go

@@ -3,6 +3,7 @@ package resultoutbox
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"net/http"
 	"net/http/httptest"
 	"sync/atomic"
@@ -33,6 +34,52 @@ func TestPublishSubmitSegmentIsIdempotent(t *testing.T) {
 	}
 }
 
+func TestUplinkWithoutPlatformMessageIDIsAValidOutboxEvent(t *testing.T) {
+	mr := miniredis.RunT(t)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	outbox := New(client)
+	event := queue.UplinkEvent{
+		Envelope: queue.Envelope{
+			SchemaVersion: queue.SchemaVersion,
+			MessageType:   queue.MessageTypeUplinkEvent,
+			ChannelID:     "channel-1",
+			CreatedAt:     time.Now().UTC(),
+		},
+		SequenceID: 1, PhoneNumber: "13800138000", DestID: "10690001",
+		Content: "UP", ReceivedAt: time.Now().UTC(),
+	}
+	if err := outbox.PublishUplink(context.Background(), event); err != nil {
+		t.Fatalf("publish uplink: %v", err)
+	}
+	streams, err := client.XRange(context.Background(), outbox.StreamName(), "-", "+").Result()
+	if err != nil || len(streams) != 1 {
+		t.Fatalf("read uplink stream: entries=%d err=%v", len(streams), err)
+	}
+	parsed, err := EventFromStreamValues(streams[0].Values)
+	if err != nil {
+		t.Fatalf("parse uplink without messageId: %v", err)
+	}
+	if parsed.EventType != "uplink" || parsed.MessageID != "" {
+		t.Fatalf("unexpected uplink envelope: %+v", parsed)
+	}
+}
+
+func TestRetryUntilCanceledRestartsAfterTransientFailure(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	calls := 0
+	err := retryUntilCanceled(ctx, time.Millisecond, func() error {
+		calls++
+		if calls == 1 {
+			return errors.New("transient Redis read timeout")
+		}
+		cancel()
+		return ctx.Err()
+	})
+	if !errors.Is(err, context.Canceled) || calls != 2 {
+		t.Fatalf("retry result err=%v calls=%d, want context canceled after 2 calls", err, calls)
+	}
+}
+
 func TestBatchCallbackSendsMultipleEventsInOneRequest(t *testing.T) {
 	mr := miniredis.RunT(t)
 	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})

gateway/internal/resultoutbox/worker.go

@@ -43,12 +43,24 @@ func (o *Outbox) Run(ctx context.Context) error {
 	if strings.TrimSpace(o.APIBaseURL) == "" {
 		return fmt.Errorf("result Outbox API base URL is required")
 	}
+	if o.BatchEnabled {
+		return retryUntilCanceled(ctx, time.Second, func() error {
+			if err := o.ensureGroup(ctx); err != nil {
+				log.Printf("gateway result Outbox group initialization failed: %v", err)
+				return err
+			}
+			if err := o.runBatches(ctx); err != nil && ctx.Err() == nil {
+				// A transient Redis read error must not permanently stop the only
+				// consumer for submit results, receipts, and uplink events.
+				log.Printf("gateway result Outbox batch consume failed: %v", err)
+				return err
+			}
+			return ctx.Err()
+		})
+	}
 	if err := o.ensureGroup(ctx); err != nil {
 		return err
 	}
-	if o.BatchEnabled {
-		return o.runBatches(ctx)
-	}
 	pool := newCallbackPool(ctx, o, o.concurrency())
 	defer pool.wait()
 	for {
@@ -276,3 +288,12 @@ func sleep(ctx context.Context, duration time.Duration) {
 	case <-timer.C:
 	}
 }
+
+func retryUntilCanceled(ctx context.Context, delay time.Duration, operation func() error) error {
+	for ctx.Err() == nil {
+		if err := operation(); err != nil && ctx.Err() == nil {
+			sleep(ctx, delay)
+		}
+	}
+	return ctx.Err()
+}