axe/internal/repository/recurring.go

package repository

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"

	"building-mgmt/internal/models"
)

// RecurringRepo — правила повторяющихся заявок и их срабатывание
// (миграции 0014/0016). Правило — шаблон будущей заявки (объект,
// категория, описание, приоритет) плюс расписание «каждые period_days
// дней/недель от anchor_at»; фоновый планировщик в cmd/api зовёт
// RunDueRules и создаёт заявки по расписанию.
//
// Проверка владельца встроена в запросы, как у проектов: clientID != ""
// ограничивает выборку, чужое правило неотличимо от несуществующего.
type RecurringRepo struct {
	db Querier
}

// NewRecurringRepo создаёт репозиторий поверх Querier (пул или транзакция).
func NewRecurringRepo(db Querier) *RecurringRepo {
	return &RecurringRepo{db: db}
}

// nextDue — ближайший запуск правила: наименьшая точка расписания
// anchor + k·period (k ≥ 0), строго после отправной точки ref.
//
// Сетка сроков строится от anchor_at и не «плывёт» после каждого
// срабатывания: опоздание на тик не накапливается. Отправная точка —
// COALESCE(last_run_at, created_at): у правила, которое ещё ни разу не
// срабатывало, первая заявка — первая точка сетки после создания
// (anchor в будущем — это сам anchor: пользователь выбрал первый запуск).
//
// periodDays ≥ 1 держится CHECK'ом в БД; защита ниже — чтобы деление
// на ноль при битых данных не уронило планировщик.
func nextDue(anchor time.Time, periodDays int, ref time.Time) time.Time {
	if periodDays < 1 {
		periodDays = 1
	}
	a, r := anchor.UTC(), ref.UTC()
	if r.Before(a) {
		return a
	}
	period := time.Duration(periodDays) * 24 * time.Hour
	// Сколько полных периодов прошло между anchor и ref, плюс один:
	// срок обязан быть строго после ref (ровно на ref — уже прошло).
	return a.Add(time.Duration(int64(r.Sub(a)/period)+1) * period)
}

// recurringColumns и recurringFrom — одна проекция для всех выборок:
// правило плюс денормализации для UI (проект, объект, адрес, имена).
const recurringColumns = `
        r.id, r.client_id, r.project_id, r.object_id,
        r.category, r.description, r.priority, COALESCE(r.floor, ''),
        r.anchor_at, r.period_days, r.is_active, r.last_run_at, r.created_at,
        COALESCE(r.created_by::text, ''),
        p.name, o.name, o.address,
        COALESCE(c.name, ''), COALESCE(cb.name, '')`
const recurringFrom = `
        FROM recurring_rules r
        JOIN projects p ON p.id = r.project_id
        JOIN project_objects o ON o.id = r.object_id
        LEFT JOIN profiles c ON c.id = r.client_id
        LEFT JOIN profiles cb ON cb.id = r.created_by`

// scanRule читает строку в модель; порядок Scan = recurringColumns.
// extra — доп. назначения в конце SELECT (COUNT(*) OVER() у списка).
func scanRule(s scanner, extra ...any) (models.RecurringRule, error) {
	var rule models.RecurringRule
	dest := []any{
		&rule.ID, &rule.ClientID, &rule.ProjectID, &rule.ObjectID,
		&rule.Category, &rule.Description, &rule.Priority, &rule.Floor,
		&rule.AnchorAt, &rule.PeriodDays, &rule.IsActive, &rule.LastRunAt, &rule.CreatedAt,
		&rule.CreatedBy,
		&rule.ProjectName, &rule.ObjectName, &rule.Address,
		&rule.ClientName, &rule.CreatedByName,
	}
	dest = append(dest, extra...)
	if err := s.Scan(dest...); err != nil {
		return rule, classify(err)
	}
	// Ближайший запуск — чистая функция расписания и отправной точки,
	// поэтому считается здесь, а не хранится (миграция бы устарела
	// при правке периода или после каждого срабатывания).
	next := nextDue(rule.AnchorAt, rule.PeriodDays, dueRef(rule))
	rule.NextRunAt = &next
	return rule, nil
}

// dueRef — отправная точка срока правила: последнее срабатывание
// или создание, если оно ещё ни разу не срабатывало.
func dueRef(rule models.RecurringRule) time.Time {
	if rule.LastRunAt != nil {
		return *rule.LastRunAt
	}
	return rule.CreatedAt
}

