276 lines
11 KiB
Go
276 lines
11 KiB
Go
package repository
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"time"
|
||
|
||
"github.com/jackc/pgx/v5"
|
||
)
|
||
|
||
// TelegramRepo — привязка Telegram, токены собственных ботов и выборка
|
||
// заявок-кандидатов на уведомление (миграции 0011–0012). Держим отдельно
|
||
// от профилей и заявок: тема одна, а SQL здесь — про «куда слать»,
|
||
// «чем слать» и «сколько уже слали».
|
||
type TelegramRepo struct {
|
||
db Querier
|
||
}
|
||
|
||
// NewTelegramRepo создаёт репозиторий поверх Querier (пул или транзакция).
|
||
func NewTelegramRepo(db Querier) *TelegramRepo {
|
||
return &TelegramRepo{db: db}
|
||
}
|
||
|
||
// Target — куда слать уведомления ("" — не настроено; NULL в базе
|
||
// нормализуется в пустую строку). ErrNotFound — профиля нет.
|
||
func (r *TelegramRepo) Target(ctx context.Context, userID string) (string, error) {
|
||
var target *string
|
||
err := r.db.QueryRow(ctx,
|
||
`SELECT telegram_target FROM profiles WHERE id = $1`, userID).Scan(&target)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return "", ErrNotFound
|
||
}
|
||
if err != nil {
|
||
return "", fmt.Errorf("telegram target: %w", err)
|
||
}
|
||
if target == nil {
|
||
return "", nil
|
||
}
|
||
return *target, nil
|
||
}
|
||
|
||
// SetTarget записывает цель уведомлений; пустая строка очищает
|
||
// (NULLIF кладёт NULL — «отключено» физически, а не пустым текстом).
|
||
// ErrNotFound — профиля нет (сессия могла пережить удаление).
|
||
func (r *TelegramRepo) SetTarget(ctx context.Context, userID, target string) error {
|
||
tag, err := r.db.Exec(ctx, `
|
||
UPDATE profiles SET telegram_target = NULLIF($2, '')
|
||
WHERE id = $1`, userID, target)
|
||
if err != nil {
|
||
return fmt.Errorf("set telegram target: %w", err)
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
return ErrNotFound
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// CreateBindCode выпускает одноразовый код привязки: прежние
|
||
// неиспользованные коды пользователя гаснут — действует ровно один
|
||
// свежий, как и токены сброса пароля (CreateResetToken).
|
||
// Возвращает срок жизни (time.Now()+ttl).
|
||
func (r *TelegramRepo) CreateBindCode(ctx context.Context, userID, codeHash string, ttl time.Duration) (time.Time, error) {
|
||
if _, err := r.db.Exec(ctx, `
|
||
DELETE FROM telegram_bind_codes
|
||
WHERE user_id = $1 AND used_at IS NULL`, userID); err != nil {
|
||
return time.Time{}, fmt.Errorf("supersede bind codes: %w", err)
|
||
}
|
||
expires := time.Now().Add(ttl)
|
||
if _, err := r.db.Exec(ctx, `
|
||
INSERT INTO telegram_bind_codes (user_id, code_hash, expires_at)
|
||
VALUES ($1, $2, $3)`, userID, codeHash, expires); err != nil {
|
||
return time.Time{}, fmt.Errorf("create bind code: %w", err)
|
||
}
|
||
return expires, nil
|
||
}
|
||
|
||
// ConsumeBindCode гасит код одним UPDATE — одноразовость и срок живут
|
||
// в самом WHERE, поэтому две параллельные привязки невозможны.
|
||
// Возвращает владельца кода; ErrNotFound — неизвестен, уже использован
|
||
// или истёк (причина наружу не раскрывается).
|
||
func (r *TelegramRepo) ConsumeBindCode(ctx context.Context, codeHash string) (string, error) {
|
||
var userID string
|
||
err := r.db.QueryRow(ctx, `
|
||
UPDATE telegram_bind_codes
|
||
SET used_at = now()
|
||
WHERE code_hash = $1 AND used_at IS NULL AND expires_at > now()
|
||
RETURNING user_id`, codeHash).Scan(&userID)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return "", ErrNotFound
|
||
}
|
||
if err != nil {
|
||
return "", fmt.Errorf("consume bind code: %w", err)
|
||
}
|
||
return userID, nil
|
||
}
|
||
|
||
// CleanupBindCodes удаляет протухшие строки (использованные тоже:
|
||
// история кодов никому не нужна, таблица должна оставаться маленькой).
|
||
func (r *TelegramRepo) CleanupBindCodes(ctx context.Context, before time.Time) (int64, error) {
|
||
tag, err := r.db.Exec(ctx,
|
||
`DELETE FROM telegram_bind_codes WHERE expires_at < $1`, before)
|
||
if err != nil {
|
||
return 0, fmt.Errorf("cleanup bind codes: %w", err)
|
||
}
|
||
return tag.RowsAffected(), nil
|
||
}
|
||
|
||
// BotTokenInfo — есть ли у пользователя свой токен бота и как его бот
|
||
// называется (имя из getMe при сохранении). Токен сюда не входит: он
|
||
// наружу не отдаётся никому, включая владельца.
|
||
// ErrNotFound — профиля нет (сессия могла пережить удаление).
|
||
func (r *TelegramRepo) BotTokenInfo(ctx context.Context, userID string) (has bool, username string, err error) {
|
||
var enc []byte
|
||
var name *string
|
||
err = r.db.QueryRow(ctx, `
|
||
SELECT telegram_bot_token, telegram_bot_username
|
||
FROM profiles WHERE id = $1`, userID).Scan(&enc, &name)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return false, "", ErrNotFound
|
||
}
|
||
if err != nil {
|
||
return false, "", fmt.Errorf("telegram bot token info: %w", err)
|
||
}
|
||
if name != nil {
|
||
username = *name
|
||
}
|
||
return enc != nil, username, nil
|
||
}
|
||
|
||
// SetBotToken сохраняет зашифрованный токен и имя бота (оба поля одной
|
||
// операцией — чтобы статус не показывал имя бота без токена и наоборот).
|
||
// ErrNotFound — профиля нет.
|
||
func (r *TelegramRepo) SetBotToken(ctx context.Context, userID string, encToken, username string) error {
|
||
tag, err := r.db.Exec(ctx, `
|
||
UPDATE profiles
|
||
SET telegram_bot_token = $2, telegram_bot_username = NULLIF($3, '')
|
||
WHERE id = $1`, userID, []byte(encToken), username)
|
||
if err != nil {
|
||
return fmt.Errorf("set telegram bot token: %w", err)
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
return ErrNotFound
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ClearBotToken убирает свой токен: уведомления перестают уходить через
|
||
// этого бота, его поллер останавливается (см. notify.Registry).
|
||
// ErrNotFound — профиля нет.
|
||
func (r *TelegramRepo) ClearBotToken(ctx context.Context, userID string) error {
|
||
tag, err := r.db.Exec(ctx, `
|
||
UPDATE profiles
|
||
SET telegram_bot_token = NULL, telegram_bot_username = NULL
|
||
WHERE id = $1`, userID)
|
||
if err != nil {
|
||
return fmt.Errorf("clear telegram bot token: %w", err)
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
return ErrNotFound
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// UserBot — пользователь со своим ботом: токен приходит зашифрованным,
|
||
// расшифровывает вызывающий (репозиторий не знает про ключ).
|
||
type UserBot struct {
|
||
UserID string
|
||
// TokenEnc — зашифрованный токен (AES-GCM от crypt.Cipher).
|
||
TokenEnc string
|
||
// Username — имя бота из getMe, чтобы показывать его до первого
|
||
// сетевого вызова.
|
||
Username string
|
||
}
|
||
|
||
// BotTokens — все сохранённые токены: читается один раз при старте,
|
||
// чтобы поднять клиенты ботов и их поллеры.
|
||
func (r *TelegramRepo) BotTokens(ctx context.Context) ([]UserBot, error) {
|
||
rows, err := r.db.Query(ctx, `
|
||
SELECT id, telegram_bot_token, telegram_bot_username
|
||
FROM profiles
|
||
WHERE telegram_bot_token IS NOT NULL
|
||
ORDER BY created_at`)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("list telegram bot tokens: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
|
||
out := make([]UserBot, 0, 8)
|
||
for rows.Next() {
|
||
var u UserBot
|
||
var enc []byte
|
||
var name *string
|
||
if err := rows.Scan(&u.UserID, &enc, &name); err != nil {
|
||
return nil, fmt.Errorf("list telegram bot tokens scan: %w", err)
|
||
}
|
||
u.TokenEnc = string(enc)
|
||
if name != nil {
|
||
u.Username = *name
|
||
}
|
||
out = append(out, u)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, fmt.Errorf("list telegram bot tokens: %w", err)
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// StaleTicket — заявка, которую пора напомнить, и куда напоминание слать.
|
||
type StaleTicket struct {
|
||
ID int64
|
||
Address string
|
||
Category string
|
||
Priority string
|
||
ClientName string
|
||
// ClientID — владелец заявки: по нему выбирается бот, через который
|
||
// уйдёт сообщение (свой токен пользователя или глобальный).
|
||
ClientID string
|
||
// Target — profiles.telegram_target: числовой chat_id или @группа.
|
||
Target string
|
||
}
|
||
|
||
// StaleTickets — кандидаты на уведомление: всё ещё 'new', созданы раньше
|
||
// cutoff, и последнее уведомление либо не уходило, либо тоже старше cutoff
|
||
// (повтор — раз в сутки). Без telegram_target и с выключенным аккаунтом —
|
||
// не попадают: слать некуда. Каким ботом слать — выбирается по ClientID
|
||
// на стороне нотификатора (личный токен пользователя или глобальный).
|
||
func (r *TelegramRepo) StaleTickets(ctx context.Context, cutoff time.Time, limit int) ([]StaleTicket, error) {
|
||
rows, err := r.db.Query(ctx, `
|
||
SELECT t.id, t.address, t.category, t.priority, p.name,
|
||
t.client_id, p.telegram_target
|
||
FROM tickets t
|
||
JOIN profiles p ON p.id = t.client_id
|
||
WHERE t.status = 'new'
|
||
AND t.created_at <= $1
|
||
AND (t.tg_notified_at IS NULL OR t.tg_notified_at <= $1)
|
||
AND p.is_active
|
||
AND p.telegram_target IS NOT NULL
|
||
ORDER BY t.created_at
|
||
LIMIT $2`, cutoff, limit)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("stale tickets: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
|
||
out := make([]StaleTicket, 0, 16)
|
||
for rows.Next() {
|
||
var s StaleTicket
|
||
if err := rows.Scan(&s.ID, &s.Address, &s.Category, &s.Priority,
|
||
&s.ClientName, &s.ClientID, &s.Target); err != nil {
|
||
return nil, fmt.Errorf("stale tickets scan: %w", err)
|
||
}
|
||
out = append(out, s)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, fmt.Errorf("stale tickets: %w", err)
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// MarkNotified фиксирует отправку: со следующего тика заявка молчит
|
||
// сутки. Условие status='new' — гонка со взятием заявки в работу:
|
||
// если её успели назначить, напоминание уже неактуально.
|
||
func (r *TelegramRepo) MarkNotified(ctx context.Context, ticketID int64) error {
|
||
tag, err := r.db.Exec(ctx, `
|
||
UPDATE tickets SET tg_notified_at = now()
|
||
WHERE id = $1 AND status = 'new'`, ticketID)
|
||
if err != nil {
|
||
return fmt.Errorf("mark notified: %w", err)
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
return ErrNotFound
|
||
}
|
||
return nil
|
||
}
|