Retrospective: Growing a 10-Person Dev Team to 50 with Deel 2026 and Slack 5.0 for Remote Onboarding

import os
import logging
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError, SlackClientError
from datetime import datetime
import time

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

# Initialize Slack 5.0 client with bot token (xoxb- prefix for Slack 5.0)
SLACK_BOT_TOKEN = os.getenv("SLACK_5_BOT_TOKEN")
if not SLACK_BOT_TOKEN:
    logger.error("Missing SLACK_5_BOT_TOKEN environment variable")
    raise ValueError("Slack 5.0 bot token not configured")

client = WebClient(token=SLACK_BOT_TOKEN)

# Define onboarding configuration for Slack 5.0
ONBOARDING_CHANNEL_PREFIX = "onboard-"
WELCOME_CHANNEL = "new-hires"
TEAM_LEAD_CHANNEL = "eng-leads"
REQUIRED_APPS = ["github", "jira", "deel", "1password"]

def retry_on_rate_limit(func, *args, **kwargs):
    """Slack 5.0 rate limits are 100 requests/min for Workspaces under 50 users, handle retries."""
    max_retries = 3
    retry_delay = 60  # Seconds to wait on rate limit
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except SlackApiError as e:
            if e.response["error"] == "rate_limited":
                logger.warning(f"Rate limited, retrying in {retry_delay}s (attempt {attempt+1}/{max_retries})")
                time.sleep(retry_delay)
                continue
            raise  # Re-raise non-rate limit errors
    logger.error("Max retries exceeded for rate limited request")
    raise SlackClientError("Exceeded max retries for Slack API request")

def provision_slack_onboarding(new_hire_email: str, team: str, start_date: str) -> dict:
    """
    Automates Slack 5.0 onboarding for new dev hires using 2026-compliant workflows.
    Args:
        new_hire_email: Corporate email of new hire
        team: Engineering sub-team (backend, frontend, devops)
        start_date: ISO formatted date string (YYYY-MM-DD)
    Returns:
        Dictionary with Slack user ID and provisioned channels
    """
    try:
        # 1. Invite user to Slack workspace (Slack 5.0 supports batch invites via email)
        logger.info(f"Inviting {new_hire_email} to Slack workspace")
        invite_resp = retry_on_rate_limit(
            client.users_invite,
            email=new_hire_email,
            channels=[],  # We'll add channels after user is created
            set_active=True
        )
        if not invite_resp["ok"]:
            raise SlackApiError("users_invite failed", invite_resp)
        user_id = invite_resp["user"]["id"]
        logger.info(f"Provisioned Slack user {user_id} for {new_hire_email}")

        # 2. Create team-specific onboarding channel (Slack 5.0 Canvas auto-attaches to new channels)
        channel_name = f"{ONBOARDING_CHANNEL_PREFIX}{user_id.split('|')[0]}"
        logger.info(f"Creating onboarding channel {channel_name}")
        channel_resp = retry_on_rate_limit(
            client.conversations_create,
            name=channel_name,
            is_private=True,
            team_id=os.getenv("SLACK_WORKSPACE_ID")
        )
        if not channel_resp["ok"]:
            raise SlackApiError("conversations_create failed", channel_resp)
        channel_id = channel_resp["channel"]["id"]

        # 3. Invite new hire and team lead to onboarding channel
        logger.info(f"Inviting {user_id} to {channel_name}")
        retry_on_rate_limit(
            client.conversations_invite,
            channel=channel_id,
            users=[user_id]
        )
        # Get team lead ID from team mapping
        team_leads = {
            "backend": "U123456",
            "frontend": "U789012",
            "devops": "U345678"
        }
        lead_id = team_leads.get(team)
        if lead_id:
            retry_on_rate_limit(
                client.conversations_invite,
                channel=channel_id,
                users=[lead_id]
            )

        # 4. Post welcome message with Slack 5.0 Canvas link (auto-generated for onboarding)
        canvas_link = f"https://your-workspace.slack.com/canvas/{channel_id}"
        welcome_msg = f"Welcome to the team! Your onboarding Canvas is here: {canvas_link}. Start date: {start_date}."
        retry_on_rate_limit(
            client.chat_postMessage,
            channel=channel_id,
            text=welcome_msg,
            username="Onboarding Bot",
            icon_emoji=":rocket:"
        )

        # 5. Add user to standard engineering channels
        standard_channels = ["eng-general", "eng-announcements", team]
        for ch in standard_channels:
            try:
                retry_on_rate_limit(
                    client.conversations_invite,
                    channel=ch,
                    users=[user_id]
                )
            except SlackApiError as e:
                logger.warning(f"Failed to add {user_id} to {ch}: {e}")
                continue

        # 6. Trigger Deel 2026 webhook to sync Slack user ID to HR record
        # (Deel 2026 supports Slack-native webhook integration)
        import requests
        deel_webhook = os.getenv("DEEL_2026_WEBHOOK_URL")
        if deel_webhook:
            requests.post(
                deel_webhook,
                json={"slack_user_id": user_id, "email": new_hire_email},
                headers={"Content-Type": "application/json"}
            )

        return {"user_id": user_id, "onboarding_channel": channel_id}

    except SlackApiError as e:
        logger.error(f"Slack API error during onboarding: {e.response['error']}")
        raise
    except Exception as e:
        logger.error(f"Unexpected error provisioning Slack onboarding: {str(e)}")
        raise

