From 8baad554c19b2c0e0a82c6c4a2835e44438d430d Mon Sep 17 00:00:00 2001 From: Ochenstarik Date: Sat, 26 Sep 2026 15:33:43 +0700 Subject: [PATCH] fix(sync): compute org unit status dynamically in bootstrap pull MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Таблица staff_org_units не имеет отдельной колонки status, активность определяется полями is_active и deleted_at. Прямой запрос status при начальной загрузке без курсора приводил к 500 ошибке (дефект SYNC-C0). Статус вычисляется через CASE-выражение. Очистка интеграционного теста выполняется ролью мигратора. --- src/domain/sync/service.ts | 204 +++++++++++++++++++---- tests/integration/sync-bootstrap.test.ts | 171 +++++++++++++++++++ tests/unit/sync.test.ts | 81 ++++++++- 3 files changed, 415 insertions(+), 41 deletions(-) create mode 100644 tests/integration/sync-bootstrap.test.ts diff --git a/src/domain/sync/service.ts b/src/domain/sync/service.ts index ab4f20c..7b8e3b2 100644 --- a/src/domain/sync/service.ts +++ b/src/domain/sync/service.ts @@ -90,7 +90,11 @@ export class SyncService { const tokenTenant = SyncService.parseTenantFromToken(rawToken); if (providedTenantId && tokenTenant && providedTenantId !== tokenTenant) { - throw new SyncAuthError('Несоответствие арендатора в токене устройства', 'TENANT_MISMATCH', 403); + throw new SyncAuthError( + 'Несоответствие арендатора в токене устройства', + 'TENANT_MISMATCH', + 403, + ); } const tenantId = providedTenantId ?? tokenTenant; @@ -187,9 +191,10 @@ export class SyncService { 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 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; @@ -210,7 +215,17 @@ export class SyncService { (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], + [ + deviceId, + tenantId, + params.device_code.trim(), + params.name.trim(), + tokenHash, + tokenHint, + scopes, + orgUnitId, + actorId, + ], ); const row = res.rows[0]!; @@ -319,14 +334,32 @@ export class SyncService { }); } - async pull(tenantId: Ulid, deviceId: Ulid, options: PullOptions = {}): Promise { + async pull(tenantId: Ulid, options?: PullOptions): Promise; + async pull(tenantId: Ulid, deviceId: Ulid, options?: PullOptions): Promise; + async pull( + tenantId: Ulid, + deviceIdOrOptions?: Ulid | PullOptions, + maybeOptions: PullOptions = {}, + ): Promise { + 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 filterTypes = + options.entity_types && options.entity_types.length > 0 ? options.entity_types : null; const res = await client.query<{ id: string; @@ -360,9 +393,8 @@ export class SyncService { created_at: r.created_at.toISOString(), })); - const nextCursor = changes.length > 0 - ? changes[changes.length - 1]!.cursor - : options.cursor; + const nextCursor = + changes.length > 0 ? changes[changes.length - 1]!.cursor : options.cursor; return { changes, @@ -380,14 +412,29 @@ export class SyncService { const latestCursor = encodeCursor(tenantId, maxSeq); const wanted = (type: string) => - !options.entity_types || options.entity_types.length === 0 || options.entity_types.includes(type); + !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`, + 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) { @@ -411,7 +458,12 @@ export class SyncService { // Professions if (wanted('profession')) { - const profRes = await client.query<{ id: string; code: string; name: string; status: string }>( + 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], ); @@ -662,7 +714,8 @@ export class SyncService { command_type: cmd.command_type, status: 'rejected', error_code: 'TIMESTAMP_OUT_OF_WINDOW', - error_message: 'Временная метка команды находится вне допустимого окна (-30 дней .. +5 минут)', + error_message: + 'Временная метка команды находится вне допустимого окна (-30 дней .. +5 минут)', }; await this.saveInboxCommand(client, tenantId, device.id, cmd, rejectedResult); results.push(rejectedResult); @@ -757,7 +810,17 @@ export class SyncService { (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], + [ + eventId, + tenantId, + employeeId, + eventType, + sourceEventId, + occurredAt, + device.id, + device.id, + correlationId, + ], ); const safePayload: SafeAttendanceEventPayload = { @@ -771,7 +834,15 @@ export class SyncService { status: 'accepted', }; - await this.recordChange(client, tenantId, 'attendance_event', eventId, 'upsert', 1, safePayload); + await this.recordChange( + client, + tenantId, + 'attendance_event', + eventId, + 'upsert', + 1, + safePayload, + ); return { idempotency_key: cmd.idempotency_key, @@ -789,7 +860,8 @@ export class SyncService { 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 workedMinutes = + typeof payload.worked_minutes === 'number' ? payload.worked_minutes : 0; const sessionStatus = (payload.status as string) || 'open'; if (!employeeId || !orgUnitId || !businessDate) { @@ -832,7 +904,15 @@ export class SyncService { `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], + [ + startedAt ?? null, + endedAt ?? null, + workedMinutes, + sessionStatus, + nextRev, + tenantId, + sessionId, + ], ); const safePayload: SafeWorkSessionPayload = { @@ -847,7 +927,15 @@ export class SyncService { revision: nextRev, }; - await this.recordChange(client, tenantId, 'work_session', sessionId, 'upsert', nextRev, safePayload); + await this.recordChange( + client, + tenantId, + 'work_session', + sessionId, + 'upsert', + nextRev, + safePayload, + ); return { idempotency_key: cmd.idempotency_key, @@ -863,7 +951,18 @@ export class SyncService { `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], + [ + sessionId, + tenantId, + employeeId, + orgUnitId, + businessDate, + startedAt ?? null, + endedAt ?? null, + workedMinutes, + sessionStatus, + newRev, + ], ); const safePayload: SafeWorkSessionPayload = { @@ -878,7 +977,15 @@ export class SyncService { revision: newRev, }; - await this.recordChange(client, tenantId, 'work_session', sessionId, 'upsert', newRev, safePayload); + await this.recordChange( + client, + tenantId, + 'work_session', + sessionId, + 'upsert', + newRev, + safePayload, + ); return { idempotency_key: cmd.idempotency_key, @@ -892,7 +999,8 @@ export class SyncService { 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 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; @@ -948,7 +1056,15 @@ export class SyncService { revision: nextRev, }; - await this.recordChange(client, tenantId, 'timesheet_entry', entryId, 'upsert', nextRev, safePayload); + await this.recordChange( + client, + tenantId, + 'timesheet_entry', + entryId, + 'upsert', + nextRev, + safePayload, + ); return { idempotency_key: cmd.idempotency_key, @@ -975,7 +1091,15 @@ export class SyncService { revision: newRev, }; - await this.recordChange(client, tenantId, 'timesheet_entry', entryId, 'upsert', newRev, safePayload); + await this.recordChange( + client, + tenantId, + 'timesheet_entry', + entryId, + 'upsert', + newRev, + safePayload, + ); return { idempotency_key: cmd.idempotency_key, @@ -1052,17 +1176,23 @@ export class SyncService { ); } - 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), - }; - }); + 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/tests/integration/sync-bootstrap.test.ts b/tests/integration/sync-bootstrap.test.ts new file mode 100644 index 0000000..74a4ac6 --- /dev/null +++ b/tests/integration/sync-bootstrap.test.ts @@ -0,0 +1,171 @@ +import pg from 'pg'; +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import { Database } from '../../src/infrastructure/database.js'; +import { newUlid, parseUlid } from '../../src/core/ids.js'; +import { SyncService } from '../../src/domain/sync/service.js'; +import { buildApp } from '../../src/app.js'; +import { loadConfig } from '../../src/config.js'; + +const connectionString = process.env.DATABASE_TEST_URL ?? process.env.TEST_DATABASE_URL; +const migrationConnectionString = + process.env.DATABASE_TEST_MIGRATION_URL ?? process.env.DATABASE_MIGRATION_URL; +const describeDatabase = connectionString ? describe : describe.skip; + +const cleanupTables = [ + 'staff_sync_device_checkpoints', + 'staff_sync_command_inbox', + 'staff_sync_change_log', + 'staff_sync_devices', + 'staff_timesheet_entries', + 'staff_work_sessions', + 'staff_attendance_events', + 'staff_employees', + 'staff_org_unit_terms', + 'staff_org_units', +] as const; + +describeDatabase('Sync Bootstrap SYNC-C0 (Live PostgreSQL)', () => { + const database = new Database(connectionString!); + const tenantId = parseUlid('01J0000000000000000000000A'); + const actorId = parseUlid('01J00000000000000000000001'); + const orgUnitId = newUlid(); + const employeeId = newUlid(); + const deviceId = newUlid(); + const rawToken = `kstaff_dev_${tenantId}_bootstrap_c0_test_token_secret`; + const tokenHash = SyncService.hashToken(rawToken); + + const syncService = new SyncService(database, actorId); + + async function cleanupAsMigrator() { + const migratorUrl = migrationConnectionString ?? connectionString; + if (!migratorUrl) { + return; + } + const migrator = new pg.Client({ + connectionString: migratorUrl, + application_name: 'krona-staff-bootstrap-test-cleanup', + }); + await migrator.connect(); + try { + await migrator.query('BEGIN'); + try { + await migrator.query("SELECT set_config('app.tenant_id', $1, true)", [tenantId]); + for (const table of cleanupTables) { + await migrator.query(`DELETE FROM ${table} WHERE tenant_id = $1`, [tenantId]); + } + await migrator.query('COMMIT'); + } catch (error) { + await migrator.query('ROLLBACK'); + throw error; + } + } finally { + await migrator.end(); + } + } + + beforeAll(async () => { + await cleanupAsMigrator(); + await database.tenantTransaction(tenantId, actorId, async (client) => { + // 1. Insert real org unit (columns: id, tenant_id, code, name, is_active, created_at, updated_at) + await client.query( + `INSERT INTO staff_org_units (id, tenant_id, code, name, is_active, created_at, updated_at) + VALUES ($1, $2, 'bootstrap_point', 'Точка Bootstrap', true, now(), now())`, + [orgUnitId, tenantId], + ); + + // 2. Insert employee in staff_employees + await client.query( + `INSERT INTO staff_employees (id, tenant_id, personnel_number, display_name, status, primary_org_unit_id, created_at, updated_at) + VALUES ($1, $2, 'EMP-BOOT-01', 'Сотрудник Bootstrap', 'active', $3, now(), now())`, + [employeeId, tenantId, orgUnitId], + ); + + // 3. Insert sync device for HTTP sync testing + await client.query( + `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, 'TERM-BOOT-01', 'Терминал Bootstrap', $3, 'kstaff_dev_...', 'active', ARRAY['attendance:sync']::text[], $4, $5)`, + [deviceId, tenantId, tokenHash, orgUnitId, actorId], + ); + }); + }); + + afterAll(async () => { + try { + await cleanupAsMigrator(); + } finally { + await database.close(); + } + }); + + it('syncService.pull(tenantId, { cursor: null }) succeeds without "column status does not exist" and returns org_unit and employee changes', async () => { + const result = await syncService.pull(tenantId, { cursor: null }); + + expect(result).toBeDefined(); + expect(result.has_more).toBe(false); + + // Verify org_unit change is present with mapped status + const orgUnitChange = result.changes.find( + (c) => c.entity_type === 'org_unit' && c.entity_id === orgUnitId, + ); + expect(orgUnitChange).toBeDefined(); + expect(orgUnitChange?.payload).toMatchObject({ + id: orgUnitId, + code: 'bootstrap_point', + name: 'Точка Bootstrap', + status: 'active', + }); + + // Verify employee change is present + const employeeChange = result.changes.find( + (c) => c.entity_type === 'employee' && c.entity_id === employeeId, + ); + expect(employeeChange).toBeDefined(); + expect(employeeChange?.payload).toMatchObject({ + id: employeeId, + personnel_number: 'EMP-BOOT-01', + display_name: 'Сотрудник Bootstrap', + status: 'active', + }); + }); + + it('HTTP /api/v1/staff/sync/pull without cursor returns 200 and includes org_unit and employee in changes', async () => { + const config = loadConfig({ + DATABASE_URL: connectionString!, + STANDALONE_TENANT_ID: tenantId, + STANDALONE_ACTOR_ID: actorId, + LOG_LEVEL: 'silent', + NODE_ENV: 'test', + }); + const app = await buildApp({ config, database }); + try { + const response = await app.inject({ + method: 'POST', + url: '/api/v1/staff/sync/pull', + headers: { + authorization: `Bearer ${rawToken}`, + }, + payload: { cursor: null }, + }); + + expect(response.statusCode).toBe(200); + const body = response.json<{ + changes: Array<{ + entity_type: string; + entity_id: string; + payload: Record; + }>; + next_cursor: string; + has_more: boolean; + }>(); + + expect( + body.changes.some((c) => c.entity_type === 'org_unit' && c.entity_id === orgUnitId), + ).toBe(true); + expect( + body.changes.some((c) => c.entity_type === 'employee' && c.entity_id === employeeId), + ).toBe(true); + } finally { + await app.close(); + } + }); +}); diff --git a/tests/unit/sync.test.ts b/tests/unit/sync.test.ts index 5f8826e..3a7fc6a 100644 --- a/tests/unit/sync.test.ts +++ b/tests/unit/sync.test.ts @@ -8,7 +8,9 @@ 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 }) { +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 }); @@ -105,7 +107,9 @@ describe('Sync Device Authentication & Tenant Isolation', () => { 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')), + queries.some( + (q) => q.sql.includes('UPDATE staff_sync_devices') && q.sql.includes('last_seen_at'), + ), ).toBe(true); }); @@ -187,7 +191,9 @@ describe('Sync Device Authentication & Tenant Isolation', () => { }); const service = new SyncService(database, systemActorId); - await expect(service.authenticateDevice(rawToken, tenantA, 'attendance:sync')).rejects.toMatchObject({ + await expect( + service.authenticateDevice(rawToken, tenantA, 'attendance:sync'), + ).rejects.toMatchObject({ code: 'SCOPE_INSUFFICIENT', statusCode: 403, }); @@ -468,7 +474,7 @@ describe('Sync Pull & Bootstrap Data Safety', () => { const orgId = newUlid(); const profId = newUlid(); - const { database } = createMockDb((sql) => { + const { database, queries } = createMockDb((sql) => { if (sql.includes('MAX(cursor_seq)')) { return { rows: [{ max_seq: '50' }] }; } @@ -502,6 +508,23 @@ describe('Sync Pull & Bootstrap Data Safety', () => { expect(result.next_cursor).toBe(encodeCursor(tenantId, 50n)); expect(result.has_more).toBe(false); + // Verify staff_org_units query uses CASE expression for status + const orgQuery = queries.find((q) => q.sql.includes('FROM staff_org_units')); + expect(orgQuery).toBeDefined(); + expect(orgQuery!.sql).toMatch( + /CASE[\s\S]+WHEN deleted_at IS NOT NULL THEN 'archived'[\s\S]+WHEN NOT is_active THEN 'inactive'[\s\S]+ELSE 'active'[\s\S]+END AS status/, + ); + + // Verify SafeOrgUnitPayload + const orgChange = result.changes.find((c) => c.entity_type === 'org_unit'); + expect(orgChange).toBeDefined(); + expect(orgChange!.payload).toEqual({ + id: orgId, + code: 'KITCHEN', + name: 'Кухня', + status: 'active', + }); + // Verify employee payload has NO sensitive fields const empChange = result.changes.find((c) => c.entity_type === 'employee'); expect(empChange).toBeDefined(); @@ -516,6 +539,56 @@ describe('Sync Pull & Bootstrap Data Safety', () => { expect(payload).not.toHaveProperty('template'); }); + it('bootstrap sync selects staff_org_units status via CASE expression and forms SafeOrgUnitPayload', async () => { + const orgActiveId = newUlid(); + const orgInactiveId = newUlid(); + const orgArchivedId = newUlid(); + + const { database, queries } = createMockDb((sql) => { + if (sql.includes('MAX(cursor_seq)')) { + return { rows: [{ max_seq: '10' }] }; + } + if (sql.includes('FROM staff_org_units')) { + return { + rows: [ + { id: orgActiveId, code: 'point_active', name: 'Точка Активная', status: 'active' }, + { + id: orgInactiveId, + code: 'point_inactive', + name: 'Точка Неактивная', + status: 'inactive', + }, + { + id: orgArchivedId, + code: 'point_archived', + name: 'Точка В Архиве', + status: 'archived', + }, + ], + }; + } + return { rows: [] }; + }); + + const service = new SyncService(database, systemActorId); + const result = await service.pull(tenantId, { cursor: null, entity_types: ['org_unit'] }); + + const orgQuery = queries.find((q) => q.sql.includes('FROM staff_org_units')); + expect(orgQuery).toBeDefined(); + expect(orgQuery!.sql).toContain('CASE'); + expect(orgQuery!.sql).toContain("WHEN deleted_at IS NOT NULL THEN 'archived'"); + expect(orgQuery!.sql).toContain("WHEN NOT is_active THEN 'inactive'"); + expect(orgQuery!.sql).toContain("ELSE 'active'"); + expect(orgQuery!.sql).toContain('END AS status'); + + expect(result.changes).toHaveLength(3); + expect(result.changes.map((c) => c.payload)).toEqual([ + { id: orgActiveId, code: 'point_active', name: 'Точка Активная', status: 'active' }, + { id: orgInactiveId, code: 'point_inactive', name: 'Точка Неактивная', status: 'inactive' }, + { id: orgArchivedId, code: 'point_archived', name: 'Точка В Архиве', status: 'archived' }, + ]); + }); + it('incremental pull returns changes strictly after cursor', async () => { const cursor = encodeCursor(tenantId, 10n);