Method processBuffer

packages/db/src/buffers/bot-buffer.ts:46–80

()

  }

  async processBuffer() {
    const lrangeStart = performance.now();
    const events = await this.redis.lrange(
      this.redisKey,
      0,
      this.batchSize - 1
    );
    const lrangeMs = performance.now() - lrangeStart;

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

    // Raw passthrough: each Redis entry is already a valid JSONEachRow
    // line. Streaming raw strings skips JSON.parse + the client's
    // re-stringify on the hot path.
    const chStart = performance.now();
    await ch.insert({
      table: TABLE_NAMES.events_bots,
      values: this.jsonEachRowStream(events),
      format: 'JSONEachRow',
      clickhouse_settings: this.getClickhouseSettings(),
    });
    const chInsertMs = performance.now() - chStart;

    const trimStart = performance.now();
    await this.redis.ltrim(this.redisKey, events.length, -1);
    const trimMs = performance.now() - trimStart;

    this.reportFlushStats({
      rowsProcessed: events.length,
      phases: { lrangeMs, chInsertMs, trimMs },
    });
  }
}