package worker

import (
	"context"
	"errors"
	"time"

	"github.com/prabhupadasays/spsays-bff/internal/repository"
	"github.com/prabhupadasays/spsays-bff/pkg/shorturl"
	"go.uber.org/zap"
)

// ShortURLWorker drains the Redis create-queue + click-queue into Postgres.
type ShortURLWorker struct {
	repo   *repository.ShortURLRepo
	queue  *shorturl.Queue
	logger *zap.Logger
}

func NewShortURLWorker(repo *repository.ShortURLRepo, queue *shorturl.Queue, logger *zap.Logger) *ShortURLWorker {
	return &ShortURLWorker{repo: repo, queue: queue, logger: logger}
}

// Run starts the create-worker and click-worker goroutines. Both exit cleanly
// when ctx is cancelled. Call from server startup.
func (w *ShortURLWorker) Run(ctx context.Context) {
	go w.runCreateLoop(ctx)
	go w.runClickLoop(ctx)
}

func (w *ShortURLWorker) runCreateLoop(ctx context.Context) {
	w.logger.Info("shorturl create-worker started")
	defer w.logger.Info("shorturl create-worker stopped")

	for {
		if ctx.Err() != nil {
			return
		}
		code, err := w.queue.DequeueCreate(ctx, 5*time.Second)
		if err != nil {
			if errors.Is(err, context.Canceled) {
				return
			}
			w.logger.Warn("create-worker dequeue", zap.Error(err))
			time.Sleep(time.Second)
			continue
		}
		if code == "" {
			continue
		}

		payload, ok := w.queue.LookupPending(ctx, code)
		if !ok {
			// Placeholder without a real payload — likely the producer crashed
			// between MarkPending and EnqueueCreate. Skip.
			w.logger.Warn("create-worker: missing pending payload", zap.String("code", code))
			continue
		}

		parsed := shorturl.ParsePath(payload.URL)
		if _, err := w.repo.Insert(ctx, code, payload, parsed); err != nil {
			if repository.IsUniqueViolation(err) {
				// Already persisted by a previous attempt; clean up the queue entry.
				_ = w.queue.UnmarkPending(ctx, code)
				continue
			}
			w.logger.Error("create-worker insert", zap.String("code", code), zap.Error(err))
			_ = w.queue.PushDeadLetter(ctx, code, err.Error())
			continue
		}

		if err := w.queue.MarkCached(ctx, code, payload.URL); err != nil {
			w.logger.Warn("create-worker mark cached", zap.String("code", code), zap.Error(err))
		}
		if err := w.queue.UnmarkPending(ctx, code); err != nil {
			w.logger.Warn("create-worker unmark pending", zap.String("code", code), zap.Error(err))
		}
	}
}

func (w *ShortURLWorker) runClickLoop(ctx context.Context) {
	w.logger.Info("shorturl click-worker started")
	defer w.logger.Info("shorturl click-worker stopped")

	for {
		if ctx.Err() != nil {
			return
		}
		ev, ok, err := w.queue.DequeueClick(ctx, 5*time.Second)
		if err != nil {
			if errors.Is(err, context.Canceled) {
				return
			}
			w.logger.Warn("click-worker dequeue", zap.Error(err))
			time.Sleep(time.Second)
			continue
		}
		if !ok {
			continue
		}

		if err := w.repo.RecordClick(ctx, ev); err != nil {
			w.logger.Warn("click-worker record", zap.String("code", ev.Code), zap.Error(err))
		}
	}
}