We Ditched HubSpot for Salesforce: 25% More Qualified Leads Postmortem

import os
import time
import logging
import hashlib
from typing import List, Dict, Optional
import requests
from simple_salesforce import Salesforce, SalesforceError
import pandas as pd
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Configure logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[logging.FileHandler('migration.log'), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# Custom exceptions for migration errors
class HubSpotAPIError(Exception):
    pass

class SalesforceAPIError(Exception):
    pass

class LeadMigrationPipeline:
    def __init__(self, hubspot_api_key: str, sf_username: str, sf_password: str, sf_security_token: str):
        self.hubspot_api_key = hubspot_api_key
        self.sf = Salesforce(
            username=sf_username,
            password=sf_password,
            security_token=sf_security_token,
            version='58.0'  # Salesforce REST API v58.0
        )
        # Configure HubSpot session with retry logic for rate limits
        self.hubspot_session = requests.Session()
        retry_strategy = Retry(
            total=5,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=['GET', 'POST']
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.hubspot_session.mount('https://', adapter)
        self.hubspot_session.mount('http://', adapter)
        self.hubspot_base_url = 'https://api.hubapi.com/crm/v3/objects'

    def _generate_lead_hash(self, email: str, domain: str) -> str:
        """Generate unique hash for lead de-duplication using email and company domain"""
        if not email or not domain:
            return ''
        return hashlib.sha256(f'{email.lower()}{domain.lower()}'.encode()).hexdigest()

    def fetch_hubspot_leads(self, limit: int = 100, after: Optional[str] = None) -> Dict:
        """Fetch leads from HubSpot API v3 with pagination handling"""
        url = f'{self.hubspot_base_url}/contacts'
        params = {
            'limit': limit,
            'properties': 'email,firstname,lastname,company,phone,lead_score,hs_lead_status',
            'hapikey': self.hubspot_api_key
        }
        if after:
            params['after'] = after
        try:
            response = self.hubspot_session.get(url, params=params, timeout=10)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            logger.error(f'HubSpot API fetch failed: {str(e)}')
            raise HubSpotAPIError(f'Failed to fetch HubSpot leads: {str(e)}')

    def transform_lead(self, hubspot_lead: Dict) -> Dict:
        """Transform HubSpot lead schema to Salesforce Lead schema"""
        properties = hubspot_lead.get('properties', {})
        email = properties.get('email')
        company = properties.get('company', '')
        domain = company.split('.')[0].lower() if company else ''
        return {
            'FirstName': properties.get('firstname', ''),
            'LastName': properties.get('lastname', ''),
            'Email': email,
            'Company': company,
            'Phone': properties.get('phone', ''),
            'LeadScore': int(properties.get('lead_score', 0)),
            'Status': properties.get('hs_lead_status', 'Open'),
            'Lead_Hash__c': self._generate_lead_hash(email, domain)  # Custom field for de-duplication
        }

    def push_to_salesforce(self, leads: List[Dict]) -> None:
        """Push batch of leads to Salesforce using Bulk API 2.0 for efficiency"""
        try:
            # Use Bulk API for batches > 200 records to avoid REST rate limits
            if len(leads) > 200:
                job = self.sf.bulk.Lead.insert(leads, batch_size=10000)
                logger.info(f'Salesforce Bulk Job {job["id"]} created for {len(leads)} leads')
            else:
                for lead in leads:
                    self.sf.Lead.upsert('Email', lead)  # Upsert on email to avoid duplicates
            logger.info(f'Successfully pushed {len(leads)} leads to Salesforce')
        except SalesforceError as e:
            logger.error(f'Salesforce push failed: {str(e)}')
            raise SalesforceAPIError(f'Failed to push leads to Salesforce: {str(e)}')

    def run_migration(self, batch_size: int = 100) -> None:
        """Orchestrate full migration with pagination and batch processing"""
        after = None
        total_migrated = 0
        logger.info('Starting HubSpot to Salesforce lead migration')
        while True:
            try:
                data = self.fetch_hubspot_leads(limit=batch_size, after=after)
                leads = data.get('results', [])
                if not leads:
                    break
                transformed_leads = [self.transform_lead(lead) for lead in leads]
                self.push_to_salesforce(transformed_leads)
                total_migrated += len(leads)
                logger.info(f'Migrated {total_migrated} leads so far')
                # Handle pagination
                paging = data.get('paging', {})
                after = paging.get('next', {}).get('after')
                if not after:
                    break
                time.sleep(0.1)  # Respect HubSpot rate limits: 100 requests per 10 seconds
            except (HubSpotAPIError, SalesforceAPIError) as e:
                logger.error(f'Migration batch failed: {str(e)}')
                # Retry once on batch failure before aborting
                time.sleep(5)
                continue
        logger.info(f'Migration complete. Total leads migrated: {total_migrated}')

if __name__ == '__main__':
    # Load credentials from environment variables
    pipeline = LeadMigrationPipeline(
        hubspot_api_key=os.getenv('HUBSPOT_API_KEY'),
        sf_username=os.getenv('SF_USERNAME'),
        sf_password=os.getenv('SF_PASSWORD'),
        sf_security_token=os.getenv('SF_SECURITY_TOKEN')
    )
    pipeline.run_migration(batch_size=100)

require('dotenv').config();
const jsforce = require('jsforce');
const { Client } = require('pg');
const { StreamingClient } = require('@salesforce/streaming');
const { logger } = require('./logger'); // Assume winston logger configured elsewhere

// Configuration constants
const SF_LOGIN_URL = 'https://login.salesforce.com';
const LEAD_TOPIC = '/topic/LeadUpdates'; // Push topic for lead changes
const BATCH_SIZE = 50;
const RECONNECT_DELAY_MS = 5000;

// Initialize PostgreSQL client for internal lead score storage
const pgClient = new Client({
  host: process.env.PG_HOST,
  port: process.env.PG_PORT,
  database: process.env.PG_DATABASE,
  user: process.env.PG_USER,
  password: process.env.PG_PASSWORD,
});

// Initialize Salesforce connection
const sfConn = new jsforce.Connection({ loginUrl: SF_LOGIN_URL });

// Custom error classes for error handling
class SalesforceStreamError extends Error {}
class PostgresSyncError extends Error {}

/**
 * Authenticate with Salesforce using environment credentials
 * @returns {Promise}
 */
async function authenticateSalesforce() {
  try {
    await sfConn.login(
      process.env.SF_USERNAME,
      process.env.SF_PASSWORD + process.env.SF_SECURITY_TOKEN
    );
    logger.info('Successfully authenticated with Salesforce');
  } catch (error) {
    logger.error(`Salesforce authentication failed: ${error.message}`);
    throw new SalesforceStreamError(`Auth failed: ${error.message}`);
  }
}

/**
 * Fetch lead score from internal PostgreSQL database
 * @param {string} leadEmail - Lead email to look up
 * @returns {Promise} Lead score (0-100)
 */
async function fetchInternalLeadScore(leadEmail) {
  try {
    await pgClient.connect();
    const res = await pgClient.query(
      'SELECT lead_score FROM internal_leads WHERE email = $1',
      [leadEmail]
    );
    return res.rows[0]?.lead_score || 0;
  } catch (error) {
    logger.error(`PostgreSQL fetch failed for ${leadEmail}: ${error.message}`);
    throw new PostgresSyncError(`Failed to fetch lead score: ${error.message}`);
  } finally {
    await pgClient.end();
  }
}

/**
 * Update Salesforce lead with internal lead score
 * @param {string} leadId - Salesforce Lead ID
 * @param {number} leadScore - New lead score
 * @returns {Promise}
 */
async function updateSalesforceLeadScore(leadId, leadScore) {
  try {
    await sfConn.sobject('Lead').update({
      Id: leadId,
      LeadScore__c: leadScore,
    });
    logger.info(`Updated Lead ${leadId} with score ${leadScore}`);
  } catch (error) {
    logger.error(`Salesforce update failed for Lead ${leadId}: ${error.message}`);
    throw new SalesforceStreamError(`Update failed: ${error.message}`);
  }
}

/**
 * Process batch of lead updates from Salesforce Streaming API
 * @param {Array} leads - Batch of lead records from streaming topic
 * @returns {Promise}
 */
async function processLeadBatch(leads) {
  logger.info(`Processing batch of ${leads.length} lead updates`);
  for (const lead of leads) {
    try {
      const internalScore = await fetchInternalLeadScore(lead.Email);
      if (internalScore !== lead.LeadScore__c) {
        await updateSalesforceLeadScore(lead.Id, internalScore);
      }
    } catch (error) {
      logger.error(`Failed to process Lead ${lead.Id}: ${error.message}`);
      // Log and continue to avoid blocking entire batch
    }
  }
}

/**
 * Subscribe to Salesforce Streaming API for real-time lead updates
 * @returns {Promise}
 */
async function subscribeToLeadUpdates() {
  const streamingClient = new StreamingClient(sfConn, [LEAD_TOPIC]);
  let batch = [];

  streamingClient.on('message', async (message) => {
    const lead = message.sobject;
    batch.push(lead);
    if (batch.length >= BATCH_SIZE) {
      await processLeadBatch(batch);
      batch = [];
    }
  });

  streamingClient.on('error', async (error) => {
    logger.error(`Streaming client error: ${error.message}`);
    logger.info(`Reconnecting in ${RECONNECT_DELAY_MS}ms...`);
    await new Promise((resolve) => setTimeout(resolve, RECONNECT_DELAY_MS));
    await subscribeToLeadUpdates(); // Reconnect recursively
  });

  try {
    await streamingClient.subscribe();
    logger.info(`Subscribed to Salesforce topic: ${LEAD_TOPIC}`);
  } catch (error) {
    logger.error(`Subscription failed: ${error.message}`);
    throw new SalesforceStreamError(`Subscribe failed: ${error.message}`);
  }
}

// Main execution
(async () => {
  try {
    await authenticateSalesforce();
    await subscribeToLeadUpdates();
  } catch (error) {
    logger.error(`Fatal error: ${error.message}`);
    process.exit(1);
  }
})();

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "os"
    "time"

    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promhttp"
    "github.com/simpleforce/simpleforce"
)

// Configuration constants
const (
    hubspotBaseURL = "https://api.hubapi.com/crm/v3"
    sfAPIVersion   = "58.0"
    checkInterval  = 30 * time.Second
    metricsPort    = ":9090"
)

// Prometheus metrics definitions
var (
    hubspotAPILatency = prometheus.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "hubspot_api_latency_ms",
            Help:    "HubSpot API request latency in milliseconds",
            Buckets: prometheus.DefBuckets,
        },
        []string{"endpoint", "status"},
    )
    salesforceAPILatency = prometheus.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "salesforce_api_latency_ms",
            Help:    "Salesforce API request latency in milliseconds",
            Buckets: prometheus.DefBuckets,
        },
        []string{"endpoint", "status"},
    )
    leadCountMismatch = prometheus.NewGauge(
        prometheus.GaugeOpts{
            Name: "lead_count_mismatch",
            Help: "Difference between HubSpot and Salesforce lead counts (HubSpot - Salesforce)",
        },
    )
    migrationSuccess = prometheus.NewCounter(
        prometheus.CounterOpts{
            Name: "migration_success_total",
            Help: "Total number of successful migration batches",
        },
    )
    migrationFailure = prometheus.NewCounter(
        prometheus.CounterOpts{
            Name: "migration_failure_total",
            Help: "Total number of failed migration batches",
        },
    )
)

