fix(ingestion): salesforce connector + dbt

src/ingestion/connectors/crm/salesforce/README.md

@@ -1,8 +1,6 @@
 # Salesforce Connector
 
-CDK-based Python connector for Salesforce CRM. Pulls data via Bulk API 2.0 with REST `/queryAll` fallback; describe-driven field discovery means no SOQL maintenance as SF orgs evolve; custom (`__c`) fields are captured into a single `custom_fields` JSON column so Bronze stays stable across orgs.
-
-Architecture adapted from the [official Airbyte Salesforce connector](https://github.com/airbytehq/airbyte/tree/master/airbyte-integrations/connectors/source-salesforce) (ELv2). Per-file copyright headers preserved.
+CDK-based Python connector for Salesforce CRM. Pulls data via the REST `/queryAll` API; describe-driven field discovery means no SOQL maintenance as SF orgs evolve; custom (`__c`) fields are captured into a single `custom_fields` JSON column so Bronze stays stable across orgs.
 
 ## Prerequisites
 
@@ -29,7 +27,7 @@ stringData:
   salesforce_client_id: ""                         # OAuth Consumer Key
   salesforce_client_secret: ""                     # OAuth Consumer Secret
   salesforce_start_date: "2024-01-01T00:00:00Z"    # Optional
-  salesforce_num_workers: "20"                     # Optional (1–50)
+  salesforce_num_workers: "1"                      # Optional (1–50), default 1
 ```
 
 ### Fields
@@ -40,11 +38,9 @@ stringData:
 | `salesforce_client_id` | Yes | OAuth Consumer Key |
 | `salesforce_client_secret` | Yes | OAuth Consumer Secret (sensitive) |
 | `salesforce_start_date` | No | Incremental sync start (ISO 8601). Defaults to two years before current date (computed in `source.py`). |
-| `salesforce_streams` | No | JSON array of sobject names to sync. Overrides curated default |
 | `salesforce_stream_slice_step` | No | Concurrent cursor window. Default `P30D` |
 | `salesforce_lookback_window` | No | Re-read window for SystemModstamp consistency. Default `PT10M` |
-| `salesforce_force_use_bulk_api` | No | Force Bulk even for unsupported types. Default `false` |
-| `salesforce_num_workers` | No | Max concurrent jobs (1–50). Default `20` |
+| `salesforce_num_workers` | No | Max concurrent jobs (1–50). Default `1` (sequential; raise once ClickHouse has headroom) |
 
 ### Automatically injected
 
@@ -67,10 +63,10 @@ cp src/ingestion/secrets/connectors/salesforce.yaml.example src/ingestion/secret
 
 ## Streams
 
-Active: 10 sobjects (curated for analytics value vs sync cost). Operator can
-override via `salesforce_streams` config; additional disabled sobjects listed
-at the bottom of `constants.CRM_STREAMS` can be re-enabled there or supplied
-ad-hoc via config.
+Active: 10 sobjects (curated for analytics value vs sync cost). The stream
+set is fixed in `constants.CRM_STREAMS`; additional disabled sobjects listed
+at the bottom of that list can be re-enabled there (a code change, so dbt
+coverage ships alongside).
 
 ### Active
 
@@ -88,7 +84,7 @@ ad-hoc via config.
 | `OpportunityContactRole` | Buyer-committee / multi-threading |
 
 All incremental via `SystemModstamp` (except `OpportunityHistory` → `CreatedDate`).
-Bulk API by default; REST fallback for sobjects with compound / base64 fields.
+All streams use the REST `/queryAll` API (paginated via `nextRecordsUrl`).
 Soft-deleted records (`IsDeleted=true`) included — `queryAll` used throughout.
 
 ### Available, disabled by default
@@ -124,10 +120,10 @@ Silver classes for Lead / Case / OpportunityContactRole / OpportunityLineItem /
 
 ```bash
 cd src/ingestion
-./airbyte-toolkit/build-connector.sh crm/salesforce   # docker build + Kind load + Airbyte definition update
-./airbyte-toolkit/connect.sh <tenant>                 # create source + connection from K8s Secret
-./run-sync.sh salesforce <tenant>                     # e2e via Argo (sync + dbt)
-./logs.sh -f latest                                   # follow
+bash reconcile-connectors/lib/cdk-build.sh crm/salesforce        # docker build + Kind load + Airbyte definition update
+bash reconcile-connectors/main.sh --connector salesforce         # create/update source + connection from K8s Secret
+./run-sync.sh salesforce <tenant>                                # e2e via Argo (sync + dbt)
+./logs.sh -f latest                                              # follow
 ```
 
 ## Troubleshooting

src/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_accounts.sql

@@ -9,7 +9,7 @@
     tags=['salesforce', 'silver:class_crm_accounts']
 ) }}
 
-SELECT * FROM (
+WITH src AS (
     SELECT
         tenant_id,
         source_id,
@@ -30,14 +30,22 @@ SELECT * FROM (
             'IsDeleted',         toString(coalesce(IsDeleted, false))
         ))                                              AS metadata,
         custom_fields,
-        parseDateTimeBestEffort(CreatedDate)            AS created_at,
-        parseDateTimeBestEffort(LastModifiedDate)       AS updated_at,
+        CreatedDate                                     AS created_at,
+        LastModifiedDate                                AS updated_at,
         data_source,
-        toUnixTimestamp64Milli(
-            parseDateTime64BestEffort(SystemModstamp)
-        )                                               AS _version
+        coalesce(toUnixTimestamp64Milli(SystemModstamp), 0) AS _version
     FROM {{ source('bronze_salesforce', 'Account') }}
 )
 {% if is_incremental() %}
-WHERE _version > coalesce((SELECT max(_version) FROM {{ this }}), 0)
+SELECT src.*
+FROM src
+LEFT JOIN (
+    SELECT tenant_id, source_id, max(_version) AS hwm
+    FROM {{ this }}
+    GROUP BY tenant_id, source_id
+) w
+  ON w.tenant_id = src.tenant_id AND w.source_id = src.source_id
+WHERE src._version > coalesce(w.hwm, 0)
+{% else %}
+SELECT * FROM src
 {% endif %}

src/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_activities.sql

@@ -25,13 +25,17 @@ WITH tasks AS (
         -- HubSpot's `properties_hs_created_by_user_id`; gold attributes
         -- non-call activities by this column.
         CAST(CreatedById AS Nullable(String))       AS created_by_user_id,
-        WhoId                                       AS contact_id,
+        -- WhoId is polymorphic: 003-prefixed = Contact, 00Q-prefixed = Lead.
+        -- Leads have no silver class; keep contact_id referentially valid.
+        CASE WHEN startsWith(coalesce(WhoId, ''), '003') THEN WhoId
+             ELSE NULL END                          AS contact_id,
         CASE WHEN startsWith(coalesce(WhatId, ''), '006') THEN WhatId
              ELSE NULL END                          AS deal_id,
         CASE WHEN startsWith(coalesce(WhatId, ''), '001') THEN WhatId
              ELSE NULL END                          AS account_id,
-        parseDateTimeBestEffort(
-            coalesce(toString(ActivityDate), toString(CreatedDate))
+        coalesce(
+            CAST(ActivityDate AS Nullable(DateTime64(3))),
+            CreatedDate
         )                                           AS timestamp,
         CASE WHEN CallType IS NOT NULL AND CallDurationInSeconds IS NOT NULL
              THEN toInt64(CallDurationInSeconds)
@@ -45,11 +49,9 @@ WITH tasks AS (
             'IsDeleted',   toString(coalesce(IsDeleted, false))
         ))                                          AS metadata,
         custom_fields,
-        parseDateTimeBestEffort(CreatedDate)        AS created_at,
+        CreatedDate                                 AS created_at,
         data_source,
-        toUnixTimestamp64Milli(
-            parseDateTime64BestEffort(SystemModstamp)
-        )                                           AS _version
+        coalesce(toUnixTimestamp64Milli(SystemModstamp), 0) AS _version
     FROM {{ source('bronze_salesforce', 'Task') }}
 ),
 events AS (
@@ -67,15 +69,20 @@ events AS (
         -- HubSpot's `properties_hs_created_by_user_id`; gold attributes
         -- non-call activities by this column.
         CAST(CreatedById AS Nullable(String))       AS created_by_user_id,
-        WhoId                                       AS contact_id,
+        -- WhoId is polymorphic: 003-prefixed = Contact, 00Q-prefixed = Lead.
+        -- Leads have no silver class; keep contact_id referentially valid.
+        CASE WHEN startsWith(coalesce(WhoId, ''), '003') THEN WhoId
+             ELSE NULL END                          AS contact_id,
         CASE WHEN startsWith(coalesce(WhatId, ''), '006') THEN WhatId
              ELSE NULL END                          AS deal_id,
         CASE WHEN startsWith(coalesce(WhatId, ''), '001') THEN WhatId
              ELSE NULL END                          AS account_id,
-        parseDateTimeBestEffort(
-            coalesce(toString(StartDateTime), toString(ActivityDate), toString(CreatedDate))
+        coalesce(
+            StartDateTime,
+            CAST(ActivityDate AS Nullable(DateTime64(3))),
+            CreatedDate
         )                                           AS timestamp,
-        toInt64OrNull(DurationInMinutes) * 60       AS duration_seconds,
+        DurationInMinutes * 60                      AS duration_seconds,
         CAST(NULL AS Nullable(String))              AS outcome,
         toJSONString(map(
             'Subject',      coalesce(toString(Subject), ''),
@@ -85,19 +92,26 @@ events AS (
             'IsDeleted',    toString(coalesce(IsDeleted, false))
         ))                                          AS metadata,
         custom_fields,
-        parseDateTimeBestEffort(CreatedDate)        AS created_at,
+        CreatedDate                                 AS created_at,
         data_source,
-        toUnixTimestamp64Milli(
-            parseDateTime64BestEffort(SystemModstamp)
-        )                                           AS _version
+        coalesce(toUnixTimestamp64Milli(SystemModstamp), 0) AS _version
     FROM {{ source('bronze_salesforce', 'Event') }}
 ),
 combined AS (
     SELECT * FROM tasks
     UNION ALL
     SELECT * FROM events
 )
-SELECT * FROM combined
 {% if is_incremental() %}
-WHERE _version > coalesce((SELECT max(_version) FROM {{ this }}), 0)
+SELECT combined.*
+FROM combined
+LEFT JOIN (
+    SELECT tenant_id, source_id, max(_version) AS hwm
+    FROM {{ this }}
+    GROUP BY tenant_id, source_id
+) w
+  ON w.tenant_id = combined.tenant_id AND w.source_id = combined.source_id
+WHERE combined._version > coalesce(w.hwm, 0)
+{% else %}
+SELECT * FROM combined
 {% endif %}

src/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_contacts.sql

@@ -9,7 +9,7 @@
     tags=['salesforce', 'silver:class_crm_contacts']
 ) }}
 
-SELECT * FROM (
+WITH src AS (
     SELECT
         tenant_id,
         source_id,
@@ -28,14 +28,22 @@ SELECT * FROM (
             'IsDeleted',  toString(coalesce(IsDeleted, false))
         ))                                              AS metadata,
         custom_fields,
-        parseDateTimeBestEffort(CreatedDate)            AS created_at,
-        parseDateTimeBestEffort(LastModifiedDate)       AS updated_at,
+        CreatedDate                                     AS created_at,
+        LastModifiedDate                                AS updated_at,
         data_source,
-        toUnixTimestamp64Milli(
-            parseDateTime64BestEffort(SystemModstamp)
-        )                                               AS _version
+        coalesce(toUnixTimestamp64Milli(SystemModstamp), 0) AS _version
     FROM {{ source('bronze_salesforce', 'Contact') }}
 )
 {% if is_incremental() %}
-WHERE _version > coalesce((SELECT max(_version) FROM {{ this }}), 0)
+SELECT src.*
+FROM src
+LEFT JOIN (
+    SELECT tenant_id, source_id, max(_version) AS hwm
+    FROM {{ this }}
+    GROUP BY tenant_id, source_id
+) w
+  ON w.tenant_id = src.tenant_id AND w.source_id = src.source_id
+WHERE src._version > coalesce(w.hwm, 0)
+{% else %}
+SELECT * FROM src
 {% endif %}

src/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_deals.sql

@@ -9,7 +9,7 @@
     tags=['salesforce', 'silver:class_crm_deals']
 ) }}
 
-SELECT * FROM (
+WITH src AS (
     SELECT
         tenant_id,
         source_id,
@@ -32,7 +32,7 @@ SELECT * FROM (
         CAST(NULL AS Nullable(Float64))                 AS acv,
         CAST(NULL AS Nullable(Float64))                 AS tcv,
         CAST(NULL AS Nullable(Float64))                 AS arr,
-        toDateOrNull(CloseDate)                         AS close_date,
+        CloseDate                                       AS close_date,
         OwnerId                                         AS owner_id,
         -- Rep who created the Opportunity record (universal SF audit field).
         -- Parallels HubSpot's `properties_hs_created_by_user_id`.
@@ -56,14 +56,22 @@ SELECT * FROM (
             'IsDeleted', if(coalesce(IsDeleted, false), 'true', 'false')
         ))                                              AS metadata,
         custom_fields,
-        parseDateTime64BestEffortOrNull(CreatedDate, 3)      AS created_at,
-        parseDateTime64BestEffortOrNull(LastModifiedDate, 3) AS updated_at,
+        CreatedDate                                     AS created_at,
+        LastModifiedDate                                AS updated_at,
         data_source,
-        toUnixTimestamp64Milli(
-            parseDateTime64BestEffort(SystemModstamp)
-        )                                               AS _version
+        coalesce(toUnixTimestamp64Milli(SystemModstamp), 0) AS _version
     FROM {{ source('bronze_salesforce', 'Opportunity') }}
 )
 {% if is_incremental() %}
-WHERE _version > coalesce((SELECT max(_version) FROM {{ this }}), 0)
+SELECT src.*
+FROM src
+LEFT JOIN (
+    SELECT tenant_id, source_id, max(_version) AS hwm
+    FROM {{ this }}
+    GROUP BY tenant_id, source_id
+) w
+  ON w.tenant_id = src.tenant_id AND w.source_id = src.source_id
+WHERE src._version > coalesce(w.hwm, 0)
+{% else %}
+SELECT * FROM src
 {% endif %}

src/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_users.sql

@@ -9,7 +9,7 @@
     tags=['salesforce', 'silver:class_crm_users']
 ) }}
 
-SELECT * FROM (
+WITH src AS (
     SELECT
         tenant_id,
         source_id,
@@ -27,18 +27,24 @@ SELECT * FROM (
         toInt64(IsActive = true)                        AS is_active,
         toJSONString(map(
             'Username',   coalesce(toString(Username), ''),
-            'ProfileId',  coalesce(toString(ProfileId), ''),
-            'UserRoleId', coalesce(toString(UserRoleId), ''),
-            'IsDeleted',  toString(coalesce(IsDeleted, false))
+            'UserRoleId', coalesce(toString(UserRoleId), '')
         ))                                              AS metadata,
         custom_fields,
         collected_at,
         data_source,
-        toUnixTimestamp64Milli(
-            parseDateTime64BestEffort(SystemModstamp)
-        )                                               AS _version
+        coalesce(toUnixTimestamp64Milli(SystemModstamp), 0) AS _version
     FROM {{ source('bronze_salesforce', 'User') }}
 )
 {% if is_incremental() %}
-WHERE _version > coalesce((SELECT max(_version) FROM {{ this }}), 0)
+SELECT src.*
+FROM src
+LEFT JOIN (
+    SELECT tenant_id, source_id, max(_version) AS hwm
+    FROM {{ this }}
+    GROUP BY tenant_id, source_id
+) w
+  ON w.tenant_id = src.tenant_id AND w.source_id = src.source_id
+WHERE src._version > coalesce(w.hwm, 0)
+{% else %}
+SELECT * FROM src
 {% endif %}

src/ingestion/connectors/crm/salesforce/pyproject.toml

@@ -5,11 +5,11 @@ build-backend = "setuptools.build_meta"
 [project]
 name = "source-salesforce-insight"
 version = "2.0.0"
-description = "Airbyte source connector for Salesforce (CDK-native, Bulk API 2.0 with REST fallback)"
+description = "Airbyte source connector for Salesforce (CDK-native, REST queryAll)"
 requires-python = ">=3.10"
 dependencies = [
-    "airbyte-cdk>=7.16.0",
-    "requests>=2.31.0",
+    "airbyte-cdk>=7.23.1,<8.0.0",
+    "requests>=2.34.2",
     "pendulum>=3.0.0",
 ]