// Package worker monta o processo que roda o bot do Telegram e os jobs.
//
// Só uma instância pode estar ativa: o Telegram em long polling não aceita
// dois consumidores e os jobs não devem rodar em duplicidade. A exclusividade
// vem de um advisory lock do Postgres (ARQUITETURA.md, 4.3).
package worker

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"runtime/debug"
	"sync"
	"sync/atomic"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"

	"botpontos/internal/app"
	"botpontos/internal/bootstrap"
	"botpontos/internal/bot"
	"botpontos/internal/config"
	"botpontos/internal/httpx"
	"botpontos/internal/rules"
)

// advisoryLockKey identifica o lock de liderança do worker no Postgres.
const advisoryLockKey int64 = 7_202_610_001

// lockCheckInterval define de quanto em quanto tempo a conexão que segura o lock é verificada.
const lockCheckInterval = 30 * time.Second

type worker struct {
	cfg    config.Config
	pool   *pgxpool.Pool
	built  *bootstrap.Built
	svc    *app.Service
	log    *slog.Logger
	leader atomic.Bool
}

func Run(ctx context.Context, cfg config.Config, logger *slog.Logger) error {
	if err := cfg.RequireTelegram(); err != nil {
		return err
	}
	built, err := bootstrap.Build(ctx, cfg, logger)
	if err != nil {
		return err
	}
	defer built.Pool.Close()

	w := &worker{cfg: cfg, pool: built.Pool, built: built, svc: built.Service, log: logger}
	if err := w.svc.SeedSettings(ctx); err != nil {
		return fmt.Errorf("gravar configurações padrão: %w", err)
	}

	health := &http.Server{
		Addr:              cfg.Worker.HealthAddr,
		Handler:           httpx.RequestLog(logger, w.healthHandler()),
		ReadHeaderTimeout: 5 * time.Second,
	}
	go func() {
		if err := health.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
			logger.Error("servidor de saúde do worker parou", "error", err)
		}
	}()
	defer func() {
		shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = health.Shutdown(shutdownCtx)
	}()

	w.lead(ctx)
	logger.Info("worker encerrado")
	return nil
}

// lead tenta assumir a liderança até o contexto ser cancelado.
func (w *worker) lead(ctx context.Context) {
	for {
		if err := w.leadOnce(ctx); err != nil && ctx.Err() == nil {
			w.log.Warn("liderança indisponível", "error", err)
		}
		select {
		case <-ctx.Done():
			return
		case <-time.After(w.cfg.Worker.LeaderRetry):
		}
	}
}

func (w *worker) leadOnce(ctx context.Context) error {
	conn, err := w.pool.Acquire(ctx)
	if err != nil {
		return fmt.Errorf("obter conexão para o lock: %w", err)
	}
	defer conn.Release()

	var acquired bool
	if err := conn.QueryRow(ctx, "SELECT pg_try_advisory_lock($1)", advisoryLockKey).Scan(&acquired); err != nil {
		return fmt.Errorf("tentar advisory lock: %w", err)
	}
	if !acquired {
		w.log.Info("outra instância do worker está ativa; aguardando")
		return nil
	}

	w.leader.Store(true)
	defer w.leader.Store(false)
	defer func() {
		unlockCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_, _ = conn.Exec(unlockCtx, "SELECT pg_advisory_unlock($1)", advisoryLockKey)
	}()
	w.log.Info("liderança obtida; worker ativo")

	activeCtx, stop := context.WithCancel(ctx)
	defer stop()

	var lockErr error
	done := make(chan struct{})
	go func() {
		defer close(done)
		w.active(activeCtx)
	}()

	ticker := time.NewTicker(lockCheckInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			stop()
			<-done
			return nil
		case <-ticker.C:
			// O lock vive na sessão: se esta conexão cair, a liderança se perde.
			if err := conn.Ping(ctx); err != nil {
				lockErr = fmt.Errorf("conexão que segura o lock caiu: %w", err)
				stop()
				<-done
				return lockErr
			}
		}
	}
}

// active roda o bot e os ciclos enquanto a instância for a líder.
func (w *worker) active(ctx context.Context) {
	w.svc.CheckScheduleIntervals(ctx)

	jobs := w.cfg.Jobs
	runner := bot.NewRunner(w.built.Telegram, bot.NewRouter(w.svc, w.log), w.log, w.cfg.Telegram.PollTimeout)

	var wg sync.WaitGroup
	start := func(fn func()) {
		wg.Add(1)
		go func() {
			defer wg.Done()
			fn()
		}()
	}
	start(func() { runner.Run(ctx) })
	start(func() {
		w.every(ctx, "reconciliação", jobs.ReconciliationInterval, func(ctx context.Context, tick time.Time) error {
			return w.svc.Reconcile(ctx, tick)
		})
	})
	start(func() {
		w.every(ctx, "despacho", jobs.DispatchInterval, func(ctx context.Context, _ time.Time) error {
			return w.svc.Dispatch(ctx)
		})
	})
	start(func() {
		w.every(ctx, "manutenção", jobs.MaintenanceInterval, func(ctx context.Context, _ time.Time) error {
			return w.svc.Maintenance(ctx)
		})
	})
	wg.Wait()
}

// every executa fn no tick atual e depois em cada tick alinhado ao relógio
// (ex.: :00, :15, :30, :45). Cada ciclo roda na sua própria goroutine; se uma
// execução passar do próximo tick, ele é pulado em vez de acumular (8.2).
func (w *worker) every(ctx context.Context, name string, interval time.Duration, fn func(context.Context, time.Time) error) {
	w.runSafely(ctx, name, rules.Tick(time.Now(), interval), fn)
	for {
		next := rules.NextTick(time.Now(), interval)
		select {
		case <-ctx.Done():
			return
		case <-time.After(time.Until(next)):
		}
		w.runSafely(ctx, name, next, fn)
	}
}

func (w *worker) runSafely(ctx context.Context, name string, tick time.Time, fn func(context.Context, time.Time) error) {
	defer func() {
		if v := recover(); v != nil {
			w.log.Error("panic no ciclo", "job", name, "panic", v, "stack", string(debug.Stack()))
		}
	}()
	started := time.Now()
	if err := fn(ctx, tick); err != nil && ctx.Err() == nil {
		w.log.Error("ciclo falhou", "job", name, "error", err)
		return
	}
	w.log.Debug("ciclo concluído", "job", name, "duration_ms", time.Since(started).Milliseconds())
}

func (w *worker) healthHandler() http.Handler {
	mux := http.NewServeMux()
	mux.HandleFunc("GET /health/live", func(rw http.ResponseWriter, _ *http.Request) {
		httpx.WriteJSON(rw, http.StatusOK, map[string]string{"status": "ok"})
	})
	mux.HandleFunc("GET /health/ready", func(rw http.ResponseWriter, req *http.Request) {
		if !w.leader.Load() {
			httpx.WriteProblem(rw, http.StatusServiceUnavailable, "Worker em espera (outra instância está ativa)", "STANDBY")
			return
		}
		ctx, cancel := context.WithTimeout(req.Context(), 2*time.Second)
		defer cancel()
		if err := w.pool.Ping(ctx); err != nil {
			httpx.WriteProblem(rw, http.StatusServiceUnavailable, "Banco de dados indisponível", "DB_UNAVAILABLE")
			return
		}
		httpx.WriteJSON(rw, http.StatusOK, map[string]string{"status": "ok", "role": "leader"})
	})
	return mux
}