package redis

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"strings"
	"sync"
	"time"

	goredis "github.com/redis/go-redis/v9"

	"bit.admedia.com/scm/ad/adpilot-indexing-commit-intel.com/internal/config"
	contractevents "bit.admedia.com/scm/ad/adpilot-indexing-commit-intel.com/internal/contracts/events"
	"bit.admedia.com/scm/ad/adpilot-indexing-commit-intel.com/internal/contracts/validate"
)

const defaultConsumerName = "consumer-1"

// GraphDeltaProcessor handles validated graph.delta.ready events.
type GraphDeltaProcessor interface {
	ProcessGraphDeltaReady(ctx context.Context, event contractevents.GraphDeltaReadyEvent) error
	RecordGraphDeltaFailure(ctx context.Context, event contractevents.GraphDeltaReadyEvent, message string) error
}

// Consumer reads commit intelligence input streams from Redis.
type Consumer struct {
	redis         *Client
	cfg           config.Config
	deltaHandler  GraphDeltaProcessor
	logger        *slog.Logger
	consumerName  string
	consumerGroup string
	workers       int

	seenMu sync.Mutex
	seen   map[string]struct{}

	workerSem chan struct{}
}

// NewConsumer creates a Redis Streams consumer for the commit intelligence service.
func NewConsumer(
	client *Client,
	cfg config.Config,
	deltaHandler GraphDeltaProcessor,
	logger *slog.Logger,
) *Consumer {
	if logger == nil {
		logger = slog.Default()
	}

	workers := cfg.Analysis.Workers
	if workers < 1 {
		workers = 1
	}

	group := cfg.Stream.Group
	if strings.TrimSpace(group) == "" {
		group = cfg.Redis.ConsumerGroup
	}
	if strings.TrimSpace(group) == "" {
		group = "commit-intelligence-service"
	}

	consumerName := cfg.Stream.ConsumerName
	if strings.TrimSpace(consumerName) == "" {
		name, err := os.Hostname()
		if err != nil || strings.TrimSpace(name) == "" {
			consumerName = defaultConsumerName
		} else {
			consumerName = name
		}
	}

	return &Consumer{
		redis:         client,
		cfg:           cfg,
		deltaHandler:  deltaHandler,
		logger:        logger,
		consumerName:  consumerName,
		consumerGroup: group,
		workers:       workers,
		seen:          make(map[string]struct{}),
		workerSem:     make(chan struct{}, workers),
	}
}

// Run starts stream consumer loops and blocks until ctx is cancelled.
func (c *Consumer) Run(ctx context.Context) error {
	var wg sync.WaitGroup
	wg.Add(2)

	go func() {
		defer wg.Done()
		if err := c.ConsumeGraphDeltaReady(ctx); err != nil && !errors.Is(err, context.Canceled) {
			c.logger.Error("graph.delta.ready consumer stopped", "error", err)
		}
	}()

	go func() {
		defer wg.Done()
		if err := c.ConsumeGraphArtifactReady(ctx); err != nil && !errors.Is(err, context.Canceled) {
			c.logger.Error("graph.artifact.ready consumer stopped", "error", err)
		}
	}()

	wg.Wait()
	return ctx.Err()
}

// ConsumeGraphDeltaReady reads graph.delta.ready events and runs commit analysis.
func (c *Consumer) ConsumeGraphDeltaReady(ctx context.Context) error {
	stream := c.cfg.Redis.StreamName(contractevents.StreamGraphDeltaReady)
	return c.consumeStream(ctx, stream, c.handleGraphDeltaReady)
}

// ConsumeGraphArtifactReady reads graph.artifact.ready events and ACKs without analysis.
func (c *Consumer) ConsumeGraphArtifactReady(ctx context.Context) error {
	stream := c.cfg.Redis.StreamName(contractevents.StreamGraphArtifactReady)
	return c.consumeStream(ctx, stream, c.handleGraphArtifactReady)
}

type messageHandler func(ctx context.Context, payload string) (ack bool, err error)