// CreateRule — новое правило. Вызывающий обязан уже проверить владение
// объектом (OwnedObject) и активность проекта/объекта — как при создании
// заявки (handlers.TicketHandler.Create).
func (r *RecurringRepo) CreateRule(ctx context.Context, rule *models.RecurringRule) error {
	err := r.db.QueryRow(ctx,
		`INSERT INTO recurring_rules
             (client_id, project_id, object_id, category, description,
              priority, floor, anchor_at, period_days, created_by)
         VALUES ($1, $2, $3, $4, $5, $6, NULLIF($7, ''), $8, $9, NULLIF($10, '')::uuid)
         RETURNING id, created_at`,
		rule.ClientID, rule.ProjectID, rule.ObjectID, rule.Category,
		rule.Description, rule.Priority, rule.Floor, rule.AnchorAt,
		rule.PeriodDays, rule.CreatedBy,
	).Scan(&rule.ID, &rule.CreatedAt)
	return classify(err)
}

// GetRule — правило по id; clientID != "" ограничивает выборку
// правилами этого клиента (чужое → ErrNotFound).
func (r *RecurringRepo) GetRule(ctx context.Context, id int64, clientID string) (*models.RecurringRule, error) {
	query := `SELECT ` + recurringColumns + recurringFrom + ` WHERE r.id = $1`
	args := []any{id}
	if clientID != "" {
		query += ` AND r.client_id = $2`
		args = append(args, clientID)
	}

	rule, err := scanRule(r.db.QueryRow(ctx, query, args...))
	if err != nil {
		return nil, fmt.Errorf("recurring rule %d: %w", id, err)
	}
	return &rule, nil
}

// ListRules — страница правил: все (clientID пуст, для админа) или только
// свои; projectID != 0 сужает до одного проекта (секция в проекте).
// Пустое окно досчитывается тем же WHERE: окно total его не знает.
func (r *RecurringRepo) ListRules(ctx context.Context, clientID string, projectID int64, page Page) ([]models.RecurringRule, int, error) {
	page = page.normalize()
	order, err := orderBy(page, RecurringSorts, "created_at", "r.id DESC")
	if err != nil {
		return nil, 0, err
	}

	where := `1=1`
	var whereArgs []any
	if clientID != "" {
		where += ` AND r.client_id = $` + itoa(len(whereArgs)+1)
		whereArgs = append(whereArgs, clientID)
	}
	if projectID != 0 {
		where += ` AND r.project_id = $` + itoa(len(whereArgs)+1)
		whereArgs = append(whereArgs, projectID)
	}

	query := `SELECT ` + recurringColumns + `, COUNT(*) OVER() AS total` + recurringFrom + `
        WHERE ` + where + order +
		` LIMIT $` + itoa(len(whereArgs)+1) + ` OFFSET $` + itoa(len(whereArgs)+2)
	args := append(append([]any{}, whereArgs...), page.Limit, page.Offset)

	rows, err := r.db.Query(ctx, query, args...)
	if err != nil {
		return nil, 0, classify(err)
	}
	defer rows.Close()

	var rules []models.RecurringRule
	total := 0
	for rows.Next() {
		rule, err := scanRule(rows, &total)
		if err != nil {
			return nil, 0, err
		}
		rules = append(rules, rule)
	}
	if err := rows.Err(); err != nil {
		return nil, 0, classify(err)
	}
	if len(rules) == 0 {
		if err := r.db.QueryRow(ctx,
			`SELECT COUNT(*)`+recurringFrom+` WHERE `+where, whereArgs...).Scan(&total); err != nil {
			return nil, 0, classify(err)
		}
	}
	return rules, total, nil
}

// RulePatch — изменяемые поля правила в UpdateRule: nil означает
// «поле не трогаем» (COALESCE в самом UPDATE). Объект (object_id) не
// входит: перенос правила в другой проект — это удаление и создание,
// иначе легко увести правило в чужой проект.
type RulePatch struct {
	Floor       *string
	Category    *string
	Description *string
	Priority    *string
	AnchorAt    *time.Time
	PeriodDays  *int
	IsActive    *bool
}

// UpdateRule — частичное обновление правила: применятся только
// переданные поля. Чужое или несуществующее → ErrNotFound: выборка
// уже подрезана clientID. Срок не трогается — он пересчитывается сам
// от COALESCE(last_run_at, created_at) с новым расписанием (nextDue).
func (r *RecurringRepo) UpdateRule(ctx context.Context, id int64, clientID string, patch RulePatch) error {
	query := `UPDATE recurring_rules
              SET floor = NULLIF(COALESCE($2, floor), ''),
                  category = COALESCE($3, category),
                  description = COALESCE($4, description),
                  priority = COALESCE($5, priority),
                  anchor_at = COALESCE($6, anchor_at),
                  period_days = COALESCE($7, period_days),
                  is_active = COALESCE($8, is_active)
              WHERE id = $1`
	args := []any{id, patch.Floor, patch.Category, patch.Description,
		patch.Priority, patch.AnchorAt, patch.PeriodDays, patch.IsActive}
	if clientID != "" {
		query += ` AND client_id = $9`
		args = append(args, clientID)
	}

	tag, err := r.db.Exec(ctx, query, args...)
	if err != nil {
		return classify(err)
	}
	if tag.RowsAffected() == 0 {
		return notFound("recurring rule %d", id)
	}
	return nil
}

