broker

func (*MemoryBroker) Publish(ctx context.Context, topic string, payload []byte) error
{
	m.subMu.RLock()
	handlers := m.subs[topic]
	m.subMu.RUnlock()

	for _, h := range handlers {
		go func(fn func([]byte) error) {
			_ = fn(payload)
		}(h)
	}
	return nil
}

func (*NATSBroker) Subscribe(topic string, handler func([]byte) error) error
{
	ctx := context.Background()
	streamName := sanitizeStreamName(topic)

	_, err := b.js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
		Name:     streamName,
		Subjects: []string{topic},
	})
	if err != nil {
		return fmt.Errorf("create stream: %w", err)
	}

	consumer, err := b.js.CreateOrUpdateConsumer(ctx, streamName, jetstream.ConsumerConfig{
		Durable:       b.group,
		FilterSubject: topic,
		AckPolicy:     jetstream.AckExplicitPolicy,
		AckWait:       b.ackWait,
		MaxDeliver:    b.maxDeliver,
	})
	if err != nil {
		return fmt.Errorf("create consumer: %w", err)
	}

	_, err = consumer.Consume(func(msg jetstream.Msg) {
		if err := handler(msg.Data()); err != nil {
			_ = msg.Nak()
			return
		}
		_ = msg.Ack()
	})
	return err
}

func (*RedisBroker) readLoop(topic string, handler func([]byte) error)
{
	dead := topic + ".dead"
	r.reclaimPending(topic, dead, handler)

	for {
		select {
		case <-r.ctx.Done():
			return
		default:
		}

		streams, err := r.client.XReadGroup(r.ctx, &redis.XReadGroupArgs{
			Group:    r.group,
			Consumer: r.consumer,
			Streams:  []string{topic, ">"},
			Count:    10,
			Block:    r.blockTimeout,
		}).Result()
		if err != nil {
			if r.ctx.Err() != nil {
				return
			}
			continue
		}

		for _, stream := range streams {
			for _, msg := range stream.Messages {
				r.processMessage(topic, dead, msg, handler)
			}
		}
	}
}

func (*RedisBroker) processMessage(topic, dead string, msg redis.XMessage, handler func([]byte) error)
{
	payload := extractPayload(msg)

	if err := handler(payload); err == nil {
		r.client.XAck(r.ctx, topic, r.group, msg.ID)
		return
	}

	pending, err := r.client.XPendingExt(r.ctx, &redis.XPendingExtArgs{
		Stream:   topic,
		Group:    r.group,
		Start:    msg.ID,
		End:      msg.ID,
		Count:    1,
		Consumer: r.consumer,
	}).Result()
	if err != nil || len(pending) == 0 {
		return
	}

	if int(pending[0].RetryCount) >= r.maxRetries {
		r.client.XAdd(r.ctx, &redis.XAddArgs{
			Stream: dead,
			Values: map[string]any{"payload": payload, "original_id": msg.ID},
		})
		r.client.XAck(r.ctx, topic, r.group, msg.ID)
	}
}

func (*RedisBroker) reclaimPending(topic, dead string, handler func([]byte) error)
{
	start := "0-0"
	for {
		msgs, next, err := r.client.XAutoClaim(r.ctx, &redis.XAutoClaimArgs{
			Stream:   topic,
			Group:    r.group,
			Consumer: r.consumer,
			MinIdle:  r.claimTimeout,
			Start:    start,
			Count:    10,
		}).Result()
		if err != nil {
			return
		}
		for _, msg := range msgs {
			r.processMessage(topic, dead, msg, handler)
		}
		if next == "0-0" || next == "" || len(msgs) < 10 {
			return
		}
		start = next
	}
}