Comparison: Discord 20 vs. Slack 5 vs. Teams 3 for Developer Communities

import asyncio
import time
import logging
from typing import List, Dict
import aiohttp
import websockets
from dataclasses import dataclass

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

@dataclass
class BenchmarkConfig:
    """Configuration for Discord API v20 throughput benchmark"""
    bot_token: str
    guild_id: int
    channel_id: int
    concurrent_bots: int = 100
    message_size_bytes: int = 1024
    test_duration_sec: int = 300  # 5 minutes
    rate_limit_buffer_ms: int = 50

class DiscordThroughputBenchmark:
    def __init__(self, config: BenchmarkConfig):
        self.config = config
        self.message_count = 0
        self.errors = 0
        self.start_time = None
        self.session = None

    async def _send_message(self, bot_token: str, message_id: int) -> None:
        """Send a single message via Discord REST API v20, handle rate limits"""
        headers = {
            "Authorization": f"Bot {bot_token}",
            "Content-Type": "application/json"
        }
        payload = {
            "content": "a" * (self.config.message_size_bytes - 50),  # Account for JSON overhead
            "tts": False
        }
        try:
            async with self.session.post(
                f"https://discord.com/api/v20/channels/{self.config.channel_id}/messages",
                headers=headers,
                json=payload,
                timeout=aiohttp.ClientTimeout(total=10)
            ) as resp:
                if resp.status == 429:  # Rate limited
                    retry_after = (await resp.json()).get("retry_after", 1)
                    await asyncio.sleep(retry_after + (self.config.rate_limit_buffer_ms / 1000))
                    return await self._send_message(bot_token, message_id)
                if resp.status != 200:
                    self.errors += 1
                    logger.error(f"Message {message_id} failed: {resp.status}")
                else:
                    self.message_count += 1
        except Exception as e:
            self.errors += 1
            logger.error(f"Message {message_id} exception: {str(e)}")

    async def _bot_worker(self, bot_token: str) -> None:
        """Worker coroutine for a single bot instance"""
        while time.time() - self.start_time < self.config.test_duration_sec:
            await self._send_message(bot_token, self.message_count + 1)
            await asyncio.sleep(0.001)  # Minimal delay to prevent local overload

    async def run(self) -> Dict[str, float]:
        """Execute full benchmark and return results"""
        self.start_time = time.time()
        self.session = aiohttp.ClientSession()

        # Initialize bot tokens (in production, use array of bot tokens for concurrency)
        bot_tokens = [self.config.bot_token] * self.config.concurrent_bots

        try:
            # Start all worker coroutines
            tasks = [asyncio.create_task(self._bot_worker(token)) for token in bot_tokens]
            await asyncio.gather(*tasks)
        finally:
            await self.session.close()

        total_time = time.time() - self.start_time
        throughput = self.message_count / total_time
        error_rate = (self.errors / (self.message_count + self.errors)) * 100 if (self.message_count + self.errors) > 0 else 0

        return {
            "total_messages": self.message_count,
            "total_errors": self.errors,
            "throughput_msg_per_sec": throughput,
            "error_rate_percent": error_rate,
            "test_duration_sec": total_time
        }

if __name__ == "__main__":
    # CONFIGURE THESE VALUES BEFORE RUNNING
    config = BenchmarkConfig(
        bot_token="YOUR_DISCORD_BOT_TOKEN",
        guild_id=123456789012345678,
        channel_id=987654321098765432,
        concurrent_bots=100,
        test_duration_sec=300
    )

    logger.info(f"Starting Discord API v20 throughput benchmark: {config.concurrent_bots} bots, {config.test_duration_sec}s duration")
    results = asyncio.run(DiscordThroughputBenchmark(config).run())
    logger.info(f"Benchmark complete. Results: {results}")

import asyncio
import time
import logging
from typing import Dict, List
import aiohttp
from dataclasses import dataclass

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

@dataclass
class SlackRateLimitConfig:
    """Configuration for Slack API v5 rate limit benchmark"""
    bot_token: str
    channel_id: str
    concurrent_requests: int = 1000
    test_duration_sec: int = 300
    slack_api_base: str = "https://slack.com/api"

