sxchnge/backend/shared/messaging/reliable/transport.go

package reliable

import (
	"context"
	"strings"
	"time"

	"git.sendico.io/tech/sxchnge/shared/errors"
)

const ReservedHeaderPrefix = "sxchnge-runtime-"

const (
	HeaderTransportMessageID       = ReservedHeaderPrefix + "transport-message-id"
	HeaderOriginalSubject          = ReservedHeaderPrefix + "original-subject"
	HeaderOriginalBusinessMsgID    = ReservedHeaderPrefix + "original-business-message-id"
	HeaderOriginalTransportMsgID   = ReservedHeaderPrefix + "original-transport-message-id"
	HeaderFirstSeenAt              = ReservedHeaderPrefix + "first-seen-at"
	HeaderLastDeliveryAt           = ReservedHeaderPrefix + "last-delivery-at"
	HeaderDeliveryAttempts         = ReservedHeaderPrefix + "delivery-attempts"
	HeaderConsumerDurable          = ReservedHeaderPrefix + "consumer"
	HeaderParkReason               = ReservedHeaderPrefix + "park-reason"
	HeaderParkedRecordID           = ReservedHeaderPrefix + "parked-record-id"
	HeaderReplaySource             = ReservedHeaderPrefix + "replay-source"
	HeaderReplayAttempt            = ReservedHeaderPrefix + "replay-attempt"
	HeaderReplayCount              = ReservedHeaderPrefix + "replay-count"
	HeaderLastReplayAt             = ReservedHeaderPrefix + "last-replay-at"
	HeaderLastReplayTransportMsgID = ReservedHeaderPrefix + "last-replay-transport-message-id"
	replaySourceParked             = "parked"
)

type Record struct {
	// Subject identifies the logical topic or routing key for the record.
	Subject string
	// Payload carries the application bytes. Callers and adapters must treat payload data as read-only.
	Payload []byte
	// Headers carries case-insensitive metadata. Public publishes normalize keys to lowercase, preserve per-key value order,
	// treat nil and empty maps as equivalent, drop empty value slices, and reject caller-supplied keys that use ReservedHeaderPrefix.
	Headers map[string][]string
	// MessageID is optional and caller-supplied. When non-empty, it is the logical message identity used for correlation and
	// redelivery contract checks, and runtime-owned durable paths must preserve it unchanged end to end.
	MessageID string
}

// Transport is the durable data-plane capability.
// Publish accepts public records only.
// OpenConsumer returns a subscription whose deliveries obey the Delivery and Subscription contracts below.
type Transport interface {
	Publish(ctx context.Context, record Record) error
	OpenConsumer(ctx context.Context, settings DurableSettings, consumer DurableConsumerSettings) (Subscription, error)
}

// Admin provisions durable transport resources.
// EnsureProducerResources and EnsureConsumerResources must be idempotent.
type Admin interface {
	EnsureProducerResources(ctx context.Context, settings DurableSettings) error
	EnsureConsumerResources(ctx context.Context, settings DurableSettings, consumer DurableConsumerSettings) error
}

// Subscription fetches durable deliveries.
// Fetch rejects limit <= 0 with merrors.InvalidArgument.
// Fetch may return fewer than limit deliveries, including zero deliveries with nil error, when fewer records are available
// or the wait window expires.
// Close is idempotent and concurrency-safe. After Close, Fetch returns capability.ErrClosed and no further successful
// Fetch calls are allowed.
type Subscription interface {
	Fetch(ctx context.Context, limit int, wait time.Duration) ([]Delivery, error)
	Close() error
}

// Delivery represents one delivery attempt of a durable record.
// Attempt is 1-based.
// A redelivery is any later delivery of the same logical message after a prior delivery was not successfully acked.
// When Record.MessageID is non-empty, that exact value defines logical message identity for contract purposes.
// Ack is idempotent on the same delivery instance. Repeated calls after a successful ack return nil.
// If a delivery is not acked, it remains eligible for redelivery under adapter policy.
// After Subscription.Close, Ack on a previously fetched but unacked delivery returns capability.ErrClosed unless it already succeeded earlier.
type Delivery interface {
	Record() Record
	Attempt() int
	Ack(ctx context.Context) error
}

type runtimePublisher interface {
	PublishRuntime(ctx context.Context, record Record) error
}

func publishRuntimeRecord(ctx context.Context, transport Transport, record Record) error {
	if transport == nil {
		return nil
	}
	if publisher, ok := transport.(runtimePublisher); ok {
		return publisher.PublishRuntime(ctx, record)
	}
	return transport.Publish(ctx, record)
}

func normalizeRecord(record Record, allowReserved bool) (Record, error) {
	record.Subject = strings.TrimSpace(record.Subject)
	record.MessageID = strings.TrimSpace(record.MessageID)
	if record.Subject == "" {
		return Record{}, merrors.InvalidArgument("record subject is required", "record.subject")
	}
	record.Headers = normalizeHeaders(record.Headers, allowReserved)
	return record, nil
}

func NormalizePublicRecord(record Record) (Record, error) {
	if err := validatePublicRecord(record); err != nil {
		return Record{}, err
	}
	return normalizeRecord(record, false)
}

func NormalizeRuntimeRecord(record Record) (Record, error) {
	return normalizeRecord(record, true)
}

func normalizeHeaders(headers map[string][]string, allowReserved bool) map[string][]string {
	if len(headers) == 0 {
		return nil
	}
	result := make(map[string][]string, len(headers))
	for key, values := range headers {
		key = normalizeHeaderKey(key)
		if key == "" {
			continue
		}
		if !allowReserved && strings.HasPrefix(key, ReservedHeaderPrefix) {
			continue
		}
		normalized := make([]string, 0, len(values))
		for _, value := range values {
			if trimmed := strings.TrimSpace(value); trimmed != "" {
				normalized = append(normalized, trimmed)
			}
		}
		if len(normalized) == 0 {
			continue
		}
		result[key] = append(result[key], normalized...)
	}
	if len(result) == 0 {
		return nil
	}
	return result
}

func validatePublicRecord(record Record) error {
	for key := range record.Headers {
		normalized := normalizeHeaderKey(key)
		if normalized == "" {
			return merrors.InvalidArgument("record header key is empty", "record.headers")
		}
		if strings.HasPrefix(normalized, ReservedHeaderPrefix) {
			return merrors.InvalidArgument("record header key uses reserved runtime prefix", "record.headers")
		}
	}
	return nil
}

func ValidatePublicRecord(record Record) error {
	return validatePublicRecord(record)
}

func normalizeHeaderKey(key string) string {
	return strings.ToLower(strings.TrimSpace(key))
}

func cloneHeaderMap(headers map[string][]string) map[string][]string {
	if len(headers) == 0 {
		return nil
	}
	result := make(map[string][]string, len(headers))
	for key, values := range headers {
		result[key] = append([]string(nil), values...)
	}
	return result
}

func CloneHeaders(headers map[string][]string) map[string][]string {
	return cloneHeaderMap(headers)
}

func headerValue(headers map[string][]string, key string) string {
	if len(headers) == 0 {
		return ""
	}
	values := headers[normalizeHeaderKey(key)]
	if len(values) == 0 {
		return ""
	}
	return strings.TrimSpace(values[0])
}

func HeaderValue(headers map[string][]string, key string) string {
	return headerValue(headers, key)
}