krona-staff/src/sync.ts
Ochenstarik 5ee75177b9 feat(sync): add offline sync backend foundation
- Add PostgreSQL migrations with strict RLS policies (0011)
- Introduce SyncService for handling auth, token-based registration
- Add payload size and replay attack mitigations
- Provide HTTP fastify routes (pull, push, ack, device management)
- Validate openapi contracts and add comprehensive tests
2026-09-26 13:10:27 +07:00

457 lines
15 KiB
TypeScript

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;
}