Configure async ClickHouse inserts for pageview events (#23274)

## Context

Pageview tracking goes through the `trackAnalytics` mutation on the
metadata API and is persisted through the unified event pipeline before
the mutation resolves.

ClickHouse inserts already use:

```text
async_insert = 1
wait_for_async_insert = 1
```

`async_insert` lets ClickHouse buffer and batch small inserts, but
`wait_for_async_insert = 1` still keeps the API request open until that
buffer is flushed successfully. For sparse pageview inserts, the buffer
timeout can therefore account for most of the request duration and
contribute to metadata API tail latency.

## What this changes

- Adds a named `ClickHouseService.insert` option for overriding
`async_insert_busy_timeout_max_ms`.
- Caps the pageview buffer wait at 100 ms.
- Keeps `wait_for_async_insert = 1`.
- Leaves workspace, object, usage, application-log, and other event
inserts on the existing default timeout.

## Why this approach

This removes the avoidable buffer wait from the pageview request path
without changing the delivery guarantees of the event pipeline.

In particular, this does **not** use `wait_for_async_insert = 0` or
fire-and-forget writes. The API still receives an acknowledgement only
after ClickHouse flushes the pageview successfully, and insert/schema
errors still propagate through the existing handling.

The 100 ms value caps only the batching wait. It does not impose a 100
ms deadline on the complete ClickHouse request.

## Expected impact

- Lower ClickHouse span duration for pageview tracking.
- Lower tail latency for metadata API requests that emit pageviews.
- No behavior or durability change for other event types.

The trade-off is that pageviews may be flushed in smaller batches. The
setting remains scoped to the pageview table so higher-value event
streams keep their current batching behavior.

## Testing

- Added coverage for the optional ClickHouse busy-timeout setting.
- Added coverage verifying that only pageview inserts receive the 100 ms
override.
- Existing insert failure/retry behavior remains covered.
- `twenty-server` typecheck passes.
- Focused test result: 23 tests passed.

## Post-deploy verification

- Compare pageview ClickHouse span p95/p99 before and after deployment.
- Compare metadata API p95/p99.
- Check ClickHouse asynchronous-insert failures.
- Watch ClickHouse part creation and merge pressure for unexpected
growth.

<!-- This is an auto-generated description by cubic. -->
<a
href="https://cubic.dev/pr/twentyhq/twenty/pull/23274?utm_source=github"
target="_blank" rel="noopener noreferrer"
data-no-image-dialog="true"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source
media="(prefers-color-scheme: light)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img
alt="Review in cubic"
src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
<!-- End of auto-generated description by cubic. -->

packages/twenty-server/src/database/clickHouse/clickHouse.service.spec.ts

@@ -112,6 +112,25 @@ describe('ClickHouseService', () => {
       });
     });
 
