Build a LinkedIn Outreach Sequence

// Search for leads using LinkedIn search
// 1. Build your search URL on LinkedIn first (e.g., filter by title, location, company)
// 2. Copy the search URL from your browser
const searchUrl = 'https://www.linkedin.com/search/results/people/?keywords=CTO&geoUrn=%5B103644278%5D&origin=FACETED_SEARCH';

const searchResults = await fetch('https://api.edges.run/v1/actions/linkedin-search-people/run/live', {
  method: 'POST',
  headers: {
    'X-API-Key': EDGES_API_KEY,
    'Content-Type': 'application/json'
  },
  body: JSON.stringify({
    identity_mode: 'auto',
    input: {
      linkedin_people_search_url: searchUrl
    }
  })
});

const leads = await searchResults.json();
// Each result contains: linkedin_profile_url, full_name, headline, etc.

import requests

# Search for leads using LinkedIn search
# 1. Build your search URL on LinkedIn first (e.g., filter by title, location, company)
# 2. Copy the search URL from your browser
search_url = 'https://www.linkedin.com/search/results/people/?keywords=CTO&geoUrn=%5B103644278%5D&origin=FACETED_SEARCH'

response = requests.post(
    'https://api.edges.run/v1/actions/linkedin-search-people/run/live',
    headers={
        'X-API-Key': EDGES_API_KEY,
        'Content-Type': 'application/json'
    },
    json={
        'identity_mode': 'auto',
        'input': {
            'linkedin_people_search_url': search_url
        }
    }
)

leads = response.json()
# Each result contains: linkedin_profile_url, full_name, headline, etc.

// Get linkedin_profile_id from integration meta (recommended)
async function getIntegrationProfileId(identityId, integration = 'linkedin') {
  const response = await fetch(
    `https://api.edges.run/v1/identities/${identityId}/integrations/${integration}`,
    {
      headers: { 'X-API-Key': EDGES_API_KEY }
    }
  );
  
  const data = await response.json();
  return data.meta?.linkedin_profile_id;  // Already available!
}

// Cache this when identity is created/connected
const userProfileId = await getIntegrationProfileId('identity_abc123');

import requests

def get_integration_profile_id(identity_id: str, integration: str = 'linkedin') -> int:
    """Get linkedin_profile_id from integration meta (recommended)."""
    response = requests.get(
        f'https://api.edges.run/v1/identities/{identity_id}/integrations/{integration}',
        headers={'X-API-Key': EDGES_API_KEY}
    )
    
    data = response.json()
    return data.get('meta', {}).get('linkedin_profile_id')  # Already available!

# Cache this when identity is created/connected
user_profile_id = get_integration_profile_id('identity_abc123')

async function executeWithErrorHandling(identityId, lead, actionFn) {
  try {
    const result = await actionFn();
    return { success: true, result };
  } catch (error) {
    const errorLabel = error.error_label || 'UNKNOWN';
    
    switch (errorLabel) {
      case 'LIMIT_REACHED':
        // Retry tomorrow - use postponed_until if available
        const retryAt = error.postponed_until || tomorrow();
        await updateLead(lead.id, { next_action_at: retryAt });
        break;
        
      case 'ALREADY_CONNECTED':
        // Skip connect step, go to message
        await updateLead(lead.id, { 
          connection_status: 'CONNECTED', 
          current_step: 3  // Skip to message step
        });
        break;
        
      case 'INVITATION_PENDING':
        // Already sent, just wait
        await updateLead(lead.id, { connection_status: 'PENDING' });
        break;
        
      case 'NOT_CONNECTED':
        // Can't message - need to connect first
        await updateLead(lead.id, { current_step: 1 });  // Go to connect step
        break;
        
      case 'PROFILE_NOT_ACCESSIBLE':
        // Remove from sequence
        await updateLead(lead.id, { 
          sequence_status: 'ARCHIVED', 
          paused_reason: 'PROFILE_INACCESSIBLE' 
        });
        break;
        
      case 'AUTH_EXPIRED':
        // Pause all leads for this identity
        await pauseIdentityLeads(identityId, 'AUTH_EXPIRED');
        break;
        
      default:
        // Log and investigate
        console.error('Unexpected error:', error);
        await logOutreachError(lead.id, error);
    }
    
    return { success: false, errorLabel };
  }
}

function tomorrow() {
  const date = new Date();
  date.setDate(date.getDate() + 1);
  date.setHours(9, 0, 0, 0);  // 9am tomorrow
  return date.toISOString();
}

from datetime import datetime, timedelta