func init() {
    // Register all Prometheus metrics
    prometheus.MustRegister(hubspotAPILatency)
    prometheus.MustRegister(salesforceAPILatency)
    prometheus.MustRegister(leadCountMismatch)
    prometheus.MustRegister(migrationSuccess)
    prometheus.MustRegister(migrationFailure)
}

// HubSpotClient wraps HubSpot API interactions
type HubSpotClient struct {
    apiKey string
    client *http.Client
}

// NewHubSpotClient creates a new HubSpot API client
func NewHubSpotClient(apiKey string) *HubSpotClient {
    return &HubSpotClient{
        apiKey: apiKey,
        client: &http.Client{Timeout: 10 * time.Second},
    }
}

// GetLeadCount fetches total lead count from HubSpot
func (h *HubSpotClient) GetLeadCount(ctx context.Context) (int, error) {
    start := time.Now()
    url := fmt.Sprintf("%s/objects/contacts?limit=1&hapikey=%s", hubspotBaseURL, h.apiKey)
    req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
    if err != nil {
        return 0, fmt.Errorf("failed to create request: %w", err)
    }
    resp, err := h.client.Do(req)
    if err != nil {
        hubspotAPILatency.WithLabelValues("/objects/contacts", "error").Observe(float64(time.Since(start).Milliseconds()))
        return 0, fmt.Errorf("request failed: %w", err)
    }
    defer resp.Body.Close()
    hubspotAPILatency.WithLabelValues("/objects/contacts", fmt.Sprintf("%d", resp.StatusCode)).Observe(float64(time.Since(start).Milliseconds()))
    if resp.StatusCode != http.StatusOK {
        return 0, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
    }
    var result struct {
        Total int `json:"total"`
    }
    if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
        return 0, fmt.Errorf("failed to decode response: %w", err)
    }
    return result.Total, nil
}

