Method processBuffer

packages/db/src/buffers/event-buffer.ts:176–252

()

  }

  async processBuffer() {
    const redis = getRedisCache();

    const lrangeStart = performance.now();
    const queueEvents = await redis.lrange(
      this.queueKey,
      0,
      this.batchSize - 1
    );
    const lrangeMs = performance.now() - lrangeStart;

    if (queueEvents.length === 0) {
      this.reportFlushStats({ rowsProcessed: 0, phases: { lrangeMs } });
      return;
    }

    // We don't need to JSON.parse the events at all — they're already
    // valid JSONEachRow lines (one stringified event per Redis entry).
    // The client's custom `json.stringify` (set in CLICKHOUSE_OPTIONS)
    // passes strings through unchanged, so the bytes go straight from
    // Redis → CH HTTP body. This skips:
    //   - JSON.parse × N (50–300ms for N=100k)
    //   - The @clickhouse/client's internal JSON.stringify × N (same)
    //   - All the intermediate object allocations (saves ~200MB heap)
    //
    // We still need `project_id` per row for the per-project pub/sub.
    // extractProjectId() does an indexOf-based fast path that's ~50×
    // faster than JSON.parse, and falls back to a real parse on the
    // rare line where `project_id` appears more than once (e.g. a
    // user-supplied `properties.project_id`) — so the count is always
    // attributed to the top-level field, never a nested one.
    const countByProject = new Map<string, number>();
    const yieldEvery = this.getYieldInterval(queueEvents.length, {
      min: 1000,
      max: 5000,
    });
    for (let i = 0; i < queueEvents.length; i++) {
      const projectId = extractProjectId(queueEvents[i]!);
      if (projectId) {
        countByProject.set(
          projectId,
          (countByProject.get(projectId) ?? 0) + 1,
        );
      }
      if ((i + 1) % yieldEvery === 0) {
        await this.yieldToEventLoop();
      }
    }

    const chStart = performance.now();
    await this.parallelLimit(
      this.chunks(queueEvents, this.chunkSize),
      (chunk) =>
        ch.insert({
          table: 'events',
          // Stream the raw JSONEachRow lines straight through — already
          // serialized in Redis, no client-side parse/stringify needed.
          values: this.jsonEachRowStream(chunk),