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-code-parser.com/internal/config"
	contractevents "bit.admedia.com/scm/ad/adpilot-indexing-code-parser.com/internal/contracts/events"
	"bit.admedia.com/scm/ad/adpilot-indexing-code-parser.com/internal/contracts/validate"
	"bit.admedia.com/scm/ad/adpilot-indexing-code-parser.com/internal/service/parser"
)

const defaultConsumer = "consumer-1"

// FileChangedProcessor handles validated files.changed events.
type FileChangedProcessor interface {
	ProcessFileChanged(ctx context.Context, event contractevents.FileChangedEvent) error
}

// RepoSnapshotProcessor handles validated repo.snapshot.ready events.
type RepoSnapshotProcessor interface {
	ProcessRepoSnapshotReady(ctx context.Context, event contractevents.RepoSnapshotReadyEvent) error
}

// CommitsChangedProcessor handles validated commits.changed events.
type CommitsChangedProcessor interface {
	ProcessCommitsChanged(ctx context.Context, event contractevents.CommitsChangedEvent) error
}

// DocsParseCompletedProcessor handles validated docs.parse.completed events.
type DocsParseCompletedProcessor interface {
	ProcessDocsParseCompleted(ctx context.Context, event contractevents.DocsParseCompletedEvent) error
}

// Consumer reads indexing events from Redis Streams and delegates to the parser service.
type Consumer struct {
	redis              *Client
	redisCfg           config.RedisConfig
	streamCfg          config.StreamConsumerConfig
	fileHandler        FileChangedProcessor
	snapshotHandler    RepoSnapshotProcessor
	docsHandler        DocsParseCompletedProcessor
	commitsHandler     CommitsChangedProcessor
	logger             *slog.Logger
	consumerName       string
	consumerGroup      string

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

// NewConsumer creates a Redis Streams consumer for the parser service.
func NewConsumer(client *Client, cfg config.Config, parserService *parser.Service, logger *slog.Logger) *Consumer {
	return NewConsumerWithHandlers(client, cfg, parserService, parserService, parserService, parserService, logger)
}

// NewConsumerWithHandler creates a consumer with a custom files.changed handler (for tests).
func NewConsumerWithHandler(client *Client, cfg config.Config, handler FileChangedProcessor, logger *slog.Logger) *Consumer {
	return NewConsumerWithHandlers(client, cfg, handler, nil, nil, nil, logger)
}

// NewConsumerWithHandlers creates a consumer with custom stream handlers (for tests).
func NewConsumerWithHandlers(
	client *Client,
	cfg config.Config,
	fileHandler FileChangedProcessor,
	snapshotHandler RepoSnapshotProcessor,
	docsHandler DocsParseCompletedProcessor,
	commitsHandler CommitsChangedProcessor,
	logger *slog.Logger,
) *Consumer {
	if logger == nil {
		logger = slog.Default()
	}

	group := cfg.Stream.Group
	if strings.TrimSpace(group) == "" {
		group = "code-parser-service"
	}

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

	return &Consumer{
		redis:           client,
		redisCfg:        cfg.Redis,
		streamCfg:       cfg.Stream,
		fileHandler:     fileHandler,
		snapshotHandler: snapshotHandler,
		docsHandler:     docsHandler,
		commitsHandler:  commitsHandler,
		logger:          logger,
		consumerName:    consumerName,
		consumerGroup:   group,
		seen:            make(map[string]struct{}),
	}
}

var _ parser.EventConsumer = (*Consumer)(nil)

// 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.ConsumeFilesChanged(ctx); err != nil && !errors.Is(err, context.Canceled) {
			c.logger.Error("files.changed consumer stopped", "error", err)
		}
	}()

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

	if c.commitsHandler != nil {
		wg.Add(1)
		go func() {
			defer wg.Done()
			if err := c.ConsumeCommitsChanged(ctx); err != nil && !errors.Is(err, context.Canceled) {
				c.logger.Error("commits.changed consumer stopped", "error", err)
			}
		}()
	}

	if c.docsHandler != nil {
		wg.Add(1)
		go func() {
			defer wg.Done()
			if err := c.ConsumeDocsParseCompleted(ctx); err != nil && !errors.Is(err, context.Canceled) {
				c.logger.Error("docs.parse.completed consumer stopped", "error", err)
			}
		}()
	}

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