func (c *Consumer) consumeStream(ctx context.Context, stream string, handle messageHandler) error {
	if err := c.ensureGroup(ctx, stream); err != nil {
		return err
	}

	rdb := c.redis.Underlying()
	block := time.Duration(c.cfg.Stream.BlockMs) * time.Millisecond
	if block <= 0 {
		block = 2 * time.Second
	}
	count := c.cfg.Stream.ReadCount
	if count <= 0 {
		count = 10
	}

	for {
		if err := ctx.Err(); err != nil {
			return err
		}

		c.autoclaimPending(ctx, stream, handle)

		streams, err := rdb.XReadGroup(ctx, &goredis.XReadGroupArgs{
			Group:    c.consumerGroup,
			Consumer: c.consumerName,
			Streams:  []string{stream, ">"},
			Count:    count,
			Block:    block,
		}).Result()
		if err != nil {
			if errors.Is(err, goredis.Nil) || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
				if err := ctx.Err(); err != nil {
					return err
				}
				continue
			}
			c.logger.Error("xreadgroup failed", "stream", stream, "error", err)
			continue
		}

		for _, s := range streams {
			for _, msg := range s.Messages {
				c.processMessage(ctx, stream, msg, handle)
			}
		}
	}
}

func (c *Consumer) autoclaimPending(ctx context.Context, stream string, handle messageHandler) {
	rdb := c.redis.Underlying()
	minIdle := time.Duration(c.cfg.Stream.AutoclaimMinIdleMs) * time.Millisecond
	if minIdle <= 0 {
		minIdle = 60 * time.Second
	}
	claimCount := c.cfg.Stream.AutoclaimCount
	if claimCount <= 0 {
		claimCount = 10
	}

	result, _, err := rdb.XAutoClaim(ctx, &goredis.XAutoClaimArgs{
		Stream:   stream,
		Group:    c.consumerGroup,
		Consumer: c.consumerName,
		MinIdle:  minIdle,
		Start:    "0-0",
		Count:    claimCount,
	}).Result()
	if err != nil && !errors.Is(err, goredis.Nil) {
		c.logger.Error("xautoclaim failed", "stream", stream, "error", err)
		return
	}
	for _, msg := range result {
		c.processMessage(ctx, stream, msg, handle)
	}
}

func (c *Consumer) ensureGroup(ctx context.Context, stream string) error {
	err := c.redis.Underlying().XGroupCreateMkStream(ctx, stream, c.consumerGroup, "$").Err()
	if err == nil {
		return nil
	}
	if strings.Contains(err.Error(), "BUSYGROUP") {
		return nil
	}
	return fmt.Errorf("create consumer group on %s: %w", stream, err)
}

func (c *Consumer) processMessage(ctx context.Context, stream string, msg goredis.XMessage, handle messageHandler) {
	payload, ok := msg.Values["payload"].(string)
	if !ok || strings.TrimSpace(payload) == "" {
		c.logger.Error("missing payload field", "stream", stream, "message_id", msg.ID)
		_ = c.publishDLQ(ctx, stream, msg.ID, jsonMarshalFields(msg.Values), "missing payload field", 1, "")
		c.ackMessage(ctx, stream, msg.ID)
		return
	}

	ack, err := handle(ctx, payload)
	if err != nil {
		deliveryCount, countErr := c.deliveryCount(ctx, stream, msg.ID)
		if countErr != nil {
			deliveryCount = 1
		}
		maxRetries := c.cfg.Stream.MaxRetries
		if maxRetries <= 0 {
			maxRetries = 5
		}

		if deliveryCount >= maxRetries {
			eventID := extractEventID(payload)
			_ = c.publishDLQ(ctx, stream, msg.ID, payload, err.Error(), deliveryCount, eventID)
			c.logger.Error("moved stream message to dlq",
				"stream", stream,
				"message_id", msg.ID,
				"event_id", eventID,
				"attempt_count", deliveryCount,
				"error", err,
			)
			c.recordGraphDeltaFailureAfterDLQ(ctx, stream, payload, err.Error())
			c.ackMessage(ctx, stream, msg.ID)
			return
		}

		c.logger.Error("transient stream processing failure",
			"stream", stream,
			"message_id", msg.ID,
			"attempt_count", deliveryCount,
			"error", err,
		)
		return
	}

	if ack {
		c.ackMessage(ctx, stream, msg.ID)
	}
}

func (c *Consumer) deliveryCount(ctx context.Context, stream, messageID string) (int, error) {
	pending, err := c.redis.Underlying().XPendingExt(ctx, &goredis.XPendingExtArgs{
		Stream: stream,
		Group:  c.consumerGroup,
		Start:  messageID,
		End:    messageID,
		Count:  1,
	}).Result()
	if err != nil {
		return 0, err
	}
	if len(pending) == 0 {
		return 1, nil
	}
	return int(pending[0].RetryCount) + 1, nil
}