// SalesforceClient wraps Salesforce API interactions
type SalesforceClient struct {
    client *simpleforce.Client
}

// NewSalesforceClient creates a new Salesforce API client
func NewSalesforceClient(username, password, token string) (*SalesforceClient, error) {
    client := simpleforce.NewClient("https://login.salesforce.com", sfAPIVersion)
    if err := client.Login(username, password+token); err != nil {
        return nil, fmt.Errorf("salesforce login failed: %w", err)
    }
    return &SalesforceClient{client: client}, nil
}

// GetLeadCount fetches total lead count from Salesforce
func (s *SalesforceClient) GetLeadCount(ctx context.Context) (int, error) {
    start := time.Now()
    query := "SELECT COUNT() FROM Lead"
    resp, err := s.client.Query(query)
    if err != nil {
        salesforceAPILatency.WithLabelValues("query", "error").Observe(float64(time.Since(start).Milliseconds()))
        return 0, fmt.Errorf("query failed: %w", err)
    }
    salesforceAPILatency.WithLabelValues("query", "200").Observe(float64(time.Since(start).Milliseconds()))
    return resp.TotalSize, nil
}

func main() {
    // Load environment variables
    hubspotAPIKey := os.Getenv("HUBSPOT_API_KEY")
    sfUsername := os.Getenv("SF_USERNAME")
    sfPassword := os.Getenv("SF_PASSWORD")
    sfToken := os.Getenv("SF_SECURITY_TOKEN")
    if hubspotAPIKey == "" || sfUsername == "" || sfPassword == "" || sfToken == "" {
        log.Fatal("Missing required environment variables")
    }

    // Initialize clients
    hubspotClient := NewHubSpotClient(hubspotAPIKey)
    sfClient, err := NewSalesforceClient(sfUsername, sfPassword, sfToken)
    if err != nil {
        log.Fatalf("Failed to initialize Salesforce client: %v", err)
    }

    // Start Prometheus metrics server
    http.Handle("/metrics", promhttp.Handler())
    go func() {
        log.Printf("Starting metrics server on %s", metricsPort)
        if err := http.ListenAndServe(metricsPort, nil); err != nil {
            log.Fatalf("Metrics server failed: %v", err)
        }
    }()

    // Run periodic health checks
    ticker := time.NewTicker(checkInterval)
    defer ticker.Stop()
    for {
        select {
        case <-ticker.C:
            ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
            defer cancel()

            // Check HubSpot lead count
            hubspotCount, err := hubspotClient.GetLeadCount(ctx)
            if err != nil {
                log.Printf("HubSpot count check failed: %v", err)
                migrationFailure.Inc()
                continue
            }

            // Check Salesforce lead count
            sfCount, err := sfClient.GetLeadCount(ctx)
            if err != nil {
                log.Printf("Salesforce count check failed: %v", err)
                migrationFailure.Inc()
                continue
            }

            // Update mismatch metric
            mismatch := hubspotCount - sfCount
            leadCountMismatch.Set(float64(mismatch))
            log.Printf("Lead count check: HubSpot=%d, Salesforce=%d, Mismatch=%d", hubspotCount, sfCount, mismatch)

            migrationSuccess.Inc()
        }
    }
}