// ConsumeFilesChanged reads files.changed events and triggers parse jobs.
func (c *Consumer) ConsumeFilesChanged(ctx context.Context) error {
	stream := c.redisCfg.StreamName(contractevents.StreamFilesChanged)
	return c.consumeStream(ctx, stream, c.handleFileChanged)
}

// ConsumeRepoSnapshotReady reads repo.snapshot.ready events.
func (c *Consumer) ConsumeRepoSnapshotReady(ctx context.Context) error {
	stream := c.redisCfg.StreamName(contractevents.StreamRepoSnapshotReady)
	return c.consumeStream(ctx, stream, c.handleRepoSnapshotReady)
}

// ConsumeCommitsChanged reads commits.changed events and triggers graph delta computation.
func (c *Consumer) ConsumeCommitsChanged(ctx context.Context) error {
	stream := c.redisCfg.StreamName(contractevents.StreamCommitsChanged)
	return c.consumeStream(ctx, stream, c.handleCommitsChanged)
}

// ConsumeDocsParseCompleted reads docs.parse.completed events and retries snapshot finalization.
func (c *Consumer) ConsumeDocsParseCompleted(ctx context.Context) error {
	stream := c.redisCfg.StreamName(contractevents.StreamDocsParseCompleted)
	return c.consumeStream(ctx, stream, c.handleDocsParseCompleted)
}

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.streamCfg.BlockMs) * time.Millisecond
	if block <= 0 {
		block = 2 * time.Second
	}
	count := c.streamCfg.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.streamCfg.AutoclaimMinIdleMs) * time.Millisecond
	if minIdle <= 0 {
		minIdle = 60 * time.Second
	}
	claimCount := c.streamCfg.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 {
	return retryWhileRedisLoading(ctx, c.logger, "ensure_group:"+stream, func() 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 {
		if isRetryableStreamError(err) {
			c.logger.Debug("retryable stream processing failure",
				"stream", stream,
				"message_id", msg.ID,
				"error", err,
			)
			return
		}

		deliveryCount, countErr := c.deliveryCount(ctx, stream, msg.ID)
		if countErr != nil {
			deliveryCount = 1
		}
		maxRetries := c.streamCfg.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.recordCodeFileFailureAfterDLQ(ctx, stream, payload)
			c.recordCommitChangedFailureAfterDLQ(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 := c.dlqStreamName(originalStream)
	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) dlqStreamName(stream string) string {
	return stream + ".dlq"
}

func (c *Consumer) recordCommitChangedFailureAfterDLQ(ctx context.Context, stream, payload, message string) {
	if stream != c.redisCfg.StreamName(contractevents.StreamCommitsChanged) {
		return
	}

	recorder, ok := c.commitsHandler.(interface {
		RecordCommitChangedFailure(context.Context, contractevents.CommitsChangedEvent, string) error
	})
	if !ok {
		return
	}

	var event contractevents.CommitsChangedEvent
	if err := json.Unmarshal([]byte(payload), &event); err != nil {
		c.logger.Error("record commit changed failure after dlq: unmarshal failed", "error", err)
		return
	}
	if err := recorder.RecordCommitChangedFailure(ctx, event, message); err != nil {
		c.logger.Error("record commit changed failure after dlq failed",
			"repo_id", event.RepoID,
			"snapshot_id", event.SnapshotID,
			"commit_sha", event.CommitSHA,
			"error", err,
		)
	}
}

func (c *Consumer) recordCodeFileFailureAfterDLQ(ctx context.Context, stream, payload string) {
	if stream != c.redisCfg.StreamName(contractevents.StreamFilesChanged) {
		return
	}

	recorder, ok := c.fileHandler.(interface {
		RecordCodeFileFailure(context.Context, contractevents.FileChangedEvent) error
	})
	if !ok {
		return
	}

	var event contractevents.FileChangedEvent
	if err := json.Unmarshal([]byte(payload), &event); err != nil {
		c.logger.Error("record code file failure after dlq: unmarshal failed", "error", err)
		return
	}
	if event.FileKind != "" && event.FileKind != contractevents.FileKindCode {
		return
	}
	if err := recorder.RecordCodeFileFailure(ctx, event); err != nil {
		c.logger.Error("record code file failure after dlq failed",
			"repo_id", event.RepoID,
			"snapshot_id", event.SnapshotID,
			"file_path", event.FilePath,
			"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) handleFileChanged(ctx context.Context, payload string) (bool, error) {
	var event contractevents.FileChangedEvent
	if err := json.Unmarshal([]byte(payload), &event); err != nil {
		c.logger.Error("unmarshal files.changed", "error", err)
		return true, nil
	}

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

	if err := validate.FileChanged(event); err != nil {
		c.logger.Error("invalid files.changed event", "event_id", event.EventID, "error", err)
		return true, nil
	}

	if event.FileKind != "" && event.FileKind != contractevents.FileKindCode {
		c.markSeen(event.EventID)
		return true, nil
	}

	c.logger.Info("processing files.changed",
		"event_id", event.EventID,
		"repo_id", event.RepoID,
		"snapshot_id", event.SnapshotID,
		"file_path", event.FilePath,
		"language", event.Language,
	)

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

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

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

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

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

	c.logger.Info("snapshot ready",
		"event_id", event.EventID,
		"repo_id", event.RepoID,
		"snapshot_id", event.SnapshotID,
		"file_count", event.FileCount,
	)

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

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

func (c *Consumer) handleDocsParseCompleted(ctx context.Context, payload string) (bool, error) {
	if c.docsHandler == nil {
		return true, nil
	}

	var event contractevents.DocsParseCompletedEvent
	if err := json.Unmarshal([]byte(payload), &event); err != nil {
		c.logger.Error("unmarshal docs.parse.completed", "error", err)
		return true, nil
	}

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

	if err := validate.DocsParseCompleted(event); err != nil {
		c.logger.Error("invalid docs.parse.completed event", "event_id", event.EventID, "error", err)
		return true, nil
	}

	c.logger.Info("docs parse completed",
		"event_id", event.EventID,
		"repo_id", event.RepoID,
		"snapshot_id", event.SnapshotID,
		"processed_docs_files", event.ProcessedDocsFiles,
		"expected_docs_files", event.ExpectedDocsFiles,
	)

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

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

func (c *Consumer) handleCommitsChanged(ctx context.Context, payload string) (bool, error) {
	if c.commitsHandler == nil {
		return true, nil
	}

	var event contractevents.CommitsChangedEvent
	if err := json.Unmarshal([]byte(payload), &event); err != nil {
		c.logger.Error("unmarshal commits.changed", "error", err)
		return true, nil
	}

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

	if err := validate.CommitsChanged(event); err != nil {
		c.logger.Error("invalid commits.changed event", "event_id", event.EventID, "error", err)
		return true, nil
	}

	c.logger.Info("processing commits.changed",
		"event_id", event.EventID,
		"repo_id", event.RepoID,
		"snapshot_id", event.SnapshotID,
		"commit_sha", event.CommitSHA,
		"parent_sha", event.ParentSHA,
	)

	if err := c.commitsHandler.ProcessCommitsChanged(ctx, event); err != nil {
		if errors.Is(err, parser.ErrCommitOrderNotReady) || errors.Is(err, parser.ErrCommitArtifactsNotReady) {
			return false, err
		}
		return false, err
	}

	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)
}

func isRetryableStreamError(err error) bool {
	return errors.Is(err, parser.ErrCommitOrderNotReady) ||
		errors.Is(err, parser.ErrCommitArtifactsNotReady) ||
		errors.Is(err, parser.ErrIndexingRunNotReady)
}