package postgres import ( "context" "time" "botpontos/internal/domain" ) func notificationCols(p string) string { return cols(p, "id", "user_id") + ", to_char(" + p + "reference_date, 'YYYY-MM-DD'), " + cols(p, "type", "scheduled_at", "predicted_at", "status", "attempts", "locked_at", "sent_at", "telegram_message_id", "error_message", "created_at", "updated_at") } func readNotification(row scanner) (domain.Notification, error) { var n domain.Notification var typ, status string if err := row.Scan(&n.ID, &n.UserID, &n.ReferenceDate, &typ, &n.ScheduledAt, &n.PredictedAt, &status, &n.Attempts, &n.LockedAt, &n.SentAt, &n.TelegramMessageID, &n.ErrorMessage, &n.CreatedAt, &n.UpdatedAt); err != nil { return n, mapErr(err) } n.Type = domain.NotificationType(typ) n.Status = domain.NotificationStatus(status) return n, nil } func (s *Store) GetNotification(ctx context.Context, id int64) (domain.Notification, error) { return readNotification(s.pool.QueryRow(ctx, `SELECT `+notificationCols("")+` FROM notifications WHERE id = $1`, id)) } // FindNotification devolve o lembrete do usuário para o dia e tipo; found=false se não existir. func (s *Store) FindNotification(ctx context.Context, userID int64, day string, t domain.NotificationType) (domain.Notification, bool, error) { n, err := readNotification(s.pool.QueryRow(ctx, ` SELECT `+notificationCols("")+` FROM notifications WHERE user_id = $1 AND reference_date = $2::text::date AND type::text = $3`, userID, day, string(t))) if err == domain.ErrNotFound { return n, false, nil } return n, err == nil, err } // InsertNotification cria o lembrete se ainda não existir um para (usuário, dia, tipo). // inserted=false devolve o lembrete que já existia, sem alterá-lo. func (s *Store) InsertNotification(ctx context.Context, n domain.Notification) (domain.Notification, bool, error) { created, err := readNotification(s.pool.QueryRow(ctx, ` INSERT INTO notifications (user_id, reference_date, type, scheduled_at, predicted_at, status, error_message) VALUES ($1, $2::text::date, $3::text::notification_type, $4, $5, $6::text::notification_status, $7) ON CONFLICT (user_id, reference_date, type) DO NOTHING RETURNING `+notificationCols(""), n.UserID, n.ReferenceDate, string(n.Type), n.ScheduledAt, n.PredictedAt, string(n.Status), n.ErrorMessage)) if err == domain.ErrNotFound { existing, _, err := s.FindNotification(ctx, n.UserID, n.ReferenceDate, n.Type) return existing, false, err } return created, err == nil, err } // CancelScheduled cancela o lembrete agendado de um tipo no dia. func (s *Store) CancelScheduled(ctx context.Context, userID int64, day string, t domain.NotificationType) (bool, error) { tag, err := s.pool.Exec(ctx, ` UPDATE notifications SET status = 'cancelled' WHERE user_id = $1 AND reference_date = $2::text::date AND type::text = $3 AND status = 'scheduled'`, userID, day, string(t)) return tag.RowsAffected() > 0, err } // CancelScheduledFrom cancela todos os lembretes agendados do usuário a partir do dia. func (s *Store) CancelScheduledFrom(ctx context.Context, userID int64, day string) (int64, error) { tag, err := s.pool.Exec(ctx, ` UPDATE notifications SET status = 'cancelled' WHERE user_id = $1 AND reference_date >= $2::text::date AND status = 'scheduled'`, userID, day) return tag.RowsAffected(), err } // CancelNotification cancela um lembrete específico, só se ainda estiver agendado. func (s *Store) CancelNotification(ctx context.Context, id int64) (domain.Notification, error) { return readNotification(s.pool.QueryRow(ctx, ` UPDATE notifications SET status = 'cancelled', error_message = 'cancelado pelo painel' WHERE id = $1 AND status = 'scheduled' RETURNING `+notificationCols(""), id)) } // ExpireOverdue marca como expired o que passou da tolerância de atraso (9.3, etapa 1). func (s *Store) ExpireOverdue(ctx context.Context, before time.Time) ([]domain.Notification, error) { rows, err := s.pool.Query(ctx, ` UPDATE notifications SET status = 'expired', error_message = 'atraso acima da tolerância' WHERE status = 'scheduled' AND scheduled_at < $1 RETURNING `+notificationCols(""), before) return collect(rows, err, readNotification) } // ClaimDue reserva os lembretes vencidos (scheduled → sending) sem duplicidade (6.3, etapa 2). func (s *Store) ClaimDue(ctx context.Context, now time.Time, limit int) ([]domain.Notification, error) { rows, err := s.pool.Query(ctx, ` WITH due AS ( SELECT id FROM notifications WHERE status = 'scheduled' AND scheduled_at <= $1 ORDER BY scheduled_at LIMIT $2 FOR UPDATE SKIP LOCKED ) UPDATE notifications n SET status = 'sending', locked_at = now(), attempts = n.attempts + 1 FROM due WHERE n.id = due.id RETURNING `+notificationCols("n."), now, limit) return collect(rows, err, readNotification) } // ClaimFailed reserva um lembrete failed para o reenvio manual (9.5). A troca // de status é atômica: duplo clique ou dois admins resultam num envio só. func (s *Store) ClaimFailed(ctx context.Context, id int64) (domain.Notification, error) { return readNotification(s.pool.QueryRow(ctx, ` UPDATE notifications SET status = 'sending', locked_at = now(), attempts = attempts + 1, error_message = NULL WHERE id = $1 AND status = 'failed' RETURNING `+notificationCols(""), id)) } func (s *Store) MarkSent(ctx context.Context, id, messageID int64, at time.Time) error { return expectOne(s.pool.Exec(ctx, ` UPDATE notifications SET status = 'sent', sent_at = $3, telegram_message_id = $2, error_message = NULL WHERE id = $1 AND status = 'sending'`, id, messageID, at)) } // MarkFinished encerra um lembrete em envio com failed, expired ou cancelled. func (s *Store) MarkFinished(ctx context.Context, id int64, status domain.NotificationStatus, message string) error { return expectOne(s.pool.Exec(ctx, ` UPDATE notifications SET status = $2::text::notification_status, error_message = $3 WHERE id = $1 AND status = 'sending'`, id, string(status), message)) } // FailStuckSending marca como failed o que está em sending há tempo demais (crash no envio). func (s *Store) FailStuckSending(ctx context.Context, before time.Time) ([]domain.Notification, error) { rows, err := s.pool.Query(ctx, ` UPDATE notifications SET status = 'failed', error_message = 'envio interrompido (reinício ou travamento)' WHERE status = 'sending' AND locked_at < $1 RETURNING `+notificationCols(""), before) return collect(rows, err, readNotification) } type NotificationView struct { Notification domain.Notification UserName string } type NotificationFilter struct { Day *string Status *string Type *string UserID *int64 Limit int } func (s *Store) ListNotifications(ctx context.Context, f NotificationFilter) ([]NotificationView, error) { if f.Limit <= 0 || f.Limit > 500 { f.Limit = 200 } rows, err := s.pool.Query(ctx, ` SELECT `+notificationCols("n.")+`, u.display_name FROM notifications n JOIN users u ON u.id = n.user_id WHERE ($1::text IS NULL OR n.reference_date = $1::text::date) AND ($2::text IS NULL OR n.status::text = $2) AND ($3::text IS NULL OR n.type::text = $3) AND ($4::bigint IS NULL OR n.user_id = $4) ORDER BY n.reference_date DESC, n.scheduled_at DESC LIMIT $5`, f.Day, f.Status, f.Type, f.UserID, f.Limit) return collect(rows, err, func(row scanner) (NotificationView, error) { var v NotificationView var typ, status string n := &v.Notification err := row.Scan(&n.ID, &n.UserID, &n.ReferenceDate, &typ, &n.ScheduledAt, &n.PredictedAt, &status, &n.Attempts, &n.LockedAt, &n.SentAt, &n.TelegramMessageID, &n.ErrorMessage, &n.CreatedAt, &n.UpdatedAt, &v.UserName) n.Type = domain.NotificationType(typ) n.Status = domain.NotificationStatus(status) return v, err }) }