Function insertProfilesBatch

packages/db/src/services/import.service.ts:142–174

(
  profiles: IClickhouseProfile[],
  projectId: string
)

 * Used by Mixpanel (and other providers) to import user profiles during an import job.
 */
export async function insertProfilesBatch(
  profiles: IClickhouseProfile[],
  projectId: string
): Promise<{ inserted: number }> {
  if (profiles.length === 0) {
    return { inserted: 0 };
  }

  const normalized = profiles.map((p) => ({
    id: p.id,
    project_id: projectId,
    first_name: p.first_name ?? '',
    last_name: p.last_name ?? '',
    email: p.email ?? '',
    avatar: p.avatar ?? '',
    is_external: p.is_external ?? true,
    properties: Object.fromEntries(
      Object.entries(p.properties || {}).filter(
        (kv): kv is [string, string] => kv[1] != null && kv[1] !== ''
      )
    ) as Record<string, string>,
    created_at: p.created_at,
    last_seen_at: p.last_seen_at ?? p.created_at,
  }));

  await ch.insert({
    table: TABLE_NAMES.profiles,
    values: normalized,
    format: 'JSONEachRow',
  });

  return { inserted: normalized.length };
}

/**
 * Delete all staging data for an import. Used to get a clean slate on retry