Background Job: ActivityCloserJob

Назначение

ActivityCloserJob — фоновый polling-worker модуля sessions/views, который закрывает по неактивности открытые Session и PropertyObjectView: переводит Active → Inactive, считает Duration, ставит EndedAt и публикует session-ended / property-object-view-ended в outbox. Это реализация механизма закрытия ADR-0056 п.3 (simplicity-first: PostgreSQL-only, без Redis/TTL — вместо online-expiry периодический скан + построчная блокировка FOR UPDATE SKIP LOCKED).

Worker решает две задачи в одном прогоне:

1. Протухшие сессии — Active-сессии, у которых now() − LastActivityAt > InactivityThreshold: закрыть сессию и все её ещё открытые просмотры одной транзакцией (INV-SES-005, INV-LV-007).
2. «Висящие» просмотры — Active-PropertyObjectView, чья сессия ещё Active (heartbeat сессии идёт), но по самому просмотру давно нет сигнала (now() − PropertyObjectView.LastActivityAt > InactivityThreshold): закрыть только просмотр, сессию не трогать (INV-LV-004).

Metadata	Значение
Owning team	PIN Engagement
Статус документа	draft
Область действия	контекст user-activity, таблицы user_activity.sessions / user_activity.property_object_views, проект Pin.UserActivity.Background
Assumptions/ограничения	Duration = LastActivityAt − CreatedAt неточна на границе цикла (не учитывает хвост после последнего heartbeat) — приемлемо для истории/агрегатов, точная сессионная аналитика — dev-stats (ADR-0057); целевая нагрузка 1–10k активных ⇒ тысячи открытых строк, партиционирования нет
Зависимости	PostgreSQL (user_activity); Kafka (через outbox); события потребляют PropertyObjectViewEndedConsumer (self, property-object-view-stats.md), dev-stats, crm
Change context	greenfield; заявлен в ../README.md («Background jobs») и domain-доках как будущий worker

Классификация

Поле	Значение
Сервис	user-activity
Тип job	Recurring (фоновый воркер, внутренний таймер)
Trigger	внутренний таймер каждые PollingInterval (~1 мин, конфиг)
Критичность	medium (трекинг вне денежного пути; деградация задерживает историю/агрегаты, не блокирует каталог)

Расписание И Locking

Аспект	Правило
Schedule	каждые PollingInterval (default 1 мин; UTC); лаг закрытия ≤ InactivityThreshold + один polling-цикл
Distributed lock	не требуется — конкурентность даёт построчная блокировка FOR UPDATE SKIP LOCKED: несколько инстансов берут непересекающиеся строки, дубль-закрытия нет (INV-SES-006, INV-LV-008); Redis отсутствует (ADR-0056). Опционально ограничить одним инстансом — PostgreSQL advisory lock Pin:lock:job:scheduled:activity-closer, но по построению он не нужен
Batch size	BatchSize строк на выборку (default 500); прогон повторяет выборку, пока батч полон — большой backlog разгребается за несколько итераций
Max duration	MaxRunDuration (default 5 мин) — предохранитель; при превышении прогон завершается штатно, остаток берёт следующий tick
Cancellation	отмена таймера/хоста учитывается между записями — graceful stop, без обрыва открытой транзакции

Входная Модель ActivityCloserOptions (config)

Внешнего runtime-входа нет; поведение задаётся конфигом (bind + валидация на старте хоста).

Поле	Тип	Обязательность	Источник	Описание
InactivityThreshold	TimeSpan	Да	config (ADR-0056 п.3)	Порог неактивности 10–15 мин (default 12 мин). Общий для сессий и просмотров.
PollingInterval	TimeSpan	Да	config	Период запуска (~1 мин).
BatchSize	int	Да	config	Размер выборки строк за проход (> 0, default 500).
MaxRunDuration	TimeSpan	Да	config	Верхняя граница одного прогона (default 5 мин).

Actor — system (worker); контекст вне tenant (ADR-0052 п.2): скан идёт по всем Active-строкам безотносительно UserId/AnonymousId (tenant/record-level фильтры не применяются) — это задокументированный system-maintenance cross-owner scan, публичного входа у job нет.