async def execute_with_error_handling(identity_id: str, lead, action_fn):
    try:
        result = await action_fn()
        return {'success': True, 'result': result}
    except EdgesError as error:
        error_label = getattr(error, 'error_label', 'UNKNOWN')
        
        if error_label == 'LIMIT_REACHED':
            # Retry tomorrow - use postponed_until if available
            retry_at = getattr(error, 'postponed_until', None) or tomorrow()
            await update_lead(lead.id, next_action_at=retry_at)
            
        elif error_label == 'ALREADY_CONNECTED':
            # Skip connect step, go to message
            await update_lead(lead.id, connection_status='CONNECTED', current_step=3)
            
        elif error_label == 'INVITATION_PENDING':
            # Already sent, just wait
            await update_lead(lead.id, connection_status='PENDING')
            
        elif error_label == 'NOT_CONNECTED':
            # Can't message - need to connect first
            await update_lead(lead.id, current_step=1)
            
        elif error_label == 'PROFILE_NOT_ACCESSIBLE':
            # Remove from sequence
            await update_lead(lead.id, 
                sequence_status='ARCHIVED', 
                paused_reason='PROFILE_INACCESSIBLE'
            )
            
        elif error_label == 'AUTH_EXPIRED':
            # Pause all leads for this identity
            await pause_identity_leads(identity_id, 'AUTH_EXPIRED')
            
        else:
            # Log and investigate
            print(f'Unexpected error: {error}')
            await log_outreach_error(lead.id, error)
        
        return {'success': False, 'error_label': error_label}

def tomorrow():
    date = datetime.now() + timedelta(days=1)
    return date.replace(hour=9, minute=0, second=0, microsecond=0).isoformat()

