How We Grew Slack vs Figma: What You Need to Know

import os
import time
import logging
from slack_sdk.web import WebClient
from slack_sdk.errors import SlackApiError
from dataclasses import dataclass
from typing import List, Dict

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

@dataclass
class BenchmarkConfig:
    """Configuration for Slack message throughput benchmark"""
    workspace_token: str
    channel_id: str
    num_messages: int = 1000
    batch_size: int = 50
    rate_limit_per_sec: int = 100  # Slack's tier 3 rate limit

class SlackThroughputBenchmark:
    def __init__(self, config: BenchmarkConfig):
        self.config = config
        self.client = WebClient(token=config.workspace_token)
        self.success_count = 0
        self.error_count = 0
        self.latencies: List[float] = []

    def send_batch(self, messages: List[str]) -> None:
        """Send a batch of messages with error handling and latency tracking"""
        for msg in messages:
            start_time = time.perf_counter()
            try:
                response = self.client.chat_postMessage(
                    channel=self.config.channel_id,
                    text=msg,
                    unfurl_links=False
                )
                if response["ok"]:
                    self.success_count += 1
                    latency = (time.perf_counter() - start_time) * 1000  # ms
                    self.latencies.append(latency)
                else:
                    self.error_count += 1
                    logger.warning(f"Slack API returned non-ok response: {response}")
            except SlackApiError as e:
                self.error_count += 1
                logger.error(f"Slack API error: {e.response['error']}")
                if e.response["error"] == "rate_limited":
                    retry_after = int(e.response.headers.get("Retry-After", 1))
                    logger.info(f"Rate limited, retrying after {retry_after}s")
                    time.sleep(retry_after)
            except Exception as e:
                self.error_count += 1
                logger.error(f"Unexpected error sending message: {str(e)}")

    def run(self) -> Dict[str, float]:
        """Execute full benchmark and return metrics"""
        logger.info(f"Starting Slack throughput benchmark: {self.config.num_messages} messages")
        start_time = time.perf_counter()

        # Generate test messages
        test_messages = [f"Benchmark message {i} - {time.time_ns()}" for i in range(self.config.num_messages)]

        # Send in batches with rate limiting
        for i in range(0, len(test_messages), self.config.batch_size):
            batch = test_messages[i:i + self.config.batch_size]
            self.send_batch(batch)
            # Enforce rate limit: sleep if we're exceeding per-second limit
            time.sleep(max(0, (self.config.batch_size / self.config.rate_limit_per_sec) - (time.perf_counter() - start_time) % 1))

        total_time = time.perf_counter() - start_time
        throughput = self.success_count / total_time if total_time > 0 else 0
        p99_latency = sorted(self.latencies)[int(0.99 * len(self.latencies))] if self.latencies else 0

        return {
            "total_time_sec": total_time,
            "throughput_msg_per_sec": throughput,
            "success_count": self.success_count,
            "error_count": self.error_count,
            "p99_latency_ms": p99_latency,
            "median_latency_ms": sorted(self.latencies)[len(self.latencies)//2] if self.latencies else 0
        }

if __name__ == "__main__":
    # Load config from environment variables (never hardcode tokens!)
    config = BenchmarkConfig(
        workspace_token=os.getenv("SLACK_WORKSPACE_TOKEN"),
        channel_id=os.getenv("SLACK_BENCH_CHANNEL_ID"),
        num_messages=int(os.getenv("BENCH_NUM_MESSAGES", 1000)),
        batch_size=int(os.getenv("BENCH_BATCH_SIZE", 50))
    )

    if not config.workspace_token or not config.channel_id:
        logger.error("Missing SLACK_WORKSPACE_TOKEN or SLACK_BENCH_CHANNEL_ID env vars")
        exit(1)

    benchmark = SlackThroughputBenchmark(config)
    results = benchmark.run()

    logger.info("=== Benchmark Results ===")
    for key, value in results.items():
        logger.info(f"{key}: {value:.2f}")

// Figma Plugin: Concurrent Edit Conflict Benchmark
// Run via Figma Plugin Editor, requires Figma Enterprise API access
// @ts-ignore: Figma API types are injected at runtime
declare const figma: any;
declare const console: any;

import { FigmaApi } from '@figma/api';
import { performance } from 'perf_hooks';

// Configuration interface for benchmark
interface BenchmarkConfig {
    apiToken: string;
    fileId: string;
    numEditors: number;
    editsPerEditor: number;
    nodeCount: number; // Number of vector nodes to create per editor
}

// Metrics interface to track results
interface BenchmarkMetrics {
    totalEdits: number;
    conflictCount: number;
    medianLatencyMs: number;
    p99LatencyMs: number;
    successRate: number;
}

class FigmaConcurrencyBenchmark {
    private config: BenchmarkConfig;
    private api: FigmaApi;
    private latencies: number[] = [];
    private conflictCount: number = 0;
    private successCount: number = 0;

    constructor(config: BenchmarkConfig) {
        this.config = config;
        this.api = new FigmaApi({ personalAccessToken: config.apiToken });
    }

    /**
     * Create a test vector node with random properties
     */
    private createTestNode(index: number): any {
        return {
            type: 'RECTANGLE',
            name: `BenchmarkRect_${index}_${Date.now()}`,
            x: Math.random() * 1000,
            y: Math.random() * 1000,
            width: 100 + Math.random() * 200,
            height: 100 + Math.random() * 200,
            fills: [{ type: 'SOLID', color: { r: Math.random(), g: Math.random(), b: Math.random() } }]
        };
    }

    /**
     * Simulate a single editor's edit session with conflict tracking
     */
    private async simulateEditor(editorId: number): Promise {
        const editorToken = `${this.config.apiToken}_editor${editorId}`; // Simulate per-editor tokens
        const editorApi = new FigmaApi({ personalAccessToken: editorToken });

        for (let i = 0; i < this.config.editsPerEditor; i++) {
            const startTime = performance.now();
            try {
                // Fetch current file version to check for conflicts
                const fileResp = await editorApi.getFile(this.config.fileId);
                const currentVersion = fileResp.data.version;

                // Create new node
                const newNode = this.createTestNode(i);

                // Post edit with version check (simulate optimistic locking)
                const editResp = await editorApi.postFileNodes(this.config.fileId, {
                    nodes: [newNode],
                    branch: false
                });

                if (editResp.data.version === currentVersion) {
                    this.conflictCount++;
                    console.warn(`Editor ${editorId}: Conflict detected, version unchanged`);
                } else {
                    this.successCount++;
                }

                const latency = performance.now() - startTime;
                this.latencies.push(latency);
            } catch (err: any) {
                console.error(`Editor ${editorId} error: ${err.message}`);
                // Retry once on rate limit
                if (err.response?.status === 429) {
                    await new Promise(resolve => setTimeout(resolve, 1000));
                    i--; // Retry this edit
                }
            }
        }
    }

    /**
     * Run full benchmark across all simulated editors
     */
    async run(): Promise {
        console.log(`Starting Figma concurrency benchmark: ${this.config.numEditors} editors, ${this.config.editsPerEditor} edits each`);
        const startTime = performance.now();

        // Initialize file with base nodes
        const initNodes = Array.from({ length: this.config.nodeCount }, (_, i) => this.createTestNode(i));
        await this.api.postFileNodes(this.config.fileId, { nodes: initNodes, branch: false });

        // Run editors concurrently
        const editorPromises = Array.from({ length: this.config.numEditors }, (_, i) => this.simulateEditor(i));
        await Promise.all(editorPromises);

        const totalTime = performance.now() - startTime;
        const sortedLatencies = [...this.latencies].sort((a, b) => a - b);
        const totalEdits = this.config.numEditors * this.config.editsPerEditor;

        return {
            totalEdits,
            conflictCount: this.conflictCount,
            medianLatencyMs: sortedLatencies[Math.floor(sortedLatencies.length / 2)] || 0,
            p99LatencyMs: sortedLatencies[Math.floor(sortedLatencies.length * 0.99)] || 0,
            successRate: (this.successCount / totalEdits) * 100
        };
    }
}

// Main execution (runs in Figma plugin context)
(async () => {
    const config: BenchmarkConfig = {
        apiToken: process.env.FIGMA_API_TOKEN || '',
        fileId: process.env.FIGMA_BENCH_FILE_ID || '',
        numEditors: parseInt(process.env.BENCH_NUM_EDITORS || '10'),
        editsPerEditor: parseInt(process.env.BENCH_EDITS_PER_EDITOR || '50'),
        nodeCount: parseInt(process.env.BENCH_NODE_COUNT || '100')
    };

    if (!config.apiToken || !config.fileId) {
        console.error('Missing FIGMA_API_TOKEN or FIGMA_BENCH_FILE_ID');
        figma.closePlugin();
        return;
    }

    const benchmark = new FigmaConcurrencyBenchmark(config);
    try {
        const results = await benchmark.run();
        console.log('=== Figma Benchmark Results ===');
        console.log(JSON.stringify(results, null, 2));
    } catch (err: any) {
        console.error('Benchmark failed:', err.message);
    } finally {
        figma.closePlugin();
    }
})();

import os
import argparse
from dataclasses import dataclass
from typing import Dict, Optional

# Pricing data as of Q1 2024 (publicly available enterprise pricing)
SLACK_PRICING = {
    "pro": 12.50,  # per user per month
    "business_plus": 18.00,
    "enterprise_grid": 25.00,  # Custom, but avg for 10k+ users
    "slack_connect": 5.00  # per external user per month
}

FIGMA_PRICING = {
    "starter": 0.00,
    "professional": 12.00,
    "organization": 45.00,  # per editor per month
    "enterprise": 75.00
}

@dataclass
class TeamConfig:
    """Team configuration for TCO calculation"""
    name: str
    total_users: int
    slack_users: int  # Number of users with Slack access
    figma_editors: int  # Number of Figma editors (viewers are free)
    external_slack_users: int  # External users via Slack Connect
    figma_viewer_ratio: float  # 0.0-1.0, % of total users who are Figma viewers
    use_slack_enterprise: bool
    use_figma_enterprise: bool

class CollaborationTCOCalculator:
    def __init__(self, team_config: TeamConfig):
        self.config = team_config

    def calculate_slack_cost(self) -> Dict[str, float]:
        """Calculate annual Slack TCO"""
        if self.config.use_slack_enterprise:
            per_user_rate = SLACK_PRICING["enterprise_grid"]
        else:
            # Assume 70% Business Plus, 30% Pro for non-enterprise
            per_user_rate = (SLACK_PRICING["business_plus"] * 0.7) + (SLACK_PRICING["pro"] * 0.3)

        user_cost = self.config.slack_users * per_user_rate * 12
        connect_cost = self.config.external_slack_users * SLACK_PRICING["slack_connect"] * 12
        # Add 15% for integrations, support, storage
        total_cost = (user_cost + connect_cost) * 1.15

        return {
            "annual_user_cost": user_cost,
            "annual_connect_cost": connect_cost,
            "annual_total_slack": total_cost,
            "per_user_monthly": per_user_rate
        }

    def calculate_figma_cost(self) -> Dict[str, float]:
        """Calculate annual Figma TCO"""
        if self.config.use_figma_enterprise:
            editor_rate = FIGMA_PRICING["enterprise"]
        else:
            # Assume 60% Organization, 40% Professional for non-enterprise
            editor_rate = (FIGMA_PRICING["organization"] * 0.6) + (FIGMA_PRICING["professional"] * 0.4)

        editor_cost = self.config.figma_editors * editor_rate * 12
        # Viewers are free, but add 10% for admin, SSO, audit logs
        total_cost = editor_cost * 1.10

        return {
            "annual_editor_cost": editor_cost,
            "annual_total_figma": total_cost,
            "per_editor_monthly": editor_rate,
            "free_viewer_count": int(self.config.total_users * self.config.figma_viewer_ratio)
        }

    def calculate_total_tco(self) -> Dict[str, float]:
        """Calculate total cross-tool TCO"""
        slack_costs = self.calculate_slack_cost()
        figma_costs = self.calculate_figma_cost()

        return {
            **slack_costs,
            **figma_costs,
            "total_annual_tco": slack_costs["annual_total_slack"] + figma_costs["annual_total_figma"],
            "slack_percent_of_total": (slack_costs["annual_total_slack"] / (slack_costs["annual_total_slack"] + figma_costs["annual_total_figma"])) * 100,
            "figma_percent_of_total": (figma_costs["annual_total_figma"] / (slack_costs["annual_total_slack"] + figma_costs["annual_total_figma"])) * 100
        }

def main():
    parser = argparse.ArgumentParser(description='Calculate Slack vs Figma TCO for engineering teams')
    parser.add_argument('--team-size', type=int, required=True, help='Total team size')
    parser.add_argument('--slack-users', type=int, help='Users with Slack access (default: team-size)')
    parser.add_argument('--figma-editors', type=int, help='Figma editors (default: 30% of team-size)')
    parser.add_argument('--external-slack-users', type=int, default=0, help='External Slack Connect users')
    parser.add_argument('--enterprise', action='store_true', help='Use enterprise pricing for both tools')

    args = parser.parse_args()

    config = TeamConfig(
        name="Engineering Team",
        total_users=args.team_size,
        slack_users=args.slack_users or args.team_size,
        figma_editors=args.figma_editors or int(args.team_size * 0.3),
        external_slack_users=args.external_slack_users,
        figma_viewer_ratio=0.4,  # 40% of team are viewers
        use_slack_enterprise=args.enterprise,
        use_figma_enterprise=args.enterprise
    )

    calculator = CollaborationTCOCalculator(config)
    tco = calculator.calculate_total_tco()

    print(f"=== TCO for {config.total_users} User Team ===")
    print(f"Slack Annual Cost: ${tco['annual_total_slack']:,.2f} ({tco['slack_percent_of_total']:.1f}% of total)")
    print(f"Figma Annual Cost: ${tco['annual_total_figma']:,.2f} ({tco['figma_percent_of_total']:.1f}% of total)")
    print(f"Total Annual TCO: ${tco['total_annual_tco']:,.2f}")
    print(f"\nSlack Details:")
    print(f"  Per User Monthly: ${tco['per_user_monthly']:.2f}")
    print(f"  External Connect Cost: ${tco['annual_connect_cost']:,.2f}")
    print(f"\nFigma Details:")
    print(f"  Per Editor Monthly: ${tco['per_editor_monthly']:.2f}")
    print(f"  Free Viewers: {tco['free_viewer_count']}")

if __name__ == "__main__":
    # Example usage: python tco_calculator.py --team-size 1000 --enterprise
    main()

import boto3
from datetime import datetime

cloudwatch = boto3.client('cloudwatch')
def log_slack_latency(latency_ms):
    cloudwatch.put_metric_data(
        Namespace='Slack/Benchmarks',
        MetricData=[{
            'MetricName': 'MessageLatencyMs',
            'Value': latency_ms,
            'Unit': 'Milliseconds',
            'Timestamp': datetime.utcnow()
        }]
    )

import csv

def export_tco_to_csv(tco: Dict[str, float], filename: str = 'tco_report.csv'):
    with open(filename, 'w', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=tco.keys())
        writer.writeheader()
        writer.writerow(tco)

# Slack CLI workflow step
const updateFigmaFile = async (fileId, token) => {
    const figmaApi = new FigmaApi({ personalAccessToken: token });
    await figmaApi.postFileNodes(fileId, {
        nodes: [{ type: 'RECTANGLE', name: 'Approved Design' }]
    });
    return { status: 'success' };
};