Function appendEvents

src/node/services/analytics/etl.ts:167–219

(conn: DuckDBConnection, events: IngestEvent[])

}

export async function appendEvents(conn: DuckDBConnection, events: IngestEvent[]): Promise<void> {
  assert(INSERT_EVENT_SQL.length > 0, "appendEvents: expected INSERT_EVENT_SQL to remain defined");

  if (events.length === 0) {
    return;
  }

  const appender = await conn.createAppender("events");

  try {
    assert(
      appender instanceof DuckDBAppender,
      "appendEvents: expected createAppender to return a DuckDBAppender"
    );

    for (const event of events) {
      const row = event.row;
      appendVarcharOrNull(appender, row.workspace_id);
      appendVarcharOrNull(appender, row.project_path);
      appendVarcharOrNull(appender, row.project_name);
      appendVarcharOrNull(appender, row.workspace_name);
      appendVarcharOrNull(appender, row.parent_workspace_id);
      appendVarcharOrNull(appender, row.agent_id);
      appendBigIntOrNull(appender, row.timestamp);
      appendDateOrNull(appender, event.date);
      appendVarcharOrNull(appender, row.model);
      appendVarcharOrNull(appender, row.thinking_level);
      appender.appendInteger(row.input_tokens);
      appender.appendInteger(row.output_tokens);
      appender.appendInteger(row.reasoning_tokens);
      appender.appendInteger(row.cached_tokens);
      appender.appendInteger(row.cache_create_tokens);
      appender.appendDouble(row.input_cost_usd);
      appender.appendDouble(row.output_cost_usd);
      appender.appendDouble(row.reasoning_cost_usd);
      appender.appendDouble(row.cached_cost_usd);
      appender.appendDouble(row.total_cost_usd);
      appendDoubleOrNull(appender, row.duration_ms);
      appendDoubleOrNull(appender, row.ttft_ms);
      appendDoubleOrNull(appender, row.streaming_ms);
      appendDoubleOrNull(appender, row.tool_execution_ms);
      appendDoubleOrNull(appender, row.output_tps);
      appendIntegerOrNull(appender, row.response_index);
      appender.appendBoolean(row.is_sub_agent);
      appendVarcharOrNull(appender, row.tool_name);
      appender.endRow();
    }

    appender.flushSync();
  } finally {
    appender.closeSync();
  }
}

const INSERT_DELEGATION_ROLLUP_SQL = `
INSERT OR REPLACE INTO delegation_rollups (