from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter

retry_strategy = Retry(
    total=5,
    backoff_factor=1,  # 1s, 2s, 4s, 8s, 16s
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=['GET', 'POST'],
    backoff_jitter=0.2  # ±20% jitter
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount('https://', adapter)

const job = await conn.bulk.createJob('Lead', 'insert', {
  externalIdFieldName: 'Email',  // Upsert on Email
  contentType: 'JSON',
  concurrencyMode: 'Parallel'
});
const batch = job.createBatch();
batch.execute(leads);
batch.on('response', (res) => {
  if (res.numberRecordsFailed > 0) {
    logger.error(`Batch failed ${res.numberRecordsFailed} records`);
    // Retry failed records
  }
});

import (
  "go.opentelemetry.io/otel"
  "go.opentelemetry.io/otel/exporters/jaeger"
  "go.opentelemetry.io/otel/sdk/resource"
  sdktrace "go.opentelemetry.io/otel/sdk/trace"
)

func initTracer() {
  exporter, err := jaeger.New(jaeger.WithCollectorEndpoint(jaeger.WithEndpoint("http://jaeger:14268/api/traces")))
  if err != nil {
    log.Fatal(err)
  }
  tp := sdktrace.NewTracerProvider(
    sdktrace.WithBatcher(exporter),
    sdktrace.WithResource(resource.NewWithAttributes(
      "service.name", "crm-sync-health",
    )),
  )
  otel.SetTracerProvider(tp)
}