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 type { StorageContext } from '@/lib/uploads/config'
 import { hasCloudStorage } from '@/lib/uploads/core/storage-service'
 import { verifyFileAccess } from '@/app/api/files/authorization'
@@ -84,10 +86,26 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
 
     logger.info(`Generated download URL for ${storageContext ?? 'inferred'} 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)