// Package kafkaconsumer subscribes to the edu.notification.requested topic
// (ISSUE-053 final 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 the G16 rule (edu.<domain>.<aggregate>.<action>); the
// previous abstract name edu.notification.events / edu.msg.notification.events
// is deprecated (see president-final-rulings §1.5).
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.requested", "invalid")
		// Commit to skip the poison message.
		_ = c.reader.CommitMessages(ctx, msg)
		return
	}

	// 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("notification.requested", "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("notification.requested", "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("notification.requested", "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})
}