Skip to main content
EVOKORE// BROWSE
>

./browse/prompts

20 NODES
🤖system prompt•7 months ago

team-communication-protocols

Structured messaging protocols for agent team communication

coding
⭐1
# Team Communication Protocols Protocols for effective communication between agent teammates, including message type selection, plan approval workflows, shutdown procedures, and common anti-patterns to avoid. ## When to Use This Skill - Establishing communication norms for a new team - Choosing between message types (message, broadcast, shutdown_request) - Handling plan approval workflows - Managing graceful team shutdown - Discovering teammate identities and capabilities ## Message Type Selection ### `message` (Direct Message) — Default Choice Send to a single specific teammate: ```json { "type": "message", "recipient": "implementer-1", "content": "Your API endpoint is ready. You can now build the frontend form.", "summary": "API endpoint ready for frontend" } ``` **Use for**: Task updates, coordination, questions, integration notifications. ### `broadcast` — Use Sparingly Send to ALL teammates simultaneously: ```json { "type": "broadcast", "content": "Critical: shared types file has been updated. Pull latest before continuing.", "summary": "Shared types updated" } ``` **Use ONLY for**: Critical blockers affecting everyone, major changes to shared resources. **Why sparingly?**: Each broadcast sends N separate messages (one per teammate), consuming API resources proportional to team size. ### `shutdown_request` — Graceful Termination Request a teammate to shut down: ```json { "type": "shutdown_request", "recipient": "reviewer-1", "content": "Review complete, shutting down team." } ``` The teammate responds with `shutdown_response` (approve or reject with reason). ## Communication Anti-Patterns | Anti-Pattern | Problem | Better Approach | | --------------------------------------- | ---------------------------------------- | -------------------------------------- | | Broadcasting routine updates | Wastes resources, noise | Direct message to affected teammate | | Sending JSON status messages | Not designed for structured data | Use TaskUpdate to update task status | | Not communicating at integration points | Teammates build against stale interfaces | Message when your interface is ready | | Micromanaging via messages | Overwhelms teammates, slows work | Check in at milestones, not every step | | Using UUIDs instead of names | Hard to read, error-prone | Always use teammate names | | Ignoring idle teammates | Wasted capacity | Assign new work or shut down | ## Plan Approval Workflow When a teammate is spawned with `plan_mode_required`: 1. Teammate creates a plan using read-only exploration tools 2. Teammate calls `ExitPlanMode` which sends a `plan_approval_request` to the lead 3. Lead reviews the plan 4. Lead responds with `plan_approval_response`: **Approve**: ```json { "type": "plan_approval_response", "request_id": "abc-123", "recipient": "implementer-1", "approve": true } ``` **Reject with feedback**: ```json { "type": "plan_approval_response", "request_id": "abc-123", "recipient": "implementer-1", "approve": false, "content": "Please add error handling for the API calls" } ``` ## Shutdown Protocol ### Graceful Shutdown Sequence 1. **Lead sends shutdown_request** to each teammate 2. **Teammate receives request** as a JSON message with `type: "shutdown_request"` 3. **Teammate responds** with `shutdown_response`: - `approve: true` — Teammate saves state and exits - `approve: false` + reason — Teammate continues working 4. **Lead handles rejections** — Wait for teammate to finish, then retry 5. **After all teammates shut down** — Call `Teammate` cleanup ### Handling Rejections If a teammate rejects shutdown: - Check their reason (usually "still working on task") - Wait for their current task to complete - Retry shutdown request - If urgent, user can force shutdown ## Teammate Discovery Find team members by reading the config file: **Location**: `~/.claude/teams/{team-name}/config.json` **Structure**: ```json { "members": [ { "name": "security-reviewer", "agentId": "uuid-here", "agentType": "team-reviewer" }, { "name": "perf-reviewer", "agentId": "uuid-here", "agentType": "team-reviewer" } ] } ``` **Always use `name`** for messaging and task assignment. Never use `agentId` directly.
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

microservices-patterns

Design microservices architectures with service boundaries,

coding
⭐1
# Microservices Patterns Master microservices architecture patterns including service boundaries, inter-service communication, data management, and resilience patterns for building distributed systems. ## When to Use This Skill - Decomposing monoliths into microservices - Designing service boundaries and contracts - Implementing inter-service communication - Managing distributed data and transactions - Building resilient distributed systems - Implementing service discovery and load balancing - Designing event-driven architectures ## Core Concepts ### 1. Service Decomposition Strategies **By Business Capability** - Organize services around business functions - Each service owns its domain - Example: OrderService, PaymentService, InventoryService **By Subdomain (DDD)** - Core domain, supporting subdomains - Bounded contexts map to services - Clear ownership and responsibility **Strangler Fig Pattern** - Gradually extract from monolith - New functionality as microservices - Proxy routes to old/new systems ### 2. Communication Patterns **Synchronous (Request/Response)** - REST APIs - gRPC - GraphQL **Asynchronous (Events/Messages)** - Event streaming (Kafka) - Message queues (RabbitMQ, SQS) - Pub/Sub patterns ### 3. Data Management **Database Per Service** - Each service owns its data - No shared databases - Loose coupling **Saga Pattern** - Distributed transactions - Compensating actions - Eventual consistency ### 4. Resilience Patterns **Circuit Breaker** - Fail fast on repeated errors - Prevent cascade failures **Retry with Backoff** - Transient fault handling - Exponential backoff **Bulkhead** - Isolate resources - Limit impact of failures ## Service Decomposition Patterns ### Pattern 1: By Business Capability ```python # E-commerce example # Order Service class OrderService: """Handles order lifecycle.""" async def create_order(self, order_data: dict) -> Order: order = Order.create(order_data) # Publish event for other services await self.event_bus.publish( OrderCreatedEvent( order_id=order.id, customer_id=order.customer_id, items=order.items, total=order.total ) ) return order # Payment Service (separate service) class PaymentService: """Handles payment processing.""" async def process_payment(self, payment_request: PaymentRequest) -> PaymentResult: # Process payment result = await self.payment_gateway.charge( amount=payment_request.amount, customer=payment_request.customer_id ) if result.success: await self.event_bus.publish( PaymentCompletedEvent( order_id=payment_request.order_id, transaction_id=result.transaction_id ) ) return result # Inventory Service (separate service) class InventoryService: """Handles inventory management.""" async def reserve_items(self, order_id: str, items: List[OrderItem]) -> ReservationResult: # Check availability for item in items: available = await self.inventory_repo.get_available(item.product_id) if available < item.quantity: return ReservationResult( success=False, error=f"Insufficient inventory for {item.product_id}" ) # Reserve items reservation = await self.create_reservation(order_id, items) await self.event_bus.publish( InventoryReservedEvent( order_id=order_id, reservation_id=reservation.id ) ) return ReservationResult(success=True, reservation=reservation) ``` ### Pattern 2: API Gateway ```python from fastapi import FastAPI, HTTPException, Depends import httpx from circuitbreaker import circuit app = FastAPI() class APIGateway: """Central entry point for all client requests.""" def __init__(self): self.order_service_url = "http://order-service:8000" self.payment_service_url = "http://payment-service:8001" self.inventory_service_url = "http://inventory-service:8002" self.http_client = httpx.AsyncClient(timeout=5.0) @circuit(failure_threshold=5, recovery_timeout=30) async def call_order_service(self, path: str, method: str = "GET", **kwargs): """Call order service with circuit breaker.""" response = await self.http_client.request( method, f"{self.order_service_url}{path}", **kwargs ) response.raise_for_status() return response.json() async def create_order_aggregate(self, order_id: str) -> dict: """Aggregate data from multiple services.""" # Parallel requests order, payment, inventory = await asyncio.gather( self.call_order_service(f"/orders/{order_id}"), self.call_payment_service(f"/payments/order/{order_id}"), self.call_inventory_service(f"/reservations/order/{order_id}"), return_exceptions=True ) # Handle partial failures result = {"order": order} if not isinstance(payment, Exception): result["payment"] = payment if not isinstance(inventory, Exception): result["inventory"] = inventory return result @app.post("/api/orders") async def create_order( order_data: dict, gateway: APIGateway = Depends() ): """API Gateway endpoint.""" try: # Route to order service order = await gateway.call_order_service( "/orders", method="POST", json=order_data ) return {"order": order} except httpx.HTTPError as e: raise HTTPException(status_code=503, detail="Order service unavailable") ``` ## Communication Patterns ### Pattern 1: Synchronous REST Communication ```python # Service A calls Service B import httpx from tenacity import retry, stop_after_attempt, wait_exponential class ServiceClient: """HTTP client with retries and timeout.""" def __init__(self, base_url: str): self.base_url = base_url self.client = httpx.AsyncClient( timeout=httpx.Timeout(5.0, connect=2.0), limits=httpx.Limits(max_keepalive_connections=20) ) @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10) ) async def get(self, path: str, **kwargs): """GET with automatic retries.""" response = await self.client.get(f"{self.base_url}{path}", **kwargs) response.raise_for_status() return response.json() async def post(self, path: str, **kwargs): """POST request.""" response = await self.client.post(f"{self.base_url}{path}", **kwargs) response.raise_for_status() return response.json() # Usage payment_client = ServiceClient("http://payment-service:8001") result = await payment_client.post("/payments", json=payment_data) ``` ### Pattern 2: Asynchronous Event-Driven ```python # Event-driven communication with Kafka from aiokafka import AIOKafkaProducer, AIOKafkaConsumer import json from dataclasses import dataclass, asdict from datetime import datetime @dataclass class DomainEvent: event_id: str event_type: str aggregate_id: str occurred_at: datetime data: dict class EventBus: """Event publishing and subscription.""" def __init__(self, bootstrap_servers: List[str]): self.bootstrap_servers = bootstrap_servers self.producer = None async def start(self): self.producer = AIOKafkaProducer( bootstrap_servers=self.bootstrap_servers, value_serializer=lambda v: json.dumps(v).encode() ) await self.producer.start() async def publish(self, event: DomainEvent): """Publish event to Kafka topic.""" topic = event.event_type await self.producer.send_and_wait( topic, value=asdict(event), key=event.aggregate_id.encode() ) async def subscribe(self, topic: str, handler: callable): """Subscribe to events.""" consumer = AIOKafkaConsumer( topic, bootstrap_servers=self.bootstrap_servers, value_deserializer=lambda v: json.loads(v.decode()), group_id="my-service" ) await consumer.start() try: async for message in consumer: event_data = message.value await handler(event_data) finally: await consumer.stop() # Order Service publishes event async def create_order(order_data: dict): order = await save_order(order_data) event = DomainEvent( event_id=str(uuid.uuid4()), event_type="OrderCreated", aggregate_id=order.id, occurred_at=datetime.now(), data={ "order_id": order.id, "customer_id": order.customer_id, "total": order.total } ) await event_bus.publish(event) # Inventory Service listens for OrderCreated async def handle_order_created(event_data: dict): """React to order creation.""" order_id = event_data["data"]["order_id"] items = event_data["data"]["items"] # Reserve inventory await reserve_inventory(order_id, items) ``` ### Pattern 3: Saga Pattern (Distributed Transactions) ```python # Saga orchestration for order fulfillment from enum import Enum from typing import List, Callable class SagaStep: """Single step in saga.""" def __init__( self, name: str, action: Callable, compensation: Callable ): self.name = name self.action = action self.compensation = compensation class SagaStatus(Enum): PENDING = "pending" COMPLETED = "completed" COMPENSATING = "compensating" FAILED = "failed" class OrderFulfillmentSaga: """Orchestrated saga for order fulfillment.""" def __init__(self): self.steps: List[SagaStep] = [ SagaStep( "create_order", action=self.create_order, compensation=self.cancel_order ), SagaStep( "reserve_inventory", action=self.reserve_inventory, compensation=self.release_inventory ), SagaStep( "process_payment", action=self.process_payment, compensation=self.refund_payment ), SagaStep( "confirm_order", action=self.confirm_order, compensation=self.cancel_order_confirmation ) ] async def execute(self, order_data: dict) -> SagaResult: """Execute saga steps.""" completed_steps = [] context = {"order_data": order_data} try: for step in self.steps: # Execute step result = await step.action(context) if not result.success: # Compensate await self.compensate(completed_steps, context) return SagaResult( status=SagaStatus.FAILED, error=result.error ) completed_steps.append(step) context.update(result.data) return SagaResult(status=SagaStatus.COMPLETED, data=context) except Exception as e: # Compensate on error await self.compensate(completed_steps, context) return SagaResult(status=SagaStatus.FAILED, error=str(e)) async def compensate(self, completed_steps: List[SagaStep], context: dict): """Execute compensating actions in reverse order.""" for step in reversed(completed_steps): try: await step.compensation(context) except Exception as e: # Log compensation failure print(f"Compensation failed for {step.name}: {e}") # Step implementations async def create_order(self, context: dict) -> StepResult: order = await order_service.create(context["order_data"]) return StepResult(success=True, data={"order_id": order.id}) async def cancel_order(self, context: dict): await order_service.cancel(context["order_id"]) async def reserve_inventory(self, context: dict) -> StepResult: result = await inventory_service.reserve( context["order_id"], context["order_data"]["items"] ) return StepResult( success=result.success, data={"reservation_id": result.reservation_id} ) async def release_inventory(self, context: dict): await inventory_service.release(context["reservation_id"]) async def process_payment(self, context: dict) -> StepResult: result = await payment_service.charge( context["order_id"], context["order_data"]["total"] ) return StepResult( success=result.success, data={"transaction_id": result.transaction_id}, error=result.error ) async def refund_payment(self, context: dict): await payment_service.refund(context["transaction_id"]) ``` ## Resilience Patterns ### Circuit Breaker Pattern ```python from enum import Enum from datetime import datetime, timedelta from typing import Callable, Any class CircuitState(Enum): CLOSED = "closed" # Normal operation OPEN = "open" # Failing, reject requests HALF_OPEN = "half_open" # Testing if recovered class CircuitBreaker: """Circuit breaker for service calls.""" def __init__( self, failure_threshold: int = 5, recovery_timeout: int = 30, success_threshold: int = 2 ): self.failure_threshold = failure_threshold self.recovery_timeout = recovery_timeout self.success_threshold = success_threshold self.failure_count = 0 self.success_count = 0 self.state = CircuitState.CLOSED self.opened_at = None async def call(self, func: Callable, *args, **kwargs) -> Any: """Execute function with circuit breaker.""" if self.state == CircuitState.OPEN: if self._should_attempt_reset(): self.state = CircuitState.HALF_OPEN else: raise CircuitBreakerOpenError("Circuit breaker is open") try: result = await func(*args, **kwargs) self._on_success() return result except Exception as e: self._on_failure() raise def _on_success(self): """Handle successful call.""" self.failure_count = 0 if self.state == CircuitState.HALF_OPEN: self.success_count += 1 if self.success_count >= self.success_threshold: self.state = CircuitState.CLOSED self.success_count = 0 def _on_failure(self): """Handle failed call.""" self.failure_count += 1 if self.failure_count >= self.failure_threshold: self.state = CircuitState.OPEN self.opened_at = datetime.now() if self.state == CircuitState.HALF_OPEN: self.state = CircuitState.OPEN self.opened_at = datetime.now() def _should_attempt_reset(self) -> bool: """Check if enough time passed to try again.""" return ( datetime.now() - self.opened_at > timedelta(seconds=self.recovery_timeout) ) # Usage breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=30) async def call_payment_service(payment_data: dict): return await breaker.call( payment_client.process_payment, payment_data ) ``` ## Resources - **references/service-decomposition-guide.md**: Breaking down monoliths - **references/communication-patterns.md**: Sync vs async patterns - **references/saga-implementation.md**: Distributed transactions - **assets/circuit-breaker.py**: Production circuit breaker - **assets/event-bus-template.py**: Kafka event bus implementation - **assets/api-gateway-template.py**: Complete API gateway ## Best Practices 1. **Service Boundaries**: Align with business capabilities 2. **Database Per Service**: No shared databases 3. **API Contracts**: Versioned, backward compatible 4. **Async When Possible**: Events over direct calls 5. **Circuit Breakers**: Fail fast on service failures 6. **Distributed Tracing**: Track requests across services 7. **Service Registry**: Dynamic service discovery 8. **Health Checks**: Liveness and readiness probes ## Common Pitfalls - **Distributed Monolith**: Tightly coupled services - **Chatty Services**: Too many inter-service calls - **Shared Databases**: Tight coupling through data - **No Circuit Breakers**: Cascade failures - **Synchronous Everything**: Tight coupling, poor resilience - **Premature Microservices**: Starting with microservices - **Ignoring Network Failures**: Assuming reliable network - **No Compensation Logic**: Can't undo failed transactions
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

data-storytelling

Transform data into compelling narratives using visualization,

data
⭐1
# Data Storytelling Transform raw data into compelling narratives that drive decisions and inspire action. ## When to Use This Skill - Presenting analytics to executives - Creating quarterly business reviews - Building investor presentations - Writing data-driven reports - Communicating insights to non-technical audiences - Making recommendations based on data ## Core Concepts ### 1. Story Structure ``` Setup → Conflict → Resolution Setup: Context and baseline Conflict: The problem or opportunity Resolution: Insights and recommendations ``` ### 2. Narrative Arc ``` 1. Hook: Grab attention with surprising insight 2. Context: Establish the baseline 3. Rising Action: Build through data points 4. Climax: The key insight 5. Resolution: Recommendations 6. Call to Action: Next steps ``` ### 3. Three Pillars | Pillar | Purpose | Components | | ------------- | -------- | -------------------------------- | | **Data** | Evidence | Numbers, trends, comparisons | | **Narrative** | Meaning | Context, causation, implications | | **Visuals** | Clarity | Charts, diagrams, highlights | ## Story Frameworks ### Framework 1: The Problem-Solution Story ```markdown # Customer Churn Analysis ## The Hook "We're losing $2.4M annually to preventable churn." ## The Context - Current churn rate: 8.5% (industry average: 5%) - Average customer lifetime value: $4,800 - 500 customers churned last quarter ## The Problem Analysis of churned customers reveals a pattern: - 73% churned within first 90 days - Common factor: < 3 support interactions - Low feature adoption in first month ## The Insight [Show engagement curve visualization] Customers who don't engage in the first 14 days are 4x more likely to churn. ## The Solution 1. Implement 14-day onboarding sequence 2. Proactive outreach at day 7 3. Feature adoption tracking ## Expected Impact - Reduce early churn by 40% - Save $960K annually - Payback period: 3 months ## Call to Action Approve $50K budget for onboarding automation. ``` ### Framework 2: The Trend Story ```markdown # Q4 Performance Analysis ## Where We Started Q3 ended with $1.2M MRR, 15% below target. Team morale was low after missed goals. ## What Changed [Timeline visualization] - Oct: Launched self-serve pricing - Nov: Reduced friction in signup - Dec: Added customer success calls ## The Transformation [Before/after comparison chart] | Metric | Q3 | Q4 | Change | |----------------|--------|--------|--------| | Trial → Paid | 8% | 15% | +87% | | Time to Value | 14 days| 5 days | -64% | | Expansion Rate | 2% | 8% | +300% | ## Key Insight Self-serve + high-touch creates compound growth. Customers who self-serve AND get a success call have 3x higher expansion rate. ## Going Forward Double down on hybrid model. Target: $1.8M MRR by Q2. ``` ### Framework 3: The Comparison Story ```markdown # Market Opportunity Analysis ## The Question Should we expand into EMEA or APAC first? ## The Comparison [Side-by-side market analysis] ### EMEA - Market size: $4.2B - Growth rate: 8% - Competition: High - Regulatory: Complex (GDPR) - Language: Multiple ### APAC - Market size: $3.8B - Growth rate: 15% - Competition: Moderate - Regulatory: Varied - Language: Multiple ## The Analysis [Weighted scoring matrix visualization] | Factor | Weight | EMEA Score | APAC Score | | ----------- | ------ | ---------- | ---------- | | Market Size | 25% | 5 | 4 | | Growth | 30% | 3 | 5 | | Competition | 20% | 2 | 4 | | Ease | 25% | 2 | 3 | | **Total** | | **2.9** | **4.1** | ## The Recommendation APAC first. Higher growth, less competition. Start with Singapore hub (English, business-friendly). Enter EMEA in Year 2 with localization ready. ## Risk Mitigation - Timezone coverage: Hire 24/7 support - Cultural fit: Local partnerships - Payment: Multi-currency from day 1 ``` ## Visualization Techniques ### Technique 1: Progressive Reveal ```markdown Start simple, add layers: Slide 1: "Revenue is growing" [single line chart] Slide 2: "But growth is slowing" [add growth rate overlay] Slide 3: "Driven by one segment" [add segment breakdown] Slide 4: "Which is saturating" [add market share] Slide 5: "We need new segments" [add opportunity zones] ``` ### Technique 2: Contrast and Compare ```markdown Before/After: ┌─────────────────┬─────────────────┐ │ BEFORE │ AFTER │ │ │ │ │ Process: 5 days│ Process: 1 day │ │ Errors: 15% │ Errors: 2% │ │ Cost: $50/unit │ Cost: $20/unit │ └─────────────────┴─────────────────┘ This/That (emphasize difference): ┌─────────────────────────────────────┐ │ CUSTOMER A vs B │ │ ┌──────────┐ ┌──────────┐ │ │ │ ████████ │ │ ██ │ │ │ │ $45,000 │ │ $8,000 │ │ │ │ LTV │ │ LTV │ │ │ └──────────┘ └──────────┘ │ │ Onboarded No onboarding │ └─────────────────────────────────────┘ ``` ### Technique 3: Annotation and Highlight ```python import matplotlib.pyplot as plt import pandas as pd fig, ax = plt.subplots(figsize=(12, 6)) # Plot the main data ax.plot(dates, revenue, linewidth=2, color='#2E86AB') # Add annotation for key events ax.annotate( 'Product Launch\n+32% spike', xy=(launch_date, launch_revenue), xytext=(launch_date, launch_revenue * 1.2), fontsize=10, arrowprops=dict(arrowstyle='->', color='#E63946'), color='#E63946' ) # Highlight a region ax.axvspan(growth_start, growth_end, alpha=0.2, color='green', label='Growth Period') # Add threshold line ax.axhline(y=target, color='gray', linestyle='--', label=f'Target: ${target:,.0f}') ax.set_title('Revenue Growth Story', fontsize=14, fontweight='bold') ax.legend() ``` ## Presentation Templates ### Template 1: Executive Summary Slide ``` ┌─────────────────────────────────────────────────────────────┐ │ KEY INSIGHT │ │ ══════════════════════════════════════════════════════════│ │ │ │ "Customers who complete onboarding in week 1 │ │ have 3x higher lifetime value" │ │ │ ├──────────────────────┬──────────────────────────────────────┤ │ │ │ │ THE DATA │ THE IMPLICATION │ │ │ │ │ Week 1 completers: │ ✓ Prioritize onboarding UX │ │ • LTV: $4,500 │ ✓ Add day-1 success milestones │ │ • Retention: 85% │ ✓ Proactive week-1 outreach │ │ • NPS: 72 │ │ │ │ Investment: $75K │ │ Others: │ Expected ROI: 8x │ │ • LTV: $1,500 │ │ │ • Retention: 45% │ │ │ • NPS: 34 │ │ │ │ │ └──────────────────────┴──────────────────────────────────────┘ ``` ### Template 2: Data Story Flow ``` Slide 1: THE HEADLINE "We can grow 40% faster by fixing onboarding" Slide 2: THE CONTEXT Current state metrics Industry benchmarks Gap analysis Slide 3: THE DISCOVERY What the data revealed Surprising finding Pattern identification Slide 4: THE DEEP DIVE Root cause analysis Segment breakdowns Statistical significance Slide 5: THE RECOMMENDATION Proposed actions Resource requirements Timeline Slide 6: THE IMPACT Expected outcomes ROI calculation Risk assessment Slide 7: THE ASK Specific request Decision needed Next steps ``` ### Template 3: One-Page Dashboard Story ```markdown # Monthly Business Review: January 2024 ## THE HEADLINE Revenue up 15% but CAC increasing faster than LTV ## KEY METRICS AT A GLANCE ┌────────┬────────┬────────┬────────┐ │ MRR │ NRR │ CAC │ LTV │ │ $125K │ 108% │ $450 │ $2,200 │ │ ▲15% │ ▲3% │ ▲22% │ ▲8% │ └────────┴────────┴────────┴────────┘ ## WHAT'S WORKING ✓ Enterprise segment growing 25% MoM ✓ Referral program driving 30% of new logos ✓ Support satisfaction at all-time high (94%) ## WHAT NEEDS ATTENTION ✗ SMB acquisition cost up 40% ✗ Trial conversion down 5 points ✗ Time-to-value increased by 3 days ## ROOT CAUSE [Mini chart showing SMB vs Enterprise CAC trend] SMB paid ads becoming less efficient. CPC up 35% while conversion flat. ## RECOMMENDATION 1. Shift $20K/mo from paid to content 2. Launch SMB self-serve trial 3. A/B test shorter onboarding ## NEXT MONTH'S FOCUS - Launch content marketing pilot - Complete self-serve MVP - Reduce time-to-value to < 7 days ``` ## Writing Techniques ### Headlines That Work ```markdown BAD: "Q4 Sales Analysis" GOOD: "Q4 Sales Beat Target by 23% - Here's Why" BAD: "Customer Churn Report" GOOD: "We're Losing $2.4M to Preventable Churn" BAD: "Marketing Performance" GOOD: "Content Marketing Delivers 4x ROI vs. Paid" Formula: [Specific Number] + [Business Impact] + [Actionable Context] ``` ### Transition Phrases ```markdown Building the narrative: • "This leads us to ask..." • "When we dig deeper..." • "The pattern becomes clear when..." • "Contrast this with..." Introducing insights: • "The data reveals..." • "What surprised us was..." • "The inflection point came when..." • "The key finding is..." Moving to action: • "This insight suggests..." • "Based on this analysis..." • "The implication is clear..." • "Our recommendation is..." ``` ### Handling Uncertainty ```markdown Acknowledge limitations: • "With 95% confidence, we can say..." • "The sample size of 500 shows..." • "While correlation is strong, causation requires..." • "This trend holds for [segment], though [caveat]..." Present ranges: • "Impact estimate: $400K-$600K" • "Confidence interval: 15-20% improvement" • "Best case: X, Conservative: Y" ``` ## Best Practices ### Do's - **Start with the "so what"** - Lead with insight - **Use the rule of three** - Three points, three comparisons - **Show, don't tell** - Let data speak - **Make it personal** - Connect to audience goals - **End with action** - Clear next steps ### Don'ts - **Don't data dump** - Curate ruthlessly - **Don't bury the insight** - Front-load key findings - **Don't use jargon** - Match audience vocabulary - **Don't show methodology first** - Context, then method - **Don't forget the narrative** - Numbers need meaning ## Resources - [Storytelling with Data (Cole Nussbaumer)](https://www.storytellingwithdata.com/) - [The Pyramid Principle (Barbara Minto)](https://www.amazon.com/Pyramid-Principle-Logic-Writing-Thinking/dp/0273710516) - [Resonate (Nancy Duarte)](https://www.duarte.com/resonate/)
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

mtls-configuration

Configure mutual TLS (mTLS) for zero-trust service-to-service