Алгоритм

Порог t = InactivityThreshold, отсечка cutoff = now() − t. Порядок фаз обязателен: сначала просмотры, затем сессии (INV-LV-007 — у Inactive-сессии не остаётся Active-просмотров).

1. Контекст. Actor = system, correlationId = jobRunId (Guid v7 на прогон), вне tenant. Идемпотентность обеспечивается фильтром Status = Active + построчной блокировкой (шаг 3), а не ключом.
2. Проверка. Внешних данных нет — проверяется только конфиг при старте (InactivityThreshold ∈ [10 мин, 15 мин], BatchSize > 0) → иначе fail-fast хоста. Существование/статус сущностей проверять нечего: выборка сама отбирает нужное состояние.
3. Что читаем (под построчной блокировкой FOR UPDATE SKIP LOCKED, батч BatchSize).
   • Фаза A — висящие просмотры: Active-просмотры с LastActivityAt < cutoff, по LastActivityAt asc (индекс ix_property_object_views_active_last_activity). Для актуального UserId в payload — сессии этих просмотров батчем (без N+1; см. property-object-view.md «UserId — join в worker-е»).
   • Фаза B — протухшие сессии: Active-сессии с LastActivityAt < cutoff, по LastActivityAt asc (индекс ix_sessions_active_last_activity); затем на каждую — её оставшиеся Active-просмотры (под той же блокировкой). SKIP LOCKED пропускает строки, взятые параллельным инстансом/предыдущим незакоммиченным прогоном.
4. Бизнес-правила и переходы (дом полей — session.md/property-object-view.md, здесь только маппинг):
   • закрытие просмотра: guard Status == Active; Status: Active → Inactive; EndedAt = now(); Duration = LastActivityAt − CreatedAt (check chk_property_object_views_duration_nonneg); фиксируется факт PropertyObjectViewEnded.
   • закрытие сессии: guard Status == Active; Status: Active → Inactive; EndedAt = now(); Duration = LastActivityAt − CreatedAt; предварительно закрываются все оставшиеся Active-просмотры сессии (тем же правилом); фиксируется факт SessionEnded.
5. Что меняется. Закрытые (переведённые в Inactive) property_object_views и sessions. Порядок с учётом FK PropertyObjectView.SessionId → sessions (Restrict): просмотр закрывается раньше/вместе со своей сессией.
6. Какие события фиксируются (через транзакционный outbox тем же согласованным изменением):
   • на каждый закрытый PropertyObjectView — PropertyObjectViewEndedKafkaEvent (ContractVersion = 1, topic pin.user-activity.property-object-view-ended.v1, AggregateType = "PropertyObjectView"); payload/поля — дом property-object-view.md (PropertyObjectViewId, PropertyObjectId, SessionId, AnonymousId, UserId? из join, ViewSource, ViewedAt = CreatedAt, LastActivityAt, DurationSeconds, FilterSnapshot?);
   • на каждую закрытую Session — SessionEndedKafkaEvent (ContractVersion = 1, topic pin.user-activity.session-ended.v1); поля — дом session.md (SessionId, AnonymousId, UserId?, CreatedAt, LastActivityAt, DurationSeconds). Device.IpAddress/UserAgent в payload не включаются.
7. Согласованность. Фаза A — одно согласованное изменение на батч висящих просмотров; фаза B — одно согласованное изменение на сессию (сессия + её просмотры + весь outbox атомарно, INV-SES-005); outbox-publisher доставит события в Kafka после коммита.
8. Результат. Job возвращает не наружу, а метрики прогона (закрыто сессий/просмотров, длительность, остался ли backlog). Пустой прогон — no-op success. Инфраструктурная ошибка на записи — лог + переход к следующей (частичный прогресс закоммичен), остаток берёт следующий tick.

Граница Транзакции