func (c *Consumer) publishDLQ(ctx context.Context, originalStream, messageID, payload, errMsg string, attemptCount int, eventID string) error {
	dlqStream := originalStream + ".dlq"
	fields := map[string]interface{}{
		"original_stream":     originalStream,
		"original_message_id": messageID,
		"event_id":            eventID,
		"payload":             payload,
		"error":               errMsg,
		"attempt_count":       attemptCount,
		"failed_at":           time.Now().UTC().Format(time.RFC3339),
	}
	return c.redis.Underlying().XAdd(ctx, &goredis.XAddArgs{
		Stream: dlqStream,
		Values: fields,
	}).Err()
}

func (c *Consumer) recordGraphDeltaFailureAfterDLQ(ctx context.Context, stream, payload, message string) {
	if c.deltaHandler == nil || stream != c.cfg.Redis.StreamName(contractevents.StreamGraphDeltaReady) {
		return
	}

	var event contractevents.GraphDeltaReadyEvent
	if err := json.Unmarshal([]byte(payload), &event); err != nil {
		c.logger.Error("record graph delta failure after dlq: unmarshal failed", "error", err)
		return
	}
	if err := c.deltaHandler.RecordGraphDeltaFailure(ctx, event, message); err != nil {
		c.logger.Error("record graph delta failure after dlq failed",
			"repo_id", event.RepoID,
			"target_commit_sha", event.TargetCommitSHA,
			"error", err,
		)
	}
}

func (c *Consumer) ackMessage(ctx context.Context, stream, messageID string) {
	if err := c.redis.Underlying().XAck(ctx, stream, c.consumerGroup, messageID).Err(); err != nil {
		c.logger.Error("xack failed", "stream", stream, "message_id", messageID, "error", err)
	}
}

func (c *Consumer) handleGraphDeltaReady(ctx context.Context, payload string) (bool, error) {
	var event contractevents.GraphDeltaReadyEvent
	if err := json.Unmarshal([]byte(payload), &event); err != nil {
		c.logger.Error("unmarshal graph.delta.ready", "error", err)
		return true, nil
	}

	if c.alreadySeen(event.EventID) {
		return true, nil
	}

	if err := validate.GraphDeltaReady(event); err != nil {
		c.logger.Error("invalid graph.delta.ready event", "event_id", event.EventID, "error", err)
		return true, nil
	}

	c.logger.Info("processing graph.delta.ready",
		"event_id", event.EventID,
		"repo_id", event.RepoID,
		"base_commit_sha", event.BaseCommitSHA,
		"target_commit_sha", event.TargetCommitSHA,
	)

	if c.deltaHandler == nil {
		c.logger.Error("graph.delta.ready handler not configured", "event_id", event.EventID)
		return true, nil
	}

	c.workerSem <- struct{}{}
	defer func() { <-c.workerSem }()

	if err := c.deltaHandler.ProcessGraphDeltaReady(ctx, event); err != nil {
		return false, err
	}

	c.markSeen(event.EventID)
	return true, nil
}

func (c *Consumer) handleGraphArtifactReady(ctx context.Context, payload string) (bool, error) {
	var event contractevents.GraphArtifactReadyEvent
	if err := json.Unmarshal([]byte(payload), &event); err != nil {
		c.logger.Error("unmarshal graph.artifact.ready", "error", err)
		return true, nil
	}

	if c.alreadySeen(event.EventID) {
		return true, nil
	}

	if err := validate.GraphArtifactReady(event); err != nil {
		c.logger.Error("invalid graph.artifact.ready event", "event_id", event.EventID, "error", err)
		return true, nil
	}

	c.logger.Info("graph.artifact.ready ack",
		"event_id", event.EventID,
		"repo_id", event.RepoID,
		"artifact_id", event.ArtifactID,
	)

	c.markSeen(event.EventID)
	return true, nil
}

func (c *Consumer) alreadySeen(eventID string) bool {
	if strings.TrimSpace(eventID) == "" {
		return false
	}

	c.seenMu.Lock()
	defer c.seenMu.Unlock()
	_, ok := c.seen[eventID]
	return ok
}

func (c *Consumer) markSeen(eventID string) {
	if strings.TrimSpace(eventID) == "" {
		return
	}

	c.seenMu.Lock()
	defer c.seenMu.Unlock()
	c.seen[eventID] = struct{}{}
}

func extractEventID(payload string) string {
	var envelope struct {
		EventID string `json:"event_id"`
	}
	_ = json.Unmarshal([]byte(payload), &envelope)
	return envelope.EventID
}

func jsonMarshalFields(values map[string]interface{}) string {
	data, err := json.Marshal(values)
	if err != nil {
		return "{}"
	}
	return string(data)
}