lislgosms/gateway/internal/resultoutbox/batch.go

package resultoutbox

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strings"
	"time"

	"github.com/redis/go-redis/v9"
)

type callbackBatch struct {
	BatchID           string               `json:"batchId"`
	GatewayInstanceID string               `json:"gatewayInstanceId"`
	CreatedAt         time.Time            `json:"createdAt"`
	Events            []callbackBatchEvent `json:"events"`
}
type callbackBatchEvent struct {
	EventID string          `json:"eventId"`
	Type    string          `json:"type"`
	Payload json.RawMessage `json:"payload"`
}
type callbackBatchResponse struct {
	BatchID string                `json:"batchId"`
	Results []callbackEventResult `json:"results"`
}
type callbackEventResult struct {
	EventID   string `json:"eventId"`
	Accepted  bool   `json:"accepted"`
	Retryable bool   `json:"retryable,omitempty"`
	ErrorCode string `json:"errorCode,omitempty"`
}

func (o *Outbox) runBatches(ctx context.Context) error {
	for ctx.Err() == nil {
		messages, _, err := o.Redis.XAutoClaim(ctx, &redis.XAutoClaimArgs{Stream: o.stream(), Group: o.group(), Consumer: o.consumer(), MinIdle: o.minIdle(), Start: "0-0", Count: int64(o.batchSize())}).Result()
		if err != nil && !errors.Is(err, redis.Nil) {
			return err
		}
		if len(messages) == 0 {
			streams, readErr := o.Redis.XReadGroup(ctx, &redis.XReadGroupArgs{Group: o.group(), Consumer: o.consumer(), Streams: []string{o.stream(), ">"}, Count: int64(o.batchSize()), Block: o.batchWait()}).Result()
			if errors.Is(readErr, redis.Nil) {
				continue
			}
			if readErr != nil {
				return readErr
			}
			for _, stream := range streams {
				messages = append(messages, stream.Messages...)
			}
		}
		if len(messages) == 0 {
			continue
		}
		if err := o.processBatch(ctx, messages); err != nil {
			o.batchRetries.Add(int64(len(messages)))
			sleep(ctx, 100*time.Millisecond)
		}
	}
	return ctx.Err()
}

func (o *Outbox) processBatch(ctx context.Context, messages []redis.XMessage) error {
	batch := callbackBatch{BatchID: fmt.Sprintf("CB-%d", time.Now().UnixNano()), GatewayInstanceID: o.GatewayInstanceID, CreatedAt: time.Now().UTC()}
	byEvent := make(map[string]redis.XMessage, len(messages))
	for _, message := range messages {
		event, err := EventFromStreamValues(message.Values)
		if err != nil {
			_ = o.deadLetter(ctx, message, "INVALID_ENVELOPE", err.Error())
			continue
		}
		batch.Events = append(batch.Events, callbackBatchEvent{EventID: event.EventID, Type: event.EventType, Payload: event.Payload})
		byEvent[event.EventID] = message
	}
	if len(batch.Events) == 0 {
		return nil
	}
	body, err := json.Marshal(batch)
	if err != nil {
		return err
	}
	if len(body) > 1024*1024 {
		return fmt.Errorf("callback batch exceeds 1MB")
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(o.APIBaseURL, "/")+"/gateway/events/batch", bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	client := &http.Client{Timeout: o.httpTimeout()}
	resp, err := client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		data, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
		return fmt.Errorf("batch callback returned %d: %s", resp.StatusCode, strings.TrimSpace(string(data)))
	}
	var result callbackBatchResponse
	if err := json.NewDecoder(io.LimitReader(resp.Body, 1024*1024)).Decode(&result); err != nil {
		return err
	}
	if result.BatchID != batch.BatchID {
		return fmt.Errorf("callback batchId mismatch")
	}
	o.batchRequests.Add(1)
	o.batchEvents.Add(int64(len(batch.Events)))
	seen := make(map[string]struct{}, len(result.Results))
	for _, item := range result.Results {
		message, ok := byEvent[item.EventID]
		if !ok {
			continue
		}
		seen[item.EventID] = struct{}{}
		if item.Accepted {
			if err := o.ackDelete(ctx, message.ID); err != nil {
				return err
			}
			continue
		}
		if !item.Retryable {
			if err := o.deadLetter(ctx, message, item.ErrorCode, "non-retryable callback result"); err != nil {
				return err
			}
		}
	}
	for eventID := range byEvent {
		if _, ok := seen[eventID]; !ok {
			return fmt.Errorf("batch response omitted event %s", eventID)
		}
	}
	return nil
}

func (o *Outbox) ackDelete(ctx context.Context, id string) error {
	return acknowledgeAndDeleteScript.Run(ctx, o.Redis, []string{o.stream()}, o.group(), id).Err()
}
func (o *Outbox) deadLetter(ctx context.Context, message redis.XMessage, code, detail string) error {
	stream := o.DeadLetterStream
	if stream == "" {
		stream = o.stream() + ".dead"
	}
	pipe := o.Redis.TxPipeline()
	pipe.XAdd(ctx, &redis.XAddArgs{Stream: stream, Values: map[string]any{"sourceId": message.ID, "errorCode": code, "detail": detail, "data": fmt.Sprint(message.Values["data"])}})
	pipe.XAck(ctx, o.stream(), o.group(), message.ID)
	pipe.XDel(ctx, o.stream(), message.ID)
	if _, err := pipe.Exec(ctx); err != nil {
		return err
	}
	o.deadLetters.Add(1)
	return nil
}
func (o *Outbox) batchSize() int {
	if o.BatchSize < 1 {
		return 50
	}
	return min(o.BatchSize, 100)
}
func (o *Outbox) batchWait() time.Duration {
	if o.BatchWait <= 0 {
		return 10 * time.Millisecond
	}
	return o.BatchWait
}