feat(observability): extend audit log, PostHog, and storage metering coverage

apps/realtime/src/handlers/variables.ts

@@ -1,3 +1,4 @@
+import { AuditAction, AuditResourceType, recordAudit } from '@sim/audit'
 import { db } from '@sim/db'
 import { workflow } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
@@ -18,6 +19,9 @@ type PendingVariable = {
  latest: { variableId: string; field: string; value: any; timestamp: number }
  timeout: NodeJS.Timeout
  opToSocket: Map<string, string>
+ /** Most recent writer, used as the audit actor when the coalesced update is persisted. */
+ actorId: string
+ actorName?: string
 }
 
 // Keyed by `${workflowId}:${variableId}:${field}`
@@ -177,6 +181,8 @@ export function setupVariablesHandlers(socket: AuthenticatedSocket, roomManager:
  if (existing) {
  clearTimeout(existing.timeout)
  existing.latest = { variableId, field, value, timestamp }
+ existing.actorId = session.userId
+ existing.actorName = session.userName
  if (operationId) existing.opToSocket.set(operationId, socket.id)
  existing.timeout = setTimeout(async () => {
  await flushVariableUpdate(workflowId, existing, roomManager)
@@ -196,6 +202,8 @@ export function setupVariablesHandlers(socket: AuthenticatedSocket, roomManager:
  latest: { variableId, field, value, timestamp },
  timeout,
  opToSocket,
+ actorId: session.userId,
+ actorName: session.userName,
  })
  }
  } catch (error) {
@@ -231,7 +239,11 @@ async function flushVariableUpdate(
 
  try {
  const workflowExists = await db
- .select({ id: workflow.id })
+ .select({
+ id: workflow.id,
+ name: workflow.name,
+ workspaceId: workflow.workspaceId,
+ })
  .from(workflow)
  .where(eq(workflow.id, workflowId))
  .limit(1)
@@ -294,6 +306,19 @@ async function flushVariableUpdate(
  })
 
  if (updateSuccessful) {
+ const workflowRow = workflowExists[0]
+ recordAudit({
+ workspaceId: workflowRow.workspaceId ?? null,
+ actorId: pending.actorId,
+ actorName: pending.actorName,
+ action: AuditAction.WORKFLOW_VARIABLES_UPDATED,
+ resourceType: AuditResourceType.WORKFLOW,
+ resourceId: workflowId,
+ resourceName: workflowRow.name ?? undefined,
+ description: `Updated workflow variables`,
+ metadata: { variableId, field },
+ })
+
  // Broadcast to room excluding all senders (works cross-pod via Redis adapter)
  const senderSocketIds = [...pending.opToSocket.values()]
  const broadcastPayload = {

apps/sim/app/api/auth/oauth/token/route.ts

@@ -1,3 +1,4 @@
+import { AuditAction, AuditResourceType, recordAudit } from '@sim/audit'
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
 import { type NextRequest, NextResponse } from 'next/server'
@@ -11,6 +12,7 @@ import { AuthType, checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { ATLASSIAN_SERVICE_ACCOUNT_PROVIDER_ID } from '@/lib/oauth/types'
+import { captureServerEvent } from '@/lib/posthog/server'
 import {
  getAtlassianServiceAccountSecret,
  getCredential,
@@ -96,6 +98,24 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
  )
  }
 
+ recordAudit({
+ actorId: auth.userId,
+ action: AuditAction.CREDENTIAL_ACCESSED,
+ resourceType: AuditResourceType.CREDENTIAL,
+ resourceId: providerId,
+ description: `Accessed OAuth credential for provider ${providerId}`,
+ metadata: {
+ provider: providerId,
+ credentialType: 'oauth',
+ credentialAccountUserId,
+ },
+ request,
+ })
+ captureServerEvent(auth.userId, 'credential_used', {
+ credential_type: 'oauth',
+ provider_id: providerId,
+ })
+
  return NextResponse.json({ accessToken }, { status: 200 })
  } catch (error) {
  const message = getErrorMessage(error, 'Failed to get OAuth token')
@@ -120,9 +140,39 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
  return NextResponse.json({ error: authz.error || 'Unauthorized' }, { status: 403 })
  }
 
+ const saActorId = authz.requesterUserId
+ const saWorkspaceId = resolved.workspaceId ?? authz.workspaceId ?? null
+ const emitServiceAccountAccess = () => {
+ if (!saActorId) return
+ recordAudit({
+ workspaceId: saWorkspaceId,
+ actorId: saActorId,
+ action: AuditAction.CREDENTIAL_ACCESSED,
+ resourceType: AuditResourceType.CREDENTIAL,
+ resourceId: resolved.credentialId ?? credentialId,
+ description: `Accessed service account credential for provider ${resolved.providerId ?? 'unknown'}`,
+ metadata: {
+ provider: resolved.providerId,
+ credentialType: 'service_account',
+ },
+ request,
+ })
+ captureServerEvent(
+ saActorId,
+ 'credential_used',
+ {
+ credential_type: 'service_account',
+ provider_id: resolved.providerId ?? 'unknown',
+ ...(saWorkspaceId ? { workspace_id: saWorkspaceId } : {}),
+ },
+ saWorkspaceId ? { groups: { workspace: saWorkspaceId } } : undefined
+ )
+ }
+
  try {
  if (resolved.providerId === ATLASSIAN_SERVICE_ACCOUNT_PROVIDER_ID) {
  const secret = await getAtlassianServiceAccountSecret(resolved.credentialId)
+ emitServiceAccountAccess()
  return NextResponse.json(
  {
  accessToken: secret.apiToken,
@@ -137,6 +187,7 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
  scopes ?? [],
  impersonateEmail
  )
+ emitServiceAccountAccess()
  return NextResponse.json({ accessToken }, { status: 200 })
  } catch (error) {
  logger.error(`[${requestId}] Service account token error:`, error)
@@ -165,13 +216,42 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
  return NextResponse.json({ error: 'Credential not found' }, { status: 404 })
  }
 
+ const oauthActorId = authz.requesterUserId
+ const oauthWorkspaceId = authz.workspaceId ?? null
+
  try {
  const { accessToken } = await refreshTokenIfNeeded(
  requestId,
  credential,
  resolvedCredentialId
  )
 
+ if (oauthActorId) {
+ recordAudit({
+ workspaceId: oauthWorkspaceId,
+ actorId: oauthActorId,
+ action: AuditAction.CREDENTIAL_ACCESSED,
+ resourceType: AuditResourceType.CREDENTIAL,
+ resourceId: resolvedCredentialId,
+ description: `Accessed OAuth credential for provider ${credential.providerId}`,
+ metadata: {
+ provider: credential.providerId,
+ credentialType: 'oauth',
+ },
+ request,
+ })
+ captureServerEvent(
+ oauthActorId,
+ 'credential_used',
+ {
+ credential_type: 'oauth',
+ provider_id: credential.providerId,
+ ...(oauthWorkspaceId ? { workspace_id: oauthWorkspaceId } : {}),
+ },
+ oauthWorkspaceId ? { groups: { workspace: oauthWorkspaceId } } : undefined
+ )
+ }
+
  let instanceUrl: string | undefined
  if (credential.providerId === 'salesforce' && credential.scope) {
  const instanceMatch = credential.scope.match(SALESFORCE_INSTANCE_URL_REGEX)
@@ -247,13 +327,42 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
  return NextResponse.json({ error: 'No access token available' }, { status: 400 })
  }
 
+ const actorId = authz.requesterUserId
+ const workspaceId = authz.workspaceId ?? null
+
  try {
  const { accessToken } = await refreshTokenIfNeeded(
  requestId,
  credential,
  resolvedCredentialId
  )
 
+ if (actorId) {
+ recordAudit({
+ workspaceId,
+ actorId,
+ action: AuditAction.CREDENTIAL_ACCESSED,
+ resourceType: AuditResourceType.CREDENTIAL,
+ resourceId: resolvedCredentialId,
+ description: `Accessed OAuth credential for provider ${credential.providerId}`,
+ metadata: {
+ provider: credential.providerId,
+ credentialType: 'oauth',
+ },
+ request,
+ })
+ captureServerEvent(
+ actorId,
+ 'credential_used',
+ {
+ credential_type: 'oauth',
+ provider_id: credential.providerId,
+ ...(workspaceId ? { workspace_id: workspaceId } : {}),
+ },
+ workspaceId ? { groups: { workspace: workspaceId } } : undefined
+ )
+ }
+
  // For Salesforce, extract instanceUrl from the scope field
  let instanceUrl: string | undefined
  if (credential.providerId === 'salesforce' && credential.scope) {

apps/sim/app/api/billing/switch-plan/route.ts

@@ -1,3 +1,4 @@
+import { AuditAction, AuditResourceType, recordAudit } from '@sim/audit'
 import { db } from '@sim/db'
 import { subscription as subscriptionTable } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
@@ -174,6 +175,29 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
  { set: { plan: targetPlanName } }
  )
 
+ if (isOrgScopedSubscription(sub, userId)) {
+ recordAudit({
+ actorId: userId,
+ action: AuditAction.ORG_PLAN_CONVERTED,
+ resourceType: AuditResourceType.ORGANIZATION,
+ resourceId: sub.referenceId,
+ description: `Plan converted from ${sub.plan ?? 'unknown'} to ${targetPlanName}`,
+ metadata: {
+ organizationId: sub.referenceId,
+ subscriptionId: sub.id,
+ fromPlan: sub.plan,
+ toPlan: targetPlanName,
+ interval: targetInterval,
+ },
+ request,
+ })
+ captureServerEvent(userId, 'plan_converted', {
+ organization_id: sub.referenceId,
+ from_plan: sub.plan ?? 'unknown',
+ to_plan: targetPlanName,
+ })
+ }
+
  return NextResponse.json({ success: true, plan: targetPlanName, interval: targetInterval })
  } catch (error) {
  logger.error('Failed to switch subscription', {

apps/sim/app/api/environment/route.ts

@@ -14,6 +14,7 @@ import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { syncPersonalEnvCredentialsForUser } from '@/lib/credentials/environment'
 import type { EnvironmentVariable } from '@/lib/environment/api'
+import { captureServerEvent } from '@/lib/posthog/server'
 
 const logger = createLogger('EnvironmentAPI')
 
@@ -89,6 +90,11 @@ export const POST = withRouteHandler(async (req: NextRequest) => {
  request: req,
  })
 
+ captureServerEvent(session.user.id, 'environment_updated', {
+ key_count: Object.keys(variables).length,
+ scope: 'personal',
+ })
+
  return NextResponse.json({ success: true })
  } catch (error) {
  logger.error(`[${requestId}] Error updating environment variables`, error)

apps/sim/app/api/files/download/route.ts

@@ -1,9 +1,11 @@
+import { AuditAction, AuditResourceType, recordAudit } from '@sim/audit'
 import { createLogger } from '@sim/logger'
 import { type NextRequest, NextResponse } from 'next/server'
 import { fileDownloadContract } from '@/lib/api/contracts/storage-transfer'
 import { getValidationErrorMessage, parseRequest } from '@/lib/api/server'
 import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { captureServerEvent } from '@/lib/posthog/server'
 import { hasCloudStorage } from '@/lib/uploads/core/storage-service'
 import { inferContextFromKey } from '@/lib/uploads/utils/file-utils'
 import { verifyFileAccess } from '@/app/api/files/authorization'
@@ -80,10 +82,26 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
 
  logger.info(`Generated download URL for ${storageContext} file: ${key}`)
 
+ const downloadName = name || key.split('/').pop() || 'download'
+ recordAudit({
+ workspaceId: null,
+ actorId: userId,
+ action: AuditAction.FILE_DOWNLOADED,
+ resourceType: AuditResourceType.FILE,
+ resourceName: downloadName,
+ description: `Downloaded file "${downloadName}"`,
+ metadata: { key, fileName: downloadName, context: storageContext },
+ request,
+ })
+ captureServerEvent(userId, 'file_downloaded', {
+ is_bulk: false,
+ file_count: 1,
+ })
+
  return NextResponse.json({
  downloadUrl,
  expiresIn: null,
- fileName: name || key.split('/').pop() || 'download',
+ fileName: downloadName,
  })
  } catch (error) {
  logger.error('Error in file download endpoint:', error)