fix(database): make live PostgreSQL workflows valid

Route every transactional outbox write through one complete event-envelope helper so occurred_at is always populated. Separate employee position and profession lookup parameters to avoid PostgreSQL 42P08, and add unit plus live-database regression coverage.
2026-09-26 14:47:40 +07:00 · 2026-09-26 14:47:40 +07:00 · b289dd6d63
commit b289dd6d63
parent fd1ec3d819
5 changed files with 289 additions and 74 deletions
--- a/src/core/outbox.ts
+++ b/src/core/outbox.ts
@ -0,0 +1,45 @@
+import type { PoolClient } from 'pg';
+import { newUlid, type Ulid } from './ids.js';
+
+export interface OutboxEventInput<TPayload extends Record<string, unknown>> {
+  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<TPayload extends Record<string, unknown>>(
+  client: Pick<PoolClient, 'query'>,
+  event: OutboxEventInput<TPayload>,
+): Promise<Ulid> {
+  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;
+}
--- 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(),
+  await insertOutboxEvent(client, {
    tenantId,
-      runId,
+    eventType: 'staff.payroll.run_calculated',
+    subjectType: 'PayrollRun',
+    subjectId: parseUlid(runId),
+    actorType: 'user',
    actorId,
-      { period_start: from, period_end: to, total_accrued_minor: totalAccruedMinor.toString() },
+    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(),
+  await insertOutboxEvent(client, {
    tenantId,
-      runId,
+    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
--- 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,
+          VALUES ($1, $2::varchar(26), $3, $4, $5::text,
            COALESCE($6, (SELECT id FROM staff_professions
-                         WHERE tenant_id = $2 AND name = $5 AND status = 'active' LIMIT 1)),
+                          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,
+          const outboxId = await insertOutboxEvent(client, {
+            tenantId: context.tenantId,
            eventType,
-              employeeId,
-              context.actorId,
-              { attendance_event_id: eventId, channel: 'platform_notification' },
-              context.correlationId,
-              occurredAt.toISOString(),
-            ],
-          );
+            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,
-                {
+            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,
              },
-                context.correlationId,
-              ],
-            );
+              correlationId: context.correlationId,
+            });
          }
          return { kind: 'saved' as const, session: result };
        },
--- a/tests/integration/live-postgres-remediation.test.ts
+++ b/tests/integration/live-postgres-remediation.test.ts
@ -0,0 +1,126 @@
+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 guardToken = process.env.DATABASE_TEST_GUARD_TOKEN;
+const describeDatabase =
+  connectionString && isolatedDatabase && guardToken ? 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 role = await database.pool.query<{ rolsuper: boolean; rolbypassrls: boolean }>(
+      'SELECT rolsuper, rolbypassrls FROM pg_roles WHERE rolname = current_user',
+    );
+    expect(role.rows[0]).toEqual({ rolsuper: false, rolbypassrls: false });
+    const guard = await database.pool.query<{ marker: string }>(
+      'SELECT marker FROM krona_staff_test_guard WHERE marker = $1',
+      [guardToken],
+    );
+    expect(guard.rows).toEqual([{ marker: guardToken }]);
+    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 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();
+    }
+  });
+});
--- a/tests/unit/outbox.test.ts
+++ 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);
+  });
+});