Элемент	Правило
Unit of work	Фаза A — per-batch (висящие просмотры); фаза B — per-session (сессия + её просмотры + outbox)
Сохраняемое состояние	sessions (Status/EndedAt/Duration), property_object_views (то же), записи outbox (session-ended/property-object-view-ended)
External calls	Внутри транзакции внешних вызовов нет; публикация в Kafka — after-commit через outbox-publisher (at-least-once)
Частичная ошибка	continue: сбой на одной сессии/батче логируется и не откатывает уже закоммиченные; строка остаётся Active и повторно берётся следующим tick (SKIP LOCKED снимет lock после rollback)

Идемпотентность И Retry

Сценарий	Правило
Job restart	Checkpoint не нужен — «прогресс» это сам факт Status = Inactive; рестарт заново фильтрует Active по cutoff, уже закрытые не попадают (INV-SES-006, INV-LV-008)
Параллельные инстансы	Безопасны: FOR UPDATE SKIP LOCKED разводит строки; одну сессию/просмотр закрывает ровно один инстанс
Duplicate item	Невозможно в рамках прогона: выбираются только Active; повторный проход по той же строке даёт 0 совпадений
Повторяемая ошибка	Инфраструктурный сбой при фиксации изменений — строка не закрыта, откат снимает lock, следующий tick повторяет; без бесконечного retry внутри прогона
Дубликат события	Payload property-object-view-ended дедуплицирует потребитель по PropertyObjectViewId (INV-LVS-003); outbox гарантирует at-least-once, не exactly-once

Пред/постусловия И Согласование С Онлайн-операциями

Предусловие прогона: есть Active-строки; конфиг валиден (InactivityThreshold ∈ [10 мин, 15 мин], BatchSize > 0). Постусловие: в обработанных батчах не остаётся Active-строк старше cutoff; на каждую закрытую строку в outbox лежит событие; частичный/прерванный прогон оставляет незакрытые строки для следующего tick (прогресс = сам факт Status = Inactive).

Гонка со встречными онлайн-операциями за одну строку (worker держит построчную блокировку выбранной строки, онлайн-команды сериализуются за ним) — бизнес-исход детерминирован:

• Heartbeat vs закрытие сессии/просмотра. Если worker закрыл первым, встречный session-heartbeat видит 0 Active-строк → авто-рестарт (новая сессия, INV-SES-003), а property-object-view-heartbeat возвращает Accepted = false → клиент переоткрывает просмотр (INV-LV-003). Если heartbeat успел первым — LastActivityAt сдвинулся за cutoff, и строка в этот прогон не выбирается. Двойного закрытия и потери активности нет.
• link-session (Switched) vs закрытие. Обе операции закрывают одну сессию под построчной блокировкой: сериализуются, закрывает ровно одна сторона (guard Status == Active), вторая видит уже Inactive и не дублирует переход (INV-SES-006). Если worker закрыл сессию перед link — link не находит Active-сессии устройства → USER_ACTIVITY_SESSION_NOT_FOUND (клиент вызывает start-session).
• Проигравший всегда наблюдает согласованное конечное состояние (Inactive + событие ушло ровно раз); повторный расчёт Duration или второй session-ended/property-object-view-ended на ту же строку исключён.

Наблюдаемость

Тип	Правило
Logs	job started (jobRunId, cutoff); batch processed (фаза, rows, closed); item failed (id, тип, exception); job completed (closedSessions, closedViews, durationMs, backlogRemaining). AnonymousId/IpAddress в логи не пишутся (PII)
Metrics	activity_closer_run_duration_ms (histogram); activity_closer_sessions_closed_total / _views_closed_total (counter); activity_closer_batch_backlog (gauge — осталось после прогона); activity_closer_errors_total; activity_closer_close_lag_seconds (now − LastActivityAt на закрытии — контроль SLO)
Traces	span на прогон с jobRunId, атрибуты phase, batchSize, closedCount; вложенные span на транзакцию сессии
Alerts	«залипший backlog» — Active-строки с LastActivityAt старше 2 × InactivityThreshold (worker не успевает/стоит); activity_closer_errors_total растёт; отсутствие успешного прогона > 5 мин; outbox backlog > 1000 (см. ../README.md «Alerts»)

SLO Точности