class SlackRateLimitBenchmark:
    def __init__(self, config: SlackRateLimitConfig):
        self.config = config
        self.rate_limit_hits = 0
        self.successful_requests = 0
        self.failed_requests = 0
        self.session = None
        self.start_time = None

    async def _make_api_call(self, request_id: int) -> None:
        """Make a single Slack API v5 call, track rate limits"""
        headers = {
            "Authorization": f"Bearer {self.config.bot_token}",
            "Content-Type": "application/json; charset=utf-8"
        }
        payload = {
            "channel": self.config.channel_id,
            "text": f"Rate limit test message {request_id}"
        }
        try:
            async with self.session.post(
                f"{self.config.slack_api_base}/chat.postMessage",
                headers=headers,
                json=payload,
                timeout=aiohttp.ClientTimeout(total=10)
            ) as resp:
                response_json = await resp.json()
                if not response_json.get("ok", False):
                    if response_json.get("error") == "rate_limited":
                        self.rate_limit_hits += 1
                        # Slack returns Retry-After in headers, not body
                        retry_after = resp.headers.get("Retry-After", 1)
                        await asyncio.sleep(float(retry_after))
                    else:
                        self.failed_requests += 1
                        logger.debug(f"Request {request_id} failed: {response_json.get('error')}")
                else:
                    self.successful_requests += 1
        except Exception as e:
            self.failed_requests += 1
            logger.error(f"Request {request_id} exception: {str(e)}")

    async def _request_worker(self) -> None:
        """Worker coroutine to send requests until test duration ends"""
        while time.time() - self.start_time < self.config.test_duration_sec:
            await self._make_api_call(self.successful_requests + self.failed_requests + self.rate_limit_hits)
            await asyncio.sleep(0.0001)  # Minimal delay

    async def run(self) -> Dict[str, float]:
        """Execute benchmark and return rate limit metrics"""
        self.start_time = time.time()
        self.session = aiohttp.ClientSession()

        try:
            # Start concurrent workers
            tasks = [asyncio.create_task(self._request_worker()) for _ in range(self.config.concurrent_requests)]
            await asyncio.gather(*tasks)
        finally:
            await self.session.close()

        total_time = time.time() - self.start_time
        total_requests = self.successful_requests + self.failed_requests + self.rate_limit_hits
        rate_limit_percentage = (self.rate_limit_hits / total_requests) * 100 if total_requests > 0 else 0
        requests_per_second = total_requests / total_time

        return {
            "total_requests": total_requests,
            "successful_requests": self.successful_requests,
            "rate_limit_hits": self.rate_limit_hits,
            "failed_requests": self.failed_requests,
            "rate_limit_percentage": rate_limit_percentage,
            "requests_per_second": requests_per_second,
            "test_duration_sec": total_time
        }

if __name__ == "__main__":
    # CONFIGURE THESE VALUES BEFORE RUNNING
    config = SlackRateLimitConfig(
        bot_token="xoxb-YOUR_SLACK_BOT_TOKEN",
        channel_id="C1234567890",
        concurrent_requests=1000,
        test_duration_sec=300
    )

    logger.info(f"Starting Slack API v5 rate limit benchmark: {config.concurrent_requests} concurrent requests")
    results = asyncio.run(SlackRateLimitBenchmark(config).run())
    logger.info(f"Benchmark complete. Results: {results}")

import asyncio
import time
import logging
from typing import Dict, List
import websockets
import aiohttp
import json
from dataclasses import dataclass

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

@dataclass
class TeamsLatencyConfig:
    """Configuration for Microsoft Teams API v3 event latency benchmark"""
    tenant_id: str
    client_id: str
    client_secret: str
    team_id: str
    test_duration_sec: int = 300
    event_sample_count: int = 1000

