feat(server): unify workspace-event ingestion behind one EventSink pipeline (#21197)

## Why The five event-log streams (`workspaceEvent`, `pageview`, `objectEvent`, `usageEvent`, `applicationLog`) each wrote to ClickHouse through their own fire-and-forget writer (`AuditService`, `UsageEventWriterService`, and the `application-logs` driver), with the per-type knowledge (table names, normalization, access rules) spread across several modules. Three of them reimplemented the same ClickHouse insert, and the read side, the live stream, and the producers lived in different modules under two different names. This consolidates them into one `core-modules/event-logs/` subsystem (emit, write, live, read), with the per-type config in a single registry so adding an event type is roughly one file. The base Logs settings tab and free application logs shipped separately in #21180 (merged). This PR adds the unified backend, the registry, and the viewer's live mode and entitlement gating. ## Pipeline ```mermaid flowchart TB subgraph PROD["Producers"] A["auth, billing, impersonation,<br/>webhook, custom-domain"] U["usage listener"] F["logic-function executor (app logs)"] R["record CRUD (entity events)"] end EM["EventLogEmitterService<br/>createContext().insert* / dispatch()"] EQ(["entityEventsToDbQueue<br/>(existing, shared with timeline)"]) CIE["CreateEventLogFromInternalEvent"] SINK["WorkspaceEventSinkService.ingest()"] C1["ClickHouseEventSink"] C2["ConsoleEventSink"] LIVE["EventLogLiveService.publishWatched()<br/>(presence-gated)"] CH[("ClickHouse, 5 tables, async_insert")] CHAN(["WORKSPACE_EVENTS_CHANNEL"]) RS["EventLogsService (registry-driven read)"] LR["EventLogsLiveResolver"] UI["Settings > Logs"] A --> EM U --> EM F --> EM EM -->|direct| SINK R --> EQ --> CIE -->|ingest| SINK SINK --> C1 --> CH SINK --> C2 SINK --> LIVE -.->|if a viewer is watching| CHAN --> LR --> UI CH --> RS --> UI ``` ## What it does - Producers call `EventLogEmitterService.createContext().insert*()`, which builds a typed `WorkspaceEventEnvelope` and writes it through `WorkspaceEventSinkService` to the configured sinks (ClickHouse, Console) plus a presence-gated live fan-out. Record/CRUD events reach the same sink through the existing `entityEventsToDbQueue`. There is no dedicated queue; ClickHouse `async_insert` batches server-side. Writes are best-effort, as on main today. - `EVENT_LOG_TYPES[table]` is the per-type source of truth: the ClickHouse table, the required entitlement, the free-text filter column, and the row-to-GraphQL mapping. Read row shapes derive from the write rows. - Four modules along their dependency boundaries: `EventLogEmitterModule` (producer API), `EventLogIngestionModule` (sink layer), `EventLogLiveModule` (fan-out), and `EventLogsViewerModule` (the entitlement-gated GraphQL read, which is where billing/enterprise/permissions stay so producers stay light). - Logs viewer: per-table columns, filters (text, date, record), live mode, and an upgrade card that points to Billing on Cloud or the Admin Panel on self-hosted. Application logs are free on every plan; the other four require the `AUDIT_LOGS` entitlement (with a `NO_ENTITLEMENT` fallback to the upgrade card). - Renames `AuditService` to `EventLogEmitterService`, and the generic `Monitoring` event to a typed `Impersonation` event (`level` + `action`). - Removes `UsageEventWriterService`, the `application-logs` driver/module, and `AuditService`'s direct inserts. ## Durability Writes are best-effort, the same as main today (the old writers were fire-and-forget). A dedicated queue was tried mid-PR and removed: `async_insert` already batches server-side, so the queue only added durability, which isn't a requirement right now. The `EventSink` seam keeps a durable transport (e.g. a Redis-Streams buffer) easy to add later without touching producers. ## Out of scope S3 peer sink (seam only), Postgres or any second read path, `ReplicatedMergeTree`, ClickHouse table-schema changes, and the record-data `EVENT_STREAM_CHANNEL` (unchanged, separate concern). ## Testing Unit tests cover the registry definitions and row normalization, the entitlement gating, the envelope builders, and the producers. Integration tests cover the write paths (record create produces an `objectEvent`; the track mutation produces a `workspaceEvent`) and the read/query path across all five tables. Verified with typecheck, lint, a server boot, and GraphQL/SDK codegen.
2026-06-06 10:32:56 +02:00
parent 6c65d26ced
commit 91f2f08995
148 changed files with 2919 additions and 2132 deletions
@@ -2,7 +2,8 @@ import { Module } from '@nestjs/common';
 import { TypeOrmModule } from '@nestjs/typeorm';

 import { ApplicationRegistrationVariableEntity } from 'src/engine/core-modules/application/application-registration-variable/application-registration-variable.entity';
-import { AuditModule } from 'src/engine/core-modules/audit/audit.module';
+import { EventLogEmitterModule } from 'src/engine/core-modules/event-logs/emit/event-log-emitter.module';
+import { EventLogLiveModule } from 'src/engine/core-modules/event-logs/live/event-log-live.module';
 import { TokenModule } from 'src/engine/core-modules/auth/token/token.module';
 import { BillingModule } from 'src/engine/core-modules/billing/billing.module';
 import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
@@ -15,7 +16,8 @@ import { WorkspaceCacheModule } from 'src/engine/workspace-cache/workspace-cache
@Module({
  imports: [
    ThrottlerModule,
-    AuditModule,
+    EventLogEmitterModule,
+    EventLogLiveModule,
    TokenModule,
    SecretEncryptionModule,
    SubscriptionsModule,
@@ -17,13 +17,13 @@ import {
  type LogicFunctionTranspileResult,
 } from 'src/engine/core-modules/logic-function/logic-function-drivers/interfaces/logic-function-driver.interface';

-import { ApplicationLogsService } from 'src/engine/core-modules/application-logs/application-logs.service';
-import { parseApplicationLogLines } from 'src/engine/core-modules/application-logs/utils/parse-application-log-lines';
+import { buildApplicationLogEnvelopes } from 'src/engine/core-modules/event-logs/producers/application-log/build-application-log-envelopes';
+import { parseApplicationLogLines } from 'src/engine/core-modules/event-logs/producers/application-log/parse-application-log-lines';
 import { ApplicationRegistrationVariableEntity } from 'src/engine/core-modules/application/application-registration-variable/application-registration-variable.entity';
 import type { FlatApplicationVariable } from 'src/engine/metadata-modules/flat-application-variable/types/flat-application-variable.type';
 import { FlatApplication } from 'src/engine/core-modules/application/types/flat-application.type';
-import { AuditService } from 'src/engine/core-modules/audit/services/audit.service';
-import { LOGIC_FUNCTION_EXECUTED_EVENT } from 'src/engine/core-modules/audit/utils/events/workspace-event/logic-function/logic-function-executed';
+import { EventLogEmitterService } from 'src/engine/core-modules/event-logs/emit/event-log-emitter.service';
+import { LOGIC_FUNCTION_EXECUTED_EVENT } from 'src/engine/core-modules/event-logs/emit/events/workspace-event/logic-function/logic-function-executed';
 import { ApplicationTokenService } from 'src/engine/core-modules/auth/token/services/application-token.service';
 import { BillingUsageService } from 'src/engine/core-modules/billing/services/billing-usage.service';
 import { BillingService } from 'src/engine/core-modules/billing/services/billing.service';
@@ -47,6 +47,7 @@ import {
 import { FlatLogicFunction } from 'src/engine/metadata-modules/logic-function/types/flat-logic-function.type';
 import { SubscriptionChannel } from 'src/engine/subscriptions/enums/subscription-channel.enum';
 import { SubscriptionService } from 'src/engine/subscriptions/subscription.service';
+import { EventLogLiveService } from 'src/engine/core-modules/event-logs/live/event-log-live.service';
 import { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service';
 import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter';
 import { cleanServerUrl } from 'src/utils/clean-server-url';
@@ -78,8 +79,8 @@ export class LogicFunctionExecutorService {
    private readonly applicationTokenService: ApplicationTokenService,
    private readonly secretEncryptionService: SecretEncryptionService,
    private readonly subscriptionService: SubscriptionService,
-    private readonly auditService: AuditService,
-    private readonly applicationLogsService: ApplicationLogsService,
+    private readonly eventLogLiveService: EventLogLiveService,
+    private readonly eventLogEmitterService: EventLogEmitterService,
    private readonly workspaceEventEmitter: WorkspaceEventEmitter,
    private readonly billingService: BillingService,
    private readonly billingUsageService: BillingUsageService,
@@ -333,22 +334,11 @@ export class LogicFunctionExecutorService {
      [DEFAULT_APP_ACCESS_TOKEN_NAME]: applicationAccessToken.token,
      [DEFAULT_API_KEY_NAME]: applicationAccessToken.token,
      APPLICATION_ID: flatApplication.id,
-      // Server variables first, workspace variables override. Workspace-level
-      // values let a specific tenant customize a server default.
      ...serverVariables,
      ...workspaceVariables,
    };
  }

-  // Resolves encrypted server-level variables (ApplicationRegistrationVariable)
-  // for the application's registration. Returns an empty object when the
-  // application isn't linked to a registration (legacy LOCAL apps).
-  //
-  // Runs on every logic function execution — the query is indexed on
-  // applicationRegistrationId and filters unfilled rows server-side. Most
-  // apps have 0-3 server variables so the round-trip is cheap, but if this
-  // becomes a hot path, move to a WorkspaceCacheProvider mirroring
-  // WorkspaceApplicationVariableMapCacheService.
  private async buildServerVariableEnvMap(
    applicationRegistrationId: string | null,
  ): Promise<Record<string, string>> {
@@ -366,16 +356,6 @@ export class LogicFunctionExecutorService {

    const envMap: Record<string, string> = {};

-    // ApplicationRegistrationVariable.encryptedValue is always written
-    // encrypted (ApplicationRegistrationVariableService.createVariable and
-    // .updateVariable call encrypt unconditionally), independent of
-    // `isSecret`. `isSecret` is display metadata — the storage contract is
-    // not conditional, so decryption isn't either.
-    //
-    // Registration variables are server-level config — any installed
-    // application across any workspace must be able to read them — so they
-    // use the instance-scoped versioned envelope (no workspaceId in the HKDF
-    // info).
    for (const variable of serverVariables) {
      if (variable.encryptedValue !== '') {
        envMap[variable.key] = this.secretEncryptionService.decryptVersioned(
@@ -387,6 +367,46 @@ export class LogicFunctionExecutorService {
    return envMap;
  }

+  private async publishLogicFunctionLogsToCli({
+    result,
+    flatApplication,
+    flatLogicFunction,
+    workspaceId,
+  }: {
+    result: LogicFunctionExecuteResult;
+    workspaceId: string;
+    flatLogicFunction: FlatLogicFunction;
+    flatApplication: FlatApplication;
+  }): Promise<void> {
+    try {
+      const isWatched = await this.eventLogLiveService.isWatched(
+        workspaceId,
+        SubscriptionChannel.LOGIC_FUNCTION_LOGS_CHANNEL,
+      );
+
+      if (!isWatched) {
+        return;
+      }
+
+      await this.subscriptionService.publish({
+        channel: SubscriptionChannel.LOGIC_FUNCTION_LOGS_CHANNEL,
+        workspaceId,
+        payload: {
+          logicFunctionLogs: {
+            logs: result.logs,
+            id: flatLogicFunction.id,
+            name: flatLogicFunction.name,
+            universalIdentifier: flatLogicFunction.universalIdentifier,
+            applicationId: flatApplication.id,
+            applicationUniversalIdentifier: flatApplication.universalIdentifier,
+          },
+        },
+      });
+    } catch (error) {
+      this.logger.error('Failed to publish logic function logs', error);
+    }
+  }
+
  private async handleExecutionResult({
    result,
    flatApplication,
@@ -410,26 +430,22 @@ export class LogicFunctionExecutorService {
      executionId,
    }));

-    void this.applicationLogsService.writeLogs(logEntries).catch((error) => {
-      this.logger.error('Failed to persist application logs', error);
-    });
+    if (this.eventLogEmitterService.isEnabled()) {
+      void this.eventLogEmitterService
+        .dispatch(buildApplicationLogEnvelopes(logEntries))
+        .catch((error) => {
+          this.logger.error('Failed to record application logs', error);
+        });
+    }

-    await this.subscriptionService.publish({
-      channel: SubscriptionChannel.LOGIC_FUNCTION_LOGS_CHANNEL,
+    void this.publishLogicFunctionLogsToCli({
+      result,
+      flatApplication,
+      flatLogicFunction,
      workspaceId,
-      payload: {
-        logicFunctionLogs: {
-          logs: result.logs,
-          id: flatLogicFunction.id,
-          name: flatLogicFunction.name,
-          universalIdentifier: flatLogicFunction.universalIdentifier,
-          applicationId: flatApplication.id,
-          applicationUniversalIdentifier: flatApplication.universalIdentifier,
-        },
-      },
    });

-    void this.auditService
+    void this.eventLogEmitterService
      .createContext({
        workspaceId,
      })