Показатель	Цель	Обоснование
Лаг закрытия	≤ InactivityThreshold + PollingInterval (≈ 11–16 мин) для 99% строк	polling-delta ADR-0056; проверяется activity_closer_close_lag_seconds
Отсутствие «вечных» Active	нет Active с LastActivityAt старше 2 × InactivityThreshold	алерт + метрика backlog
Погрешность Duration	недоучёт ≤ PollingInterval (хвост после последнего heartbeat)	документированное упрощение; точная аналитика — dev-stats
Идемпотентность	0 повторных закрытий/дублей доменного перехода при рестартах и N инстансах	фильтр Active + FOR UPDATE SKIP LOCKED
Доставка событий	at-least-once на каждый закрытый агрегат; дедуп у потребителя	outbox + PropertyObjectViewId-идемпотентность консюмера

Связанный Runbook

Сценарий	Runbook
Backlog растёт (worker не успевает)	увеличить BatchSize/частоту, проверить блокировки PostgreSQL и лаг outbox-publisher
Worker стоит (нет прогонов)	проверить hosted-service/pod, конфиг PollingInterval, advisory lock (если включён)
Массовый рост errors_total	проверить доступность PostgreSQL/схемы, откаты транзакций; строки останутся Active и закроются после восстановления

Тесты

Тест	Ожидание
Empty batch	нет Active старше cutoff — no-op success, событий нет
Протухшая сессия с открытыми просмотрами	сессия Inactive, все её просмотры Inactive, Duration посчитан, outbox session-ended + property-object-view-ended на каждый (INV-LV-007)
Висящий просмотр в живой сессии	просмотр Inactive + property-object-view-ended; сессия остаётся Active
Идемпотентность/рестарт	повторный прогон по тем же данным закрывает 0 строк, дублей событий нет
Параллельные инстансы	две копии worker-а не закрывают одну строку дважды (FOR UPDATE SKIP LOCKED)
UserId в payload	property-object-view-ended.UserId = актуальный Session.UserId (сессия связана после открытия просмотра)
Cancellation	отмена прогона между записями — graceful stop, открытая транзакция не рвётся, закоммиченное сохранено
Партиал-фейл	сбой фиксации на одной сессии не откатывает соседние; сбойная остаётся Active

Обратные ссылки

Автоссылки: где используется этот документ.

• Домен (CQRS) (2): Entity / Aggregate: PropertyObjectView, Entity / Aggregate: Session
• API (1): PUT /api/user-activity/property-object-views/{id}/heartbeat — Продление активности просмотра объекта недвижимости
• Use Cases (2): UC-ACT-001. Трекинг сессии гостя и пользователя, UC-ACT-002. Трекинг длительности и числа просмотров объекта недвижимости
• События (2): Event Contract: pin.user-activity.property-object-view-ended (PropertyObjectViewEndedKafkaEvent), Event Contract: pin.user-activity.session-ended (SessionEndedKafkaEvent)
• Консюмеры (4): Consumer: DevStatsPropertyObjectViewEndedConsumer, Consumer: DevStatsSessionEndedConsumer, Consumer: Identity Sign-Ins (UserActivitySignedInConsumer), Consumer: User-Activity Popularity (pin-catalog-popularity)
• Фоновые задачи (4): Background Job: AccessFanOutJob — веерная рассылка UserAccessChangedKafkaEvent при смене состава прав роли, Background Job: ActivityRetentionJob, Background Job: CallTimeoutReaperJob, Background Job: SnapshotDetachRetryJob

Связанные Документы

• Дом полей/инвариантов: ../domain/session.md (INV-SES-005/006, Session.Close), ../domain/property-object-view.md (INV-LV-004/007/008, payload property-object-view-ended), ../domain/property-object-view-stats.md (потребитель PropertyObjectViewEndedConsumer).
• Обзор/эксплуатация: ../README.md («Background jobs», «Alerts»); соседний worker — activity-retention (очистка Inactive старше 12 мес).
• События (будущие контракты): ../events/session-ended.md, ../events/property-object-view-ended.md.
• Решения: ADR-0052 (вне tenant), ADR-0054 (модули sessions/views), ADR-0056 п.3 (polling-закрытие, PostgreSQL-only), ADR-0057 (dev-stats — точная аналитика).