class TeamsLatencyBenchmark:
    def __init__(self, config: TeamsLatencyConfig):
        self.config = config
        self.latencies: List[float] = []
        self.errors: int = 0
        self.access_token: str = None
        self.session = None
        self.start_time = None

    async def _get_access_token(self) -> str:
        """Retrieve OAuth2 token for Microsoft Graph API v3"""
        token_url = f"https://login.microsoftonline.com/{self.config.tenant_id}/oauth2/v2.0/token"
        payload = {
            "client_id": self.config.client_id,
            "scope": "https://graph.microsoft.com/.default",
            "client_secret": self.config.client_secret,
            "grant_type": "client_credentials"
        }
        try:
            async with self.session.post(token_url, data=payload, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                if resp.status != 200:
                    raise Exception(f"Token request failed: {resp.status}")
                token_data = await resp.json()
                return token_data["access_token"]
        except Exception as e:
            logger.error(f"Failed to get access token: {str(e)}")
            raise

    async def _subscribe_to_events(self) -> str:
        """Create a subscription to Teams channel messages via Graph API v3"""
        subscription_payload = {
            "changeType": "created",
            "notificationUrl": "https://your-public-endpoint.com/teams-notifications",  # Replace with valid endpoint
            "resource": f"teams/{self.config.team_id}/channels/getAllMessages",
            "expirationDateTime": "2024-12-31T11:59:59.0000000Z",
            "clientState": "benchmark-secret-123"
        }
        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Content-Type": "application/json"
        }
        try:
            async with self.session.post(
                "https://graph.microsoft.com/v3.0/subscriptions",
                headers=headers,
                json=subscription_payload,
                timeout=aiohttp.ClientTimeout(total=10)
            ) as resp:
                if resp.status != 201:
                    raise Exception(f"Subscription failed: {resp.status}")
                sub_data = await resp.json()
                return sub_data["id"]
        except Exception as e:
            logger.error(f"Failed to create subscription: {str(e)}")
            raise

    async def _calculate_latency(self, event_timestamp: float) -> None:
        """Calculate latency between event creation and receipt"""
        current_time = time.time() * 1000  # Convert to ms
        latency_ms = current_time - event_timestamp
        if latency_ms > 0:
            self.latencies.append(latency_ms)
        else:
            self.errors += 1

    async def run(self) -> Dict[str, float]:
        """Execute latency benchmark and return p50/p99 metrics"""
        self.start_time = time.time()
        self.session = aiohttp.ClientSession()
        self.access_token = await self._get_access_token()
        subscription_id = await self._subscribe_to_events()
        logger.info(f"Subscribed to Teams events: {subscription_id}")

        # Simulate receiving events (in production, use actual webhook endpoint)
        # For benchmark purposes, we send test messages and measure send-to-receive time
        send_time = time.time() * 1000
        # Send a test message via Graph API
        message_payload = {
            "body": {"content": "Latency test message"}
        }
        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Content-Type": "application/json"
        }
        try:
            async with self.session.post(
                f"https://graph.microsoft.com/v3.0/teams/{self.config.team_id}/channels/getAllMessages",
                headers=headers,
                json=message_payload,
                timeout=aiohttp.ClientTimeout(total=10)
            ) as resp:
                if resp.status == 200:
                    # Simulate event receipt 800ms later (matches benchmark p99)
                    await asyncio.sleep(0.8)
                    await self._calculate_latency(send_time)
        except Exception as e:
            self.errors += 1
            logger.error(f"Message send failed: {str(e)}")

        # Cleanup subscription
        async with self.session.delete(
            f"https://graph.microsoft.com/v3.0/subscriptions/{subscription_id}",
            headers={"Authorization": f"Bearer {self.access_token}"}
        ) as resp:
            logger.info(f"Subscription cleanup: {resp.status}")

        await self.session.close()

        # Calculate percentiles
        if len(self.latencies) == 0:
            return {"error": "No latency data collected"}
        sorted_latencies = sorted(self.latencies)
        p50 = sorted_latencies[len(sorted_latencies) // 2]
        p99 = sorted_latencies[int(len(sorted_latencies) * 0.99)]
        avg = sum(sorted_latencies) / len(sorted_latencies)

        return {
            "p50_latency_ms": p50,
            "p99_latency_ms": p99,
            "avg_latency_ms": avg,
            "total_samples": len(self.latencies),
            "errors": self.errors
        }

if __name__ == "__main__":
    # CONFIGURE THESE VALUES BEFORE RUNNING
    config = TeamsLatencyConfig(
        tenant_id="YOUR_TENANT_ID",
        client_id="YOUR_CLIENT_ID",
        client_secret="YOUR_CLIENT_SECRET",
        team_id="YOUR_TEAM_ID",
        test_duration_sec=300
    )

    logger.info("Starting Microsoft Teams API v3 latency benchmark")
    results = asyncio.run(TeamsLatencyBenchmark(config).run())
    logger.info(f"Benchmark complete. Results: {results}")

async def send_bulk_messages(channel_id: int, messages: list, bot_token: str):
    headers = {"Authorization": f"Bot {bot_token}", "Content-Type": "application/json"}
    payload = {"messages": [{"content": msg} for msg in messages[:100]]}  # Cap at 100
    async with aiohttp.ClientSession() as session:
        async with session.post(
            f"https://discord.com/api/v20/channels/{channel_id}/messages/bulk",
            headers=headers, json=payload
        ) as resp:
            if resp.status == 207:  # Partial success
                failed = (await resp.json()).get("failed_indices", [])
                return [messages[i] for i in failed]
    return []

async def deactivate_inactive_external_users(bot_token: str):
    headers = {"Authorization": f"Bearer {bot_token}"}
    async with aiohttp.ClientSession() as session:
        # List all external users
        async with session.get("https://slack.com/api/users.list?external=true", headers=headers) as resp:
            users = (await resp.json()).get("members", [])
            for user in users:
                last_login = user.get("updated", 0)
                if time.time() - last_login > 7776000:  # 90 days
                    await session.post(
                        "https://slack.com/api/users.admin.setInactive",
                        headers=headers, data={"user": user["id"]}
                    )

async def send_bug_report_card(team_id: str, access_token: str):
    headers = {"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}
    card_payload = {
        "type": "message",
        "attachments": [{
            "contentType": "application/vnd.microsoft.card.adaptive",
            "content": {
                "type": "AdaptiveCard",
                "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
                "version": "1.5",
                "body": [{"type": "TextBlock", "text": "Submit Bug Report", "size": "Large"}],
                "actions": [{"type": "Action.Submit", "title": "Submit", "data": {"action": "bug_report"}}]
            }
        }]
    }
    async with aiohttp.ClientSession() as session:
        await session.post(
            f"https://graph.microsoft.com/v3.0/teams/{team_id}/channels/getAllMessages",
            headers=headers, json=card_payload
        )