diff --git a/src/core/outbox.ts b/src/core/outbox.ts new file mode 100644 index 0000000..047daf0 --- /dev/null +++ b/src/core/outbox.ts @@ -0,0 +1,45 @@ +import type { PoolClient } from 'pg'; +import { newUlid, type Ulid } from './ids.js'; + +export interface OutboxEventInput> { + tenantId: Ulid; + eventType: string; + subjectType: string; + subjectId: Ulid; + actorType: 'user' | 'service'; + actorId: Ulid; + payload: TPayload; + correlationId: Ulid; + causationId?: Ulid; + occurredAt?: string; + eventVersion?: number; +} + +export async function insertOutboxEvent>( + client: Pick, + event: OutboxEventInput, +): Promise { + const id = newUlid(); + const occurredAt = event.occurredAt ?? new Date().toISOString(); + await client.query( + `INSERT INTO staff_outbox + (id, tenant_id, event_type, event_version, subject_type, subject_id, + actor_type, actor_id, payload, correlation_id, causation_id, occurred_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, $10, $11, $12)`, + [ + id, + event.tenantId, + event.eventType, + event.eventVersion ?? 1, + event.subjectType, + event.subjectId, + event.actorType, + event.actorId, + JSON.stringify(event.payload), + event.correlationId, + event.causationId ?? null, + occurredAt, + ], + ); + return id; +} diff --git a/src/domain/payroll/service.ts b/src/domain/payroll/service.ts index 4759974..9cd51a6 100644 --- a/src/domain/payroll/service.ts +++ b/src/domain/payroll/service.ts @@ -1,6 +1,7 @@ import type { PoolClient } from 'pg'; import type { RequestContext } from '../../core/context.js'; import { newUlid, parseUlid } from '../../core/ids.js'; +import { insertOutboxEvent } from '../../core/outbox.js'; import { calculateDay, type DayParticipant, type PayrollLine } from './calculator.js'; import { cateringPreset } from './presets.js'; import type { RuleNode } from './rule.js'; @@ -767,21 +768,20 @@ export async function calculatePayrollPeriod( [tenantId, runId, totalAccruedMinor.toString(), calculatedAt], ); - await client.query( - `INSERT INTO staff_outbox - (id, tenant_id, event_type, event_version, subject_type, subject_id, - actor_type, actor_id, payload, correlation_id) - VALUES ($1, $2, 'staff.payroll.run_calculated', 1, 'PayrollRun', $3, - 'user', $4, $5, $6)`, - [ - newUlid(), - tenantId, - runId, - actorId, - { period_start: from, period_end: to, total_accrued_minor: totalAccruedMinor.toString() }, - correlationId, - ], - ); + await insertOutboxEvent(client, { + tenantId, + eventType: 'staff.payroll.run_calculated', + subjectType: 'PayrollRun', + subjectId: parseUlid(runId), + actorType: 'user', + actorId, + payload: { + period_start: from, + period_end: to, + total_accrued_minor: totalAccruedMinor.toString(), + }, + correlationId, + }); await client.query( `INSERT INTO staff_audit_log @@ -914,25 +914,20 @@ export async function approvePayrollRun( [tenantId, runId, actorId, approvedAt], ); - await client.query( - `INSERT INTO staff_outbox - (id, tenant_id, event_type, event_version, subject_type, subject_id, - actor_type, actor_id, payload, correlation_id) - VALUES ($1, $2, 'staff.payroll.run_approved', 1, 'PayrollRun', $3, - 'user', $4, $5, $6)`, - [ - newUlid(), - tenantId, - runId, - actorId, - { - period_start: run.period_start, - period_end: run.period_end, - total_accrued_minor: run.total_accrued_minor, - }, - correlationId, - ], - ); + await insertOutboxEvent(client, { + tenantId, + eventType: 'staff.payroll.run_approved', + subjectType: 'PayrollRun', + subjectId: parseUlid(runId), + actorType: 'user', + actorId, + payload: { + period_start: run.period_start, + period_end: run.period_end, + total_accrued_minor: run.total_accrued_minor, + }, + correlationId, + }); await client.query( `INSERT INTO staff_audit_log diff --git a/src/portal.ts b/src/portal.ts index c00335b..052625f 100644 --- a/src/portal.ts +++ b/src/portal.ts @@ -3,6 +3,7 @@ import type { FastifyInstance, FastifyRequest } from 'fastify'; import type { AppConfig } from './config.js'; import type { RequestContext } from './core/context.js'; import { newUlid, parseUlid } from './core/ids.js'; +import { insertOutboxEvent } from './core/outbox.js'; import type { IdentityPort } from './core/ports.js'; import { PlatformIdentityAdapter, StandaloneIdentityAdapter } from './adapters/identity.js'; import type { Database } from './infrastructure/database.js'; @@ -454,12 +455,14 @@ export async function registerPortalRoutes( context.tenantId, context.actorId, async (client) => { + const position = request.body.position?.trim() ?? ''; const saved = await client.query( `INSERT INTO staff_employees (id, tenant_id, personnel_number, display_name, position, profession_id, gender, status) - VALUES ($1, $2, $3, $4, $5, - COALESCE($6, (SELECT id FROM staff_professions - WHERE tenant_id = $2 AND name = $5 AND status = 'active' LIMIT 1)), + VALUES ($1, $2::varchar(26), $3, $4, $5::text, + COALESCE($6, (SELECT id FROM staff_professions + WHERE tenant_id = $2::varchar(26) AND name = $8::varchar(160) + AND status = 'active' LIMIT 1)), $7, 'active') ON CONFLICT (tenant_id, personnel_number) DO NOTHING RETURNING id, core_user_id, personnel_number, display_name, position, profession_id, gender, status`, @@ -468,9 +471,10 @@ export async function registerPortalRoutes( context.tenantId, personnelNumber, displayName, - request.body.position?.trim() ?? '', + position, request.body.profession_id ?? null, request.body.gender ?? '', + position, ], ); if (saved.rowCount === 0) return null; @@ -795,25 +799,19 @@ export async function registerPortalRoutes( ); if (inserted.rowCount === 0) return { id: eventId, duplicate: true }; - const outboxId = newUlid(); const eventType = request.body.event_type === 'check_in' ? 'attendance.clock.in' : 'attendance.clock.out'; - await client.query( - `INSERT INTO staff_outbox - (id, tenant_id, event_type, event_version, subject_type, subject_id, - actor_type, actor_id, payload, correlation_id, occurred_at) - VALUES ($1, $2, $3, 1, 'Employee', $4, 'service', $5, $6, $7, $8)`, - [ - outboxId, - context.tenantId, - eventType, - employeeId, - context.actorId, - { attendance_event_id: eventId, channel: 'platform_notification' }, - context.correlationId, - occurredAt.toISOString(), - ], - ); + const outboxId = await insertOutboxEvent(client, { + tenantId: context.tenantId, + eventType, + subjectType: 'Employee', + subjectId: employeeId, + actorType: 'service', + actorId: context.actorId, + payload: { attendance_event_id: eventId, channel: 'platform_notification' }, + correlationId: context.correlationId, + occurredAt: occurredAt.toISOString(), + }); return { id: eventId, outbox_id: outboxId, duplicate: false }; }, ); @@ -1134,25 +1132,20 @@ export async function registerPortalRoutes( context.actorId, ], ); - await client.query( - `INSERT INTO staff_outbox - (id, tenant_id, event_type, event_version, subject_type, subject_id, - actor_type, actor_id, payload, correlation_id) - VALUES ($1, $2, 'staff.work_session.confirmed', 1, 'WorkSession', $3, - 'user', $4, $5, $6)`, - [ - newUlid(), - context.tenantId, - sessionId, - context.actorId, - { - employee_id: current.employee_id, - business_date: current.business_date, - worked_minutes: workedMinutes, - }, - context.correlationId, - ], - ); + await insertOutboxEvent(client, { + tenantId: context.tenantId, + eventType: 'staff.work_session.confirmed', + subjectType: 'WorkSession', + subjectId: sessionId, + actorType: 'user', + actorId: context.actorId, + payload: { + employee_id: current.employee_id, + business_date: current.business_date, + worked_minutes: workedMinutes, + }, + correlationId: context.correlationId, + }); } return { kind: 'saved' as const, session: result }; }, diff --git a/tests/integration/live-postgres-remediation.test.ts b/tests/integration/live-postgres-remediation.test.ts new file mode 100644 index 0000000..eaf4d64 --- /dev/null +++ b/tests/integration/live-postgres-remediation.test.ts @@ -0,0 +1,127 @@ +import { describe, expect, it } from 'vitest'; +import { buildApp } from '../../src/app.js'; +import { loadConfig } from '../../src/config.js'; +import { newUlid } from '../../src/core/ids.js'; +import { insertOutboxEvent } from '../../src/core/outbox.js'; +import { Database } from '../../src/infrastructure/database.js'; + +const connectionString = process.env.DATABASE_TEST_URL; +const isolatedDatabase = process.env.DATABASE_TEST_ISOLATED === 'true'; +const describeDatabase = connectionString && isolatedDatabase ? describe : describe.skip; + +describeDatabase('live PostgreSQL audit remediations', () => { + it('creates an employee and completes the outbox-producing shift and payroll workflow', async () => { + const database = new Database(connectionString!); + const tenantId = newUlid(); + const actorId = newUlid(); + const managerToken = 'm'.repeat(64); + const orgUnitId = newUlid(); + const sessionId = newUlid(); + const currentDatabase = await database.pool.query<{ name: string }>( + 'SELECT current_database() AS name', + ); + expect(currentDatabase.rows[0]?.name).toMatch(/^krona_staff_audit_[a-z0-9_]+$/); + const existingData = await database.pool.query<{ has_data: boolean }>( + `SELECT EXISTS (SELECT 1 FROM staff_employees) + OR EXISTS (SELECT 1 FROM staff_outbox) + OR EXISTS (SELECT 1 FROM staff_audit_log) + OR EXISTS (SELECT 1 FROM staff_work_sessions) + OR EXISTS (SELECT 1 FROM staff_payroll_runs) AS has_data`, + ); + expect(existingData.rows[0]?.has_data).toBe(false); + const probeOutboxId = await database.tenantTransaction(tenantId, actorId, (client) => + insertOutboxEvent(client, { + tenantId, + eventType: 'staff.audit_probe.created', + subjectType: 'AuditProbe', + subjectId: newUlid(), + actorType: 'user', + actorId, + payload: { source: 'live-postgres-test' }, + correlationId: newUlid(), + }), + ); + const role = await database.pool.query( + 'SELECT rolsuper, rolbypassrls FROM pg_roles WHERE rolname = current_user', + ); + expect(role.rows[0]).toEqual({ rolsuper: false, rolbypassrls: false }); + const probe = await database.tenantTransaction(tenantId, actorId, (client) => + client.query<{ occurred_at: Date }>('SELECT occurred_at FROM staff_outbox WHERE id = $1', [ + probeOutboxId, + ]), + ); + expect(probe.rows[0]?.occurred_at).toBeTruthy(); + + const app = await buildApp({ + config: loadConfig({ + DATABASE_URL: connectionString!, + STANDALONE_TENANT_ID: tenantId, + STANDALONE_ACTOR_ID: actorId, + STANDALONE_MANAGER_TOKEN: managerToken, + STANDALONE_TERMINAL_TOKEN: 't'.repeat(64), + STANDALONE_ACCESS: 'manager', + LOG_LEVEL: 'silent', + NODE_ENV: 'test', + }), + database, + }); + const authorization = { authorization: `Bearer ${managerToken}` }; + try { + const employee = await app.inject({ + method: 'POST', + url: '/api/v1/staff/employees', + headers: authorization, + payload: { + personnel_number: `AUDIT-${newUlid()}`, + display_name: 'Тестовый сотрудник', + position: 'Тестовая профессия', + }, + }); + expect(employee.statusCode, employee.body).toBe(201); + const employeeId = employee.json<{ id: string }>().id; + + await database.tenantTransaction(tenantId, actorId, async (client) => { + await client.query( + `INSERT INTO staff_org_units (id, tenant_id, code, name) + VALUES ($1, $2, $3, 'Тестовая точка')`, + [orgUnitId, tenantId, `audit-${newUlid().toLowerCase()}`], + ); + 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) + VALUES ($1, $2, $3, $4, '2040-01-15', '2040-01-15T01:00:00Z', + '2040-01-15T09:00:00Z', 480, 'pending_review')`, + [sessionId, tenantId, employeeId, orgUnitId], + ); + }); + + const confirmed = await app.inject({ + method: 'PATCH', + url: `/api/v1/staff/work-sessions/${sessionId}`, + headers: authorization, + payload: { status: 'confirmed' }, + }); + expect(confirmed.statusCode).toBe(200); + + const calculated = await app.inject({ + method: 'POST', + url: '/api/v1/staff/payroll/calculate', + headers: authorization, + payload: { month: '2040-01' }, + }); + expect(calculated.statusCode).toBe(200); + const runId = calculated.json<{ run_id: string }>().run_id; + + const approved = await app.inject({ + method: 'POST', + url: `/api/v1/staff/payroll/runs/${runId}/approve`, + headers: authorization, + payload: {}, + }); + expect(approved.statusCode).toBe(200); + } finally { + await app.close(); + } + }); +}); diff --git a/tests/unit/outbox.test.ts b/tests/unit/outbox.test.ts new file mode 100644 index 0000000..4ef731f --- /dev/null +++ b/tests/unit/outbox.test.ts @@ -0,0 +1,56 @@ +import { describe, expect, it, vi } from 'vitest'; +import { insertOutboxEvent } from '../../src/core/outbox.js'; +import { parseUlid } from '../../src/core/ids.js'; + +describe('transactional outbox helper', () => { + it('writes a complete platform event envelope with an occurrence time', async () => { + const query = vi.fn(async (...args: [string, readonly unknown[] | undefined]) => { + void args; + return { rows: [], rowCount: 1 }; + }); + const tenantId = parseUlid('01J0000000000000000000000A'); + const actorId = parseUlid('01J0000000000000000000001A'); + const subjectId = parseUlid('01J0000000000000000000002A'); + const correlationId = parseUlid('01J0000000000000000000003A'); + + await insertOutboxEvent({ query } as never, { + tenantId, + eventType: 'staff.test.created', + subjectType: 'TestSubject', + subjectId, + actorType: 'user', + actorId, + payload: { state: 'created' }, + correlationId, + }); + + expect(query).toHaveBeenCalledOnce(); + const [sql, parameters] = query.mock.calls[0] as unknown as [string, unknown[]]; + expect(sql).toContain('causation_id, occurred_at'); + expect(parameters[8]).toBe('{"state":"created"}'); + expect(parameters[11]).toMatch(/^\d{4}-\d{2}-\d{2}T/); + }); + + it('preserves a source occurrence time for imported or terminal events', async () => { + const query = vi.fn(async (...args: [string, readonly unknown[] | undefined]) => { + void args; + return { rows: [], rowCount: 1 }; + }); + const occurredAt = '2026-09-24T01:00:00.000Z'; + + await insertOutboxEvent({ query } as never, { + tenantId: parseUlid('01J0000000000000000000000A'), + eventType: 'attendance.clock.in', + subjectType: 'Employee', + subjectId: parseUlid('01J0000000000000000000002A'), + actorType: 'service', + actorId: parseUlid('01J0000000000000000000001A'), + payload: { source: 'terminal' }, + correlationId: parseUlid('01J0000000000000000000003A'), + occurredAt, + }); + + const parameters = query.mock.calls[0]?.[1] as readonly unknown[]; + expect(parameters[11]).toBe(occurredAt); + }); +});