233 lines
6.5 KiB
Go
233 lines
6.5 KiB
Go
// 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
|
|
}
|