architecture
⭐1
# mTLS Configuration Comprehensive guide to implementing mutual TLS for zero-trust service mesh communication. ## When to Use This Skill - Implementing zero-trust networking - Securing service-to-service communication - Certificate rotation and management - Debugging TLS handshake issues - Compliance requirements (PCI-DSS, HIPAA) - Multi-cluster secure communication ## Core Concepts ### 1. mTLS Flow ``` ┌─────────┐ ┌─────────┐ │ Service │ │ Service │ │ A │ │ B │ └────┬────┘ └────┬────┘ │ │ ┌────┴────┐ TLS Handshake ┌────┴────┐ │ Proxy │◄───────────────────────────►│ Proxy │ │(Sidecar)│ 1. ClientHello │(Sidecar)│ │ │ 2. ServerHello + Cert │ │ │ │ 3. Client Cert │ │ │ │ 4. Verify Both Certs │ │ │ │ 5. Encrypted Channel │ │ └─────────┘ └─────────┘ ``` ### 2. Certificate Hierarchy ``` Root CA (Self-signed, long-lived) │ ├── Intermediate CA (Cluster-level) │ │ │ ├── Workload Cert (Service A) │ └── Workload Cert (Service B) │ └── Intermediate CA (Multi-cluster) │ └── Cross-cluster certs ``` ## Templates ### Template 1: Istio mTLS (Strict Mode) ```yaml # Enable strict mTLS mesh-wide apiVersion: security.istio.io/v1beta1 kind: PeerAuthentication metadata: name: default namespace: istio-system spec: mtls: mode: STRICT --- # Namespace-level override (permissive for migration) apiVersion: security.istio.io/v1beta1 kind: PeerAuthentication metadata: name: default namespace: legacy-namespace spec: mtls: mode: PERMISSIVE --- # Workload-specific policy apiVersion: security.istio.io/v1beta1 kind: PeerAuthentication metadata: name: payment-service namespace: production spec: selector: matchLabels: app: payment-service mtls: mode: STRICT portLevelMtls: 8080: mode: STRICT 9090: mode: DISABLE # Metrics port, no mTLS ``` ### Template 2: Istio Destination Rule for mTLS ```yaml apiVersion: networking.istio.io/v1beta1 kind: DestinationRule metadata: name: default namespace: istio-system spec: host: "*.local" trafficPolicy: tls: mode: ISTIO_MUTUAL --- # TLS to external service apiVersion: networking.istio.io/v1beta1 kind: DestinationRule metadata: name: external-api spec: host: api.external.com trafficPolicy: tls: mode: SIMPLE caCertificates: /etc/certs/external-ca.pem --- # Mutual TLS to external service apiVersion: networking.istio.io/v1beta1 kind: DestinationRule metadata: name: partner-api spec: host: api.partner.com trafficPolicy: tls: mode: MUTUAL clientCertificate: /etc/certs/client.pem privateKey: /etc/certs/client-key.pem caCertificates: /etc/certs/partner-ca.pem ``` ### Template 3: Cert-Manager with Istio ```yaml # Install cert-manager issuer for Istio apiVersion: cert-manager.io/v1 kind: ClusterIssuer metadata: name: istio-ca spec: ca: secretName: istio-ca-secret --- # Create Istio CA secret apiVersion: v1 kind: Secret metadata: name: istio-ca-secret namespace: cert-manager type: kubernetes.io/tls data: tls.crt: <base64-encoded-ca-cert> tls.key: <base64-encoded-ca-key> --- # Certificate for workload apiVersion: cert-manager.io/v1 kind: Certificate metadata: name: my-service-cert namespace: my-namespace spec: secretName: my-service-tls duration: 24h renewBefore: 8h issuerRef: name: istio-ca kind: ClusterIssuer commonName: my-service.my-namespace.svc.cluster.local dnsNames: - my-service - my-service.my-namespace - my-service.my-namespace.svc - my-service.my-namespace.svc.cluster.local usages: - server auth - client auth ``` ### Template 4: SPIFFE/SPIRE Integration ```yaml # SPIRE Server configuration apiVersion: v1 kind: ConfigMap metadata: name: spire-server namespace: spire data: server.conf: | server { bind_address = "0.0.0.0" bind_port = "8081" trust_domain = "example.org" data_dir = "/run/spire/data" log_level = "INFO" ca_ttl = "168h" default_x509_svid_ttl = "1h" } plugins { DataStore "sql" { plugin_data { database_type = "sqlite3" connection_string = "/run/spire/data/datastore.sqlite3" } } NodeAttestor "k8s_psat" { plugin_data { clusters = { "demo-cluster" = { service_account_allow_list = ["spire:spire-agent"] } } } } KeyManager "memory" { plugin_data {} } UpstreamAuthority "disk" { plugin_data { key_file_path = "/run/spire/secrets/bootstrap.key" cert_file_path = "/run/spire/secrets/bootstrap.crt" } } } --- # SPIRE Agent DaemonSet (abbreviated) apiVersion: apps/v1 kind: DaemonSet metadata: name: spire-agent namespace: spire spec: selector: matchLabels: app: spire-agent template: spec: containers: - name: spire-agent image: ghcr.io/spiffe/spire-agent:1.8.0 volumeMounts: - name: spire-agent-socket mountPath: /run/spire/sockets volumes: - name: spire-agent-socket hostPath: path: /run/spire/sockets type: DirectoryOrCreate ``` ### Template 5: Linkerd mTLS (Automatic) ```yaml # Linkerd enables mTLS automatically # Verify with: # linkerd viz edges deployment -n my-namespace # For external services without mTLS apiVersion: policy.linkerd.io/v1beta1 kind: Server metadata: name: external-api namespace: my-namespace spec: podSelector: matchLabels: app: my-app port: external-api proxyProtocol: HTTP/1 # or TLS for passthrough --- # Skip TLS for specific port apiVersion: v1 kind: Service metadata: name: my-service annotations: config.linkerd.io/skip-outbound-ports: "3306" # MySQL ``` ## Certificate Rotation ```bash # Istio - Check certificate expiry istioctl proxy-config secret deploy/my-app -o json | \ jq '.dynamicActiveSecrets[0].secret.tlsCertificate.certificateChain.inlineBytes' | \ tr -d '"' | base64 -d | openssl x509 -text -noout # Force certificate rotation kubectl rollout restart deployment/my-app # Check Linkerd identity linkerd identity -n my-namespace ``` ## Debugging mTLS Issues ```bash # Istio - Check if mTLS is enabled istioctl authn tls-check my-service.my-namespace.svc.cluster.local # Verify peer authentication kubectl get peerauthentication --all-namespaces # Check destination rules kubectl get destinationrule --all-namespaces # Debug TLS handshake istioctl proxy-config log deploy/my-app --level debug kubectl logs deploy/my-app -c istio-proxy | grep -i tls # Linkerd - Check mTLS status linkerd viz edges deployment -n my-namespace linkerd viz tap deploy/my-app --to deploy/my-backend ``` ## Best Practices ### Do's - **Start with PERMISSIVE** - Migrate gradually to STRICT - **Monitor certificate expiry** - Set up alerts - **Use short-lived certs** - 24h or less for workloads - **Rotate CA periodically** - Plan for CA rotation - **Log TLS errors** - For debugging and audit ### Don'ts - **Don't disable mTLS** - For convenience in production - **Don't ignore cert expiry** - Automate rotation - **Don't use self-signed certs** - Use proper CA hierarchy - **Don't skip verification** - Verify the full chain ## Resources - [Istio Security](https://istio.io/latest/docs/concepts/security/) - [SPIFFE/SPIRE](https://spiffe.io/) - [cert-manager](https://cert-manager.io/) - [Zero Trust Architecture (NIST)](https://www.nist.gov/publications/zero-trust-architecture)
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

screen-reader-testing

Test web applications with screen readers including VoiceOver,

coding
⭐1
# Screen Reader Testing Practical guide to testing web applications with screen readers for comprehensive accessibility validation. ## When to Use This Skill - Validating screen reader compatibility - Testing ARIA implementations - Debugging assistive technology issues - Verifying form accessibility - Testing dynamic content announcements - Ensuring navigation accessibility ## Core Concepts ### 1. Major Screen Readers | Screen Reader | Platform | Browser | Usage | | ------------- | --------- | -------------- | ----- | | **VoiceOver** | macOS/iOS | Safari | ~15% | | **NVDA** | Windows | Firefox/Chrome | ~31% | | **JAWS** | Windows | Chrome/IE | ~40% | | **TalkBack** | Android | Chrome | ~10% | | **Narrator** | Windows | Edge | ~4% | ### 2. Testing Priority ``` Minimum Coverage: 1. NVDA + Firefox (Windows) 2. VoiceOver + Safari (macOS) 3. VoiceOver + Safari (iOS) Comprehensive Coverage: + JAWS + Chrome (Windows) + TalkBack + Chrome (Android) + Narrator + Edge (Windows) ``` ### 3. Screen Reader Modes | Mode | Purpose | When Used | | ------------------ | ---------------------- | ----------------- | | **Browse/Virtual** | Read content | Default reading | | **Focus/Forms** | Interact with controls | Filling forms | | **Application** | Custom widgets | ARIA applications | ## VoiceOver (macOS) ### Setup ``` Enable: System Preferences → Accessibility → VoiceOver Toggle: Cmd + F5 Quick Toggle: Triple-press Touch ID ``` ### Essential Commands ``` Navigation: VO = Ctrl + Option (VoiceOver modifier) VO + Right Arrow Next element VO + Left Arrow Previous element VO + Shift + Down Enter group VO + Shift + Up Exit group Reading: VO + A Read all from cursor Ctrl Stop speaking VO + B Read current paragraph Interaction: VO + Space Activate element VO + Shift + M Open menu Tab Next focusable element Shift + Tab Previous focusable element Rotor (VO + U): Navigate by: Headings, Links, Forms, Landmarks Left/Right Arrow Change rotor category Up/Down Arrow Navigate within category Enter Go to item Web Specific: VO + Cmd + H Next heading VO + Cmd + J Next form control VO + Cmd + L Next link VO + Cmd + T Next table ``` ### Testing Checklist ```markdown ## VoiceOver Testing Checklist ### Page Load - [ ] Page title announced - [ ] Main landmark found - [ ] Skip link works ### Navigation - [ ] All headings discoverable via rotor - [ ] Heading levels logical (H1 → H2 → H3) - [ ] Landmarks properly labeled - [ ] Skip links functional ### Links & Buttons - [ ] Link purpose clear - [ ] Button actions described - [ ] New window/tab announced ### Forms - [ ] All labels read with inputs - [ ] Required fields announced - [ ] Error messages read - [ ] Instructions available - [ ] Focus moves to errors ### Dynamic Content - [ ] Alerts announced immediately - [ ] Loading states communicated - [ ] Content updates announced - [ ] Modals trap focus correctly ### Tables - [ ] Headers associated with cells - [ ] Table navigation works - [ ] Complex tables have captions ``` ### Common Issues & Fixes ```html <!-- Issue: Button not announcing purpose --> <button><svg>...</svg></button> <!-- Fix --> <button aria-label="Close dialog"><svg aria-hidden="true">...</svg></button> <!-- Issue: Dynamic content not announced --> <div id="results">New results loaded</div> <!-- Fix --> <div id="results" role="status" aria-live="polite">New results loaded</div> <!-- Issue: Form error not read --> <input type="email" /> <span class="error">Invalid email</span> <!-- Fix --> <input type="email" aria-invalid="true" aria-describedby="email-error" /> <span id="email-error" role="alert">Invalid email</span> ``` ## NVDA (Windows) ### Setup ``` Download: nvaccess.org Start: Ctrl + Alt + N Stop: Insert + Q ``` ### Essential Commands ``` Navigation: Insert = NVDA modifier Down Arrow Next line Up Arrow Previous line Tab Next focusable Shift + Tab Previous focusable Reading: NVDA + Down Arrow Say all Ctrl Stop speech NVDA + Up Arrow Current line Headings: H Next heading Shift + H Previous heading 1-6 Heading level 1-6 Forms: F Next form field B Next button E Next edit field X Next checkbox C Next combo box Links: K Next link U Next unvisited link V Next visited link Landmarks: D Next landmark Shift + D Previous landmark Tables: T Next table Ctrl + Alt + Arrows Navigate cells Elements List (NVDA + F7): Shows all links, headings, form fields, landmarks ``` ### Browse vs Focus Mode ``` NVDA automatically switches modes: - Browse Mode: Arrow keys navigate content - Focus Mode: Arrow keys control interactive elements Manual switch: NVDA + Space Watch for: - "Browse mode" announcement when navigating - "Focus mode" when entering form fields - Application role forces forms mode ``` ### Testing Script ```markdown ## NVDA Test Script ### Initial Load 1. Navigate to page 2. Let page finish loading 3. Press Insert + Down to read all 4. Note: Page title, main content identified? ### Landmark Navigation 1. Press D repeatedly 2. Check: All main areas reachable? 3. Check: Landmarks properly labeled? ### Heading Navigation 1. Press Insert + F7 → Headings 2. Check: Logical heading structure? 3. Press H to navigate headings 4. Check: All sections discoverable? ### Form Testing 1. Press F to find first form field 2. Check: Label read? 3. Fill in invalid data 4. Submit form 5. Check: Errors announced? 6. Check: Focus moved to error? ### Interactive Elements 1. Tab through all interactive elements 2. Check: Each announces role and state 3. Activate buttons with Enter/Space 4. Check: Result announced? ### Dynamic Content 1. Trigger content update 2. Check: Change announced? 3. Open modal 4. Check: Focus trapped? 5. Close modal 6. Check: Focus returns? ``` ## JAWS (Windows) ### Essential Commands ``` Start: Desktop shortcut or Ctrl + Alt + J Virtual Cursor: Auto-enabled in browsers Navigation: Arrow keys Navigate content Tab Next focusable Insert + Down Read all Ctrl Stop speech Quick Keys: H Next heading T Next table F Next form field B Next button G Next graphic L Next list ; Next landmark Forms Mode: Enter Enter forms mode Numpad + Exit forms mode F5 List form fields Lists: Insert + F7 Link list Insert + F6 Heading list Insert + F5 Form field list Tables: Ctrl + Alt + Arrows Table navigation ``` ## TalkBack (Android) ### Setup ``` Enable: Settings → Accessibility → TalkBack Toggle: Hold both volume buttons 3 seconds ``` ### Gestures ``` Explore: Drag finger across screen Next: Swipe right Previous: Swipe left Activate: Double tap Scroll: Two finger swipe Reading Controls (swipe up then right): - Headings - Links - Controls - Characters - Words - Lines - Paragraphs ``` ## Common Test Scenarios ### 1. Modal Dialog ```html <!-- Accessible modal structure --> <div role="dialog" aria-modal="true" aria-labelledby="dialog-title" aria-describedby="dialog-desc" > <h2 id="dialog-title">Confirm Delete</h2> <p id="dialog-desc">This action cannot be undone.</p> <button>Cancel</button> <button>Delete</button> </div> ``` ```javascript // Focus management function openModal(modal) { // Store last focused element lastFocus = document.activeElement; // Move focus to modal modal.querySelector("h2").focus(); // Trap focus modal.addEventListener("keydown", trapFocus); } function closeModal(modal) { // Return focus lastFocus.focus(); } function trapFocus(e) { if (e.key === "Tab") { const focusable = modal.querySelectorAll( 'button, [href], input, select, textarea, [tabindex]:not([tabindex="-1"])', ); const first = focusable[0]; const last = focusable[focusable.length - 1]; if (e.shiftKey && document.activeElement === first) { last.focus(); e.preventDefault(); } else if (!e.shiftKey && document.activeElement === last) { first.focus(); e.preventDefault(); } } if (e.key === "Escape") { closeModal(modal); } } ``` ### 2. Live Regions ```html <!-- Status messages (polite) --> <div role="status" aria-live="polite" aria-atomic="true"> <!-- Content updates will be announced after current speech --> </div> <!-- Alerts (assertive) --> <div role="alert" aria-live="assertive"> <!-- Content updates interrupt current speech --> </div> <!-- Progress updates --> <div role="progressbar" aria-valuenow="75" aria-valuemin="0" aria-valuemax="100" aria-label="Upload progress" ></div> <!-- Log (additions only) --> <div role="log" aria-live="polite" aria-relevant="additions"> <!-- New messages announced, removals not --> </div> ``` ### 3. Tab Interface ```html <div role="tablist" aria-label="Product information"> <button role="tab" id="tab-1" aria-selected="true" aria-controls="panel-1"> Description </button> <button role="tab" id="tab-2" aria-selected="false" aria-controls="panel-2" tabindex="-1" > Reviews </button> </div> <div role="tabpanel" id="panel-1" aria-labelledby="tab-1"> Product description content... </div> <div role="tabpanel" id="panel-2" aria-labelledby="tab-2" hidden> Reviews content... </div> ``` ```javascript // Tab keyboard navigation tablist.addEventListener("keydown", (e) => { const tabs = [...tablist.querySelectorAll('[role="tab"]')]; const index = tabs.indexOf(document.activeElement); let newIndex; switch (e.key) { case "ArrowRight": newIndex = (index + 1) % tabs.length; break; case "ArrowLeft": newIndex = (index - 1 + tabs.length) % tabs.length; break; case "Home": newIndex = 0; break; case "End": newIndex = tabs.length - 1; break; default: return; } tabs[newIndex].focus(); activateTab(tabs[newIndex]); e.preventDefault(); }); ``` ## Debugging Tips ```javascript // Log what screen reader sees function logAccessibleName(element) { const computed = window.getComputedStyle(element); console.log({ role: element.getAttribute("role") || element.tagName, name: element.getAttribute("aria-label") || element.getAttribute("aria-labelledby") || element.textContent, state: { expanded: element.getAttribute("aria-expanded"), selected: element.getAttribute("aria-selected"), checked: element.getAttribute("aria-checked"), disabled: element.disabled, }, visible: computed.display !== "none" && computed.visibility !== "hidden", }); } ``` ## Best Practices ### Do's - **Test with actual screen readers** - Not just simulators - **Use semantic HTML first** - ARIA is supplemental - **Test in browse and focus modes** - Different experiences - **Verify focus management** - Especially for SPAs - **Test keyboard only first** - Foundation for SR testing ### Don'ts - **Don't assume one SR is enough** - Test multiple - **Don't ignore mobile** - Growing user base - **Don't test only happy path** - Test error states - **Don't skip dynamic content** - Most common issues - **Don't rely on visual testing** - Different experience ## Resources - [VoiceOver User Guide](https://support.apple.com/guide/voiceover/welcome/mac) - [NVDA User Guide](https://www.nvaccess.org/files/nvda/documentation/userGuide.html) - [JAWS Documentation](https://support.freedomscientific.com/Products/Blindness/JAWS) - [WebAIM Screen Reader Survey](https://webaim.org/projects/screenreadersurvey/)
👍1
👁️1
🤖 Auto-discovered
🤖system prompt•7 months ago

service-mesh-observability

Implement comprehensive observability for service meshes including

architecture
⭐1
# Service Mesh Observability Complete guide to observability patterns for Istio, Linkerd, and service mesh deployments. ## When to Use This Skill - Setting up distributed tracing across services - Implementing service mesh metrics and dashboards - Debugging latency and error issues - Defining SLOs for service communication - Visualizing service dependencies - Troubleshooting mesh connectivity ## Core Concepts ### 1. Three Pillars of Observability ``` ┌─────────────────────────────────────────────────────┐ │ Observability │ ├─────────────────┬─────────────────┬─────────────────┤ │ Metrics │ Traces │ Logs │ │ │ │ │ │ • Request rate │ • Span context │ • Access logs │ │ • Error rate │ • Latency │ • Error details │ │ • Latency P50 │ • Dependencies │ • Debug info │ │ • Saturation │ • Bottlenecks │ • Audit trail │ └─────────────────┴─────────────────┴─────────────────┘ ``` ### 2. Golden Signals for Mesh | Signal | Description | Alert Threshold | | -------------- | ------------------------- | ----------------- | | **Latency** | Request duration P50, P99 | P99 > 500ms | | **Traffic** | Requests per second | Anomaly detection | | **Errors** | 5xx error rate | > 1% | | **Saturation** | Resource utilization | > 80% | ## Templates ### Template 1: Istio with Prometheus & Grafana ```yaml # Install Prometheus apiVersion: v1 kind: ConfigMap metadata: name: prometheus namespace: istio-system data: prometheus.yml: | global: scrape_interval: 15s scrape_configs: - job_name: 'istio-mesh' kubernetes_sd_configs: - role: endpoints namespaces: names: - istio-system relabel_configs: - source_labels: [__meta_kubernetes_service_name] action: keep regex: istio-telemetry --- # ServiceMonitor for Prometheus Operator apiVersion: monitoring.coreos.com/v1 kind: ServiceMonitor metadata: name: istio-mesh namespace: istio-system spec: selector: matchLabels: app: istiod endpoints: - port: http-monitoring interval: 15s ``` ### Template 2: Key Istio Metrics Queries ```promql # Request rate by service sum(rate(istio_requests_total{reporter="destination"}[5m])) by (destination_service_name) # Error rate (5xx) sum(rate(istio_requests_total{reporter="destination", response_code=~"5.."}[5m])) / sum(rate(istio_requests_total{reporter="destination"}[5m])) * 100 # P99 latency histogram_quantile(0.99, sum(rate(istio_request_duration_milliseconds_bucket{reporter="destination"}[5m])) by (le, destination_service_name)) # TCP connections sum(istio_tcp_connections_opened_total{reporter="destination"}) by (destination_service_name) # Request size histogram_quantile(0.99, sum(rate(istio_request_bytes_bucket{reporter="destination"}[5m])) by (le, destination_service_name)) ``` ### Template 3: Jaeger Distributed Tracing ```yaml # Jaeger installation for Istio apiVersion: install.istio.io/v1alpha1 kind: IstioOperator spec: meshConfig: enableTracing: true defaultConfig: tracing: sampling: 100.0 # 100% in dev, lower in prod zipkin: address: jaeger-collector.istio-system:9411 --- # Jaeger deployment apiVersion: apps/v1 kind: Deployment metadata: name: jaeger namespace: istio-system spec: selector: matchLabels: app: jaeger template: metadata: labels: app: jaeger spec: containers: - name: jaeger image: jaegertracing/all-in-one:1.50 ports: - containerPort: 5775 # UDP - containerPort: 6831 # Thrift - containerPort: 6832 # Thrift - containerPort: 5778 # Config - containerPort: 16686 # UI - containerPort: 14268 # HTTP - containerPort: 14250 # gRPC - containerPort: 9411 # Zipkin env: - name: COLLECTOR_ZIPKIN_HOST_PORT value: ":9411" ``` ### Template 4: Linkerd Viz Dashboard ```bash # Install Linkerd viz extension linkerd viz install | kubectl apply -f - # Access dashboard linkerd viz dashboard # CLI commands for observability # Top requests linkerd viz top deploy/my-app # Per-route metrics linkerd viz routes deploy/my-app --to deploy/backend # Live traffic inspection linkerd viz tap deploy/my-app --to deploy/backend # Service edges (dependencies) linkerd viz edges deployment -n my-namespace ``` ### Template 5: Grafana Dashboard JSON ```json { "dashboard": { "title": "Service Mesh Overview", "panels": [ { "title": "Request Rate", "type": "graph", "targets": [ { "expr": "sum(rate(istio_requests_total{reporter=\"destination\"}[5m])) by (destination_service_name)", "legendFormat": "{{destination_service_name}}" } ] }, { "title": "Error Rate", "type": "gauge", "targets": [ { "expr": "sum(rate(istio_requests_total{response_code=~\"5..\"}[5m])) / sum(rate(istio_requests_total[5m])) * 100" } ], "fieldConfig": { "defaults": { "thresholds": { "steps": [ { "value": 0, "color": "green" }, { "value": 1, "color": "yellow" }, { "value": 5, "color": "red" } ] } } } }, { "title": "P99 Latency", "type": "graph", "targets": [ { "expr": "histogram_quantile(0.99, sum(rate(istio_request_duration_milliseconds_bucket{reporter=\"destination\"}[5m])) by (le, destination_service_name))", "legendFormat": "{{destination_service_name}}" } ] }, { "title": "Service Topology", "type": "nodeGraph", "targets": [ { "expr": "sum(rate(istio_requests_total{reporter=\"destination\"}[5m])) by (source_workload, destination_service_name)" } ] } ] } } ``` ### Template 6: Kiali Service Mesh Visualization ```yaml # Kiali installation apiVersion: kiali.io/v1alpha1 kind: Kiali metadata: name: kiali namespace: istio-system spec: auth: strategy: anonymous # or openid, token deployment: accessible_namespaces: - "**" external_services: prometheus: url: http://prometheus.istio-system:9090 tracing: url: http://jaeger-query.istio-system:16686 grafana: url: http://grafana.istio-system:3000 ``` ### Template 7: OpenTelemetry Integration ```yaml # OpenTelemetry Collector for mesh apiVersion: v1 kind: ConfigMap metadata: name: otel-collector-config data: config.yaml: | receivers: otlp: protocols: grpc: endpoint: 0.0.0.0:4317 http: endpoint: 0.0.0.0:4318 zipkin: endpoint: 0.0.0.0:9411 processors: batch: timeout: 10s exporters: jaeger: endpoint: jaeger-collector:14250 tls: insecure: true prometheus: endpoint: 0.0.0.0:8889 service: pipelines: traces: receivers: [otlp, zipkin] processors: [batch] exporters: [jaeger] metrics: receivers: [otlp] processors: [batch] exporters: [prometheus] --- # Istio Telemetry v2 with OTel apiVersion: telemetry.istio.io/v1alpha1 kind: Telemetry metadata: name: mesh-default namespace: istio-system spec: tracing: - providers: - name: otel randomSamplingPercentage: 10 ``` ## Alerting Rules ```yaml apiVersion: monitoring.coreos.com/v1 kind: PrometheusRule metadata: name: mesh-alerts namespace: istio-system spec: groups: - name: mesh.rules rules: - alert: HighErrorRate expr: | sum(rate(istio_requests_total{response_code=~"5.."}[5m])) by (destination_service_name) / sum(rate(istio_requests_total[5m])) by (destination_service_name) > 0.05 for: 5m labels: severity: critical annotations: summary: "High error rate for {{ $labels.destination_service_name }}" - alert: HighLatency expr: | histogram_quantile(0.99, sum(rate(istio_request_duration_milliseconds_bucket[5m])) by (le, destination_service_name)) > 1000 for: 5m labels: severity: warning annotations: summary: "High P99 latency for {{ $labels.destination_service_name }}" - alert: MeshCertExpiring expr: | (certmanager_certificate_expiration_timestamp_seconds - time()) / 86400 < 7 labels: severity: warning annotations: summary: "Mesh certificate expiring in less than 7 days" ``` ## Best Practices ### Do's - **Sample appropriately** - 100% in dev, 1-10% in prod - **Use trace context** - Propagate headers consistently - **Set up alerts** - For golden signals - **Correlate metrics/traces** - Use exemplars - **Retain strategically** - Hot/cold storage tiers ### Don'ts - **Don't over-sample** - Storage costs add up - **Don't ignore cardinality** - Limit label values - **Don't skip dashboards** - Visualize dependencies - **Don't forget costs** - Monitor observability costs ## Resources - [Istio Observability](https://istio.io/latest/docs/tasks/observability/) - [Linkerd Observability](https://linkerd.io/2.14/features/dashboard/) - [OpenTelemetry](https://opentelemetry.io/) - [Kiali](https://kiali.io/)
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

context-driven-development

