diff --git a/api/openapi.yaml b/api/openapi.yaml index b538654..6e768a0 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -660,6 +660,271 @@ paths: '204': { description: Подписка удалена } '401': { description: Требуется авторизация } '403': { description: Недостаточно прав } + /sync/devices: + post: + summary: Регистрация нового устройства синхронизации + operationId: registerSyncDevice + tags: [Sync] + security: [{ bearerAuth: [] }] + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [device_code, name] + properties: + device_code: { type: string, minLength: 1, maxLength: 64 } + name: { type: string, minLength: 1, maxLength: 255 } + org_unit_id: { $ref: '#/components/schemas/Ulid' } + allowed_scopes: + type: array + items: { type: string } + responses: + '201': + description: Устройство зарегистрировано, выдан токен + content: + application/json: + schema: + type: object + required: [device, token] + properties: + device: + type: object + required: [id, tenant_id, device_code, name, token_hint, status, allowed_scopes, registered_at, created_by] + properties: + id: { $ref: '#/components/schemas/Ulid' } + tenant_id: { $ref: '#/components/schemas/Ulid' } + device_code: { type: string } + name: { type: string } + token_hint: { type: string } + status: { type: string, enum: [active, revoked, pending] } + allowed_scopes: { type: array, items: { type: string } } + org_unit_id: + oneOf: + - { $ref: '#/components/schemas/Ulid' } + - { type: 'null' } + registered_at: { type: string, format: date-time } + created_by: { $ref: '#/components/schemas/Ulid' } + token: { type: string } + '400': { $ref: '#/components/responses/BadRequest' } + '401': { description: Требуется авторизация руководителя } + '403': { description: Недостаточно прав } + /sync/devices/{deviceId}/revoke: + post: + summary: Отозвать токен и доступ устройства + operationId: revokeSyncDevice + tags: [Sync] + security: [{ bearerAuth: [] }] + parameters: + - in: path + name: deviceId + required: true + schema: { $ref: '#/components/schemas/Ulid' } + responses: + '200': + description: Устройство отозвано + content: + application/json: + schema: + type: object + required: [id, status, revoked_at] + properties: + id: { $ref: '#/components/schemas/Ulid' } + status: { type: string, enum: [revoked] } + revoked_at: { type: string, format: date-time } + '401': { description: Требуется авторизация руководителя } + '403': { description: Недостаточно прав } + '404': { description: Устройство не найдено } + /sync/status: + get: + summary: Получить статус сервера синхронизации и чекпоинта устройства + operationId: getSyncStatus + tags: [Sync] + security: [{ bearerAuth: [] }] + responses: + '200': + description: Текущее состояние синхронизации + content: + application/json: + schema: + type: object + required: [server_time_utc, latest_cursor, schema_version] + properties: + server_time_utc: { type: string, format: date-time } + latest_cursor: { type: string } + device_checkpoint: + oneOf: + - type: object + properties: + last_acked_cursor: + oneOf: + - { type: string } + - { type: 'null' } + acknowledged_at: + oneOf: + - { type: string, format: date-time } + - { type: 'null' } + - { type: 'null' } + schema_version: { type: integer } + '401': { description: Недействительный токен устройства } + '403': { description: Устройство отозвано или заблокировано } + /sync/pull: + post: + summary: Получить изменения сущностей после заданного курсора или начальный снимок + operationId: pullSyncChanges + tags: [Sync] + security: [{ bearerAuth: [] }] + requestBody: + required: false + content: + application/json: + schema: + type: object + properties: + cursor: + oneOf: + - { type: string } + - { type: 'null' } + limit: { type: integer, minimum: 1, maximum: 100, default: 100 } + entity_types: + type: array + items: { type: string } + responses: + '200': + description: Пакет изменений или начальный снимок + content: + application/json: + schema: + type: object + required: [changes, next_cursor, has_more] + properties: + changes: + type: array + items: + type: object + required: [cursor, entity_type, entity_id, operation, revision, payload, created_at] + properties: + cursor: { type: string } + entity_type: { type: string } + entity_id: { type: string } + operation: { type: string, enum: [upsert, delete] } + revision: { type: integer } + payload: { type: object } + created_at: { type: string, format: date-time } + next_cursor: { type: string } + has_more: { type: boolean } + '400': { $ref: '#/components/responses/BadRequest' } + '401': { description: Недействительный токен устройства } + '403': { description: Устройство отозвано или заблокировано } + /sync/push: + post: + summary: Пакетная отправка команд от локального устройства + operationId: pushSyncCommands + tags: [Sync] + security: [{ bearerAuth: [] }] + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [commands] + properties: + commands: + type: array + minItems: 1 + maxItems: 100 + items: + type: object + required: [command_type, schema_version, idempotency_key, correlation_id, created_at, payload] + properties: + command_type: { type: string } + schema_version: { type: integer } + idempotency_key: { type: string, minLength: 1, maxLength: 128 } + correlation_id: { type: string } + causation_id: + oneOf: + - { type: string } + - { type: 'null' } + created_at: { type: string, format: date-time } + payload: { type: object } + responses: + '200': + description: Результаты применения команд + content: + application/json: + schema: + type: object + required: [results] + properties: + results: + type: array + items: + type: object + required: [idempotency_key, command_type, status] + properties: + idempotency_key: { type: string } + command_type: { type: string } + status: { type: string, enum: [applied, rejected, conflict] } + result: + oneOf: + - { type: object } + - { type: 'null' } + conflict: + oneOf: + - type: object + required: [conflict_code, entity_type, entity_id, expected_revision, actual_revision, safe_description, client_action] + properties: + conflict_code: { type: string, enum: [REVISION_MISMATCH] } + entity_type: { type: string } + entity_id: { type: string } + expected_revision: { type: integer } + actual_revision: { type: integer } + safe_description: { type: string } + client_action: { type: string, enum: [FETCH_LATEST_AND_RETRY] } + - { type: 'null' } + error_code: + oneOf: + - { type: string } + - { type: 'null' } + error_message: + oneOf: + - { type: string } + - { type: 'null' } + '400': { $ref: '#/components/responses/BadRequest' } + '401': { description: Недействительный токен устройства } + '403': { description: Устройство отозвано или заблокировано } + /sync/ack: + post: + summary: Подтвердить получение и обработку изменений до курсора + operationId: ackSyncCheckpoint + tags: [Sync] + security: [{ bearerAuth: [] }] + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [cursor] + properties: + cursor: { type: string, minLength: 5 } + responses: + '200': + description: Чекпоинт зафиксирован + content: + application/json: + schema: + type: object + required: [acknowledged, cursor, acknowledged_at] + properties: + acknowledged: { type: boolean } + cursor: { type: string } + acknowledged_at: { type: string, format: date-time } + '400': { $ref: '#/components/responses/BadRequest' } + '401': { description: Недействительный токен устройства } + '403': { description: Устройство отозвано или заблокировано } components: securitySchemes: bearerAuth: diff --git a/docs/OFFLINE-SYNC-PROTOCOL.md b/docs/OFFLINE-SYNC-PROTOCOL.md new file mode 100644 index 0000000..bbd474c --- /dev/null +++ b/docs/OFFLINE-SYNC-PROTOCOL.md @@ -0,0 +1,255 @@ +# Протокол безопасной офлайн-синхронизации Krona Staff + +## 1. Архитектурный контекст и инварианты + +Krona Staff является модулем платформы Krona Business Platform, а не изолированной системой пользователей, прав и ролей: +1. **Единственный источник истины (Single Source of Truth)**: Центральный сервер Krona Staff и Platform Core являются владельцами эталонных данных. Локальные терминалы, планшеты и клиенты являются исполнительными терминалами и не могут становиться вторым источником истины. +2. **User != Employee**: Учётная запись пользователя платформы (`User`) отделена от карточки сотрудника (`Employee`). +3. **Строгая изоляция арендаторов (Tenant Isolation)**: Все сущности привязаны к `tenant_id` (ULID). Новые таблицы включают и форсируют PostgreSQL Row Level Security (`FORCE ROW LEVEL SECURITY`), гарантируя невозможность утечки данных между организациями даже на уровне прямого SQL. +4. **Безопасность логов**: Секретные токены, хэши паролей, биометрия и персональные данные (ПДН) никогда не попадают в серверные логи. + +--- + +## 2. Жизненный цикл устройства и аутентификация + +### 2.1. Регистрация устройства (`POST /api/v1/staff/sync/devices`) +Регистрация выполняется администратором или руководителем предприятия (требуется сессия руководителя или токен с правом `staff:sync:manage` / `staff:attendance:manage`). + +**Запрос:** +```json +{ + "device_code": "TERM-MAIN-01", + "name": "Главный вход — Терминал", + "org_unit_id": "01J00000000000000000000001", + "allowed_scopes": ["attendance:sync"] +} +``` + +**Ответ (201 Created):** +```json +{ + "device": { + "id": "01J9X1Z2A3B4C5D6E7F8G9H0J1", + "tenant_id": "01J00000000000000000000000", + "device_code": "TERM-MAIN-01", + "name": "Главный вход — Терминал", + "token_hint": "kstaff_dev_01J00...", + "status": "active", + "allowed_scopes": ["attendance:sync"], + "org_unit_id": "01J00000000000000000000001", + "registered_at": "2026-09-26T12:00:00.000Z", + "created_by": "01J00000000000000000000099" + }, + "token": "kstaff_dev_01J00000000000000000000000_3f2a1b9c8d7e6f..." +} +``` + +> [!IMPORTANT] +> Сырой токен `token` возвращается клиенту ровно один раз при регистрации. В базе данных (`staff_sync_devices`) сохраняется исключительно криптографический хэш SHA-256 (`token_hash`) и безопасная маска (`token_hint`). + +### 2.2. Аутентификация запросов устройства +Каждый запрос от локального терминала к API синхронизации должен содержать заголовок: +```http +Authorization: Bearer +``` +Сервер: +1. Вычисляет `SHA-256(device_token)`. +2. Извлекает `tenant_id` из префикса токена или заголовка `X-Tenant-ID`. +3. Открывает транзакцию с RLS контекстом `database.tenantTransaction(tenantId, actorId, ...)`. +4. Находит устройство по `token_hash` в таблице `staff_sync_devices`. +5. Проверяет `status = 'active'` и наличие скоупа `attendance:sync`. +6. Асинхронно обновляет `last_seen_at`. + +### 2.3. Отзыв устройства (`POST /api/v1/staff/sync/devices/:deviceId/revoke`) +При компрометации устройства, утере или списании администратор выполняет отзыв. Статус устройства в БД переводится в `revoked`, фиксируется `revoked_at = now()`. + +Все последующие запросы с токеном отозванного устройства немедленно отклоняются со статусом **403 Forbidden** и кодом ошибки `DEVICE_REVOKED`: +```json +{ + "statusCode": 403, + "error": "Forbidden", + "message": "Устройство отозвано" +} +``` + +--- + +## 3. Монотонный непрозрачный курсор изменений + +### 3.1. Формат курсора +Курсор представляет собой непрозрачную строку вида: +```text +cur__ +``` +Например: `cur_105_7a9f8b2c1d`. +- ``: Монотонный идентификатор последовательности изменений в таблице `staff_sync_change_log` (`cursor_seq BIGSERIAL`). +- ``: Криптографическая контрольная подпись от арендатора, последовательности и серверной соли (`SHA-256(tenant_id + seq + salt)`). + +Свойства курсора: +- **Монотонность**: Гарантирует строго возрастающий порядок чтения журнала изменений. +- **Независимость от времени**: Не зависит от системных часов терминала и защищён от сдвигов локального времени. +- **Целостность и изоляция**: Попытка передать повреждённый курсор или курсор другого арендатора немедленно отклоняется с ошибкой `INVALID_CURSOR` (400 Bad Request). + +### 3.2. Начальная загрузка (Bootstrap, `cursor = null`) +Когда терминал настраивается впервые или выполняет полный сброс локальной базы данных, он отправляет `POST /api/v1/staff/sync/pull` с пустым или отсутствующим курсором (`cursor: null`): +1. Сервер формирует снимок текущих **актуальных** данных для поддерживаемых сущностей (`org_unit`, `profession`, `employee`, `work_session`, `timesheet_entry`, `planned_shift`). +2. Для каждой сущности формируется безопасная полезная нагрузка с операцией `upsert`. +3. В ответе возвращается `next_cursor`, соответствующий текущему максимальному `cursor_seq` журнала изменений сервера. +4. После применения начального снимка терминал фиксирует `next_cursor` и переходит в режим инкрементального получения дельт. + +### 3.3. Инкрементальная загрузка (Pull, `POST /api/v1/staff/sync/pull`) +Запрос: +```json +{ + "cursor": "cur_105_7a9f8b2c1d", + "limit": 50, + "entity_types": ["employee", "work_session"] +} +``` +Ответ: +```json +{ + "changes": [ + { + "cursor": "cur_106_3c4d5e6f7a", + "entity_type": "employee", + "entity_id": "01J9X1Z2A3B4C5D6E7F8G9H0J2", + "operation": "upsert", + "revision": 2, + "payload": { + "id": "01J9X1Z2A3B4C5D6E7F8G9H0J2", + "personnel_number": "EMP-042", + "display_name": "Иванов Иван Иванович", + "status": "active", + "profession_id": "01J00000000000000000000010", + "position": "Повар-универсал" + }, + "created_at": "2026-09-26T12:05:00.000Z" + } + ], + "next_cursor": "cur_106_3c4d5e6f7a", + "has_more": false +} +``` + +### 3.4. Подтверждение изменений (Ack, `POST /api/v1/staff/sync/ack`) +После успешного локального сохранения изменений устройство подтверждает обработанный курсор: +```json +{ + "cursor": "cur_106_3c4d5e6f7a" +} +``` +Сервер сохраняет контрольную точку в таблице `staff_sync_device_checkpoints`. Это позволяет отслеживать отставание устройств и безопасно выполнять очистку журнала изменений. + +--- + +## 4. Пакетная отправка команд (Push) и идемпотентность + +### 4.1. Контракт отправки (`POST /api/v1/staff/sync/push`) +Клиент отправляет накопленные в офлайне команды пачками. +- Ограничение пакета: от 1 до 100 команд (`BATCH_SIZE_EXCEEDED` при превышении). +- Максимальный размер полезной нагрузки: не более 64 КБ на команду (`PAYLOAD_TOO_LARGE`). +- Максимальный размер HTTP-запроса: 2 МБ. +- Временное окно защиты от Replay: `created_at` команды должно быть не старше 30 дней и не более чем на 5 минут в будущем (`TIMESTAMP_OUT_OF_WINDOW`). + +**Пример запроса:** +```json +{ + "commands": [ + { + "command_type": "attendance_event", + "schema_version": 1, + "idempotency_key": "cmd-term01-778899", + "correlation_id": "01J9X1Z2A3B4C5D6E7F8G9H0AA", + "created_at": "2026-09-26T12:10:00.000Z", + "payload": { + "employee_id": "01J9X1Z2A3B4C5D6E7F8G9H0J2", + "event_type": "check_in", + "occurred_at": "2026-09-26T12:09:58.000Z", + "source_event_id": "scan-123456" + } + } + ] +} +``` + +### 4.2. Идемпотентность (`staff_sync_command_inbox`) +Таблица `staff_sync_command_inbox` содержит уникальный ключ: +```sql +UNIQUE (tenant_id, device_id, idempotency_key) +``` +При повторной отправке команды с тем же `idempotency_key`: +1. Мутация повторно не исполняется. +2. Сервер извлекает ранее зафиксированный результат (`status`, `result`, `conflict`, `error_code`, `error_message`) и возвращает его клиенту. +3. Это гарантирует полную безопасность при сбоях связи и автоматических повторах (Retry). + +--- + +## 5. Обнаружение конфликтов + +Для сущностей с контролем версий (`work_session`, `timesheet_entry`) используется оптимистическая блокировка через поле `revision` (целое число > 0). + +Если терминал пытается применить мутацию на основе устаревшей ревизии (например, смена уже была подтверждена управляющим на сервере), команда **не перезаписывает** данные. + +**Ответ с конфликтом (в составе `results`):** +```json +{ + "idempotency_key": "cmd-term01-778900", + "command_type": "upsert_work_session", + "status": "conflict", + "conflict": { + "conflict_code": "REVISION_MISMATCH", + "entity_type": "work_session", + "entity_id": "01J9X1Z2A3B4C5D6E7F8G9H0WS", + "expected_revision": 2, + "actual_revision": 3, + "safe_description": "Ревизия рабочей смены на сервере изменилась", + "client_action": "FETCH_LATEST_AND_RETRY" + } +} +``` + +Клиент, получив конфликт, обязан: +1. Запросить актуальное состояние сущности через `POST /sync/pull`. +2. Обновить локальное представление. +3. При необходимости повторить действие с актуальной ревизией или уведомить оператора. + +--- + +## 6. Минимальный безопасный набор данных + +Терминал получает **исключительно оперативные данные**, необходимые для фиксации посещаемости и графиков: + +| Сущность | Разрешённые поля | Запрещённые данные | +| :--- | :--- | :--- | +| `org_unit` | `id`, `code`, `name`, `status` | Финансовые показатели, выручка, арендные договоры | +| `profession` | `id`, `code`, `name`, `status` | Тарифные сетки, оклады | +| `employee` | `id`, `personnel_number`, `display_name`, `status`, `profession_id`, `position` | **Зарплаты, ставки, паспортные данные, СНИЛС, телефоны, адреса, биометрия** | +| `work_session` | `id`, `employee_id`, `org_unit_id`, `business_date`, `started_at`, `ended_at`, `worked_minutes`, `status`, `revision` | Стоимость смены, начисления, штрафы | +| `timesheet_entry` | `id`, `employee_id`, `work_date`, `minutes_worked`, `status`, `revision` | Расчётные листы, итоговые выплаты | +| `planned_shift` | `id`, `employee_id`, `org_unit_id`, `business_date`, `planned_minutes`, `start_time`, `end_time` | Плановая стоимость ФОТ смены | +| `attendance_event` | `id`, `employee_id`, `event_type`, `occurred_at`, `source_event_id`, `device_id`, `org_unit_id`, `status` | Сырые биометрические сэмплы сканера | + +> [!CAUTION] +> **Финансовые данные и заработная плата НИКОГДА не передаются на локальный терминал!** +> Это исключает риск утечки конфиденциальной финансовой информации при физическом хищении терминала или компрометации его локального хранилища. + +--- + +## 7. Безопасность биометрии + +1. **Запрет передачи через Sync API**: Биометрические шаблоны (отпечатки пальцев, векторы лиц) строго запрещено передавать через интерфейсы синхронизации Krona Staff. +2. **Локальная верификация**: Если терминал оснащён биометрическим датчиком (например, ZKTeco SLK20R), он выполняет сопоставление биометрии локально в изолированном модуле или безопасном анклаве устройства. +3. **Фиксация только факта**: Сервер получает только сформированное событие `attendance_event` с проверенным `employee_id`, типом отметки (`check_in`/`check_out`) и временем фиксации. + +--- + +## 8. Политика хранения данных (Retention) + +Для предотвращения бесконечного роста таблиц синхронизации предусмотрена функция очистки: +```sql +SELECT * FROM staff_sync_cleanup_retention(30); +``` +- Удаляет записи `staff_sync_change_log` старше заданного количества дней (по умолчанию 30). +- Удаляет обработанные команды `staff_sync_command_inbox` старше заданного количества дней. +- Активные устройства, отстающие более чем на окно хранения (30 дней), при следующем обращении с устаревшим курсором получают требование повторной начальной инициализации (Bootstrap). diff --git a/migrations/0011_offline_sync_backend.down.sql b/migrations/0011_offline_sync_backend.down.sql new file mode 100644 index 0000000..c705e90 --- /dev/null +++ b/migrations/0011_offline_sync_backend.down.sql @@ -0,0 +1,5 @@ +DROP FUNCTION IF EXISTS staff_sync_cleanup_retention(INTEGER); +DROP TABLE IF EXISTS staff_sync_device_checkpoints; +DROP TABLE IF EXISTS staff_sync_command_inbox; +DROP TABLE IF EXISTS staff_sync_change_log; +DROP TABLE IF EXISTS staff_sync_devices; diff --git a/migrations/0011_offline_sync_backend.up.sql b/migrations/0011_offline_sync_backend.up.sql new file mode 100644 index 0000000..6426b25 --- /dev/null +++ b/migrations/0011_offline_sync_backend.up.sql @@ -0,0 +1,141 @@ +-- Offline synchronization backend foundation for Krona Staff local terminals and clients. +-- Protected by PostgreSQL Row Level Security (RLS) and strict tenant isolation. + +CREATE TABLE staff_sync_devices ( + id VARCHAR(26) PRIMARY KEY CHECK (id ~ '^[0-9A-HJKMNP-TV-Z]{26}$'), + tenant_id VARCHAR(26) NOT NULL CHECK (tenant_id ~ '^[0-9A-HJKMNP-TV-Z]{26}$'), + device_code VARCHAR(64) NOT NULL, + name VARCHAR(255) NOT NULL, + token_hash VARCHAR(64) NOT NULL CHECK (token_hash ~ '^[0-9a-f]{64}$'), + token_hint VARCHAR(16) NOT NULL, + status VARCHAR(20) NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'revoked', 'pending')), + allowed_scopes TEXT[] NOT NULL DEFAULT ARRAY['attendance:sync'], + org_unit_id VARCHAR(26), + registered_at TIMESTAMPTZ NOT NULL DEFAULT now(), + last_seen_at TIMESTAMPTZ, + revoked_at TIMESTAMPTZ, + created_by VARCHAR(26) NOT NULL CHECK (created_by ~ '^[0-9A-HJKMNP-TV-Z]{26}$'), + UNIQUE (tenant_id, id), + UNIQUE (tenant_id, device_code), + UNIQUE (tenant_id, token_hash), + FOREIGN KEY (tenant_id, org_unit_id) REFERENCES staff_org_units (tenant_id, id) ON DELETE SET NULL +); + +CREATE INDEX staff_sync_devices_token_hash_idx + ON staff_sync_devices (tenant_id, token_hash); +CREATE INDEX staff_sync_devices_status_idx + ON staff_sync_devices (tenant_id, status); + +ALTER TABLE staff_sync_devices ENABLE ROW LEVEL SECURITY; +ALTER TABLE staff_sync_devices FORCE ROW LEVEL SECURITY; +CREATE POLICY staff_sync_devices_tenant_isolation + ON staff_sync_devices + USING (tenant_id = current_setting('app.tenant_id', true)) + WITH CHECK (tenant_id = current_setting('app.tenant_id', true)); + +CREATE TABLE staff_sync_change_log ( + id VARCHAR(26) PRIMARY KEY CHECK (id ~ '^[0-9A-HJKMNP-TV-Z]{26}$'), + tenant_id VARCHAR(26) NOT NULL CHECK (tenant_id ~ '^[0-9A-HJKMNP-TV-Z]{26}$'), + cursor_seq BIGSERIAL NOT NULL, + entity_type VARCHAR(64) NOT NULL, + entity_id VARCHAR(26) NOT NULL, + operation VARCHAR(20) NOT NULL CHECK (operation IN ('upsert', 'delete')), + revision INTEGER NOT NULL DEFAULT 1 CHECK (revision > 0), + payload JSONB NOT NULL DEFAULT '{}'::jsonb, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (tenant_id, id) +); + +CREATE INDEX staff_sync_change_log_cursor_idx + ON staff_sync_change_log (tenant_id, cursor_seq); +CREATE INDEX staff_sync_change_log_entity_idx + ON staff_sync_change_log (tenant_id, entity_type, entity_id); + +ALTER TABLE staff_sync_change_log ENABLE ROW LEVEL SECURITY; +ALTER TABLE staff_sync_change_log FORCE ROW LEVEL SECURITY; +CREATE POLICY staff_sync_change_log_tenant_isolation + ON staff_sync_change_log + USING (tenant_id = current_setting('app.tenant_id', true)) + WITH CHECK (tenant_id = current_setting('app.tenant_id', true)); + +CREATE TABLE staff_sync_command_inbox ( + id VARCHAR(26) PRIMARY KEY CHECK (id ~ '^[0-9A-HJKMNP-TV-Z]{26}$') , + tenant_id VARCHAR(26) NOT NULL CHECK (tenant_id ~ '^[0-9A-HJKMNP-TV-Z]{26}$'), + device_id VARCHAR(26) NOT NULL, + command_type VARCHAR(64) NOT NULL, + schema_version INTEGER NOT NULL DEFAULT 1, + idempotency_key VARCHAR(128) NOT NULL, + correlation_id VARCHAR(36) NOT NULL, + causation_id VARCHAR(36), + payload JSONB NOT NULL DEFAULT '{}'::jsonb, + status VARCHAR(20) NOT NULL CHECK (status IN ('applied', 'rejected', 'conflict')), + result JSONB, + conflict JSONB, + error_code VARCHAR(64), + error_message TEXT, + created_at TIMESTAMPTZ NOT NULL, + processed_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (tenant_id, id), + UNIQUE (tenant_id, device_id, idempotency_key), + FOREIGN KEY (tenant_id, device_id) REFERENCES staff_sync_devices (tenant_id, id) ON DELETE CASCADE +); + +CREATE INDEX staff_sync_command_inbox_device_idx + ON staff_sync_command_inbox (tenant_id, device_id, processed_at DESC); + +ALTER TABLE staff_sync_command_inbox ENABLE ROW LEVEL SECURITY; +ALTER TABLE staff_sync_command_inbox FORCE ROW LEVEL SECURITY; +CREATE POLICY staff_sync_command_inbox_tenant_isolation + ON staff_sync_command_inbox + USING (tenant_id = current_setting('app.tenant_id', true)) + WITH CHECK (tenant_id = current_setting('app.tenant_id', true)); + +CREATE TABLE staff_sync_device_checkpoints ( + device_id VARCHAR(26) NOT NULL, + tenant_id VARCHAR(26) NOT NULL CHECK (tenant_id ~ '^[0-9A-HJKMNP-TV-Z]{26}$'), + last_acked_cursor_seq BIGINT NOT NULL DEFAULT 0, + last_acked_cursor TEXT, + acknowledged_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (tenant_id, device_id), + FOREIGN KEY (tenant_id, device_id) REFERENCES staff_sync_devices (tenant_id, id) ON DELETE CASCADE +); + +ALTER TABLE staff_sync_device_checkpoints ENABLE ROW LEVEL SECURITY; +ALTER TABLE staff_sync_device_checkpoints FORCE ROW LEVEL SECURITY; +CREATE POLICY staff_sync_device_checkpoints_tenant_isolation + ON staff_sync_device_checkpoints + USING (tenant_id = current_setting('app.tenant_id', true)) + WITH CHECK (tenant_id = current_setting('app.tenant_id', true)); + +CREATE OR REPLACE FUNCTION staff_sync_cleanup_retention(p_days INTEGER DEFAULT 30) +RETURNS TABLE(purged_change_logs BIGINT, purged_commands BIGINT) +LANGUAGE plpgsql +AS $$ +DECLARE + v_change_logs BIGINT; + v_commands BIGINT; +BEGIN + WITH deleted_changes AS ( + DELETE FROM staff_sync_change_log + WHERE created_at < now() - (p_days || ' days')::INTERVAL + RETURNING 1 + ) + SELECT count(*) INTO v_change_logs FROM deleted_changes; + + WITH deleted_cmds AS ( + DELETE FROM staff_sync_command_inbox + WHERE processed_at < now() - (p_days || ' days')::INTERVAL + RETURNING 1 + ) + SELECT count(*) INTO v_commands FROM deleted_cmds; + + RETURN QUERY SELECT v_change_logs, v_commands; +END; +$$; + +GRANT SELECT, INSERT, UPDATE, DELETE ON staff_sync_devices TO krona_staff_app; +GRANT SELECT, INSERT, UPDATE, DELETE ON staff_sync_change_log TO krona_staff_app; +GRANT USAGE, SELECT ON SEQUENCE staff_sync_change_log_cursor_seq_seq TO krona_staff_app; +GRANT SELECT, INSERT, UPDATE, DELETE ON staff_sync_command_inbox TO krona_staff_app; +GRANT SELECT, INSERT, UPDATE, DELETE ON staff_sync_device_checkpoints TO krona_staff_app; +GRANT EXECUTE ON FUNCTION staff_sync_cleanup_retention(INTEGER) TO krona_staff_app; diff --git a/module.json b/module.json index ee2e0c4..6cdfc59 100644 --- a/module.json +++ b/module.json @@ -30,7 +30,8 @@ "staff:payroll:calculate", "staff:payroll:approve", "staff:payout:write", - "staff:payslip:read_own" + "staff:payslip:read_own", + "staff:sync:manage" ], "entities": [ "Employee", @@ -42,7 +43,8 @@ "PayRule", "WorkSession", "PayrollRun", - "Payout" + "Payout", + "SyncDevice" ], "events_emitted": [ "attendance.clock.in", diff --git a/src/app.ts b/src/app.ts index 5063848..22eb4de 100644 --- a/src/app.ts +++ b/src/app.ts @@ -8,6 +8,7 @@ import { Database } from './infrastructure/database.js'; import { newUlid } from './core/ids.js'; import { SERVICE_NAME, SERVICE_VERSION } from './version.js'; import { registerPortalRoutes } from './portal.js'; +import { registerSyncRoutes } from './sync.js'; import { portalCssEnhanced as portalCss, portalHtml, @@ -130,6 +131,7 @@ export async function buildApp({ ); api.get('/openapi.json', async () => app.swagger()); await registerPortalRoutes(api, config, database); + await registerSyncRoutes(api, config, database); }, { prefix: apiPrefix }, ); diff --git a/src/domain/sync/cursor.ts b/src/domain/sync/cursor.ts new file mode 100644 index 0000000..1d983e8 --- /dev/null +++ b/src/domain/sync/cursor.ts @@ -0,0 +1,51 @@ +import crypto from 'node:crypto'; + +export class InvalidCursorError extends Error { + readonly code = 'INVALID_CURSOR'; + constructor(message = 'Курсор синхронизации недействителен или повреждён') { + super(message); + this.name = 'InvalidCursorError'; + } +} + +const CURSOR_PREFIX = 'cur_'; +const CURSOR_SALT = 'krona_staff_sync_cursor_salt_v1'; + +export function encodeCursor(tenantId: string, seq: bigint | number | string): string { + const seqStr = String(seq); + const hash = crypto + .createHash('sha256') + .update(`${tenantId}:${seqStr}:${CURSOR_SALT}`) + .digest('hex') + .slice(0, 12); + return `${CURSOR_PREFIX}${seqStr}_${hash}`; +} + +export function decodeCursor(tenantId: string, cursor: string): bigint { + if (typeof cursor !== 'string' || !cursor) { + throw new InvalidCursorError('Курсор синхронизации не задан'); + } + const match = cursor.match(/^cur_(\d+)_([0-9a-f]{12})$/); + if (!match) { + throw new InvalidCursorError('Неверный формат курсора синхронизации'); + } + const seqStr = match[1]!; + const candidateHash = match[2]!; + const expectedHash = crypto + .createHash('sha256') + .update(`${tenantId}:${seqStr}:${CURSOR_SALT}`) + .digest('hex') + .slice(0, 12); + + const candidateBuffer = Buffer.from(candidateHash, 'utf8'); + const expectedBuffer = Buffer.from(expectedHash, 'utf8'); + + if ( + candidateBuffer.length !== expectedBuffer.length || + !crypto.timingSafeEqual(candidateBuffer, expectedBuffer) + ) { + throw new InvalidCursorError('Подпись курсора синхронизации недействительна для данного арендатора'); + } + + return BigInt(seqStr); +} diff --git a/src/domain/sync/service.ts b/src/domain/sync/service.ts new file mode 100644 index 0000000..ab4f20c --- /dev/null +++ b/src/domain/sync/service.ts @@ -0,0 +1,1068 @@ +import crypto from 'node:crypto'; +import type { PoolClient } from 'pg'; +import { newUlid, parseUlid, type Ulid } from '../../core/ids.js'; +import type { Database } from '../../infrastructure/database.js'; +import { decodeCursor, encodeCursor } from './cursor.js'; +import type { + SafeAttendanceEventPayload, + SafeEmployeePayload, + SafeOrgUnitPayload, + SafePlannedShiftPayload, + SafeProfessionPayload, + SafeTimesheetEntryPayload, + SafeWorkSessionPayload, + SyncAckResult, + SyncChange, + SyncCommand, + SyncCommandResult, + SyncConflict, + SyncDevice, + SyncDeviceRegistrationResult, + SyncDeviceStatus, + SyncOperation, + SyncPullResult, + SyncStatus, +} from './types.js'; + +export class SyncAuthError extends Error { + readonly code: string; + readonly statusCode: number; + constructor(message: string, code = 'DEVICE_UNAUTHORIZED', statusCode = 401) { + super(message); + this.name = 'SyncAuthError'; + this.code = code; + this.statusCode = statusCode; + } +} + +export class SyncValidationError extends Error { + readonly code: string; + readonly statusCode: number; + constructor(message: string, code = 'SYNC_VALIDATION_ERROR', statusCode = 400) { + super(message); + this.name = 'SyncValidationError'; + this.code = code; + this.statusCode = statusCode; + } +} + +export interface RegisterDeviceParams { + device_code: string; + name: string; + org_unit_id?: string | null; + allowed_scopes?: string[]; +} + +export interface PullOptions { + cursor?: string | null; + limit?: number; + entity_types?: string[]; +} + +export class SyncService { + constructor( + private readonly database: Database, + private readonly systemActorId: Ulid, + ) {} + + static hashToken(token: string): string { + return crypto.createHash('sha256').update(token, 'utf8').digest('hex'); + } + + static parseTenantFromToken(token: string): Ulid | null { + const match = token.match(/^kstaff_dev_([0-9A-HJKMNP-TV-Z]{26})_[0-9a-f]{32,64}$/); + if (!match) return null; + try { + return parseUlid(match[1]!); + } catch { + return null; + } + } + + async authenticateDevice( + rawToken: string, + providedTenantId?: Ulid, + requiredScope = 'attendance:sync', + ): Promise { + if (!rawToken || typeof rawToken !== 'string') { + throw new SyncAuthError('Требуется токен авторизации устройства', 'DEVICE_UNAUTHORIZED', 401); + } + + const tokenTenant = SyncService.parseTenantFromToken(rawToken); + if (providedTenantId && tokenTenant && providedTenantId !== tokenTenant) { + throw new SyncAuthError('Несоответствие арендатора в токене устройства', 'TENANT_MISMATCH', 403); + } + + const tenantId = providedTenantId ?? tokenTenant; + if (!tenantId) { + throw new SyncAuthError( + 'Не удалось определить арендатора для устройства', + 'DEVICE_UNAUTHORIZED', + 401, + ); + } + + const tokenHash = SyncService.hashToken(rawToken); + + return this.database.tenantTransaction(tenantId, this.systemActorId, async (client) => { + const res = await client.query<{ + id: string; + tenant_id: string; + device_code: string; + name: string; + token_hash: string; + token_hint: string; + status: SyncDeviceStatus; + allowed_scopes: string[]; + org_unit_id: string | null; + registered_at: Date; + last_seen_at: Date | null; + revoked_at: Date | null; + created_by: string; + }>( + `SELECT id, tenant_id, device_code, name, token_hash, token_hint, status, + allowed_scopes, org_unit_id, registered_at, last_seen_at, revoked_at, created_by + FROM staff_sync_devices + WHERE tenant_id = $1 AND token_hash = $2`, + [tenantId, tokenHash], + ); + + const row = res.rows[0]; + if (!row) { + throw new SyncAuthError('Недействительный токен устройства', 'DEVICE_UNAUTHORIZED', 401); + } + + if (row.status === 'revoked') { + throw new SyncAuthError('Устройство отозвано', 'DEVICE_REVOKED', 403); + } + + if (row.status !== 'active') { + throw new SyncAuthError('Устройство неактивно', 'DEVICE_INACTIVE', 403); + } + + if (requiredScope && !row.allowed_scopes.includes(requiredScope)) { + throw new SyncAuthError( + 'Устройство не имеет необходимых прав синхронизации', + 'SCOPE_INSUFFICIENT', + 403, + ); + } + + await client.query( + `UPDATE staff_sync_devices + SET last_seen_at = now() + WHERE tenant_id = $1 AND id = $2`, + [tenantId, row.id], + ); + + return { + id: parseUlid(row.id), + tenant_id: parseUlid(row.tenant_id), + device_code: row.device_code, + name: row.name, + token_hash: row.token_hash, + token_hint: row.token_hint, + status: row.status, + allowed_scopes: row.allowed_scopes, + org_unit_id: row.org_unit_id ? parseUlid(row.org_unit_id) : null, + registered_at: row.registered_at.toISOString(), + last_seen_at: new Date().toISOString(), + revoked_at: row.revoked_at ? row.revoked_at.toISOString() : null, + created_by: parseUlid(row.created_by), + }; + }); + } + + async registerDevice( + tenantId: Ulid, + actorId: Ulid, + params: RegisterDeviceParams, + ): Promise { + if (!params.device_code?.trim() || !params.name?.trim()) { + throw new SyncValidationError('Код и наименование устройства обязательны'); + } + + const deviceId = newUlid(); + const rawSecret = crypto.randomBytes(32).toString('hex'); + const rawToken = `kstaff_dev_${tenantId}_${rawSecret}`; + const tokenHash = SyncService.hashToken(rawToken); + const tokenHint = rawToken.slice(0, 16); + const scopes = params.allowed_scopes && params.allowed_scopes.length > 0 + ? params.allowed_scopes + : ['attendance:sync']; + + const orgUnitId = params.org_unit_id ? parseUlid(params.org_unit_id) : null; + + return this.database.tenantTransaction(tenantId, actorId, async (client) => { + const res = await client.query<{ + id: string; + tenant_id: string; + device_code: string; + name: string; + token_hint: string; + status: SyncDeviceStatus; + allowed_scopes: string[]; + org_unit_id: string | null; + registered_at: Date; + created_by: string; + }>( + `INSERT INTO staff_sync_devices + (id, tenant_id, device_code, name, token_hash, token_hint, status, allowed_scopes, org_unit_id, created_by) + VALUES ($1, $2, $3, $4, $5, $6, 'active', $7, $8, $9) + RETURNING id, tenant_id, device_code, name, token_hint, status, allowed_scopes, org_unit_id, registered_at, created_by`, + [deviceId, tenantId, params.device_code.trim(), params.name.trim(), tokenHash, tokenHint, scopes, orgUnitId, actorId], + ); + + const row = res.rows[0]!; + return { + device: { + id: parseUlid(row.id), + tenant_id: parseUlid(row.tenant_id), + device_code: row.device_code, + name: row.name, + token_hint: row.token_hint, + status: row.status, + allowed_scopes: row.allowed_scopes, + org_unit_id: row.org_unit_id ? parseUlid(row.org_unit_id) : null, + registered_at: row.registered_at.toISOString(), + created_by: parseUlid(row.created_by), + }, + token: rawToken, + }; + }); + } + + async revokeDevice( + tenantId: Ulid, + actorId: Ulid, + deviceId: Ulid, + ): Promise<{ id: Ulid; status: SyncDeviceStatus; revoked_at: string }> { + return this.database.tenantTransaction(tenantId, actorId, async (client) => { + const res = await client.query<{ id: string; status: SyncDeviceStatus; revoked_at: Date }>( + `UPDATE staff_sync_devices + SET status = 'revoked', revoked_at = now() + WHERE tenant_id = $1 AND id = $2 + RETURNING id, status, revoked_at`, + [tenantId, deviceId], + ); + + const row = res.rows[0]; + if (!row) { + throw new SyncValidationError('Устройство не найдено', 'DEVICE_NOT_FOUND', 404); + } + + return { + id: parseUlid(row.id), + status: row.status, + revoked_at: row.revoked_at.toISOString(), + }; + }); + } + + async getStatus(tenantId: Ulid, deviceId: Ulid): Promise { + return this.database.tenantTransaction(tenantId, deviceId, async (client) => { + const seqRes = await client.query<{ max_seq: string | null }>( + `SELECT MAX(cursor_seq) AS max_seq FROM staff_sync_change_log WHERE tenant_id = $1`, + [tenantId], + ); + const maxSeq = seqRes.rows[0]?.max_seq ? BigInt(seqRes.rows[0].max_seq) : 0n; + const latestCursor = encodeCursor(tenantId, maxSeq); + + const cpRes = await client.query<{ + last_acked_cursor: string | null; + acknowledged_at: Date | null; + }>( + `SELECT last_acked_cursor, acknowledged_at + FROM staff_sync_device_checkpoints + WHERE tenant_id = $1 AND device_id = $2`, + [tenantId, deviceId], + ); + + const cp = cpRes.rows[0]; + return { + server_time_utc: new Date().toISOString(), + latest_cursor: latestCursor, + device_checkpoint: cp + ? { + last_acked_cursor: cp.last_acked_cursor, + acknowledged_at: cp.acknowledged_at ? cp.acknowledged_at.toISOString() : null, + } + : null, + schema_version: 1, + }; + }); + } + + async acknowledge(tenantId: Ulid, deviceId: Ulid, cursor: string): Promise { + const seq = decodeCursor(tenantId, cursor); + + return this.database.tenantTransaction(tenantId, deviceId, async (client) => { + const res = await client.query<{ acknowledged_at: Date }>( + `INSERT INTO staff_sync_device_checkpoints + (tenant_id, device_id, last_acked_cursor_seq, last_acked_cursor, acknowledged_at) + VALUES ($1, $2, $3, $4, now()) + ON CONFLICT (tenant_id, device_id) + DO UPDATE SET + last_acked_cursor_seq = EXCLUDED.last_acked_cursor_seq, + last_acked_cursor = EXCLUDED.last_acked_cursor, + acknowledged_at = EXCLUDED.acknowledged_at + RETURNING acknowledged_at`, + [tenantId, deviceId, seq.toString(), cursor], + ); + + const row = res.rows[0]!; + return { + acknowledged: true, + cursor, + acknowledged_at: row.acknowledged_at.toISOString(), + }; + }); + } + + async pull(tenantId: Ulid, deviceId: Ulid, options: PullOptions = {}): Promise { + const limit = Math.min(Math.max(1, options.limit ?? 100), 100); + + return this.database.tenantTransaction(tenantId, deviceId, async (client) => { + // 1. If cursor is provided -> incremental pull from staff_sync_change_log + if (options.cursor) { + const seq = decodeCursor(tenantId, options.cursor); + const filterTypes = options.entity_types && options.entity_types.length > 0 ? options.entity_types : null; + + const res = await client.query<{ + id: string; + cursor_seq: string; + entity_type: string; + entity_id: string; + operation: SyncOperation; + revision: number; + payload: Record; + created_at: Date; + }>( + `SELECT id, cursor_seq, entity_type, entity_id, operation, revision, payload, created_at + FROM staff_sync_change_log + WHERE tenant_id = $1 AND cursor_seq > $2 + AND ($3::text[] IS NULL OR entity_type = ANY($3)) + ORDER BY cursor_seq ASC + LIMIT $4`, + [tenantId, seq.toString(), filterTypes, limit + 1], + ); + + const hasMore = res.rows.length > limit; + const pageRows = res.rows.slice(0, limit); + + const changes: SyncChange[] = pageRows.map((r) => ({ + cursor: encodeCursor(tenantId, r.cursor_seq), + entity_type: r.entity_type, + entity_id: r.entity_id, + operation: r.operation, + revision: Number(r.revision), + payload: r.payload, + created_at: r.created_at.toISOString(), + })); + + const nextCursor = changes.length > 0 + ? changes[changes.length - 1]!.cursor + : options.cursor; + + return { + changes, + next_cursor: nextCursor, + has_more: hasMore, + }; + } + + // 2. If cursor is omitted/null -> initial bootstrap of current active snapshot + const seqRes = await client.query<{ max_seq: string | null }>( + `SELECT MAX(cursor_seq) AS max_seq FROM staff_sync_change_log WHERE tenant_id = $1`, + [tenantId], + ); + const maxSeq = seqRes.rows[0]?.max_seq ? BigInt(seqRes.rows[0].max_seq) : 0n; + const latestCursor = encodeCursor(tenantId, maxSeq); + + const wanted = (type: string) => + !options.entity_types || options.entity_types.length === 0 || options.entity_types.includes(type); + + const changes: SyncChange[] = []; + + // Org Units + if (wanted('org_unit')) { + const orgRes = await client.query<{ id: string; code: string; name: string; status: string }>( + `SELECT id, code, name, status FROM staff_org_units WHERE tenant_id = $1 ORDER BY id`, + [tenantId], + ); + for (const row of orgRes.rows) { + const payload: SafeOrgUnitPayload = { + id: row.id, + code: row.code, + name: row.name, + status: row.status, + }; + changes.push({ + cursor: latestCursor, + entity_type: 'org_unit', + entity_id: row.id, + operation: 'upsert', + revision: 1, + payload: payload as unknown as Record, + created_at: new Date().toISOString(), + }); + } + } + + // Professions + if (wanted('profession')) { + const profRes = await client.query<{ id: string; code: string; name: string; status: string }>( + `SELECT id, code, name, status FROM staff_professions WHERE tenant_id = $1 ORDER BY id`, + [tenantId], + ); + for (const row of profRes.rows) { + const payload: SafeProfessionPayload = { + id: row.id, + code: row.code, + name: row.name, + status: row.status, + }; + changes.push({ + cursor: latestCursor, + entity_type: 'profession', + entity_id: row.id, + operation: 'upsert', + revision: 1, + payload: payload as unknown as Record, + created_at: new Date().toISOString(), + }); + } + } + + // Employees (NO salaries, NO personal PII, NO biometrics) + if (wanted('employee')) { + const empRes = await client.query<{ + id: string; + personnel_number: string; + display_name: string; + status: string; + profession_id: string | null; + position: string; + }>( + `SELECT id, personnel_number, display_name, status, profession_id, position + FROM staff_employees + WHERE tenant_id = $1 + ORDER BY id`, + [tenantId], + ); + for (const row of empRes.rows) { + const payload: SafeEmployeePayload = { + id: row.id, + personnel_number: row.personnel_number, + display_name: row.display_name, + status: row.status, + profession_id: row.profession_id, + position: row.position, + }; + changes.push({ + cursor: latestCursor, + entity_type: 'employee', + entity_id: row.id, + operation: 'upsert', + revision: 1, + payload: payload as unknown as Record, + created_at: new Date().toISOString(), + }); + } + } + + // Work Sessions + if (wanted('work_session')) { + const wsRes = await client.query<{ + id: string; + employee_id: string; + org_unit_id: string; + business_date: Date; + started_at: Date | null; + ended_at: Date | null; + worked_minutes: number; + status: string; + revision: number; + }>( + `SELECT id, employee_id, org_unit_id, business_date, started_at, ended_at, worked_minutes, status, revision + FROM staff_work_sessions + WHERE tenant_id = $1 + ORDER BY id`, + [tenantId], + ); + for (const row of wsRes.rows) { + const payload: SafeWorkSessionPayload = { + id: row.id, + employee_id: row.employee_id, + org_unit_id: row.org_unit_id, + business_date: row.business_date.toISOString().slice(0, 10), + started_at: row.started_at ? row.started_at.toISOString() : null, + ended_at: row.ended_at ? row.ended_at.toISOString() : null, + worked_minutes: row.worked_minutes, + status: row.status, + revision: row.revision, + }; + changes.push({ + cursor: latestCursor, + entity_type: 'work_session', + entity_id: row.id, + operation: 'upsert', + revision: row.revision, + payload: payload as unknown as Record, + created_at: new Date().toISOString(), + }); + } + } + + // Timesheet Entries + if (wanted('timesheet_entry')) { + const tsRes = await client.query<{ + id: string; + employee_id: string; + work_date: Date; + minutes_worked: number; + status: string; + revision: number; + }>( + `SELECT id, employee_id, work_date, minutes_worked, status, revision + FROM staff_timesheet_entries + WHERE tenant_id = $1 + ORDER BY id`, + [tenantId], + ); + for (const row of tsRes.rows) { + const payload: SafeTimesheetEntryPayload = { + id: row.id, + employee_id: row.employee_id, + work_date: row.work_date.toISOString().slice(0, 10), + minutes_worked: row.minutes_worked, + status: row.status, + revision: row.revision, + }; + changes.push({ + cursor: latestCursor, + entity_type: 'timesheet_entry', + entity_id: row.id, + operation: 'upsert', + revision: row.revision, + payload: payload as unknown as Record, + created_at: new Date().toISOString(), + }); + } + } + + // Planned Shifts (from staff_work_items WHERE item_kind = 'planned_shift') + if (wanted('planned_shift')) { + const shiftRes = await client.query<{ + id: string; + employee_id: string; + org_unit_id: string; + business_date: Date; + attributes: { + planned_minutes?: number; + start_time?: string | null; + end_time?: string | null; + }; + }>( + `SELECT id, employee_id, org_unit_id, business_date, attributes + FROM staff_work_items + WHERE tenant_id = $1 AND item_kind = 'planned_shift' + ORDER BY id`, + [tenantId], + ); + for (const row of shiftRes.rows) { + const payload: SafePlannedShiftPayload = { + id: row.id, + employee_id: row.employee_id, + org_unit_id: row.org_unit_id, + business_date: row.business_date.toISOString().slice(0, 10), + planned_minutes: Number(row.attributes?.planned_minutes ?? 0), + start_time: row.attributes?.start_time ?? null, + end_time: row.attributes?.end_time ?? null, + }; + changes.push({ + cursor: latestCursor, + entity_type: 'planned_shift', + entity_id: row.id, + operation: 'upsert', + revision: 1, + payload: payload as unknown as Record, + created_at: new Date().toISOString(), + }); + } + } + + const hasMore = changes.length > limit; + const pageChanges = changes.slice(0, limit); + + return { + changes: pageChanges, + next_cursor: latestCursor, + has_more: hasMore, + }; + }); + } + + async push( + tenantId: Ulid, + device: SyncDevice, + commands: SyncCommand[], + ): Promise { + if (!Array.isArray(commands) || commands.length === 0) { + throw new SyncValidationError('Пакет команд не может быть пустым', 'BATCH_EMPTY'); + } + + if (commands.length > 100) { + throw new SyncValidationError( + 'Превышен максимальный размер пакета (максимум 100 команд)', + 'BATCH_SIZE_EXCEEDED', + ); + } + + return this.database.tenantTransaction(tenantId, device.id, async (client) => { + const results: SyncCommandResult[] = []; + + for (const cmd of commands) { + if (!cmd.idempotency_key?.trim()) { + results.push({ + idempotency_key: cmd.idempotency_key ?? '', + command_type: cmd.command_type ?? 'unknown', + status: 'rejected', + error_code: 'MISSING_IDEMPOTENCY_KEY', + error_message: 'idempotency_key обязателен для каждой команды', + }); + continue; + } + + // 1. Max payload limit: 64KB per command + const payloadStr = JSON.stringify(cmd.payload ?? {}); + if (Buffer.byteLength(payloadStr, 'utf8') > 65536) { + const rejectedResult: SyncCommandResult = { + idempotency_key: cmd.idempotency_key, + command_type: cmd.command_type, + status: 'rejected', + error_code: 'PAYLOAD_TOO_LARGE', + error_message: 'Размер полезной нагрузки команды превышает 64 КБ', + }; + await this.saveInboxCommand(client, tenantId, device.id, cmd, rejectedResult); + results.push(rejectedResult); + continue; + } + + // 2. Replay attack protection: window check (not older than 30 days, not > 5 min in future) + const createdAt = new Date(cmd.created_at); + const now = Date.now(); + const maxPast = 30 * 24 * 60 * 60 * 1000; + const maxFuture = 5 * 60 * 1000; + const timeDiff = now - createdAt.getTime(); + + if (isNaN(createdAt.getTime()) || timeDiff > maxPast || timeDiff < -maxFuture) { + const rejectedResult: SyncCommandResult = { + idempotency_key: cmd.idempotency_key, + command_type: cmd.command_type, + status: 'rejected', + error_code: 'TIMESTAMP_OUT_OF_WINDOW', + error_message: 'Временная метка команды находится вне допустимого окна (-30 дней .. +5 минут)', + }; + await this.saveInboxCommand(client, tenantId, device.id, cmd, rejectedResult); + results.push(rejectedResult); + continue; + } + + // 3. Idempotency check: returns cached result if already processed + const existingRes = await client.query<{ + status: 'applied' | 'rejected' | 'conflict'; + result: Record | null; + conflict: SyncConflict | null; + error_code: string | null; + error_message: string | null; + }>( + `SELECT status, result, conflict, error_code, error_message + FROM staff_sync_command_inbox + WHERE tenant_id = $1 AND device_id = $2 AND idempotency_key = $3`, + [tenantId, device.id, cmd.idempotency_key], + ); + + if (existingRes.rows.length > 0) { + const existing = existingRes.rows[0]!; + results.push({ + idempotency_key: cmd.idempotency_key, + command_type: cmd.command_type, + status: existing.status, + result: existing.result, + conflict: existing.conflict, + error_code: existing.error_code, + error_message: existing.error_message, + }); + continue; + } + + // 4. Execute command + const executionResult = await this.executeCommand(client, tenantId, device, cmd); + await this.saveInboxCommand(client, tenantId, device.id, cmd, executionResult); + results.push(executionResult); + } + + return results; + }); + } + + private async executeCommand( + client: PoolClient, + tenantId: Ulid, + device: SyncDevice, + cmd: SyncCommand, + ): Promise { + const payload = cmd.payload ?? {}; + + switch (cmd.command_type) { + case 'attendance_event': + case 'record_attendance': { + const employeeId = payload.employee_id as string; + const eventType = payload.event_type as string; + const occurredAt = payload.occurred_at as string; + const sourceEventId = (payload.source_event_id as string) || cmd.idempotency_key; + const orgUnitId = (payload.org_unit_id as string) || device.org_unit_id; + + if (!employeeId || !['check_in', 'check_out'].includes(eventType) || !occurredAt) { + return { + idempotency_key: cmd.idempotency_key, + command_type: cmd.command_type, + status: 'rejected', + error_code: 'INVALID_ATTENDANCE_PAYLOAD', + error_message: 'employee_id, event_type (check_in/check_out) и occurred_at обязательны', + }; + } + + // Verify employee belongs to tenant + const empCheck = await client.query<{ id: string }>( + `SELECT id FROM staff_employees WHERE tenant_id = $1 AND id = $2`, + [tenantId, employeeId], + ); + if (empCheck.rowCount === 0) { + return { + idempotency_key: cmd.idempotency_key, + command_type: cmd.command_type, + status: 'rejected', + error_code: 'EMPLOYEE_NOT_FOUND', + error_message: 'Сотрудник не найден в данном предприятии', + }; + } + + const eventId = (payload.event_id as string) || newUlid(); + const correlationId = cmd.correlation_id || newUlid(); + + await client.query( + `INSERT INTO staff_attendance_events + (id, tenant_id, employee_id, event_type, source, source_event_id, occurred_at, device_id, actor_id, correlation_id) + VALUES ($1, $2, $3, $4, 'terminal', $5, $6, $7, $8, $9) + ON CONFLICT (tenant_id, source, source_event_id) DO NOTHING`, + [eventId, tenantId, employeeId, eventType, sourceEventId, occurredAt, device.id, device.id, correlationId], + ); + + const safePayload: SafeAttendanceEventPayload = { + id: eventId, + employee_id: employeeId, + event_type: eventType as 'check_in' | 'check_out', + occurred_at: occurredAt, + source_event_id: sourceEventId, + device_id: device.id, + org_unit_id: orgUnitId ? String(orgUnitId) : null, + status: 'accepted', + }; + + await this.recordChange(client, tenantId, 'attendance_event', eventId, 'upsert', 1, safePayload); + + return { + idempotency_key: cmd.idempotency_key, + command_type: cmd.command_type, + status: 'applied', + result: { event_id: eventId, status: 'accepted' }, + }; + } + + case 'upsert_work_session': { + const sessionId = (payload.id as string) || newUlid(); + const employeeId = payload.employee_id as string; + const orgUnitId = (payload.org_unit_id as string) || device.org_unit_id; + const businessDate = payload.business_date as string; + const clientRevision = typeof payload.revision === 'number' ? payload.revision : 1; + const startedAt = payload.started_at as string | null | undefined; + const endedAt = payload.ended_at as string | null | undefined; + const workedMinutes = typeof payload.worked_minutes === 'number' ? payload.worked_minutes : 0; + const sessionStatus = (payload.status as string) || 'open'; + + if (!employeeId || !orgUnitId || !businessDate) { + return { + idempotency_key: cmd.idempotency_key, + command_type: cmd.command_type, + status: 'rejected', + error_code: 'INVALID_WORK_SESSION_PAYLOAD', + error_message: 'employee_id, org_unit_id и business_date обязательны', + }; + } + + const existingRes = await client.query<{ id: string; revision: number; status: string }>( + `SELECT id, revision, status FROM staff_work_sessions WHERE tenant_id = $1 AND id = $2`, + [tenantId, sessionId], + ); + + if (existingRes.rows.length > 0) { + const existing = existingRes.rows[0]!; + if (existing.revision !== clientRevision) { + const conflict: SyncConflict = { + conflict_code: 'REVISION_MISMATCH', + entity_type: 'work_session', + entity_id: sessionId, + expected_revision: clientRevision, + actual_revision: existing.revision, + safe_description: 'Ревизия рабочей смены на сервере изменилась', + client_action: 'FETCH_LATEST_AND_RETRY', + }; + return { + idempotency_key: cmd.idempotency_key, + command_type: cmd.command_type, + status: 'conflict', + conflict, + }; + } + + const nextRev = existing.revision + 1; + await client.query( + `UPDATE staff_work_sessions + SET started_at = $1, ended_at = $2, worked_minutes = $3, status = $4, revision = $5, updated_at = now() + WHERE tenant_id = $6 AND id = $7`, + [startedAt ?? null, endedAt ?? null, workedMinutes, sessionStatus, nextRev, tenantId, sessionId], + ); + + const safePayload: SafeWorkSessionPayload = { + id: sessionId, + employee_id: employeeId, + org_unit_id: orgUnitId, + business_date: businessDate, + started_at: startedAt ?? null, + ended_at: endedAt ?? null, + worked_minutes: workedMinutes, + status: sessionStatus, + revision: nextRev, + }; + + await this.recordChange(client, tenantId, 'work_session', sessionId, 'upsert', nextRev, safePayload); + + return { + idempotency_key: cmd.idempotency_key, + command_type: cmd.command_type, + status: 'applied', + result: { id: sessionId, revision: nextRev, status: 'applied' }, + }; + } + + // New work session insert + const newRev = clientRevision; + await client.query( + `INSERT INTO staff_work_sessions + (id, tenant_id, employee_id, org_unit_id, business_date, started_at, ended_at, worked_minutes, status, revision) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`, + [sessionId, tenantId, employeeId, orgUnitId, businessDate, startedAt ?? null, endedAt ?? null, workedMinutes, sessionStatus, newRev], + ); + + const safePayload: SafeWorkSessionPayload = { + id: sessionId, + employee_id: employeeId, + org_unit_id: orgUnitId, + business_date: businessDate, + started_at: startedAt ?? null, + ended_at: endedAt ?? null, + worked_minutes: workedMinutes, + status: sessionStatus, + revision: newRev, + }; + + await this.recordChange(client, tenantId, 'work_session', sessionId, 'upsert', newRev, safePayload); + + return { + idempotency_key: cmd.idempotency_key, + command_type: cmd.command_type, + status: 'applied', + result: { id: sessionId, revision: newRev, status: 'applied' }, + }; + } + + case 'upsert_timesheet_entry': { + const entryId = (payload.id as string) || newUlid(); + const employeeId = payload.employee_id as string; + const workDate = payload.work_date as string; + const minutesWorked = typeof payload.minutes_worked === 'number' ? payload.minutes_worked : 0; + const tsStatus = (payload.status as string) || 'draft'; + const clientRevision = typeof payload.revision === 'number' ? payload.revision : 1; + + if (!employeeId || !workDate) { + return { + idempotency_key: cmd.idempotency_key, + command_type: cmd.command_type, + status: 'rejected', + error_code: 'INVALID_TIMESHEET_PAYLOAD', + error_message: 'employee_id и work_date обязательны', + }; + } + + const existingRes = await client.query<{ id: string; revision: number; status: string }>( + `SELECT id, revision, status FROM staff_timesheet_entries WHERE tenant_id = $1 AND id = $2`, + [tenantId, entryId], + ); + + if (existingRes.rows.length > 0) { + const existing = existingRes.rows[0]!; + if (existing.revision !== clientRevision) { + const conflict: SyncConflict = { + conflict_code: 'REVISION_MISMATCH', + entity_type: 'timesheet_entry', + entity_id: entryId, + expected_revision: clientRevision, + actual_revision: existing.revision, + safe_description: 'Ревизия записи табеля на сервере изменилась', + client_action: 'FETCH_LATEST_AND_RETRY', + }; + return { + idempotency_key: cmd.idempotency_key, + command_type: cmd.command_type, + status: 'conflict', + conflict, + }; + } + + const nextRev = existing.revision + 1; + await client.query( + `UPDATE staff_timesheet_entries + SET minutes_worked = $1, status = $2, revision = $3, updated_at = now() + WHERE tenant_id = $4 AND id = $5`, + [minutesWorked, tsStatus, nextRev, tenantId, entryId], + ); + + const safePayload: SafeTimesheetEntryPayload = { + id: entryId, + employee_id: employeeId, + work_date: workDate, + minutes_worked: minutesWorked, + status: tsStatus, + revision: nextRev, + }; + + await this.recordChange(client, tenantId, 'timesheet_entry', entryId, 'upsert', nextRev, safePayload); + + return { + idempotency_key: cmd.idempotency_key, + command_type: cmd.command_type, + status: 'applied', + result: { id: entryId, revision: nextRev, status: 'applied' }, + }; + } + + const newRev = clientRevision; + await client.query( + `INSERT INTO staff_timesheet_entries + (id, tenant_id, employee_id, work_date, minutes_worked, status, revision, updated_by) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`, + [entryId, tenantId, employeeId, workDate, minutesWorked, tsStatus, newRev, device.id], + ); + + const safePayload: SafeTimesheetEntryPayload = { + id: entryId, + employee_id: employeeId, + work_date: workDate, + minutes_worked: minutesWorked, + status: tsStatus, + revision: newRev, + }; + + await this.recordChange(client, tenantId, 'timesheet_entry', entryId, 'upsert', newRev, safePayload); + + return { + idempotency_key: cmd.idempotency_key, + command_type: cmd.command_type, + status: 'applied', + result: { id: entryId, revision: newRev, status: 'applied' }, + }; + } + + default: + return { + idempotency_key: cmd.idempotency_key, + command_type: cmd.command_type, + status: 'rejected', + error_code: 'UNKNOWN_COMMAND_TYPE', + error_message: `Неизвестный тип команды: ${cmd.command_type}`, + }; + } + } + + private async saveInboxCommand( + client: PoolClient, + tenantId: Ulid, + deviceId: Ulid, + cmd: SyncCommand, + result: SyncCommandResult, + ): Promise { + const inboxId = newUlid(); + const correlationId = cmd.correlation_id || newUlid(); + const createdAt = cmd.created_at ? new Date(cmd.created_at) : new Date(); + const validCreatedAt = isNaN(createdAt.getTime()) ? new Date() : createdAt; + + await client.query( + `INSERT INTO staff_sync_command_inbox + (id, tenant_id, device_id, command_type, schema_version, idempotency_key, correlation_id, + causation_id, payload, status, result, conflict, error_code, error_message, created_at, processed_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, now()) + ON CONFLICT (tenant_id, device_id, idempotency_key) DO NOTHING`, + [ + inboxId, + tenantId, + deviceId, + cmd.command_type || 'unknown', + cmd.schema_version || 1, + cmd.idempotency_key, + correlationId, + cmd.causation_id ?? null, + JSON.stringify(cmd.payload ?? {}), + result.status, + result.result ? JSON.stringify(result.result) : null, + result.conflict ? JSON.stringify(result.conflict) : null, + result.error_code ?? null, + result.error_message ?? null, + validCreatedAt, + ], + ); + } + + private async recordChange( + client: PoolClient, + tenantId: Ulid, + entityType: string, + entityId: string, + operation: SyncOperation, + revision: number, + payload: unknown, + ): Promise { + const changeId = newUlid(); + await client.query( + `INSERT INTO staff_sync_change_log + (id, tenant_id, entity_type, entity_id, operation, revision, payload, created_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, now())`, + [changeId, tenantId, entityType, entityId, operation, revision, JSON.stringify(payload)], + ); + } + + async cleanupRetention(retentionDays = 30): Promise<{ purged_change_logs: number; purged_commands: number }> { + return this.database.tenantTransaction(this.systemActorId, this.systemActorId, async (client) => { + const res = await client.query<{ purged_change_logs: string; purged_commands: string }>( + `SELECT purged_change_logs, purged_commands FROM staff_sync_cleanup_retention($1)`, + [retentionDays], + ); + const row = res.rows[0]; + return { + purged_change_logs: Number(row?.purged_change_logs ?? 0), + purged_commands: Number(row?.purged_commands ?? 0), + }; + }); + } +} diff --git a/src/domain/sync/types.ts b/src/domain/sync/types.ts new file mode 100644 index 0000000..9a80018 --- /dev/null +++ b/src/domain/sync/types.ts @@ -0,0 +1,165 @@ +import type { Ulid } from '../../core/ids.js'; + +export type SyncDeviceStatus = 'active' | 'revoked' | 'pending'; +export type SyncOperation = 'upsert' | 'delete'; +export type SyncCommandStatus = 'applied' | 'rejected' | 'conflict'; + +export interface SyncDevice { + id: Ulid; + tenant_id: Ulid; + device_code: string; + name: string; + token_hash: string; + token_hint: string; + status: SyncDeviceStatus; + allowed_scopes: string[]; + org_unit_id: Ulid | null; + registered_at: string; + last_seen_at: string | null; + revoked_at: string | null; + created_by: Ulid; +} + +export interface SyncDeviceRegistrationResult { + device: { + id: Ulid; + tenant_id: Ulid; + device_code: string; + name: string; + token_hint: string; + status: SyncDeviceStatus; + allowed_scopes: string[]; + org_unit_id: Ulid | null; + registered_at: string; + created_by: Ulid; + }; + token: string; +} + +export interface SyncConflict { + conflict_code: 'REVISION_MISMATCH'; + entity_type: string; + entity_id: string; + expected_revision: number; + actual_revision: number; + safe_description: string; + client_action: 'FETCH_LATEST_AND_RETRY'; +} + +export interface SyncCommand

