package app import ( "context" "errors" "sync" "time" "botpontos/internal/adapters/postgres" "botpontos/internal/domain" "botpontos/internal/rules" ) const stateOutage = "pontomais_outage" type outageState struct { Since time.Time `json:"since"` LastError string `json:"last_error"` } // reconJob é um usuário que deve ser consultado neste tick. type reconJob struct { ru postgres.ReconUser loc *time.Location day string window domain.QueryWindow state domain.UserDay } type reconResult struct { job reconJob points []domain.Point err error skipped bool // não chegou a ser consultado (prazo do ciclo esgotado) } // Reconcile é o ciclo de 15 min (8.1, 8.2, 9.4, 9.12): seleciona quem está // na janela e com a próxima consulta vencida, consulta a PontoMais com // concorrência limitada e aplica as regras. func (s *Service) Reconcile(ctx context.Context, tick time.Time) error { global, err := s.GlobalWindow(ctx) if err != nil { return err } users, err := s.store.ListReconciliationUsers(ctx) if err != nil { return err } var due []reconJob for _, ru := range users { loc := ru.User.Location() day := domain.LocalDay(tick, loc) window, _ := rules.EffectiveWindow(global, ru.Schedule) if !rules.InWindow(tick, window, day, loc) { continue } if business, _ := s.IsBusinessDay(ctx, tick, loc); !business { continue } state, found, err := s.store.GetUserDay(ctx, ru.User.ID, day) if err != nil { return err } if found && (state.NextCheckAt == nil || state.NextCheckAt.After(tick)) { continue } due = append(due, reconJob{ru: ru, loc: loc, day: day, window: window, state: state}) } if len(due) == 0 { return nil } started := s.now() results := s.fetchAll(ctx, due) var successes, failures []reconResult for _, r := range results { switch { case r.skipped: case r.err == nil || errors.Is(r.err, domain.ErrInvalidPointHistory): successes = append(successes, r) // a API respondeu (mesmo com dado inválido) default: failures = append(failures, r) } } var outage outageState outageActive, err := s.store.GetState(ctx, stateOutage, &outage) if err != nil { s.log.Warn("falha ao ler estado de queda", "error", err) } attempted := len(successes) + len(failures) s.metrics.ReconciliationCycle(s.now().Sub(started), attempted, len(failures)) switch { case attempted >= 2 && len(successes) == 0: // Todos falharam: queda da PontoMais. Ninguém soma falha (9.4). if !outageActive { s.startOutage(ctx, len(failures), failures[0].err) } return nil case outageActive && len(successes) == 0: // Queda em andamento e só um usuário consultado: não conta. return nil case outageActive: s.endOutage(ctx, outage) } for _, r := range successes { if r.err != nil { s.handleInvalidHistory(ctx, r.job, tick, r.err) continue } if err := s.applyPoints(ctx, r.job, r.points, tick); err != nil { s.log.Error("falha ao aplicar pontos", "user_id", r.job.ru.User.ID, "error", err) } } for _, r := range failures { s.recordFailure(ctx, r.job, r.err) } return nil } func (s *Service) fetchAll(ctx context.Context, jobs []reconJob) []reconResult { deadline, cancel := context.WithTimeout(ctx, s.cfg.Jobs.ReconciliationDeadline) defer cancel() results := make([]reconResult, len(jobs)) sem := make(chan struct{}, s.cfg.Jobs.ReconciliationWorkers) var wg sync.WaitGroup for i, job := range jobs { wg.Add(1) go func(i int, job reconJob) { defer wg.Done() results[i].job = job select { case sem <- struct{}{}: case <-deadline.Done(): results[i].skipped = true return } defer func() { <-sem }() requestStart := time.Now() points, err := s.points.History(deadline, job.ru.Credential, job.loc) s.metrics.PointsRequest(time.Since(requestStart), err) if err != nil && deadline.Err() != nil && ctx.Err() == nil { // Prazo do ciclo esgotado: fica para o próximo tick, sem contar falha. results[i].skipped = true return } results[i].points, results[i].err = points, err }(i, job) } wg.Wait() return results } func (s *Service) startOutage(ctx context.Context, users int, cause error) { state := outageState{Since: s.now(), LastError: shortError(cause)} if err := s.store.SetState(ctx, stateOutage, state); err != nil { s.log.Warn("falha ao registrar queda", "error", err) } s.audit(ctx, WorkerActor, nil, domain.EventPointAPIOutageStarted, false, map[string]any{"users": users, "error": state.LastError}) s.notifyAdmins(ctx, msgOutageStarted(users, cause)) } func (s *Service) endOutage(ctx context.Context, outage outageState) { if err := s.store.DeleteState(ctx, stateOutage); err != nil { s.log.Warn("falha ao encerrar queda", "error", err) } duration := s.now().Sub(outage.Since) s.audit(ctx, WorkerActor, nil, domain.EventPointAPIOutageEnded, true, map[string]any{"minutes": int(duration.Minutes())}) s.notifyAdmins(ctx, msgOutageEnded(duration)) } // recordFailure aplica a regra das 3 falhas seguidas (9.4). O usuário volta a // ser consultado no próximo tick, porque o estado do dia não muda. func (s *Service) recordFailure(ctx context.Context, job reconJob, cause error) { u := job.ru.User failures, err := s.store.RecordCredentialFailure(ctx, u.ID, shortError(cause), s.now()) if err != nil { s.log.Error("falha ao registrar falha da PontoMais", "user_id", u.ID, "error", err) return } s.audit(ctx, WorkerActor, &u.ID, domain.EventPointAPIFailure, false, map[string]any{"consecutive_failures": failures, "error": shortError(cause)}) if failures < s.cfg.Jobs.APIFailureThreshold { return } reason := domain.PauseAPIFailures if err := s.store.SetNotifications(ctx, u.ID, false, &reason); err != nil { s.log.Error("falha ao pausar lembretes", "user_id", u.ID, "error", err) return } s.audit(ctx, WorkerActor, &u.ID, domain.EventNotificationsPausedAPI, false, map[string]any{"consecutive_failures": failures}) s.notifyUser(ctx, u, msgPausedAPIFailures(failures)) s.notifyAdmins(ctx, msgAdminUserPaused(u, failures, cause)) } // handleInvalidHistory: a API respondeu com dado ilegível. Não conta como falha; // consulta de novo mais devagar. func (s *Service) handleInvalidHistory(ctx context.Context, job reconJob, tick time.Time, cause error) { u := job.ru.User s.audit(ctx, WorkerActor, &u.ID, domain.EventInvalidPointHistory, false, map[string]any{"error": shortError(cause)}) next := tick.Add(s.cfg.Jobs.SlowCheckInterval) state := job.state if next.After(job.window.End.On(job.day, job.loc)) { state.NextCheckAt = nil } else { state.NextCheckAt = &next } if err := s.store.SaveUserDay(ctx, state); err != nil { s.log.Error("falha ao salvar o dia", "user_id", u.ID, "error", err) } } // applyPoints cria, mantém ou cancela os lembretes e calcula a próxima consulta. func (s *Service) applyPoints(ctx context.Context, job reconJob, points []domain.Point, tick time.Time) error { u, schedule := job.ru.User, job.ru.Schedule if job.ru.Credential.ConsecutiveFailures > 0 || job.ru.Credential.LastError != nil { if err := s.store.ResetCredentialFailures(ctx, u.ID); err != nil { return err } } wd := rules.CurrentWorkday(points, job.day, job.loc) previous := job.state var saida2At *time.Time switch { case wd.Invalid: if previous.LastPointCount != wd.Count() || previous.UpdatedAt.IsZero() { s.audit(ctx, WorkerActor, &u.ID, domain.EventInvalidPointHistory, false, map[string]any{"reason": "pontos repetidos ou fora de ordem", "points": wd.Count()}) } case wd.Overtime || wd.Count() >= 4: cancelled, err := s.store.CancelScheduledFrom(ctx, u.ID, job.day) if err != nil { return err } if cancelled > 0 { s.audit(ctx, WorkerActor, &u.ID, domain.EventNotificationCancelled, true, map[string]any{"reason": "jornada encerrada", "count": cancelled}) } case wd.Count() == 3: cancelled, err := s.store.CancelScheduled(ctx, u.ID, job.day, domain.NotificationEntrada2) if err != nil { return err } if cancelled { s.audit(ctx, WorkerActor, &u.ID, domain.EventNotificationCancelled, true, map[string]any{"type": "entrada2", "reason": "retorno do almoço detectado"}) } predicted, err := rules.Saida2Time(wd, schedule.DailyWorkMinutes) if err != nil { if previous.LastPointCount != 3 { s.audit(ctx, WorkerActor, &u.ID, domain.EventInvalidWorkday, false, map[string]any{"error": err.Error()}) } break } n, err := s.ensureReminder(ctx, job, domain.NotificationSaida2, predicted, schedule.Saida2LeadMinutes) if err != nil { return err } if n.Status != domain.NotificationCancelled { saida2At = &n.ScheduledAt } case wd.Count() == 2: predicted, err := rules.Entrada2Time(wd, schedule.LunchMinutes) if err != nil { return err } if _, err := s.ensureReminder(ctx, job, domain.NotificationEntrada2, predicted, schedule.Entrada2LeadMinutes); err != nil { return err } } plan := rules.PlanNextCheck(rules.PlanInput{ Tick: tick, Day: job.day, Location: job.loc, Window: job.window, Workday: wd, Resumed: previous.ResumedBy != nil, Saida2At: saida2At, Reconciliation: s.cfg.Jobs.ReconciliationInterval, Slow: s.cfg.Jobs.SlowCheckInterval, ExitConfirmFrom: s.cfg.Jobs.ExitConfirmFrom, ExitConfirmTo: s.cfg.Jobs.ExitConfirmTo, }) // Se o dia foi retomado (/ativarponto ou painel) durante a consulta, a // retomada vence: o próximo tick recalcula com o estado novo. if current, found, err := s.store.GetUserDay(ctx, u.ID, job.day); err == nil && found && previous.ResumedBy == nil && current.ResumedBy != nil { return nil } next := previous next.Status = plan.Status next.NextCheckAt = plan.NextCheckAt next.LastPointCount = wd.Count() if wd.Count() == 0 { next.ZeroPointChecks++ } if err := s.store.SaveUserDay(ctx, next); err != nil { return err } if plan.Status == domain.DayNoPoints && previous.Status != domain.DayNoPoints { s.audit(ctx, WorkerActor, &u.ID, domain.EventDayNoPoints, true, nil) } if plan.Status == domain.DayNotWorking && previous.Status != domain.DayNotWorking { s.audit(ctx, WorkerActor, &u.ID, domain.EventDayNotWorked, true, nil) } return nil } // ensureReminder cria o lembrete do dia se ainda não existir (um existente não // é alterado). Se o horário do aviso já passou da tolerância, ele nasce // expired e é só auditado (9.3). func (s *Service) ensureReminder(ctx context.Context, job reconJob, t domain.NotificationType, predicted time.Time, lead int) (domain.Notification, error) { u := job.ru.User now := s.now() n := domain.Notification{ UserID: u.ID, ReferenceDate: job.day, Type: t, ScheduledAt: rules.ReminderAt(predicted, lead), PredictedAt: predicted, Status: domain.NotificationScheduled, } delay := now.Sub(n.ScheduledAt) if delay > s.cfg.Jobs.LateTolerance { n.Status = domain.NotificationExpired n.ErrorMessage = ptr("detectado com atraso acima da tolerância") } saved, inserted, err := s.store.InsertNotification(ctx, n) if err != nil || !inserted { return saved, err } details := map[string]any{ "type": string(t), "scheduled_at": saved.ScheduledAt, "predicted_at": saved.PredictedAt, } if saved.Status == domain.NotificationExpired { details["delay_minutes"] = int(delay.Minutes()) s.audit(ctx, WorkerActor, &u.ID, domain.EventNotificationExpired, false, details) } else { s.audit(ctx, WorkerActor, &u.ID, domain.EventNotificationScheduled, true, details) } return saved, nil } // CheckScheduleIntervals avisa, na subida do worker, sobre jornadas que ficaram // fora da regra 9.8 após mudança do intervalo de reconciliação. func (s *Service) CheckScheduleIntervals(ctx context.Context) { items, err := s.store.ListActiveSchedules(ctx) if err != nil { s.log.Warn("falha ao verificar jornadas", "error", err) return } var names []string for _, item := range items { if !rules.ScheduleFitsInterval(item.Schedule, s.cfg.Jobs.ReconciliationInterval) { names = append(names, item.User.DisplayName) s.audit(ctx, WorkerActor, &item.User.ID, domain.EventScheduleWindowViolation, false, nil) } } if len(names) > 0 { s.notifyAdmins(ctx, msgScheduleViolations(names, s.cfg.Jobs.ReconciliationInterval)) } }