Creates and maintains project context artifacts (product.md,

coding
⭐1
# Context-Driven Development Guide for implementing and maintaining context as a managed artifact alongside code, enabling consistent AI interactions and team alignment through structured project documentation. ## When to Use This Skill - Setting up new projects with Conductor - Understanding the relationship between context artifacts - Maintaining consistency across AI-assisted development sessions - Onboarding team members to an existing Conductor project - Deciding when to update context documents - Managing greenfield vs brownfield project contexts ## Core Philosophy Context-Driven Development treats project context as a first-class artifact managed alongside code. Instead of relying on ad-hoc prompts or scattered documentation, establish a persistent, structured foundation that informs all AI interactions. Key principles: 1. **Context precedes code**: Define what you're building and how before implementation 2. **Living documentation**: Context artifacts evolve with the project 3. **Single source of truth**: One canonical location for each type of information 4. **AI alignment**: Consistent context produces consistent AI behavior ## The Workflow Follow the **Context → Spec & Plan → Implement** workflow: 1. **Context Phase**: Establish or verify project context artifacts exist and are current 2. **Specification Phase**: Define requirements and acceptance criteria for work units 3. **Planning Phase**: Break specifications into phased, actionable tasks 4. **Implementation Phase**: Execute tasks following established workflow patterns ## Artifact Relationships ### product.md - Defines WHAT and WHY Purpose: Captures product vision, goals, target users, and business context. Contents: - Product name and one-line description - Problem statement and solution approach - Target user personas - Core features and capabilities - Success metrics and KPIs - Product roadmap (high-level) Update when: - Product vision or goals change - New major features are planned - Target audience shifts - Business priorities evolve ### product-guidelines.md - Defines HOW to Communicate Purpose: Establishes brand voice, messaging standards, and communication patterns. Contents: - Brand voice and tone guidelines - Terminology and glossary - Error message conventions - User-facing copy standards - Documentation style Update when: - Brand guidelines change - New terminology is introduced - Communication patterns need refinement ### tech-stack.md - Defines WITH WHAT Purpose: Documents technology choices, dependencies, and architectural decisions. Contents: - Primary languages and frameworks - Key dependencies with versions - Infrastructure and deployment targets - Development tools and environment - Testing frameworks - Code quality tools Update when: - Adding new dependencies - Upgrading major versions - Changing infrastructure - Adopting new tools or patterns ### workflow.md - Defines HOW to Work Purpose: Establishes development practices, quality gates, and team workflows. Contents: - Development methodology (TDD, etc.) - Git workflow and commit conventions - Code review requirements - Testing requirements and coverage targets - Quality assurance gates - Deployment procedures Update when: - Team practices evolve - Quality standards change - New workflow patterns are adopted ### tracks.md - Tracks WHAT'S HAPPENING Purpose: Registry of all work units with status and metadata. Contents: - Active tracks with current status - Completed tracks with completion dates - Track metadata (type, priority, assignee) - Links to individual track directories Update when: - New tracks are created - Track status changes - Tracks are completed or archived See [references/artifact-templates.md](references/artifact-templates.md) for copy-paste starter templates. ## Context Maintenance Principles ### Keep Artifacts Synchronized Ensure changes in one artifact reflect in related documents: - New feature in product.md → Update tech-stack.md if new dependencies needed - Completed track → Update product.md to reflect new capabilities - Workflow change → Update all affected track plans ### Update tech-stack.md When Adding Dependencies Before adding any new dependency: 1. Check if existing dependencies solve the need 2. Document the rationale for new dependencies 3. Add version constraints 4. Note any configuration requirements ### Update product.md When Features Complete After completing a feature track: 1. Move feature from "planned" to "implemented" in product.md 2. Update any affected success metrics 3. Document any scope changes from original plan ### Verify Context Before Implementation Before starting any track: 1. Read all context artifacts 2. Flag any outdated information 3. Propose updates before proceeding 4. Confirm context accuracy with stakeholders ## Greenfield vs Brownfield Handling ### Greenfield Projects (New) For new projects: 1. Run `/conductor:setup` to create all artifacts interactively 2. Answer questions about product vision, tech preferences, and workflow 3. Generate initial style guides for chosen languages 4. Create empty tracks registry Characteristics: - Full control over context structure - Define standards before code exists - Establish patterns early ### Brownfield Projects (Existing) For existing codebases: 1. Run `/conductor:setup` with existing codebase detection 2. System analyzes existing code, configs, and documentation 3. Pre-populate artifacts based on discovered patterns 4. Review and refine generated context Characteristics: - Extract implicit context from existing code - Reconcile existing patterns with desired patterns - Document technical debt and modernization plans - Preserve working patterns while establishing standards ## Benefits ### Team Alignment - New team members onboard faster with explicit context - Consistent terminology and conventions across the team - Shared understanding of product goals and technical decisions ### AI Consistency - AI assistants produce aligned outputs across sessions - Reduced need to re-explain context in each interaction - Predictable behavior based on documented standards ### Institutional Memory - Decisions and rationale are preserved - Context survives team changes - Historical context informs future decisions ### Quality Assurance - Standards are explicit and verifiable - Deviations from context are detectable - Quality gates are documented and enforceable ## Directory Structure ``` conductor/ ├── index.md # Navigation hub linking all artifacts ├── product.md # Product vision and goals ├── product-guidelines.md # Communication standards ├── tech-stack.md # Technology preferences ├── workflow.md # Development practices ├── tracks.md # Work unit registry ├── setup_state.json # Resumable setup state ├── code_styleguides/ # Language-specific conventions │ ├── python.md │ ├── typescript.md │ └── ... └── tracks/ └── <track-id>/ ├── spec.md ├── plan.md ├── metadata.json └── index.md ``` ## Context Lifecycle 1. **Creation**: Initial setup via `/conductor:setup` 2. **Validation**: Verify before each track 3. **Evolution**: Update as project grows 4. **Synchronization**: Keep artifacts aligned 5. **Archival**: Document historical decisions ## Context Validation Checklist Before starting implementation on any track, validate context: ### Product Context - [ ] product.md reflects current product vision - [ ] Target users are accurately described - [ ] Feature list is up to date - [ ] Success metrics are defined ### Technical Context - [ ] tech-stack.md lists all current dependencies - [ ] Version numbers are accurate - [ ] Infrastructure targets are correct - [ ] Development tools are documented ### Workflow Context - [ ] workflow.md describes current practices - [ ] Quality gates are defined - [ ] Coverage targets are specified - [ ] Commit conventions are documented ### Track Context - [ ] tracks.md shows all active work - [ ] No stale or abandoned tracks - [ ] Dependencies between tracks are noted ## Common Anti-Patterns Avoid these context management mistakes: ### Stale Context Problem: Context documents become outdated and misleading. Solution: Update context as part of each track's completion process. ### Context Sprawl Problem: Information scattered across multiple locations. Solution: Use the defined artifact structure; resist creating new document types. ### Implicit Context Problem: Relying on knowledge not captured in artifacts. Solution: If you reference something repeatedly, add it to the appropriate artifact. ### Context Hoarding Problem: One person maintains context without team input. Solution: Review context artifacts in pull requests; make updates collaborative. ### Over-Specification Problem: Context becomes so detailed it's impossible to maintain. Solution: Keep artifacts focused on decisions that affect AI behavior and team alignment. ## Integration with Development Tools ### IDE Integration Configure your IDE to display context files prominently: - Pin conductor/product.md for quick reference - Add tech-stack.md to project notes - Create snippets for common patterns from style guides ### Git Hooks Consider pre-commit hooks that: - Warn when dependencies change without tech-stack.md update - Remind to update product.md when feature branches merge - Validate context artifact syntax ### CI/CD Integration Include context validation in pipelines: - Check tech-stack.md matches actual dependencies - Verify links in context documents resolve - Ensure tracks.md status matches git branch state ## Session Continuity Conductor supports multi-session development through context persistence: ### Starting a New Session 1. Read index.md to orient yourself 2. Check tracks.md for active work 3. Review relevant track's plan.md for current task 4. Verify context artifacts are current ### Ending a Session 1. Update plan.md with current progress 2. Note any blockers or decisions made 3. Commit in-progress work with clear status 4. Update tracks.md if status changed ### Handling Interruptions If interrupted mid-task: 1. Mark task as `[~]` with note about stopping point 2. Commit work-in-progress to feature branch 3. Document any uncommitted decisions in plan.md ## Best Practices 1. **Read context first**: Always read relevant artifacts before starting work 2. **Small updates**: Make incremental context changes, not massive rewrites 3. **Link decisions**: Reference context when making implementation choices 4. **Version context**: Commit context changes alongside code changes 5. **Review context**: Include context artifact reviews in code reviews 6. **Validate regularly**: Run context validation checklist before major work 7. **Communicate changes**: Notify team when context artifacts change significantly 8. **Preserve history**: Use git to track context evolution over time 9. **Question staleness**: If context feels wrong, investigate and update 10. **Keep it actionable**: Every context item should inform a decision or behavior
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

godot-gdscript-patterns

Master Godot 4 GDScript patterns including signals, scenes, state

coding
⭐1
# Godot GDScript Patterns Production patterns for Godot 4.x game development with GDScript, covering architecture, signals, scenes, and optimization. ## When to Use This Skill - Building games with Godot 4 - Implementing game systems in GDScript - Designing scene architecture - Managing game state - Optimizing GDScript performance - Learning Godot best practices ## Core Concepts ### 1. Godot Architecture ``` Node: Base building block ├── Scene: Reusable node tree (saved as .tscn) ├── Resource: Data container (saved as .tres) ├── Signal: Event communication └── Group: Node categorization ``` ### 2. GDScript Basics ```gdscript class_name Player extends CharacterBody2D # Signals signal health_changed(new_health: int) signal died # Exports (Inspector-editable) @export var speed: float = 200.0 @export var max_health: int = 100 @export_range(0, 1) var damage_reduction: float = 0.0 @export_group("Combat") @export var attack_damage: int = 10 @export var attack_cooldown: float = 0.5 # Onready (initialized when ready) @onready var sprite: Sprite2D = $Sprite2D @onready var animation: AnimationPlayer = $AnimationPlayer @onready var hitbox: Area2D = $Hitbox # Private variables (convention: underscore prefix) var _health: int var _can_attack: bool = true func _ready() -> void: _health = max_health func _physics_process(delta: float) -> void: var direction := Input.get_vector("left", "right", "up", "down") velocity = direction * speed move_and_slide() func take_damage(amount: int) -> void: var actual_damage := int(amount * (1.0 - damage_reduction)) _health = max(_health - actual_damage, 0) health_changed.emit(_health) if _health <= 0: died.emit() ``` ## Patterns ### Pattern 1: State Machine ```gdscript # state_machine.gd class_name StateMachine extends Node signal state_changed(from_state: StringName, to_state: StringName) @export var initial_state: State var current_state: State var states: Dictionary = {} func _ready() -> void: # Register all State children for child in get_children(): if child is State: states[child.name] = child child.state_machine = self child.process_mode = Node.PROCESS_MODE_DISABLED # Start initial state if initial_state: current_state = initial_state current_state.process_mode = Node.PROCESS_MODE_INHERIT current_state.enter() func _process(delta: float) -> void: if current_state: current_state.update(delta) func _physics_process(delta: float) -> void: if current_state: current_state.physics_update(delta) func _unhandled_input(event: InputEvent) -> void: if current_state: current_state.handle_input(event) func transition_to(state_name: StringName, msg: Dictionary = {}) -> void: if not states.has(state_name): push_error("State '%s' not found" % state_name) return var previous_state := current_state previous_state.exit() previous_state.process_mode = Node.PROCESS_MODE_DISABLED current_state = states[state_name] current_state.process_mode = Node.PROCESS_MODE_INHERIT current_state.enter(msg) state_changed.emit(previous_state.name, current_state.name) ``` ```gdscript # state.gd class_name State extends Node var state_machine: StateMachine func enter(_msg: Dictionary = {}) -> void: pass func exit() -> void: pass func update(_delta: float) -> void: pass func physics_update(_delta: float) -> void: pass func handle_input(_event: InputEvent) -> void: pass ``` ```gdscript # player_idle.gd class_name PlayerIdle extends State @export var player: Player func enter(_msg: Dictionary = {}) -> void: player.animation.play("idle") func physics_update(_delta: float) -> void: var direction := Input.get_vector("left", "right", "up", "down") if direction != Vector2.ZERO: state_machine.transition_to("Move") func handle_input(event: InputEvent) -> void: if event.is_action_pressed("attack"): state_machine.transition_to("Attack") elif event.is_action_pressed("jump"): state_machine.transition_to("Jump") ``` ### Pattern 2: Autoload Singletons ```gdscript # game_manager.gd (Add to Project Settings > Autoload) extends Node signal game_started signal game_paused(is_paused: bool) signal game_over(won: bool) signal score_changed(new_score: int) enum GameState { MENU, PLAYING, PAUSED, GAME_OVER } var state: GameState = GameState.MENU var score: int = 0: set(value): score = value score_changed.emit(score) var high_score: int = 0 func _ready() -> void: process_mode = Node.PROCESS_MODE_ALWAYS _load_high_score() func _input(event: InputEvent) -> void: if event.is_action_pressed("pause") and state == GameState.PLAYING: toggle_pause() func start_game() -> void: score = 0 state = GameState.PLAYING game_started.emit() func toggle_pause() -> void: var is_paused := state != GameState.PAUSED if is_paused: state = GameState.PAUSED get_tree().paused = true else: state = GameState.PLAYING get_tree().paused = false game_paused.emit(is_paused) func end_game(won: bool) -> void: state = GameState.GAME_OVER if score > high_score: high_score = score _save_high_score() game_over.emit(won) func add_score(points: int) -> void: score += points func _load_high_score() -> void: if FileAccess.file_exists("user://high_score.save"): var file := FileAccess.open("user://high_score.save", FileAccess.READ) high_score = file.get_32() func _save_high_score() -> void: var file := FileAccess.open("user://high_score.save", FileAccess.WRITE) file.store_32(high_score) ``` ```gdscript # event_bus.gd (Global signal bus) extends Node # Player events signal player_spawned(player: Node2D) signal player_died(player: Node2D) signal player_health_changed(health: int, max_health: int) # Enemy events signal enemy_spawned(enemy: Node2D) signal enemy_died(enemy: Node2D, position: Vector2) # Item events signal item_collected(item_type: StringName, value: int) signal powerup_activated(powerup_type: StringName) # Level events signal level_started(level_number: int) signal level_completed(level_number: int, time: float) signal checkpoint_reached(checkpoint_id: int) ``` ### Pattern 3: Resource-based Data ```gdscript # weapon_data.gd class_name WeaponData extends Resource @export var name: StringName @export var damage: int @export var attack_speed: float @export var range: float @export_multiline var description: String @export var icon: Texture2D @export var projectile_scene: PackedScene @export var sound_attack: AudioStream ``` ```gdscript # character_stats.gd class_name CharacterStats extends Resource signal stat_changed(stat_name: StringName, new_value: float) @export var max_health: float = 100.0 @export var attack: float = 10.0 @export var defense: float = 5.0 @export var speed: float = 200.0 # Runtime values (not saved) var _current_health: float func _init() -> void: _current_health = max_health func get_current_health() -> float: return _current_health func take_damage(amount: float) -> float: var actual_damage := maxf(amount - defense, 1.0) _current_health = maxf(_current_health - actual_damage, 0.0) stat_changed.emit("health", _current_health) return actual_damage func heal(amount: float) -> void: _current_health = minf(_current_health + amount, max_health) stat_changed.emit("health", _current_health) func duplicate_for_runtime() -> CharacterStats: var copy := duplicate() as CharacterStats copy._current_health = copy.max_health return copy ``` ```gdscript # Using resources class_name Character extends CharacterBody2D @export var base_stats: CharacterStats @export var weapon: WeaponData var stats: CharacterStats func _ready() -> void: # Create runtime copy to avoid modifying the resource stats = base_stats.duplicate_for_runtime() stats.stat_changed.connect(_on_stat_changed) func attack() -> void: if weapon: print("Attacking with %s for %d damage" % [weapon.name, weapon.damage]) func _on_stat_changed(stat_name: StringName, value: float) -> void: if stat_name == "health" and value <= 0: die() ``` ### Pattern 4: Object Pooling ```gdscript # object_pool.gd class_name ObjectPool extends Node @export var pooled_scene: PackedScene @export var initial_size: int = 10 @export var can_grow: bool = true var _available: Array[Node] = [] var _in_use: Array[Node] = [] func _ready() -> void: _initialize_pool() func _initialize_pool() -> void: for i in initial_size: _create_instance() func _create_instance() -> Node: var instance := pooled_scene.instantiate() instance.process_mode = Node.PROCESS_MODE_DISABLED instance.visible = false add_child(instance) _available.append(instance) # Connect return signal if exists if instance.has_signal("returned_to_pool"): instance.returned_to_pool.connect(_return_to_pool.bind(instance)) return instance func get_instance() -> Node: var instance: Node if _available.is_empty(): if can_grow: instance = _create_instance() _available.erase(instance) else: push_warning("Pool exhausted and cannot grow") return null else: instance = _available.pop_back() instance.process_mode = Node.PROCESS_MODE_INHERIT instance.visible = true _in_use.append(instance) if instance.has_method("on_spawn"): instance.on_spawn() return instance func _return_to_pool(instance: Node) -> void: if not instance in _in_use: return _in_use.erase(instance) if instance.has_method("on_despawn"): instance.on_despawn() instance.process_mode = Node.PROCESS_MODE_DISABLED instance.visible = false _available.append(instance) func return_all() -> void: for instance in _in_use.duplicate(): _return_to_pool(instance) ``` ```gdscript # pooled_bullet.gd class_name PooledBullet extends Area2D signal returned_to_pool @export var speed: float = 500.0 @export var lifetime: float = 5.0 var direction: Vector2 var _timer: float func on_spawn() -> void: _timer = lifetime func on_despawn() -> void: direction = Vector2.ZERO func initialize(pos: Vector2, dir: Vector2) -> void: global_position = pos direction = dir.normalized() rotation = direction.angle() func _physics_process(delta: float) -> void: position += direction * speed * delta _timer -= delta if _timer <= 0: returned_to_pool.emit() func _on_body_entered(body: Node2D) -> void: if body.has_method("take_damage"): body.take_damage(10) returned_to_pool.emit() ``` ### Pattern 5: Component System ```gdscript # health_component.gd class_name HealthComponent extends Node signal health_changed(current: int, maximum: int) signal damaged(amount: int, source: Node) signal healed(amount: int) signal died @export var max_health: int = 100 @export var invincibility_time: float = 0.0 var current_health: int: set(value): var old := current_health current_health = clampi(value, 0, max_health) if current_health != old: health_changed.emit(current_health, max_health) var _invincible: bool = false func _ready() -> void: current_health = max_health func take_damage(amount: int, source: Node = null) -> int: if _invincible or current_health <= 0: return 0 var actual := mini(amount, current_health) current_health -= actual damaged.emit(actual, source) if current_health <= 0: died.emit() elif invincibility_time > 0: _start_invincibility() return actual func heal(amount: int) -> int: var actual := mini(amount, max_health - current_health) current_health += actual if actual > 0: healed.emit(actual) return actual func _start_invincibility() -> void: _invincible = true await get_tree().create_timer(invincibility_time).timeout _invincible = false ``` ```gdscript # hitbox_component.gd class_name HitboxComponent extends Area2D signal hit(hurtbox: HurtboxComponent) @export var damage: int = 10 @export var knockback_force: float = 200.0 var owner_node: Node func _ready() -> void: owner_node = get_parent() area_entered.connect(_on_area_entered) func _on_area_entered(area: Area2D) -> void: if area is HurtboxComponent: var hurtbox := area as HurtboxComponent if hurtbox.owner_node != owner_node: hit.emit(hurtbox) hurtbox.receive_hit(self) ``` ```gdscript # hurtbox_component.gd class_name HurtboxComponent extends Area2D signal hurt(hitbox: HitboxComponent) @export var health_component: HealthComponent var owner_node: Node func _ready() -> void: owner_node = get_parent() func receive_hit(hitbox: HitboxComponent) -> void: hurt.emit(hitbox) if health_component: health_component.take_damage(hitbox.damage, hitbox.owner_node) ``` ### Pattern 6: Scene Management ```gdscript # scene_manager.gd (Autoload) extends Node signal scene_loading_started(scene_path: String) signal scene_loading_progress(progress: float) signal scene_loaded(scene: Node) signal transition_started signal transition_finished @export var transition_scene: PackedScene @export var loading_scene: PackedScene var _current_scene: Node var _transition: CanvasLayer var _loader: ResourceLoader func _ready() -> void: _current_scene = get_tree().current_scene if transition_scene: _transition = transition_scene.instantiate() add_child(_transition) _transition.visible = false func change_scene(scene_path: String, with_transition: bool = true) -> void: if with_transition: await _play_transition_out() _load_scene(scene_path) func change_scene_packed(scene: PackedScene, with_transition: bool = true) -> void: if with_transition: await _play_transition_out() _swap_scene(scene.instantiate()) func _load_scene(path: String) -> void: scene_loading_started.emit(path) # Check if already loaded if ResourceLoader.has_cached(path): var scene := load(path) as PackedScene _swap_scene(scene.instantiate()) return # Async loading ResourceLoader.load_threaded_request(path) while true: var progress := [] var status := ResourceLoader.load_threaded_get_status(path, progress) match status: ResourceLoader.THREAD_LOAD_IN_PROGRESS: scene_loading_progress.emit(progress[0]) await get_tree().process_frame ResourceLoader.THREAD_LOAD_LOADED: var scene := ResourceLoader.load_threaded_get(path) as PackedScene _swap_scene(scene.instantiate()) return _: push_error("Failed to load scene: %s" % path) return func _swap_scene(new_scene: Node) -> void: if _current_scene: _current_scene.queue_free() _current_scene = new_scene get_tree().root.add_child(_current_scene) get_tree().current_scene = _current_scene scene_loaded.emit(_current_scene) await _play_transition_in() func _play_transition_out() -> void: if not _transition: return transition_started.emit() _transition.visible = true if _transition.has_method("transition_out"): await _transition.transition_out() else: await get_tree().create_timer(0.3).timeout func _play_transition_in() -> void: if not _transition: transition_finished.emit() return if _transition.has_method("transition_in"): await _transition.transition_in() else: await get_tree().create_timer(0.3).timeout _transition.visible = false transition_finished.emit() ``` ### Pattern 7: Save System ```gdscript # save_manager.gd (Autoload) extends Node const SAVE_PATH := "user://savegame.save" const ENCRYPTION_KEY := "your_secret_key_here" signal save_completed signal load_completed signal save_error(message: String) func save_game(data: Dictionary) -> void: var file := FileAccess.open_encrypted_with_pass( SAVE_PATH, FileAccess.WRITE, ENCRYPTION_KEY ) if file == null: save_error.emit("Could not open save file") return var json := JSON.stringify(data) file.store_string(json) file.close() save_completed.emit() func load_game() -> Dictionary: if not FileAccess.file_exists(SAVE_PATH): return {} var file := FileAccess.open_encrypted_with_pass( SAVE_PATH, FileAccess.READ, ENCRYPTION_KEY ) if file == null: save_error.emit("Could not open save file") return {} var json := file.get_as_text() file.close() var parsed := JSON.parse_string(json) if parsed == null: save_error.emit("Could not parse save data") return {} load_completed.emit() return parsed func delete_save() -> void: if FileAccess.file_exists(SAVE_PATH): DirAccess.remove_absolute(SAVE_PATH) func has_save() -> bool: return FileAccess.file_exists(SAVE_PATH) ``` ```gdscript # saveable.gd (Attach to saveable nodes) class_name Saveable extends Node @export var save_id: String func _ready() -> void: if save_id.is_empty(): save_id = str(get_path()) func get_save_data() -> Dictionary: var parent := get_parent() var data := {"id": save_id} if parent is Node2D: data["position"] = {"x": parent.position.x, "y": parent.position.y} if parent.has_method("get_custom_save_data"): data.merge(parent.get_custom_save_data()) return data func load_save_data(data: Dictionary) -> void: var parent := get_parent() if data.has("position") and parent is Node2D: parent.position = Vector2(data.position.x, data.position.y) if parent.has_method("load_custom_save_data"): parent.load_custom_save_data(data) ``` ## Performance Tips ```gdscript # 1. Cache node references @onready var sprite := $Sprite2D # Good # $Sprite2D in _process() # Bad - repeated lookup # 2. Use object pooling for frequent spawning # See Pattern 4 # 3. Avoid allocations in hot paths var _reusable_array: Array = [] func _process(_delta: float) -> void: _reusable_array.clear() # Reuse instead of creating new # 4. Use static typing func calculate(value: float) -> float: # Good return value * 2.0 # 5. Disable processing when not needed func _on_off_screen() -> void: set_process(false) set_physics_process(false) ``` ## Best Practices ### Do's - **Use signals for decoupling** - Avoid direct references - **Type everything** - Static typing catches errors - **Use resources for data** - Separate data from logic - **Pool frequently spawned objects** - Avoid GC hitches - **Use Autoloads sparingly** - Only for truly global systems ### Don'ts - **Don't use `get_node()` in loops** - Cache ref
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

employment-contract-templates

Create employment contracts, offer letters, and HR policy documents

coding
⭐1
# Employment Contract Templates Templates and patterns for creating legally sound employment documentation including contracts, offer letters, and HR policies. ## When to Use This Skill - Drafting employment contracts - Creating offer letters - Writing employee handbooks - Developing HR policies - Standardizing employment documentation - Onboarding documentation ## Core Concepts ### 1. Employment Document Types | Document | Purpose | When Used | | ----------------------- | ----------------------- | ------------- | | **Offer Letter** | Initial job offer | Pre-hire | | **Employment Contract** | Formal agreement | Hire | | **Employee Handbook** | Policies & procedures | Onboarding | | **NDA** | Confidentiality | Before access | | **Non-Compete** | Competition restriction | Hire/Exit | ### 2. Key Legal Considerations ``` Employment Relationship: ├── At-Will vs. Contract ├── Employee vs. Contractor ├── Full-Time vs. Part-Time ├── Exempt vs. Non-Exempt └── Jurisdiction-Specific Requirements ``` **DISCLAIMER: These templates are for informational purposes only and do not constitute legal advice. Consult with qualified legal counsel before using any employment documents.** ## Templates ### Template 1: Offer Letter ```markdown # EMPLOYMENT OFFER LETTER [Company Letterhead] Date: [DATE] [Candidate Name] [Address] [City, State ZIP] Dear [Candidate Name], We are pleased to extend an offer of employment for the position of [JOB TITLE] at [COMPANY NAME]. We believe your skills and experience will be valuable additions to our team. ## Position Details **Title:** [Job Title] **Department:** [Department] **Reports To:** [Manager Name/Title] **Location:** [Office Location / Remote] **Start Date:** [Proposed Start Date] **Employment Type:** [Full-Time/Part-Time], [Exempt/Non-Exempt] ## Compensation **Base Salary:** $[AMOUNT] per [year/hour], paid [bi-weekly/semi-monthly/monthly] **Bonus:** [Eligible for annual bonus of up to X% based on company and individual performance / Not applicable] **Equity:** [X shares of stock options vesting over 4 years with 1-year cliff / Not applicable] ## Benefits You will be eligible for our standard benefits package, including: - Health insurance (medical, dental, vision) effective [date] - 401(k) with [X]% company match - [x] days paid time off per year - [x] paid holidays - [Other benefits] Full details will be provided during onboarding. ## Contingencies This offer is contingent upon: - Successful completion of background check - Verification of your right to work in [Country] - Execution of required employment documents including: - Confidentiality Agreement - [Non-Compete Agreement, if applicable] - [IP Assignment Agreement] ## At-Will Employment Please note that employment with [Company Name] is at-will. This means that either you or the Company may terminate the employment relationship at any time, with or without cause or notice. This offer letter does not constitute a contract of employment for any specific period. ## Acceptance To accept this offer, please sign below and return by [DEADLINE DATE]. This offer will expire if not accepted by that date. We are excited about the possibility of you joining our team. If you have any questions, please contact [HR Contact] at [email/phone]. Sincerely, --- [Hiring Manager Name] [Title] [Company Name] --- ## ACCEPTANCE I accept this offer of employment and agree to the terms stated above. Signature: ************\_************ Printed Name: ************\_************ Date: ************\_************ Anticipated Start Date: ************\_************ ``` ### Template 2: Employment Agreement (Contract Position) ```markdown # EMPLOYMENT AGREEMENT This Employment Agreement ("Agreement") is entered into as of [DATE] ("Effective Date") by and between: **Employer:** [COMPANY LEGAL NAME], a [State] [corporation/LLC] with principal offices at [Address] ("Company") **Employee:** [EMPLOYEE NAME], an individual residing at [Address] ("Employee") ## 1. EMPLOYMENT 1.1 **Position.** The Company agrees to employ Employee as [JOB TITLE], reporting to [Manager Title]. Employee accepts such employment subject to the terms of this Agreement. 1.2 **Duties.** Employee shall perform duties consistent with their position, including but not limited to: - [Primary duty 1] - [Primary duty 2] - [Primary duty 3] - Other duties as reasonably assigned 1.3 **Best Efforts.** Employee agrees to devote their full business time, attention, and best efforts to the Company's business during employment. 1.4 **Location.** Employee's primary work location shall be [Location/Remote]. [Travel requirements, if any.] ## 2. TERM 2.1 **Employment Period.** This Agreement shall commence on [START DATE] and continue until terminated as provided herein. 2.2 **At-Will Employment.** [FOR AT-WILL STATES] Notwithstanding anything herein, employment is at-will and may be terminated by either party at any time, with or without cause or notice. [OR FOR FIXED TERM:] 2.2 **Fixed Term.** This Agreement is for a fixed term of [X] months/years, ending on [END DATE], unless terminated earlier as provided herein or extended by mutual written agreement. ## 3. COMPENSATION 3.1 **Base Salary.** Employee shall receive a base salary of $[AMOUNT] per year, payable in accordance with the Company's standard payroll practices, subject to applicable withholdings. 3.2 **Bonus.** Employee may be eligible for an annual discretionary bonus of up to [X]% of base salary, based on [criteria]. Bonus payments are at Company's sole discretion and require active employment at payment date. 3.3 **Equity.** [If applicable] Subject to Board approval and the Company's equity incentive plan, Employee shall be granted [X shares/options] under the terms of a separate Stock Option Agreement. 3.4 **Benefits.** Employee shall be entitled to participate in benefit plans offered to similarly situated employees, subject to plan terms and eligibility requirements. 3.5 **Expenses.** Company shall reimburse Employee for reasonable business expenses incurred in accordance with Company policy. ## 4. CONFIDENTIALITY 4.1 **Confidential Information.** Employee acknowledges access to confidential and proprietary information including: trade secrets, business plans, customer lists, financial data, technical information, and other non-public information ("Confidential Information"). 4.2 **Non-Disclosure.** During and after employment, Employee shall not disclose, use, or permit use of any Confidential Information except as required for their duties or with prior written consent. 4.3 **Return of Materials.** Upon termination, Employee shall immediately return all Company property and Confidential Information in any form. 4.4 **Survival.** Confidentiality obligations survive termination indefinitely for trade secrets and for [3] years for other Confidential Information. ## 5. INTELLECTUAL PROPERTY 5.1 **Work Product.** All inventions, discoveries, works, and developments created by Employee during employment, relating to Company's business, or using Company resources ("Work Product") shall be Company's sole property. 5.2 **Assignment.** Employee hereby assigns to Company all rights in Work Product, including all intellectual property rights. 5.3 **Assistance.** Employee agrees to execute documents and take actions necessary to perfect Company's rights in Work Product. 5.4 **Prior Inventions.** Attached as Exhibit A is a list of any prior inventions that Employee wishes to exclude from this Agreement. ## 6. NON-COMPETITION AND NON-SOLICITATION [NOTE: Enforceability varies by jurisdiction. Consult local counsel.] 6.1 **Non-Competition.** During employment and for [12] months after termination, Employee shall not, directly or indirectly, engage in any business competitive with Company's business within [Geographic Area]. 6.2 **Non-Solicitation of Customers.** During employment and for [12] months after termination, Employee shall not solicit any customer of the Company for competing products or services. 6.3 **Non-Solicitation of Employees.** During employment and for [12] months after termination, Employee shall not recruit or solicit any Company employee to leave Company employment. ## 7. TERMINATION 7.1 **By Company for Cause.** Company may terminate immediately for Cause, defined as: (a) Material breach of this Agreement (b) Conviction of a felony (c) Fraud, dishonesty, or gross misconduct (d) Failure to perform duties after written notice and cure period 7.2 **By Company Without Cause.** Company may terminate without Cause upon [30] days written notice. 7.3 **By Employee.** Employee may terminate upon [30] days written notice. 7.4 **Severance.** [If applicable] Upon termination without Cause, Employee shall receive [X] weeks base salary as severance, contingent upon execution of a release agreement. 7.5 **Effect of Termination.** Upon termination: - All compensation earned through termination date shall be paid - Unvested equity shall be forfeited - Benefits terminate per plan terms - Sections 4, 5, 6, 8, and 9 survive termination ## 8. GENERAL PROVISIONS 8.1 **Entire Agreement.** This Agreement constitutes the entire agreement and supersedes all prior negotiations, representations, and agreements. 8.2 **Amendments.** This Agreement may be amended only by written agreement signed by both parties. 8.3 **Governing Law.** This Agreement shall be governed by the laws of [State], without regard to conflicts of law principles. 8.4 **Dispute Resolution.** [Arbitration clause or jurisdiction selection] 8.5 **Severability.** If any provision is unenforceable, it shall be modified to the minimum extent necessary, and remaining provisions shall remain in effect. 8.6 **Notices.** Notices shall be in writing and delivered to addresses above. 8.7 **Assignment.** Employee may not assign this Agreement. Company may assign to a successor. 8.8 **Waiver.** Failure to enforce any provision shall not constitute waiver. ## 9. ACKNOWLEDGMENTS Employee acknowledges: - Having read and understood this Agreement - Having opportunity to consult with counsel - Agreeing to all terms voluntarily --- IN WITNESS WHEREOF, the parties have executed this Agreement as of the Effective Date. **[COMPANY NAME]** By: ************\_************ Name: [Authorized Signatory] Title: [Title] Date: ************\_************ **EMPLOYEE** Signature: ************\_************ Name: [Employee Name] Date: ************\_************ --- ## EXHIBIT A: PRIOR INVENTIONS [Employee to list any prior inventions, if any, or write "None"] --- ``` ### Template 3: Employee Handbook Policy Section ```markdown # EMPLOYEE HANDBOOK - POLICY SECTION ## EMPLOYMENT POLICIES ### Equal Employment Opportunity [Company Name] is an equal opportunity employer. We do not discriminate based on race, color, religion, sex, sexual orientation, gender identity, national origin, age, disability, veteran status, or any other protected characteristic. This policy applies to all employment practices including: - Recruitment and hiring - Compensation and benefits - Training and development - Promotions and transfers - Termination ### Anti-Harassment Policy [Company Name] is committed to providing a workplace free from harassment. Harassment based on any protected characteristic is strictly prohibited. **Prohibited Conduct Includes:** - Unwelcome sexual advances or requests for sexual favors - Offensive comments, jokes, or slurs - Physical conduct such as assault or unwanted touching - Visual conduct such as displaying offensive images - Threatening, intimidating, or hostile acts **Reporting Procedure:** 1. Report to your manager, HR, or any member of leadership 2. Reports may be made verbally or in writing 3. Anonymous reports are accepted via [hotline/email] **Investigation:** All reports will be promptly investigated. Retaliation against anyone who reports harassment is strictly prohibited and will result in disciplinary action up to termination. ### Work Hours and Attendance **Standard Hours:** [8:00 AM - 5:00 PM, Monday through Friday] **Core Hours:** [10:00 AM - 3:00 PM] - Employees expected to be available **Flexible Work:** [Policy on remote work, flexible scheduling] **Attendance Expectations:** - Notify your manager as soon as possible if you will be absent - Excessive unexcused absences may result in disciplinary action - [x] unexcused absences in [Y] days considered excessive ### Paid Time Off (PTO) **PTO Accrual:** | Years of Service | Annual PTO Days | |------------------|-----------------| | 0-2 years | 15 days | | 3-5 years | 20 days | | 6+ years | 25 days | **PTO Guidelines:** - PTO accrues per pay period - Maximum accrual: [X] days (use it or lose it after) - Request PTO at least [2] weeks in advance - Manager approval required - PTO may not be taken during [blackout periods] ### Sick Leave - [x] days sick leave per year - May be used for personal illness or family member care - Doctor's note required for absences exceeding [3] days ### Holidays The following paid holidays are observed: - New Year's Day - Martin Luther King Jr. Day - Presidents Day - Memorial Day - Independence Day - Labor Day - Thanksgiving Day - Day after Thanksgiving - Christmas Day - [Floating holiday] ### Code of Conduct All employees are expected to: - Act with integrity and honesty - Treat colleagues, customers, and partners with respect - Protect company confidential information - Avoid conflicts of interest - Comply with all laws and regulations - Report any violations of this code **Violations may result in disciplinary action up to and including termination.** ### Technology and Communication **Acceptable Use:** - Company technology is for business purposes - Limited personal use is permitted if it doesn't interfere with work - No illegal activities or viewing inappropriate content **Monitoring:** - Company reserves the right to monitor company systems - Employees should have no expectation of privacy on company devices **Security:** - Use strong passwords and enable 2FA - Report security incidents immediately - Lock devices when unattended ### Social Media Policy **Personal Social Media:** - Clearly state opinions are your own, not the company's - Do not share confidential company information - Be respectful and professional **Company Social Media:** - Only authorized personnel may post on behalf of the company - Follow brand guidelines - Escalate negative comments to [Marketing/PR] --- ## ACKNOWLEDGMENT I acknowledge that I have received a copy of the Employee Handbook and understand that: 1. I am responsible for reading and understanding its contents 2. The handbook does not create a contract of employment 3. Policies may be changed at any time at the company's discretion 4. Employment is at-will [if applicable] I agree to abide by the policies and procedures outlined in this handbook. Employee Signature: ************\_************ Employee Name (Print): ************\_************ Date: ************\_************ ``` ## Best Practices ### Do's - **Consult legal counsel** - Employment law varies by jurisdiction - **Keep copies signed** - Document all agreements - **Update regularly** - Laws and policies change - **Be clear and specific** - Avoid ambiguity - **Train managers** - On policies and procedures ### Don'ts - **Don't use generic templates** - Customize for your jurisdiction - **Don't make promises** - That could create implied contracts - **Don't discriminate** - In language or application - **Don't forget at-will language** - Where applicable - **Don't skip review** - Have legal counsel review all documents ## Resources - [SHRM Employment Templates](https://www.shrm.org/) - [Department of Labor](https://www.dol.gov/) - [EEOC Guidance](https://www.eeoc.gov/) - State-specific labor departments
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

incident-runbook-templates

Create structured incident response runbooks with step-by-step

coding
⭐1
# Incident Runbook Templates Production-ready templates for incident response runbooks covering detection, triage, mitigation, resolution, and communication. ## When to Use This Skill - Creating incident response procedures - Building service-specific runbooks - Establishing escalation paths - Documenting recovery procedures - Responding to active incidents - Onboarding on-call engineers ## Core Concepts ### 1. Incident Severity Levels | Severity | Impact | Response Time | Example | | -------- | -------------------------- | ----------------- | ----------------------- | | **SEV1** | Complete outage, data loss | 15 min | Production down | | **SEV2** | Major degradation | 30 min | Critical feature broken | | **SEV3** | Minor impact | 2 hours | Non-critical bug | | **SEV4** | Minimal impact | Next business day | Cosmetic issue | ### 2. Runbook Structure ``` 1. Overview & Impact 2. Detection & Alerts 3. Initial Triage 4. Mitigation Steps 5. Root Cause Investigation 6. Resolution Procedures 7. Verification & Rollback 8. Communication Templates 9. Escalation Matrix ``` ## Runbook Templates ### Template 1: Service Outage Runbook ````markdown # [Service Name] Outage Runbook ## Overview **Service**: Payment Processing Service **Owner**: Platform Team **Slack**: #payments-incidents **PagerDuty**: payments-oncall ## Impact Assessment - [ ] Which customers are affected? - [ ] What percentage of traffic is impacted? - [ ] Are there financial implications? - [ ] What's the blast radius? ## Detection ### Alerts - `payment_error_rate > 5%` (PagerDuty) - `payment_latency_p99 > 2s` (Slack) - `payment_success_rate < 95%` (PagerDuty) ### Dashboards - [Payment Service Dashboard](https://grafana/d/payments) - [Error Tracking](https://sentry.io/payments) - [Dependency Status](https://status.stripe.com) ## Initial Triage (First 5 Minutes) ### 1. Assess Scope ```bash # Check service health kubectl get pods -n payments -l app=payment-service # Check recent deployments kubectl rollout history deployment/payment-service -n payments # Check error rates curl -s "http://prometheus:9090/api/v1/query?query=sum(rate(http_requests_total{status=~'5..'}[5m]))" ``` ```` ### 2. Quick Health Checks - [ ] Can you reach the service? `curl -I https://api.company.com/payments/health` - [ ] Database connectivity? Check connection pool metrics - [ ] External dependencies? Check Stripe, bank API status - [ ] Recent changes? Check deploy history ### 3. Initial Classification | Symptom | Likely Cause | Go To Section | | -------------------- | ------------------- | ------------- | | All requests failing | Service down | Section 4.1 | | High latency | Database/dependency | Section 4.2 | | Partial failures | Code bug | Section 4.3 | | Spike in errors | Traffic surge | Section 4.4 | ## Mitigation Procedures ### 4.1 Service Completely Down ```bash # Step 1: Check pod status kubectl get pods -n payments # Step 2: If pods are crash-looping, check logs kubectl logs -n payments -l app=payment-service --tail=100 # Step 3: Check recent deployments kubectl rollout history deployment/payment-service -n payments # Step 4: ROLLBACK if recent deploy is suspect kubectl rollout undo deployment/payment-service -n payments # Step 5: Scale up if resource constrained kubectl scale deployment/payment-service -n payments --replicas=10 # Step 6: Verify recovery kubectl rollout status deployment/payment-service -n payments ``` ### 4.2 High Latency ```bash # Step 1: Check database connections kubectl exec -n payments deploy/payment-service -- \ curl localhost:8080/metrics | grep db_pool # Step 2: Check slow queries (if DB issue) psql -h $DB_HOST -U $DB_USER -c " SELECT pid, now() - query_start AS duration, query FROM pg_stat_activity WHERE state = 'active' AND duration > interval '5 seconds' ORDER BY duration DESC;" # Step 3: Kill long-running queries if needed psql -h $DB_HOST -U $DB_USER -c "SELECT pg_terminate_backend(pid);" # Step 4: Check external dependency latency curl -w "@curl-format.txt" -o /dev/null -s https://api.stripe.com/v1/health # Step 5: Enable circuit breaker if dependency is slow kubectl set env deployment/payment-service \ STRIPE_CIRCUIT_BREAKER_ENABLED=true -n payments ``` ### 4.3 Partial Failures (Specific Errors) ```bash # Step 1: Identify error pattern kubectl logs -n payments -l app=payment-service --tail=500 | \ grep -i error | sort | uniq -c | sort -rn | head -20 # Step 2: Check error tracking # Go to Sentry: https://sentry.io/payments # Step 3: If specific endpoint, enable feature flag to disable curl -X POST https://api.company.com/internal/feature-flags \ -d '{"flag": "DISABLE_PROBLEMATIC_FEATURE", "enabled": true}' # Step 4: If data issue, check recent data changes psql -h $DB_HOST -c " SELECT * FROM audit_log WHERE table_name = 'payment_methods' AND created_at > now() - interval '1 hour';" ``` ### 4.4 Traffic Surge ```bash # Step 1: Check current request rate kubectl top pods -n payments # Step 2: Scale horizontally kubectl scale deployment/payment-service -n payments --replicas=20 # Step 3: Enable rate limiting kubectl set env deployment/payment-service \ RATE_LIMIT_ENABLED=true \ RATE_LIMIT_RPS=1000 -n payments # Step 4: If attack, block suspicious IPs kubectl apply -f - <<EOF apiVersion: networking.k8s.io/v1 kind: NetworkPolicy metadata: name: block-suspicious namespace: payments spec: podSelector: matchLabels: app: payment-service ingress: - from: - ipBlock: cidr: 0.0.0.0/0 except: - 192.168.1.0/24 # Suspicious range EOF ``` ## Verification Steps ```bash # Verify service is healthy curl -s https://api.company.com/payments/health | jq # Verify error rate is back to normal curl -s "http://prometheus:9090/api/v1/query?query=sum(rate(http_requests_total{status=~'5..'}[5m]))" | jq '.data.result[0].value[1]' # Verify latency is acceptable curl -s "http://prometheus:9090/api/v1/query?query=histogram_quantile(0.99,sum(rate(http_request_duration_seconds_bucket[5m]))by(le))" | jq # Smoke test critical flows ./scripts/smoke-test-payments.sh ``` ## Rollback Procedures ```bash # Rollback Kubernetes deployment kubectl rollout undo deployment/payment-service -n payments # Rollback database migration (if applicable) ./scripts/db-rollback.sh $MIGRATION_VERSION # Rollback feature flag curl -X POST https://api.company.com/internal/feature-flags \ -d '{"flag": "NEW_PAYMENT_FLOW", "enabled": false}' ``` ## Escalation Matrix | Condition | Escalate To | Contact | | ----------------------------- | ------------------- | ------------------- | | > 15 min unresolved SEV1 | Engineering Manager | @manager (Slack) | | Data breach suspected | Security Team | #security-incidents | | Financial impact > $10k | Finance + Legal | @finance-oncall | | Customer communication needed | Support Lead | @support-lead | ## Communication Templates ### Initial Notification (Internal) ``` 🚨 INCIDENT: Payment Service Degradation Severity: SEV2 Status: Investigating Impact: ~20% of payment requests failing Start Time: [TIME] Incident Commander: [NAME] Current Actions: - Investigating root cause - Scaling up service - Monitoring dashboards Updates in #payments-incidents ``` ### Status Update ``` 📊 UPDATE: Payment Service Incident Status: Mitigating Impact: Reduced to ~5% failure rate Duration: 25 minutes Actions Taken: - Rolled back deployment v2.3.4 → v2.3.3 - Scaled service from 5 → 10 replicas Next Steps: - Continuing to monitor - Root cause analysis in progress ETA to Resolution: ~15 minutes ``` ### Resolution Notification ``` ✅ RESOLVED: Payment Service Incident Duration: 45 minutes Impact: ~5,000 affected transactions Root Cause: Memory leak in v2.3.4 Resolution: - Rolled back to v2.3.3 - Transactions auto-retried successfully Follow-up: - Postmortem scheduled for [DATE] - Bug fix in progress ``` ```` ### Template 2: Database Incident Runbook ```markdown # Database Incident Runbook ## Quick Reference | Issue | Command | |-------|---------| | Check connections | `SELECT count(*) FROM pg_stat_activity;` | | Kill query | `SELECT pg_terminate_backend(pid);` | | Check replication lag | `SELECT extract(epoch from (now() - pg_last_xact_replay_timestamp()));` | | Check locks | `SELECT * FROM pg_locks WHERE NOT granted;` | ## Connection Pool Exhaustion ```sql -- Check current connections SELECT datname, usename, state, count(*) FROM pg_stat_activity GROUP BY datname, usename, state ORDER BY count(*) DESC; -- Identify long-running connections SELECT pid, usename, datname, state, query_start, query FROM pg_stat_activity WHERE state != 'idle' ORDER BY query_start; -- Terminate idle connections SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE state = 'idle' AND query_start < now() - interval '10 minutes'; ```` ## Replication Lag ```sql -- Check lag on replica SELECT CASE WHEN pg_last_wal_receive_lsn() = pg_last_wal_replay_lsn() THEN 0 ELSE extract(epoch from now() - pg_last_xact_replay_timestamp()) END AS lag_seconds; -- If lag > 60s, consider: -- 1. Check network between primary/replica -- 2. Check replica disk I/O -- 3. Consider failover if unrecoverable ``` ## Disk Space Critical ```bash # Check disk usage df -h /var/lib/postgresql/data # Find large tables psql -c "SELECT relname, pg_size_pretty(pg_total_relation_size(relid)) FROM pg_catalog.pg_statio_user_tables ORDER BY pg_total_relation_size(relid) DESC LIMIT 10;" # VACUUM to reclaim space psql -c "VACUUM FULL large_table;" # If emergency, delete old data or expand disk ``` ``` ## Best Practices ### Do's - **Keep runbooks updated** - Review after every incident - **Test runbooks regularly** - Game days, chaos engineering - **Include rollback steps** - Always have an escape hatch - **Document assumptions** - What must be true for steps to work - **Link to dashboards** - Quick access during stress ### Don'ts - **Don't assume knowledge** - Write for 3 AM brain - **Don't skip verification** - Confirm each step worked - **Don't forget communication** - Keep stakeholders informed - **Don't work alone** - Escalate early - **Don't skip postmortems** - Learn from every incident ## Resources - [Google SRE Book - Incident Management](https://sre.google/sre-book/managing-incidents/) - [PagerDuty Incident Response](https://response.pagerduty.com/) - [Atlassian Incident Management](https://www.atlassian.com/incident-management) ```
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

on-call-handoff-patterns

Master on-call shift handoffs with context transfer, escalation

coding
⭐1
# On-Call Handoff Patterns Effective patterns for on-call shift transitions, ensuring continuity, context transfer, and reliable incident response across shifts. ## When to Use This Skill - Transitioning on-call responsibilities - Writing shift handoff summaries - Documenting ongoing investigations - Establishing on-call rotation procedures - Improving handoff quality - Onboarding new on-call engineers ## Core Concepts ### 1. Handoff Components | Component | Purpose | | -------------------------- | ----------------------- | | **Active Incidents** | What's currently broken | | **Ongoing Investigations** | Issues being debugged | | **Recent Changes** | Deployments, configs | | **Known Issues** | Workarounds in place | | **Upcoming Events** | Maintenance, releases | ### 2. Handoff Timing ``` Recommended: 30 min overlap between shifts Outgoing: ├── 15 min: Write handoff document └── 15 min: Sync call with incoming Incoming: ├── 15 min: Review handoff document ├── 15 min: Sync call with outgoing └── 5 min: Verify alerting setup ``` ## Templates ### Template 1: Shift Handoff Document ````markdown # On-Call Handoff: Platform Team **Outgoing**: @alice (2024-01-15 to 2024-01-22) **Incoming**: @bob (2024-01-22 to 2024-01-29) **Handoff Time**: 2024-01-22 09:00 UTC --- ## 🔴 Active Incidents ### None currently active No active incidents at handoff time. --- ## 🟡 Ongoing Investigations ### 1. Intermittent API Timeouts (ENG-1234) **Status**: Investigating **Started**: 2024-01-20 **Impact**: ~0.1% of requests timing out **Context**: - Timeouts correlate with database backup window (02:00-03:00 UTC) - Suspect backup process causing lock contention - Added extra logging in PR #567 (deployed 01/21) **Next Steps**: - [ ] Review new logs after tonight's backup - [ ] Consider moving backup window if confirmed **Resources**: - Dashboard: [API Latency](https://grafana/d/api-latency) - Thread: #platform-eng (01/20, 14:32) --- ### 2. Memory Growth in Auth Service (ENG-1235) **Status**: Monitoring **Started**: 2024-01-18 **Impact**: None yet (proactive) **Context**: - Memory usage growing ~5% per day - No memory leak found in profiling - Suspect connection pool not releasing properly **Next Steps**: - [ ] Review heap dump from 01/21 - [ ] Consider restart if usage > 80% **Resources**: - Dashboard: [Auth Service Memory](https://grafana/d/auth-memory) - Analysis doc: [Memory Investigation](https://docs/eng-1235) --- ## 🟢 Resolved This Shift ### Payment Service Outage (2024-01-19) - **Duration**: 23 minutes - **Root Cause**: Database connection exhaustion - **Resolution**: Rolled back v2.3.4, increased pool size - **Postmortem**: [POSTMORTEM-89](https://docs/postmortem-89) - **Follow-up tickets**: ENG-1230, ENG-1231 --- ## 📋 Recent Changes ### Deployments | Service | Version | Time | Notes | | ------------ | ------- | ----------- | -------------------------- | | api-gateway | v3.2.1 | 01/21 14:00 | Bug fix for header parsing | | user-service | v2.8.0 | 01/20 10:00 | New profile features | | auth-service | v4.1.2 | 01/19 16:00 | Security patch | ### Configuration Changes - 01/21: Increased API rate limit from 1000 to 1500 RPS - 01/20: Updated database connection pool max from 50 to 75 ### Infrastructure - 01/20: Added 2 nodes to Kubernetes cluster - 01/19: Upgraded Redis from 6.2 to 7.0 --- ## ⚠️ Known Issues & Workarounds ### 1. Slow Dashboard Loading **Issue**: Grafana dashboards slow on Monday mornings **Workaround**: Wait 5 min after 08:00 UTC for cache warm-up **Ticket**: OPS-456 (P3) ### 2. Flaky Integration Test **Issue**: `test_payment_flow` fails intermittently in CI **Workaround**: Re-run failed job (usually passes on retry) **Ticket**: ENG-1200 (P2) --- ## 📅 Upcoming Events | Date | Event | Impact | Contact | | ----------- | -------------------- | ------------------- | ------------- | | 01/23 02:00 | Database maintenance | 5 min read-only | @dba-team | | 01/24 14:00 | Major release v5.0 | Monitor closely | @release-team | | 01/25 | Marketing campaign | 2x traffic expected | @platform | --- ## 📞 Escalation Reminders | Issue Type | First Escalation | Second Escalation | | --------------- | -------------------- | ----------------- | | Payment issues | @payments-oncall | @payments-manager | | Auth issues | @auth-oncall | @security-team | | Database issues | @dba-team | @infra-manager | | Unknown/severe | @engineering-manager | @vp-engineering | --- ## 🔧 Quick Reference ### Common Commands ```bash # Check service health kubectl get pods -A | grep -v Running # Recent deployments kubectl get events --sort-by='.lastTimestamp' | tail -20 # Database connections psql -c "SELECT count(*) FROM pg_stat_activity;" # Clear cache (emergency only) redis-cli FLUSHDB ``` ```` ### Important Links - [Runbooks](https://wiki/runbooks) - [Service Catalog](https://wiki/services) - [Incident Slack](https://slack.com/incidents) - [PagerDuty](https://pagerduty.com/schedules) --- ## Handoff Checklist ### Outgoing Engineer - [x] Document active incidents - [x] Document ongoing investigations - [x] List recent changes - [x] Note known issues - [x] Add upcoming events - [x] Sync with incoming engineer ### Incoming Engineer - [ ] Read this document - [ ] Join sync call - [ ] Verify PagerDuty is routing to you - [ ] Verify Slack notifications working - [ ] Check VPN/access working - [ ] Review critical dashboards ```` ### Template 2: Quick Handoff (Async) ```markdown # Quick Handoff: @alice → @bob ## TL;DR - No active incidents - 1 investigation ongoing (API timeouts, see ENG-1234) - Major release tomorrow (01/24) - be ready for issues ## Watch List 1. API latency around 02:00-03:00 UTC (backup window) 2. Auth service memory (restart if > 80%) ## Recent - Deployed api-gateway v3.2.1 yesterday (stable) - Increased rate limits to 1500 RPS ## Coming Up - 01/23 02:00 - DB maintenance (5 min read-only) - 01/24 14:00 - v5.0 release ## Questions? I'll be available on Slack until 17:00 today. ```` ### Template 3: Incident Handoff (Mid-Incident) ```markdown # INCIDENT HANDOFF: Payment Service Degradation **Incident Start**: 2024-01-22 08:15 UTC **Current Status**: Mitigating **Severity**: SEV2 --- ## Current State - Error rate: 15% (down from 40%) - Mitigation in progress: scaling up pods - ETA to resolution: ~30 min ## What We Know 1. Root cause: Memory pressure on payment-service pods 2. Triggered by: Unusual traffic spike (3x normal) 3. Contributing: Inefficient query in checkout flow ## What We've Done - Scaled payment-service from 5 → 15 pods - Enabled rate limiting on checkout endpoint - Disabled non-critical features ## What Needs to Happen 1. Monitor error rate - should reach <1% in ~15 min 2. If not improving, escalate to @payments-manager 3. Once stable, begin root cause investigation ## Key People - Incident Commander: @alice (handing off) - Comms Lead: @charlie - Technical Lead: @bob (incoming) ## Communication - Status page: Updated at 08:45 - Customer support: Notified - Exec team: Aware ## Resources - Incident channel: #inc-20240122-payment - Dashboard: [Payment Service](https://grafana/d/payments) - Runbook: [Payment Degradation](https://wiki/runbooks/payments) --- **Incoming on-call (@bob) - Please confirm you have:** - [ ] Joined #inc-20240122-payment - [ ] Access to dashboards - [ ] Understand current state - [ ] Know escalation path ``` ## Handoff Sync Meeting ### Agenda (15 minutes) ```markdown ## Handoff Sync: @alice → @bob 1. **Active Issues** (5 min) - Walk through any ongoing incidents - Discuss investigation status - Transfer context and theories 2. **Recent Changes** (3 min) - Deployments to watch - Config changes - Known regressions 3. **Upcoming Events** (3 min) - Maintenance windows - Expected traffic changes - Releases planned 4. **Questions** (4 min) - Clarify anything unclear - Confirm access and alerting - Exchange contact info ``` ## On-Call Best Practices ### Before Your Shift ```markdown ## Pre-Shift Checklist ### Access Verification - [ ] VPN working - [ ] kubectl access to all clusters - [ ] Database read access - [ ] Log aggregator access (Splunk/Datadog) - [ ] PagerDuty app installed and logged in ### Alerting Setup - [ ] PagerDuty schedule shows you as primary - [ ] Phone notifications enabled - [ ] Slack notifications for incident channels - [ ] Test alert received and acknowledged ### Knowledge Refresh - [ ] Review recent incidents (past 2 weeks) - [ ] Check service changelog - [ ] Skim critical runbooks - [ ] Know escalation contacts ### Environment Ready - [ ] Laptop charged and accessible - [ ] Phone charged - [ ] Quiet space available for calls - [ ] Secondary contact identified (if traveling) ``` ### During Your Shift ```markdown ## Daily On-Call Routine ### Morning (start of day) - [ ] Check overnight alerts - [ ] Review dashboards for anomalies - [ ] Check for any P0/P1 tickets created - [ ] Skim incident channels for context ### Throughout Day - [ ] Respond to alerts within SLA - [ ] Document investigation progress - [ ] Update team on significant issues - [ ] Triage incoming pages ### End of Day - [ ] Hand off any active issues - [ ] Update investigation docs - [ ] Note anything for next shift ``` ### After Your Shift ```markdown ## Post-Shift Checklist - [ ] Complete handoff document - [ ] Sync with incoming on-call - [ ] Verify PagerDuty routing changed - [ ] Close/update investigation tickets - [ ] File postmortems for any incidents - [ ] Take time off if shift was stressful ``` ## Escalation Guidelines ### When to Escalate ```markdown ## Escalation Triggers ### Immediate Escalation - SEV1 incident declared - Data breach suspected - Unable to diagnose within 30 min - Customer or legal escalation received ### Consider Escalation - Issue spans multiple teams - Requires expertise you don't have - Business impact exceeds threshold - You're uncertain about next steps ### How to Escalate 1. Page the appropriate escalation path 2. Provide brief context in Slack 3. Stay engaged until escalation acknowledges 4. Hand off cleanly, don't just disappear ``` ## Best Practices ### Do's - **Document everything** - Future you will thank you - **Escalate early** - Better safe than sorry - **Take breaks** - Alert fatigue is real - **Keep handoffs synchronous** - Async loses context - **Test your setup** - Before incidents, not during ### Don'ts - **Don't skip handoffs** - Context loss causes incidents - **Don't hero** - Escalate when needed - **Don't ignore alerts** - Even if they seem minor - **Don't work sick** - Swap shifts instead - **Don't disappear** - Stay reachable during shift ## Resources - [Google SRE - Being On-Call](https://sre.google/sre-book/being-on-call/) - [PagerDuty On-Call Guide](https://www.pagerduty.com/resources/learn/on-call-management/) - [Increment On-Call Issue](https://increment.com/on-call/)
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

postmortem-writing

Write effective blameless postmortems with root cause analysis,

coding
⭐1
# Postmortem Writing Comprehensive guide to writing effective, blameless postmortems that drive organizational learning and prevent incident recurrence. ## When to Use This Skill - Conducting post-incident reviews - Writing postmortem documents - Facilitating blameless postmortem meetings - Identifying root causes and contributing factors - Creating actionable follow-up items - Building organizational learning culture ## Core Concepts ### 1. Blameless Culture | Blame-Focused | Blameless | | ------------------------ | --------------------------------- | | "Who caused this?" | "What conditions allowed this?" | | "Someone made a mistake" | "The system allowed this mistake" | | Punish individuals | Improve systems | | Hide information | Share learnings | | Fear of speaking up | Psychological safety | ### 2. Postmortem Triggers - SEV1 or SEV2 incidents - Customer-facing outages > 15 minutes - Data loss or security incidents - Near-misses that could have been severe - Novel failure modes - Incidents requiring unusual intervention ## Quick Start ### Postmortem Timeline ``` Day 0: Incident occurs Day 1-2: Draft postmortem document Day 3-5: Postmortem meeting Day 5-7: Finalize document, create tickets Week 2+: Action item completion Quarterly: Review patterns across incidents ``` ## Templates ### Template 1: Standard Postmortem ```markdown # Postmortem: [Incident Title] **Date**: 2024-01-15 **Authors**: @alice, @bob **Status**: Draft | In Review | Final **Incident Severity**: SEV2 **Incident Duration**: 47 minutes ## Executive Summary On January 15, 2024, the payment processing service experienced a 47-minute outage affecting approximately 12,000 customers. The root cause was a database connection pool exhaustion triggered by a configuration change in deployment v2.3.4. The incident was resolved by rolling back to v2.3.3 and increasing connection pool limits. **Impact**: - 12,000 customers unable to complete purchases - Estimated revenue loss: $45,000 - 847 support tickets created - No data loss or security implications ## Timeline (All times UTC) | Time | Event | | ----- | ----------------------------------------------- | | 14:23 | Deployment v2.3.4 completed to production | | 14:31 | First alert: `payment_error_rate > 5%` | | 14:33 | On-call engineer @alice acknowledges alert | | 14:35 | Initial investigation begins, error rate at 23% | | 14:41 | Incident declared SEV2, @bob joins | | 14:45 | Database connection exhaustion identified | | 14:52 | Decision to rollback deployment | | 14:58 | Rollback to v2.3.3 initiated | | 15:10 | Rollback complete, error rate dropping | | 15:18 | Service fully recovered, incident resolved | ## Root Cause Analysis ### What Happened The v2.3.4 deployment included a change to the database query pattern that inadvertently removed connection pooling for a frequently-called endpoint. Each request opened a new database connection instead of reusing pooled connections. ### Why It Happened 1. **Proximate Cause**: Code change in `PaymentRepository.java` replaced pooled `DataSource` with direct `DriverManager.getConnection()` calls. 2. **Contributing Factors**: - Code review did not catch the connection handling change - No integration tests specifically for connection pool behavior - Staging environment has lower traffic, masking the issue - Database connection metrics alert threshold was too high (90%) 3. **5 Whys Analysis**: - Why did the service fail? → Database connections exhausted - Why were connections exhausted? → Each request opened new connection - Why did each request open new connection? → Code bypassed connection pool - Why did code bypass connection pool? → Developer unfamiliar with codebase patterns - Why was developer unfamiliar? → No documentation on connection management patterns ### System Diagram ``` [Client] → [Load Balancer] → [Payment Service] → [Database] ↓ Connection Pool (broken) ↓ Direct connections (cause) ``` ## Detection ### What Worked - Error rate alert fired within 8 minutes of deployment - Grafana dashboard clearly showed connection spike - On-call response was swift (2 minute acknowledgment) ### What Didn't Work - Database connection metric alert threshold too high - No deployment-correlated alerting - Canary deployment would have caught this earlier ### Detection Gap The deployment completed at 14:23, but the first alert didn't fire until 14:31 (8 minutes). A deployment-aware alert could have detected the issue faster. ## Response ### What Worked - On-call engineer quickly identified database as the issue - Rollback decision was made decisively - Clear communication in incident channel ### What Could Be Improved - Took 10 minutes to correlate issue with recent deployment - Had to manually check deployment history - Rollback took 12 minutes (could be faster) ## Impact ### Customer Impact - 12,000 unique customers affected - Average impact duration: 35 minutes - 847 support tickets (23% of affected users) - Customer satisfaction score dropped 12 points ### Business Impact - Estimated revenue loss: $45,000 - Support cost: ~$2,500 (agent time) - Engineering time: ~8 person-hours ### Technical Impact - Database primary experienced elevated load - Some replica lag during incident - No permanent damage to systems ## Lessons Learned ### What Went Well 1. Alerting detected the issue before customer reports 2. Team collaborated effectively under pressure 3. Rollback procedure worked smoothly 4. Communication was clear and timely ### What Went Wrong 1. Code review missed critical change 2. Test coverage gap for connection pooling 3. Staging environment doesn't reflect production traffic 4. Alert thresholds were not tuned properly ### Where We Got Lucky 1. Incident occurred during business hours with full team available 2. Database handled the load without failing completely 3. No other incidents occurred simultaneously ## Action Items | Priority | Action | Owner | Due Date | Ticket | |----------|--------|-------|----------|--------| | P0 | Add integration test for connection pool behavior | @alice | 2024-01-22 | ENG-1234 | | P0 | Lower database connection alert threshold to 70% | @bob | 2024-01-17 | OPS-567 | | P1 | Document connection management patterns | @alice | 2024-01-29 | DOC-89 | | P1 | Implement deployment-correlated alerting | @bob | 2024-02-05 | OPS-568 | | P2 | Evaluate canary deployment strategy | @charlie | 2024-02-15 | ENG-1235 | | P2 | Load test staging with production-like traffic | @dave | 2024-02-28 | QA-123 | ## Appendix ### Supporting Data #### Error Rate Graph [Link to Grafana dashboard snapshot] #### Database Connection Graph [Link to metrics] ### Related Incidents - 2023-11-02: Similar connection issue in User Service (POSTMORTEM-42) ### References - [Connection Pool Best Practices](internal-wiki/connection-pools) - [Deployment Runbook](internal-wiki/deployment-runbook) ``` ### Template 2: 5 Whys Analysis ```markdown # 5 Whys Analysis: [Incident] ## Problem Statement Payment service experienced 47-minute outage due to database connection exhaustion. ## Analysis ### Why #1: Why did the service fail? **Answer**: Database connections were exhausted, causing all new requests to fail. **Evidence**: Metrics showed connection count at 100/100 (max), with 500+ pending requests. --- ### Why #2: Why were database connections exhausted? **Answer**: Each incoming request opened a new database connection instead of using the connection pool. **Evidence**: Code diff shows direct `DriverManager.getConnection()` instead of pooled `DataSource`. --- ### Why #3: Why did the code bypass the connection pool? **Answer**: A developer refactored the repository class and inadvertently changed the connection acquisition method. **Evidence**: PR #1234 shows the change, made while fixing a different bug. --- ### Why #4: Why wasn't this caught in code review? **Answer**: The reviewer focused on the functional change (the bug fix) and didn't notice the infrastructure change. **Evidence**: Review comments only discuss business logic. --- ### Why #5: Why isn't there a safety net for this type of change? **Answer**: We lack automated tests that verify connection pool behavior and lack documentation about our connection patterns. **Evidence**: Test suite has no tests for connection handling; wiki has no article on database connections. ## Root Causes Identified 1. **Primary**: Missing automated tests for infrastructure behavior 2. **Secondary**: Insufficient documentation of architectural patterns 3. **Tertiary**: Code review checklist doesn't include infrastructure considerations ## Systemic Improvements | Root Cause | Improvement | Type | | ------------- | --------------------------------- | ---------- | | Missing tests | Add infrastructure behavior tests | Prevention | | Missing docs | Document connection patterns | Prevention | | Review gaps | Update review checklist | Detection | | No canary | Implement canary deployments | Mitigation | ``` ### Template 3: Quick Postmortem (Minor Incidents) ```markdown # Quick Postmortem: [Brief Title] **Date**: 2024-01-15 | **Duration**: 12 min | **Severity**: SEV3 ## What Happened API latency spiked to 5s due to cache miss storm after cache flush. ## Timeline - 10:00 - Cache flush initiated for config update - 10:02 - Latency alerts fire - 10:05 - Identified as cache miss storm - 10:08 - Enabled cache warming - 10:12 - Latency normalized ## Root Cause Full cache flush for minor config update caused thundering herd. ## Fix - Immediate: Enabled cache warming - Long-term: Implement partial cache invalidation (ENG-999) ## Lessons Don't full-flush cache in production; use targeted invalidation. ``` ## Facilitation Guide ### Running a Postmortem Meeting ```markdown ## Meeting Structure (60 minutes) ### 1. Opening (5 min) - Remind everyone of blameless culture - "We're here to learn, not to blame" - Review meeting norms ### 2. Timeline Review (15 min) - Walk through events chronologically - Ask clarifying questions - Identify gaps in timeline ### 3. Analysis Discussion (20 min) - What failed? - Why did it fail? - What conditions allowed this? - What would have prevented it? ### 4. Action Items (15 min) - Brainstorm improvements - Prioritize by impact and effort - Assign owners and due dates ### 5. Closing (5 min) - Summarize key learnings - Confirm action item owners - Schedule follow-up if needed ## Facilitation Tips - Keep discussion on track - Redirect blame to systems - Encourage quiet participants - Document dissenting views - Time-box tangents ``` ## Anti-Patterns to Avoid | Anti-Pattern | Problem | Better Approach | | ----------------------- | -------------------------- | ------------------------------- | | **Blame game** | Shuts down learning | Focus on systems | | **Shallow analysis** | Doesn't prevent recurrence | Ask "why" 5 times | | **No action items** | Waste of time | Always have concrete next steps | | **Unrealistic actions** | Never completed | Scope to achievable tasks | | **No follow-up** | Actions forgotten | Track in ticketing system | ## Best Practices ### Do's - **Start immediately** - Memory fades fast - **Be specific** - Exact times, exact errors - **Include graphs** - Visual evidence - **Assign owners** - No orphan action items - **Share widely** - Organizational learning ### Don'ts - **Don't name and shame** - Ever - **Don't skip small incidents** - They reveal patterns - **Don't make it a blame doc** - That kills learning - **Don't create busywork** - Actions should be meaningful - **Don't skip follow-up** - Verify actions completed ## Resources - [Google SRE - Postmortem Culture](https://sre.google/sre-book/postmortem-culture/) - [Etsy's Blameless Postmortems](https://codeascraft.com/2012/05/22/blameless-postmortems/) - [PagerDuty Postmortem Guide](https://postmortems.pagerduty.com/)
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

prompt-engineering-patterns

Master advanced prompt engineering techniques to maximize LLM

coding
⭐1
# Prompt Engineering Patterns Master advanced prompt engineering techniques to maximize LLM performance, reliability, and controllability. ## When to Use This Skill - Designing complex prompts for production LLM applications - Optimizing prompt performance and consistency - Implementing structured reasoning patterns (chain-of-thought, tree-of-thought) - Building few-shot learning systems with dynamic example selection - Creating reusable prompt templates with variable interpolation - Debugging and refining prompts that produce inconsistent outputs - Implementing system prompts for specialized AI assistants - Using structured outputs (JSON mode) for reliable parsing ## Core Capabilities ### 1. Few-Shot Learning - Example selection strategies (semantic similarity, diversity sampling) - Balancing example count with context window constraints - Constructing effective demonstrations with input-output pairs - Dynamic example retrieval from knowledge bases - Handling edge cases through strategic example selection ### 2. Chain-of-Thought Prompting - Step-by-step reasoning elicitation - Zero-shot CoT with "Let's think step by step" - Few-shot CoT with reasoning traces - Self-consistency techniques (sampling multiple reasoning paths) - Verification and validation steps ### 3. Structured Outputs - JSON mode for reliable parsing - Pydantic schema enforcement - Type-safe response handling - Error handling for malformed outputs ### 4. Prompt Optimization - Iterative refinement workflows - A/B testing prompt variations - Measuring prompt performance metrics (accuracy, consistency, latency) - Reducing token usage while maintaining quality - Handling edge cases and failure modes ### 5. Template Systems - Variable interpolation and formatting - Conditional prompt sections - Multi-turn conversation templates - Role-based prompt composition - Modular prompt components ### 6. System Prompt Design - Setting model behavior and constraints - Defining output formats and structure - Establishing role and expertise - Safety guidelines and content policies - Context setting and background information ## Quick Start ```python from langchain_anthropic import ChatAnthropic from langchain_core.prompts import ChatPromptTemplate from pydantic import BaseModel, Field # Define structured output schema class SQLQuery(BaseModel): query: str = Field(description="The SQL query") explanation: str = Field(description="Brief explanation of what the query does") tables_used: list[str] = Field(description="List of tables referenced") # Initialize model with structured output llm = ChatAnthropic(model="claude-sonnet-4-6") structured_llm = llm.with_structured_output(SQLQuery) # Create prompt template prompt = ChatPromptTemplate.from_messages([ ("system", """You are an expert SQL developer. Generate efficient, secure SQL queries. Always use parameterized queries to prevent SQL injection. Explain your reasoning briefly."""), ("user", "Convert this to SQL: {query}") ]) # Create chain chain = prompt | structured_llm # Use result = await chain.ainvoke({ "query": "Find all users who registered in the last 30 days" }) print(result.query) print(result.explanation) ``` ## Key Patterns ### Pattern 1: Structured Output with Pydantic ```python from anthropic import Anthropic from pydantic import BaseModel, Field from typing import Literal import json class SentimentAnalysis(BaseModel): sentiment: Literal["positive", "negative", "neutral"] confidence: float = Field(ge=0, le=1) key_phrases: list[str] reasoning: str async def analyze_sentiment(text: str) -> SentimentAnalysis: """Analyze sentiment with structured output.""" client = Anthropic() message = client.messages.create( model="claude-sonnet-4-6", max_tokens=500, messages=[{ "role": "user", "content": f"""Analyze the sentiment of this text. Text: {text} Respond with JSON matching this schema: {{ "sentiment": "positive" | "negative" | "neutral", "confidence": 0.0-1.0, "key_phrases": ["phrase1", "phrase2"], "reasoning": "brief explanation" }}""" }] ) return SentimentAnalysis(**json.loads(message.content[0].text)) ``` ### Pattern 2: Chain-of-Thought with Self-Verification ```python from langchain_core.prompts import ChatPromptTemplate cot_prompt = ChatPromptTemplate.from_template(""" Solve this problem step by step. Problem: {problem} Instructions: 1. Break down the problem into clear steps 2. Work through each step showing your reasoning 3. State your final answer 4. Verify your answer by checking it against the original problem Format your response as: ## Steps [Your step-by-step reasoning] ## Answer [Your final answer] ## Verification [Check that your answer is correct] """) ``` ### Pattern 3: Few-Shot with Dynamic Example Selection ```python from langchain_voyageai import VoyageAIEmbeddings from langchain_core.example_selectors import SemanticSimilarityExampleSelector from langchain_chroma import Chroma # Create example selector with semantic similarity example_selector = SemanticSimilarityExampleSelector.from_examples( examples=[ {"input": "How do I reset my password?", "output": "Go to Settings > Security > Reset Password"}, {"input": "Where can I see my order history?", "output": "Navigate to Account > Orders"}, {"input": "How do I contact support?", "output": "Click Help > Contact Us or email support@example.com"}, ], embeddings=VoyageAIEmbeddings(model="voyage-3-large"), vectorstore_cls=Chroma, k=2 # Select 2 most similar examples ) async def get_few_shot_prompt(query: str) -> str: """Build prompt with dynamically selected examples.""" examples = await example_selector.aselect_examples({"input": query}) examples_text = "\n".join( f"User: {ex['input']}\nAssistant: {ex['output']}" for ex in examples ) return f"""You are a helpful customer support assistant. Here are some example interactions: {examples_text} Now respond to this query: User: {query} Assistant:""" ``` ### Pattern 4: Progressive Disclosure Start with simple prompts, add complexity only when needed: ```python PROMPT_LEVELS = { # Level 1: Direct instruction "simple": "Summarize this article: {text}", # Level 2: Add constraints "constrained": """Summarize this article in 3 bullet points, focusing on: - Key findings - Main conclusions - Practical implications Article: {text}""", # Level 3: Add reasoning "reasoning": """Read this article carefully. 1. First, identify the main topic and thesis 2. Then, extract the key supporting points 3. Finally, summarize in 3 bullet points Article: {text} Summary:""", # Level 4: Add examples "few_shot": """Read articles and provide concise summaries. Example: Article: "New research shows that regular exercise can reduce anxiety by up to 40%..." Summary: • Regular exercise reduces anxiety by up to 40% • 30 minutes of moderate activity 3x/week is sufficient • Benefits appear within 2 weeks of starting Now summarize this article: Article: {text} Summary:""" } ``` ### Pattern 5: Error Recovery and Fallback ```python from pydantic import BaseModel, ValidationError import json class ResponseWithConfidence(BaseModel): answer: str confidence: float sources: list[str] alternative_interpretations: list[str] = [] ERROR_RECOVERY_PROMPT = """ Answer the question based on the context provided. Context: {context} Question: {question} Instructions: 1. If you can answer confidently (>0.8), provide a direct answer 2. If you're somewhat confident (0.5-0.8), provide your best answer with caveats 3. If you're uncertain (<0.5), explain what information is missing 4. Always provide alternative interpretations if the question is ambiguous Respond in JSON: {{ "answer": "your answer or 'I cannot determine this from the context'", "confidence": 0.0-1.0, "sources": ["relevant context excerpts"], "alternative_interpretations": ["if question is ambiguous"] }} """ async def answer_with_fallback( context: str, question: str, llm ) -> ResponseWithConfidence: """Answer with error recovery and fallback.""" prompt = ERROR_RECOVERY_PROMPT.format(context=context, question=question) try: response = await llm.ainvoke(prompt) return ResponseWithConfidence(**json.loads(response.content)) except (json.JSONDecodeError, ValidationError) as e: # Fallback: try to extract answer without structure simple_prompt = f"Based on: {context}\n\nAnswer: {question}" simple_response = await llm.ainvoke(simple_prompt) return ResponseWithConfidence( answer=simple_response.content, confidence=0.5, sources=["fallback extraction"], alternative_interpretations=[] ) ``` ### Pattern 6: Role-Based System Prompts ```python SYSTEM_PROMPTS = { "analyst": """You are a senior data analyst with expertise in SQL, Python, and business intelligence. Your responsibilities: - Write efficient, well-documented queries - Explain your analysis methodology - Highlight key insights and recommendations - Flag any data quality concerns Communication style: - Be precise and technical when discussing methodology - Translate technical findings into business impact - Use clear visualizations when helpful""", "assistant": """You are a helpful AI assistant focused on accuracy and clarity. Core principles: - Always cite sources when making factual claims - Acknowledge uncertainty rather than guessing - Ask clarifying questions when the request is ambiguous - Provide step-by-step explanations for complex topics Constraints: - Do not provide medical, legal, or financial advice - Redirect harmful requests appropriately - Protect user privacy""", "code_reviewer": """You are a senior software engineer conducting code reviews. Review criteria: - Correctness: Does the code work as intended? - Security: Are there any vulnerabilities? - Performance: Are there efficiency concerns? - Maintainability: Is the code readable and well-structured? - Best practices: Does it follow language idioms? Output format: 1. Summary assessment (approve/request changes) 2. Critical issues (must fix) 3. Suggestions (nice to have) 4. Positive feedback (what's done well)""" } ``` ## Integration Patterns ### With RAG Systems ```python RAG_PROMPT = """You are a knowledgeable assistant that answers questions based on provided context. Context (retrieved from knowledge base): {context} Instructions: 1. Answer ONLY based on the provided context 2. If the context doesn't contain the answer, say "I don't have information about that in my knowledge base" 3. Cite specific passages using [1], [2] notation 4. If the question is ambiguous, ask for clarification Question: {question} Answer:""" ``` ### With Validation and Verification ```python VALIDATED_PROMPT = """Complete the following task: Task: {task} After generating your response, verify it meets ALL these criteria: ✓ Directly addresses the original request ✓ Contains no factual errors ✓ Is appropriately detailed (not too brief, not too verbose) ✓ Uses proper formatting ✓ Is safe and appropriate If verification fails on any criterion, revise before responding. Response:""" ``` ## Performance Optimization ### Token Efficiency ```python # Before: Verbose prompt (150+ tokens) verbose_prompt = """ I would like you to please take the following text and provide me with a comprehensive summary of the main points. The summary should capture the key ideas and important details while being concise and easy to understand. """ # After: Concise prompt (30 tokens) concise_prompt = """Summarize the key points concisely: {text} Summary:""" ``` ### Caching Common Prefixes ```python from anthropic import Anthropic client = Anthropic() # Use prompt caching for repeated system prompts response = client.messages.create( model="claude-sonnet-4-6", max_tokens=1000, system=[ { "type": "text", "text": LONG_SYSTEM_PROMPT, "cache_control": {"type": "ephemeral"} } ], messages=[{"role": "user", "content": user_query}] ) ``` ## Best Practices 1. **Be Specific**: Vague prompts produce inconsistent results 2. **Show, Don't Tell**: Examples are more effective than descriptions 3. **Use Structured Outputs**: Enforce schemas with Pydantic for reliability 4. **Test Extensively**: Evaluate on diverse, representative inputs 5. **Iterate Rapidly**: Small changes can have large impacts 6. **Monitor Performance**: Track metrics in production 7. **Version Control**: Treat prompts as code with proper versioning 8. **Document Intent**: Explain why prompts are structured as they are ## Common Pitfalls - **Over-engineering**: Starting with complex prompts before trying simple ones - **Example pollution**: Using examples that don't match the target task - **Context overflow**: Exceeding token limits with excessive examples - **Ambiguous instructions**: Leaving room for multiple interpretations - **Ignoring edge cases**: Not testing on unusual or boundary inputs - **No error handling**: Assuming outputs will always be well-formed - **Hardcoded values**: Not parameterizing prompts for reuse ## Success Metrics Track these KPIs for your prompts: - **Accuracy**: Correctness of outputs - **Consistency**: Reproducibility across similar inputs - **Latency**: Response time (P50, P95, P99) - **Token Usage**: Average tokens per request - **Success Rate**: Percentage of valid, parseable outputs - **User Satisfaction**: Ratings and feedback ## Resources - [Anthropic Prompt Engineering Guide](https://docs.anthropic.com/en/docs/build-with-claude/prompt-engineering) - [Claude Prompt Caching](https://docs.anthropic.com/en/docs/build-with-claude/prompt-caching) - [OpenAI Prompt Engineering](https://platform.openai.com/docs/guides/prompt-engineering) - [LangChain Prompts](https://python.langchain.com/docs/concepts/prompts/)
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

billing-automation

Build automated billing systems for recurring payments, invoicing,

business
⭐1
# Billing Automation Master automated billing systems including recurring billing, invoice generation, dunning management, proration, and tax calculation. ## When to Use This Skill - Implementing SaaS subscription billing - Automating invoice generation and delivery - Managing failed payment recovery (dunning) - Calculating prorated charges for plan changes - Handling sales tax, VAT, and GST - Processing usage-based billing - Managing billing cycles and renewals ## Core Concepts ### 1. Billing Cycles **Common Intervals:** - Monthly (most common for SaaS) - Annual (discounted long-term) - Quarterly - Weekly - Custom (usage-based, per-seat) ### 2. Subscription States ``` trial → active → past_due → canceled → paused → resumed ``` ### 3. Dunning Management Automated process to recover failed payments through: - Retry schedules - Customer notifications - Grace periods - Account restrictions ### 4. Proration Adjusting charges when: - Upgrading/downgrading mid-cycle - Adding/removing seats - Changing billing frequency ## Quick Start ```python from billing import BillingEngine, Subscription # Initialize billing engine billing = BillingEngine() # Create subscription subscription = billing.create_subscription( customer_id="cus_123", plan_id="plan_pro_monthly", billing_cycle_anchor=datetime.now(), trial_days=14 ) # Process billing cycle billing.process_billing_cycle(subscription.id) ``` ## Subscription Lifecycle Management ```python from datetime import datetime, timedelta from enum import Enum class SubscriptionStatus(Enum): TRIAL = "trial" ACTIVE = "active" PAST_DUE = "past_due" CANCELED = "canceled" PAUSED = "paused" class Subscription: def __init__(self, customer_id, plan, billing_cycle_day=None): self.id = generate_id() self.customer_id = customer_id self.plan = plan self.status = SubscriptionStatus.TRIAL self.current_period_start = datetime.now() self.current_period_end = self.current_period_start + timedelta(days=plan.trial_days or 30) self.billing_cycle_day = billing_cycle_day or self.current_period_start.day self.trial_end = datetime.now() + timedelta(days=plan.trial_days) if plan.trial_days else None def start_trial(self, trial_days): """Start trial period.""" self.status = SubscriptionStatus.TRIAL self.trial_end = datetime.now() + timedelta(days=trial_days) self.current_period_end = self.trial_end def activate(self): """Activate subscription after trial or immediately.""" self.status = SubscriptionStatus.ACTIVE self.current_period_start = datetime.now() self.current_period_end = self.calculate_next_billing_date() def mark_past_due(self): """Mark subscription as past due after failed payment.""" self.status = SubscriptionStatus.PAST_DUE # Trigger dunning workflow def cancel(self, at_period_end=True): """Cancel subscription.""" if at_period_end: self.cancel_at_period_end = True # Will cancel when current period ends else: self.status = SubscriptionStatus.CANCELED self.canceled_at = datetime.now() def calculate_next_billing_date(self): """Calculate next billing date based on interval.""" if self.plan.interval == 'month': return self.current_period_start + timedelta(days=30) elif self.plan.interval == 'year': return self.current_period_start + timedelta(days=365) elif self.plan.interval == 'week': return self.current_period_start + timedelta(days=7) ``` ## Billing Cycle Processing ```python class BillingEngine: def process_billing_cycle(self, subscription_id): """Process billing for a subscription.""" subscription = self.get_subscription(subscription_id) # Check if billing is due if datetime.now() < subscription.current_period_end: return # Generate invoice invoice = self.generate_invoice(subscription) # Attempt payment payment_result = self.charge_customer( subscription.customer_id, invoice.total ) if payment_result.success: # Payment successful invoice.mark_paid() subscription.advance_billing_period() self.send_invoice(invoice) else: # Payment failed subscription.mark_past_due() self.start_dunning_process(subscription, invoice) def generate_invoice(self, subscription): """Generate invoice for billing period.""" invoice = Invoice( customer_id=subscription.customer_id, subscription_id=subscription.id, period_start=subscription.current_period_start, period_end=subscription.current_period_end ) # Add subscription line item invoice.add_line_item( description=subscription.plan.name, amount=subscription.plan.amount, quantity=subscription.quantity or 1 ) # Add usage-based charges if applicable if subscription.has_usage_billing: usage_charges = self.calculate_usage_charges(subscription) invoice.add_line_item( description="Usage charges", amount=usage_charges ) # Calculate tax tax = self.calculate_tax(invoice.subtotal, subscription.customer) invoice.tax = tax invoice.finalize() return invoice def charge_customer(self, customer_id, amount): """Charge customer using saved payment method.""" customer = self.get_customer(customer_id) try: # Charge using payment processor charge = stripe.Charge.create( customer=customer.stripe_id, amount=int(amount * 100), # Convert to cents currency='usd' ) return PaymentResult(success=True, transaction_id=charge.id) except stripe.error.CardError as e: return PaymentResult(success=False, error=str(e)) ``` ## Dunning Management ```python class DunningManager: """Manage failed payment recovery.""" def __init__(self): self.retry_schedule = [ {'days': 3, 'email_template': 'payment_failed_first'}, {'days': 7, 'email_template': 'payment_failed_reminder'}, {'days': 14, 'email_template': 'payment_failed_final'} ] def start_dunning_process(self, subscription, invoice): """Start dunning process for failed payment.""" dunning_attempt = DunningAttempt( subscription_id=subscription.id, invoice_id=invoice.id, attempt_number=1, next_retry=datetime.now() + timedelta(days=3) ) # Send initial failure notification self.send_dunning_email(subscription, 'payment_failed_first') # Schedule retries self.schedule_retries(dunning_attempt) def retry_payment(self, dunning_attempt): """Retry failed payment.""" subscription = self.get_subscription(dunning_attempt.subscription_id) invoice = self.get_invoice(dunning_attempt.invoice_id) # Attempt payment again result = self.charge_customer(subscription.customer_id, invoice.total) if result.success: # Payment succeeded invoice.mark_paid() subscription.status = SubscriptionStatus.ACTIVE self.send_dunning_email(subscription, 'payment_recovered') dunning_attempt.mark_resolved() else: # Still failing dunning_attempt.attempt_number += 1 if dunning_attempt.attempt_number < len(self.retry_schedule): # Schedule next retry next_retry_config = self.retry_schedule[dunning_attempt.attempt_number] dunning_attempt.next_retry = datetime.now() + timedelta(days=next_retry_config['days']) self.send_dunning_email(subscription, next_retry_config['email_template']) else: # Exhausted retries, cancel subscription subscription.cancel(at_period_end=False) self.send_dunning_email(subscription, 'subscription_canceled') def send_dunning_email(self, subscription, template): """Send dunning notification to customer.""" customer = self.get_customer(subscription.customer_id) email_content = self.render_template(template, { 'customer_name': customer.name, 'amount_due': subscription.plan.amount, 'update_payment_url': f"https://app.example.com/billing" }) send_email( to=customer.email, subject=email_content['subject'], body=email_content['body'] ) ``` ## Proration ```python class ProrationCalculator: """Calculate prorated charges for plan changes.""" @staticmethod def calculate_proration(old_plan, new_plan, period_start, period_end, change_date): """Calculate proration for plan change.""" # Days in current period total_days = (period_end - period_start).days # Days used on old plan days_used = (change_date - period_start).days # Days remaining on new plan days_remaining = (period_end - change_date).days # Calculate prorated amounts unused_amount = (old_plan.amount / total_days) * days_remaining new_plan_amount = (new_plan.amount / total_days) * days_remaining # Net charge/credit proration = new_plan_amount - unused_amount return { 'old_plan_credit': -unused_amount, 'new_plan_charge': new_plan_amount, 'net_proration': proration, 'days_used': days_used, 'days_remaining': days_remaining } @staticmethod def calculate_seat_proration(current_seats, new_seats, price_per_seat, period_start, period_end, change_date): """Calculate proration for seat changes.""" total_days = (period_end - period_start).days days_remaining = (period_end - change_date).days # Additional seats charge additional_seats = new_seats - current_seats prorated_amount = (additional_seats * price_per_seat / total_days) * days_remaining return { 'additional_seats': additional_seats, 'prorated_charge': max(0, prorated_amount), # No refund for removing seats mid-cycle 'effective_date': change_date } ``` ## Tax Calculation ```python class TaxCalculator: """Calculate sales tax, VAT, GST.""" def __init__(self): # Tax rates by region self.tax_rates = { 'US_CA': 0.0725, # California sales tax 'US_NY': 0.04, # New York sales tax 'GB': 0.20, # UK VAT 'DE': 0.19, # Germany VAT 'FR': 0.20, # France VAT 'AU': 0.10, # Australia GST } def calculate_tax(self, amount, customer): """Calculate applicable tax.""" # Determine tax jurisdiction jurisdiction = self.get_tax_jurisdiction(customer) if not jurisdiction: return 0 # Get tax rate tax_rate = self.tax_rates.get(jurisdiction, 0) # Calculate tax tax = amount * tax_rate return { 'tax_amount': tax, 'tax_rate': tax_rate, 'jurisdiction': jurisdiction, 'tax_type': self.get_tax_type(jurisdiction) } def get_tax_jurisdiction(self, customer): """Determine tax jurisdiction based on customer location.""" if customer.country == 'US': # US: Tax based on customer state return f"US_{customer.state}" elif customer.country in ['GB', 'DE', 'FR']: # EU: VAT return customer.country elif customer.country == 'AU': # Australia: GST return 'AU' else: return None def get_tax_type(self, jurisdiction): """Get type of tax for jurisdiction.""" if jurisdiction.startswith('US_'): return 'Sales Tax' elif jurisdiction in ['GB', 'DE', 'FR']: return 'VAT' elif jurisdiction == 'AU': return 'GST' return 'Tax' def validate_vat_number(self, vat_number, country): """Validate EU VAT number.""" # Use VIES API for validation # Returns True if valid, False otherwise pass ``` ## Invoice Generation ```python class Invoice: def __init__(self, customer_id, subscription_id=None): self.id = generate_invoice_number() self.customer_id = customer_id self.subscription_id = subscription_id self.status = 'draft' self.line_items = [] self.subtotal = 0 self.tax = 0 self.total = 0 self.created_at = datetime.now() def add_line_item(self, description, amount, quantity=1): """Add line item to invoice.""" line_item = { 'description': description, 'unit_amount': amount, 'quantity': quantity, 'total': amount * quantity } self.line_items.append(line_item) self.subtotal += line_item['total'] def finalize(self): """Finalize invoice and calculate total.""" self.total = self.subtotal + self.tax self.status = 'open' self.finalized_at = datetime.now() def mark_paid(self): """Mark invoice as paid.""" self.status = 'paid' self.paid_at = datetime.now() def to_pdf(self): """Generate PDF invoice.""" from reportlab.pdfgen import canvas # Generate PDF # Include: company info, customer info, line items, tax, total pass def to_html(self): """Generate HTML invoice.""" template = """ <!DOCTYPE html> <html> <head><title>Invoice #{invoice_number}</title></head> <body> <h1>Invoice #{invoice_number}</h1> <p>Date: {date}</p> <h2>Bill To:</h2> <p>{customer_name}<br>{customer_address}</p> <table> <tr><th>Description</th><th>Quantity</th><th>Amount</th></tr> {line_items} </table> <p>Subtotal: ${subtotal}</p> <p>Tax: ${tax}</p> <h3>Total: ${total}</h3> </body> </html> """ return template.format( invoice_number=self.id, date=self.created_at.strftime('%Y-%m-%d'), customer_name=self.customer.name, customer_address=self.customer.address, line_items=self.render_line_items(), subtotal=self.subtotal, tax=self.tax, total=self.total ) ``` ## Usage-Based Billing ```python class UsageBillingEngine: """Track and bill for usage.""" def track_usage(self, customer_id, metric, quantity): """Track usage event.""" UsageRecord.create( customer_id=customer_id, metric=metric, quantity=quantity, timestamp=datetime.now() ) def calculate_usage_charges(self, subscription, period_start, period_end): """Calculate charges for usage in billing period.""" usage_records = UsageRecord.get_for_period( subscription.customer_id, period_start, period_end ) total_usage = sum(record.quantity for record in usage_records) # Tiered pricing if subscription.plan.pricing_model == 'tiered': charge = self.calculate_tiered_pricing(total_usage, subscription.plan.tiers) # Per-unit pricing elif subscription.plan.pricing_model == 'per_unit': charge = total_usage * subscription.plan.unit_price # Volume pricing elif subscription.plan.pricing_model == 'volume': charge = self.calculate_volume_pricing(total_usage, subscription.plan.tiers) return charge def calculate_tiered_pricing(self, total_usage, tiers): """Calculate cost using tiered pricing.""" charge = 0 remaining = total_usage for tier in sorted(tiers, key=lambda x: x['up_to']): tier_usage = min(remaining, tier['up_to'] - tier['from']) charge += tier_usage * tier['unit_price'] remaining -= tier_usage if remaining <= 0: break return charge ``` ## Resources - **references/billing-cycles.md**: Billing cycle management - **references/dunning-management.md**: Failed payment recovery - **references/proration.md**: Prorated charge calculations - **references/tax-calculation.md**: Tax/VAT/GST handling - **references/invoice-lifecycle.md**: Invoice state management - **assets/billing-state-machine.yaml**: Billing workflow - **assets/invoice-template.html**: Invoice templates - **assets/dunning-policy.yaml**: Dunning configuration ## Best Practices 1. **Automate Everything**: Minimize manual intervention 2. **Clear Communication**: Notify customers of billing events 3. **Flexible Retry Logic**: Balance recovery with customer experience 4. **Accurate Proration**: Fair calculation for plan changes 5. **Tax Compliance**: Calculate correct tax for jurisdiction 6. **Audit Trail**: Log all billing events 7. **Graceful Degradation**: Handle edge cases without breaking ## Common Pitfalls - **Incorrect Proration**: Not accounting for partial periods - **Missing Tax**: Forgetting to add tax to invoices - **Aggressive Dunning**: Canceling too quickly - **No Notifications**: Not informing customers of failures - **Hardcoded Cycles**: Not supporting custom billing dates
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

stripe-integration

Implement Stripe payment processing for robust, PCI-compliant

business
⭐1
# Stripe Integration Master Stripe payment processing integration for robust, PCI-compliant payment flows including checkout, subscriptions, webhooks, and refunds. ## When to Use This Skill - Implementing payment processing in web/mobile applications - Setting up subscription billing systems - Handling one-time payments and recurring charges - Processing refunds and disputes - Managing customer payment methods - Implementing SCA (Strong Customer Authentication) for European payments - Building marketplace payment flows with Stripe Connect ## Core Concepts ### 1. Payment Flows **Checkout Sessions** - Recommended for most integrations - Supports all UI paths: - Stripe-hosted checkout page - Embedded checkout form - Custom UI with Elements (Payment Element, Express Checkout Element) using `ui_mode='custom'` - Provides built-in checkout capabilities (line items, discounts, tax, shipping, address collection, saved payment methods, and checkout lifecycle events) - Lower integration and maintenance burden than Payment Intents **Payment Intents (Bespoke control)** - You calculate the final amount with taxes, discounts, subscriptions, and currency conversion yourself. - More complex implementation and long-term maintenance burden - Requires Stripe.js for PCI compliance **Setup Intents (Save Payment Methods)** - Collect payment method without charging - Used for subscriptions and future payments - Requires customer confirmation ### 2. Webhooks **Critical Events:** - `payment_intent.succeeded`: Payment completed - `payment_intent.payment_failed`: Payment failed - `customer.subscription.updated`: Subscription changed - `customer.subscription.deleted`: Subscription canceled - `charge.refunded`: Refund processed - `invoice.payment_succeeded`: Subscription payment successful ### 3. Subscriptions **Components:** - **Product**: What you're selling - **Price**: How much and how often - **Subscription**: Customer's recurring payment - **Invoice**: Generated for each billing cycle ### 4. Customer Management - Create and manage customer records - Store multiple payment methods - Track customer metadata - Manage billing details ## Quick Start ```python import stripe stripe.api_key = "sk_test_..." # Create a checkout session session = stripe.checkout.Session.create( line_items=[{ 'price_data': { 'currency': 'usd', 'product_data': { 'name': 'Premium Subscription', }, 'unit_amount': 2000, # $20.00 'recurring': { 'interval': 'month', }, }, 'quantity': 1, }], mode='subscription', success_url='https://yourdomain.com/success?session_id={CHECKOUT_SESSION_ID}', cancel_url='https://yourdomain.com/cancel' ) # Redirect user to session.url print(session.url) ``` ## Payment Implementation Patterns ### Pattern 1: One-Time Payment (Hosted Checkout) ```python def create_checkout_session(amount, currency='usd'): """Create a one-time payment checkout session.""" try: session = stripe.checkout.Session.create( line_items=[{ 'price_data': { 'currency': currency, 'product_data': { 'name': 'Blue T-shirt', 'images': ['https://example.com/product.jpg'], }, 'unit_amount': amount, # Amount in cents }, 'quantity': 1, }], mode='payment', success_url='https://yourdomain.com/success?session_id={CHECKOUT_SESSION_ID}', cancel_url='https://yourdomain.com/cancel', metadata={ 'order_id': 'order_123', 'user_id': 'user_456' } ) return session except stripe.error.StripeError as e: # Handle error print(f"Stripe error: {e.user_message}") raise ``` ### Pattern 2: Elements with Checkout Sessions ```python def create_checkout_session_for_elements(amount, currency='usd'): """Create a checkout session configured for Payment Element.""" session = stripe.checkout.Session.create( mode='payment', ui_mode='custom', line_items=[{ 'price_data': { 'currency': currency, 'product_data': {'name': 'Blue T-shirt'}, 'unit_amount': amount, }, 'quantity': 1, }], return_url='https://yourdomain.com/complete?session_id={CHECKOUT_SESSION_ID}' ) return session.client_secret # Send to frontend ``` ```javascript const stripe = Stripe("pk_test_..."); const appearance = { theme: "stripe" }; const checkout = stripe.initCheckout({ clientSecret, elementsOptions: { appearance }, }); const loadActionsResult = await checkout.loadActions(); if (loadActionsResult.type === "success") { const { actions } = loadActionsResult; const session = actions.getSession(); const button = document.getElementById("pay-button"); const checkoutContainer = document.getElementById("checkout-container"); const emailInput = document.getElementById("email"); const emailErrors = document.getElementById("email-errors"); const errors = document.getElementById("confirm-errors"); // Display a formatted string representing the total amount checkoutContainer.append(`Total: ${session.total.total.amount}`); // Mount Payment Element const paymentElement = checkout.createPaymentElement(); paymentElement.mount("#payment-element"); // Store email for submission emailInput.addEventListener("blur", () => { actions.updateEmail(emailInput.value).then((result) => { if (result.error) emailErrors.textContent = result.error.message; }); }); // Handle form submission button.addEventListener("click", () => { actions.confirm().then((result) => { if (result.type === "error") errors.textContent = result.error.message; }); }); } ``` ### Pattern 3: Elements with Payment Intents Pattern 2 (Elements with Checkout Sessions) is Stripe's recommended approach, but you can also use Payment Intents as an alternative. ```python def create_payment_intent(amount, currency='usd', customer_id=None): """Create a payment intent for bespoke checkout UI with Payment Element.""" intent = stripe.PaymentIntent.create( amount=amount, currency=currency, customer=customer_id, automatic_payment_methods={ 'enabled': True, }, metadata={ 'integration_check': 'accept_a_payment' } ) return intent.client_secret # Send to frontend ``` ```javascript // Mount Payment Element and confirm via Payment Intents const stripe = Stripe("pk_test_..."); const appearance = { theme: "stripe" }; const elements = stripe.elements({ appearance, clientSecret }); const paymentElement = elements.create("payment"); paymentElement.mount("#payment-element"); document.getElementById("pay-button").addEventListener("click", async () => { const { error } = await stripe.confirmPayment({ elements, confirmParams: { return_url: "https://yourdomain.com/complete", }, }); if (error) { document.getElementById("errors").textContent = error.message; } }); ``` ### Pattern 4: Subscription Creation ```python def create_subscription(customer_id, price_id): """Create a subscription for a customer.""" try: subscription = stripe.Subscription.create( customer=customer_id, items=[{'price': price_id}], payment_behavior='default_incomplete', payment_settings={'save_default_payment_method': 'on_subscription'}, expand=['latest_invoice.payment_intent'], ) return { 'subscription_id': subscription.id, 'client_secret': subscription.latest_invoice.payment_intent.client_secret } except stripe.error.StripeError as e: print(f"Subscription creation failed: {e}") raise ``` ### Pattern 5: Customer Portal ```python def create_customer_portal_session(customer_id): """Create a portal session for customers to manage subscriptions.""" session = stripe.billing_portal.Session.create( customer=customer_id, return_url='https://yourdomain.com/account', ) return session.url # Redirect customer here ``` ## Webhook Handling ### Secure Webhook Endpoint ```python from flask import Flask, request import stripe app = Flask(__name__) endpoint_secret = 'whsec_...' @app.route('/webhook', methods=['POST']) def webhook(): payload = request.data sig_header = request.headers.get('Stripe-Signature') try: event = stripe.Webhook.construct_event( payload, sig_header, endpoint_secret ) except ValueError: # Invalid payload return 'Invalid payload', 400 except stripe.error.SignatureVerificationError: # Invalid signature return 'Invalid signature', 400 # Handle the event if event['type'] == 'payment_intent.succeeded': payment_intent = event['data']['object'] handle_successful_payment(payment_intent) elif event['type'] == 'payment_intent.payment_failed': payment_intent = event['data']['object'] handle_failed_payment(payment_intent) elif event['type'] == 'customer.subscription.deleted': subscription = event['data']['object'] handle_subscription_canceled(subscription) return 'Success', 200 def handle_successful_payment(payment_intent): """Process successful payment.""" customer_id = payment_intent.get('customer') amount = payment_intent['amount'] metadata = payment_intent.get('metadata', {}) # Update your database # Send confirmation email # Fulfill order print(f"Payment succeeded: {payment_intent['id']}") def handle_failed_payment(payment_intent): """Handle failed payment.""" error = payment_intent.get('last_payment_error', {}) print(f"Payment failed: {error.get('message')}") # Notify customer # Update order status def handle_subscription_canceled(subscription): """Handle subscription cancellation.""" customer_id = subscription['customer'] # Update user access # Send cancellation email print(f"Subscription canceled: {subscription['id']}") ``` ### Webhook Best Practices ```python import hashlib import hmac def verify_webhook_signature(payload, signature, secret): """Manually verify webhook signature.""" expected_sig = hmac.new( secret.encode('utf-8'), payload, hashlib.sha256 ).hexdigest() return hmac.compare_digest(signature, expected_sig) def handle_webhook_idempotently(event_id, handler): """Ensure webhook is processed exactly once.""" # Check if event already processed if is_event_processed(event_id): return # Process event try: handler() mark_event_processed(event_id) except Exception as e: log_error(e) # Stripe will retry failed webhooks raise ``` ## Customer Management ```python def create_customer(email, name, payment_method_id=None): """Create a Stripe customer.""" customer = stripe.Customer.create( email=email, name=name, payment_method=payment_method_id, invoice_settings={ 'default_payment_method': payment_method_id } if payment_method_id else None, metadata={ 'user_id': '12345' } ) return customer def attach_payment_method(customer_id, payment_method_id): """Attach a payment method to a customer.""" stripe.PaymentMethod.attach( payment_method_id, customer=customer_id ) # Set as default stripe.Customer.modify( customer_id, invoice_settings={ 'default_payment_method': payment_method_id } ) def list_customer_payment_methods(customer_id): """List all payment methods for a customer.""" payment_methods = stripe.PaymentMethod.list( customer=customer_id, type='card' ) return payment_methods.data ``` ## Refund Handling ```python def create_refund(payment_intent_id, amount=None, reason=None): """Create a refund.""" refund_params = { 'payment_intent': payment_intent_id } if amount: refund_params['amount'] = amount # Partial refund if reason: refund_params['reason'] = reason # 'duplicate', 'fraudulent', 'requested_by_customer' refund = stripe.Refund.create(**refund_params) return refund def handle_dispute(charge_id, evidence): """Update dispute with evidence.""" stripe.Dispute.modify( charge_id, evidence={ 'customer_name': evidence.get('customer_name'), 'customer_email_address': evidence.get('customer_email'), 'shipping_documentation': evidence.get('shipping_proof'), 'customer_communication': evidence.get('communication'), } ) ``` ## Testing ```python # Use test mode keys stripe.api_key = "sk_test_..." # Test card numbers TEST_CARDS = { 'success': '4242424242424242', 'declined': '4000000000000002', '3d_secure': '4000002500003155', 'insufficient_funds': '4000000000009995' } def test_payment_flow(): """Test complete payment flow.""" # Create test customer customer = stripe.Customer.create( email="test@example.com" ) # Create payment intent intent = stripe.PaymentIntent.create( amount=1000, automatic_payment_methods={ 'enabled': True }, currency='usd', customer=customer.id ) # Confirm with test card confirmed = stripe.PaymentIntent.confirm( intent.id, payment_method='pm_card_visa' # Test payment method ) assert confirmed.status == 'succeeded' ``` ## Resources - **references/checkout-flows.md**: Detailed checkout implementation - **references/webhook-handling.md**: Webhook security and processing - **references/subscription-management.md**: Subscription lifecycle - **references/customer-management.md**: Customer and payment method handling - **references/invoice-generation.md**: Invoicing and billing - **assets/stripe-client.py**: Production-ready Stripe client wrapper - **assets/webhook-handler.py**: Complete webhook processor - **assets/checkout-config.json**: Checkout configuration templates ## Best Practices 1. **Always Use Webhooks**: Don't rely solely on client-side confirmation 2. **Idempotency**: Handle webhook events idempotently 3. **Error Handling**: Gracefully handle all Stripe errors 4. **Test Mode**: Thoroughly test with test keys before production 5. **Metadata**: Use metadata to link Stripe objects to your database 6. **Monitoring**: Track payment success rates and errors 7. **PCI Compliance**: Never handle raw card data on your server 8. **SCA Ready**: Implement 3D Secure for European payments ## Common Pitfalls - **Not Verifying Webhooks**: Always verify webhook signatures - **Missing Webhook Events**: Handle all relevant webhook events - **Hardcoded Amounts**: Use cents/smallest currency unit - **No Retry Logic**: Implement retries for API calls - **Ignoring Test Mode**: Test all edge cases with test cards
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

async-python-patterns

Master Python asyncio, concurrent programming, and async/await

coding
⭐1
# Async Python Patterns Comprehensive guidance for implementing asynchronous Python applications using asyncio, concurrent programming patterns, and async/await for building high-performance, non-blocking systems. ## When to Use This Skill - Building async web APIs (FastAPI, aiohttp, Sanic) - Implementing concurrent I/O operations (database, file, network) - Creating web scrapers with concurrent requests - Developing real-time applications (WebSocket servers, chat systems) - Processing multiple independent tasks simultaneously - Building microservices with async communication - Optimizing I/O-bound workloads - Implementing async background tasks and queues ## Sync vs Async Decision Guide Before adopting async, consider whether it's the right choice for your use case. | Use Case | Recommended Approach | |----------|---------------------| | Many concurrent network/DB calls | `asyncio` | | CPU-bound computation | `multiprocessing` or thread pool | | Mixed I/O + CPU | Offload CPU work with `asyncio.to_thread()` | | Simple scripts, few connections | Sync (simpler, easier to debug) | | Web APIs with high concurrency | Async frameworks (FastAPI, aiohttp) | **Key Rule:** Stay fully sync or fully async within a call path. Mixing creates hidden blocking and complexity. ## Core Concepts ### 1. Event Loop The event loop is the heart of asyncio, managing and scheduling asynchronous tasks. **Key characteristics:** - Single-threaded cooperative multitasking - Schedules coroutines for execution - Handles I/O operations without blocking - Manages callbacks and futures ### 2. Coroutines Functions defined with `async def` that can be paused and resumed. **Syntax:** ```python async def my_coroutine(): result = await some_async_operation() return result ``` ### 3. Tasks Scheduled coroutines that run concurrently on the event loop. ### 4. Futures Low-level objects representing eventual results of async operations. ### 5. Async Context Managers Resources that support `async with` for proper cleanup. ### 6. Async Iterators Objects that support `async for` for iterating over async data sources. ## Quick Start ```python import asyncio async def main(): print("Hello") await asyncio.sleep(1) print("World") # Python 3.7+ asyncio.run(main()) ``` ## Fundamental Patterns ### Pattern 1: Basic Async/Await ```python import asyncio async def fetch_data(url: str) -> dict: """Fetch data from URL asynchronously.""" await asyncio.sleep(1) # Simulate I/O return {"url": url, "data": "result"} async def main(): result = await fetch_data("https://api.example.com") print(result) asyncio.run(main()) ``` ### Pattern 2: Concurrent Execution with gather() ```python import asyncio from typing import List async def fetch_user(user_id: int) -> dict: """Fetch user data.""" await asyncio.sleep(0.5) return {"id": user_id, "name": f"User {user_id}"} async def fetch_all_users(user_ids: List[int]) -> List[dict]: """Fetch multiple users concurrently.""" tasks = [fetch_user(uid) for uid in user_ids] results = await asyncio.gather(*tasks) return results async def main(): user_ids = [1, 2, 3, 4, 5] users = await fetch_all_users(user_ids) print(f"Fetched {len(users)} users") asyncio.run(main()) ``` ### Pattern 3: Task Creation and Management ```python import asyncio async def background_task(name: str, delay: int): """Long-running background task.""" print(f"{name} started") await asyncio.sleep(delay) print(f"{name} completed") return f"Result from {name}" async def main(): # Create tasks task1 = asyncio.create_task(background_task("Task 1", 2)) task2 = asyncio.create_task(background_task("Task 2", 1)) # Do other work print("Main: doing other work") await asyncio.sleep(0.5) # Wait for tasks result1 = await task1 result2 = await task2 print(f"Results: {result1}, {result2}") asyncio.run(main()) ``` ### Pattern 4: Error Handling in Async Code ```python import asyncio from typing import List, Optional async def risky_operation(item_id: int) -> dict: """Operation that might fail.""" await asyncio.sleep(0.1) if item_id % 3 == 0: raise ValueError(f"Item {item_id} failed") return {"id": item_id, "status": "success"} async def safe_operation(item_id: int) -> Optional[dict]: """Wrapper with error handling.""" try: return await risky_operation(item_id) except ValueError as e: print(f"Error: {e}") return None async def process_items(item_ids: List[int]): """Process multiple items with error handling.""" tasks = [safe_operation(iid) for iid in item_ids] results = await asyncio.gather(*tasks, return_exceptions=True) # Filter out failures successful = [r for r in results if r is not None and not isinstance(r, Exception)] failed = [r for r in results if isinstance(r, Exception)] print(f"Success: {len(successful)}, Failed: {len(failed)}") return successful asyncio.run(process_items([1, 2, 3, 4, 5, 6])) ``` ### Pattern 5: Timeout Handling ```python import asyncio async def slow_operation(delay: int) -> str: """Operation that takes time.""" await asyncio.sleep(delay) return f"Completed after {delay}s" async def with_timeout(): """Execute operation with timeout.""" try: result = await asyncio.wait_for(slow_operation(5), timeout=2.0) print(result) except asyncio.TimeoutError: print("Operation timed out") asyncio.run(with_timeout()) ``` ## Advanced Patterns ### Pattern 6: Async Context Managers ```python import asyncio from typing import Optional class AsyncDatabaseConnection: """Async database connection context manager.""" def __init__(self, dsn: str): self.dsn = dsn self.connection: Optional[object] = None async def __aenter__(self): print("Opening connection") await asyncio.sleep(0.1) # Simulate connection self.connection = {"dsn": self.dsn, "connected": True} return self.connection async def __aexit__(self, exc_type, exc_val, exc_tb): print("Closing connection") await asyncio.sleep(0.1) # Simulate cleanup self.connection = None async def query_database(): """Use async context manager.""" async with AsyncDatabaseConnection("postgresql://localhost") as conn: print(f"Using connection: {conn}") await asyncio.sleep(0.2) # Simulate query return {"rows": 10} asyncio.run(query_database()) ``` ### Pattern 7: Async Iterators and Generators ```python import asyncio from typing import AsyncIterator async def async_range(start: int, end: int, delay: float = 0.1) -> AsyncIterator[int]: """Async generator that yields numbers with delay.""" for i in range(start, end): await asyncio.sleep(delay) yield i async def fetch_pages(url: str, max_pages: int) -> AsyncIterator[dict]: """Fetch paginated data asynchronously.""" for page in range(1, max_pages + 1): await asyncio.sleep(0.2) # Simulate API call yield { "page": page, "url": f"{url}?page={page}", "data": [f"item_{page}_{i}" for i in range(5)] } async def consume_async_iterator(): """Consume async iterator.""" async for number in async_range(1, 5): print(f"Number: {number}") print("\nFetching pages:") async for page_data in fetch_pages("https://api.example.com/items", 3): print(f"Page {page_data['page']}: {len(page_data['data'])} items") asyncio.run(consume_async_iterator()) ``` ### Pattern 8: Producer-Consumer Pattern ```python import asyncio from asyncio import Queue from typing import Optional async def producer(queue: Queue, producer_id: int, num_items: int): """Produce items and put them in queue.""" for i in range(num_items): item = f"Item-{producer_id}-{i}" await queue.put(item) print(f"Producer {producer_id} produced: {item}") await asyncio.sleep(0.1) await queue.put(None) # Signal completion async def consumer(queue: Queue, consumer_id: int): """Consume items from queue.""" while True: item = await queue.get() if item is None: queue.task_done() break print(f"Consumer {consumer_id} processing: {item}") await asyncio.sleep(0.2) # Simulate work queue.task_done() async def producer_consumer_example(): """Run producer-consumer pattern.""" queue = Queue(maxsize=10) # Create tasks producers = [ asyncio.create_task(producer(queue, i, 5)) for i in range(2) ] consumers = [ asyncio.create_task(consumer(queue, i)) for i in range(3) ] # Wait for producers await asyncio.gather(*producers) # Wait for queue to be empty await queue.join() # Cancel consumers for c in consumers: c.cancel() asyncio.run(producer_consumer_example()) ``` ### Pattern 9: Semaphore for Rate Limiting ```python import asyncio from typing import List async def api_call(url: str, semaphore: asyncio.Semaphore) -> dict: """Make API call with rate limiting.""" async with semaphore: print(f"Calling {url}") await asyncio.sleep(0.5) # Simulate API call return {"url": url, "status": 200} async def rate_limited_requests(urls: List[str], max_concurrent: int = 5): """Make multiple requests with rate limiting.""" semaphore = asyncio.Semaphore(max_concurrent) tasks = [api_call(url, semaphore) for url in urls] results = await asyncio.gather(*tasks) return results async def main(): urls = [f"https://api.example.com/item/{i}" for i in range(20)] results = await rate_limited_requests(urls, max_concurrent=3) print(f"Completed {len(results)} requests") asyncio.run(main()) ``` ### Pattern 10: Async Locks and Synchronization ```python import asyncio class AsyncCounter: """Thread-safe async counter.""" def __init__(self): self.value = 0 self.lock = asyncio.Lock() async def increment(self): """Safely increment counter.""" async with self.lock: current = self.value await asyncio.sleep(0.01) # Simulate work self.value = current + 1 async def get_value(self) -> int: """Get current value.""" async with self.lock: return self.value async def worker(counter: AsyncCounter, worker_id: int): """Worker that increments counter.""" for _ in range(10): await counter.increment() print(f"Worker {worker_id} incremented") async def test_counter(): """Test concurrent counter.""" counter = AsyncCounter() workers = [asyncio.create_task(worker(counter, i)) for i in range(5)] await asyncio.gather(*workers) final_value = await counter.get_value() print(f"Final counter value: {final_value}") asyncio.run(test_counter()) ``` ## Real-World Applications ### Web Scraping with aiohttp ```python import asyncio import aiohttp from typing import List, Dict async def fetch_url(session: aiohttp.ClientSession, url: str) -> Dict: """Fetch single URL.""" try: async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response: text = await response.text() return { "url": url, "status": response.status, "length": len(text) } except Exception as e: return {"url": url, "error": str(e)} async def scrape_urls(urls: List[str]) -> List[Dict]: """Scrape multiple URLs concurrently.""" async with aiohttp.ClientSession() as session: tasks = [fetch_url(session, url) for url in urls] results = await asyncio.gather(*tasks) return results async def main(): urls = [ "https://httpbin.org/delay/1", "https://httpbin.org/delay/2", "https://httpbin.org/status/404", ] results = await scrape_urls(urls) for result in results: print(result) asyncio.run(main()) ``` ### Async Database Operations ```python import asyncio from typing import List, Optional # Simulated async database client class AsyncDB: """Simulated async database.""" async def execute(self, query: str) -> List[dict]: """Execute query.""" await asyncio.sleep(0.1) return [{"id": 1, "name": "Example"}] async def fetch_one(self, query: str) -> Optional[dict]: """Fetch single row.""" await asyncio.sleep(0.1) return {"id": 1, "name": "Example"} async def get_user_data(db: AsyncDB, user_id: int) -> dict: """Fetch user and related data concurrently.""" user_task = db.fetch_one(f"SELECT * FROM users WHERE id = {user_id}") orders_task = db.execute(f"SELECT * FROM orders WHERE user_id = {user_id}") profile_task = db.fetch_one(f"SELECT * FROM profiles WHERE user_id = {user_id}") user, orders, profile = await asyncio.gather(user_task, orders_task, profile_task) return { "user": user, "orders": orders, "profile": profile } async def main(): db = AsyncDB() user_data = await get_user_data(db, 1) print(user_data) asyncio.run(main()) ``` ### WebSocket Server ```python import asyncio from typing import Set # Simulated WebSocket connection class WebSocket: """Simulated WebSocket.""" def __init__(self, client_id: str): self.client_id = client_id async def send(self, message: str): """Send message.""" print(f"Sending to {self.client_id}: {message}") await asyncio.sleep(0.01) async def recv(self) -> str: """Receive message.""" await asyncio.sleep(1) return f"Message from {self.client_id}" class WebSocketServer: """Simple WebSocket server.""" def __init__(self): self.clients: Set[WebSocket] = set() async def register(self, websocket: WebSocket): """Register new client.""" self.clients.add(websocket) print(f"Client {websocket.client_id} connected") async def unregister(self, websocket: WebSocket): """Unregister client.""" self.clients.remove(websocket) print(f"Client {websocket.client_id} disconnected") async def broadcast(self, message: str): """Broadcast message to all clients.""" if self.clients: tasks = [client.send(message) for client in self.clients] await asyncio.gather(*tasks) async def handle_client(self, websocket: WebSocket): """Handle individual client connection.""" await self.register(websocket) try: async for message in self.message_iterator(websocket): await self.broadcast(f"{websocket.client_id}: {message}") finally: await self.unregister(websocket) async def message_iterator(self, websocket: WebSocket): """Iterate over messages from client.""" for _ in range(3): # Simulate 3 messages yield await websocket.recv() ``` ## Performance Best Practices ### 1. Use Connection Pools ```python import asyncio import aiohttp async def with_connection_pool(): """Use connection pool for efficiency.""" connector = aiohttp.TCPConnector(limit=100, limit_per_host=10) async with aiohttp.ClientSession(connector=connector) as session: tasks = [session.get(f"https://api.example.com/item/{i}") for i in range(50)] responses = await asyncio.gather(*tasks) return responses ``` ### 2. Batch Operations ```python async def batch_process(items: List[str], batch_size: int = 10): """Process items in batches.""" for i in range(0, len(items), batch_size): batch = items[i:i + batch_size] tasks = [process_item(item) for item in batch] await asyncio.gather(*tasks) print(f"Processed batch {i // batch_size + 1}") async def process_item(item: str): """Process single item.""" await asyncio.sleep(0.1) return f"Processed: {item}" ``` ### 3. Avoid Blocking Operations Never block the event loop with synchronous operations. A single blocking call stalls all concurrent tasks. ```python # BAD - blocks the entire event loop async def fetch_data_bad(): import time import requests time.sleep(1) # Blocks! response = requests.get(url) # Also blocks! # GOOD - use async-native libraries (e.g., httpx for async HTTP) import httpx async def fetch_data_good(url: str): await asyncio.sleep(1) async with httpx.AsyncClient() as client: response = await client.get(url) ``` **Wrapping Blocking Code with `asyncio.to_thread()` (Python 3.9+):** When you must use synchronous libraries, offload to a thread pool: ```python import asyncio from pathlib import Path async def read_file_async(path: str) -> str: """Read file without blocking event loop.""" # asyncio.to_thread() runs sync code in a thread pool return await asyncio.to_thread(Path(path).read_text) async def call_sync_library(data: dict) -> dict: """Wrap a synchronous library call.""" # Useful for sync database drivers, file I/O, CPU work return await asyncio.to_thread(sync_library.process, data) ``` **Lower-level approach with `run_in_executor()`:** ```python import asyncio import concurrent.futures from typing import Any def blocking_operation(data: Any) -> Any: """CPU-intensive blocking operation.""" import time time.sleep(1) return data * 2 async def run_in_executor(data: Any) -> Any: """Run blocking operation in thread pool.""" loop = asyncio.get_running_loop() with concurrent.futures.ThreadPoolExecutor() as pool: result = await loop.run_in_executor(pool, blocking_operation, data) return result async def main(): results = await asyncio.gather(*[run_in_executor(i) for i in range(5)]) print(results) asyncio.run(main()) ``` ## Common Pitfalls ### 1. Forgetting await ```python # Wrong - returns coroutine object, doesn't execute result = async_function() # Correct result = await async_function() ``` ### 2. Blocking the Event Loop ```python # Wrong - blocks event loop import time async def bad(): time.sleep(1) # Blocks! # Correct async def good(): await asyncio.sleep(1) # Non-blocking ``` ### 3. Not Handling Cancellation ```python async def cancelable_task(): """Task that handles cancellation.""" try: while True: await asyncio.sleep(1) print("Working...") except asyncio.CancelledError: print("Task cancelled, cleaning up...") # Perform cleanup raise # Re-raise to propagate cancellation ``` ### 4. Mixing Sync and Async Code ```python # Wrong - can't call async from sync directly def sync_function(): result = await async_function() # SyntaxError! # Correct def sync_function(): result = asyncio.run(async_function()) ``` ## Testing Async Code ```python import asyncio import pytest # Using pytest-asyncio @pytest.mark.asyncio async def test_async_function(): """Test async function.""" result = await fetch_data("https
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

attack-tree-construction

Build comprehensive attack trees to visualize threat paths. Use

security
⭐1
# Attack Tree Construction Systematic attack path visualization and analysis. ## When to Use This Skill - Visualizing complex attack scenarios - Identifying defense gaps and priorities - Communicating risks to stakeholders - Planning defensive investments - Penetration test planning - Security architecture review ## Core Concepts ### 1. Attack Tree Structure ``` [Root Goal] | ┌────────────┴────────────┐ │ │ [Sub-goal 1] [Sub-goal 2] (OR node) (AND node) │ │ ┌─────┴─────┐ ┌─────┴─────┐ │ │ │ │ [Attack] [Attack] [Attack] [Attack] (leaf) (leaf) (leaf) (leaf) ``` ### 2. Node Types | Type | Symbol | Description | | -------- | --------- | ----------------------- | | **OR** | Oval | Any child achieves goal | | **AND** | Rectangle | All children required | | **Leaf** | Box | Atomic attack step | ### 3. Attack Attributes | Attribute | Description | Values | | ------------- | ----------------------- | ------------------ | | **Cost** | Resources needed | $, $$, $$$ | | **Time** | Duration to execute | Hours, Days, Weeks | | **Skill** | Expertise required | Low, Medium, High | | **Detection** | Likelihood of detection | Low, Medium, High | ## Templates ### Template 1: Attack Tree Data Model ```python from dataclasses import dataclass, field from enum import Enum from typing import List, Dict, Optional, Union import json class NodeType(Enum): OR = "or" AND = "and" LEAF = "leaf" class Difficulty(Enum): TRIVIAL = 1 LOW = 2 MEDIUM = 3 HIGH = 4 EXPERT = 5 class Cost(Enum): FREE = 0 LOW = 1 MEDIUM = 2 HIGH = 3 VERY_HIGH = 4 class DetectionRisk(Enum): NONE = 0 LOW = 1 MEDIUM = 2 HIGH = 3 CERTAIN = 4 @dataclass class AttackAttributes: difficulty: Difficulty = Difficulty.MEDIUM cost: Cost = Cost.MEDIUM detection_risk: DetectionRisk = DetectionRisk.MEDIUM time_hours: float = 8.0 requires_insider: bool = False requires_physical: bool = False @dataclass class AttackNode: id: str name: str description: str node_type: NodeType attributes: AttackAttributes = field(default_factory=AttackAttributes) children: List['AttackNode'] = field(default_factory=list) mitigations: List[str] = field(default_factory=list) cve_refs: List[str] = field(default_factory=list) def add_child(self, child: 'AttackNode') -> None: self.children.append(child) def calculate_path_difficulty(self) -> float: """Calculate aggregate difficulty for this path.""" if self.node_type == NodeType.LEAF: return self.attributes.difficulty.value if not self.children: return 0 child_difficulties = [c.calculate_path_difficulty() for c in self.children] if self.node_type == NodeType.OR: return min(child_difficulties) else: # AND return max(child_difficulties) def calculate_path_cost(self) -> float: """Calculate aggregate cost for this path.""" if self.node_type == NodeType.LEAF: return self.attributes.cost.value if not self.children: return 0 child_costs = [c.calculate_path_cost() for c in self.children] if self.node_type == NodeType.OR: return min(child_costs) else: # AND return sum(child_costs) def to_dict(self) -> Dict: """Convert to dictionary for serialization.""" return { "id": self.id, "name": self.name, "description": self.description, "type": self.node_type.value, "attributes": { "difficulty": self.attributes.difficulty.name, "cost": self.attributes.cost.name, "detection_risk": self.attributes.detection_risk.name, "time_hours": self.attributes.time_hours, }, "mitigations": self.mitigations, "children": [c.to_dict() for c in self.children] } @dataclass class AttackTree: name: str description: str root: AttackNode version: str = "1.0" def find_easiest_path(self) -> List[AttackNode]: """Find the path with lowest difficulty.""" return self._find_path(self.root, minimize="difficulty") def find_cheapest_path(self) -> List[AttackNode]: """Find the path with lowest cost.""" return self._find_path(self.root, minimize="cost") def find_stealthiest_path(self) -> List[AttackNode]: """Find the path with lowest detection risk.""" return self._find_path(self.root, minimize="detection") def _find_path( self, node: AttackNode, minimize: str ) -> List[AttackNode]: """Recursive path finding.""" if node.node_type == NodeType.LEAF: return [node] if not node.children: return [node] if node.node_type == NodeType.OR: # Pick the best child path best_path = None best_score = float('inf') for child in node.children: child_path = self._find_path(child, minimize) score = self._path_score(child_path, minimize) if score < best_score: best_score = score best_path = child_path return [node] + (best_path or []) else: # AND # Must traverse all children path = [node] for child in node.children: path.extend(self._find_path(child, minimize)) return path def _path_score(self, path: List[AttackNode], metric: str) -> float: """Calculate score for a path.""" if metric == "difficulty": return sum(n.attributes.difficulty.value for n in path if n.node_type == NodeType.LEAF) elif metric == "cost": return sum(n.attributes.cost.value for n in path if n.node_type == NodeType.LEAF) elif metric == "detection": return sum(n.attributes.detection_risk.value for n in path if n.node_type == NodeType.LEAF) return 0 def get_all_leaf_attacks(self) -> List[AttackNode]: """Get all leaf attack nodes.""" leaves = [] self._collect_leaves(self.root, leaves) return leaves def _collect_leaves(self, node: AttackNode, leaves: List[AttackNode]) -> None: if node.node_type == NodeType.LEAF: leaves.append(node) for child in node.children: self._collect_leaves(child, leaves) def get_unmitigated_attacks(self) -> List[AttackNode]: """Find attacks without mitigations.""" return [n for n in self.get_all_leaf_attacks() if not n.mitigations] def export_json(self) -> str: """Export tree to JSON.""" return json.dumps({ "name": self.name, "description": self.description, "version": self.version, "root": self.root.to_dict() }, indent=2) ``` ### Template 2: Attack Tree Builder ```python class AttackTreeBuilder: """Fluent builder for attack trees.""" def __init__(self, name: str, description: str): self.name = name self.description = description self._node_stack: List[AttackNode] = [] self._root: Optional[AttackNode] = None def goal(self, id: str, name: str, description: str = "") -> 'AttackTreeBuilder': """Set the root goal (OR node by default).""" self._root = AttackNode( id=id, name=name, description=description, node_type=NodeType.OR ) self._node_stack = [self._root] return self def or_node(self, id: str, name: str, description: str = "") -> 'AttackTreeBuilder': """Add an OR sub-goal.""" node = AttackNode( id=id, name=name, description=description, node_type=NodeType.OR ) self._current().add_child(node) self._node_stack.append(node) return self def and_node(self, id: str, name: str, description: str = "") -> 'AttackTreeBuilder': """Add an AND sub-goal (all children required).""" node = AttackNode( id=id, name=name, description=description, node_type=NodeType.AND ) self._current().add_child(node) self._node_stack.append(node) return self def attack( self, id: str, name: str, description: str = "", difficulty: Difficulty = Difficulty.MEDIUM, cost: Cost = Cost.MEDIUM, detection: DetectionRisk = DetectionRisk.MEDIUM, time_hours: float = 8.0, mitigations: List[str] = None ) -> 'AttackTreeBuilder': """Add a leaf attack node.""" node = AttackNode( id=id, name=name, description=description, node_type=NodeType.LEAF, attributes=AttackAttributes( difficulty=difficulty, cost=cost, detection_risk=detection, time_hours=time_hours ), mitigations=mitigations or [] ) self._current().add_child(node) return self def end(self) -> 'AttackTreeBuilder': """Close current node, return to parent.""" if len(self._node_stack) > 1: self._node_stack.pop() return self def build(self) -> AttackTree: """Build the attack tree.""" if not self._root: raise ValueError("No root goal defined") return AttackTree( name=self.name, description=self.description, root=self._root ) def _current(self) -> AttackNode: if not self._node_stack: raise ValueError("No current node") return self._node_stack[-1] # Example usage def build_account_takeover_tree() -> AttackTree: """Build attack tree for account takeover scenario.""" return ( AttackTreeBuilder("Account Takeover", "Gain unauthorized access to user account") .goal("G1", "Take Over User Account") .or_node("S1", "Steal Credentials") .attack( "A1", "Phishing Attack", difficulty=Difficulty.LOW, cost=Cost.LOW, detection=DetectionRisk.MEDIUM, mitigations=["Security awareness training", "Email filtering"] ) .attack( "A2", "Credential Stuffing", difficulty=Difficulty.TRIVIAL, cost=Cost.LOW, detection=DetectionRisk.HIGH, mitigations=["Rate limiting", "MFA", "Password breach monitoring"] ) .attack( "A3", "Keylogger Malware", difficulty=Difficulty.MEDIUM, cost=Cost.MEDIUM, detection=DetectionRisk.MEDIUM, mitigations=["Endpoint protection", "MFA"] ) .end() .or_node("S2", "Bypass Authentication") .attack( "A4", "Session Hijacking", difficulty=Difficulty.MEDIUM, cost=Cost.LOW, detection=DetectionRisk.LOW, mitigations=["Secure session management", "HTTPS only"] ) .attack( "A5", "Authentication Bypass Vulnerability", difficulty=Difficulty.HIGH, cost=Cost.LOW, detection=DetectionRisk.LOW, mitigations=["Security testing", "Code review", "WAF"] ) .end() .or_node("S3", "Social Engineering") .and_node("S3.1", "Account Recovery Attack") .attack( "A6", "Gather Personal Information", difficulty=Difficulty.LOW, cost=Cost.FREE, detection=DetectionRisk.NONE ) .attack( "A7", "Call Support Desk", difficulty=Difficulty.MEDIUM, cost=Cost.FREE, detection=DetectionRisk.MEDIUM, mitigations=["Support verification procedures", "Security questions"] ) .end() .end() .build() ) ``` ### Template 3: Mermaid Diagram Generator ```python class MermaidExporter: """Export attack trees to Mermaid diagram format.""" def __init__(self, tree: AttackTree): self.tree = tree self._lines: List[str] = [] self._node_count = 0 def export(self) -> str: """Export tree to Mermaid flowchart.""" self._lines = ["flowchart TD"] self._export_node(self.tree.root, None) return "\n".join(self._lines) def _export_node(self, node: AttackNode, parent_id: Optional[str]) -> str: """Recursively export nodes.""" node_id = f"N{self._node_count}" self._node_count += 1 # Node shape based on type if node.node_type == NodeType.OR: shape = f"{node_id}(({node.name}))" elif node.node_type == NodeType.AND: shape = f"{node_id}[{node.name}]" else: # LEAF # Color based on difficulty style = self._get_leaf_style(node) shape = f"{node_id}[/{node.name}/]" self._lines.append(f" style {node_id} {style}") self._lines.append(f" {shape}") if parent_id: connector = "-->" if node.node_type != NodeType.AND else "==>" self._lines.append(f" {parent_id} {connector} {node_id}") for child in node.children: self._export_node(child, node_id) return node_id def _get_leaf_style(self, node: AttackNode) -> str: """Get style based on attack attributes.""" colors = { Difficulty.TRIVIAL: "fill:#ff6b6b", # Red - easy attack Difficulty.LOW: "fill:#ffa06b", Difficulty.MEDIUM: "fill:#ffd93d", Difficulty.HIGH: "fill:#6bcb77", Difficulty.EXPERT: "fill:#4d96ff", # Blue - hard attack } color = colors.get(node.attributes.difficulty, "fill:#gray") return color class PlantUMLExporter: """Export attack trees to PlantUML format.""" def __init__(self, tree: AttackTree): self.tree = tree def export(self) -> str: """Export tree to PlantUML.""" lines = [ "@startmindmap", f"* {self.tree.name}", ] self._export_node(self.tree.root, lines, 1) lines.append("@endmindmap") return "\n".join(lines) def _export_node(self, node: AttackNode, lines: List[str], depth: int) -> None: """Recursively export nodes.""" prefix = "*" * (depth + 1) if node.node_type == NodeType.OR: marker = "[OR]" elif node.node_type == NodeType.AND: marker = "[AND]" else: diff = node.attributes.difficulty.name marker = f"<<{diff}>>" lines.append(f"{prefix} {marker} {node.name}") for child in node.children: self._export_node(child, lines, depth + 1) ``` ### Template 4: Attack Path Analysis ```python from typing import Set, Tuple class AttackPathAnalyzer: """Analyze attack paths and coverage.""" def __init__(self, tree: AttackTree): self.tree = tree def get_all_paths(self) -> List[List[AttackNode]]: """Get all possible attack paths.""" paths = [] self._collect_paths(self.tree.root, [], paths) return paths def _collect_paths( self, node: AttackNode, current_path: List[AttackNode], all_paths: List[List[AttackNode]] ) -> None: """Recursively collect all paths.""" current_path = current_path + [node] if node.node_type == NodeType.LEAF: all_paths.append(current_path) return if not node.children: all_paths.append(current_path) return if node.node_type == NodeType.OR: # Each child is a separate path for child in node.children: self._collect_paths(child, current_path, all_paths) else: # AND # Must combine all children child_paths = [] for child in node.children: child_sub_paths = [] self._collect_paths(child, [], child_sub_paths) child_paths.append(child_sub_paths) # Combine paths from all AND children combined = self._combine_and_paths(child_paths) for combo in combined: all_paths.append(current_path + combo) def _combine_and_paths( self, child_paths: List[List[List[AttackNode]]] ) -> List[List[AttackNode]]: """Combine paths from AND node children.""" if not child_paths: return [[]] if len(child_paths) == 1: return [path for paths in child_paths for path in paths] # Cartesian product of all child path combinations result = [[]] for paths in child_paths: new_result = [] for existing in result: for path in paths: new_result.append(existing + path) result = new_result return result def calculate_path_metrics(self, path: List[AttackNode]) -> Dict: """Calculate metrics for a specific path.""" leaves = [n for n in path if n.node_type == NodeType.LEAF] total_difficulty = sum(n.attributes.difficulty.value for n in leaves) total_cost = sum(n.attributes.cost.value for n in leaves) total_time = sum(n.attributes.time_hours for n in leaves) max_detection = max((n.attributes.detection_risk.value for n in leaves), default=0) return { "steps": len(leaves), "total_difficulty": total_difficulty, "avg_difficulty": total_difficulty / len(leaves) if leaves else 0, "total_cost": total_cost, "total_time_hours": total_time, "max_detection_risk": max_detection, "requires_insider": any(n.attributes.requires_insider for n in leaves), "requires_physical": any(n.attributes.requires_physical for n in leaves), } def identify_critical_nodes(self) -> List[Tuple[AttackNode, int]]: """Find nodes that appear in the most paths.""" paths = self.get_all_paths() node_counts: Dict[str, Tuple[AttackNode, int]] = {} for path in paths: for node in path: if node.id not in node_counts: node_counts[node.id] = (node, 0) node_counts[node.id] = (node, node_counts[node.id][1] + 1) return sort
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

go-concurrency-patterns

Master Go concurrency with goroutines, channels, sync primitives,

coding
⭐1
# Go Concurrency Patterns Production patterns for Go concurrency including goroutines, channels, synchronization primitives, and context management. ## When to Use This Skill - Building concurrent Go applications - Implementing worker pools and pipelines - Managing goroutine lifecycles - Using channels for communication - Debugging race conditions - Implementing graceful shutdown ## Core Concepts ### 1. Go Concurrency Primitives | Primitive | Purpose | | ----------------- | -------------------------------- | | `goroutine` | Lightweight concurrent execution | | `channel` | Communication between goroutines | | `select` | Multiplex channel operations | | `sync.Mutex` | Mutual exclusion | | `sync.WaitGroup` | Wait for goroutines to complete | | `context.Context` | Cancellation and deadlines | ### 2. Go Concurrency Mantra ``` Don't communicate by sharing memory; share memory by communicating. ``` ## Quick Start ```go package main import ( "context" "fmt" "sync" "time" ) func main() { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() results := make(chan string, 10) var wg sync.WaitGroup // Spawn workers for i := 0; i < 3; i++ { wg.Add(1) go worker(ctx, i, results, &wg) } // Close results when done go func() { wg.Wait() close(results) }() // Collect results for result := range results { fmt.Println(result) } } func worker(ctx context.Context, id int, results chan<- string, wg *sync.WaitGroup) { defer wg.Done() select { case <-ctx.Done(): return case results <- fmt.Sprintf("Worker %d done", id): } } ``` ## Patterns ### Pattern 1: Worker Pool ```go package main import ( "context" "fmt" "sync" ) type Job struct { ID int Data string } type Result struct { JobID int Output string Err error } func WorkerPool(ctx context.Context, numWorkers int, jobs <-chan Job) <-chan Result { results := make(chan Result, len(jobs)) var wg sync.WaitGroup for i := 0; i < numWorkers; i++ { wg.Add(1) go func(workerID int) { defer wg.Done() for job := range jobs { select { case <-ctx.Done(): return default: result := processJob(job) results <- result } } }(i) } go func() { wg.Wait() close(results) }() return results } func processJob(job Job) Result { // Simulate work return Result{ JobID: job.ID, Output: fmt.Sprintf("Processed: %s", job.Data), } } // Usage func main() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() jobs := make(chan Job, 100) // Send jobs go func() { for i := 0; i < 50; i++ { jobs <- Job{ID: i, Data: fmt.Sprintf("job-%d", i)} } close(jobs) }() // Process with 5 workers results := WorkerPool(ctx, 5, jobs) for result := range results { fmt.Printf("Result: %+v\n", result) } } ``` ### Pattern 2: Fan-Out/Fan-In Pipeline ```go package main import ( "context" "sync" ) // Stage 1: Generate numbers func generate(ctx context.Context, nums ...int) <-chan int { out := make(chan int) go func() { defer close(out) for _, n := range nums { select { case <-ctx.Done(): return case out <- n: } } }() return out } // Stage 2: Square numbers (can run multiple instances) func square(ctx context.Context, in <-chan int) <-chan int { out := make(chan int) go func() { defer close(out) for n := range in { select { case <-ctx.Done(): return case out <- n * n: } } }() return out } // Fan-in: Merge multiple channels into one func merge(ctx context.Context, cs ...<-chan int) <-chan int { var wg sync.WaitGroup out := make(chan int) // Start output goroutine for each input channel output := func(c <-chan int) { defer wg.Done() for n := range c { select { case <-ctx.Done(): return case out <- n: } } } wg.Add(len(cs)) for _, c := range cs { go output(c) } // Close out after all inputs are done go func() { wg.Wait() close(out) }() return out } func main() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() // Generate input in := generate(ctx, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10) // Fan out to multiple squarers c1 := square(ctx, in) c2 := square(ctx, in) c3 := square(ctx, in) // Fan in results for result := range merge(ctx, c1, c2, c3) { fmt.Println(result) } } ``` ### Pattern 3: Bounded Concurrency with Semaphore ```go package main import ( "context" "fmt" "golang.org/x/sync/semaphore" "sync" ) type RateLimitedWorker struct { sem *semaphore.Weighted } func NewRateLimitedWorker(maxConcurrent int64) *RateLimitedWorker { return &RateLimitedWorker{ sem: semaphore.NewWeighted(maxConcurrent), } } func (w *RateLimitedWorker) Do(ctx context.Context, tasks []func() error) []error { var ( wg sync.WaitGroup mu sync.Mutex errors []error ) for _, task := range tasks { // Acquire semaphore (blocks if at limit) if err := w.sem.Acquire(ctx, 1); err != nil { return []error{err} } wg.Add(1) go func(t func() error) { defer wg.Done() defer w.sem.Release(1) if err := t(); err != nil { mu.Lock() errors = append(errors, err) mu.Unlock() } }(task) } wg.Wait() return errors } // Alternative: Channel-based semaphore type Semaphore chan struct{} func NewSemaphore(n int) Semaphore { return make(chan struct{}, n) } func (s Semaphore) Acquire() { s <- struct{}{} } func (s Semaphore) Release() { <-s } ``` ### Pattern 4: Graceful Shutdown ```go package main import ( "context" "fmt" "os" "os/signal" "sync" "syscall" "time" ) type Server struct { shutdown chan struct{} wg sync.WaitGroup } func NewServer() *Server { return &Server{ shutdown: make(chan struct{}), } } func (s *Server) Start(ctx context.Context) { // Start workers for i := 0; i < 5; i++ { s.wg.Add(1) go s.worker(ctx, i) } } func (s *Server) worker(ctx context.Context, id int) { defer s.wg.Done() defer fmt.Printf("Worker %d stopped\n", id) ticker := time.NewTicker(time.Second) defer ticker.Stop() for { select { case <-ctx.Done(): // Cleanup fmt.Printf("Worker %d cleaning up...\n", id) time.Sleep(500 * time.Millisecond) // Simulated cleanup return case <-ticker.C: fmt.Printf("Worker %d working...\n", id) } } } func (s *Server) Shutdown(timeout time.Duration) { // Signal shutdown close(s.shutdown) // Wait with timeout done := make(chan struct{}) go func() { s.wg.Wait() close(done) }() select { case <-done: fmt.Println("Clean shutdown completed") case <-time.After(timeout): fmt.Println("Shutdown timed out, forcing exit") } } func main() { // Setup signal handling ctx, cancel := context.WithCancel(context.Background()) sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) server := NewServer() server.Start(ctx) // Wait for signal sig := <-sigCh fmt.Printf("\nReceived signal: %v\n", sig) // Cancel context to stop workers cancel() // Wait for graceful shutdown server.Shutdown(5 * time.Second) } ``` ### Pattern 5: Error Group with Cancellation ```go package main import ( "context" "fmt" "golang.org/x/sync/errgroup" "net/http" ) func fetchAllURLs(ctx context.Context, urls []string) ([]string, error) { g, ctx := errgroup.WithContext(ctx) results := make([]string, len(urls)) for i, url := range urls { i, url := i, url // Capture loop variables g.Go(func() error { req, err := http.NewRequestWithContext(ctx, "GET", url, nil) if err != nil { return fmt.Errorf("creating request for %s: %w", url, err) } resp, err := http.DefaultClient.Do(req) if err != nil { return fmt.Errorf("fetching %s: %w", url, err) } defer resp.Body.Close() results[i] = fmt.Sprintf("%s: %d", url, resp.StatusCode) return nil }) } // Wait for all goroutines to complete or one to fail if err := g.Wait(); err != nil { return nil, err // First error cancels all others } return results, nil } // With concurrency limit func fetchWithLimit(ctx context.Context, urls []string, limit int) ([]string, error) { g, ctx := errgroup.WithContext(ctx) g.SetLimit(limit) // Max concurrent goroutines results := make([]string, len(urls)) var mu sync.Mutex for i, url := range urls { i, url := i, url g.Go(func() error { result, err := fetchURL(ctx, url) if err != nil { return err } mu.Lock() results[i] = result mu.Unlock() return nil }) } if err := g.Wait(); err != nil { return nil, err } return results, nil } ``` ### Pattern 6: Concurrent Map with sync.Map ```go package main import ( "sync" ) // For frequent reads, infrequent writes type Cache struct { m sync.Map } func (c *Cache) Get(key string) (interface{}, bool) { return c.m.Load(key) } func (c *Cache) Set(key string, value interface{}) { c.m.Store(key, value) } func (c *Cache) GetOrSet(key string, value interface{}) (interface{}, bool) { return c.m.LoadOrStore(key, value) } func (c *Cache) Delete(key string) { c.m.Delete(key) } // For write-heavy workloads, use sharded map type ShardedMap struct { shards []*shard numShards int } type shard struct { sync.RWMutex data map[string]interface{} } func NewShardedMap(numShards int) *ShardedMap { m := &ShardedMap{ shards: make([]*shard, numShards), numShards: numShards, } for i := range m.shards { m.shards[i] = &shard{data: make(map[string]interface{})} } return m } func (m *ShardedMap) getShard(key string) *shard { // Simple hash h := 0 for _, c := range key { h = 31*h + int(c) } return m.shards[h%m.numShards] } func (m *ShardedMap) Get(key string) (interface{}, bool) { shard := m.getShard(key) shard.RLock() defer shard.RUnlock() v, ok := shard.data[key] return v, ok } func (m *ShardedMap) Set(key string, value interface{}) { shard := m.getShard(key) shard.Lock() defer shard.Unlock() shard.data[key] = value } ``` ### Pattern 7: Select with Timeout and Default ```go func selectPatterns() { ch := make(chan int) // Timeout pattern select { case v := <-ch: fmt.Println("Received:", v) case <-time.After(time.Second): fmt.Println("Timeout!") } // Non-blocking send/receive select { case ch <- 42: fmt.Println("Sent") default: fmt.Println("Channel full, skipping") } // Priority select (check high priority first) highPriority := make(chan int) lowPriority := make(chan int) for { select { case msg := <-highPriority: fmt.Println("High priority:", msg) default: select { case msg := <-highPriority: fmt.Println("High priority:", msg) case msg := <-lowPriority: fmt.Println("Low priority:", msg) } } } } ``` ## Race Detection ```bash # Run tests with race detector go test -race ./... # Build with race detector go build -race . # Run with race detector go run -race main.go ``` ## Best Practices ### Do's - **Use context** - For cancellation and deadlines - **Close channels** - From sender side only - **Use errgroup** - For concurrent operations with errors - **Buffer channels** - When you know the count - **Prefer channels** - Over mutexes when possible ### Don'ts - **Don't leak goroutines** - Always have exit path - **Don't close from receiver** - Causes panic - **Don't use shared memory** - Unless necessary - **Don't ignore context cancellation** - Check ctx.Done() - **Don't use time.Sleep for sync** - Use proper primitives ## Resources - [Go Concurrency Patterns](https://go.dev/blog/pipelines) - [Effective Go - Concurrency](https://go.dev/doc/effective_go#concurrency) - [Go by Example - Goroutines](https://gobyexample.com/goroutines)
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

rust-async-patterns

Master Rust async programming with Tokio, async traits, error

coding
⭐1
# Rust Async Patterns Production patterns for async Rust programming with Tokio runtime, including tasks, channels, streams, and error handling. ## When to Use This Skill - Building async Rust applications - Implementing concurrent network services - Using Tokio for async I/O - Handling async errors properly - Debugging async code issues - Optimizing async performance ## Core Concepts ### 1. Async Execution Model ``` Future (lazy) → poll() → Ready(value) | Pending ↑ ↓ Waker ← Runtime schedules ``` ### 2. Key Abstractions | Concept | Purpose | | ---------- | ---------------------------------------- | | `Future` | Lazy computation that may complete later | | `async fn` | Function returning impl Future | | `await` | Suspend until future completes | | `Task` | Spawned future running concurrently | | `Runtime` | Executor that polls futures | ## Quick Start ```toml # Cargo.toml [dependencies] tokio = { version = "1", features = ["full"] } futures = "0.3" async-trait = "0.1" anyhow = "1.0" tracing = "0.1" tracing-subscriber = "0.3" ``` ```rust use tokio::time::{sleep, Duration}; use anyhow::Result; #[tokio::main] async fn main() -> Result<()> { // Initialize tracing tracing_subscriber::fmt::init(); // Async operations let result = fetch_data("https://api.example.com").await?; println!("Got: {}", result); Ok(()) } async fn fetch_data(url: &str) -> Result<String> { // Simulated async operation sleep(Duration::from_millis(100)).await; Ok(format!("Data from {}", url)) } ``` ## Patterns ### Pattern 1: Concurrent Task Execution ```rust use tokio::task::JoinSet; use anyhow::Result; // Spawn multiple concurrent tasks async fn fetch_all_concurrent(urls: Vec<String>) -> Result<Vec<String>> { let mut set = JoinSet::new(); for url in urls { set.spawn(async move { fetch_data(&url).await }); } let mut results = Vec::new(); while let Some(res) = set.join_next().await { match res { Ok(Ok(data)) => results.push(data), Ok(Err(e)) => tracing::error!("Task failed: {}", e), Err(e) => tracing::error!("Join error: {}", e), } } Ok(results) } // With concurrency limit use futures::stream::{self, StreamExt}; async fn fetch_with_limit(urls: Vec<String>, limit: usize) -> Vec<Result<String>> { stream::iter(urls) .map(|url| async move { fetch_data(&url).await }) .buffer_unordered(limit) // Max concurrent tasks .collect() .await } // Select first to complete use tokio::select; async fn race_requests(url1: &str, url2: &str) -> Result<String> { select! { result = fetch_data(url1) => result, result = fetch_data(url2) => result, } } ``` ### Pattern 2: Channels for Communication ```rust use tokio::sync::{mpsc, broadcast, oneshot, watch}; // Multi-producer, single-consumer async fn mpsc_example() { let (tx, mut rx) = mpsc::channel::<String>(100); // Spawn producer let tx2 = tx.clone(); tokio::spawn(async move { tx2.send("Hello".to_string()).await.unwrap(); }); // Consume while let Some(msg) = rx.recv().await { println!("Got: {}", msg); } } // Broadcast: multi-producer, multi-consumer async fn broadcast_example() { let (tx, _) = broadcast::channel::<String>(100); let mut rx1 = tx.subscribe(); let mut rx2 = tx.subscribe(); tx.send("Event".to_string()).unwrap(); // Both receivers get the message let _ = rx1.recv().await; let _ = rx2.recv().await; } // Oneshot: single value, single use async fn oneshot_example() -> String { let (tx, rx) = oneshot::channel::<String>(); tokio::spawn(async move { tx.send("Result".to_string()).unwrap(); }); rx.await.unwrap() } // Watch: single producer, multi-consumer, latest value async fn watch_example() { let (tx, mut rx) = watch::channel("initial".to_string()); tokio::spawn(async move { loop { // Wait for changes rx.changed().await.unwrap(); println!("New value: {}", *rx.borrow()); } }); tx.send("updated".to_string()).unwrap(); } ``` ### Pattern 3: Async Error Handling ```rust use anyhow::{Context, Result, bail}; use thiserror::Error; #[derive(Error, Debug)] pub enum ServiceError { #[error("Network error: {0}")] Network(#[from] reqwest::Error), #[error("Database error: {0}")] Database(#[from] sqlx::Error), #[error("Not found: {0}")] NotFound(String), #[error("Timeout after {0:?}")] Timeout(std::time::Duration), } // Using anyhow for application errors async fn process_request(id: &str) -> Result<Response> { let data = fetch_data(id) .await .context("Failed to fetch data")?; let parsed = parse_response(&data) .context("Failed to parse response")?; Ok(parsed) } // Using custom errors for library code async fn get_user(id: &str) -> Result<User, ServiceError> { let result = db.query(id).await?; match result { Some(user) => Ok(user), None => Err(ServiceError::NotFound(id.to_string())), } } // Timeout wrapper use tokio::time::timeout; async fn with_timeout<T, F>(duration: Duration, future: F) -> Result<T, ServiceError> where F: std::future::Future<Output = Result<T, ServiceError>>, { timeout(duration, future) .await .map_err(|_| ServiceError::Timeout(duration))? } ``` ### Pattern 4: Graceful Shutdown ```rust use tokio::signal; use tokio::sync::broadcast; use tokio_util::sync::CancellationToken; async fn run_server() -> Result<()> { // Method 1: CancellationToken let token = CancellationToken::new(); let token_clone = token.clone(); // Spawn task that respects cancellation tokio::spawn(async move { loop { tokio::select! { _ = token_clone.cancelled() => { tracing::info!("Task shutting down"); break; } _ = do_work() => {} } } }); // Wait for shutdown signal signal::ctrl_c().await?; tracing::info!("Shutdown signal received"); // Cancel all tasks token.cancel(); // Give tasks time to cleanup tokio::time::sleep(Duration::from_secs(5)).await; Ok(()) } // Method 2: Broadcast channel for shutdown async fn run_with_broadcast() -> Result<()> { let (shutdown_tx, _) = broadcast::channel::<()>(1); let mut rx = shutdown_tx.subscribe(); tokio::spawn(async move { tokio::select! { _ = rx.recv() => { tracing::info!("Received shutdown"); } _ = async { loop { do_work().await } } => {} } }); signal::ctrl_c().await?; let _ = shutdown_tx.send(()); Ok(()) } ``` ### Pattern 5: Async Traits ```rust use async_trait::async_trait; #[async_trait] pub trait Repository { async fn get(&self, id: &str) -> Result<Entity>; async fn save(&self, entity: &Entity) -> Result<()>; async fn delete(&self, id: &str) -> Result<()>; } pub struct PostgresRepository { pool: sqlx::PgPool, } #[async_trait] impl Repository for PostgresRepository { async fn get(&self, id: &str) -> Result<Entity> { sqlx::query_as!(Entity, "SELECT * FROM entities WHERE id = $1", id) .fetch_one(&self.pool) .await .map_err(Into::into) } async fn save(&self, entity: &Entity) -> Result<()> { sqlx::query!( "INSERT INTO entities (id, data) VALUES ($1, $2) ON CONFLICT (id) DO UPDATE SET data = $2", entity.id, entity.data ) .execute(&self.pool) .await?; Ok(()) } async fn delete(&self, id: &str) -> Result<()> { sqlx::query!("DELETE FROM entities WHERE id = $1", id) .execute(&self.pool) .await?; Ok(()) } } // Trait object usage async fn process(repo: &dyn Repository, id: &str) -> Result<()> { let entity = repo.get(id).await?; // Process... repo.save(&entity).await } ``` ### Pattern 6: Streams and Async Iteration ```rust use futures::stream::{self, Stream, StreamExt}; use async_stream::stream; // Create stream from async iterator fn numbers_stream() -> impl Stream<Item = i32> { stream! { for i in 0..10 { tokio::time::sleep(Duration::from_millis(100)).await; yield i; } } } // Process stream async fn process_stream() { let stream = numbers_stream(); // Map and filter let processed: Vec<_> = stream .filter(|n| futures::future::ready(*n % 2 == 0)) .map(|n| n * 2) .collect() .await; println!("{:?}", processed); } // Chunked processing async fn process_in_chunks() { let stream = numbers_stream(); let mut chunks = stream.chunks(3); while let Some(chunk) = chunks.next().await { println!("Processing chunk: {:?}", chunk); } } // Merge multiple streams async fn merge_streams() { let stream1 = numbers_stream(); let stream2 = numbers_stream(); let merged = stream::select(stream1, stream2); merged .for_each(|n| async move { println!("Got: {}", n); }) .await; } ``` ### Pattern 7: Resource Management ```rust use std::sync::Arc; use tokio::sync::{Mutex, RwLock, Semaphore}; // Shared state with RwLock (prefer for read-heavy) struct Cache { data: RwLock<HashMap<String, String>>, } impl Cache { async fn get(&self, key: &str) -> Option<String> { self.data.read().await.get(key).cloned() } async fn set(&self, key: String, value: String) { self.data.write().await.insert(key, value); } } // Connection pool with semaphore struct Pool { semaphore: Semaphore, connections: Mutex<Vec<Connection>>, } impl Pool { fn new(size: usize) -> Self { Self { semaphore: Semaphore::new(size), connections: Mutex::new((0..size).map(|_| Connection::new()).collect()), } } async fn acquire(&self) -> PooledConnection<'_> { let permit = self.semaphore.acquire().await.unwrap(); let conn = self.connections.lock().await.pop().unwrap(); PooledConnection { pool: self, conn: Some(conn), _permit: permit } } } struct PooledConnection<'a> { pool: &'a Pool, conn: Option<Connection>, _permit: tokio::sync::SemaphorePermit<'a>, } impl Drop for PooledConnection<'_> { fn drop(&mut self) { if let Some(conn) = self.conn.take() { let pool = self.pool; tokio::spawn(async move { pool.connections.lock().await.push(conn); }); } } } ``` ## Debugging Tips ```rust // Enable tokio-console for runtime debugging // Cargo.toml: tokio = { features = ["tracing"] } // Run: RUSTFLAGS="--cfg tokio_unstable" cargo run // Then: tokio-console // Instrument async functions use tracing::instrument; #[instrument(skip(pool))] async fn fetch_user(pool: &PgPool, id: &str) -> Result<User> { tracing::debug!("Fetching user"); // ... } // Track task spawning let span = tracing::info_span!("worker", id = %worker_id); tokio::spawn(async move { // Enters span when polled }.instrument(span)); ``` ## Best Practices ### Do's - **Use `tokio::select!`** - For racing futures - **Prefer channels** - Over shared state when possible - **Use `JoinSet`** - For managing multiple tasks - **Instrument with tracing** - For debugging async code - **Handle cancellation** - Check `CancellationToken` ### Don'ts - **Don't block** - Never use `std::thread::sleep` in async - **Don't hold locks across awaits** - Causes deadlocks - **Don't spawn unboundedly** - Use semaphores for limits - **Don't ignore errors** - Propagate with `?` or log - **Don't forget Send bounds** - For spawned futures ## Resources - [Tokio Tutorial](https://tokio.rs/tokio/tutorial) - [Async Book](https://rust-lang.github.io/async-book/) - [Tokio Console](https://github.com/tokio-rs/console)
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

interaction-design

Design and implement microinteractions, motion design, transitions,

coding
⭐1
# Interaction Design Create engaging, intuitive interactions through motion, feedback, and thoughtful state transitions that enhance usability and delight users. ## When to Use This Skill - Adding microinteractions to enhance user feedback - Implementing smooth page and component transitions - Designing loading states and skeleton screens - Creating gesture-based interactions - Building notification and toast systems - Implementing drag-and-drop interfaces - Adding scroll-triggered animations - Designing hover and focus states ## Core Principles ### 1. Purposeful Motion Motion should communicate, not decorate: - **Feedback**: Confirm user actions occurred - **Orientation**: Show where elements come from/go to - **Focus**: Direct attention to important changes - **Continuity**: Maintain context during transitions ### 2. Timing Guidelines | Duration | Use Case | | --------- | ----------------------------------------- | | 100-150ms | Micro-feedback (hovers, clicks) | | 200-300ms | Small transitions (toggles, dropdowns) | | 300-500ms | Medium transitions (modals, page changes) | | 500ms+ | Complex choreographed animations | ### 3. Easing Functions ```css /* Common easings */ --ease-out: cubic-bezier(0.16, 1, 0.3, 1); /* Decelerate - entering */ --ease-in: cubic-bezier(0.55, 0, 1, 0.45); /* Accelerate - exiting */ --ease-in-out: cubic-bezier(0.65, 0, 0.35, 1); /* Both - moving between */ --spring: cubic-bezier(0.34, 1.56, 0.64, 1); /* Overshoot - playful */ ``` ## Quick Start: Button Microinteraction ```tsx import { motion } from "framer-motion"; export function InteractiveButton({ children, onClick }) { return ( <motion.button onClick={onClick} whileHover={{ scale: 1.02 }} whileTap={{ scale: 0.98 }} transition={{ type: "spring", stiffness: 400, damping: 17 }} className="px-4 py-2 bg-blue-600 text-white rounded-lg" > {children} </motion.button> ); } ``` ## Interaction Patterns ### 1. Loading States **Skeleton Screens**: Preserve layout while loading ```tsx function CardSkeleton() { return ( <div className="animate-pulse"> <div className="h-48 bg-gray-200 rounded-lg" /> <div className="mt-4 h-4 bg-gray-200 rounded w-3/4" /> <div className="mt-2 h-4 bg-gray-200 rounded w-1/2" /> </div> ); } ``` **Progress Indicators**: Show determinate progress ```tsx function ProgressBar({ progress }: { progress: number }) { return ( <div className="h-2 bg-gray-200 rounded-full overflow-hidden"> <motion.div className="h-full bg-blue-600" initial={{ width: 0 }} animate={{ width: `${progress}%` }} transition={{ ease: "easeOut" }} /> </div> ); } ``` ### 2. State Transitions **Toggle with smooth transition**: ```tsx function Toggle({ checked, onChange }) { return ( <button role="switch" aria-checked={checked} onClick={() => onChange(!checked)} className={` relative w-12 h-6 rounded-full transition-colors duration-200 ${checked ? "bg-blue-600" : "bg-gray-300"} `} > <motion.span className="absolute top-1 left-1 w-4 h-4 bg-white rounded-full shadow" animate={{ x: checked ? 24 : 0 }} transition={{ type: "spring", stiffness: 500, damping: 30 }} /> </button> ); } ``` ### 3. Page Transitions **Framer Motion layout animations**: ```tsx import { AnimatePresence, motion } from "framer-motion"; function PageTransition({ children, key }) { return ( <AnimatePresence mode="wait"> <motion.div key={key} initial={{ opacity: 0, y: 20 }} animate={{ opacity: 1, y: 0 }} exit={{ opacity: 0, y: -20 }} transition={{ duration: 0.3 }} > {children} </motion.div> </AnimatePresence> ); } ``` ### 4. Feedback Patterns **Ripple effect on click**: ```tsx function RippleButton({ children, onClick }) { const [ripples, setRipples] = useState([]); const handleClick = (e) => { const rect = e.currentTarget.getBoundingClientRect(); const ripple = { x: e.clientX - rect.left, y: e.clientY - rect.top, id: Date.now(), }; setRipples((prev) => [...prev, ripple]); setTimeout(() => { setRipples((prev) => prev.filter((r) => r.id !== ripple.id)); }, 600); onClick?.(e); }; return ( <button onClick={handleClick} className="relative overflow-hidden"> {children} {ripples.map((ripple) => ( <span key={ripple.id} className="absolute bg-white/30 rounded-full animate-ripple" style={{ left: ripple.x, top: ripple.y }} /> ))} </button> ); } ``` ### 5. Gesture Interactions **Swipe to dismiss**: ```tsx function SwipeCard({ children, onDismiss }) { return ( <motion.div drag="x" dragConstraints={{ left: 0, right: 0 }} onDragEnd={(_, info) => { if (Math.abs(info.offset.x) > 100) { onDismiss(); } }} className="cursor-grab active:cursor-grabbing" > {children} </motion.div> ); } ``` ## CSS Animation Patterns ### Keyframe Animations ```css @keyframes fadeIn { from { opacity: 0; transform: translateY(10px); } to { opacity: 1; transform: translateY(0); } } @keyframes pulse { 0%, 100% { opacity: 1; } 50% { opacity: 0.5; } } @keyframes spin { to { transform: rotate(360deg); } } .animate-fadeIn { animation: fadeIn 0.3s ease-out; } .animate-pulse { animation: pulse 2s ease-in-out infinite; } .animate-spin { animation: spin 1s linear infinite; } ``` ### CSS Transitions ```css .card { transition: transform 0.2s ease-out, box-shadow 0.2s ease-out; } .card:hover { transform: translateY(-4px); box-shadow: 0 12px 24px rgba(0, 0, 0, 0.1); } ``` ## Accessibility Considerations ```css /* Respect user motion preferences */ @media (prefers-reduced-motion: reduce) { *, *::before, *::after { animation-duration: 0.01ms !important; animation-iteration-count: 1 !important; transition-duration: 0.01ms !important; } } ``` ```tsx function AnimatedComponent() { const prefersReducedMotion = window.matchMedia( "(prefers-reduced-motion: reduce)", ).matches; return ( <motion.div animate={{ opacity: 1 }} transition={{ duration: prefersReducedMotion ? 0 : 0.3 }} /> ); } ``` ## Best Practices 1. **Performance First**: Use `transform` and `opacity` for smooth 60fps 2. **Reduce Motion Support**: Always respect `prefers-reduced-motion` 3. **Consistent Timing**: Use a timing scale across the app 4. **Natural Physics**: Prefer spring animations over linear 5. **Interruptible**: Allow users to cancel long animations 6. **Progressive Enhancement**: Work without JS animations 7. **Test on Devices**: Performance varies significantly ## Common Issues - **Janky Animations**: Avoid animating `width`, `height`, `top`, `left` - **Over-animation**: Too much motion causes fatigue - **Blocking Interactions**: Never prevent user input during animations - **Memory Leaks**: Clean up animation listeners on unmount - **Flash of Content**: Use `will-change` sparingly for optimization ## Resources - [Framer Motion Documentation](https://www.framer.com/motion/) - [CSS Animation Guide](https://web.dev/animations-guide/) - [Material Design Motion](https://m3.material.io/styles/motion/overview) - [GSAP Animation Library](https://greensock.com/gsap/)
👍0
👁️0
🤖 Auto-discovered