1a77a8264b
Таблица staff_org_units не имеет отдельной колонки status, активность определяется полями is_active и deleted_at. Прямой запрос status при начальной загрузке без курсора приводил к 500 ошибке (дефект SYNC-C0). Статус вычисляется через CASE-выражение. Очистка интеграционного теста выполняется ролью мигратора.
953 lines
32 KiB
TypeScript
953 lines
32 KiB
TypeScript
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<SyncDevice> {
|
||
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<SyncDeviceRegistrationResult> {
|
||
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<SyncStatus> {
|
||
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<SyncAckResult> {
|
||
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, options?: PullOptions): Promise<SyncPullResult>;
|
||
async pull(tenantId: Ulid, deviceId: Ulid, options?: PullOptions): Promise<SyncPullResult>;
|
||
async pull(
|
||
tenantId: Ulid,
|
||
deviceIdOrOptions?: Ulid | PullOptions,
|
||
maybeOptions: PullOptions = {},
|
||
): Promise<SyncPullResult> {
|
||
let deviceId: Ulid;
|
||
let options: PullOptions;
|
||
|
||
if (typeof deviceIdOrOptions === 'string') {
|
||
deviceId = deviceIdOrOptions;
|
||
options = maybeOptions ?? {};
|
||
} else {
|
||
deviceId = this.systemActorId;
|
||
options = deviceIdOrOptions ?? {};
|
||
}
|
||
|
||
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<string, unknown>;
|
||
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,
|
||
CASE
|
||
WHEN deleted_at IS NOT NULL THEN 'archived'
|
||
WHEN NOT is_active THEN 'inactive'
|
||
ELSE 'active'
|
||
END AS 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<string, unknown>,
|
||
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<string, unknown>,
|
||
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<string, unknown>,
|
||
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<string, unknown>,
|
||
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<string, unknown>,
|
||
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<string, unknown>,
|
||
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<SyncCommandResult[]> {
|
||
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<string, unknown> | 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<SyncCommandResult> {
|
||
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':
|
||
case 'upsert_timesheet_entry': {
|
||
return {
|
||
idempotency_key: cmd.idempotency_key,
|
||
command_type: cmd.command_type,
|
||
status: 'rejected',
|
||
error_code: 'COMMAND_NOT_SUPPORTED',
|
||
error_message:
|
||
'Команда не поддерживается: прямое создание или изменение смен и табелей с устройств запрещено',
|
||
};
|
||
}
|
||
|
||
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<void> {
|
||
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<void> {
|
||
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),
|
||
};
|
||
},
|
||
);
|
||
}
|
||
}
|