// Sync conversations and detect replies efficiently
async function syncConversations(identityId) {
  const userProfileId = await getUserProfileId(identityId);
  
  const response = await fetch('https://api.edges.run/v1/actions/linkedin-extract-conversations/run/live', {
    method: 'POST',
    headers: {
      'X-API-Key': EDGES_API_KEY,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({
      identity_ids: [identityId]
    })
  });
  
  const data = await response.json();
  
  for (const conv of data) {
    const lastSenderId = conv.last_message?.linkedin_profile_id;
    
    // If last message sender != user, the lead replied!
    if (lastSenderId && lastSenderId !== userProfileId) {
      const leadProfileId = conv.participants[0]?.linkedin_profile_id;
      await markLeadAsReplied(identityId, leadProfileId);
      // NO NEED to call extract-messages for reply detection!
    }
    
    // Cache conversation for de-dup
    await upsertConversation({
      identity_id: identityId,
      linkedin_thread_id: conv.linkedin_thread_id,
      participant_profile_id: conv.participants[0]?.linkedin_profile_id,
      last_message_sender_id: lastSenderId,
      last_message_at: conv.last_message?.delivered_at
    });
  }
}

async def sync_conversations(identity_id: str):
    """Sync conversations and detect replies efficiently."""
    user_profile_id = await get_user_profile_id(identity_id)
    
    response = requests.post(
        'https://api.edges.run/v1/actions/linkedin-extract-conversations/run/live',
        headers={
            'X-API-Key': EDGES_API_KEY,
            'Content-Type': 'application/json'
        },
        json={
            'identity_ids': [identity_id]
        }
    )
    
    data = response.json()
    
    for conv in data:
        last_sender_id = conv.get('last_message', {}).get('linkedin_profile_id')
        
        # If last message sender != user, the lead replied!
        if last_sender_id and last_sender_id != user_profile_id:
            lead_profile_id = conv['participants'][0].get('linkedin_profile_id')
            await mark_lead_as_replied(identity_id, lead_profile_id)
            # NO NEED to call extract-messages for reply detection!
        
        # Cache conversation for de-dup
        await upsert_conversation(
            identity_id=identity_id,
            linkedin_thread_id=conv['linkedin_thread_id'],
            participant_profile_id=conv['participants'][0].get('linkedin_profile_id'),
            last_message_sender_id=last_sender_id,
            last_message_at=conv.get('last_message', {}).get('delivered_at')
        )

async function sendConnectionRequest(identityId, lead, message = null) {
  const response = await fetch('https://api.edges.run/v1/actions/linkedin-connect-profile/run/live', {
    method: 'POST',
    headers: {
      'X-API-Key': EDGES_API_KEY,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({
      identity_ids: [identityId],
      parameters: {
        message: message  // Optional, 300 char limit. Only 5/month with note!
      },
      input: {
        linkedin_profile_url: lead.linkedin_profile_url
      }
    })
  });
  
  if (!response.ok) {
    const error = await response.json();
    throw error;
  }
  
  const data = await response.json();
  
  // Update lead status
  await updateLead(lead.id, {
    connection_status: 'PENDING',
    connection_sent_at: new Date().toISOString(),
    last_contacted_at: new Date().toISOString()
  });
  
  // Log the action
  await logOutreach(lead.id, 'CONNECT', data.run_id);
  
  return data;
}

async def send_connection_request(identity_id: str, lead, message: str = None):
    response = requests.post(
        'https://api.edges.run/v1/actions/linkedin-connect-profile/run/live',
        headers={
            'X-API-Key': EDGES_API_KEY,
            'Content-Type': 'application/json'
        },
        json={
            'identity_ids': [identity_id],
            'parameters': {
                'message': message  # Optional, 300 char limit. Only 5/month with note!
            },
            'input': {
                'linkedin_profile_url': lead.linkedin_profile_url
            }
        }
    )
    
    if not response.ok:
        raise EdgesError(response.json())
    
    data = response.json()
    
    # Update lead status
    await update_lead(lead.id,
        connection_status='PENDING',
        connection_sent_at=datetime.now().isoformat(),
        last_contacted_at=datetime.now().isoformat()
    )
    
    # Log the action
    await log_outreach(lead.id, 'CONNECT', data['run_id'])
    
    return data

async function sendMessage(identityId, lead, messageText) {
  const response = await fetch('https://api.edges.run/v1/actions/linkedin-message-profile/run/live', {
    method: 'POST',
    headers: {
      'X-API-Key': EDGES_API_KEY,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({
      identity_ids: [identityId],
      parameters: {
        message: messageText
      },
      input: {
        linkedin_profile_url: lead.linkedin_profile_url
      }
    })
  });
  
  if (!response.ok) {
    const error = await response.json();
    throw error;
  }
  
  const data = await response.json();
  
  // Update lead status
  await updateLead(lead.id, {
    sequence_status: 'WAITING_REPLY',
    last_contacted_at: new Date().toISOString()
  });
  
  // Log the action
  await logOutreach(lead.id, 'MESSAGE', data.run_id);
  
  return data;
}

async def send_message(identity_id: str, lead, message_text: str):
    response = requests.post(
        'https://api.edges.run/v1/actions/linkedin-message-profile/run/live',
        headers={
            'X-API-Key': EDGES_API_KEY,
            'Content-Type': 'application/json'
        },
        json={
            'identity_ids': [identity_id],
            'parameters': {
                'message': message_text
            },
            'input': {
                'linkedin_profile_url': lead.linkedin_profile_url
            }
        }
    )
    
    if not response.ok:
        raise EdgesError(response.json())
    
    data = response.json()
    
    # Update lead status
    await update_lead(lead.id,
        sequence_status='WAITING_REPLY',
        last_contacted_at=datetime.now().isoformat()
    )
    
    # Log the action
    await log_outreach(lead.id, 'MESSAGE', data['run_id'])
    
    return data

async function canSendOutreach(identityId, lead) {
  // Check 1: Already replied?
  if (lead.sequence_status === 'REPLIED') {
    return { canSend: false, reason: 'ALREADY_REPLIED' };
  }
  
  // Check 2: Pending connection request?
  if (lead.connection_status === 'PENDING') {
    const daysSinceSent = daysBetween(lead.connection_sent_at, new Date());
    if (daysSinceSent < 21) {
      return { canSend: false, reason: 'CONNECTION_PENDING' };
    }
  }
  
  // Check 3: Existing conversation? (from synced data)
  const existingConv = await db.conversations.findFirst({
    where: {
      identity_id: identityId,
      participant_profile_id: lead.linkedin_profile_id
    }
  });
  
  if (existingConv) {
    const userProfileId = await getUserProfileId(identityId);
    
    // Lead replied - last message is from them
    if (existingConv.last_message_sender_id !== userProfileId) {
      await updateLead(lead.id, { sequence_status: 'REPLIED' });
      return { canSend: false, reason: 'LEAD_REPLIED' };
    }
    
    // We already messaged, check cooldown
    const daysSinceMessage = daysBetween(existingConv.last_message_at, new Date());
    if (daysSinceMessage < 7) {
      return { canSend: false, reason: 'COOLDOWN_ACTIVE' };
    }
  }
  
  return { canSend: true };
}

function daysBetween(date1, date2) {
  const diffMs = new Date(date2) - new Date(date1);
  return Math.floor(diffMs / (1000 * 60 * 60 * 24));
}

from datetime import datetime

async def can_send_outreach(identity_id: str, lead):
    # Check 1: Already replied?
    if lead.sequence_status == 'REPLIED':
        return {'can_send': False, 'reason': 'ALREADY_REPLIED'}
    
    # Check 2: Pending connection request?
    if lead.connection_status == 'PENDING':
        days_since_sent = days_between(lead.connection_sent_at, datetime.now())
        if days_since_sent < 21:
            return {'can_send': False, 'reason': 'CONNECTION_PENDING'}
    
    # Check 3: Existing conversation? (from synced data)
    existing_conv = await db.conversations.find_first(
        identity_id=identity_id,
        participant_profile_id=lead.linkedin_profile_id
    )
    
    if existing_conv:
        user_profile_id = await get_user_profile_id(identity_id)
        
        # Lead replied - last message is from them
        if existing_conv.last_message_sender_id != user_profile_id:
            await update_lead(lead.id, sequence_status='REPLIED')
            return {'can_send': False, 'reason': 'LEAD_REPLIED'}
        
        # We already messaged, check cooldown
        days_since_message = days_between(existing_conv.last_message_at, datetime.now())
        if days_since_message < 7:
            return {'can_send': False, 'reason': 'COOLDOWN_ACTIVE'}
    
    return {'can_send': True}

def days_between(date1, date2):
    diff = date2 - date1
    return diff.days

function getSendTime(lead) {
  // Default business hours: 9am - 6pm in lead's timezone
  const targetHour = 9 + Math.floor(Math.random() * 9);  // Random hour 9-17
  const targetMinute = Math.floor(Math.random() * 60);
  
  // Get lead's timezone offset (you'd lookup from location)
  const leadTimezone = getTimezoneFromLocation(lead.location) || 'America/New_York';
  
  const now = new Date();
  const sendTime = new Date(now);
  
  // Set to target time in lead's timezone
  sendTime.setHours(targetHour, targetMinute, 0, 0);
  
  // If it's past business hours today, schedule for tomorrow
  const nowInLeadTz = new Date(now.toLocaleString('en-US', { timeZone: leadTimezone }));
  if (nowInLeadTz.getHours() >= 18) {
    sendTime.setDate(sendTime.getDate() + 1);
  }
  
  // Skip weekends
  while (sendTime.getDay() === 0 || sendTime.getDay() === 6) {
    sendTime.setDate(sendTime.getDate() + 1);
  }
  
  return sendTime;
}

// Helper: Map location to timezone (implement based on your data)
function getTimezoneFromLocation(location) {
  const timezoneMap = {
    'San Francisco': 'America/Los_Angeles',
    'New York': 'America/New_York',
    'London': 'Europe/London',
    'Paris': 'Europe/Paris',
    // Add more mappings
  };
  
  for (const [city, tz] of Object.entries(timezoneMap)) {
    if (location?.includes(city)) return tz;
  }
  return null;
}

from datetime import datetime, timedelta
import random

def get_send_time(lead):
    """Calculate optimal send time in lead's business hours."""
    # Default business hours: 9am - 6pm in lead's timezone
    target_hour = 9 + random.randint(0, 8)  # Random hour 9-17
    target_minute = random.randint(0, 59)
    
    # Get lead's timezone (you'd lookup from location)
    lead_timezone = get_timezone_from_location(lead.location) or 'America/New_York'
    
    now = datetime.now()
    send_time = now.replace(hour=target_hour, minute=target_minute, second=0, microsecond=0)
    
    # If it's past business hours today, schedule for tomorrow
    if now.hour >= 18:
        send_time += timedelta(days=1)
    
    # Skip weekends (0=Monday, 6=Sunday in Python)
    while send_time.weekday() >= 5:  # Saturday=5, Sunday=6
        send_time += timedelta(days=1)
    
    return send_time

def get_timezone_from_location(location: str) -> str:
    """Map location to timezone. Implement based on your data."""
    timezone_map = {
        'San Francisco': 'America/Los_Angeles',
        'New York': 'America/New_York',
        'London': 'Europe/London',
        'Paris': 'Europe/Paris',
        # Add more mappings
    }
    
    if not location:
        return None
        
    for city, tz in timezone_map.items():
        if city in location:
            return tz
    return None

// Send async action with custom_data
async function sendAsyncMessage(identityId, lead) {
  const response = await fetch('https://api.edges.run/v1/actions/linkedin-message-profile/run/async', {
    method: 'POST',
    headers: {
      'X-API-Key': EDGES_API_KEY,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({
      identity_ids: [identityId],
      parameters: {
        message: 'Hi! I wanted to connect...'
      },
      inputs: [{
        linkedin_profile_url: lead.linkedin_profile_url,
        custom_data: {
          lead_id: lead.id,
          step: 'initial_message',
          identity_id: identityId
        }
      }],
      callback: {
        url: 'https://your-app.com/webhooks/edges'
      }
    })
  });
  
  return response.json();
}

// Handle callback in your webhook endpoint
app.post('/webhooks/edges', async (req, res) => {
  const { run_id, status, output, custom_data } = req.body;
  
  // custom_data contains exactly what you sent
  const { lead_id, step, identity_id } = custom_data;
  
  if (status === 'SUCCESS') {
    await updateLead(lead_id, {
      sequence_status: 'WAITING_REPLY',
      last_contacted_at: new Date().toISOString()
    });
  } else {
    await logOutreachError(lead_id, output.error_label);
  }
  
  res.status(200).send('OK');
});

import requests
from flask import Flask, request

# Send async action with custom_data
async def send_async_message(identity_id: str, lead):
    response = requests.post(
        'https://api.edges.run/v1/actions/linkedin-message-profile/run/async',
        headers={
            'X-API-Key': EDGES_API_KEY,
            'Content-Type': 'application/json'
        },
        json={
            'identity_ids': [identity_id],
            'parameters': {
                'message': 'Hi! I wanted to connect...'
            },
            'inputs': [{
                'linkedin_profile_url': lead.linkedin_profile_url,
                'custom_data': {
                    'lead_id': str(lead.id),
                    'step': 'initial_message',
                    'identity_id': identity_id
                }
            }],
            'callback': {
                'url': 'https://your-app.com/webhooks/edges'
            }
        }
    )
    return response.json()

# Handle callback in your webhook endpoint
app = Flask(__name__)

@app.route('/webhooks/edges', methods=['POST'])
async def handle_edges_callback():
    data = request.json
    run_id = data['run_id']
    status = data['status']
    output = data.get('output', {})
    custom_data = data.get('custom_data', {})
    
    # custom_data contains exactly what you sent
    lead_id = custom_data.get('lead_id')
    step = custom_data.get('step')
    identity_id = custom_data.get('identity_id')
    
    if status == 'SUCCESS':
        await update_lead(lead_id,
            sequence_status='WAITING_REPLY',
            last_contacted_at=datetime.now().isoformat()
        )
    else:
        await log_outreach_error(lead_id, output.get('error_label'))
    
    return 'OK', 200

// Schedule daily connections sync (use /run/schedule)
// Or call this on a cron job using /run/async
async function syncConnections(identityId) {
  const response = await fetch('https://api.edges.run/v1/actions/linkedin-extract-connections/run/live', {
    method: 'POST',
    headers: {
      'X-API-Key': EDGES_API_KEY,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({
      identity_ids: [identityId]
    })
  });
  
  const connections = await response.json();
  
  // Update your connections cache
  for (const conn of connections) {
    await upsertConnection({
      identity_id: identityId,
      linkedin_profile_id: conn.linkedin_profile_id,
      linkedin_profile_url: conn.linkedin_profile_url,
      connected_at: conn.connected_at
    });
  }
  
  return connections.length;
}

async def sync_connections(identity_id: str):
    """Sync connections - call daily or after identity setup."""
    response = requests.post(
        'https://api.edges.run/v1/actions/linkedin-extract-connections/run/live',
        headers={
            'X-API-Key': EDGES_API_KEY,
            'Content-Type': 'application/json'
        },
        json={
            'identity_ids': [identity_id]
        }
    )
    
    connections = response.json()
    
    # Update your connections cache
    for conn in connections:
        await upsert_connection(
            identity_id=identity_id,
            linkedin_profile_id=conn['linkedin_profile_id'],
            linkedin_profile_url=conn['linkedin_profile_url'],
            connected_at=conn.get('connected_at')
        )
    
    return len(connections)

async function executeOutreachWithSync(identityId, lead) {
  // Step 1: Fresh sync for this specific lead
  // Note: extractConversations() wraps the API call from Section 8
  const conversations = await extractConversations(identityId, { max_results: 50 });
  
  // Step 2: Check if lead replied since last sync
  const userProfileId = await getUserProfileId(identityId);
  
  for (const conv of conversations) {
    const participantId = conv.participants[0]?.linkedin_profile_id;
    
    if (participantId === lead.linkedin_profile_id) {
      if (conv.last_message?.linkedin_profile_id !== userProfileId) {
        // Lead replied! Abort outreach
        await updateLead(lead.id, { sequence_status: 'REPLIED' });
        return { status: 'skipped', reason: 'LEAD_REPLIED_SINCE_SYNC' };
      }
    }
  }
  
  // Step 3: Safe to send
  return await sendMessage(identityId, lead);
}

async def execute_outreach_with_sync(identity_id: str, lead):
    """Sync conversations immediately before sending to prevent race conditions."""
    # Step 1: Fresh sync for this specific lead
    # Note: extract_conversations() wraps the API call from Section 8
    conversations = await extract_conversations(identity_id, max_results=50)
    
    # Step 2: Check if lead replied since last sync
    user_profile_id = await get_user_profile_id(identity_id)
    
    for conv in conversations:
        participant_id = conv['participants'][0].get('linkedin_profile_id')
        
        if participant_id == lead.linkedin_profile_id:
            last_sender = conv.get('last_message', {}).get('linkedin_profile_id')
            if last_sender != user_profile_id:
                # Lead replied! Abort outreach
                await update_lead(lead.id, sequence_status='REPLIED')
                return {'status': 'skipped', 'reason': 'LEAD_REPLIED_SINCE_SYNC'}
    
    # Step 3: Safe to send
    return await send_message(identity_id, lead)

// Webhook handler for identity status changes
app.post('/webhooks/edges/integration', async (req, res) => {
  const { identity_uid, event_type, integration_type } = req.body;
  
  if (integration_type !== 'linkedin') {
    return res.status(200).send('OK');
  }
  
  switch (event_type) {
    case 'AUTH_EXPIRED':
      await handleIdentityAuthExpired(identity_uid);
      break;
    case 'AUTH_SUCCESS':
      await handleIdentityAuthRestored(identity_uid);
      break;
  }
  
  res.status(200).send('OK');
});

async function handleIdentityAuthExpired(identityUid) {
  // 1. Pause all leads in active sequences for this identity
  await db.leads.updateMany({
    where: {
      identity_id: identityUid,
      sequence_status: { in: ['IN_SEQUENCE', 'WAITING_REPLY'] }
    },
    data: {
      sequence_status: 'PAUSED',
      paused_reason: 'IDENTITY_AUTH_EXPIRED',
      paused_at: new Date()
    }
  });
  
  // 2. Notify user to re-authenticate
  const user = await getUserFromIdentity(identityUid);
  await sendNotification(user.id, {
    type: 'LINKEDIN_AUTH_EXPIRED',
    message: 'Your LinkedIn connection expired. Please reconnect to resume sequences.',
    action_url: '/settings/integrations'
  });
}

async function handleIdentityAuthRestored(identityUid) {
  // Resume paused leads
  await db.leads.updateMany({
    where: {
      identity_id: identityUid,
      sequence_status: 'PAUSED',
      paused_reason: 'IDENTITY_AUTH_EXPIRED'
    },
    data: {
      sequence_status: 'IN_SEQUENCE',
      paused_reason: null,
      next_action_at: new Date()  // Reschedule immediately
    }
  });
}

from flask import Flask, request

app = Flask(__name__)

@app.route('/webhooks/edges/integration', methods=['POST'])
async def handle_integration_webhook():
    data = request.json
    identity_uid = data['identity_uid']
    event_type = data['event_type']
    integration_type = data['integration_type']
    
    if integration_type != 'linkedin':
        return 'OK', 200
    
    if event_type == 'AUTH_EXPIRED':
        await handle_identity_auth_expired(identity_uid)
    elif event_type == 'AUTH_SUCCESS':
        await handle_identity_auth_restored(identity_uid)
    
    return 'OK', 200

async def handle_identity_auth_expired(identity_uid: str):
    """Pause all sequences when LinkedIn auth expires."""
    # 1. Pause all leads in active sequences for this identity
    await db.execute("""
        UPDATE leads 
        SET sequence_status = 'PAUSED',
            paused_reason = 'IDENTITY_AUTH_EXPIRED',
            paused_at = NOW()
        WHERE identity_id = %s 
        AND sequence_status IN ('IN_SEQUENCE', 'WAITING_REPLY')
    """, [identity_uid])
    
    # 2. Notify user to re-authenticate
    user = await get_user_from_identity(identity_uid)
    await send_notification(
        user_id=user.id,
        notification_type='LINKEDIN_AUTH_EXPIRED',
        message='Your LinkedIn connection expired. Please reconnect to resume sequences.',
        action_url='/settings/integrations'
    )

async def handle_identity_auth_restored(identity_uid: str):
    """Resume paused leads when LinkedIn auth is restored."""
    await db.execute("""
        UPDATE leads 
        SET sequence_status = 'IN_SEQUENCE',
            paused_reason = NULL,
            next_action_at = NOW()
        WHERE identity_id = %s 
        AND sequence_status = 'PAUSED'
        AND paused_reason = 'IDENTITY_AUTH_EXPIRED'
    """, [identity_uid])

async function addLeadToSequence(identityId, lead) {
  // Step 1: Check if already connected
  // Note: Helper functions wrap the API calls shown in previous sections
  const connections = await extractConnections(identityId, { max_results: 1000 });
  
  const isConnected = connections.some(
    conn => conn.linkedin_profile_id === lead.linkedin_profile_id
  );
  
  if (isConnected) {
    lead.connection_status = 'CONNECTED';
    lead.current_step = 3;  // Skip to message step
    await saveLead(lead);
    return;
  }
  
  // Step 2: Check if conversation exists (lead may have messaged first)
  const conversations = await extractConversations(identityId, { max_results: 100 });
  
  for (const conv of conversations) {
    if (conv.participants[0]?.linkedin_profile_id === lead.linkedin_profile_id) {
      // Existing conversation! Check who initiated
      const userProfileId = await getUserProfileId(identityId);
      
      if (conv.last_message?.linkedin_profile_id === lead.linkedin_profile_id) {
        lead.sequence_status = 'REPLIED';
        lead.replied_at = conv.last_message.delivered_at;
        await saveLead(lead);
        return;  // Don't start sequence, human should handle
      }
    }
  }
  
  // Step 3: Normal start
  lead.sequence_status = 'IN_SEQUENCE';
  lead.current_step = 0;
  lead.next_action_at = new Date();
  await saveLead(lead);
}

async def add_lead_to_sequence(identity_id: str, lead):
    """Add a lead to sequence with edge case handling."""
    # Step 1: Check if already connected
    # Note: Helper functions wrap the API calls shown in previous sections
    connections = await extract_connections(identity_id, max_results=1000)
    
    is_connected = any(
        conn['linkedin_profile_id'] == lead.linkedin_profile_id 
        for conn in connections
    )
    
    if is_connected:
        lead.connection_status = 'CONNECTED'
        lead.current_step = 3  # Skip to message step
        await save_lead(lead)
        return
    
    # Step 2: Check if conversation exists (lead may have messaged first)
    conversations = await extract_conversations(identity_id, max_results=100)
    
    for conv in conversations:
        participant_id = conv['participants'][0].get('linkedin_profile_id')
        
        if participant_id == lead.linkedin_profile_id:
            # Existing conversation! Check who initiated
            user_profile_id = await get_user_profile_id(identity_id)
            last_sender = conv.get('last_message', {}).get('linkedin_profile_id')
            
            if last_sender == lead.linkedin_profile_id:
                lead.sequence_status = 'REPLIED'
                lead.replied_at = conv['last_message']['delivered_at']
                await save_lead(lead)
                return  # Don't start sequence, human should handle
    
    # Step 3: Normal start
    lead.sequence_status = 'IN_SEQUENCE'
    lead.current_step = 0
    lead.next_action_at = datetime.now()
    await save_lead(lead)

// Main orchestrator - runs daily via cron
async function runDailyOutreachForAllIdentities() {
  // Get all active identities
  const identities = await db.identities.findMany({
    where: { status: 'ACTIVE', linkedin_connected: true }
  });

  for (const identity of identities) {
    try {
      await runOutreachForIdentity(identity.id);
    } catch (error) {
      console.error(`Failed for identity ${identity.id}:`, error);
      // Continue with other identities
    }
  }
}

async function runOutreachForIdentity(identityId) {
  // Step 1: Sync conversations to detect replies (Section 8)
  await syncConversations(identityId);

  // Step 2: Get leads due for action today
  const leads = await db.leads.findMany({
    where: {
      identity_id: identityId,
      sequence_status: 'IN_SEQUENCE',
      next_action_at: { lte: new Date() }
    }
  });

  for (const lead of leads) {
    // Step 3: De-dup check (Section 10)
    const { canSend, reason } = await canSendOutreach(identityId, lead);
    if (!canSend) {
      console.log(`Skipping lead ${lead.id}: ${reason}`);
      continue;
    }

    // Step 4: Sync-before-send for race condition prevention (Section 14)
    const replyCheck = await checkForRecentReply(identityId, lead);
    if (replyCheck.hasReplied) {
      await updateLead(lead.id, { sequence_status: 'REPLIED' });
      continue;
    }

    // Step 5: Calculate send time for business hours (Section 11)
    const sendTime = getSendTime(lead);
    if (sendTime > new Date()) {
      await updateLead(lead.id, { next_action_at: sendTime });
      continue;
    }

    // Step 6: Execute the action (Section 9) with error handling (Section 7)
    await executeWithErrorHandling(identityId, lead, async () => {
      return await executeSequenceStep(identityId, lead);
    });
  }
}

async function executeSequenceStep(identityId, lead) {
  const SEQUENCE_STEPS = [
    { step: 0, action: 'visit', delayDays: 1 },
    { step: 1, action: 'connect', delayDays: 3 },
    { step: 2, action: 'follow_up_connect', delayDays: 4 },  // If not connected after 3 days
    { step: 3, action: 'message', delayDays: 7 },  // After connection accepted
    { step: 4, action: 'follow_up_1', delayDays: 7 },
    { step: 5, action: 'follow_up_2', delayDays: 7 },
    { step: 6, action: 'archive', delayDays: 0 }
  ];

  const currentStep = SEQUENCE_STEPS[lead.current_step];
  
  // Check connection status for message steps
  if (['message', 'follow_up_1', 'follow_up_2'].includes(currentStep.action)) {
    if (lead.connection_status !== 'CONNECTED') {
      // Not connected yet - check if connection was accepted
      // Note: Helper functions wrap API calls (see Section 9)
      const connections = await extractConnections(identityId, { max_results: 100 });
      const isNowConnected = connections.some(
        c => c.linkedin_profile_id === lead.linkedin_profile_id
      );
      
      if (isNowConnected) {
        await updateLead(lead.id, { connection_status: 'CONNECTED' });
      } else {
        // Still not connected - wait or move to follow-up connect
        return;
      }
    }
  }
  
  switch (currentStep.action) {
    case 'visit':
      await visitProfile(identityId, lead);
      break;

    case 'connect':
      await sendConnectionRequest(identityId, lead);
      break;

    case 'message':
    case 'follow_up_1':
    case 'follow_up_2':
      const template = MESSAGE_TEMPLATES[currentStep.action];
      const message = personalizeMessage(template, lead);
      await sendMessage(identityId, lead, message);
      break;

    case 'archive':
      await updateLead(lead.id, { sequence_status: 'ARCHIVED' });
      return;
  }

  // Advance to next step
  await updateLead(lead.id, {
    current_step: lead.current_step + 1,
    next_action_at: addDays(new Date(), currentStep.delayDays),
    last_contacted_at: new Date()
  });
}

// Message templates
const MESSAGE_TEMPLATES = {
  message: "Hi {{first_name}}, I noticed you're working at {{company_name}}. I'd love to connect and share some ideas about...",
  follow_up_1: "Hi {{first_name}}, just following up on my previous message. Would love to hear your thoughts on...",
  follow_up_2: "{{first_name}}, last follow-up from me! If now isn't a good time, no worries. Feel free to reach out whenever..."
};

function personalizeMessage(template, lead) {
  return template
    .replace('{{first_name}}', lead.full_name?.split(' ')[0] || 'there')
    .replace('{{company_name}}', lead.company_name || 'your company');
}

function addDays(date, days) {
  const result = new Date(date);
  result.setDate(result.getDate() + days);
  return result;
}

from datetime import datetime, timedelta

# Main orchestrator - runs daily via cron
async def run_daily_outreach_for_all_identities():
    """Process outreach for all active identities."""
    identities = await db.identities.find_many(
        status='ACTIVE', linkedin_connected=True
    )
    
    for identity in identities:
        try:
            await run_outreach_for_identity(identity.id)
        except Exception as e:
            print(f"Failed for identity {identity.id}: {e}")

async def run_outreach_for_identity(identity_id: str):
    """Process all due leads for a single identity."""
    # Step 1: Sync conversations to detect replies (Section 8)
    await sync_conversations(identity_id)
    
    # Step 2: Get leads due for action today
    leads = await db.leads.find_many(
        identity_id=identity_id,
        sequence_status='IN_SEQUENCE',
        next_action_at__lte=datetime.now()
    )
    
    for lead in leads:
        # Step 3: De-dup check (Section 10)
        result = await can_send_outreach(identity_id, lead)
        if not result['can_send']:
            print(f"Skipping lead {lead.id}: {result['reason']}")
            continue
        
        # Step 4: Sync-before-send for race condition prevention (Section 14)
        if await check_for_recent_reply(identity_id, lead):
            await update_lead(lead.id, sequence_status='REPLIED')
            continue
        
        # Step 5: Calculate send time for business hours (Section 11)
        send_time = get_send_time(lead)
        if send_time > datetime.now():
            await update_lead(lead.id, next_action_at=send_time)
            continue
        
        # Step 6: Execute with error handling (Section 7)
        await execute_with_error_handling(
            identity_id, lead, 
            lambda: execute_sequence_step(identity_id, lead)
        )

SEQUENCE_STEPS = [
    {'step': 0, 'action': 'visit', 'delay_days': 1},
    {'step': 1, 'action': 'connect', 'delay_days': 3},
    {'step': 2, 'action': 'follow_up_connect', 'delay_days': 4},
    {'step': 3, 'action': 'message', 'delay_days': 7},
    {'step': 4, 'action': 'follow_up_1', 'delay_days': 7},
    {'step': 5, 'action': 'follow_up_2', 'delay_days': 7},
    {'step': 6, 'action': 'archive', 'delay_days': 0}
]

MESSAGE_TEMPLATES = {
    'message': "Hi {{first_name}}, I noticed you're working at {{company_name}}. I'd love to connect and share some ideas about...",
    'follow_up_1': "Hi {{first_name}}, just following up on my previous message. Would love to hear your thoughts on...",
    'follow_up_2': "{{first_name}}, last follow-up from me! If now isn't a good time, no worries. Feel free to reach out whenever..."
}

async def execute_sequence_step(identity_id: str, lead):
    """Execute the current step in the lead's sequence."""
    current_step = SEQUENCE_STEPS[lead.current_step]
    action = current_step['action']
    
    # Check connection status for message steps
    if action in ['message', 'follow_up_1', 'follow_up_2']:
        if lead.connection_status != 'CONNECTED':
            # Note: Helper functions wrap API calls (see Section 9)
            connections = await extract_connections(identity_id, max_results=100)
            is_now_connected = any(
                c['linkedin_profile_id'] == lead.linkedin_profile_id 
                for c in connections
            )
            
            if is_now_connected:
                await update_lead(lead.id, connection_status='CONNECTED')
            else:
                return  # Still not connected - wait
    
    if action == 'visit':
        await visit_profile(identity_id, lead)
    elif action == 'connect':
        await send_connection_request(identity_id, lead)
    elif action in ['message', 'follow_up_1', 'follow_up_2']:
        template = MESSAGE_TEMPLATES[action]
        message = personalize_message(template, lead)
        await send_message(identity_id, lead, message)
    elif action == 'archive':
        await update_lead(lead.id, sequence_status='ARCHIVED')
        return
    
    # Advance to next step
    await update_lead(lead.id,
        current_step=lead.current_step + 1,
        next_action_at=datetime.now() + timedelta(days=current_step['delay_days']),
        last_contacted_at=datetime.now()
    )

def personalize_message(template: str, lead) -> str:
    """Replace placeholders with lead data."""
    first_name = lead.full_name.split(' ')[0] if lead.full_name else 'there'
    company = lead.company_name or 'your company'
    
    return template \
        .replace('{{first_name}}', first_name) \
        .replace('{{company_name}}', company)

async function executeAction(identityId, lead, dryRun = false) {
  if (dryRun) {
    // Log what WOULD happen without calling API
    console.log(`DRY RUN: Would send ${lead.current_step} to ${lead.full_name}`);
    return { status: 'dry_run', action: lead.current_step };
  }
  
  // Real execution
  return await executeSequenceStep(identityId, lead);
}

async def execute_action(identity_id: str, lead, dry_run: bool = False):
    if dry_run:
        # Log what WOULD happen without calling API
        print(f"DRY RUN: Would send step {lead.current_step} to {lead.full_name}")
        return {'status': 'dry_run', 'action': lead.current_step}
    
    # Real execution
    return await execute_sequence_step(identity_id, lead)

// Jest example - mock Edges responses
describe('Outreach Sequence', () => {
  beforeEach(() => {
    jest.spyOn(global, 'fetch').mockImplementation((url) => {
      if (url.includes('extract-conversations')) {
        return Promise.resolve({
          ok: true,
          json: () => Promise.resolve({
            output: {
              results: [{
                linkedin_thread_id: '2-test-thread',
                last_message: {
                  linkedin_profile_id: 123456789,  // Simulated lead reply
                  delivered_at: '2024-01-20T10:00:00Z'
                },
                participants: [{ linkedin_profile_id: 123456789 }]
              }]
            }
          })
        });
      }
      // Add more mock responses...
    });
  });

  test('detects lead reply correctly', async () => {
    const result = await syncConversations('identity_123');
    // Assert lead was marked as replied
  });
});

# pytest example - mock Edges responses
import pytest
from unittest.mock import patch, MagicMock

@pytest.fixture
def mock_edges_client():
    with patch('requests.post') as mock_post:
        mock_response = MagicMock()
        mock_response.ok = True
        mock_response.json.return_value = {
            'output': {
                'results': [{
                    'linkedin_thread_id': '2-test-thread',
                    'last_message': {
                        'linkedin_profile_id': 123456789,  # Simulated lead reply
                        'delivered_at': '2024-01-20T10:00:00Z'
                    },
                    'participants': [{'linkedin_profile_id': 123456789}]
                }]
            }
        }
        mock_post.return_value = mock_response
        yield mock_post

def test_detects_lead_reply(mock_edges_client):
    # Your test here
    result = sync_conversations('identity_123')
    # Assert lead was marked as replied