// DeleteRule — удалить правило; чужое или несуществующее → ErrNotFound.
// История срабатываний уходит каскадом (0014).
func (r *RecurringRepo) DeleteRule(ctx context.Context, id int64, clientID string) error {
	query := `DELETE FROM recurring_rules WHERE id = $1`
	args := []any{id}
	if clientID != "" {
		query += ` AND client_id = $2`
		args = append(args, clientID)
	}

	tag, err := r.db.Exec(ctx, query, args...)
	if err != nil {
		return classify(err)
	}
	if tag.RowsAffected() == 0 {
		return notFound("recurring rule %d", id)
	}
	return nil
}

// ============ история срабатываний ============

// ListRuns — история срабатываний правила (сначала свежие). Заявку
// показываем с адресом/категорией на момент создания; если её удалили
// каскадом (проект), строка истории остаётся, поля пустые (LEFT JOIN).
// Владение проверяет вызывающий — GetRule до этого, здесь только правило.
func (r *RecurringRepo) ListRuns(ctx context.Context, ruleID int64, page Page) ([]models.RecurringRun, int, error) {
	page = page.normalize()
	order, err := orderBy(page, RecurringRunSorts, "run_at", "rr.id DESC")
	if err != nil {
		return nil, 0, err
	}

	rows, err := r.db.Query(ctx,
		`SELECT rr.id, rr.rule_id, rr.ticket_id, rr.run_at,
                COALESCE(t.address, ''), COALESCE(t.category, ''),
                COUNT(*) OVER() AS total
         FROM recurring_runs rr
         LEFT JOIN tickets t ON t.id = rr.ticket_id
         WHERE rr.rule_id = $1`+order+`
         LIMIT $2 OFFSET $3`,
		ruleID, page.Limit, page.Offset)
	if err != nil {
		return nil, 0, classify(err)
	}
	defer rows.Close()

	var runs []models.RecurringRun
	total := 0
	for rows.Next() {
		var run models.RecurringRun
		if err := rows.Scan(&run.ID, &run.RuleID, &run.TicketID, &run.RunAt,
			&run.Address, &run.Category, &total); err != nil {
			return nil, 0, classify(err)
		}
		runs = append(runs, run)
	}
	if err := rows.Err(); err != nil {
		return nil, 0, classify(err)
	}
	if len(runs) == 0 {
		if err := r.db.QueryRow(ctx,
			`SELECT COUNT(*) FROM recurring_runs WHERE rule_id = $1`,
			ruleID).Scan(&total); err != nil {
			return nil, 0, classify(err)
		}
	}
	return runs, total, nil
}

// ============ планировщик ============

// DueRun — результат срабатывания одного правила: что создалось и куда
// слать уведомление (Target — profiles.telegram_target, "" — не настроено).
type DueRun struct {
	RuleID   int64
	TicketID int64
	ClientID string
	Target   string
	Address  string
	Category string
	// Description — текст правила: в уведомлении он объясняет, зачем
	// заявка появилась («осмотр вентиляции по графику»).
	Description string
	Priority    string
}

// RunDueRules — главный метод планировщика: создаёт заявки по всем
// правилам, у которых наступил срок (точка сетки anchor_at + k·period).
//
// Условия срабатывания (всё перепроверяется под блокировкой строки
// правила, поэтому параллельный прогон не создаст дубль):
//   - правило активно, проект и объект включены;
//   - срок наступил: nextDue(anchor_at, period_days,
//     COALESCE(last_run_at, created_at)) <= now — отправная точка —
//     создание, поэтому первая заявка появится первой точкой сетки
//     после заведения правила, а не сразу;
//   - прошлая заявка по этому правилу закрыта — пока открыта, новую не
//     создаём (ждём закрытия), срок при этом не сдвигается: после
//     закрытия следующий тик создаст заявку сразу.
//
// Каждое правило обрабатывается в своей транзакции: ошибка одного не
// откатывает остальные. Возвращает созданные заявки и первую ошибку
// (если была) — вызывающий шлёт уведомления по created, ошибки идут в лог.
func (r *RecurringRepo) RunDueRules(ctx context.Context, now time.Time) ([]DueRun, error) {
	b, ok := r.db.(Beginer)
	if !ok {
		return nil, errors.New("repository: querier cannot begin a transaction")
	}

	ids, err := r.dueRuleIDs(ctx, now)
	if err != nil {
		return nil, err
	}

	var runs []DueRun
	var firstErr error
	for _, id := range ids {
		run, err := r.runRule(ctx, b, id, "", false, now)
		if err != nil {
			if firstErr == nil {
				firstErr = err
			}
			continue
		}
		if run != nil {
			runs = append(runs, *run)
		}
	}
	return runs, firstErr
}

