diff --git a/packages/db/src/buffers/session-buffer.ts b/packages/db/src/buffers/session-buffer.ts index 0fc221ad..701ee81c 100644 --- a/packages/db/src/buffers/session-buffer.ts +++ b/packages/db/src/buffers/session-buffer.ts @@ -191,14 +191,23 @@ export class SessionBuffer extends BaseBuffer { if (events.length === 0) return; - const sessions = events.map((e) => getSafeJson(e)); + const sessions = events + .map((e) => getSafeJson(e)) + .map((session) => { + return { + ...session, + duration: Math.max(0, session?.duration || 0), + }; + }); - // Insert to ClickHouse - await ch.insert({ - table: TABLE_NAMES.sessions, - values: sessions, - format: 'JSONEachRow', - }); + for (const chunk of this.chunks(sessions, 1000)) { + // Insert to ClickHouse + await ch.insert({ + table: TABLE_NAMES.sessions, + values: chunk, + format: 'JSONEachRow', + }); + } // Only remove events after successful insert await this.redis.ltrim(this.redisKey, events.length, -1);