if __name__ == "__main__":
    # Example usage for a new backend hire
    try:
        result = provision_slack_onboarding(
            new_hire_email="new.dev@company.com",
            team="backend",
            start_date="2026-03-15"
        )
        logger.info(f"Onboarding complete: {result}")
    except Exception as e:
        logger.error(f"Onboarding failed: {str(e)}")

import os
import logging
import requests
from datetime import datetime
from typing import Dict, List
import time

# Configure logging for Deel 2026 audit compliance
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("deel_onboarding.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# Deel 2026 API configuration (OAuth 2.0, required for 2026+ compliance)
DEEL_API_BASE = "https://api.deel.com/v2026"
DEEL_CLIENT_ID = os.getenv("DEEL_2026_CLIENT_ID")
DEEL_CLIENT_SECRET = os.getenv("DEEL_2026_CLIENT_SECRET")
DEEL_WEBHOOK_SECRET = os.getenv("DEEL_2026_WEBHOOK_SECRET")

if not all([DEEL_CLIENT_ID, DEEL_CLIENT_SECRET]):
    logger.error("Missing Deel 2026 OAuth credentials")
    raise ValueError("Deel 2026 client credentials not configured")

class Deel2026Client:
    """Client for Deel 2026 HR API, compliant with global tax regulations."""
    def __init__(self):
        self.access_token = None
        self.token_expiry = 0
        self.session = requests.Session()
        self.session.headers.update({"Content-Type": "application/json"})

    def _refresh_token(self) -> None:
        """Refresh OAuth 2.0 token for Deel 2026 API (tokens last 1 hour)."""
        if self.access_token and time.time() < self.token_expiry:
            return
        logger.info("Refreshing Deel 2026 access token")
        try:
            resp = self.session.post(
                f"{DEEL_API_BASE}/oauth/token",
                json={
                    "client_id": DEEL_CLIENT_ID,
                    "client_secret": DEEL_CLIENT_SECRET,
                    "grant_type": "client_credentials",
                    "scope": "onboarding:write payroll:write compliance:read"
                }
            )
            resp.raise_for_status()
            data = resp.json()
            self.access_token = data["access_token"]
            self.token_expiry = time.time() + data["expires_in"] - 60  # Buffer 1 minute
            self.session.headers.update({"Authorization": f"Bearer {self.access_token}"})
            logger.info("Deel 2026 token refreshed successfully")
        except requests.exceptions.RequestException as e:
            logger.error(f"Failed to refresh Deel token: {str(e)}")
            raise

    def create_contractor(self, hire_data: Dict) -> Dict:
        """
        Create a new contractor record in Deel 2026 with automated tax form collection.
        Args:
            hire_data: Dictionary with name, email, country, start_date, team, equity
        Returns:
            Deel contractor ID and onboarding link
        """
        self._refresh_token()
        required_fields = ["email", "full_name", "country", "start_date", "team"]
        for field in required_fields:
            if field not in hire_data:
                raise ValueError(f"Missing required field {field} in hire data")

        logger.info(f"Creating Deel 2026 contractor for {hire_data['email']}")
        try:
            # Deel 2026 supports batch tax form generation based on hire country
            payload = {
                "contractor": {
                    "email": hire_data["email"],
                    "full_name": hire_data["full_name"],
                    "country": hire_data["country"],
                    "start_date": hire_data["start_date"],
                    "contractor_type": "individual",
                    "payment_currency": "USD",
                    "hourly_rate": hire_data.get("hourly_rate", 0),
                    "equity_grant": hire_data.get("equity", {}),
                    "team": hire_data["team"],
                    "custom_fields": {
                        "slack_user_id": hire_data.get("slack_user_id", ""),
                        "github_username": hire_data.get("github_username", "")
                    }
                },
                "onboarding_settings": {
                    "auto_collect_tax_forms": True,
                    "send_welcome_email": True,
                    "required_documents": ["id_proof", "tax_form", "bank_details"]
                }
            }
            resp = self.session.post(
                f"{DEEL_API_BASE}/contractors",
                json=payload
            )
            resp.raise_for_status()
            result = resp.json()
            logger.info(f"Created Deel contractor {result['contractor']['id']} for {hire_data['email']}")
            return {
                "contractor_id": result["contractor"]["id"],
                "onboarding_link": result["onboarding_link"],
                "tax_forms_required": result["required_tax_forms"]
            }
        except requests.exceptions.HTTPError as e:
            logger.error(f"Deel API error creating contractor: {e.response.text}")
            raise
        except Exception as e:
            logger.error(f"Unexpected error creating Deel contractor: {str(e)}")
            raise

    def sync_payroll_adjustment(self, contractor_id: str, adjustment: Dict) -> None:
        """Sync one-time payroll adjustments (e.g., signing bonus) via Deel 2026 API."""
        self._refresh_token()
        logger.info(f"Syncing payroll adjustment for contractor {contractor_id}")
        try:
            resp = self.session.post(
                f"{DEEL_API_BASE}/contractors/{contractor_id}/payroll-adjustments",
                json=adjustment
            )
            resp.raise_for_status()
            logger.info(f"Payroll adjustment synced for {contractor_id}")
        except requests.exceptions.HTTPError as e:
            logger.error(f"Failed to sync payroll adjustment: {e.response.text}")
            raise

def handle_deel_webhook(webhook_data: Dict, signature: str) -> bool:
    """
    Validate and process Deel 2026 webhooks (e.g., tax form submitted, onboarding complete).
    Args:
        webhook_data: Parsed JSON from Deel webhook
        signature: X-Deel-Signature header for validation
    Returns:
        True if webhook processed successfully
    """
    import hmac
    import hashlib
    # Validate webhook signature (Deel 2026 uses HMAC-SHA256)
    if not DEEL_WEBHOOK_SECRET:
        logger.error("Missing Deel webhook secret")
        return False
    expected_sig = hmac.new(
        DEEL_WEBHOOK_SECRET.encode(),
        str(webhook_data).encode(),
        hashlib.sha256
    ).hexdigest()
    if not hmac.compare_digest(expected_sig, signature):
        logger.warning("Invalid Deel webhook signature")
        return False

    event_type = webhook_data.get("event_type")
    if event_type == "contractor.onboarding_complete":
        contractor_id = webhook_data["data"]["contractor_id"]
        logger.info(f"Deel onboarding complete for contractor {contractor_id}")
        # Trigger Slack 5.0 notification to team lead
        return True
    elif event_type == "contractor.tax_form_submitted":
        logger.info(f"Tax form submitted for contractor {webhook_data['data']['contractor_id']}")
        return True
    else:
        logger.info(f"Unhandled Deel webhook event: {event_type}")
        return True

if __name__ == "__main__":
    # Example: Create a new contractor for Deel 2026
    client = Deel2026Client()
    try:
        hire_data = {
            "email": "new.dev@company.com",
            "full_name": "Alex Smith",
            "country": "US",
            "start_date": "2026-03-15",
            "team": "backend",
            "hourly_rate": 85,
            "equity": {"amount": 10000, "type": "ISO"},
            "slack_user_id": "U123456"
        }
        result = client.create_contractor(hire_data)
        logger.info(f"Deel onboarding link: {result['onboarding_link']}")
    except Exception as e:
        logger.error(f"Deel contractor creation failed: {str(e)}")

import os
import logging
from datetime import datetime, timedelta
from typing import List, Dict
import pandas as pd
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
import requests

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Initialize clients (reuse earlier config)
SLACK_BOT_TOKEN = os.getenv("SLACK_5_BOT_TOKEN")
DEEL_API_BASE = "https://api.deel.com/v2026"
DEEL_ACCESS_TOKEN = os.getenv("DEEL_2026_ACCESS_TOKEN")

slack_client = WebClient(token=SLACK_BOT_TOKEN)
deel_headers = {"Authorization": f"Bearer {DEEL_ACCESS_TOKEN}", "Content-Type": "application/json"}

class OnboardingMetricsCalculator:
    """Calculate onboarding metrics from Slack 5.0 and Deel 2026 data."""
    def __init__(self, start_date: str, end_date: str):
        self.start_date = datetime.fromisoformat(start_date)
        self.end_date = datetime.fromisoformat(end_date)
        self.metrics = {
            "total_hires": 0,
            "time_to_productive": [],
            "churned_hires": 0,
            "onboarding_costs": [],
            "slack_ticket_volume": 0
        }

    def _get_slack_new_hires(self) -> List[Dict]:
        """Fetch new Slack users from Slack 5.0 API within date range."""
        logger.info("Fetching new Slack users from Slack 5.0")
        new_hires = []
        try:
            # Slack 5.0 supports pagination for user lists
            cursor = None
            while True:
                params = {"limit": 200, "cursor": cursor}
                resp = slack_client.users_list(**params)
                if not resp["ok"]:
                    raise SlackApiError("users_list failed", resp)
                for user in resp["members"]:
                    # Filter for users created in date range, not bots
                    created_ts = datetime.fromtimestamp(float(user["created"]))
                    if self.start_date <= created_ts <= self.end_date and not user["is_bot"]:
                        new_hires.append({
                            "user_id": user["id"],
                            "email": user["profile"]["email"],
                            "created": created_ts.isoformat()
                        })
                cursor = resp.get("response_metadata", {}).get("next_cursor")
                if not cursor:
                    break
            logger.info(f"Found {len(new_hires)} new Slack users in date range")
            return new_hires
        except SlackApiError as e:
            logger.error(f"Slack API error fetching users: {e.response['error']}")
            raise

    def _get_deel_hire_data(self, emails: List[str]) -> Dict[str, Dict]:
        """Fetch Deel 2026 contractor data for given emails."""
        logger.info("Fetching Deel 2026 contractor data")
        deel_data = {}
        try:
            # Deel 2026 supports batch lookup by email
            resp = requests.post(
                f"{DEEL_API_BASE}/contractors/batch-lookup",
                headers=deel_headers,
                json={"emails": emails}
            )
            resp.raise_for_status()
            for contractor in resp.json()["contractors"]:
                deel_data[contractor["email"]] = {
                    "start_date": contractor["start_date"],
                    "onboarding_complete": contractor["onboarding_complete"],
                    "country": contractor["country"],
                    "equity": contractor["equity_grant"]["amount"]
                }
            logger.info(f"Fetched Deel data for {len(deel_data)} contractors")
            return deel_data
        except requests.exceptions.RequestException as e:
            logger.error(f"Deel API error fetching contractors: {str(e)}")
            raise

    def _calculate_time_to_productive(self, slack_user_id: str, start_date: str) -> int:
        """
        Calculate time-to-productive (TTP) in days: days between start date and first merged PR.
        Uses Slack 5.0 to fetch GitHub PR activity via integrated app.
        """
        logger.info(f"Calculating TTP for Slack user {slack_user_id}")
        try:
            # Slack 5.0 integrated GitHub app: list user's PR messages in eng channels
            channels = ["backend", "frontend", "devops", "eng-general"]
            first_pr_date = None
            for channel in channels:
                cursor = None
                while True:
                    resp = slack_client.conversations_history(
                        channel=channel,
                        cursor=cursor,
                        limit=200
                    )
                    if not resp["ok"]:
                        break
                    for msg in resp["messages"]:
                        # Check if message is a GitHub PR merge notification
                        if msg.get("app_id") == "github" and "merged" in msg.get("text", "").lower():
                            # Check if PR author is the new hire (match Slack user to GitHub via Deel custom field)
                            # Simplified for example: assume first merged PR after start date is TTP
                            msg_ts = datetime.fromtimestamp(float(msg["ts"]))
                            if msg_ts >= datetime.fromisoformat(start_date) and not first_pr_date:
                                first_pr_date = msg_ts
                                break
                    cursor = resp.get("response_metadata", {}).get("next_cursor")
                    if not cursor or first_pr_date:
                        break
                if first_pr_date:
                    break
            if not first_pr_date:
                return None  # No PR found, exclude from TTP
            start = datetime.fromisoformat(start_date)
            return (first_pr_date - start).days
        except SlackApiError as e:
            logger.error(f"Error calculating TTP: {e.response['error']}")
            return None

    def calculate_metrics(self) -> Dict:
        """Run full metrics calculation pipeline."""
        logger.info("Starting onboarding metrics calculation")
        # 1. Get new Slack hires
        slack_hires = self._get_slack_new_hires()
        self.metrics["total_hires"] = len(slack_hires)
        if not slack_hires:
            logger.warning("No hires found in date range")
            return self.metrics

        # 2. Get Deel data for hires
        emails = [h["email"] for h in slack_hires]
        deel_data = self._get_deel_hire_data(emails)

        # 3. Calculate TTP and churn
        for hire in slack_hires:
            email = hire["email"]
            if email not in deel_data:
                logger.warning(f"No Deel data for {email}, skipping")
                continue
            deel_hire = deel_data[email]
            # Churn: if onboarding not complete after 90 days
            start_date = datetime.fromisoformat(deel_hire["start_date"])
            if not deel_hire["onboarding_complete"] and (datetime.now() - start_date).days > 90:
                self.metrics["churned_hires"] += 1
                continue
            # Calculate TTP
            ttp = self._calculate_time_to_productive(hire["user_id"], deel_hire["start_date"])
            if ttp:
                self.metrics["time_to_productive"].append(ttp)

        # 4. Calculate cost per hire (Deel 2026 payroll + Slack 5.0 seat costs)
        # Slack 5.0 seat cost: $12.50/user/month, Deel cost: $50/contractor/month
        slack_seat_cost = 12.50 * ((self.end_date - self.start_date).days / 30)
        deel_cost = 50 * self.metrics["total_hires"]
        total_cost = slack_seat_cost + deel_cost + 2100  # Fixed onboarding tool costs
        self.metrics["onboarding_costs"] = total_cost / self.metrics["total_hires"] if self.metrics["total_hires"] else 0

        # 5. Calculate Slack ticket volume (onboarding-related)
        # Count messages in #it-help with "onboard" keyword
        try:
            resp = slack_client.conversations_history(
                channel="it-help",
                oldest=self.start_date.timestamp(),
                latest=self.end_date.timestamp(),
                limit=1000
            )
            if resp["ok"]:
                self.metrics["slack_ticket_volume"] = sum(
                    1 for msg in resp["messages"] if "onboard" in msg.get("text", "").lower()
                )
        except SlackApiError as e:
            logger.error(f"Error fetching IT tickets: {e.response['error']}")

        # Aggregate TTP stats
        if self.metrics["time_to_productive"]:
            self.metrics["avg_ttp"] = sum(self.metrics["time_to_productive"]) / len(self.metrics["time_to_productive"])
            self.metrics["p95_ttp"] = pd.Series(self.metrics["time_to_productive"]).quantile(0.95)
        else:
            self.metrics["avg_ttp"] = 0
            self.metrics["p95_ttp"] = 0

        logger.info(f"Metrics calculation complete: {self.metrics}")
        return self.metrics

if __name__ == "__main__":
    # Calculate metrics for Q1 2026 (scaled to 50 devs)
    calculator = OnboardingMetricsCalculator(
        start_date="2026-01-01",
        end_date="2026-03-31"
    )
    try:
        metrics = calculator.calculate_metrics()
        print(f"Total Hires: {metrics['total_hires']}")
        print(f"Average TTP: {metrics['avg_ttp']:.1f} days")
        print(f"Churn Rate: {(metrics['churned_hires']/metrics['total_hires'])*100:.1f}%")
        print(f"Cost Per Hire: ${metrics['onboarding_costs']:.2f}")
        print(f"Slack IT Tickets: {metrics['slack_ticket_volume']}")
    except Exception as e:
        logger.error(f"Metrics calculation failed: {str(e)}")

# Short snippet to check pending Slack invites (Slack 5.0 API)
pending_invites = client.users_list(include_pending_invites=True)
for invite in pending_invites["pending_invites"]:
    if (datetime.now() - datetime.fromtimestamp(invite["created"])).days > 1:
        client.users_invite_resend(email=invite["email"])

# Deel 2026 custom field update via API
client.update_contractor(
    contractor_id="CTR-12345",
    custom_fields={
        "github_username": "new-dev",
        "ssh_key_fingerprint": "SHA256:abc123"
    }
)

# Alert to Slack if avg TTP exceeds 14 days
if metrics["avg_ttp"] > 14:
    slack_client.chat_postMessage(
        channel="eng-leads",
        text=f"⚠️ Avg TTP is {metrics['avg_ttp']:.1f} days, exceeds 14 day threshold"
    )