// RunRuleNow — ручной запуск одного правила (POST /recurring/:id/run).
// То же срабатывание, что у планировщика, но без ожидания срока:
// заявка создаётся сразу. Остальные рубежи остаются — правило, проект
// и объект включены, прошлая заявка закрыта (иначе дубль): нарушение —
// ErrConflict с причиной, чужое правило — ErrNotFound (clientID).
// last_run_at сдвигается как обычно — следующий срок отсчитывается
// от ручного запуска.
func (r *RecurringRepo) RunRuleNow(ctx context.Context, id int64, clientID string, now time.Time) (*DueRun, error) {
	b, ok := r.db.(Beginer)
	if !ok {
		return nil, errors.New("repository: querier cannot begin a transaction")
	}
	return r.runRule(ctx, b, id, clientID, true, now)
}

// dueRuleIDs — кандидаты на срабатывание: активные правила, чей проект
// и объект включены, у которых срок наступил (созревание —
// в Go: сетку сроков anchor_at + k·period не выразить в SQL без
// прижатия дней). Проверка «нет открытой заявки» стоит в runRule —
// там она атомарна вместе с созданием.
func (r *RecurringRepo) dueRuleIDs(ctx context.Context, now time.Time) ([]int64, error) {
	rows, err := r.db.Query(ctx,
		`SELECT r.id, r.anchor_at, r.period_days,
                COALESCE(r.last_run_at, r.created_at)
         FROM recurring_rules r
         JOIN projects p ON p.id = r.project_id
         JOIN project_objects o ON o.id = r.object_id
         WHERE r.is_active AND p.is_active AND o.is_active
         ORDER BY r.id`)
	if err != nil {
		return nil, classify(err)
	}
	defer rows.Close()

	var ids []int64
	for rows.Next() {
		var id int64
		var anchor, ref time.Time
		var period int
		if err := rows.Scan(&id, &anchor, &period, &ref); err != nil {
			return nil, classify(err)
		}
		if !nextDue(anchor, period, ref).After(now) {
			ids = append(ids, id)
		}
	}
	return ids, rows.Err()
}

