feat: event-trigger (Вариант A) — TriggerTypeEvent, event-dispatcher, reconcileEvent

2026-03-19 13:44:35 +03:00
parent 1aac3f5093
commit a379091b8a
9 changed files with 534 additions and 6 deletions
@@ -0,0 +1,197 @@
+// Изменено: 2026-03-19
+// Dispatcher — управляет AMQP consumer-ами.
+// Для каждого Trigger{type:event} держит один goroutine-consumer:
+//   - Subscribe(key, queue, targetURL) — открывает consumer на очередь
+//   - Unsubscribe(key)                 — закрывает consumer
+//
+// При получении сообщения из очереди: POST на targetURL с телом сообщения.
+// 2xx → ack, остальное → nack (requeue=true).
+// При потере AMQP соединения — переподключение с экспоненциальной задержкой.
+
+package main
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/go-logr/logr"
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+// consumerEntry — запись об активном consumer-е.
+type consumerEntry struct {
+	queue     string
+	targetURL string
+	cancel    context.CancelFunc
+}
+
+// Dispatcher управляет пулом AMQP consumer-ов (один на каждый event Trigger).
+type Dispatcher struct {
+	rabbitURL  string
+	logger     logr.Logger
+	httpClient *http.Client
+
+	mu        sync.Mutex
+	consumers map[string]*consumerEntry // key = namespace/name триггера
+}
+
+// NewDispatcher создаёт Dispatcher.
+func NewDispatcher(rabbitURL string, logger logr.Logger) *Dispatcher {
+	return &Dispatcher{
+		rabbitURL:  rabbitURL,
+		logger:     logger.WithName("dispatcher"),
+		httpClient: &http.Client{Timeout: 30 * time.Second},
+		consumers:  make(map[string]*consumerEntry),
+	}
+}
+
+// Subscribe регистрирует consumer на очередь queue.
+// key — уникальный идентификатор триггера (namespace/name).
+// targetURL — внутренний HTTP URL функции (http://svc.ns.svc.cluster.local:8080/).
+// Если consumer для этого key уже есть — он переподписывается с новыми параметрами.
+func (d *Dispatcher) Subscribe(key, queue, targetURL string) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	// Если уже подписан — отменяем старый и создаём новый
+	if entry, ok := d.consumers[key]; ok {
+		if entry.queue == queue && entry.targetURL == targetURL {
+			return // ничего не изменилось
+		}
+		entry.cancel()
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	entry := &consumerEntry{queue: queue, targetURL: targetURL, cancel: cancel}
+	d.consumers[key] = entry
+
+	go d.runConsumer(ctx, key, queue, targetURL)
+	d.logger.Info("subscribed", "key", key, "queue", queue, "target", targetURL)
+}
+
+// Unsubscribe закрывает consumer для триггера key.
+func (d *Dispatcher) Unsubscribe(key string) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	if entry, ok := d.consumers[key]; ok {
+		entry.cancel()
+		delete(d.consumers, key)
+		d.logger.Info("unsubscribed", "key", key)
+	}
+}
+
+// runConsumer — горутина одного consumer-а.
+// Держит AMQP соединение и channel. При разрыве переподключается.
+// Завершается когда ctx отменён.
+func (d *Dispatcher) runConsumer(ctx context.Context, key, queue, targetURL string) {
+	backoff := time.Second
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+		err := d.consumeLoop(ctx, queue, targetURL)
+		if ctx.Err() != nil {
+			return // нормальное завершение
+		}
+		d.logger.Error(err, "consumer loop error, reconnecting", "key", key, "backoff", backoff)
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(backoff):
+		}
+		// Экспоненциальная задержка, максимум 30 секунд
+		if backoff < 30*time.Second {
+			backoff *= 2
+		}
+	}
+}
+
+// consumeLoop устанавливает соединение, объявляет очередь и читает сообщения.
+// Возвращает ошибку при потере соединения (вызывающий перезапустит).
+func (d *Dispatcher) consumeLoop(ctx context.Context, queue, targetURL string) error {
+	conn, err := amqp.Dial(d.rabbitURL)
+	if err != nil {
+		return fmt.Errorf("amqp dial: %w", err)
+	}
+	defer conn.Close()
+
+	ch, err := conn.Channel()
+	if err != nil {
+		return fmt.Errorf("amqp channel: %w", err)
+	}
+	defer ch.Close()
+
+	// Объявляем очередь как durable — она переживёт рестарт брокера
+	_, err = ch.QueueDeclare(queue, true, false, false, false, nil)
+	if err != nil {
+		return fmt.Errorf("queue declare %q: %w", queue, err)
+	}
+
+	// prefetch=1: не берём следующее сообщение пока не обработали текущее
+	if err = ch.Qos(1, 0, false); err != nil {
+		return fmt.Errorf("qos: %w", err)
+	}
+
+	msgs, err := ch.Consume(queue, "" /*auto-tag*/, false /*autoAck*/, false, false, false, nil)
+	if err != nil {
+		return fmt.Errorf("consume: %w", err)
+	}
+
+	connClose := conn.NotifyClose(make(chan *amqp.Error, 1))
+	d.logger.Info("consuming", "queue", queue, "target", targetURL)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case amqpErr := <-connClose:
+			return fmt.Errorf("connection closed: %v", amqpErr)
+		case msg, ok := <-msgs:
+			if !ok {
+				return fmt.Errorf("messages channel closed")
+			}
+			d.handleMessage(ctx, msg, targetURL)
+		}
+	}
+}
+
+// handleMessage выполняет HTTP POST на targetURL, ack/nack по результату.
+func (d *Dispatcher) handleMessage(ctx context.Context, msg amqp.Delivery, targetURL string) {
+	logger := d.logger.WithValues("target", targetURL, "contentType", msg.ContentType)
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, targetURL, bytes.NewReader(msg.Body))
+	if err != nil {
+		logger.Error(err, "failed to build request, nacking")
+		_ = msg.Nack(false, true)
+		return
+	}
+
+	contentType := msg.ContentType
+	if contentType == "" {
+		contentType = "application/octet-stream"
+	}
+	req.Header.Set("Content-Type", contentType)
+	req.ContentLength = int64(len(msg.Body))
+
+	resp, err := d.httpClient.Do(req)
+	if err != nil {
+		logger.Error(err, "http call failed, nacking")
+		_ = msg.Nack(false, true)
+		return
+	}
+	resp.Body.Close()
+
+	if resp.StatusCode >= 200 && resp.StatusCode < 300 {
+		_ = msg.Ack(false)
+		logger.V(1).Info("message delivered", "status", resp.StatusCode)
+	} else {
+		// Функция вернула ошибку — requeue для повтора
+		logger.Info("function returned non-2xx, nacking", "status", resp.StatusCode)
+		_ = msg.Nack(false, true)
+	}
+}