254 lines
8.2 KiB
Go
254 lines
8.2 KiB
Go
// Package kafkaconsumer subscribes to the edu.notify.notification.sent topic
|
|
// (ARB-013 canonical topic name) on behalf of push-gateway. Each consumed
|
|
// NotificationRequested event is dispatched to the Hub for delivery to online
|
|
// clients. At-least-once delivery is enforced via manual commit; idempotency
|
|
// is provided by Redis SETNX on event_id (TTL 24h).
|
|
//
|
|
// Topic naming follows ARB-013 (edu.<domain>.<aggregate>.<action>); the
|
|
// previous name edu.notification.requested is deprecated (v2 alignment).
|
|
package kafkaconsumer
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"time"
|
|
|
|
"github.com/edu-cloud/push-gateway/internal/hub"
|
|
"github.com/edu-cloud/push-gateway/internal/observability"
|
|
"github.com/edu-cloud/push-gateway/internal/redisclient"
|
|
"github.com/segmentio/kafka-go"
|
|
)
|
|
|
|
// MaxRetries is the per-message retry count before sending to the dead-letter
|
|
// topic. After MaxRetries failures the message is committed (skipped) and
|
|
// logged; a future P6 enhancement can republish to a DLQ topic.
|
|
const MaxRetries = 3
|
|
|
|
// NotificationRequested is the message shape produced by msg (ai10). Field
|
|
// names mirror the JSON wire format produced by msg's Outbox publisher.
|
|
type NotificationRequested struct {
|
|
EventID string `json:"event_id"`
|
|
UserID string `json:"user_id"`
|
|
EventType string `json:"event_type"`
|
|
Channel string `json:"channel"`
|
|
Title string `json:"title"`
|
|
Content string `json:"content"`
|
|
Data json.RawMessage `json:"data"`
|
|
Broadcast bool `json:"broadcast"`
|
|
OccurredAt int64 `json:"occurred_at"`
|
|
}
|
|
|
|
// Consumer wraps a kafka.Reader for the notification topic.
|
|
type Consumer struct {
|
|
reader *kafka.Reader
|
|
hub *hub.Hub
|
|
redis *redisclient.Client
|
|
metrics *observability.Metrics
|
|
topic string
|
|
groupID string
|
|
dlqTopic string
|
|
}
|
|
|
|
// Config configures the Kafka consumer.
|
|
type Config struct {
|
|
Brokers []string
|
|
Topic string
|
|
GroupID string
|
|
DLQTopic string
|
|
}
|
|
|
|
// New creates a Consumer for the configured topic and group.
|
|
func New(cfg Config, h *hub.Hub, r *redisclient.Client, m *observability.Metrics) *Consumer {
|
|
if cfg.DLQTopic == "" {
|
|
cfg.DLQTopic = cfg.Topic + ".dlq"
|
|
}
|
|
reader := kafka.NewReader(kafka.ReaderConfig{
|
|
Brokers: cfg.Brokers,
|
|
Topic: cfg.Topic,
|
|
GroupID: cfg.GroupID,
|
|
MinBytes: 1,
|
|
MaxBytes: 10 * 1024 * 1024,
|
|
CommitInterval: 1 * time.Second, // periodic background commit
|
|
StartOffset: kafka.LastOffset, // skip historical backlog on first start
|
|
})
|
|
return &Consumer{
|
|
reader: reader,
|
|
hub: h,
|
|
redis: r,
|
|
metrics: m,
|
|
topic: cfg.Topic,
|
|
groupID: cfg.GroupID,
|
|
dlqTopic: cfg.DLQTopic,
|
|
}
|
|
}
|
|
|
|
// Run blocks until ctx is canceled, consuming messages and dispatching them.
|
|
// Errors are logged but do not stop the consumer unless ctx is canceled.
|
|
func (c *Consumer) Run(ctx context.Context) error {
|
|
observability.Logger().Info("kafka consumer starting",
|
|
"topic", c.topic, "group", c.groupID, "brokers", c.reader.Config().Brokers)
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
default:
|
|
}
|
|
msg, err := c.reader.ReadMessage(ctx)
|
|
if err != nil {
|
|
if errors.Is(err, io.EOF) || errors.Is(err, context.Canceled) {
|
|
return nil
|
|
}
|
|
observability.Logger().Warn("kafka read failed", "err", err)
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil
|
|
case <-time.After(2 * time.Second):
|
|
}
|
|
continue
|
|
}
|
|
c.metrics.IncKafkaConsumed(c.topic, fmt.Sprintf("%d", msg.Partition))
|
|
c.processMessage(ctx, msg)
|
|
}
|
|
}
|
|
|
|
// Close releases the underlying reader. Should be called after Run returns.
|
|
func (c *Consumer) Close() error {
|
|
return c.reader.Close()
|
|
}
|
|
|
|
// HealthCheck verifies that the Kafka broker is reachable. Used by /readyz as
|
|
// a soft-failure probe: a failing check marks the consumer as degraded but
|
|
// does not return 503 (per ARB-015 §17.4 / ISSUE-058).
|
|
//
|
|
// The check issues a metadata request for the consumer topic; on success the
|
|
// broker is considered reachable. This is cheaper than a Lag() call which
|
|
// requires partition assignment to have completed.
|
|
func (c *Consumer) HealthCheck(ctx context.Context) error {
|
|
conn, err := kafka.DialContext(ctx, "tcp", c.reader.Config().Brokers[0])
|
|
if err != nil {
|
|
return fmt.Errorf("kafka: dial broker: %w", err)
|
|
}
|
|
defer conn.Close()
|
|
partitions, err := conn.ReadPartitions(c.topic)
|
|
if err != nil {
|
|
return fmt.Errorf("kafka: read partitions: %w", err)
|
|
}
|
|
if len(partitions) == 0 {
|
|
return fmt.Errorf("kafka: topic %q has no partitions", c.topic)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Topic returns the configured topic name. Exposed for diagnostics.
|
|
func (c *Consumer) Topic() string {
|
|
return c.topic
|
|
}
|
|
|
|
// GroupID returns the configured consumer group name.
|
|
func (c *Consumer) GroupID() string {
|
|
return c.groupID
|
|
}
|
|
|
|
// processMessage handles a single Kafka message with retry and idempotency.
|
|
func (c *Consumer) processMessage(ctx context.Context, msg kafka.Message) {
|
|
var event NotificationRequested
|
|
if err := json.Unmarshal(msg.Value, &event); err != nil {
|
|
observability.Logger().Warn("kafka: invalid message payload",
|
|
"topic", c.topic, "partition", msg.Partition, "offset", msg.Offset, "err", err)
|
|
c.metrics.IncPushed("notification.invalid", "invalid")
|
|
// Commit to skip the poison message.
|
|
_ = c.reader.CommitMessages(ctx, msg)
|
|
return
|
|
}
|
|
|
|
// eventLabel is the metrics label for IncPushed. Prefer event_type (the
|
|
// actual event name like "ExamExtended"); fall back to a generic label.
|
|
eventLabel := event.EventType
|
|
if eventLabel == "" {
|
|
eventLabel = "notification.sent"
|
|
}
|
|
|
|
// Idempotency: skip already-processed events by event_id.
|
|
fresh, err := c.redis.DedupEventId(ctx, event.EventID)
|
|
if err != nil {
|
|
observability.Logger().Warn("kafka: dedup check failed, processing anyway",
|
|
"event_id", event.EventID, "err", err)
|
|
}
|
|
if !fresh {
|
|
observability.Logger().Debug("kafka: duplicate event skipped",
|
|
"event_id", event.EventID)
|
|
c.metrics.IncPushed(eventLabel, "duplicate")
|
|
_ = c.reader.CommitMessages(ctx, msg)
|
|
return
|
|
}
|
|
|
|
// Retry up to MaxRetries on dispatch failure.
|
|
for attempt := 1; attempt <= MaxRetries; attempt++ {
|
|
if err := c.dispatch(ctx, event); err == nil {
|
|
c.metrics.IncPushed(eventLabel, "delivered")
|
|
_ = c.reader.CommitMessages(ctx, msg)
|
|
return
|
|
} else if attempt < MaxRetries {
|
|
observability.Logger().Warn("kafka: dispatch retry",
|
|
"event_id", event.EventID, "attempt", attempt, "err", err)
|
|
time.Sleep(time.Duration(attempt) * time.Second)
|
|
}
|
|
}
|
|
// MaxRetries exhausted: send to DLQ (best-effort) and commit.
|
|
observability.Logger().Error("kafka: dispatch failed after retries, sending to DLQ",
|
|
"event_id", event.EventID, "topic", c.topic)
|
|
c.sendToDLQ(ctx, msg.Value)
|
|
c.metrics.IncPushed(eventLabel, "dlq")
|
|
_ = c.reader.CommitMessages(ctx, msg)
|
|
}
|
|
|
|
// dispatch delivers the event to local Hub connections and, when the user is
|
|
// not local, asks Redis to fan out via Pub/Sub.
|
|
func (c *Consumer) dispatch(ctx context.Context, event NotificationRequested) error {
|
|
payload, err := json.Marshal(map[string]any{
|
|
"type": "message",
|
|
"event": event.EventType,
|
|
"data": event.Data,
|
|
"timestamp": time.Now().UTC().Format(time.RFC3339),
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("marshal payload: %w", err)
|
|
}
|
|
|
|
if event.Broadcast {
|
|
// Broadcast: deliver locally + publish to other instances.
|
|
c.hub.Broadcast(payload)
|
|
return c.redis.PublishBroadcast(ctx, redisclient.CrossInstanceMessage{
|
|
Event: event.EventType,
|
|
Data: event.Data,
|
|
})
|
|
}
|
|
|
|
// Directed: try local first, fall back to Redis Pub/Sub.
|
|
if c.hub.HasUser(event.UserID) {
|
|
if c.hub.SendToUser(event.UserID, payload) > 0 {
|
|
return nil
|
|
}
|
|
}
|
|
return c.redis.PublishUser(ctx, event.UserID, redisclient.CrossInstanceMessage{
|
|
Event: event.EventType,
|
|
Data: event.Data,
|
|
})
|
|
}
|
|
|
|
// sendToDLQ publishes a raw message to the dead-letter topic. Best-effort;
|
|
// failure is logged but does not block commit.
|
|
func (c *Consumer) sendToDLQ(ctx context.Context, value []byte) {
|
|
conn, err := kafka.DialLeader(ctx, "tcp", c.reader.Config().Brokers[0], c.dlqTopic, 0)
|
|
if err != nil {
|
|
observability.Logger().Error("kafka: DLQ dial failed", "topic", c.dlqTopic, "err", err)
|
|
return
|
|
}
|
|
defer conn.Close()
|
|
_ = conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
|
|
_, _ = conn.WriteMessages(kafka.Message{Value: value})
|
|
}
|