// runRule — срабатывание одного правила в транзакции.
//
// Условия рубежей общие для планировщика и ручного запуска, реакция —
// разная: планировщик (force=false) пропускает правило молча (nil, nil),
// срок не сдвигается, следующий тик проверит снова; ручной запуск
// (force=true) получает ErrNotFound или ErrConflict с причиной — её
// хендлер отдаёт клиенту.
//
// «Срок ещё не наступил» — условие только планировщика
// (nextDue(...) <= now), поэтому ручной запуск созревания не ждёт.
// Отправная точка — COALESCE(last_run_at, created_at): правило, которое
// ещё ни разу не срабатывало, создаёт первую заявку первой точкой
// сетки после создания.
//
// clientID != "" ограничивает выборку правилами клиента — чужое
// неотличимо от несуществующего на этом шаге, а не «прочитал и потом
// запустил». FOR UPDATE сериализует параллельные запуски (второй
// инстанс, два ручных запроса): проверка «нет открытой заявки» и
// создание заявки — одна атомарная операция.
func (r *RecurringRepo) runRule(ctx context.Context, b Beginer, ruleID int64, clientID string, force bool, now time.Time) (*DueRun, error) {
	tx, err := b.Begin(ctx)
	if err != nil {
		return nil, classify(err)
	}
	defer func() { _ = tx.Rollback(ctx) }() // после Commit — безопасный no-op

	// Причина конфликта: планировщику — молчаливый пропуск, ручному
	// запуску — ErrConflict (→ 409) с текстом причины.
	conflict := func(reason error) (*DueRun, error) {
		if !force {
			return nil, nil
		}
		return nil, fmt.Errorf("%w: %w", ErrConflict, reason)
	}

	query := `SELECT r.id, r.client_id, r.project_id, r.object_id,
                 r.category, r.description, r.priority, COALESCE(r.floor, ''),
                 COALESCE(r.created_by::text, ''), o.address,
                 COALESCE(pf.telegram_target, ''),
                 r.is_active, p.is_active, o.is_active,
                 r.anchor_at, r.period_days,
                 COALESCE(r.last_run_at, r.created_at)
          FROM recurring_rules r
          JOIN projects p ON p.id = r.project_id
          JOIN project_objects o ON o.id = r.object_id
          LEFT JOIN profiles pf ON pf.id = r.client_id
          WHERE r.id = $1`
	args := []any{ruleID}
	if clientID != "" {
		query += ` AND r.client_id = $` + itoa(len(args)+1)
		args = append(args, clientID)
	}
	query += ` FOR UPDATE OF r`

	var run DueRun
	var projectID, objectID int64
	var floor, createdBy string
	var ruleActive, projectActive, objectActive bool
	var anchor, ref time.Time
	var period int
	err = tx.QueryRow(ctx, query, args...).
		Scan(&run.RuleID, &run.ClientID, &projectID, &objectID,
			&run.Category, &run.Description, &run.Priority, &floor,
			&createdBy, &run.Address, &run.Target,
			&ruleActive, &projectActive, &objectActive,
			&anchor, &period, &ref)
	if errors.Is(err, pgx.ErrNoRows) {
		// Планировщику: правила нет или чужое (для него созревание
		// уже отфильтровано в dueRuleIDs).
		if !force {
			return nil, nil
		}
		return nil, notFound("recurring rule %d", ruleID)
	}
	if err != nil {
		return nil, classify(err)
	}

	// Рубежи активности: правило выключено — пауза, проект/объект
	// выключены — заявки некуда ставить.
	if !ruleActive {
		return conflict(ErrRuleNotActive)
	}
	if !projectActive {
		return conflict(ErrProjectNotActive)
	}
	if !objectActive {
		return conflict(ErrObjectNotActive)
	}

	// Срок не наступил — планировщик ждёт (ручной запуск эту
	// проверку обходит: force). Проверка под блокировкой строки —
	// после сдвига срока другим инстансом она не «протухает».
	if !force && nextDue(anchor, period, ref).After(now) {
		return nil, nil
	}

	// Прошлая заявка по правилу ещё открыта: ждём её закрытия — без
	// дублей. Срок при этом не сдвигается: после закрытия следующий
	// тик создаст заявку сразу же.
	var open bool
	if err := tx.QueryRow(ctx,
		`SELECT EXISTS (
             SELECT 1 FROM recurring_runs rr
             JOIN tickets t ON t.id = rr.ticket_id
             WHERE rr.rule_id = $1 AND t.status <> 'closed')`,
		ruleID).Scan(&open); err != nil {
		return nil, classify(err)
	}
	if open {
		return conflict(ErrRuleOpen)
	}

	// Заявка: снимок адреса с объекта, статус new (дефолт колонки).
	var ticketID int64
	err = tx.QueryRow(ctx,
		`INSERT INTO tickets
             (client_id, project_id, object_id, address, floor,
              category, description, priority)
         VALUES ($1, $2, $3, $4, NULLIF($5, ''), $6, $7, $8)
         RETURNING id`,
		run.ClientID, projectID, objectID, run.Address, floor,
		run.Category, run.Description, run.Priority,
	).Scan(&ticketID)
	if err != nil {
		return nil, classify(err)
	}
	run.TicketID = ticketID

	// Журнал: автор — тот, кто завёл правило (fallback — владелец),
	// чтобы «кто создал» читалось в истории заявки.
	actor := createdBy
	if actor == "" {
		actor = run.ClientID
	}
	if _, err := tx.Exec(ctx,
		`INSERT INTO ticket_events (ticket_id, actor_id, action, from_status, to_status)
         VALUES ($1, $2, 'created', NULL, 'new')`,
		ticketID, actor); err != nil {
		return nil, classify(err)
	}

	// История срабатывания и сдвиг срока — в той же транзакции, что и
	// заявка: «заявка есть, а правило думает, что не запускалось»
	// невозможно.
	if _, err := tx.Exec(ctx,
		`INSERT INTO recurring_runs (rule_id, ticket_id, run_at)
         VALUES ($1, $2, $3)`,
		ruleID, ticketID, now); err != nil {
		return nil, classify(err)
	}
	if _, err := tx.Exec(ctx,
		`UPDATE recurring_rules SET last_run_at = $2 WHERE id = $1`,
		ruleID, now); err != nil {
		return nil, classify(err)
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, classify(err)
	}
	return &run, nil
}