> { + command_type: string; + schema_version: number; + idempotency_key: string; + correlation_id: string; + causation_id?: string | null; + created_at: string; + payload: P; +} + +export interface SyncCommandResult { + idempotency_key: string; + command_type: string; + status: SyncCommandStatus; + result?: Record | null; + conflict?: SyncConflict | null; + error_code?: string | null; + error_message?: string | null; +} + +export interface SyncChange> { + cursor: string; + entity_type: string; + entity_id: string; + operation: SyncOperation; + revision: number; + payload: T; + created_at: string; +} + +export interface SyncStatus { + server_time_utc: string; + latest_cursor: string; + device_checkpoint: { + last_acked_cursor: string | null; + acknowledged_at: string | null; + } | null; + schema_version: number; +} + +export interface SyncPullResult { + changes: SyncChange[]; + next_cursor: string; + has_more: boolean; +} + +export interface SyncAckResult { + acknowledged: boolean; + cursor: string; + acknowledged_at: string; +} + +// Minimal safe payload schemas (no salary, no personal PII, no biometrics) +export interface SafeOrgUnitPayload { + id: string; + code: string; + name: string; + status: string; +} + +export interface SafeProfessionPayload { + id: string; + code: string; + name: string; + status: string; +} + +export interface SafeEmployeePayload { + id: string; + personnel_number: string; + display_name: string; + status: string; + profession_id: string | null; + position: string; +} + +export interface SafeWorkSessionPayload { + id: string; + employee_id: string; + org_unit_id: string; + business_date: string; + started_at: string | null; + ended_at: string | null; + worked_minutes: number; + status: string; + revision: number; +} + +export interface SafeTimesheetEntryPayload { + id: string; + employee_id: string; + work_date: string; + minutes_worked: number; + status: string; + revision: number; +} + +export interface SafePlannedShiftPayload { + id: string; + employee_id: string; + org_unit_id: string; + business_date: string; + planned_minutes: number; + start_time: string | null; + end_time: string | null; +} + +export interface SafeAttendanceEventPayload { + id: string; + employee_id: string; + event_type: 'check_in' | 'check_out'; + occurred_at: string; + source_event_id: string; + device_id: string | null; + org_unit_id: string | null; + status: string; +} diff --git a/src/sync.ts b/src/sync.ts new file mode 100644 index 0000000..06d23f4 --- /dev/null +++ b/src/sync.ts @@ -0,0 +1,457 @@ +import type { FastifyInstance, FastifyRequest } from 'fastify'; +import type { AppConfig } from './config.js'; +import { parseUlid, type Ulid } from './core/ids.js'; +import type { Database } from './infrastructure/database.js'; +import { PlatformIdentityAdapter, StandaloneIdentityAdapter } from './adapters/identity.js'; +import type { IdentityPort } from './core/ports.js'; +import { + SyncAuthError, + SyncService, + SyncValidationError, +} from './domain/sync/service.js'; +import { InvalidCursorError } from './domain/sync/cursor.js'; +import type { SyncCommand } from './domain/sync/types.js'; + +interface PullBody { + cursor?: string | null; + limit?: number; + entity_types?: string[]; +} + +interface PushBody { + commands: SyncCommand[]; +} + +interface AckBody { + cursor: string; +} + +interface RegisterDeviceBody { + device_code: string; + name: string; + org_unit_id?: string | null; + allowed_scopes?: string[]; +} + +interface RevokeParams { + deviceId: string; +} + +const ulidRegex = /^[0-9A-HJKMNP-TV-Z]{26}$/; + +export async function registerSyncRoutes( + api: FastifyInstance, + config: AppConfig, + database: Database, +) { + const baseIdentity: IdentityPort = + config.STAFF_RUNTIME_MODE === 'platform' + ? new PlatformIdentityAdapter() + : new StandaloneIdentityAdapter(config); + + const defaultActorId = parseUlid(config.STANDALONE_ACTOR_ID); + const syncService = new SyncService(database, defaultActorId); + + const extractBearerToken = (request: FastifyRequest): string => { + const authHeader = request.headers.authorization; + if (!authHeader?.startsWith('Bearer ')) { + throw new SyncAuthError('Требуется токен авторизации устройства', 'DEVICE_UNAUTHORIZED', 401); + } + return authHeader.slice(7).trim(); + }; + + const resolveTenantId = (request: FastifyRequest): Ulid | undefined => { + const header = request.headers['x-tenant-id']?.toString(); + if (header && ulidRegex.test(header)) { + return parseUlid(header); + } + if (config.STANDALONE_TENANT_ID && ulidRegex.test(config.STANDALONE_TENANT_ID)) { + return parseUlid(config.STANDALONE_TENANT_ID); + } + return undefined; + }; + + const requireManagerContext = async (request: FastifyRequest) => { + const authHeader = request.headers.authorization; + const cookieHeader = request.headers.cookie; + const sessionCookieName = '__Host-krona_staff_session'; + + let token = authHeader?.startsWith('Bearer ') ? authHeader.slice(7).trim() : undefined; + if (!token && cookieHeader) { + for (const part of cookieHeader.split(';')) { + const sep = part.indexOf('='); + if (sep > 0 && part.slice(0, sep).trim() === sessionCookieName) { + try { + token = decodeURIComponent(part.slice(sep + 1).trim()); + } catch { + // ignore malformed cookie + } + break; + } + } + } + + try { + const context = await baseIdentity.resolve(token, parseUlid(request.id)); + if ( + !context.permissions.has('staff:attendance:manage') && + !context.permissions.has('staff:sync:manage') + ) { + throw request.server.httpErrors.forbidden('Недостаточно прав для управления устройствами синхронизации'); + } + return context; + } catch (error: unknown) { + if (typeof error === 'object' && error !== null && 'statusCode' in error) { + throw error; + } + throw request.server.httpErrors.unauthorized('Требуется действующий ключ доступа руководителя'); + } + }; + + // 1. Device Registration (Manager endpoint) + api.post<{ Body: RegisterDeviceBody }>( + '/sync/devices', + { + schema: { + tags: ['Sync'], + summary: 'Регистрация нового устройства синхронизации', + body: { + type: 'object', + additionalProperties: false, + required: ['device_code', 'name'], + properties: { + device_code: { type: 'string', minLength: 1, maxLength: 64 }, + name: { type: 'string', minLength: 1, maxLength: 255 }, + org_unit_id: { type: 'string', pattern: '^[0-9A-HJKMNP-TV-Z]{26}$' }, + allowed_scopes: { + type: 'array', + items: { type: 'string' }, + default: ['attendance:sync'], + }, + }, + }, + response: { + 201: { + type: 'object', + required: ['device', 'token'], + properties: { + device: { + type: 'object', + required: [ + 'id', + 'tenant_id', + 'device_code', + 'name', + 'token_hint', + 'status', + 'allowed_scopes', + 'registered_at', + 'created_by', + ], + properties: { + id: { type: 'string' }, + tenant_id: { type: 'string' }, + device_code: { type: 'string' }, + name: { type: 'string' }, + token_hint: { type: 'string' }, + status: { type: 'string' }, + allowed_scopes: { type: 'array', items: { type: 'string' } }, + org_unit_id: { type: 'string', nullable: true }, + registered_at: { type: 'string' }, + created_by: { type: 'string' }, + }, + }, + token: { type: 'string' }, + }, + }, + }, + }, + }, + async (request, reply) => { + const manager = await requireManagerContext(request); + const result = await syncService.registerDevice(manager.tenantId, manager.actorId, request.body); + return reply.code(201).send(result); + }, + ); + + // 2. Revoke Device (Manager endpoint) + api.post<{ Params: RevokeParams }>( + '/sync/devices/:deviceId/revoke', + { + schema: { + tags: ['Sync'], + summary: 'Отзыв доступа устройства синхронизации', + params: { + type: 'object', + additionalProperties: false, + required: ['deviceId'], + properties: { + deviceId: { type: 'string', pattern: '^[0-9A-HJKMNP-TV-Z]{26}$' }, + }, + }, + response: { + 200: { + type: 'object', + required: ['id', 'status', 'revoked_at'], + properties: { + id: { type: 'string' }, + status: { type: 'string', const: 'revoked' }, + revoked_at: { type: 'string' }, + }, + }, + }, + }, + }, + async (request, reply) => { + const manager = await requireManagerContext(request); + const deviceId = parseUlid(request.params.deviceId); + const result = await syncService.revokeDevice(manager.tenantId, manager.actorId, deviceId); + return reply.code(200).send(result); + }, + ); + + // 3. Sync Status (Device endpoint) + api.get( + '/sync/status', + { + schema: { + tags: ['Sync'], + summary: 'Получить текущий статус синхронизации сервера и чекпоинта устройства', + response: { + 200: { + type: 'object', + required: ['server_time_utc', 'latest_cursor', 'schema_version'], + properties: { + server_time_utc: { type: 'string' }, + latest_cursor: { type: 'string' }, + device_checkpoint: { + type: 'object', + nullable: true, + properties: { + last_acked_cursor: { type: 'string', nullable: true }, + acknowledged_at: { type: 'string', nullable: true }, + }, + }, + schema_version: { type: 'integer' }, + }, + }, + }, + }, + }, + async (request, reply) => { + try { + const token = extractBearerToken(request); + const tenantId = resolveTenantId(request); + const device = await syncService.authenticateDevice(token, tenantId); + const status = await syncService.getStatus(device.tenant_id, device.id); + return reply.code(200).send(status); + } catch (error) { + handleSyncError(error, request); + } + }, + ); + + // 4. Sync Pull (Device endpoint) + api.post<{ Body: PullBody }>( + '/sync/pull', + { + schema: { + tags: ['Sync'], + summary: 'Получение изменений после указанного курсора (или начальный bootstrap при пустом курсоре)', + body: { + type: 'object', + additionalProperties: false, + properties: { + cursor: { type: 'string', nullable: true }, + limit: { type: 'integer', minimum: 1, maximum: 100, default: 100 }, + entity_types: { + type: 'array', + items: { type: 'string' }, + }, + }, + }, + response: { + 200: { + type: 'object', + required: ['changes', 'next_cursor', 'has_more'], + properties: { + changes: { + type: 'array', + items: { + type: 'object', + required: ['cursor', 'entity_type', 'entity_id', 'operation', 'revision', 'payload', 'created_at'], + properties: { + cursor: { type: 'string' }, + entity_type: { type: 'string' }, + entity_id: { type: 'string' }, + operation: { type: 'string', enum: ['upsert', 'delete'] }, + revision: { type: 'integer' }, + payload: { type: 'object', additionalProperties: true }, + created_at: { type: 'string' }, + }, + }, + }, + next_cursor: { type: 'string' }, + has_more: { type: 'boolean' }, + }, + }, + }, + }, + }, + async (request, reply) => { + try { + const token = extractBearerToken(request); + const tenantId = resolveTenantId(request); + const device = await syncService.authenticateDevice(token, tenantId); + const pullResult = await syncService.pull(device.tenant_id, device.id, request.body ?? {}); + return reply.code(200).send(pullResult); + } catch (error) { + handleSyncError(error, request); + } + }, + ); + + // 5. Sync Push (Device endpoint) + api.post<{ Body: PushBody }>( + '/sync/push', + { + bodyLimit: 2097152, // 2MB max request size limit + schema: { + tags: ['Sync'], + summary: 'Пакетная отправка команд от устройства на сервер', + body: { + type: 'object', + additionalProperties: false, + required: ['commands'], + properties: { + commands: { + type: 'array', + minItems: 1, + maxItems: 100, + items: { + type: 'object', + required: ['command_type', 'schema_version', 'idempotency_key', 'correlation_id', 'created_at', 'payload'], + properties: { + command_type: { type: 'string' }, + schema_version: { type: 'integer', default: 1 }, + idempotency_key: { type: 'string', minLength: 1, maxLength: 128 }, + correlation_id: { type: 'string' }, + causation_id: { type: 'string', nullable: true }, + created_at: { type: 'string' }, + payload: { type: 'object', additionalProperties: true }, + }, + }, + }, + }, + }, + response: { + 200: { + type: 'object', + required: ['results'], + properties: { + results: { + type: 'array', + items: { + type: 'object', + required: ['idempotency_key', 'command_type', 'status'], + properties: { + idempotency_key: { type: 'string' }, + command_type: { type: 'string' }, + status: { type: 'string', enum: ['applied', 'rejected', 'conflict'] }, + result: { type: 'object', nullable: true, additionalProperties: true }, + conflict: { + type: 'object', + nullable: true, + properties: { + conflict_code: { type: 'string' }, + entity_type: { type: 'string' }, + entity_id: { type: 'string' }, + expected_revision: { type: 'integer' }, + actual_revision: { type: 'integer' }, + safe_description: { type: 'string' }, + client_action: { type: 'string' }, + }, + }, + error_code: { type: 'string', nullable: true }, + error_message: { type: 'string', nullable: true }, + }, + }, + }, + }, + }, + }, + }, + }, + async (request, reply) => { + try { + const token = extractBearerToken(request); + const tenantId = resolveTenantId(request); + const device = await syncService.authenticateDevice(token, tenantId); + const results = await syncService.push(device.tenant_id, device, request.body?.commands ?? []); + return reply.code(200).send({ results }); + } catch (error) { + handleSyncError(error, request); + } + }, + ); + + // 6. Sync Ack (Device endpoint) + api.post<{ Body: AckBody }>( + '/sync/ack', + { + schema: { + tags: ['Sync'], + summary: 'Подтверждение приёма и обработки изменений устройством до заданного курсора', + body: { + type: 'object', + additionalProperties: false, + required: ['cursor'], + properties: { + cursor: { type: 'string', minLength: 5 }, + }, + }, + response: { + 200: { + type: 'object', + required: ['acknowledged', 'cursor', 'acknowledged_at'], + properties: { + acknowledged: { type: 'boolean' }, + cursor: { type: 'string' }, + acknowledged_at: { type: 'string' }, + }, + }, + }, + }, + }, + async (request, reply) => { + try { + const token = extractBearerToken(request); + const tenantId = resolveTenantId(request); + const device = await syncService.authenticateDevice(token, tenantId); + const result = await syncService.acknowledge(device.tenant_id, device.id, request.body.cursor); + return reply.code(200).send(result); + } catch (error) { + handleSyncError(error, request); + } + }, + ); +} + +function handleSyncError(error: unknown, request: FastifyRequest): never { + if (error instanceof SyncAuthError) { + if (error.statusCode === 403) { + throw request.server.httpErrors.forbidden(error.message); + } + throw request.server.httpErrors.unauthorized(error.message); + } + if (error instanceof InvalidCursorError) { + throw request.server.httpErrors.badRequest(error.message); + } + if (error instanceof SyncValidationError) { + if (error.statusCode === 404) { + throw request.server.httpErrors.notFound(error.message); + } + throw request.server.httpErrors.badRequest(error.message); + } + throw error; +} diff --git a/tests/integration/openapi-runtime.test.ts b/tests/integration/openapi-runtime.test.ts index 38cc92b..6092743 100644 --- a/tests/integration/openapi-runtime.test.ts +++ b/tests/integration/openapi-runtime.test.ts @@ -53,7 +53,8 @@ describe('OpenAPI runtime contract', () => { const runtimePath = path .replace('{employeeId}', '01J00000000000000000000000') .replace('{workDate}', '2026-09-24') - .replace('{runId}', '01J00000000000000000000000'); + .replace('{runId}', '01J00000000000000000000000') + .replace('{deviceId}', '01J00000000000000000000000'); const operations = documented.paths[path] as Record; const methods = ['get', 'post', 'put', 'patch', 'delete'] as const; const method = methods.find((candidate) => candidate in operations); diff --git a/tests/integration/tenant-isolation.test.ts b/tests/integration/tenant-isolation.test.ts index c6ef04a..a954a3a 100644 --- a/tests/integration/tenant-isolation.test.ts +++ b/tests/integration/tenant-isolation.test.ts @@ -16,6 +16,9 @@ describeDatabase('PostgreSQL tenant isolation', () => { const employeeIdA = newUlid(); const subscriptionIdA = newUlid(); const deliveryIdA = newUlid(); + const deviceIdA = newUlid(); + const changeIdA = newUlid(); + const commandIdA = newUlid(); beforeAll(async () => { const role = await database.pool.query( @@ -26,6 +29,10 @@ describeDatabase('PostgreSQL tenant isolation', () => { afterAll(async () => { await database.tenantTransaction(tenantA, actorA, async (client) => { + await client.query('DELETE FROM staff_sync_device_checkpoints WHERE tenant_id = $1', [tenantA]); + await client.query('DELETE FROM staff_sync_command_inbox WHERE tenant_id = $1', [tenantA]); + await client.query('DELETE FROM staff_sync_change_log WHERE tenant_id = $1', [tenantA]); + await client.query('DELETE FROM staff_sync_devices WHERE tenant_id = $1', [tenantA]); await client.query('DELETE FROM staff_notification_deliveries WHERE id = $1', [deliveryIdA]); await client.query('DELETE FROM staff_push_subscriptions WHERE id = $1', [subscriptionIdA]); await client.query('DELETE FROM staff_outbox WHERE id = $1', [eventId]); @@ -161,5 +168,103 @@ describeDatabase('PostgreSQL tenant isolation', () => { ), ), ).rejects.toThrow(); + + // Offline sync devices tenant isolation + await database.tenantTransaction(tenantA, actorA, async (client) => { + await client.query( + `INSERT INTO staff_sync_devices (id, tenant_id, device_code, name, token_hash, token_hint, status, created_by) + VALUES ($1, $2, 'DEV-A', 'Terminal A', '0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef', 'kstaff_dev_test', 'active', $3)`, + [deviceIdA, tenantA, actorA], + ); + }); + + const devVisibleToB = await database.tenantTransaction(tenantB, actorB, (client) => + client.query('SELECT id FROM staff_sync_devices WHERE id = $1', [deviceIdA]), + ); + expect(devVisibleToB.rowCount).toBe(0); + + await expect( + database.tenantTransaction(tenantB, actorB, (client) => + client.query( + `INSERT INTO staff_sync_devices (id, tenant_id, device_code, name, token_hash, token_hint, status, created_by) + VALUES ($1, $2, 'DEV-B-FAIL', 'Fail Terminal', 'fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210', 'hint', 'active', $3)`, + [newUlid(), tenantA, actorB], + ), + ), + ).rejects.toThrow(); + + // Offline sync change log tenant isolation + await database.tenantTransaction(tenantA, actorA, async (client) => { + await client.query( + `INSERT INTO staff_sync_change_log (id, tenant_id, entity_type, entity_id, operation, revision, payload) + VALUES ($1, $2, 'employee', $3, 'upsert', 1, '{}'::jsonb)`, + [changeIdA, tenantA, employeeIdA], + ); + }); + + const changeVisibleToB = await database.tenantTransaction(tenantB, actorB, (client) => + client.query('SELECT id FROM staff_sync_change_log WHERE id = $1', [changeIdA]), + ); + expect(changeVisibleToB.rowCount).toBe(0); + + await expect( + database.tenantTransaction(tenantB, actorB, (client) => + client.query( + `INSERT INTO staff_sync_change_log (id, tenant_id, entity_type, entity_id, operation, revision, payload) + VALUES ($1, $2, 'employee', $3, 'upsert', 1, '{}'::jsonb)`, + [newUlid(), tenantA, employeeIdA], + ), + ), + ).rejects.toThrow(); + + // Offline sync command inbox tenant isolation + await database.tenantTransaction(tenantA, actorA, async (client) => { + await client.query( + `INSERT INTO staff_sync_command_inbox + (id, tenant_id, device_id, command_type, schema_version, idempotency_key, correlation_id, payload, status, created_at) + VALUES ($1, $2, $3, 'attendance_event', 1, 'idemp-cmd-a', $4, '{}'::jsonb, 'applied', now())`, + [commandIdA, tenantA, deviceIdA, newUlid()], + ); + }); + + const cmdVisibleToB = await database.tenantTransaction(tenantB, actorB, (client) => + client.query('SELECT id FROM staff_sync_command_inbox WHERE id = $1', [commandIdA]), + ); + expect(cmdVisibleToB.rowCount).toBe(0); + + await expect( + database.tenantTransaction(tenantB, actorB, (client) => + client.query( + `INSERT INTO staff_sync_command_inbox + (id, tenant_id, device_id, command_type, schema_version, idempotency_key, correlation_id, payload, status, created_at) + VALUES ($1, $2, $3, 'attendance_event', 1, 'idemp-cmd-b', $4, '{}'::jsonb, 'applied', now())`, + [newUlid(), tenantA, deviceIdA, newUlid()], + ), + ), + ).rejects.toThrow(); + + // Offline sync checkpoints tenant isolation + await database.tenantTransaction(tenantA, actorA, async (client) => { + await client.query( + `INSERT INTO staff_sync_device_checkpoints (tenant_id, device_id, last_acked_cursor_seq, last_acked_cursor) + VALUES ($1, $2, 10, 'cur_10_test')`, + [tenantA, deviceIdA], + ); + }); + + const cpVisibleToB = await database.tenantTransaction(tenantB, actorB, (client) => + client.query('SELECT device_id FROM staff_sync_device_checkpoints WHERE device_id = $1', [deviceIdA]), + ); + expect(cpVisibleToB.rowCount).toBe(0); + + await expect( + database.tenantTransaction(tenantB, actorB, (client) => + client.query( + `INSERT INTO staff_sync_device_checkpoints (tenant_id, device_id, last_acked_cursor_seq, last_acked_cursor) + VALUES ($1, $2, 10, 'cur_10_fail')`, + [tenantA, deviceIdA], + ), + ), + ).rejects.toThrow(); }); }); diff --git a/tests/unit/sync.test.ts b/tests/unit/sync.test.ts new file mode 100644 index 0000000..5f8826e --- /dev/null +++ b/tests/unit/sync.test.ts @@ -0,0 +1,778 @@ +import crypto from 'node:crypto'; +import { describe, expect, it, vi } from 'vitest'; +import { newUlid, parseUlid } from '../../src/core/ids.js'; +import type { Database } from '../../src/infrastructure/database.js'; +import { decodeCursor, encodeCursor, InvalidCursorError } from '../../src/domain/sync/cursor.js'; +import { SyncService } from '../../src/domain/sync/service.js'; +import type { SyncCommand } from '../../src/domain/sync/types.js'; +import { buildApp } from '../../src/app.js'; +import { loadConfig } from '../../src/config.js'; + +function createMockDb(queryHandler?: (sql: string, params: unknown[]) => { rows: unknown[]; rowCount?: number }) { + const queries: Array<{ sql: string; params: unknown[] }> = []; + const query = vi.fn(async (sql: string, params: unknown[] = []) => { + queries.push({ sql, params }); + if (queryHandler) { + const res = queryHandler(sql, params); + return { rows: res.rows, rowCount: res.rowCount ?? res.rows.length }; + } + return { rows: [], rowCount: 0 }; + }); + + const database = { + ping: vi.fn(async () => undefined), + close: vi.fn(async () => undefined), + tenantTransaction: vi.fn( + async ( + _tenantId: string, + _actorId: string, + run: (client: { query: typeof query }) => Promise, + ) => run({ query }), + ), + } as unknown as Database; + return { database, queries }; +} + +describe('Sync Cursor', () => { + const tenantA = parseUlid('01J0000000000000000000000A'); + const tenantB = parseUlid('01J0000000000000000000000B'); + + it('encodes and decodes monotonic cursor correctly', () => { + const cursor = encodeCursor(tenantA, 42n); + expect(cursor).toMatch(/^cur_42_[0-9a-f]{12}$/); + const decoded = decodeCursor(tenantA, cursor); + expect(decoded).toBe(42n); + }); + + it('rejects cursor with invalid format or tampered signature', () => { + expect(() => decodeCursor(tenantA, 'invalid')).toThrowError(InvalidCursorError); + expect(() => decodeCursor(tenantA, 'cur_42')).toThrowError(InvalidCursorError); + expect(() => decodeCursor(tenantA, 'cur_abc_1234567890ab')).toThrowError(InvalidCursorError); + + // Tampered seq with same signature + const valid = encodeCursor(tenantA, 42n); + const tampered = valid.replace('cur_42_', 'cur_43_'); + expect(() => decodeCursor(tenantA, tampered)).toThrowError(InvalidCursorError); + }); + + it('enforces tenant isolation in cursor verification', () => { + const cursorA = encodeCursor(tenantA, 100n); + // Passing cursor of tenantA to tenantB must fail + expect(() => decodeCursor(tenantB, cursorA)).toThrowError(InvalidCursorError); + }); +}); + +describe('Sync Device Authentication & Tenant Isolation', () => { + const tenantA = parseUlid('01J0000000000000000000000A'); + const tenantB = parseUlid('01J0000000000000000000000B'); + const systemActorId = parseUlid('01J00000000000000000000001'); + + it('authenticates active device and updates last_seen_at', async () => { + const deviceId = newUlid(); + const rawSecret = crypto.randomBytes(32).toString('hex'); + const rawToken = `kstaff_dev_${tenantA}_${rawSecret}`; + const tokenHash = SyncService.hashToken(rawToken); + + const { database, queries } = createMockDb((sql) => { + if (sql.includes('SELECT id, tenant_id')) { + return { + rows: [ + { + id: deviceId, + tenant_id: tenantA, + device_code: 'TERM-01', + name: 'Terminal 1', + token_hash: tokenHash, + token_hint: 'kstaff_dev_01J...', + status: 'active', + allowed_scopes: ['attendance:sync'], + org_unit_id: null, + registered_at: new Date('2026-09-26T10:00:00Z'), + last_seen_at: null, + revoked_at: null, + created_by: systemActorId, + }, + ], + }; + } + return { rows: [] }; + }); + + const service = new SyncService(database, systemActorId); + const device = await service.authenticateDevice(rawToken, tenantA); + + expect(device.id).toBe(deviceId); + expect(device.tenant_id).toBe(tenantA); + expect(device.status).toBe('active'); + expect( + queries.some((q) => q.sql.includes('UPDATE staff_sync_devices') && q.sql.includes('last_seen_at')), + ).toBe(true); + }); + + it('rejects revoked device with 403 DEVICE_REVOKED', async () => { + const rawToken = `kstaff_dev_${tenantA}_${crypto.randomBytes(32).toString('hex')}`; + const tokenHash = SyncService.hashToken(rawToken); + + const { database } = createMockDb((sql) => { + if (sql.includes('SELECT id, tenant_id')) { + return { + rows: [ + { + id: newUlid(), + tenant_id: tenantA, + device_code: 'TERM-01', + name: 'Terminal 1', + token_hash: tokenHash, + token_hint: 'hint', + status: 'revoked', + allowed_scopes: ['attendance:sync'], + org_unit_id: null, + registered_at: new Date(), + last_seen_at: null, + revoked_at: new Date(), + created_by: systemActorId, + }, + ], + }; + } + return { rows: [] }; + }); + + const service = new SyncService(database, systemActorId); + await expect(service.authenticateDevice(rawToken, tenantA)).rejects.toMatchObject({ + name: 'SyncAuthError', + code: 'DEVICE_REVOKED', + statusCode: 403, + message: 'Устройство отозвано', + }); + }); + + it('prevents reading or authenticating across different tenants (cross-tenant rejection)', async () => { + const rawToken = `kstaff_dev_${tenantA}_${crypto.randomBytes(32).toString('hex')}`; + const { database } = createMockDb(); + const service = new SyncService(database, systemActorId); + + // Explicitly providing tenantB for a token configured for tenantA + await expect(service.authenticateDevice(rawToken, tenantB)).rejects.toMatchObject({ + code: 'TENANT_MISMATCH', + statusCode: 403, + }); + }); + + it('rejects device without required sync scope', async () => { + const rawToken = `kstaff_dev_${tenantA}_${crypto.randomBytes(32).toString('hex')}`; + const { database } = createMockDb((sql) => { + if (sql.includes('SELECT id, tenant_id')) { + return { + rows: [ + { + id: newUlid(), + tenant_id: tenantA, + device_code: 'TERM-VIEWER', + name: 'Viewer', + token_hash: SyncService.hashToken(rawToken), + token_hint: 'hint', + status: 'active', + allowed_scopes: ['display:read_only'], + org_unit_id: null, + registered_at: new Date(), + last_seen_at: null, + revoked_at: null, + created_by: systemActorId, + }, + ], + }; + } + return { rows: [] }; + }); + + const service = new SyncService(database, systemActorId); + await expect(service.authenticateDevice(rawToken, tenantA, 'attendance:sync')).rejects.toMatchObject({ + code: 'SCOPE_INSUFFICIENT', + statusCode: 403, + }); + }); +}); + +describe('Sync Push & Idempotency', () => { + const tenantId = parseUlid('01J0000000000000000000000A'); + const systemActorId = parseUlid('01J00000000000000000000001'); + const deviceId = parseUlid('01J0000000000000000000000D'); + const employeeId = parseUlid('01J0000000000000000000000E'); + + const testDevice = { + id: deviceId, + tenant_id: tenantId, + device_code: 'TERM-01', + name: 'Terminal 1', + token_hash: 'hash', + token_hint: 'hint', + status: 'active' as const, + allowed_scopes: ['attendance:sync'], + org_unit_id: null, + registered_at: new Date().toISOString(), + last_seen_at: null, + revoked_at: null, + created_by: systemActorId, + }; + + it('processes attendance_event and inserts into staff_sync_command_inbox', async () => { + const { database, queries } = createMockDb((sql) => { + if (sql.includes('SELECT status, result, conflict')) { + return { rows: [] }; // Not yet in inbox + } + if (sql.includes('SELECT id FROM staff_employees')) { + return { rows: [{ id: employeeId }] }; + } + return { rows: [] }; + }); + + const service = new SyncService(database, systemActorId); + const command: SyncCommand = { + command_type: 'attendance_event', + schema_version: 1, + idempotency_key: 'cmd-test-1', + correlation_id: newUlid(), + created_at: new Date().toISOString(), + payload: { + employee_id: employeeId, + event_type: 'check_in', + occurred_at: new Date().toISOString(), + source_event_id: 'scanner-evt-001', + }, + }; + + const results = await service.push(tenantId, testDevice, [command]); + expect(results).toHaveLength(1); + expect(results[0]?.status).toBe('applied'); + expect(results[0]?.result).toMatchObject({ status: 'accepted' }); + + expect(queries.some((q) => q.sql.includes('INSERT INTO staff_attendance_events'))).toBe(true); + expect(queries.some((q) => q.sql.includes('INSERT INTO staff_sync_command_inbox'))).toBe(true); + expect(queries.some((q) => q.sql.includes('INSERT INTO staff_sync_change_log'))).toBe(true); + }); + + it('returns cached result on repeated idempotency_key without re-executing', async () => { + const cachedResult = { event_id: newUlid(), status: 'accepted' }; + + const { database, queries } = createMockDb((sql) => { + if (sql.includes('SELECT status, result, conflict')) { + // Return already processed command + return { + rows: [ + { + status: 'applied', + result: cachedResult, + conflict: null, + error_code: null, + error_message: null, + }, + ], + }; + } + return { rows: [] }; + }); + + const service = new SyncService(database, systemActorId); + const command: SyncCommand = { + command_type: 'attendance_event', + schema_version: 1, + idempotency_key: 'cmd-duplicate', + correlation_id: newUlid(), + created_at: new Date().toISOString(), + payload: { + employee_id: employeeId, + event_type: 'check_in', + occurred_at: new Date().toISOString(), + source_event_id: 'scanner-evt-001', + }, + }; + + const results = await service.push(tenantId, testDevice, [command]); + expect(results).toHaveLength(1); + expect(results[0]?.status).toBe('applied'); + expect(results[0]?.result).toEqual(cachedResult); + + // Did NOT execute insert into staff_attendance_events again! + expect(queries.some((q) => q.sql.includes('INSERT INTO staff_attendance_events'))).toBe(false); + }); + + it('returns structured conflict on revision mismatch for work_session', async () => { + const sessionId = newUlid(); + + const { database } = createMockDb((sql) => { + if (sql.includes('SELECT status, result, conflict')) { + return { rows: [] }; // Not in inbox + } + if (sql.includes('SELECT id, revision, status FROM staff_work_sessions')) { + // Server has revision 3 + return { rows: [{ id: sessionId, revision: 3, status: 'confirmed' }] }; + } + return { rows: [] }; + }); + + const service = new SyncService(database, systemActorId); + const command: SyncCommand = { + command_type: 'upsert_work_session', + schema_version: 1, + idempotency_key: 'cmd-session-conflict', + correlation_id: newUlid(), + created_at: new Date().toISOString(), + payload: { + id: sessionId, + employee_id: employeeId, + org_unit_id: newUlid(), + business_date: '2026-09-26', + revision: 2, // Client expects revision 2, but server has 3 + worked_minutes: 480, + }, + }; + + const results = await service.push(tenantId, testDevice, [command]); + expect(results).toHaveLength(1); + expect(results[0]?.status).toBe('conflict'); + expect(results[0]?.conflict).toEqual({ + conflict_code: 'REVISION_MISMATCH', + entity_type: 'work_session', + entity_id: sessionId, + expected_revision: 2, + actual_revision: 3, + safe_description: 'Ревизия рабочей смены на сервере изменилась', + client_action: 'FETCH_LATEST_AND_RETRY', + }); + }); + + it('returns structured conflict on revision mismatch for timesheet_entry', async () => { + const entryId = newUlid(); + + const { database } = createMockDb((sql) => { + if (sql.includes('SELECT status, result, conflict')) { + return { rows: [] }; + } + if (sql.includes('SELECT id, revision, status FROM staff_timesheet_entries')) { + return { rows: [{ id: entryId, revision: 4, status: 'approved' }] }; + } + return { rows: [] }; + }); + + const service = new SyncService(database, systemActorId); + const command: SyncCommand = { + command_type: 'upsert_timesheet_entry', + schema_version: 1, + idempotency_key: 'cmd-ts-conflict', + correlation_id: newUlid(), + created_at: new Date().toISOString(), + payload: { + id: entryId, + employee_id: employeeId, + work_date: '2026-09-26', + minutes_worked: 500, + revision: 1, // Mismatch: expects 1, server has 4 + }, + }; + + const results = await service.push(tenantId, testDevice, [command]); + expect(results).toHaveLength(1); + expect(results[0]?.status).toBe('conflict'); + expect(results[0]?.conflict).toEqual({ + conflict_code: 'REVISION_MISMATCH', + entity_type: 'timesheet_entry', + entity_id: entryId, + expected_revision: 1, + actual_revision: 4, + safe_description: 'Ревизия записи табеля на сервере изменилась', + client_action: 'FETCH_LATEST_AND_RETRY', + }); + }); + + it('enforces batch size limits and payload size limits', async () => { + const { database } = createMockDb(); + const service = new SyncService(database, systemActorId); + + // Empty batch + await expect(service.push(tenantId, testDevice, [])).rejects.toMatchObject({ + code: 'BATCH_EMPTY', + statusCode: 400, + }); + + // Batch > 100 + const largeBatch: SyncCommand[] = Array.from({ length: 101 }, (_, i) => ({ + command_type: 'attendance_event', + schema_version: 1, + idempotency_key: `key-${i}`, + correlation_id: newUlid(), + created_at: new Date().toISOString(), + payload: {}, + })); + await expect(service.push(tenantId, testDevice, largeBatch)).rejects.toMatchObject({ + code: 'BATCH_SIZE_EXCEEDED', + statusCode: 400, + }); + + // Single payload > 64KB + const giantPayload = { data: 'x'.repeat(70000) }; + const commandWithGiantPayload: SyncCommand = { + command_type: 'attendance_event', + schema_version: 1, + idempotency_key: 'giant-payload', + correlation_id: newUlid(), + created_at: new Date().toISOString(), + payload: giantPayload, + }; + const results = await service.push(tenantId, testDevice, [commandWithGiantPayload]); + expect(results[0]?.status).toBe('rejected'); + expect(results[0]?.error_code).toBe('PAYLOAD_TOO_LARGE'); + }); + + it('rejects commands with timestamps outside the replay window', async () => { + const { database } = createMockDb(); + const service = new SyncService(database, systemActorId); + + // 40 days old (> 30 days) + const oldDate = new Date(Date.now() - 40 * 24 * 60 * 60 * 1000).toISOString(); + const oldCmd: SyncCommand = { + command_type: 'attendance_event', + schema_version: 1, + idempotency_key: 'cmd-old', + correlation_id: newUlid(), + created_at: oldDate, + payload: {}, + }; + + // 10 minutes in future (> 5 minutes) + const futureDate = new Date(Date.now() + 10 * 60 * 1000).toISOString(); + const futureCmd: SyncCommand = { + command_type: 'attendance_event', + schema_version: 1, + idempotency_key: 'cmd-future', + correlation_id: newUlid(), + created_at: futureDate, + payload: {}, + }; + + const results = await service.push(tenantId, testDevice, [oldCmd, futureCmd]); + expect(results[0]?.status).toBe('rejected'); + expect(results[0]?.error_code).toBe('TIMESTAMP_OUT_OF_WINDOW'); + expect(results[1]?.status).toBe('rejected'); + expect(results[1]?.error_code).toBe('TIMESTAMP_OUT_OF_WINDOW'); + }); +}); + +describe('Sync Pull & Bootstrap Data Safety', () => { + const tenantId = parseUlid('01J0000000000000000000000A'); + const systemActorId = parseUlid('01J00000000000000000000001'); + const deviceId = parseUlid('01J0000000000000000000000D'); + + it('bootstrap (cursor = null) returns safe entities without salary, PII, or biometrics', async () => { + const empId = newUlid(); + const orgId = newUlid(); + const profId = newUlid(); + + const { database } = createMockDb((sql) => { + if (sql.includes('MAX(cursor_seq)')) { + return { rows: [{ max_seq: '50' }] }; + } + if (sql.includes('FROM staff_org_units')) { + return { rows: [{ id: orgId, code: 'KITCHEN', name: 'Кухня', status: 'active' }] }; + } + if (sql.includes('FROM staff_professions')) { + return { rows: [{ id: profId, code: 'CHEF', name: 'Шеф', status: 'active' }] }; + } + if (sql.includes('FROM staff_employees')) { + return { + rows: [ + { + id: empId, + personnel_number: 'EMP-01', + display_name: 'Петров Петр', + status: 'active', + profession_id: profId, + position: 'Шеф-повар', + }, + ], + }; + } + return { rows: [] }; + }); + + const service = new SyncService(database, systemActorId); + const result = await service.pull(tenantId, deviceId, { cursor: null }); + + expect(result.changes).toHaveLength(3); + expect(result.next_cursor).toBe(encodeCursor(tenantId, 50n)); + expect(result.has_more).toBe(false); + + // Verify employee payload has NO sensitive fields + const empChange = result.changes.find((c) => c.entity_type === 'employee'); + expect(empChange).toBeDefined(); + const payload = empChange!.payload; + expect(payload).toHaveProperty('personnel_number'); + expect(payload).toHaveProperty('display_name'); + expect(payload).not.toHaveProperty('salary'); + expect(payload).not.toHaveProperty('hourly_rate'); + expect(payload).not.toHaveProperty('passport'); + expect(payload).not.toHaveProperty('phone'); + expect(payload).not.toHaveProperty('biometrics'); + expect(payload).not.toHaveProperty('template'); + }); + + it('incremental pull returns changes strictly after cursor', async () => { + const cursor = encodeCursor(tenantId, 10n); + + const { database } = createMockDb((sql) => { + if (sql.includes('FROM staff_sync_change_log')) { + return { + rows: [ + { + id: newUlid(), + cursor_seq: '11', + entity_type: 'employee', + entity_id: newUlid(), + operation: 'upsert', + revision: 2, + payload: { id: 'test', display_name: 'Тест' }, + created_at: new Date('2026-09-26T12:00:00Z'), + }, + ], + }; + } + return { rows: [] }; + }); + + const service = new SyncService(database, systemActorId); + const result = await service.pull(tenantId, deviceId, { cursor }); + + expect(result.changes).toHaveLength(1); + expect(result.changes[0]?.cursor).toBe(encodeCursor(tenantId, 11n)); + expect(result.next_cursor).toBe(encodeCursor(tenantId, 11n)); + expect(result.has_more).toBe(false); + }); +}); + +describe('Sync API HTTP Endpoints', () => { + const tenantId = '01J00000000000000000000000'; + const managerToken = 'a'.repeat(64); + + const config = loadConfig({ + DATABASE_URL: 'postgres://localhost/staff', + STANDALONE_TENANT_ID: tenantId, + STANDALONE_ACTOR_ID: '01J00000000000000000000001', + STANDALONE_MANAGER_TOKEN: managerToken, + STANDALONE_TERMINAL_TOKEN: 'b'.repeat(64), + LOG_LEVEL: 'silent', + NODE_ENV: 'test', + }); + + it('registers device with manager authorization', async () => { + const deviceId = newUlid(); + const { database } = createMockDb((sql) => { + if (sql.includes('INSERT INTO staff_sync_devices')) { + return { + rows: [ + { + id: deviceId, + tenant_id: tenantId, + device_code: 'TERM-API-01', + name: 'Касса 1', + token_hint: 'kstaff_dev_...', + status: 'active', + allowed_scopes: ['attendance:sync'], + org_unit_id: null, + registered_at: new Date(), + created_by: config.STANDALONE_ACTOR_ID, + }, + ], + }; + } + return { rows: [] }; + }); + + const app = await buildApp({ config, database }); + await app.ready(); + + const response = await app.inject({ + method: 'POST', + url: '/api/v1/staff/sync/devices', + headers: { + authorization: `Bearer ${managerToken}`, + }, + payload: { + device_code: 'TERM-API-01', + name: 'Касса 1', + }, + }); + + expect(response.statusCode).toBe(201); + const body = response.json<{ device: { device_code: string }; token: string }>(); + expect(body.device.device_code).toBe('TERM-API-01'); + expect(body.token).toMatch(/^kstaff_dev_/); + + await app.close(); + }); + + it('rejects unauthenticated requests to device sync endpoints', async () => { + const { database } = createMockDb(); + const app = await buildApp({ config, database }); + await app.ready(); + + const response = await app.inject({ + method: 'GET', + url: '/api/v1/staff/sync/status', + }); + + expect(response.statusCode).toBe(401); + await app.close(); + }); + + it('handles checkpoint acknowledgment', async () => { + const rawToken = `kstaff_dev_${tenantId}_${crypto.randomBytes(32).toString('hex')}`; + const deviceId = newUlid(); + const cursor = encodeCursor(tenantId, 55n); + + const { database } = createMockDb((sql) => { + if (sql.includes('FROM staff_sync_devices')) { + return { + rows: [ + { + id: deviceId, + tenant_id: tenantId, + device_code: 'TERM-01', + name: 'Term', + token_hash: SyncService.hashToken(rawToken), + token_hint: 'hint', + status: 'active', + allowed_scopes: ['attendance:sync'], + org_unit_id: null, + registered_at: new Date(), + last_seen_at: null, + revoked_at: null, + created_by: config.STANDALONE_ACTOR_ID, + }, + ], + }; + } + if (sql.includes('INSERT INTO staff_sync_device_checkpoints')) { + return { rows: [{ acknowledged_at: new Date('2026-09-26T12:00:00Z') }] }; + } + return { rows: [] }; + }); + + const app = await buildApp({ config, database }); + await app.ready(); + + const response = await app.inject({ + method: 'POST', + url: '/api/v1/staff/sync/ack', + headers: { + authorization: `Bearer ${rawToken}`, + }, + payload: { + cursor, + }, + }); + + expect(response.statusCode).toBe(200); + const body = response.json<{ acknowledged: boolean; cursor: string }>(); + expect(body.acknowledged).toBe(true); + expect(body.cursor).toBe(cursor); + + await app.close(); + }); +}); + +describe('Security & Privacy Audit (No secrets, no PII, no biometrics in logs or state)', () => { + const tenantId = parseUlid('01J0000000000000000000000A'); + const actorId = parseUlid('01J00000000000000000000001'); + + it('guarantees raw device token is never stored in the database, only SHA-256 hash', async () => { + const { database, queries } = createMockDb((sql) => { + if (sql.includes('INSERT INTO staff_sync_devices')) { + return { + rows: [ + { + id: newUlid(), + tenant_id: tenantId, + device_code: 'DEV-SEC', + name: 'Secure Terminal', + token_hint: 'kstaff_dev_...', + status: 'active', + allowed_scopes: ['attendance:sync'], + org_unit_id: null, + registered_at: new Date(), + created_by: actorId, + }, + ], + }; + } + return { rows: [] }; + }); + + const service = new SyncService(database, actorId); + const regResult = await service.registerDevice(tenantId, actorId, { + device_code: 'DEV-SEC', + name: 'Secure Terminal', + }); + + const rawToken = regResult.token; + expect(rawToken).toMatch(/^kstaff_dev_/); + + // Verify SQL query parameters + const insertQuery = queries.find((q) => q.sql.includes('INSERT INTO staff_sync_devices')); + expect(insertQuery).toBeDefined(); + + // Raw token MUST NOT be present in parameters + const paramsString = JSON.stringify(insertQuery!.params); + expect(paramsString).not.toContain(rawToken); + + // Only the hash should be stored + const expectedHash = SyncService.hashToken(rawToken); + expect(paramsString).toContain(expectedHash); + }); + + it('sanitizes and excludes sensitive employee attributes from sync payloads', async () => { + const { database } = createMockDb((sql) => { + if (sql.includes('MAX(cursor_seq)')) { + return { rows: [{ max_seq: '1' }] }; + } + if (sql.includes('FROM staff_employees')) { + return { + rows: [ + { + id: newUlid(), + personnel_number: 'EMP-99', + display_name: 'Секретный Сотрудник', + status: 'active', + profession_id: null, + position: 'Аналитик', + }, + ], + }; + } + return { rows: [] }; + }); + + const service = new SyncService(database, actorId); + const result = await service.pull(tenantId, newUlid(), { cursor: null }); + + const serialized = JSON.stringify(result); + // Explicit forbidden terms check + const forbiddenTerms = [ + 'salary', + 'wage', + 'rate', + 'passport', + 'snils', + 'inn', + 'bank_account', + 'card_number', + 'phone', + 'email', + 'biometrics', + 'fingerprint', + 'face_descriptor', + ]; + for (const term of forbiddenTerms) { + expect(serialized.toLowerCase()).not.toContain(`"${term}"`); + } + }); +}); diff --git a/tests/unit/tenant-sql.test.ts b/tests/unit/tenant-sql.test.ts index 598797b..18e7ab7 100644 --- a/tests/unit/tenant-sql.test.ts +++ b/tests/unit/tenant-sql.test.ts @@ -180,4 +180,54 @@ describe('employee push notifications migration', () => { }); }); +describe('offline sync backend migration', () => { + const sql = readFileSync('migrations/0011_offline_sync_backend.up.sql', 'utf8'); + const tables = [ + 'staff_sync_devices', + 'staff_sync_change_log', + 'staff_sync_command_inbox', + 'staff_sync_device_checkpoints', + ]; + + it.each(tables)('%s is tenant scoped and forces RLS', (table) => { + const create = sql.slice( + sql.indexOf(`CREATE TABLE ${table}`), + sql.indexOf(';', sql.indexOf(`CREATE TABLE ${table}`)), + ); + expect(create).toContain('tenant_id VARCHAR(26) NOT NULL'); + expect(sql).toContain(`ALTER TABLE ${table} ENABLE ROW LEVEL SECURITY;`); + expect(sql).toContain(`ALTER TABLE ${table} FORCE ROW LEVEL SECURITY;`); + expect(sql).toContain(`CREATE POLICY ${table}_tenant_isolation`); + expect(sql).toContain("tenant_id = current_setting('app.tenant_id', true)"); + }); + + it('declares uniqueness, cursor sequencing and idempotency constraints with tenant scope', () => { + expect(sql).toContain('UNIQUE (tenant_id, id)'); + expect(sql).toContain('UNIQUE (tenant_id, device_code)'); + expect(sql).toContain('UNIQUE (tenant_id, token_hash)'); + expect(sql).toContain('UNIQUE (tenant_id, device_id, idempotency_key)'); + expect(sql).toContain('PRIMARY KEY (tenant_id, device_id)'); + expect(sql).toContain('cursor_seq BIGSERIAL NOT NULL'); + }); + + it('declares foreign key cascades for device ownership', () => { + expect(sql).toContain( + 'FOREIGN KEY (tenant_id, device_id) REFERENCES staff_sync_devices (tenant_id, id) ON DELETE CASCADE', + ); + }); + + it('grants required operations on all sync tables and sequence to krona_staff_app', () => { + expect(sql).toContain('GRANT SELECT, INSERT, UPDATE, DELETE ON staff_sync_devices TO krona_staff_app;'); + expect(sql).toContain('GRANT SELECT, INSERT, UPDATE, DELETE ON staff_sync_change_log TO krona_staff_app;'); + expect(sql).toContain('GRANT USAGE, SELECT ON SEQUENCE staff_sync_change_log_cursor_seq_seq TO krona_staff_app;'); + expect(sql).toContain('GRANT SELECT, INSERT, UPDATE, DELETE ON staff_sync_command_inbox TO krona_staff_app;'); + expect(sql).toContain('GRANT SELECT, INSERT, UPDATE, DELETE ON staff_sync_device_checkpoints TO krona_staff_app;'); + }); + + it('declares and grants retention cleanup function', () => { + expect(sql).toContain('CREATE OR REPLACE FUNCTION staff_sync_cleanup_retention'); + expect(sql).toContain('GRANT EXECUTE ON FUNCTION staff_sync_cleanup_retention(INTEGER) TO krona_staff_app;'); + }); +}); +