Function importJob

apps/worker/src/jobs/import.ts:71–378

(job: Job<ImportQueuePayload>)

}

export async function importJob(job: Job<ImportQueuePayload>) {
  const { importId } = job.data.payload;

  const record = await db.import.findUniqueOrThrow({
    where: { id: importId },
    include: { project: true },
  });

  const jobLogger = logger.child({ importId, config: record.config });
  jobLogger.info('Starting import job');

  const providerInstance = createProvider(record, jobLogger);
  const shouldGenerateSessionIds = providerInstance.shouldGenerateSessionIds();

  try {
    const isRetry = record.currentStep !== null;
    const canResume =
      isRetry && RESUMABLE_STEPS.includes(record.currentStep as string);

    // -------------------------------------------------------
    // STAGING PHASE: clean slate on failure, run from scratch
    // -------------------------------------------------------
    if (!canResume) {
      if (isRetry) {
        jobLogger.info(
          'Retry detected before resumable phase — cleaning staging data'
        );
        await cleanupStagingData(importId);
      }

      // Phase 1: Load events into staging
      await updateImportStatus(jobLogger, job, importId, { step: 'loading' });

      const totalEvents = await providerInstance
        .getTotalEventsCount()
        .catch(() => -1);
      let processedEvents = 0;
      const eventBatch: IClickhouseEvent[] = [];

      // Profiles derived inline from events (e.g. Amplitude, which has no
      // profile export API). A bounded dedup map keeps memory flat regardless of
      // event volume; the profiles table's ReplacingMergeTree(last_seen_at)
      // collapses any duplicate rows that span flush boundaries to the latest
      // activity per id.
      const canProfileFromEvents =
        typeof (providerInstance as AmplitudeProvider)
          .transformEventToProfile === 'function';
      const PROFILE_MAP_CAP = 50_000;
      const profileMap = new Map<string, IClickhouseProfile>();
      let processedProfiles = 0;

      const flushProfiles = async () => {
        if (profileMap.size === 0) {
          return;
        }
        const values = Array.from(profileMap.values());
        await insertProfilesBatch(values, record.projectId);
        processedProfiles += values.length;