+    it('should allow overriding the asynchronous insert busy timeout', async () => {
+      const testData = [{ id: 1, name: 'test' }];
+      const result = await service.insert('test_table', testData, {
+        asyncInsertBusyTimeoutMaxMs: 100,
+      });
+
+      expect(result).toEqual({ success: true });
+      expect(mockClickHouseClient.insert).toHaveBeenCalledWith({
+        table: 'test_table',
+        values: testData,
+        format: 'JSONEachRow',
+        clickhouse_settings: {
+          async_insert: 1,
+          async_insert_busy_timeout_max_ms: 100,
+          wait_for_async_insert: 1,
+        },
+      });
+    });
+
     it('should return failure when clickhouse client is not defined', async () => {
       (service as any).mainClient = undefined;
 

packages/twenty-server/src/database/clickHouse/clickHouse.service.ts

@@ -13,6 +13,11 @@ import {
 
 import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
 
+export type ClickHouseInsertOptions = {
+  clientId?: string;
+  asyncInsertBusyTimeoutMaxMs?: number;
+};
+
 @Injectable()
 export class ClickHouseService implements OnModuleInit, OnModuleDestroy {
   private mainClient: ClickHouseClient | undefined;
@@ -135,11 +140,11 @@ export class ClickHouseService implements OnModuleInit, OnModuleDestroy {
   public async insert<T extends Record<string, any>>(
     table: string,
     values: T[],
-    clientId?: string,
+    options: ClickHouseInsertOptions = {},
   ): Promise<{ success: boolean }> {
     try {
-      const client = clientId
-        ? await this.connectToClient(clientId)
+      const client = options.clientId
+        ? await this.connectToClient(options.clientId)
         : this.mainClient;
 
       if (!client) {
@@ -149,6 +154,7 @@ export class ClickHouseService implements OnModuleInit, OnModuleDestroy {
       await this.insertInChunks(client, table, values, {
         chunkSize: 1000,
         maxMemoryMB: 4,
+        asyncInsertBusyTimeoutMaxMs: options.asyncInsertBusyTimeoutMaxMs,
       });
 
       return { success: true };
@@ -260,7 +266,11 @@ export class ClickHouseService implements OnModuleInit, OnModuleDestroy {
     client: ClickHouseClient,
     table: string,
     values: T[],
-    options: { chunkSize?: number; maxMemoryMB?: number } = {},
+    options: {
+      chunkSize?: number;
+      maxMemoryMB?: number;
+      asyncInsertBusyTimeoutMaxMs?: number;
+    } = {},
   ): Promise<void> {
     const chunkSize = options.chunkSize ?? 1000;
     const maxMemoryMB = options.maxMemoryMB;
@@ -276,6 +286,12 @@ export class ClickHouseService implements OnModuleInit, OnModuleDestroy {
         format: 'JSONEachRow',
         clickhouse_settings: {
           async_insert: 1,
+          ...(options.asyncInsertBusyTimeoutMaxMs !== undefined
+            ? {
+                async_insert_busy_timeout_max_ms:
+                  options.asyncInsertBusyTimeoutMaxMs,
+              }
+            : {}),
           wait_for_async_insert: 1,
         },
       });

packages/twenty-server/src/engine/core-modules/event-logs/ingest/clickhouse-event.sink.spec.ts

@@ -43,8 +43,14 @@ describe('ClickHouseEventSink', () => {
     await sink.write([first, second, applicationLog]);
 
     expect(insert).toHaveBeenCalledTimes(2);
-    expect(insert).toHaveBeenCalledWith('pageview', [first.row, second.row]);
-    expect(insert).toHaveBeenCalledWith('applicationLog', [applicationLog.row]);
+    expect(insert).toHaveBeenCalledWith('pageview', [first.row, second.row], {
+      asyncInsertBusyTimeoutMaxMs: 100,
+    });
+    expect(insert).toHaveBeenCalledWith(
+      'applicationLog',
+      [applicationLog.row],
+      undefined,
+    );
   });
 
   it('no-ops when ClickHouse is not configured', async () => {

packages/twenty-server/src/engine/core-modules/event-logs/ingest/clickhouse-event.sink.ts

@@ -1,8 +1,22 @@
 import { Injectable } from '@nestjs/common';
 
-import { ClickHouseService } from 'src/database/clickHouse/clickHouse.service';
+import {
+  type ClickHouseInsertOptions,
+  ClickHouseService,
+} from 'src/database/clickHouse/clickHouse.service';
 import { type EventSink } from 'src/engine/core-modules/event-logs/ingest/event-sink';
-import { type WorkspaceEventEnvelope } from 'src/engine/core-modules/event-logs/types/workspace-event-envelope.type';
+import {
+  type WorkspaceEventEnvelope,
+  type WorkspaceEventTable,
+} from 'src/engine/core-modules/event-logs/types/workspace-event-envelope.type';
+
+const CLICKHOUSE_INSERT_OPTIONS_BY_TABLE: Partial<
+  Record<WorkspaceEventTable, ClickHouseInsertOptions>
+> = {
+  pageview: {
+    asyncInsertBusyTimeoutMaxMs: 100,
+  },
+};
 
 @Injectable()
 export class ClickHouseEventSink implements EventSink {
@@ -13,7 +27,10 @@ export class ClickHouseEventSink implements EventSink {
       return;
     }
 
-    const rowsByTable = new Map<string, Record<string, unknown>[]>();
+    const rowsByTable = new Map<
+      WorkspaceEventTable,
+      Record<string, unknown>[]
+    >();
 
     for (const event of events) {
       const rows = rowsByTable.get(event.table) ?? [];
@@ -24,7 +41,11 @@ export class ClickHouseEventSink implements EventSink {
 
     await Promise.all(
       [...rowsByTable.entries()].map(async ([table, rows]) => {
-        const result = await this.clickHouseService.insert(table, rows);
+        const result = await this.clickHouseService.insert(
+          table,
+          rows,
+          CLICKHOUSE_INSERT_OPTIONS_BY_TABLE[table],
+        );
 
         if (!result.success) {
           throw new Error(