fix: fixes for mixpanel specific issues and how we handle should run import step

lindesvard · lindesvard · commit b69a0a27a0e8 · 2025-10-30T14:30:27.000+01:00
diff --git a/apps/worker/src/jobs/import.ts b/apps/worker/src/jobs/import.ts
@@ -89,13 +89,19 @@ export async function importJob(job: Job<ImportQueuePayload>) {
       (record.currentStep === 'backfilling_sessions' && record.currentBatch) ||
       undefined;
 
+    // Example:
+    // shouldRunStep(0) // currStep = 2 (should not run)
+    // shouldRunStep(1) // currStep = 2 (should not run)
+    // shouldRunStep(2) // currStep = 2 (should run)
+    // shouldRunStep(3) // currStep = 2 (should run)
     const shouldRunStep = (step: ValidStep) => {
       if (isNewImport) {
         return true;
       }
 
-      const currentStepIndex = steps[step];
-      return currentStepIndex > steps[record.currentStep as ValidStep];
+      const stepToRunIndex = steps[step];
+      const currentStepIndex = steps[record.currentStep as ValidStep];
+      return stepToRunIndex >= currentStepIndex;
     };
 
     async function whileBounds(
diff --git a/packages/db/src/services/import.service.ts b/packages/db/src/services/import.service.ts
@@ -56,11 +56,7 @@ export async function insertImportBatch(
       csvEscapeField(event.referrer_type || ''),
       csvEscapeField(event.duration ?? 0),
       csvEscapeJson(event.properties),
-      csvEscapeField(
-        event.created_at
-          ? formatClickhouseDate(event.created_at)
-          : formatClickhouseDate(new Date()),
-      ),
+      csvEscapeField(event.created_at),
       csvEscapeField(event.country || ''),
       csvEscapeField(event.city || ''),
       csvEscapeField(event.region || ''),
@@ -76,7 +72,7 @@ export async function insertImportBatch(
       csvEscapeField('\\N'), // imported_at (Nullable)
       csvEscapeField(importId),
       csvEscapeField('pending'), // import_status
-      csvEscapeField(formatClickhouseDate(new Date()).replace(/\.\d{3}$/, '')), // imported_at_meta (DateTime, not DateTime64, so no milliseconds)
+      csvEscapeField(formatClickhouseDate(new Date())), // imported_at_meta (DateTime, not DateTime64, so no milliseconds)
     ];
     return fields.join(',');
   });
@@ -326,7 +322,7 @@ export async function createSessionsStartEndEvents(
         string,
         string | number | boolean | null | undefined
       >, // From last event
-      created_at: adjustTimestamp(session.last_timestamp, 1000), // 1 second after last event
+      created_at: adjustTimestamp(session.last_timestamp, 500), // 1 second after last event
       country: firstCountry, // Same as session_start
       city: firstCity, // Same as session_start
       region: firstRegion, // Same as session_start
@@ -499,7 +495,15 @@ export async function backfillSessionsToProduction(
       os_version,
       sign,
       version,
-      properties
+      properties,
+      utm_medium,
+      utm_source,
+      utm_campaign,
+      utm_content,
+      utm_term,
+      referrer,
+      referrer_name,
+      referrer_type
     )
     SELECT 
       any(e.session_id) as id,
@@ -535,11 +539,19 @@ export async function backfillSessionsToProduction(
       argMinIf(e.os_version, e.created_at, e.name = 'session_start') as os_version,
       1 as sign,
       1 as version,
-      argMinIf(e.properties, e.created_at, e.name = 'session_start') as properties
+      argMinIf(e.properties, e.created_at, e.name = 'session_start') as properties,
+      argMinIf(e.properties['__query.utm_medium'], e.created_at, e.name = 'session_start') as utm_medium,
+      argMinIf(e.properties['__query.utm_source'], e.created_at, e.name = 'session_start') as utm_source,
+      argMinIf(e.properties['__query.utm_campaign'], e.created_at, e.name = 'session_start') as utm_campaign,
+      argMinIf(e.properties['__query.utm_content'], e.created_at, e.name = 'session_start') as utm_content,
+      argMinIf(e.properties['__query.utm_term'], e.created_at, e.name = 'session_start') as utm_term,
+      argMinIf(e.referrer, e.created_at, e.name = 'session_start') as referrer,
+      argMinIf(e.referrer_name, e.created_at, e.name = 'session_start') as referrer_name,
+      argMinIf(e.referrer_type, e.created_at, e.name = 'session_start') as referrer_type
     FROM ${TABLE_NAMES.events_imports} e
     WHERE 
       e.import_id = ${sqlstring.escape(importId)}
-      AND toDate(e.created_at) >= ${sqlstring.escape(from)}
+      AND toDate(e.created_at) = ${sqlstring.escape(from)}
       AND e.session_id != ''
     GROUP BY e.session_id
   `;
@@ -746,7 +758,7 @@ export async function updateImportStatus(
     case 'backfilling_sessions':
       data.currentStep = 'backfilling_sessions';
       data.currentBatch = options.batch;
-      data.statusMessage = `Backfilling sessions for ${options.batch}`;
+      data.statusMessage = `Aggregating sessions for ${options.batch}`;
       break;
     case 'completed':
       data.status = 'completed';
diff --git a/packages/importer/src/providers/mixpanel.test.ts b/packages/importer/src/providers/mixpanel.test.ts
@@ -1,3 +1,4 @@
+import { omit } from 'ramda';
 import { describe, expect, it } from 'vitest';
 import { MixpanelProvider } from './mixpanel';
 
@@ -120,7 +121,7 @@ describe('mixpanel', () => {
       referrer_name: 'Google',
       referrer_type: 'search',
       imported_at: expect.any(String),
-      sdk_name: 'mixpanel',
+      sdk_name: 'mixpanel (web)',
       sdk_version: '1.0.0',
     });
   });
@@ -175,4 +176,144 @@ describe('mixpanel', () => {
       'nested.level1.level2': 'value',
     });
   });
+
+  it('should handle react-native referrer', async () => {
+    const provider = new MixpanelProvider('pid', {
+      from: '2025-01-01',
+      to: '2025-01-02',
+      serviceAccount: 'sa',
+      serviceSecret: 'ss',
+      projectId: '123',
+      provider: 'mixpanel',
+      type: 'api',
+      mapScreenViewProperty: undefined,
+    });
+
+    const rawEvent = {
+      event: 'ec_search_error',
+      properties: {
+        time: 1759947367,
+        distinct_id: '3385916',
+        $browser: 'Mobile Safari',
+        $browser_version: null,
+        $city: 'Bengaluru',
+        $current_url:
+          'https://web.landeed.com/karnataka/ec-encumbrance-certificate',
+        $device: 'iPhone',
+        $device_id:
+          '199b498af1036c-0e943279a1292e-5c0f4368-51bf4-199b498af1036c',
+        $initial_referrer: 'https://www.google.com/',
+        $initial_referring_domain: 'www.google.com',
+        $insert_id: 'bclkaepeqcfuzt4v',
+        $lib_version: '2.60.0',
+        $mp_api_endpoint: 'api-js.mixpanel.com',
+        $mp_api_timestamp_ms: 1759927570699,
+        $os: 'iOS',
+        $region: 'Karnataka',
+        $screen_height: 852,
+        $screen_width: 393,
+        $search_engine: 'google',
+        $user_id: '3385916',
+        binaryReadableVersion: 'NA',
+        binaryVersion: 'NA',
+        component: '/karnataka/ec-encumbrance-certificate',
+        errMsg: 'Request failed with status code 500',
+        errType: 'SERVER_ERROR',
+        isSilentSearch: false,
+        isTimeout: false,
+        jsVersion: '0.42.0',
+        language: 'english',
+        mp_country_code: 'IN',
+        mp_lib: 'web',
+        mp_processing_time_ms: 1759927592421,
+        mp_sent_by_lib_version: '2.60.0',
+        os: 'web',
+        osVersion:
+          'Mozilla/5.0 (iPhone; CPU iPhone OS 18_7_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) GSA/388.0.811331708 Mobile/15E148 Safari/604.1',
+        phoneBrand: 'NA',
+        phoneManufacturer: 'NA',
+        phoneModel: 'NA',
+        searchUuid: '68e65d08-fd81-4ded-37d3-2b08d2bc70c3',
+        serverVersion: 'web2.0',
+        state: 17,
+        stateStr: '17',
+        statusCode: 500,
+        type: 'result_event',
+        utm_medium: 'cpc',
+        utm_source:
+          'google%26utm_medium=cpc%26utm_campaignid=21380769590%26utm_adgroupid=%26utm_adid=%26utm_term=%26utm_device=m%26utm_network=%26utm_location=9062055%26gclid=%26gad_campaignid=21374496705%26gbraid=0AAAAAoV7mTM9mWFripzQ2Od0xXAfrW6p3%26wbraid=CmAKCQjwi4PHBhCUA',
+      },
+    };
+
+    const res = provider.transformEvent(rawEvent);
+
+    expect(res.id.length).toBeGreaterThan(30);
+    expect(res.imported_at).toMatch(
+      /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$/,
+    );
+    expect(omit(['id', 'imported_at'], res)).toEqual({
+      brand: 'Apple',
+      browser: 'GSA',
+      browser_version: 'null',
+      city: 'Bengaluru',
+      country: 'IN',
+      created_at: '2025-10-08T18:16:07.000Z',
+      device: 'mobile',
+      device_id: '199b498af1036c-0e943279a1292e-5c0f4368-51bf4-199b498af1036c',
+      duration: 0,
+      latitude: null,
+      longitude: null,
+      model: 'iPhone',
+      name: 'ec_search_error',
+      origin: 'https://web.landeed.com',
+      os: 'iOS',
+      os_version: '18.7.0',
+      path: '/karnataka/ec-encumbrance-certificate',
+      profile_id: '3385916',
+      project_id: 'pid',
+      properties: {
+        __lib_version: '2.60.0',
+        '__query.gad_campaignid': '21374496705',
+        '__query.gbraid': '0AAAAAoV7mTM9mWFripzQ2Od0xXAfrW6p3',
+        '__query.utm_campaignid': '21380769590',
+        '__query.utm_device': 'm',
+        '__query.utm_location': '9062055',
+        '__query.utm_medium': 'cpc',
+        '__query.utm_source': 'google',
+        '__query.wbraid': 'CmAKCQjwi4PHBhCUA',
+        __screen: '393x852',
+        __source_insert_id: 'bclkaepeqcfuzt4v',
+        __userAgent:
+          'Mozilla/5.0 (iPhone; CPU iPhone OS 18_7_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) GSA/388.0.811331708 Mobile/15E148 Safari/604.1',
+        binaryReadableVersion: 'NA',
+        binaryVersion: 'NA',
+        component: '/karnataka/ec-encumbrance-certificate',
+        errMsg: 'Request failed with status code 500',
+        errType: 'SERVER_ERROR',
+        isSilentSearch: 'false',
+        isTimeout: 'false',
+        jsVersion: '0.42.0',
+        language: 'english',
+        os: 'web',
+        osVersion:
+          'Mozilla/5.0 (iPhone; CPU iPhone OS 18_7_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) GSA/388.0.811331708 Mobile/15E148 Safari/604.1',
+        phoneBrand: 'NA',
+        phoneManufacturer: 'NA',
+        phoneModel: 'NA',
+        searchUuid: '68e65d08-fd81-4ded-37d3-2b08d2bc70c3',
+        serverVersion: 'web2.0',
+        state: '17',
+        stateStr: '17',
+        statusCode: '500',
+        type: 'result_event',
+      },
+      referrer: 'https://www.google.com',
+      referrer_name: 'Google',
+      referrer_type: 'search',
+      region: 'Karnataka',
+      sdk_name: 'mixpanel (web)',
+      sdk_version: '1.0.0',
+      session_id: '',
+    });
+  });
 });
diff --git a/packages/importer/src/providers/mixpanel.ts b/packages/importer/src/providers/mixpanel.ts
