diff --git a/apps/docs/content/docs/platform/enterprise/audit-logs.mdx b/apps/docs/content/docs/platform/enterprise/audit-logs.mdx index 13980ad02fe..84614751b4c 100644 --- a/apps/docs/content/docs/platform/enterprise/audit-logs.mdx +++ b/apps/docs/content/docs/platform/enterprise/audit-logs.mdx @@ -105,6 +105,8 @@ Events use identifiers such as `resource.action`. These categories are represent Authentication failures and password-reset requests use an anonymous actor and record the target account separately. Impersonation and administrative session revocations identify the operator and target organization. Table-row mutations record summaries and counts without copying row contents. +When a workflow run makes the same kind of change to a table many times in quick succession, such as inserting rows in a loop, the repeats are merged into one `table.updated` entry. The merged entry keeps every row ID in `metadata.rowIds`, sums the counts, and records in `metadata.rollup` how many changes it covers and when the first and last happened. The actor of any workflow-made entry names the workflow (`metadata.actor.workflowId`) and the run (`metadata.actor.executionId`). Credential use by a workflow run is merged the same way: one `credential.accessed` entry per credential, counting the repeats. + Most audit writes are best effort: an outage or process loss can prevent persistence after an action succeeds. Deployment activation uses an awaited, idempotent insert before its outbox checkpoint. The feed is not a complete tool-call stream or a guaranteed transactional ledger for every operation. The interface and API provide no audit-entry editing or deletion. Database administrators can alter stored records, and entries are not cryptographically sealed. Use a drain to retain your own archive. diff --git a/apps/sim/lib/auth/principal.test.ts b/apps/sim/lib/auth/principal.test.ts index e3e79def78b..18a30ffef75 100644 --- a/apps/sim/lib/auth/principal.test.ts +++ b/apps/sim/lib/auth/principal.test.ts @@ -575,6 +575,7 @@ describe('principal actors', () => { kind: 'delegated', serviceId: 'executor', delegationId: 'delegation-1', + workflowId: 'workflow-1', }, attributedUserId: 'billing-owner-1', }) diff --git a/apps/sim/lib/core/application/authorized-workspace-use-case.ts b/apps/sim/lib/core/application/authorized-workspace-use-case.ts index 618f0d2fc62..5cfc0516f49 100644 --- a/apps/sim/lib/core/application/authorized-workspace-use-case.ts +++ b/apps/sim/lib/core/application/authorized-workspace-use-case.ts @@ -1,12 +1,16 @@ import { type AuditActionType, + type AuditLogParams, type AuditResourceTypeValue, + type AuditRollupSummary, recordAudit, recordAuditOnce, + recordAuditRollup, } from '@sim/audit' -import type { Principal, PrincipalAuditAttribution } from '@sim/auth/principal' +import type { Principal, PrincipalActor, PrincipalAuditAttribution } from '@sim/auth/principal' import { resolvePrincipalAuditAttribution } from '@sim/auth/principal' import { createLogger } from '@sim/logger' +import { toNumberOrNull, toStringOrNull } from '@sim/utils/coerce' import { runDeclaredContentOperation } from '@/lib/core/application/content-delivery' import type { ApplicationOperation, OperationUseCase } from '@/lib/core/application/operation' import { @@ -26,6 +30,11 @@ import type { ResourcePolicyBinding } from '@/lib/resource-policies/registry' const logger = createLogger('WorkspaceUseCaseAudit') +/** The audit actor of a workflow-execution delegation that names its run. */ +type ExecutionDelegatedActor = Extract & { + executionId: string +} + export interface WorkspaceUseCaseAuditEntry { /** Canonical affected workspace; null keeps an organization event outside the authorization workspace. */ workspaceId?: string | null @@ -40,6 +49,21 @@ export interface WorkspaceUseCaseAuditEntry { * retry after losing its response: each retry projects the same entry again. */ idempotencyKey?: string + /** + * Merges this entry with its repeats from the same workflow execution into + * one row, keeping every count and identifier (see `recordAuditRollup`). + * Applies only to an executor delegation that names its execution; every + * other caller records the entry as usual. + */ + executionRollup?: WorkspaceUseCaseAuditRollup +} + +export interface WorkspaceUseCaseAuditRollup { + /** Metadata keys summed across merged entries. */ + counterKeys: readonly string[] + /** Metadata keys collected into a `s` array on a merged row. */ + identifierKeys: readonly string[] + describe(summary: AuditRollupSummary): string } export interface AuthorizedWorkspaceUseCaseContext< @@ -119,7 +143,12 @@ export function recordProjectedUseCaseAuditEntries( organizationId?: string ): void { const attribution: PrincipalAuditAttribution = resolvePrincipalAuditAttribution(principal) - for (const { idempotencyKey, ...entry } of entries) { + const { actor } = attribution + const rollupActor = + actor.kind === 'delegated' && actor.executionId && !request + ? { ...actor, executionId: actor.executionId } + : undefined + for (const { idempotencyKey, executionRollup, ...entry } of entries) { const params = { workspaceId: entry.workspaceId === undefined ? workspaceId : entry.workspaceId, actorId: attribution.actorId, @@ -137,6 +166,10 @@ export function recordProjectedUseCaseAuditEntries( }, request, } + if (executionRollup && rollupActor && !idempotencyKey) { + recordExecutionAuditRollup(params, rollupActor, executionRollup) + continue + } if (!idempotencyKey) { recordAudit(params) continue @@ -147,6 +180,37 @@ export function recordProjectedUseCaseAuditEntries( } } +/** + * Entries merge when everything but their counters, identifiers, and the + * per-call delegation matches, so a merged row never blends two operations. + */ +function recordExecutionAuditRollup( + params: AuditLogParams & { metadata: Record }, + { delegationId, ...sharedActor }: ExecutionDelegatedActor, + rollup: WorkspaceUseCaseAuditRollup +): void { + const shared: Record = { ...params.metadata, actor: sharedActor } + const counters: Record = {} + const identifiers: Record = { delegationId } + for (const key of rollup.counterKeys) { + const value = toNumberOrNull(shared[key]) + if (value !== null) counters[key] = value + delete shared[key] + } + for (const key of rollup.identifierKeys) { + identifiers[key] = toStringOrNull(shared[key]) ?? undefined + delete shared[key] + } + recordAuditRollup({ + key: JSON.stringify(shared), + params, + rollupMetadata: shared, + counters, + identifiers, + describe: rollup.describe, + }) +} + /** * A use case that always answers `authorize`, so a caller that must run the * funnel without executing — a `HEAD` on a route declaring `headSafe: false`, diff --git a/apps/sim/lib/core/application/index.ts b/apps/sim/lib/core/application/index.ts index 0b4ad709f9e..f94e40bce27 100644 --- a/apps/sim/lib/core/application/index.ts +++ b/apps/sim/lib/core/application/index.ts @@ -8,6 +8,7 @@ export { defineAuthorizedWorkspaceUseCase, recordProjectedUseCaseAuditEntries, type WorkspaceUseCaseAuditEntry, + type WorkspaceUseCaseAuditRollup, } from '@/lib/core/application/authorized-workspace-use-case' export { FORBIDDEN_DETAIL_CODES, diff --git a/apps/sim/lib/oauth/token-resolution.test.ts b/apps/sim/lib/oauth/token-resolution.test.ts index 75542631bab..bf8351da1a8 100644 --- a/apps/sim/lib/oauth/token-resolution.test.ts +++ b/apps/sim/lib/oauth/token-resolution.test.ts @@ -75,6 +75,7 @@ describe('resolveCredentialToken', () => { expect(mockGetCredential).not.toHaveBeenCalled() expect(mockRefreshTokenIfNeeded).not.toHaveBeenCalled() expect(mockRecordAudit).not.toHaveBeenCalled() + expect(auditMockFns.mockRecordAuditRollup).not.toHaveBeenCalled() }) it('refreshes the token, records the access trail, and returns the payload', async () => { @@ -104,12 +105,14 @@ describe('resolveCredentialToken', () => { token: { accessToken: 'fresh', credentialType: 'oauth', idToken: 'id-token' }, }) expect(mockGetCredential).toHaveBeenCalledWith('req-1', 'account-1', 'owner-1') - expect(mockRecordAudit).toHaveBeenCalledWith( + expect(auditMockFns.mockRecordAuditRollup).toHaveBeenCalledWith( expect.objectContaining({ - actorId: 'user-1', - workspaceId: 'ws-1', - resourceId: 'account-1', - action: 'credential.accessed', + params: expect.objectContaining({ + actorId: 'user-1', + workspaceId: 'ws-1', + resourceId: 'account-1', + action: 'credential.accessed', + }), }) ) }) @@ -164,6 +167,7 @@ describe('resolveCredentialToken', () => { expect(result).toEqual({ ok: false, status: 401, error: 'Failed to refresh access token' }) expect(mockRecordAudit).not.toHaveBeenCalled() + expect(auditMockFns.mockRecordAuditRollup).not.toHaveBeenCalled() }) it('authorizes service-account credentials before minting a token', async () => { @@ -467,6 +471,7 @@ describe('ordinary tool credential binding', () => { expect(mockResolveServiceAccountToken).not.toHaveBeenCalled() expect(mockRefreshTokenIfNeeded).not.toHaveBeenCalled() expect(mockRecordAudit).not.toHaveBeenCalled() + expect(auditMockFns.mockRecordAuditRollup).not.toHaveBeenCalled() } ) diff --git a/apps/sim/lib/oauth/token-resolution.ts b/apps/sim/lib/oauth/token-resolution.ts index 78a5baf89fe..08b77d05d36 100644 --- a/apps/sim/lib/oauth/token-resolution.ts +++ b/apps/sim/lib/oauth/token-resolution.ts @@ -1,4 +1,4 @@ -import { AuditAction, AuditResourceType, recordAudit } from '@sim/audit' +import { AuditAction, AuditResourceType, recordAudit, recordAuditRollup } from '@sim/audit' import { type DelegatedPrincipal, resolvePrincipalSubject } from '@sim/auth/principal' import { createLogger } from '@sim/logger' import { getErrorMessage } from '@sim/utils/errors' @@ -102,7 +102,9 @@ export function validateOAuthCredentialContext( /** * Emits the semantic "credential used" trail for one resolved credential. - * Both the audit row and the analytics event are fire-and-forget. + * Both the audit row and the analytics event are fire-and-forget. A server-side + * access with no client request, such as a workflow run refreshing the token + * for every tool call, merges with its repeats into one row that counts them. */ export function recordCredentialAccess(params: { actorId: string @@ -113,19 +115,31 @@ export function recordCredentialAccess(params: { auditRequest?: CredentialAuditRequest }): void { const { actorId, workspaceId, resourceId, providerId, credentialType } = params - recordAudit({ + const credentialLabel = `${credentialType === 'oauth' ? 'OAuth' : 'service account'} credential for provider ${providerId ?? 'unknown'}` + const audit = { workspaceId, actorId, action: AuditAction.CREDENTIAL_ACCESSED, resourceType: AuditResourceType.CREDENTIAL, resourceId, - description: `Accessed ${credentialType === 'oauth' ? 'OAuth' : 'service account'} credential for provider ${providerId ?? 'unknown'}`, + description: `Accessed ${credentialLabel}`, metadata: { provider: providerId, credentialType, }, - request: params.auditRequest, - }) + } + if (params.auditRequest) { + recordAudit({ ...audit, request: params.auditRequest }) + } else { + recordAuditRollup({ + key: JSON.stringify(audit.metadata), + params: audit, + rollupMetadata: audit.metadata, + counters: {}, + identifiers: {}, + describe: ({ entries }) => `Accessed ${credentialLabel} ${entries} times`, + }) + } captureServerEvent( actorId, 'credential_used', diff --git a/apps/sim/lib/table/application/row-audit.integration.ts b/apps/sim/lib/table/application/row-audit.integration.ts index ebc871c5ebf..89ecf0bb009 100644 --- a/apps/sim/lib/table/application/row-audit.integration.ts +++ b/apps/sim/lib/table/application/row-audit.integration.ts @@ -1,3 +1,4 @@ +import { withAuditRollupScope } from '@sim/audit' import { db } from '@sim/db' import { auditLog, @@ -8,14 +9,17 @@ import { workspace, } from '@sim/db/schema' import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures' -import { createSessionPrincipal } from '@sim/testing/factories/principal.factory' +import { + createExecutorPrincipal, + createSessionPrincipal, +} from '@sim/testing/factories/principal.factory' import { storageServiceMock } from '@sim/testing/mocks/storage-service.mock' import { tableBillingMock, tableBillingMockFns } from '@sim/testing/mocks/table-billing.mock' import { tableEventsMock } from '@sim/testing/mocks/table-events.mock' import { tableTriggerMock } from '@sim/testing/mocks/table-trigger.mock' import { tableWorkflowColumnsMock } from '@sim/testing/mocks/table-workflow-columns.mock' import { generateId } from '@sim/utils/id' -import { eq } from 'drizzle-orm' +import { asc, eq } from 'drizzle-orm' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' vi.mock('@/lib/table/billing', () => tableBillingMock) @@ -173,4 +177,100 @@ describe('table mutation audit history in PostgreSQL', () => { }) expect(JSON.stringify(entry)).not.toContain(privateValue) }) + + function executorPrincipal(executionId: string) { + return createExecutorPrincipal({ + subjectUserId: userId, + workspaceId, + delegationId: generateId(), + audience: 'sim:tables', + resourceScope: { tableId }, + delegationContext: { kind: 'workflow_execution', workflowId: 'wf-audit', executionId }, + }) + } + + async function insertAsExecution(executionId: string, value: string) { + const result = await createTableRows.execute({ + principal: executorPrincipal(executionId), + input: { + tableId, + assertedWorkspaceId: workspaceId, + strictWrite: true, + dataKeying: 'ids', + kind: 'single', + data: { value }, + }, + }) + if (result.kind !== 'single') throw new Error('Expected a single-row insert') + return result.row.id + } + + async function auditRows() { + return db + .select() + .from(auditLog) + .where(eq(auditLog.workspaceId, workspaceId)) + .orderBy(asc(auditLog.createdAt)) + } + + it("merges one execution's repeated row inserts into one row that keeps every row id and count", async () => { + const rowIds: string[] = [] + await withAuditRollupScope(async () => { + for (const value of ['a', 'b', 'c']) rowIds.push(await insertAsExecution('exec-1', value)) + }) + + const entries = await auditRows() + expect(entries).toHaveLength(1) + const [entry] = entries + expect(entry).toMatchObject({ + actorId: userId, + action: 'table.updated', + resourceId: tableId, + metadata: { + op: 'insert', + operation: 'tables.rows.create', + rowsInserted: 3, + rowIds, + actor: { kind: 'delegated', workflowId: 'wf-audit', executionId: 'exec-1' }, + rollup: { entries: 3 }, + }, + }) + const metadata = entry.metadata as Record + expect(metadata.delegationIds).toHaveLength(3) + expect(new Set(metadata.delegationIds as string[]).size).toBe(3) + expect(metadata).not.toHaveProperty('rowId') + expect(entry.description).toContain('3 inserted') + }) + + it('never merges row changes from two executions', async () => { + await withAuditRollupScope(async () => { + await insertAsExecution('exec-1', 'a') + await insertAsExecution('exec-2', 'b') + await insertAsExecution('exec-1', 'c') + }) + + const entries = await auditRows() + expect(entries).toHaveLength(2) + const byExecution = new Map( + entries.map((entry) => { + const metadata = entry.metadata as { actor: { executionId: string }; rowsInserted: number } + return [metadata.actor.executionId, metadata.rowsInserted] + }) + ) + expect(Object.fromEntries(byExecution)).toEqual({ 'exec-1': 2, 'exec-2': 1 }) + }) + + it('writes a lone execution change exactly as an unmerged entry', async () => { + const insertedRowId = await withAuditRollupScope(() => insertAsExecution('exec-1', 'a')) + + const [entry] = await auditRows() + expect(entry.metadata).toMatchObject({ + op: 'insert', + rowId: insertedRowId, + rowsInserted: 1, + actor: { kind: 'delegated', workflowId: 'wf-audit', executionId: 'exec-1' }, + }) + expect(entry.metadata).not.toHaveProperty('rollup') + expect(entry.description).toBe('Inserted 1 row(s) into table "Audited table"') + }) }) diff --git a/apps/sim/lib/table/application/rows.ts b/apps/sim/lib/table/application/rows.ts index 948076cc1ee..832e40d09b5 100644 --- a/apps/sim/lib/table/application/rows.ts +++ b/apps/sim/lib/table/application/rows.ts @@ -12,6 +12,7 @@ import { capabilityGovernedPrincipalUserId, isCopilotPrincipal, mustClassifyContent, + type WorkspaceUseCaseAuditRollup, } from '@/lib/core/application' import { OrchestrationError } from '@/lib/core/orchestration/types' import { isPrivateSecretProvenanceScopeCompatible } from '@/lib/execution/durable-secret-provenance' @@ -97,6 +98,27 @@ import { predicateToStorage, resolveFilterSelectValues } from '@/lib/table/selec import { coerceRowValues } from '@/lib/table/validation' import type { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry' +const ROW_CHANGE_COUNTERS = ['rowsInserted', 'rowsUpdated', 'rowsDeleted'] as const + +/** + * Merges one workflow execution's repeats of a row change to a table into a + * single audit row that keeps every row id and count. + */ +function rowChangeRollup(tableName: string): WorkspaceUseCaseAuditRollup { + return { + counterKeys: ROW_CHANGE_COUNTERS, + identifierKeys: ['rowId'], + describe: ({ entries, counters }) => { + const changes = [ + counters.rowsInserted ? `${counters.rowsInserted} inserted` : undefined, + counters.rowsUpdated ? `${counters.rowsUpdated} updated` : undefined, + counters.rowsDeleted ? `${counters.rowsDeleted} deleted` : undefined, + ].filter((change) => change !== undefined) + return `Workflow execution changed rows in table "${tableName}" ${entries} times (${changes.join(', ') || 'no rows'})` + }, + } +} + export class TableRowsValidationError extends OrchestrationError { constructor( message: string, @@ -984,6 +1006,7 @@ export const createTableRows = defineAuthorizedTableUseCase({ resourceType: AuditResourceType.TABLE, resourceId: context.tableId, resourceName: context.table.name, + executionRollup: rowChangeRollup(context.table.name), description: `Inserted ${inserted} row(s) into table "${context.table.name}"`, metadata: { op: result.kind === 'single' ? 'insert' : 'batch_insert', @@ -1053,6 +1076,7 @@ export const replaceTableRows = defineAuthorizedTableUseCase({ resourceType: AuditResourceType.TABLE, resourceId: context.tableId, resourceName: context.table.name, + executionRollup: rowChangeRollup(context.table.name), description: `Replaced rows in table "${context.table.name}"`, metadata: { op: 'replace_rows', @@ -1225,6 +1249,7 @@ export const replaceProjectedWireRows = defineAuthorizedTableUseCase({ resourceType: AuditResourceType.TABLE, resourceId: result.table.id, resourceName: result.table.name, + executionRollup: rowChangeRollup(result.table.name), description: `Replaced rows in table "${result.table.name}"`, metadata: { op: 'replace_projected_rows', @@ -1323,6 +1348,7 @@ export const updateTableRow = defineAuthorizedTableUseCase({ resourceType: AuditResourceType.TABLE, resourceId: context.tableId, resourceName: context.table.name, + executionRollup: rowChangeRollup(context.table.name), description: `Updated a row in table "${context.table.name}"`, metadata: { op: 'update', rowId: result.row.id, rowsUpdated: 1 }, } @@ -1393,6 +1419,7 @@ export const updateTableRows = defineAuthorizedTableUseCase({ resourceType: AuditResourceType.TABLE, resourceId: context.tableId, resourceName: context.table.name, + executionRollup: rowChangeRollup(context.table.name), description: `Updated ${result.affectedCount} row(s) in table "${context.table.name}"`, metadata: { op: 'update_by_filter', rowsUpdated: result.affectedCount }, } @@ -1499,6 +1526,7 @@ export const batchUpdateTableRows = defineAuthorizedTableUseCase({ resourceType: AuditResourceType.TABLE, resourceId: context.tableId, resourceName: context.table.name, + executionRollup: rowChangeRollup(context.table.name), description: `Updated ${result.affectedCount} row(s) in table "${context.table.name}"`, metadata: { op: 'batch_update', rowsUpdated: result.affectedCount }, } @@ -1530,6 +1558,7 @@ export const deleteTableRow = defineAuthorizedTableUseCase({ resourceType: AuditResourceType.TABLE, resourceId: context.tableId, resourceName: context.table.name, + executionRollup: rowChangeRollup(context.table.name), description: `Deleted a row from table "${context.table.name}"`, metadata: { op: 'delete', rowId: result.deletedRowId, rowsDeleted: 1 }, }), @@ -1602,6 +1631,7 @@ export const deleteTableRows = defineAuthorizedTableUseCase({ resourceType: AuditResourceType.TABLE, resourceId: result.table.id, resourceName: result.table.name, + executionRollup: rowChangeRollup(result.table.name), description: `Deleted ${affected} row(s) from table "${result.table.name}"`, metadata: { op: 'bulk_delete', @@ -1688,6 +1718,7 @@ export const upsertTableRow = defineAuthorizedTableUseCase({ resourceType: AuditResourceType.TABLE, resourceId: context.tableId, resourceName: context.table.name, + executionRollup: rowChangeRollup(context.table.name), description: `${result.operation === 'insert' ? 'Inserted' : 'Updated'} a row in table "${context.table.name}"`, metadata: { op: 'upsert', diff --git a/apps/sim/lib/workflows/application/deployments.test.ts b/apps/sim/lib/workflows/application/deployments.test.ts index e5c01ba73b5..706b80eb242 100644 --- a/apps/sim/lib/workflows/application/deployments.test.ts +++ b/apps/sim/lib/workflows/application/deployments.test.ts @@ -234,6 +234,8 @@ describe('workflow deployment application use cases', () => { serviceId: 'executor', subjectUserId: 'user-1', delegationId: 'executor-1', + workflowId: 'origin-workflow', + executionId: 'execution-1', }, captureAnalytics: false, requestId: 'request-1', diff --git a/apps/sim/lib/workflows/deployment-outbox.ts b/apps/sim/lib/workflows/deployment-outbox.ts index 96c2d3fa7c0..973640be7f5 100644 --- a/apps/sim/lib/workflows/deployment-outbox.ts +++ b/apps/sim/lib/workflows/deployment-outbox.ts @@ -3,6 +3,7 @@ import type { PrincipalActor } from '@sim/auth/principal' import { db, workflowDeploymentVersion, workflow as workflowTable } from '@sim/db' import { outboxEvent, workspaceOperationReceipt } from '@sim/db/schema' import { createLogger } from '@sim/logger' +import { toStringOrNull } from '@sim/utils/coerce' import { toError } from '@sim/utils/errors' import { and, eq } from 'drizzle-orm' import { NextRequest } from 'next/server' @@ -1528,11 +1529,25 @@ function parseOptionalPrincipalActor(value: unknown): PrincipalActor | undefined serviceId, subjectUserId: parseRequiredString(record.subjectUserId, 'actor.subjectUserId'), delegationId: parseRequiredString(record.delegationId, 'actor.delegationId'), + ...parseOptionalActorRun(record), } } throw new Error(`Invalid deployment outbox actor kind: ${kind}`) } +/** The workflow run an executor delegation acted for, kept so its audit stays traceable. */ +function parseOptionalActorRun(record: Record): { + workflowId?: string + executionId?: string +} { + const workflowId = toStringOrNull(record.workflowId) + const executionId = toStringOrNull(record.executionId) + return { + ...(workflowId ? { workflowId } : {}), + ...(executionId ? { executionId } : {}), + } +} + function parseDeploymentPreparationCheckpoints(value: unknown): DeploymentPreparationCheckpoints { if (!value || typeof value !== 'object' || Array.isArray(value)) return {} const record = value as Record diff --git a/apps/sim/lib/workflows/executor/execution-core.ts b/apps/sim/lib/workflows/executor/execution-core.ts index 03a148256b5..13d7547d1eb 100644 --- a/apps/sim/lib/workflows/executor/execution-core.ts +++ b/apps/sim/lib/workflows/executor/execution-core.ts @@ -3,6 +3,7 @@ * This is the SINGLE source of truth for workflow execution */ +import { withAuditRollupScope } from '@sim/audit' import { resolvePrincipalSubject } from '@sim/auth/principal' import { db } from '@sim/db' import { organization, workspace } from '@sim/db/schema' @@ -607,9 +608,15 @@ function prefetchExecutionReads(options: ExecuteWorkflowCoreOptions): ExecutionR * volume, from opening a connection they would never use. * * Runs under `runWorkflowOutsideContentDelivery`, so a run Chat started reads what any run reads. + * + * Owns the audit rollups the run and its inline child runs record, and writes + * them before returning, since a background run's process may be frozen or + * ended once it does. */ export function executeWorkflowCore(options: ExecuteWorkflowCoreOptions): Promise { - return runWorkflowOutsideContentDelivery(() => executeWorkflowCoreInRun(options)) + return withAuditRollupScope(() => + runWorkflowOutsideContentDelivery(() => executeWorkflowCoreInRun(options)) + ) } async function executeWorkflowCoreInRun( diff --git a/packages/audit/src/index.ts b/packages/audit/src/index.ts index 67ac50cd746..777907eca40 100644 --- a/packages/audit/src/index.ts +++ b/packages/audit/src/index.ts @@ -1,5 +1,7 @@ export type { AuditLogParams } from './log' export { recordAudit, recordAuditBatch, recordAuditOnce } from './log' +export type { AuditRollupSummary } from './rollup' +export { recordAuditRollup, withAuditRollupScope } from './rollup' export type { AuditActionType, AuditResourceTypeValue } from './types' export { AuditAction, AuditResourceType } from './types' export { auditUpdatedFields } from './updated-fields' diff --git a/packages/audit/src/log.ts b/packages/audit/src/log.ts index eb784d9c9b5..8f99bfcc63f 100644 --- a/packages/audit/src/log.ts +++ b/packages/audit/src/log.ts @@ -67,13 +67,13 @@ export function recordAuditBatch(entries: AuditLogParams[]): void { * are taken as-is — lazy actor resolution is layered on top by * {@link recordAudit} only. * - * The surface comes from the ambient request context, which the route wrapper - * resolves once per request, so every entry recorded while serving a request - * is attributed without each caller passing it. It is absent outside a request. + * The surface is passed explicitly: immediate writes take it from + * {@link ambientSurface}, deferred writes from when the entry was recorded. */ function buildAuditRow( params: AuditLogParams, actor: { actorId: string | null; actorName?: string | null; actorEmail?: string | null }, + surface: string | undefined, id = generateShortId() ) { return { @@ -90,10 +90,32 @@ function buildAuditRow( metadata: params.metadata ?? {}, ipAddress: params.request ? clientIpResolver.resolve(params.request.headers) : undefined, userAgent: params.request?.headers.get('user-agent') ?? undefined, - surface: getRequestContext()?.client?.surface, + surface, } } +/** + * The surface of the request being served, which the route wrapper resolves + * once per request, so every entry recorded while serving it is attributed + * without each caller passing it. It is absent outside a request. + */ +export function ambientSurface(): string | undefined { + return getRequestContext()?.client?.surface +} + +/** + * Writes an entry recorded earlier, outside the request that produced it, with + * the surface captured when it was recorded. Throws on failure so the caller + * logs what it lost. + */ +export async function insertDeferredAuditEntry( + params: AuditLogParams, + surface: string | undefined +): Promise { + const actor = await resolveAuditActor(params) + await db.insert(auditLog).values(buildAuditRow(params, actor, surface)) +} + /** * Persists one audit row under a caller-owned idempotency key. * @@ -108,19 +130,24 @@ export async function recordAuditOnce(id: string, params: AuditLogParams): Promi const actor = await resolveAuditActor(params) await db .insert(auditLog) - .values(buildAuditRow(params, actor, id)) + .values(buildAuditRow(params, actor, ambientSurface(), id)) .onConflictDoNothing({ target: auditLog.id }) } async function insertAuditLogBatch(entries: AuditLogParams[]): Promise { if (entries.length === 0) return + const surface = ambientSurface() const rows = entries.map((params) => - buildAuditRow(params, { - actorId: params.actorId, - actorName: params.actorName, - actorEmail: params.actorEmail, - }) + buildAuditRow( + params, + { + actorId: params.actorId, + actorName: params.actorName, + actorEmail: params.actorEmail, + }, + surface + ) ) for (let index = 0; index < rows.length; index += AUDIT_BATCH_CHUNK_SIZE) { @@ -130,7 +157,7 @@ async function insertAuditLogBatch(entries: AuditLogParams[]): Promise { async function insertAuditLog(params: AuditLogParams): Promise { const actor = await resolveAuditActor(params) - await db.insert(auditLog).values(buildAuditRow(params, actor)) + await db.insert(auditLog).values(buildAuditRow(params, actor, ambientSurface())) } async function resolveAuditActor(params: AuditLogParams): Promise<{ diff --git a/packages/audit/src/rollup.ts b/packages/audit/src/rollup.ts new file mode 100644 index 00000000000..56e669dad7f --- /dev/null +++ b/packages/audit/src/rollup.ts @@ -0,0 +1,217 @@ +import { AsyncLocalStorage } from 'node:async_hooks' +import { createLogger } from '@sim/logger' +import { type AuditLogParams, ambientSurface, insertDeferredAuditEntry } from './log' + +const logger = createLogger('AuditRollup') + +/** How long the first entry of a rollup waits for repeats before it is written. */ +const ROLLUP_WINDOW_MS = 2_000 + +/** Entries merged into one row; bounds the identifier arrays its metadata collects. */ +const ROLLUP_MAX_ENTRIES = 100 + +/** Pending rollups held by the process; reaching it writes every one of them. */ +const ROLLUP_MAX_PENDING = 1_000 + +/** Rows inserted at once by one write, so a large flush never floods the pool. */ +const ROLLUP_WRITE_CONCURRENCY = 10 + +export interface AuditRollupSummary { + entries: number + counters: Record +} + +/** + * One occurrence of a repetitive audit event. Occurrences sharing a rollup key + * within {@link ROLLUP_WINDOW_MS} become one row, and nothing an occurrence + * carried is dropped: a lone occurrence is written exactly as `params`, and a + * merged row sums every counter, collects every identifier, and records how + * many occurrences it stands for and when the first and last happened. + */ +export interface AuditRollupEntry { + /** + * What must match, beyond the workspace, actor, action, resource, and surface + * the rollup key always includes, for two occurrences to share a row. + */ + key: string + /** The row a lone occurrence writes. Its request must be absent: a merged row has no single client. */ + params: Omit + /** Metadata of a merged row, before the summed counters and collected identifiers. */ + rollupMetadata: Record + /** Summed into `rollupMetadata[name]`. */ + counters: Record + /** Collected into `rollupMetadata[`${name}s`]`; an absent value is skipped. */ + identifiers: Record + describe(summary: AuditRollupSummary): string +} + +/** + * A unit of work that owns the rollups recorded inside it (a workflow run and + * the child runs it executes inline), so its end writes only its own rows. + */ +interface RollupScope { + id: number + writes: Set> +} + +interface PendingRollup { + scope: RollupScope | undefined + first: AuditRollupEntry + surface: string | undefined + firstAt: Date + lastAt: Date + entries: number + counters: Record + identifiers: Map + timer: ReturnType +} + +const pending = new Map() +const scopeStorage = new AsyncLocalStorage() +let nextScopeId = 0 + +/** + * Records a repetitive audit event, merging it with its repeats. Fire-and-forget + * like `recordAudit`. Each pending row is written when its window closes, when + * it reaches {@link ROLLUP_MAX_ENTRIES}, or when the {@link withAuditRollupScope} + * it was recorded in ends. The window timer is not unref'd, so a process that + * ends naturally still writes it. + */ +export function recordAuditRollup(entry: AuditRollupEntry): void { + const surface = ambientSurface() + const scope = scopeStorage.getStore() + const { params } = entry + const key = JSON.stringify([ + scope?.id ?? null, + surface ?? null, + params.workspaceId ?? null, + params.actorId, + params.actorName ?? null, + params.action, + params.resourceType, + params.resourceId ?? null, + entry.key, + ]) + + const existing = pending.get(key) + if (existing) { + mergeInto(existing, entry) + if (existing.entries >= ROLLUP_MAX_ENTRIES) startWrite([take(key)]) + return + } + + if (pending.size >= ROLLUP_MAX_PENDING) { + startWrite([...pending.keys()].map(take)) + } + + const now = new Date() + const rollup: PendingRollup = { + scope, + first: entry, + surface, + firstAt: now, + lastAt: now, + entries: 1, + counters: { ...entry.counters }, + identifiers: new Map(), + timer: setTimeout(() => startWrite([take(key)]), ROLLUP_WINDOW_MS), + } + collectIdentifiers(rollup, entry) + pending.set(key, rollup) +} + +/** + * Runs `work` as the owner of the rollups recorded inside it, then writes them + * and waits for those writes before returning, so work whose process may be + * frozen or ended once it returns (a background workflow run) leaves nothing + * pending. Other scopes' rollups keep their own window. Each write is one + * insert, bounded by the database's statement and connect timeouts; a failed + * one is logged, never thrown. + */ +export async function withAuditRollupScope(work: () => Promise): Promise { + const scope: RollupScope = { id: nextScopeId++, writes: new Set() } + try { + return await scopeStorage.run(scope, work) + } finally { + const keys = [...pending.entries()] + .filter(([, rollup]) => rollup.scope === scope) + .map(([key]) => key) + if (keys.length > 0) startWrite(keys.map(take)) + await Promise.all(scope.writes) + } +} + +function startWrite(rollups: PendingRollup[]): void { + const done = writePending(rollups) + const scopes = new Set(rollups.map((rollup) => rollup.scope)) + for (const scope of scopes) scope?.writes.add(done) + void done.finally(() => { + for (const scope of scopes) scope?.writes.delete(done) + }) +} + +function mergeInto(rollup: PendingRollup, entry: AuditRollupEntry): void { + rollup.entries += 1 + rollup.lastAt = new Date() + for (const [name, value] of Object.entries(entry.counters)) { + rollup.counters[name] = (rollup.counters[name] ?? 0) + value + } + collectIdentifiers(rollup, entry) +} + +function collectIdentifiers(rollup: PendingRollup, entry: AuditRollupEntry): void { + for (const [name, value] of Object.entries(entry.identifiers)) { + if (value === undefined) continue + const values = rollup.identifiers.get(name) + if (values) values.push(value) + else rollup.identifiers.set(name, [value]) + } +} + +function take(key: string): PendingRollup { + const rollup = pending.get(key) + if (!rollup) throw new Error('Audit rollup is no longer pending') + clearTimeout(rollup.timer) + pending.delete(key) + return rollup +} + +function toAuditParams(rollup: PendingRollup): AuditLogParams { + const { first } = rollup + if (rollup.entries === 1) return first.params + + const metadata: Record = { ...first.rollupMetadata, ...rollup.counters } + for (const [name, values] of rollup.identifiers) metadata[`${name}s`] = values + metadata.rollup = { + entries: rollup.entries, + firstAt: rollup.firstAt.toISOString(), + lastAt: rollup.lastAt.toISOString(), + } + return { + ...first.params, + description: first.describe({ entries: rollup.entries, counters: rollup.counters }), + metadata, + } +} + +/** + * One insert per row, so a row that cannot be written never takes others with + * it, at most {@link ROLLUP_WRITE_CONCURRENCY} at a time. + */ +async function writePending(rollups: PendingRollup[]): Promise { + for (let index = 0; index < rollups.length; index += ROLLUP_WRITE_CONCURRENCY) { + await Promise.all(rollups.slice(index, index + ROLLUP_WRITE_CONCURRENCY).map(writeRollup)) + } +} + +async function writeRollup(rollup: PendingRollup): Promise { + try { + await insertDeferredAuditEntry(toAuditParams(rollup), rollup.surface) + } catch (error) { + logger.error('Failed to record audit log rollup', { + error, + action: rollup.first.params.action, + entries: rollup.entries, + }) + } +} diff --git a/packages/auth/src/principal.ts b/packages/auth/src/principal.ts index ccd1973e3a9..c0b3978960c 100644 --- a/packages/auth/src/principal.ts +++ b/packages/auth/src/principal.ts @@ -671,6 +671,9 @@ export type PrincipalActor = serviceId: DelegatedPrincipal['serviceId'] subjectUserId?: string delegationId: string + /** The workflow, and the run of it, an executor delegation acted for. */ + workflowId?: string + executionId?: string } | { kind: 'credential_group_enrollment' @@ -791,13 +794,17 @@ export function toPrincipalActor(principal: Principal): PrincipalActor { ? { subject: principal.subject } : {}), } - case 'delegated': + case 'delegated': { + const execution = principal.serviceId === 'executor' ? principal.delegationContext : undefined return { kind: principal.kind, serviceId: principal.serviceId, ...(principal.subjectUserId ? { subjectUserId: principal.subjectUserId } : {}), delegationId: principal.delegationId, + ...(execution ? { workflowId: execution.workflowId } : {}), + ...(execution?.executionId ? { executionId: execution.executionId } : {}), } + } case 'resource_delegated': return { kind: principal.kind, diff --git a/packages/testing/src/mocks/audit.mock.ts b/packages/testing/src/mocks/audit.mock.ts index 1d7903d778b..505d7b5d8d4 100644 --- a/packages/testing/src/mocks/audit.mock.ts +++ b/packages/testing/src/mocks/audit.mock.ts @@ -273,6 +273,7 @@ const AuditResourceType = { * Controllable mock functions for `@sim/audit`. * * `mockRecordAuditOnce` resolves `undefined` by default, like the real awaited insert. + * `mockWithAuditRollupScope` runs the work it is given, like the real scope. * * @example * ```ts @@ -290,6 +291,8 @@ export const auditMockFns = { mockRecordAudit: vi.fn(), mockRecordAuditBatch: vi.fn(), mockRecordAuditOnce: vi.fn(async (_id: string, _params: unknown): Promise => undefined), + mockRecordAuditRollup: vi.fn(), + mockWithAuditRollupScope: vi.fn((work: () => Promise): Promise => work()), } /** @@ -305,6 +308,8 @@ export const auditMock = { recordAudit: auditMockFns.mockRecordAudit, recordAuditBatch: auditMockFns.mockRecordAuditBatch, recordAuditOnce: auditMockFns.mockRecordAuditOnce, + recordAuditRollup: auditMockFns.mockRecordAuditRollup, + withAuditRollupScope: auditMockFns.mockWithAuditRollupScope, /** * Real implementation, not a stub: callers under test derive their audit * metadata through it, so stubbing it would erase what the test asserts.