import redis
import json
from datetime import datetime, timedelta
from celery import Celery
from contextlib import contextmanager

app = Celery('idempotent_jobs')
redis_client = redis.Redis(host='localhost', port=6379, db=0)

class IdempotencyViolation(Exception):
    pass

@contextmanager
def idempotent_job_execution(idempotency_key, ttl_hours=24):
    """
    Context manager for idempotent job execution.
    Ensures a job with the given key can only run once within the TTL window.
    """
    lock_key = f"job_lock:{idempotency_key}"
    result_key = f"job_result:{idempotency_key}"

    # Check if we already have a result
    existing_result = redis_client.get(result_key)
    if existing_result:
        result_data = json.loads(existing_result)
        raise IdempotencyViolation(f"Job already completed: {result_data}")

    # Try to acquire lock
    lock_acquired = redis_client.set(
        lock_key,
        datetime.utcnow().isoformat(),
        nx=True,
        ex=3600  # 1 hour lock timeout
    )

    if not lock_acquired:
        raise IdempotencyViolation(f"Job already running: {idempotency_key}")

    try:
        yield result_key
    finally:
        # Always release the lock
        redis_client.delete(lock_key)

@app.task(bind=True, max_retries=3)
def process_payment_idempotent(self, user_id, subscription_id, amount_cents, billing_period):
    """
    Idempotent payment processing job.
    """
    idempotency_key = f"payment:{user_id}:{subscription_id}:{billing_period}"

    try:
        with idempotent_job_execution(idempotency_key) as result_key:
            # Your actual payment processing logic here
            payment_result = process_payment_logic(user_id, subscription_id, amount_cents)

            # Store the result with TTL
            result_data = {
                'completed_at': datetime.utcnow().isoformat(),
                'payment_id': payment_result['payment_id'],
                'status': 'success'
            }

            redis_client.set(
                result_key,
                json.dumps(result_data),
                ex=86400 * 7  # Keep results for 7 days
            )

            return result_data

    except IdempotencyViolation as e:
        # Job was already processed or is currently running
        self.retry(countdown=60 * (self.request.retries + 1))
    except Exception as e:
        # Other errors should be retried normally
        raise self.retry(exc=e, countdown=60 * (self.request.retries + 1))

def process_payment_logic(user_id, subscription_id, amount_cents):
    """
    Your actual payment processing logic.
    This should also be idempotent at the business logic level.
    """
    # Check if payment already exists for this billing period
    existing_payment = Payment.objects.filter(
        user_id=user_id,
        subscription_id=subscription_id,
        billing_period=billing_period
    ).first()

    if existing_payment:
        return {'payment_id': existing_payment.id, 'status': 'already_processed'}

    # Process the payment
    payment = create_stripe_charge(user_id, amount_cents)
    return {'payment_id': payment.id, 'status': 'processed'}

from django.db import transaction
from django.db.utils import IntegrityError

@app.task(bind=True)
def update_user_analytics_idempotent(self, user_id, event_data, event_timestamp):
    """
    Idempotent analytics update using database-level idempotency.
    """
    idempotency_key = f"analytics:{user_id}:{event_timestamp}:{hash(event_data)}"

    try:
        with transaction.atomic():
            # Try to create an idempotency record
            IdempotencyRecord.objects.create(
                key=idempotency_key,
                created_at=timezone.now()
            )

            # If we get here, this is the first time running this job
            update_user_analytics(user_id, event_data)

    except IntegrityError:
        # This job has already been processed
        pass
    except Exception as e:
        # Delete the idempotency record on failure so job can be retried
        IdempotencyRecord.objects.filter(key=idempotency_key).delete()
        raise

from celery import Celery
from kombu import Queue

# Configure Celery with multiple queues and routing
app = Celery('priority_queues')

app.conf.task_routes = {
    # Critical jobs - highest priority, dedicated workers
    'send_password_reset': {'queue': 'critical'},
    'process_payment': {'queue': 'critical'},
    'send_order_confirmation': {'queue': 'critical'},

    # High priority - important but can wait a bit
    'send_notification': {'queue': 'high'},
    'update_inventory': {'queue': 'high'},
    'detect_fraud': {'queue': 'high'},

    # Medium priority - business important but not user-facing
    'generate_report': {'queue': 'medium'},
    'sync_external_data': {'queue': 'medium'},
    'send_newsletter': {'queue': 'medium'},

    # Low priority - maintenance and cleanup
    'archive_old_logs': {'queue': 'low'},
    'cleanup_temp_files': {'queue': 'low'},
    'run_analytics': {'queue': 'low'}
}

app.conf.task_queues = (
    Queue('critical', routing_key='critical'),
    Queue('high', routing_key='high'),
    Queue('medium', routing_key='medium'),
    Queue('low', routing_key='low'),
)

# Worker configuration for different priorities
app.conf.worker_prefetch_multiplier = 1  # Important for priority queues

# Critical queue - 4 dedicated workers, low concurrency to ensure fast response
celery -A myapp worker -Q critical --concurrency=2 --prefetch-multiplier=1

# High priority - 6 workers with moderate concurrency
celery -A myapp worker -Q high --concurrency=4 --prefetch-multiplier=1

# Medium priority - 4 workers, higher concurrency OK
celery -A myapp worker -Q medium --concurrency=8 --prefetch-multiplier=2

# Low priority - 2 workers, high concurrency, can be interrupted
celery -A myapp worker -Q low --concurrency=12 --prefetch-multiplier=4

import redis
from datetime import datetime, timedelta
from enum import Enum

class JobPriority(Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"

class DynamicPriorityRouter:
    def __init__(self):
        self.redis_client = redis.Redis()

    def get_queue_for_task(self, task_name, args, kwargs):
        """
        Dynamically determine queue based on task and current system state.
        """
        # Base priority from task configuration
        base_priority = self.get_base_priority(task_name)

        # Adjust based on current conditions
        if task_name == 'send_notification':
            # Check if this is a high-value customer
            user_id = kwargs.get('user_id')
            if self.is_premium_customer(user_id):
                return JobPriority.CRITICAL.value

        if task_name == 'generate_report':
            # Check if it's business hours
            if self.is_business_hours():
                return JobPriority.HIGH.value
            else:
                return JobPriority.MEDIUM.value

        # Check system load
        if self.is_system_under_load():
            # Downgrade non-critical tasks during high load
            if base_priority == JobPriority.MEDIUM:
                return JobPriority.LOW.value

        return base_priority.value

    def is_premium_customer(self, user_id):
        # Check Redis cache for customer tier
        tier = self.redis_client.get(f"customer_tier:{user_id}")
        return tier and tier.decode() in ['premium', 'enterprise']

    def is_business_hours(self):
        now = datetime.now()
        return 9 <= now.hour <= 17  # 9 AM to 5 PM

    def is_system_under_load(self):
        # Check queue lengths
        critical_queue_length = self.redis_client.llen('celery:critical')
        high_queue_length = self.redis_client.llen('celery:high')

        return critical_queue_length > 50 or high_queue_length > 200

# Custom Celery router using dynamic priority
class CustomRouter:
    def __init__(self):
        self.priority_router = DynamicPriorityRouter()

    def route_for_task(self, task, args=None, kwargs=None):
        queue = self.priority_router.get_queue_for_task(task, args, kwargs)
        return {'queue': queue}

app.conf.task_routes = (CustomRouter(),)

from celery import Celery
import redis

def get_queue_stats():
    """
    Get comprehensive queue statistics for monitoring.
    """
    redis_client = redis.Redis()

    queues = ['critical', 'high', 'medium', 'low']
    stats = {}

    for queue in queues:
        queue_length = redis_client.llen(f'celery:{queue}')

        # Get oldest job timestamp to calculate wait time
        oldest_job = redis_client.lindex(f'celery:{queue}', 0)
        if oldest_job:
            # Parse job data to get timestamp
            # This is simplified - actual implementation depends on your serializer
            job_data = json.loads(oldest_job)
            oldest_timestamp = job_data.get('eta') or job_data.get('timestamp')
            if oldest_timestamp:
                wait_time = datetime.utcnow() - datetime.fromisoformat(oldest_timestamp)
            else:
                wait_time = None
        else:
            wait_time = None

        stats[queue] = {
            'length': queue_length,
            'oldest_job_wait_time': wait_time.total_seconds() if wait_time else 0
        }

    return stats

# Alert if critical queue wait times exceed thresholds
def check_queue_health():
    stats = get_queue_stats()

    alerts = []

    if stats['critical']['oldest_job_wait_time'] > 30:  # 30 seconds
        alerts.append(f"Critical queue delayed: {stats['critical']['oldest_job_wait_time']}s")

    if stats['high']['oldest_job_wait_time'] > 300:  # 5 minutes
        alerts.append(f"High priority queue delayed: {stats['high']['oldest_job_wait_time']}s")

    if stats['critical']['length'] > 100:
        alerts.append(f"Critical queue backed up: {stats['critical']['length']} jobs")

    return alerts

import redis
import json
from collections import defaultdict
from datetime import datetime, timedelta

def analyze_redis_memory_usage():
    """
    Analyze Redis memory usage patterns for background jobs.
    """
    r = redis.Redis(decode_responses=True)

    # Get all keys with their sizes and TTLs
    memory_stats = defaultdict(lambda: {'count': 0, 'total_size': 0, 'no_expiry': 0})

    for key in r.scan_iter(match="*"):
        key_type = r.type(key)
        key_size = r.memory_usage(key) or 0
        key_ttl = r.ttl(key)

        # Categorize keys by pattern
        if key.startswith('celery-task-meta-'):
            category = 'completed_jobs'
        elif key.startswith('_kombu.binding.'):
            category = 'celery_routing'
        elif key.startswith('failed:'):
            category = 'failed_jobs'
        elif key.startswith('result:'):
            category = 'job_results'
        elif key.startswith('lock:'):
            category = 'locks'
        else:
            category = 'other'

        stats = memory_stats[category]
        stats['count'] += 1
        stats['total_size'] += key_size

        if key_ttl == -1:  # No expiration
            stats['no_expiry'] += 1

    return dict(memory_stats)

def identify_memory_leaks():
    """
    Identify potential memory leaks in job processing.
    """
    stats = analyze_redis_memory_usage()

    issues = []

    # Check for excessive completed jobs without expiry
    completed = stats.get('completed_jobs', {})
    if completed.get('no_expiry', 0) > 1000:
        issues.append({
            'type': 'completed_jobs_no_expiry',
            'count': completed['no_expiry'],
            'size_mb': completed['total_size'] / (1024 * 1024),
            'recommendation': 'Set TTL on completed job metadata'
        })

    # Check for old failed jobs
    failed = stats.get('failed_jobs', {})
    if failed.get('total_size', 0) > 100 * 1024 * 1024:  # 100MB
        issues.append({
            'type': 'failed_jobs_accumulation',
            'count': failed['count'],
            'size_mb': failed['total_size'] / (1024 * 1024),
            'recommendation': 'Clean up old failed jobs or reduce retention'
        })

    return issues

from celery import Celery
from celery.signals import task_success, task_failure
import redis
from datetime import timedelta

app = Celery('memory_managed_jobs')

# Configure Celery with appropriate result backend settings
app.conf.update(
    # Store results but with automatic expiration
    result_backend='redis://localhost:6379/0',
    result_expires=3600,  # Results expire after 1 hour by default

    # Don't store task arguments in results by default
    result_store_errors_even_if_ignored=False,

    # Reduce result serialization overhead
    result_serializer='json',
    result_compression='gzip',

    # Clean up routing data
    worker_hijack_root_logger=False,
)

class MemoryManagedTask:
    """
    Base class for tasks with smart memory management.
    """

    def __init__(self):
        self.redis_client = redis.Redis()

    def apply_retention_policy(self, task_id, task_name, result_size_bytes):
        """
        Apply intelligent retention based on task characteristics.
        """

        # Critical tasks: keep results longer for auditing
        if task_name in ['process_payment', 'send_password_reset']:
            ttl = 86400 * 7  # 7 days

        # Regular tasks: standard retention
        elif task_name in ['send_notification', 'update_inventory']:
            ttl = 3600 * 6  # 6 hours

        # Analytics and reporting: short retention unless large
        elif task_name.startswith('analytics_'):
            if result_size_bytes > 1024 * 1024:  # 1MB
                ttl = 1800  # 30 minutes for large results
            else:
                ttl = 3600 * 12  # 12 hours for small results

        # Default: 1 hour
        else:
            ttl = 3600

        # Set TTL on result
        result_key = f"celery-task-meta-{task_id}"
        self.redis_client.expire(result_key, ttl)

        return ttl

@task_success.connect
def task_success_handler(sender=None, task_id=None, result=None, retval=None, **kwargs):
    """
    Handle successful task completion with memory management.
    """
    memory_manager = MemoryManagedTask()

    # Estimate result size
    result_json = json.dumps(result) if result else "{}"
    result_size = len(result_json.encode('utf-8'))

    # Apply retention policy
    ttl = memory_manager.apply_retention_policy(
        task_id,
        sender.name,
        result_size
    )

    # Log large results for monitoring
    if result_size > 1024 * 1024:  # 1MB
        print(f"Large result warning: {sender.name} produced {result_size/1024/1024:.2f}MB result")

@task_failure.connect
def task_failure_handler(sender=None, task_id=None, exception=None, einfo=None, **kwargs):
    """
    Handle failed tasks with appropriate retention.
    """
    memory_manager = MemoryManagedTask()

    # Failed tasks get shorter retention but enough time for debugging
    if sender.name in ['process_payment', 'send_password_reset']:
        ttl = 86400 * 3  # 3 days for critical task failures
    else:
        ttl = 86400  # 1 day for regular failures

    failure_key = f"celery-task-meta-{task_id}"
    memory_manager.redis_client.expire(failure_key, ttl)

import pickle
import gzip
from celery.backends.redis import RedisBackend

class CompressedRedisBackend(RedisBackend):
    """
    Custom Redis backend with compression for large results.
    """

    def encode(self, data):
        """
        Compress large results before storing.
        """
        serialized = super().encode(data)

        # Only compress if result is larger than 10KB
        if len(serialized) > 10240:
            compressed = gzip.compress(serialized.encode('utf-8'))

            # Only use compression if it actually saves space
            if len(compressed) < len(serialized) * 0.8:
                return {
                    'compressed': True,
                    'data': compressed
                }

        return {'compressed': False, 'data': serialized}

    def decode(self, data):
        """
        Decompress results when retrieving.
        """
        if isinstance(data, dict) and data.get('compressed'):
            decompressed = gzip.decompress(data['data']).decode('utf-8')
            return super().decode(decompressed)

        return super().decode(data.get('data', data))

# Use the compressed backend
app.conf.result_backend = 'myapp.backends.CompressedRedisBackend://localhost:6379/0'

@app.task(bind=True)
def process_large_dataset_streamed(self, dataset_params):
    """
    Process large datasets without storing full results in memory.
    """

    # Instead of returning the full result, stream it to external storage
    result_key = f"large_result:{self.request.id}"

    try:
        # Process data in chunks
        total_records = 0
        for chunk in process_dataset_in_chunks(dataset_params):
            # Store each chunk separately with its own TTL
            chunk_key = f"{result_key}:chunk:{total_records}"
            self.backend.set(chunk_key, chunk)
            self.backend.expire(chunk_key, 1800)  # 30 minutes

            total_records += len(chunk)

        # Return metadata instead of full data
        return {
            'result_type': 'streamed',
            'result_key': result_key,
            'total_records': total_records,
            'chunks': total_records // 1000 + 1,
            'expires_at': (datetime.utcnow() + timedelta(minutes=30)).isoformat()
        }

    except Exception:
        # Clean up partial results on failure
        self.cleanup_partial_results(result_key)
        raise

from celery.beat import PeriodicTask
from datetime import timedelta

@app.task
def monitor_memory_usage():
    """
    Periodic task to monitor and alert on memory usage patterns.
    """

    stats = analyze_redis_memory_usage()
    issues = identify_memory_leaks()

    # Check total memory usage
    info = redis_client.info('memory')
    used_memory_gb = info['used_memory'] / (1024**3)

    alerts = []

    if used_memory_gb > 4.0:  # Alert if using more than 4GB
        alerts.append(f"High Redis memory usage: {used_memory_gb:.2f}GB")

    # Check for growth patterns
    memory_growth = check_memory_growth_rate()
    if memory_growth > 0.1:  # Growing more than 10% per hour
        alerts.append(f"Rapid memory growth detected: {memory_growth*100:.1f}% per hour")

    # Check for keys without expiration
    no_expiry_count = sum(stat.get('no_expiry', 0) for stat in stats.values())
    if no_expiry_count > 10000:
        alerts.append(f"Too many keys without expiration: {no_expiry_count}")

    if alerts:
        send_memory_alert(alerts)

    return {
        'memory_gb': used_memory_gb,
        'issues': issues,
        'alerts': alerts
    }

# Schedule memory monitoring
app.conf.beat_schedule = {
    'monitor-memory': {
        'task': 'monitor_memory_usage',
        'schedule': timedelta(minutes=15),
    },
}

from celery import Celery
from celery.exceptions import Retry
import requests
from datetime import datetime, timedelta
import random
import redis

app = Celery('intelligent_retry')
redis_client = redis.Redis()

class RetryStrategy:
    """
    Intelligent retry strategy with exponential backoff and error classification.
    """

    # Define retriable vs non-retriable errors
    RETRIABLE_EXCEPTIONS = (
        requests.exceptions.ConnectTimeout,
        requests.exceptions.ReadTimeout,
        requests.exceptions.ConnectionError,
        ConnectionError,
        TimeoutError,
    )

    RETRIABLE_HTTP_CODES = {429, 502, 503, 504}

    NON_RETRIABLE_HTTP_CODES = {400, 401, 403, 404, 409, 422}

    def __init__(self, max_retries=5, base_delay=60):
        self.max_retries = max_retries
        self.base_delay = base_delay

    def should_retry(self, exception, retry_count):
        """
        Determine if an exception should trigger a retry.
        """
        # Never retry after max attempts
        if retry_count >= self.max_retries:
            return False

        # Check for retriable exception types
        if isinstance(exception, self.RETRIABLE_EXCEPTIONS):
            return True

        # Check HTTP errors
        if hasattr(exception, 'response') and hasattr(exception.response, 'status_code'):
            status_code = exception.response.status_code

            if status_code in self.NON_RETRIABLE_HTTP_CODES:
                return False

            if status_code in self.RETRIABLE_HTTP_CODES:
                return True

        # Default: don't retry unknown errors
        return False

    def calculate_delay(self, retry_count, jitter=True):
        """
        Calculate delay with exponential backoff and jitter.
        """
        # Exponential backoff: 60, 120, 240, 480, 960 seconds
        delay = self.base_delay * (2 ** retry_count)

        # Add jitter to prevent thundering herd
        if jitter:
            delay = delay * (0.5 + random.random() * 0.5)

        # Cap maximum delay at 30 minutes
        return min(delay, 1800)

def with_intelligent_retry(task_func):
    """
    Decorator for adding intelligent retry logic to tasks.
    """
    def wrapper(self, *args, **kwargs):
        retry_strategy = RetryStrategy()

        try:
            return task_func(self, *args, **kwargs)

        except Exception as exc:
            retry_count = self.request.retries

            if retry_strategy.should_retry(exc, retry_count):
                delay = retry_strategy.calculate_delay(retry_count)

                # Log retry attempt
                print(f"Task {self.name} failed (attempt {retry_count + 1}/{retry_strategy.max_retries}): {exc}")
                print(f"Retrying in {delay:.1f} seconds...")

                # Track retry patterns for monitoring
                retry_key = f"retry_stats:{self.name}"
                redis_client.hincrby(retry_key, 'total_retries', 1)
                redis_client.hincrby(retry_key, f'retry_attempt_{retry_count}', 1)
                redis_client.expire(retry_key, 86400)  # Keep stats for 24 hours

                raise self.retry(exc=exc, countdown=delay, max_retries=retry_strategy.max_retries)
            else:
                # Don't retry - log as permanent failure
                print(f"Task {self.name} permanently failed: {exc}")
                redis_client.hincrby(f"retry_stats:{self.name}", 'permanent_failures', 1)
                raise

    return wrapper

@app.task(bind=True, max_retries=5)
@with_intelligent_retry
def fetch_external_data(self, api_url, user_id):
    """
    Example task with intelligent retry logic.
    """

    try:
        response = requests.get(api_url, timeout=30)
        response.raise_for_status()

        # Process successful response
        return process_api_response(response.json(), user_id)

    except requests.exceptions.HTTPError as e:
        # HTTP errors will be classified by the retry strategy
        raise e
    except Exception as e:
        # Other exceptions get default handling
        raise e

from enum import Enum
from datetime import datetime, timedelta

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Circuit is open, blocking requests
    HALF_OPEN = "half_open"  # Testing if service has recovered

class CircuitBreaker:
    """
    Circuit breaker implementation for external service calls.
    """

    def __init__(self, service_name, failure_threshold=5, recovery_timeout=300, success_threshold=3):
        self.service_name = service_name
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout  # seconds
        self.success_threshold = success_threshold

        self.redis_key = f"circuit_breaker:{service_name}"

    def get_state(self):
        """Get current circuit breaker state."""
        data = redis_client.hgetall(self.redis_key)

        if not data:
            return CircuitState.CLOSED

        state = CircuitState(data.get('state', 'closed'))
        failure_count = int(data.get('failure_count', 0))
        last_failure = data.get('last_failure')
        success_count = int(data.get('success_count', 0))

        # Check if we should transition from OPEN to HALF_OPEN
        if state == CircuitState.OPEN and last_failure:
            last_failure_time = datetime.fromisoformat(last_failure)
            if datetime.utcnow() - last_failure_time > timedelta(seconds=self.recovery_timeout):
                self.set_state(CircuitState.HALF_OPEN)
                return CircuitState.HALF_OPEN

        return state

    def set_state(self, state, failure_count=0, success_count=0):
        """Set circuit breaker state."""
        redis_client.hset(self.redis_key, mapping={
            'state': state.value,
            'failure_count': failure_count,
            'success_count': success_count,
            'updated_at': datetime.utcnow().isoformat()
        })

        redis_client.expire(self.redis_key, 86400)  # Expire after 24 hours of no updates

    def record_success(self):
        """Record a successful operation."""
        state = self.get_state()

        if state == CircuitState.HALF_OPEN:
            # Increment success count
            success_count = int(redis_client.hget(self.redis_key, 'success_count') or 0) + 1
            redis_client.hset(self.redis_key, 'success_count', success_count)

            # If we've had enough successes, close the circuit
            if success_count >= self.success_threshold:
                self.set_state(CircuitState.CLOSED)

        elif state == CircuitState.OPEN:
            # Reset failure count if circuit is open but we somehow got a success
            self.set_state(CircuitState.CLOSED)

    def record_failure(self):
        """Record a failed operation."""
        failure_count = int(redis_client.hget(self.redis_key, 'failure_count') or 0) + 1

        redis_client.hset(self.redis_key, mapping={
            'failure_count': failure_count,
            'last_failure': datetime.utcnow().isoformat(),
            'success_count': 0  # Reset success count on failure
        })

        # Open circuit if failure threshold exceeded
        if failure_count >= self.failure_threshold:
            self.set_state(CircuitState.OPEN, failure_count=failure_count)

    def can_execute(self):
        """Check if requests can be executed."""
        state = self.get_state()
        return state in [CircuitState.CLOSED, CircuitState.HALF_OPEN]

class CircuitBreakerError(Exception):
    """Raised when circuit breaker prevents execution."""
    pass

@app.task(bind=True, max_retries=3)
@with_intelligent_retry
def call_external_service_with_circuit_breaker(self, service_name, api_endpoint, data):
    """
    Task that uses circuit breaker to protect external service calls.
    """
    circuit_breaker = CircuitBreaker(service_name)

    # Check if circuit breaker allows execution
    if not circuit_breaker.can_execute():
        raise CircuitBreakerError(f"Circuit breaker OPEN for {service_name}")

    try:
        # Make the external service call
        response = requests.post(api_endpoint, json=data, timeout=30)
        response.raise_for_status()

        # Record success
        circuit_breaker.record_success()

        return response.json()

    except Exception as e:
        # Record failure
        circuit_breaker.record_failure()
        raise e

@app.task
def analyze_retry_patterns():
    """
    Analyze retry patterns to identify systemic issues.
    """

    retry_stats = {}
    pattern = "retry_stats:*"

    for key in redis_client.scan_iter(match=pattern):
        task_name = key.split(':', 2)[2]  # Extract task name
        stats = redis_client.hgetall(key)

        total_retries = int(stats.get('total_retries', 0))
        permanent_failures = int(stats.get('permanent_failures', 0))

        # Calculate retry distribution
        retry_distribution = {}
        for field, count in stats.items():
            if field.startswith('retry_attempt_'):
                attempt_num = field.split('_')[2]
                retry_distribution[attempt_num] = int(count)

        retry_stats[task_name] = {
            'total_retries': total_retries,
            'permanent_failures': permanent_failures,
            'retry_distribution': retry_distribution,
            'success_rate': calculate_success_rate(task_name, total_retries, permanent_failures)
        }

    # Identify problematic patterns
    alerts = []

    for task_name, stats in retry_stats.items():
        # High retry rate indicates systemic issues
        if stats['total_retries'] > 1000:
            alerts.append(f"High retry volume for {task_name}: {stats['total_retries']} retries")

        # Low success rate after retries
        if stats['success_rate'] < 0.8:
            alerts.append(f"Low success rate for {task_name}: {stats['success_rate']:.1%}")

        # Too many jobs failing on first attempt
        first_attempt_failures = stats['retry_distribution'].get('0', 0)
        if first_attempt_failures > 500:
            alerts.append(f"High first-attempt failure rate for {task_name}: {first_attempt_failures}")

    if alerts:
        send_retry_alerts(alerts)

    return retry_stats

def calculate_success_rate(task_name, total_retries, permanent_failures):
    """
    Calculate success rate accounting for retries.
    """
    # This is simplified - you'd want to track actual success counts too
    if total_retries == 0:
        return 1.0

    # Estimate based on retry/failure ratio
    estimated_total_jobs = total_retries + permanent_failures
    estimated_successes = estimated_total_jobs - permanent_failures

    return estimated_successes / estimated_total_jobs if estimated_total_jobs > 0 else 1.0

from celery import Celery
from celery.signals import task_failure, task_success
import traceback
from datetime import datetime, timedelta
from enum import Enum
import json
import logging

# Configure comprehensive logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = Celery('monitored_jobs')

class JobOutcome(Enum):
    SUCCESS = "success"
    PARTIAL_SUCCESS = "partial_success"
    BUSINESS_FAILURE = "business_failure"
    TECHNICAL_FAILURE = "technical_failure"

class JobMonitor:
    """
    Comprehensive job monitoring and alerting system.
    """

    def __init__(self):
        self.redis_client = redis.Redis()

    def record_job_start(self, task_id, task_name, args, kwargs):
        """Record job start with full context."""
        start_record = {
            'task_id': task_id,
            'task_name': task_name,
            'started_at': datetime.utcnow().isoformat(),
            'args': json.dumps(args),
            'kwargs': json.dumps(kwargs),
            'status': 'running'
        }

        # Store with TTL to prevent accumulation
        key = f"job_execution:{task_id}"
        self.redis_client.hmset(key, start_record)
        self.redis_client.expire(key, 86400)  # 24 hours

        # Track running job count for this task type
        running_key = f"running_jobs:{task_name}"
        self.redis_client.incr(running_key)
        self.redis_client.expire(running_key, 3600)

    def record_job_outcome(self, task_id, task_name, outcome, details=None, metrics=None):
        """Record detailed job outcome."""

        outcome_record = {
            'completed_at': datetime.utcnow().isoformat(),
            'outcome': outcome.value,
            'details': json.dumps(details or {}),
            'metrics': json.dumps(metrics or {})
        }

        # Update the job execution record
        key = f"job_execution:{task_id}"
        self.redis_client.hmset(key, outcome_record)

        # Update aggregate statistics
        stats_key = f"job_stats:{task_name}"
        self.redis_client.hincrby(stats_key, outcome.value, 1)
        self.redis_client.hincrby(stats_key, 'total_executions', 1)
        self.redis_client.expire(stats_key, 86400 * 7)  # Keep for 7 days

        # Decrement running job count
        running_key = f"running_jobs:{task_name}"
        self.redis_client.decr(running_key)

        # Alert on concerning patterns
        self.check_for_alerts(task_name, outcome)

    def check_for_alerts(self, task_name, outcome):
        """Check if this outcome should trigger alerts."""

        if outcome in [JobOutcome.BUSINESS_FAILURE, JobOutcome.TECHNICAL_FAILURE]:
            # Get recent failure rate
            stats = self.get_task_stats(task_name, hours=1)
            total_recent = stats.get('total_executions', 0)
            failures = stats.get('business_failure', 0) + stats.get('technical_failure', 0)

            if total_recent > 5 and failures / total_recent > 0.5:  # 50% failure rate
                self.send_alert(f"High failure rate for {task_name}: {failures}/{total_recent}")

        # Check for stuck jobs (started but never completed)
        stuck_jobs = self.find_stuck_jobs(task_name)
        if len(stuck_jobs) > 5:
            self.send_alert(f"Detected {len(stuck_jobs)} stuck {task_name} jobs")

    def find_stuck_jobs(self, task_name, timeout_minutes=60):
        """Find jobs that started but never completed."""

        stuck_jobs = []
        pattern = "job_execution:*"

        for key in self.redis_client.scan_iter(match=pattern):
            job_data = self.redis_client.hgetall(key)

            if (job_data.get('task_name') == task_name and
                job_data.get('status') == 'running' and
                not job_data.get('completed_at')):

                # Check if job is older than timeout
                started_at = datetime.fromisoformat(job_data['started_at'])
                if datetime.utcnow() - started_at > timedelta(minutes=timeout_minutes):
                    stuck_jobs.append(job_data)

        return stuck_jobs

def monitored_task(task_func):
    """
    Decorator for comprehensive task monitoring.
    """
    def wrapper(self, *args, **kwargs):
        monitor = JobMonitor()
        task_id = self.request.id
        task_name = self.name

        # Record job start
        monitor.record_job_start(task_id, task_name, args, kwargs)

        start_time = datetime.utcnow()

        try:
            # Execute the actual task
            result = task_func(self, *args, **kwargs)

            # Calculate execution metrics
            execution_time = (datetime.utcnow() - start_time).total_seconds()

            # Analyze result to determine outcome
            outcome, details = analyze_task_result(result, task_name)

            metrics = {
                'execution_time_seconds': execution_time,
                'memory_usage': get_memory_usage(),
                'result_size_bytes': len(json.dumps(result)) if result else 0
            }

            monitor.record_job_outcome(task_id, task_name, outcome, details, metrics)

            # Log successful completion
            logger.info(f"Task {task_name}[{task_id}] completed: {outcome.value} in {execution_time:.2f}s")

            return result

        except Exception as exc:
            # Calculate execution metrics for failed job
            execution_time = (datetime.utcnow() - start_time).total_seconds()

            details = {
                'exception_type': type(exc).__name__,
                'exception_message': str(exc),
                'traceback': traceback.format_exc()
            }

            metrics = {
                'execution_time_seconds': execution_time,
                'failed': True
            }

            monitor.record_job_outcome(
                task_id,
                task_name,
                JobOutcome.TECHNICAL_FAILURE,
                details,
                metrics
            )

            # Log failure with full context
            logger.error(f"Task {task_name}[{task_id}] failed after {execution_time:.2f}s: {exc}")
            logger.error(f"Full traceback: {traceback.format_exc()}")

            # Re-raise to maintain normal Celery error handling
            raise

    return wrapper

def analyze_task_result(result, task_name):
    """
    Analyze task result to determine business outcome.
    """

    # Default to success
    outcome = JobOutcome.SUCCESS
    details = {}

    # Check for business logic indicators in result
    if isinstance(result, dict):

        # Look for explicit success/failure indicators
        if 'status' in result:
            if result['status'] in ['failed', 'error', 'business_failure']:
                outcome = JobOutcome.BUSINESS_FAILURE
                details = {'business_reason': result.get('reason', 'Unknown business failure')}
            elif result['status'] in ['partial', 'incomplete']:
                outcome = JobOutcome.PARTIAL_SUCCESS
                details = {'partial_reason': result.get('reason', 'Incomplete execution')}

        # Check for processed item counts
        if 'processed' in result and 'total' in result:
            processed = result['processed']
            total = result['total']

            if processed == 0 and total > 0:
                outcome = JobOutcome.BUSINESS_FAILURE
                details = {'reason': 'No items processed'}
            elif processed < total:
                outcome = JobOutcome.PARTIAL_SUCCESS
                details = {'reason': f'Only {processed}/{total} items processed'}

        # Check for external service errors
        if 'external_errors' in result and result['external_errors']:
            outcome = JobOutcome.PARTIAL_SUCCESS
            details = {'external_errors': result['external_errors']}

    return outcome, details

@app.task(bind=True, max_retries=3)
@monitored_task
def process_user_notifications(self, user_ids, notification_template_id):
    """
    Example task with comprehensive monitoring.
    """

    processed_count = 0
    failed_count = 0
    external_errors = []

    for user_id in user_ids:
        try:
            # Attempt to send notification
            send_notification_to_user(user_id, notification_template_id)
            processed_count += 1

        except ExternalServiceError as e:
            # External service failure - not our fault but impacts business outcome
            external_errors.append(f"User {user_id}: {str(e)}")
            failed_count += 1

        except ValidationError as e:
            # Data validation failure - indicates data quality issues
            logger.warning(f"Invalid data for user {user_id}: {e}")
            failed_count += 1

    # Return detailed result for business outcome analysis
    return {
        'status': 'success' if failed_count == 0 else ('partial' if processed_count > 0 else 'failed'),
        'processed': processed_count,
        'failed': failed_count,
        'total': len(user_ids),
        'external_errors': external_errors,
        'success_rate': processed_count / len(user_ids) if user_ids else 0
    }

@app.task
def system_health_check():
    """
    Comprehensive system health monitoring for background jobs.
    """

    monitor = JobMonitor()
    health_report = {
        'timestamp': datetime.utcnow().isoformat(),
        'overall_health': 'unknown',
        'issues': [],
        'metrics': {}
    }

    # Check for stuck jobs across all task types
    all_stuck_jobs = {}
    for task_name in get_monitored_task_names():
        stuck_jobs = monitor.find_stuck_jobs(task_name)
        if stuck_jobs:
            all_stuck_jobs[task_name] = len(stuck_jobs)

    if all_stuck_jobs:
        health_report['issues'].append({
            'type': 'stuck_jobs',
            'details': all_stuck_jobs,
            'severity': 'high' if sum(all_stuck_jobs.values()) > 20 else 'medium'
        })

    # Check recent failure rates
    high_failure_tasks = []
    for task_name in get_monitored_task_names():
        stats = monitor.get_task_stats(task_name, hours=1)
        total = stats.get('total_executions', 0)
        failures = stats.get('business_failure', 0) + stats.get('technical_failure', 0)

        if total > 10 and failures / total > 0.3:  # 30% failure rate
            high_failure_tasks.append({
                'task_name': task_name,
                'failure_rate': failures / total,
                'total_executions': total
            })

    if high_failure_tasks:
        health_report['issues'].append({
            'type': 'high_failure_rates',
            'details': high_failure_tasks,
            'severity': 'high'
        })

    # Check for missing critical jobs
    missing_critical_jobs = check_critical_job_execution()
    if missing_critical_jobs:
        health_report['issues'].append({
            'type': 'missing_critical_jobs',
            'details': missing_critical_jobs,
            'severity': 'critical'
        })

    # Overall health assessment
    if not health_report['issues']:
        health_report['overall_health'] = 'healthy'
    else:
        severity_levels = [issue['severity'] for issue in health_report['issues']]
        if 'critical' in severity_levels:
            health_report['overall_health'] = 'critical'
        elif 'high' in severity_levels:
            health_report['overall_health'] = 'unhealthy'
        else:
            health_report['overall_health'] = 'degraded'

    # Send alerts if health is poor
    if health_report['overall_health'] in ['critical', 'unhealthy']:
        send_health_alert(health_report)

    return health_report

def check_critical_job_execution():
    """
    Check that critical jobs are running on schedule.
    """

    critical_jobs = {
        'process_payments': {'max_gap_minutes': 60},
        'send_password_resets': {'max_gap_minutes': 5},
        'update_inventory': {'max_gap_minutes': 30}
    }

    missing_jobs = []

    for task_name, config in critical_jobs.items():
        # Check when this task last ran successfully
        last_success = get_last_successful_execution(task_name)

        if last_success:
            gap_minutes = (datetime.utcnow() - last_success).total_seconds() / 60
            if gap_minutes > config['max_gap_minutes']:
                missing_jobs.append({
                    'task_name': task_name,
                    'gap_minutes': gap_minutes,
                    'max_gap_minutes': config['max_gap_minutes']
                })
        else:
            # Never executed - definitely a problem
            missing_jobs.append({
                'task_name': task_name,
                'gap_minutes': float('inf'),
                'max_gap_minutes': config['max_gap_minutes']
            })

    return missing_jobs

@app.task
def business_outcome_monitoring():
    """
    Monitor business outcomes affected by background jobs.
    """

    outcomes = {}

    # Check payment processing outcomes
    payment_stats = analyze_payment_outcomes()
    if payment_stats['success_rate'] < 0.95:  # Less than 95% success
        outcomes['payment_processing'] = {
            'status': 'concerning',
            'success_rate': payment_stats['success_rate'],
            'failed_amount': payment_stats['failed_amount_cents'] / 100,
            'recommendation': 'Review payment processing failures'
        }

    # Check notification delivery rates
    notification_stats = analyze_notification_outcomes()
    if notification_stats['delivery_rate'] < 0.90:  # Less than 90% delivery
        outcomes['notifications'] = {
            'status': 'concerning',
            'delivery_rate': notification_stats['delivery_rate'],
            'undelivered_count': notification_stats['failed_count'],
            'recommendation': 'Check external notification services'
        }

    # Check data synchronization freshness
    sync_stats = analyze_sync_freshness()
    if sync_stats['avg_delay_minutes'] > 60:  # More than 1 hour delay
        outcomes['data_sync'] = {
            'status': 'concerning',
            'avg_delay_minutes': sync_stats['avg_delay_minutes'],
            'stale_records': sync_stats['stale_count'],
            'recommendation': 'Review data synchronization jobs'
        }

    # Alert on business outcome issues
    if outcomes:
        send_business_outcome_alert(outcomes)

    return outcomes

import redis
import time
import uuid
from contextlib import contextmanager
from datetime import datetime, timedelta
from celery import Celery

app = Celery('coordinated_jobs')
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

class DistributedLock:
    """
    Distributed lock implementation using Redis for job coordination.
    """

    def __init__(self, key, timeout=300, polling_interval=0.1):
        self.key = f"lock:{key}"
        self.timeout = timeout
        self.polling_interval = polling_interval
        self.identifier = str(uuid.uuid4())

    def acquire(self, blocking=True, timeout=None):
        """Acquire the distributed lock."""

        end_time = time.time() + (timeout or self.timeout)

        while time.time() < end_time:
            # Try to set the lock with our identifier and expiration
            if redis_client.set(self.key, self.identifier, nx=True, ex=self.timeout):
                return True

            if not blocking:
                return False

            time.sleep(self.polling_interval)

        return False

    def release(self):
        """Release the lock if we own it."""

        # Lua script to atomically check ownership and release
        lua_script = """
        if redis.call('get', KEYS[1]) == ARGV[1] then
            return redis.call('del', KEYS[1])
        else
            return 0
        end
        """

        return redis_client.eval(lua_script, 1, self.key, self.identifier)

    def extend(self, additional_time=300):
        """Extend the lock timeout if we own it."""

        lua_script = """
        if redis.call('get', KEYS[1]) == ARGV[1] then
            return redis.call('expire', KEYS[1], ARGV[2])
        else
            return 0
        end
        """

        return redis_client.eval(lua_script, 1, self.key, self.identifier, additional_time)

@contextmanager
def distributed_lock(lock_key, timeout=300, blocking=True, acquire_timeout=None):
    """
    Context manager for distributed locking.
    """

    lock = DistributedLock(lock_key, timeout)

    acquired = lock.acquire(blocking=blocking, timeout=acquire_timeout)
    if not acquired:
        raise RuntimeError(f"Could not acquire lock: {lock_key}")

    try:
        yield lock
    finally:
        lock.release()

class AtomicCounter:
    """
    Atomic counter operations using Redis for race-condition-safe incrementing.
    """

    def __init__(self, key, redis_client=None):
        self.key = f"counter:{key}"
        self.redis = redis_client or redis_client

    def increment(self, amount=1):
        """Atomically increment the counter."""
        return self.redis.incrby(self.key, amount)

    def decrement(self, amount=1):
        """Atomically decrement the counter."""
        return self.redis.incrby(self.key, -amount)

    def get_value(self):
        """Get current counter value."""
        value = self.redis.get(self.key)
        return int(value) if value else 0

    def set_value(self, value):
        """Set counter to specific value."""
        return self.redis.set(self.key, value)

    def conditional_decrement(self, amount=1, minimum=0):
        """
        Atomically decrement only if result would be >= minimum.
        Returns the new value or None if operation would violate minimum.
        """

        lua_script = """
        local current = tonumber(redis.call('get', KEYS[1]) or 0)
        local new_value = current - tonumber(ARGV[1])
        local minimum = tonumber(ARGV[2])

        if new_value >= minimum then
            redis.call('set', KEYS[1], new_value)
            return new_value
        else
            return nil
        end
        """

        result = self.redis.eval(lua_script, 1, self.key, amount, minimum)
        return result


@app.task(bind=True)
def update_inventory_coordinated(self, product_id, quantity_change):
    """
    Update product inventory with distributed locking to prevent race conditions.
    """

    lock_key = f"inventory_update:{product_id}"

    try:
        with distributed_lock(lock_key, timeout=60, acquire_timeout=30):
            # Now we have exclusive access to this product's inventory

            # Get current inventory (this read is now protected)
            current_inventory = get_product_inventory(product_id)

            # Validate the operation
            if quantity_change < 0 and current_inventory + quantity_change < 0:
                raise ValueError(f"Insufficient inventory. Current: {current_inventory}, Requested: {abs(quantity_change)}")

            # Perform the update atomically
            new_inventory = current_inventory + quantity_change

            # Update database within the same transaction
            with database_transaction():
                update_product_inventory(product_id, new_inventory)
                log_inventory_change(product_id, quantity_change, current_inventory, new_inventory)

            return {
                'product_id': product_id,
                'previous_inventory': current_inventory,
                'new_inventory': new_inventory,
                'change': quantity_change
            }

    except RuntimeError as e:
        # Could not acquire lock - probably another job is updating this product
        self.retry(countdown=5, max_retries=3)

@app.task(bind=True)
def process_order_with_coordination(self, order_id):
    """
    Process order with proper coordination for inventory allocation.
    """

    order = get_order(order_id)
    allocated_items = []

    try:
        # Process each item in the order
        for item in order.items:
            counter = AtomicCounter(f"inventory:{item.product_id}")

            # Atomically check and reserve inventory
            new_inventory = counter.conditional_decrement(
                amount=item.quantity,
                minimum=0
            )

            if new_inventory is None:
                # Insufficient inventory - rollback any previous allocations
                rollback_allocations(allocated_items)
                raise ValueError(f"Insufficient inventory for product {item.product_id}")

            allocated_items.append({
                'product_id': item.product_id,
                'quantity': item.quantity,
                'reserved_inventory': new_inventory
            })

        # All items successfully allocated - proceed with order processing
        finalize_order(order_id, allocated_items)

        return {
            'order_id': order_id,
            'status': 'success',
            'allocated_items': allocated_items
        }

    except Exception as e:
        # Rollback any partial allocations
        rollback_allocations(allocated_items)
        raise

def rollback_allocations(allocated_items):
    """Rollback inventory allocations on order processing failure."""

    for item in allocated_items:
        counter = AtomicCounter(f"inventory:{item['product_id']}")
        counter.increment(item['quantity'])  # Return inventory to pool

class MultiResourceLock:
    """
    Acquire multiple locks in a consistent order to prevent deadlocks.
    """

    def __init__(self, lock_keys, timeout=300):
        # Always acquire locks in sorted order to prevent deadlocks
        self.lock_keys = sorted(lock_keys)
        self.locks = [DistributedLock(key, timeout) for key in self.lock_keys]
        self.acquired_locks = []

    def acquire_all(self, timeout=None):
        """Acquire all locks or fail."""

        try:
            for lock in self.locks:
                if not lock.acquire(blocking=True, timeout=timeout):
                    # Failed to acquire - release any we got
                    self.release_all()
                    return False
                self.acquired_locks.append(lock)

            return True

        except Exception:
            self.release_all()
            raise

    def release_all(self):
        """Release all acquired locks in reverse order."""

        for lock in reversed(self.acquired_locks):
            try:
                lock.release()
            except Exception:
                # Log but continue releasing other locks
                pass

        self.acquired_locks.clear()

@contextmanager
def multi_resource_lock(lock_keys, timeout=300, acquire_timeout=None):
    """Context manager for acquiring multiple locks safely."""

    multi_lock = MultiResourceLock(lock_keys, timeout)

    if not multi_lock.acquire_all(acquire_timeout):
        raise RuntimeError(f"Could not acquire all locks: {lock_keys}")

    try:
        yield multi_lock
    finally:
        multi_lock.release_all()

@app.task(bind=True)
def transfer_inventory_between_warehouses(self, from_warehouse_id, to_warehouse_id, product_id, quantity):
    """
    Transfer inventory between warehouses with proper coordination.
    """

    # Lock both warehouses to prevent race conditions
    lock_keys = [
        f"warehouse_inventory:{from_warehouse_id}:{product_id}",
        f"warehouse_inventory:{to_warehouse_id}:{product_id}"
    ]

    try:
        with multi_resource_lock(lock_keys, acquire_timeout=30):

            # Check source warehouse has enough inventory
            source_counter = AtomicCounter(f"warehouse:{from_warehouse_id}:{product_id}")
            if source_counter.get_value() < quantity:
                raise ValueError(f"Insufficient inventory in source warehouse")

            # Perform the transfer atomically
            source_counter.decrement(quantity)
            dest_counter = AtomicCounter(f"warehouse:{to_warehouse_id}:{product_id}")
            dest_counter.increment(quantity)

            # Log the transfer
            log_inventory_transfer(from_warehouse_id, to_warehouse_id, product_id, quantity)

            return {
                'from_warehouse': from_warehouse_id,
                'to_warehouse': to_warehouse_id,
                'product_id': product_id,
                'quantity': quantity,
                'status': 'completed'
            }

    except RuntimeError as e:
        # Lock acquisition timeout - retry later
        self.retry(countdown=10, max_retries=3)

@app.task
def monitor_coordination_health():
    """
    Monitor the health of distributed coordination mechanisms.
    """

    health_report = {
        'lock_contention': {},
        'deadlock_detection': [],
        'lock_timeouts': {},
        'recommendations': []
    }

    # Analyze lock contention patterns
    lock_pattern = "lock:*"
    active_locks = []

    for key in redis_client.scan_iter(match=lock_pattern):
        lock_info = {
            'key': key,
            'ttl': redis_client.ttl(key),
            'holder': redis_client.get(key)
        }
        active_locks.append(lock_info)

    # Group by lock type to identify bottlenecks
    lock_types = {}
    for lock in active_locks:
        lock_type = lock['key'].split(':')[1]  # Extract lock type
        if lock_type not in lock_types:
            lock_types[lock_type] = 0
        lock_types[lock_type] += 1

    # Identify high contention areas
    for lock_type, count in lock_types.items():
        if count > 10:  # More than 10 concurrent locks of same type
            health_report['lock_contention'][lock_type] = {
                'active_locks': count,
                'severity': 'high' if count > 50 else 'medium'
            }

    # Check for potential deadlocks (locks held too long)
    long_held_locks = []
    for lock in active_locks:
        # If TTL is less than half the original timeout, it's been held a while
        if lock['ttl'] < 150:  # Assuming 300s original timeout
            long_held_locks.append(lock)

    if long_held_locks:
        health_report['deadlock_detection'] = long_held_locks

    # Analyze timeout patterns from job metrics
    timeout_stats = analyze_lock_timeout_patterns()
    health_report['lock_timeouts'] = timeout_stats

    # Generate recommendations
    if health_report['lock_contention']:
        health_report['recommendations'].append(
            "Consider implementing finer-grained locking or queue-based processing for high-contention resources"
        )

    if len(long_held_locks) > 5:
        health_report['recommendations'].append(
            "Review long-running operations that may be holding locks too long"
        )

    return health_report

from enum import Enum
from datetime import datetime, timedelta
import requests
from celery import Celery
from celery.exceptions import Retry

app = Celery('isolated_jobs')

class ServiceHealth(Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"
    CRITICAL = "critical"

class ExternalServiceMonitor:
    """
    Monitor external service health and implement failure isolation.
    """

    def __init__(self):
        self.redis_client = redis.Redis()

        # Define service health thresholds
        self.services = {
            'payment_processor': {
                'url': 'https://api.payments.example.com/health',
                'timeout': 5,
                'critical': True,  # Critical services get priority treatment
                'failure_threshold': 3,
                'recovery_time': 300  # 5 minutes
            },
            'image_optimizer': {
                'url': 'https://api.images.example.com/status',
                'timeout': 10,
                'critical': False,
                'failure_threshold': 5,
                'recovery_time': 600  # 10 minutes
            },
            'email_service': {
                'url': 'https://api.email.example.com/health',
                'timeout': 3,
                'critical': True,
                'failure_threshold': 2,
                'recovery_time': 180  # 3 minutes
            }
        }

    def check_service_health(self, service_name):
        """Check and update service health status."""

        if service_name not in self.services:
            return ServiceHealth.HEALTHY

        service_config = self.services[service_name]
        health_key = f"service_health:{service_name}"

        try:
            # Attempt health check
            response = requests.get(
                service_config['url'],
                timeout=service_config['timeout']
            )

            if response.status_code == 200:
                # Service is healthy - reset failure count
                self.redis_client.delete(f"failures:{service_name}")
                self.redis_client.hset(health_key, mapping={
                    'status': ServiceHealth.HEALTHY.value,
                    'last_check': datetime.utcnow().isoformat(),
                    'consecutive_failures': 0
                })
                return ServiceHealth.HEALTHY
            else:
                # Non-200 response - increment failure count
                return self._record_service_failure(service_name)

        except (requests.exceptions.RequestException, TimeoutError):
            # Network/timeout error - increment failure count
            return self._record_service_failure(service_name)

    def _record_service_failure(self, service_name):
        """Record service failure and update health status."""

        failure_key = f"failures:{service_name}"
        health_key = f"service_health:{service_name}"

        # Increment failure count with sliding window
        current_failures = self.redis_client.incr(failure_key)
        self.redis_client.expire(failure_key, 3600)  # Reset count after 1 hour

        service_config = self.services[service_name]

        # Determine health status based on failure count
        if current_failures >= service_config['failure_threshold']:
            if service_config['critical']:
                status = ServiceHealth.CRITICAL
            else:
                status = ServiceHealth.UNHEALTHY
        else:
            status = ServiceHealth.DEGRADED

        # Update health record
        self.redis_client.hset(health_key, mapping={
            'status': status.value,
            'last_check': datetime.utcnow().isoformat(),
            'consecutive_failures': current_failures
        })

        return status

    def get_service_health(self, service_name):
        """Get current service health status."""

        health_key = f"service_health:{service_name}"
        health_data = self.redis_client.hgetall(health_key)

        if not health_data:
            return ServiceHealth.HEALTHY

        # Check if we need to re-evaluate based on time
        last_check = health_data.get('last_check')
        if last_check:
            last_check_time = datetime.fromisoformat(last_check)
            if datetime.utcnow() - last_check_time > timedelta(minutes=5):
                # Stale data - perform new check
                return self.check_service_health(service_name)

        status = health_data.get('status', ServiceHealth.HEALTHY.value)
        return ServiceHealth(status)

def with_service_health_check(service_name):
    """
    Decorator to check service health before executing dependent jobs.
    """
    def decorator(task_func):
        def wrapper(self, *args, **kwargs):
            monitor = ExternalServiceMonitor()
            health = monitor.get_service_health(service_name)

            if health == ServiceHealth.CRITICAL:
                # Critical service down - fail fast, don't retry immediately
                raise ServiceUnavailableError(f"Service {service_name} is critically unhealthy")

            elif health == ServiceHealth.UNHEALTHY:
                # Service degraded - reduce concurrency and add delays
                if hasattr(self.request, 'retries') and self.request.retries < 2:
                    # Add exponential backoff for degraded services
                    delay = 60 * (2 ** self.request.retries)
                    raise self.retry(countdown=delay, max_retries=3)
                else:
                    # Too many retries - give up
                    raise ServiceUnavailableError(f"Service {service_name} remained unhealthy after retries")

            # Service is healthy or degraded but retryable - proceed
            try:
                return task_func(self, *args, **kwargs)
            except Exception as e:
                # Update service health based on execution results
                if is_service_related_error(e, service_name):
                    monitor.check_service_health(service_name)
                raise

        return wrapper
    return decorator

class ServiceUnavailableError(Exception):
    """Exception raised when required service is unavailable."""
    pass

@app.task(bind=True, max_retries=5)
@with_service_health_check('payment_processor')
def process_payment_with_isolation(self, user_id, amount_cents, payment_method_id):
    """
    Payment processing with service health monitoring and isolation.
    """

    try:
        # Call payment processor
        payment_response = call_payment_api(user_id, amount_cents, payment_method_id)

        # Payment successful - update business logic
        record_successful_payment(user_id, amount_cents, payment_response['payment_id'])

        return {
            'status': 'success',
            'payment_id': payment_response['payment_id'],
            'amount_cents': amount_cents
        }

    except ServiceUnavailableError:
        # Service health check failed - don't retry immediately
        # This job will be retried when service health improves
        raise

    except PaymentProcessingError as e:
        # Business logic error from payment processor
        if e.is_retriable():
            # Temporary payment issue - retry with backoff
            raise self.retry(exc=e, countdown=120, max_retries=3)
        else:
            # Permanent failure - don't retry
            record_failed_payment(user_id, amount_cents, str(e))
            raise

# Configure separate worker pools for different service dependencies
app.conf.task_routes = {
    # Critical services get dedicated, smaller pools to prevent resource exhaustion
    'process_payment_with_isolation': {
        'queue': 'payments',
        'options': {'priority': 10}
    },
    'send_password_reset': {
        'queue': 'critical_emails',
        'options': {'priority': 10}
    },

    # Non-critical external services get separate pools
    'optimize_product_images': {
        'queue': 'image_processing',
        'options': {'priority': 3}
    },
    'sync_analytics_data': {
        'queue': 'analytics',
        'options': {'priority': 1}
    }
}

# Deploy workers with different concurrency settings
# Critical services: Low concurrency to prevent thundering herd
# celery -A myapp worker -Q payments --concurrency=2 --prefetch-multiplier=1

# Non-critical services: Higher concurrency, can fail without impact
# celery -A myapp worker -Q image_processing --concurrency=8 --prefetch-multiplier=4

class SystemHealthCoordinator:
    """
    Coordinate system-wide responses to cascading failures.
    """

    def __init__(self):
        self.redis_client = redis.Redis()
        self.service_monitor = ExternalServiceMonitor()

    def assess_system_health(self):
        """Assess overall system health and determine protective actions."""

        health_status = {
            'overall_health': 'healthy',
            'critical_services_down': [],
            'degraded_services': [],
            'protective_actions': []
        }

        # Check all monitored services
        for service_name in self.service_monitor.services.keys():
            service_health = self.service_monitor.get_service_health(service_name)
            service_config = self.service_monitor.services[service_name]

            if service_health == ServiceHealth.CRITICAL:
                if service_config['critical']:
                    health_status['critical_services_down'].append(service_name)
                    health_status['overall_health'] = 'critical'
            elif service_health in [ServiceHealth.UNHEALTHY, ServiceHealth.DEGRADED]:
                health_status['degraded_services'].append(service_name)
                if health_status['overall_health'] == 'healthy':
                    health_status['overall_health'] = 'degraded'

        # Determine protective actions
        if health_status['critical_services_down']:
            health_status['protective_actions'].extend([
                'enable_circuit_breakers',
                'reduce_retry_rates',
                'activate_degraded_mode'
            ])

        if len(health_status['degraded_services']) > 2:
            health_status['protective_actions'].extend([
                'increase_timeouts',
                'reduce_concurrency',
                'defer_non_critical_jobs'
            ])

        return health_status

    def execute_protective_actions(self, actions):
        """Execute system-wide protective actions."""

        for action in actions:
            if action == 'enable_circuit_breakers':
                self._enable_circuit_breakers()
            elif action == 'reduce_retry_rates':
                self._reduce_retry_rates()
            elif action == 'activate_degraded_mode':
                self._activate_degraded_mode()
            elif action == 'defer_non_critical_jobs':
                self._defer_non_critical_jobs()

    def _enable_circuit_breakers(self):
        """Enable circuit breakers for all external services."""

        circuit_breaker_config = {
            'enabled': True,
            'failure_threshold': 3,
            'recovery_timeout': 300,
            'enabled_at': datetime.utcnow().isoformat()
        }

        self.redis_client.hset('system_config:circuit_breakers', mapping=circuit_breaker_config)

    def _reduce_retry_rates(self):
        """Reduce retry rates system-wide to prevent thundering herd."""

        retry_config = {
            'base_delay_multiplier': 2.0,  # Double retry delays
            'max_retries_reduction': 0.5,  # Reduce max retries by half
            'enabled_at': datetime.utcnow().isoformat()
        }

        self.redis_client.hset('system_config:retry_reduction', mapping=retry_config)

    def _defer_non_critical_jobs(self):
        """Defer non-critical jobs to preserve resources."""

        defer_config = {
            'defer_queues': ['analytics', 'image_processing', 'reporting'],
            'defer_duration_minutes': 30,
            'enabled_at': datetime.utcnow().isoformat()
        }

        self.redis_client.hset('system_config:job_deferral', mapping=defer_config)

@app.task
def system_health_monitor():
    """
    Periodic system health monitoring and cascade prevention.
    """

    coordinator = SystemHealthCoordinator()
    health_status = coordinator.assess_system_health()

    # Execute protective actions if needed
    if health_status['protective_actions']:
        coordinator.execute_protective_actions(health_status['protective_actions'])

        # Alert operations team
        send_system_health_alert(health_status)

    return health_status

# Schedule health monitoring every 2 minutes
app.conf.beat_schedule = {
    'system-health-monitor': {
        'task': 'system_health_monitor',
        'schedule': timedelta(minutes=2),
    },
}

@app.task
def orchestrate_system_recovery():
    """
    Orchestrate gradual system recovery after cascading failures.
    """

    coordinator = SystemHealthCoordinator()
    health_status = coordinator.assess_system_health()

    if health_status['overall_health'] == 'healthy':
        # All services healthy - begin recovery process

        recovery_plan = {
            'phase_1': 'restore_critical_job_processing',
            'phase_2': 'gradually_increase_concurrency',
            'phase_3': 'resume_non_critical_jobs',
            'phase_4': 'disable_protective_measures'
        }

        execute_recovery_phase(recovery_plan['phase_1'])

    return {'status': 'recovery_initiated', 'health': health_status}

def execute_recovery_phase(phase):
    """Execute specific recovery phase with monitoring."""

    if phase == 'restore_critical_job_processing':
        # Re-enable critical job queues with reduced concurrency
        enable_queue_processing(['payments', 'critical_emails'], concurrency=1)

        # Monitor for 5 minutes before proceeding
        schedule_recovery_check('phase_2', delay_minutes=5)

    elif phase == 'gradually_increase_concurrency':
        # Gradually increase worker concurrency
        increase_queue_concurrency(['payments', 'critical_emails'], target_concurrency=2)

        schedule_recovery_check('phase_3', delay_minutes=10)

    elif phase == 'resume_non_critical_jobs':
        # Resume non-critical job processing
        enable_queue_processing(['image_processing', 'analytics'], concurrency=4)

        schedule_recovery_check('phase_4', delay_minutes=15)

    elif phase == 'disable_protective_measures':
        # Disable protective measures and return to normal operation
        coordinator = SystemHealthCoordinator()
        coordinator.disable_protective_measures()

import json
from datetime import datetime, timedelta
from collections import defaultdict, Counter
from celery import Celery
from celery.signals import task_failure
import redis

app = Celery('managed_dead_letters')
redis_client = redis.Redis()

class DeadLetterManager:
    """
    Comprehensive dead letter queue management with analysis and recovery.
    """

    def __init__(self):
        self.redis_client = redis.Redis()

        # Define failure categories for analysis
        self.failure_categories = {
            'network_timeout': [
                'requests.exceptions.ConnectTimeout',
                'requests.exceptions.ReadTimeout',
                'ConnectionError',
                'TimeoutError'
            ],
            'service_unavailable': [
                'requests.exceptions.ConnectionError',
                'ServiceUnavailableError',
                'HTTPError: 503',
                'HTTPError: 502'
            ],
            'authentication_error': [
                'AuthenticationError',
                'HTTPError: 401',
                'HTTPError: 403'
            ],
            'business_logic_error': [
                'ValidationError',
                'BusinessRuleViolation',
                'InsufficientFunds',
                'InvalidStateTransition'
            ],
            'data_error': [
                'KeyError',
                'AttributeError',
                'ValueError',
                'TypeError'
            ]
        }

    def record_dead_letter(self, task_id, task_name, args, kwargs, exception_info, retry_count):
        """Record detailed information about failed jobs."""

        dead_letter_record = {
            'task_id': task_id,
            'task_name': task_name,
            'args': json.dumps(args),
            'kwargs': json.dumps(kwargs),
            'exception_type': exception_info.get('type', 'Unknown'),
            'exception_message': exception_info.get('message', ''),
            'exception_traceback': exception_info.get('traceback', ''),
            'retry_count': retry_count,
            'failed_at': datetime.utcnow().isoformat(),
            'category': self._categorize_failure(exception_info),
            'recoverable': self._assess_recoverability(exception_info, task_name),
            'business_impact': self._assess_business_impact(task_name, args, kwargs)
        }

        # Store in dead letter queue with metadata
        dead_letter_key = f"dead_letter:{task_id}"
        self.redis_client.hmset(dead_letter_key, dead_letter_record)

        # Add to category index for analysis
        category_key = f"dead_letters_by_category:{dead_letter_record['category']}"
        self.redis_client.sadd(category_key, task_id)

        # Add to task type index
        task_type_key = f"dead_letters_by_task:{task_name}"
        self.redis_client.sadd(task_type_key, task_id)

        # Set expiration (keep for 90 days by default)
        self.redis_client.expire(dead_letter_key, 86400 * 90)
        self.redis_client.expire(category_key, 86400 * 90)
        self.redis_client.expire(task_type_key, 86400 * 90)

        return dead_letter_record

    def _categorize_failure(self, exception_info):
        """Categorize the failure type for analysis."""

        exception_type = exception_info.get('type', '')
        exception_message = exception_info.get('message', '')

        for category, patterns in self.failure_categories.items():
            for pattern in patterns:
                if pattern in exception_type or pattern in exception_message:
                    return category

        return 'unknown'

    def _assess_recoverability(self, exception_info, task_name):
        """Determine if a failed job might be recoverable."""

        category = self._categorize_failure(exception_info)

        # Network and service issues are often recoverable
        if category in ['network_timeout', 'service_unavailable']:
            return 'likely_recoverable'

        # Authentication might be recoverable if credentials are fixed
        if category == 'authentication_error':
            return 'conditionally_recoverable'

        # Business logic and data errors are usually not recoverable without changes
        if category in ['business_logic_error', 'data_error']:
            return 'unlikely_recoverable'

        return 'unknown'

    def _assess_business_impact(self, task_name, args, kwargs):
        """Assess the business impact of the failed job."""

        # Critical business processes
        if task_name in ['process_payment', 'send_password_reset', 'create_user_account']:
            return 'high'

        # Important but not critical
        if task_name in ['send_order_confirmation', 'update_inventory', 'send_notification']:
            return 'medium'

        # Analytics and reporting
        if task_name.startswith('analytics_') or 'report' in task_name.lower():
            return 'low'

        return 'unknown'

    def analyze_dead_letters(self, time_window_hours=24):
        """Analyze patterns in dead letter queue."""

        analysis = {
            'total_failed_jobs': 0,
            'failure_by_category': Counter(),
            'failure_by_task': Counter(),
            'recoverable_jobs': [],
            'high_impact_failures': [],
            'trending_failures': [],
            'recommendations': []
        }

        # Get all dead letters from time window
        cutoff_time = datetime.utcnow() - timedelta(hours=time_window_hours)

        dead_letter_keys = list(self.redis_client.scan_iter(match="dead_letter:*"))

        for key in dead_letter_keys:
            dead_letter_data = self.redis_client.hgetall(key)

            if not dead_letter_data:
                continue

            failed_at = datetime.fromisoformat(dead_letter_data['failed_at'])
            if failed_at < cutoff_time:
                continue

            analysis['total_failed_jobs'] += 1

            # Count by category
            category = dead_letter_data.get('category', 'unknown')
            analysis['failure_by_category'][category] += 1

            # Count by task type
            task_name = dead_letter_data.get('task_name', 'unknown')
            analysis['failure_by_task'][task_name] += 1

            # Identify recoverable jobs
            if dead_letter_data.get('recoverable') == 'likely_recoverable':
                analysis['recoverable_jobs'].append({
                    'task_id': dead_letter_data['task_id'],
                    'task_name': task_name,
                    'failed_at': dead_letter_data['failed_at'],
                    'category': category
                })

            # Identify high-impact failures
            if dead_letter_data.get('business_impact') == 'high':
                analysis['high_impact_failures'].append({
                    'task_id': dead_letter_data['task_id'],
                    'task_name': task_name,
                    'failed_at': dead_letter_data['failed_at'],
                    'exception_message': dead_letter_data.get('exception_message', '')
                })

        # Generate recommendations based on analysis
        analysis['recommendations'] = self._generate_recommendations(analysis)

        return analysis

    def _generate_recommendations(self, analysis):
        """Generate actionable recommendations based on failure analysis."""

        recommendations = []

        # Check for high rates of recoverable failures
        recoverable_count = len(analysis['recoverable_jobs'])
        if recoverable_count > 10:
            recommendations.append({
                'type': 'recoverable_failures',
                'priority': 'high',
                'message': f'{recoverable_count} jobs failed with recoverable errors. Consider implementing better retry logic.',
                'action': 'review_retry_configuration'
            })

        # Check for specific failure patterns
        if analysis['failure_by_category']['network_timeout'] > 20:
            recommendations.append({
                'type': 'network_issues',
                'priority': 'medium',
                'message': 'High number of network timeout failures. Consider increasing timeout values or implementing circuit breakers.',
                'action': 'review_network_configuration'
            })

        # Check for task-specific issues
        for task_name, failure_count in analysis['failure_by_task'].most_common(3):
            if failure_count > 15:
                recommendations.append({
                    'type': 'task_specific_failure',
                    'priority': 'high',
                    'message': f'Task {task_name} has {failure_count} failures. Investigate task implementation.',
                    'action': f'debug_task:{task_name}'
                })

        # Check for high-impact failures
        high_impact_count = len(analysis['high_impact_failures'])
        if high_impact_count > 5:
            recommendations.append({
                'type': 'business_impact',
                'priority': 'critical',
                'message': f'{high_impact_count} high-impact jobs failed. Immediate attention required.',
                'action': 'escalate_to_oncall'
            })

        return recommendations

@task_failure.connect
def handle_task_failure(sender=None, task_id=None, exception=None, traceback=None, einfo=None, **kwargs):
    """Handle task failures and record in dead letter queue."""

    dead_letter_manager = DeadLetterManager()

    # Extract exception information
    exception_info = {
        'type': type(exception).__name__ if exception else 'Unknown',
        'message': str(exception) if exception else 'No message',
        'traceback': str(traceback) if traceback else ''
    }

    # Get retry count from task request
    retry_count = getattr(sender.request, 'retries', 0) if hasattr(sender, 'request') else 0

    # Record in dead letter queue
    dead_letter_manager.record_dead_letter(
        task_id=task_id,
        task_name=sender.name if sender else 'unknown',
        args=getattr(sender.request, 'args', []) if hasattr(sender, 'request') else [],
        kwargs=getattr(sender.request, 'kwargs', {}) if hasattr(sender, 'request') else {},
        exception_info=exception_info,
        retry_count=retry_count
    )

@app.task
def dead_letter_recovery_workflow():
    """
    Automated workflow for recovering failed jobs where appropriate.
    """

    manager = DeadLetterManager()
    analysis = manager.analyze_dead_letters(time_window_hours=1)  # Check last hour

    recovery_results = {
        'attempted_recoveries': 0,
        'successful_recoveries': 0,
        'failed_recoveries': 0,
        'recovery_details': []
    }

    # Process recoverable jobs
    for recoverable_job in analysis['recoverable_jobs']:

        # Get full job details
        dead_letter_key = f"dead_letter:{recoverable_job['task_id']}"
        job_data = manager.redis_client.hgetall(dead_letter_key)

        if not job_data:
            continue

        # Check if enough time has passed since failure (avoid immediate retry)
        failed_at = datetime.fromisoformat(job_data['failed_at'])
        if datetime.utcnow() - failed_at < timedelta(minutes=30):
            continue  # Wait at least 30 minutes before recovery attempt

        # Check if we've already attempted recovery
        recovery_key = f"recovery_attempt:{recoverable_job['task_id']}"
        if manager.redis_client.exists(recovery_key):
            continue  # Already attempted recovery

        # Mark recovery attempt
        manager.redis_client.set(recovery_key, datetime.utcnow().isoformat(), ex=86400)

        try:
            # Attempt to re-queue the job
            task_name = job_data['task_name']
            args = json.loads(job_data.get('args', '[]'))
            kwargs = json.loads(job_data.get('kwargs', '{}'))

            # Re-queue with special marking for monitoring
            kwargs['_recovery_attempt'] = True
            kwargs['_original_failure_time'] = job_data['failed_at']

            # Send to appropriate queue based on task
            app.send_task(task_name, args=args, kwargs=kwargs)

            recovery_results['attempted_recoveries'] += 1
            recovery_results['successful_recoveries'] += 1

            recovery_results['recovery_details'].append({
                'task_id': recoverable_job['task_id'],
                'task_name': task_name,
                'status': 're_queued',
                'recovery_time': datetime.utcnow().isoformat()
            })

        except Exception as e:
            recovery_results['attempted_recoveries'] += 1
            recovery_results['failed_recoveries'] += 1

            recovery_results['recovery_details'].append({
                'task_id': recoverable_job['task_id'],
                'task_name': job_data['task_name'],
                'status': 'recovery_failed',
                'error': str(e),
                'recovery_time': datetime.utcnow().isoformat()
            })

    return recovery_results

@app.task
def dead_letter_health_report():
    """
    Generate comprehensive dead letter queue health report.
    """

    manager = DeadLetterManager()

    # Analyze different time windows
    hourly_analysis = manager.analyze_dead_letters(time_window_hours=1)
    daily_analysis = manager.analyze_dead_letters(time_window_hours=24)
    weekly_analysis = manager.analyze_dead_letters(time_window_hours=168)

    health_report = {
        'timestamp': datetime.utcnow().isoformat(),
        'summary': {
            'hourly_failures': hourly_analysis['total_failed_jobs'],
            'daily_failures': daily_analysis['total_failed_jobs'],
            'weekly_failures': weekly_analysis['total_failed_jobs']
        },
        'trends': analyze_failure_trends(hourly_analysis, daily_analysis, weekly_analysis),
        'top_failing_tasks': daily_analysis['failure_by_task'].most_common(5),
        'failure_categories': dict(daily_analysis['failure_by_category']),
        'actionable_items': {
            'high_impact_failures': len(daily_analysis['high_impact_failures']),
            'recoverable_failures': len(daily_analysis['recoverable_jobs']),
            'recommendations': daily_analysis['recommendations']
        },
        'health_score': calculate_dead_letter_health_score(daily_analysis)
    }

    # Send alerts if health score is poor
    if health_report['health_score'] < 70:
        send_dead_letter_alert(health_report)

    return health_report

def calculate_dead_letter_health_score(analysis):
    """Calculate a health score based on failure patterns."""

    score = 100

    # Penalize high failure rates
    if analysis['total_failed_jobs'] > 100:
        score -= 20
    elif analysis['total_failed_jobs'] > 50:
        score -= 10

    # Penalize high-impact failures
    high_impact_count = len(analysis['high_impact_failures'])
    score -= min(high_impact_count * 5, 30)

    # Penalize unrecoverable failures
    total_failures = analysis['total_failed_jobs']
    recoverable_count = len(analysis['recoverable_jobs'])

    if total_failures > 0:
        recoverable_ratio = recoverable_count / total_failures
        if recoverable_ratio < 0.3:  # Less than 30% recoverable
            score -= 15

    # Bonus for low failure rates
    if analysis['total_failed_jobs'] < 5:
        score += 10

    return max(0, score)

# Schedule regular dead letter analysis
app.conf.beat_schedule = {
    'dead-letter-analysis': {
        'task': 'dead_letter_health_report',
        'schedule': timedelta(hours=6),
    },
    'dead-letter-recovery': {
        'task': 'dead_letter_recovery_workflow',
        'schedule': timedelta(hours=2),
    },
}

import gzip
import json
import hashlib
import tempfile
from datetime import datetime, timedelta
from celery import Celery
from io import BytesIO
import boto3
import redis

app = Celery('payload_managed_jobs')
redis_client = redis.Redis()
s3_client = boto3.client('s3')

class PayloadManager:
    """
    Intelligent payload management with size-based handling strategies.
    """

    def __init__(self, compression_threshold=10240, external_storage_threshold=1048576):
        self.compression_threshold = compression_threshold  # 10KB
        self.external_storage_threshold = external_storage_threshold  # 1MB
        self.redis_client = redis.Redis()

        # Configure S3 for large payload storage
        self.s3_bucket = 'job-payloads-bucket'  # Replace with your bucket
        self.s3_prefix = 'job-payloads/'

    def prepare_payload(self, payload_data):
        """
        Prepare payload for job processing with size-appropriate handling.
        """

        # Serialize payload to measure size
        serialized_payload = json.dumps(payload_data, default=str)
        payload_size = len(serialized_payload.encode('utf-8'))

        payload_info = {
            'original_size_bytes': payload_size,
            'strategy': 'direct',
            'compressed': False,
            'external_storage': False,
            'payload_id': None
        }

        # Determine handling strategy based on size
        if payload_size < self.compression_threshold:
            # Small payload - send directly
            payload_info['processed_payload'] = payload_data

        elif payload_size < self.external_storage_threshold:
            # Medium payload - compress
            compressed_data = self._compress_payload(serialized_payload)
            compression_ratio = len(compressed_data) / payload_size

            if compression_ratio < 0.8:  # Compression saves at least 20%
                payload_info.update({
                    'strategy': 'compressed',
                    'compressed': True,
                    'compressed_size_bytes': len(compressed_data),
                    'compression_ratio': compression_ratio,
                    'processed_payload': compressed_data
                })
            else:
                # Compression not effective - use direct
                payload_info['processed_payload'] = payload_data

        else:
            # Large payload - use external storage
            payload_id = self._store_payload_externally(serialized_payload)
            payload_info.update({
                'strategy': 'external_storage',
                'external_storage': True,
                'payload_id': payload_id,
                'processed_payload': {'_external_payload_id': payload_id}
            })

        return payload_info

    def retrieve_payload(self, processed_payload, payload_info):
        """
        Retrieve original payload data based on processing strategy.
        """

        if payload_info['strategy'] == 'direct':
            return processed_payload

        elif payload_info['strategy'] == 'compressed':
            return self._decompress_payload(processed_payload)

        elif payload_info['strategy'] == 'external_storage':
            payload_id = processed_payload.get('_external_payload_id')
            if not payload_id:
                raise ValueError("Missing external payload ID")
            return self._retrieve_payload_externally(payload_id)

        else:
            raise ValueError(f"Unknown payload strategy: {payload_info['strategy']}")

    def _compress_payload(self, serialized_data):
        """Compress payload data using gzip."""

        with BytesIO() as buffer:
            with gzip.GzipFile(fileobj=buffer, mode='wb') as gz_file:
                gz_file.write(serialized_data.encode('utf-8'))
            return buffer.getvalue()

    def _decompress_payload(self, compressed_data):
        """Decompress payload data."""

        with BytesIO(compressed_data) as buffer:
            with gzip.GzipFile(fileobj=buffer, mode='rb') as gz_file:
                decompressed_data = gz_file.read().decode('utf-8')
                return json.loads(decompressed_data)

    def _store_payload_externally(self, serialized_data):
        """Store large payload in S3 and return reference ID."""

        payload_id = f"{datetime.utcnow().strftime('%Y%m%d')}/{hashlib.sha256(serialized_data.encode()).hexdigest()[:16]}"
        s3_key = f"{self.s3_prefix}{payload_id}.json"

        try:
            # Store in S3 with metadata
            s3_client.put_object(
                Bucket=self.s3_bucket,
                Key=s3_key,
                Body=serialized_data.encode('utf-8'),
                ContentType='application/json',
                Metadata={
                    'created_at': datetime.utcnow().isoformat(),
                    'size_bytes': str(len(serialized_data.encode('utf-8')))
                }
            )

            # Store reference in Redis with TTL
            reference_key = f"payload_ref:{payload_id}"
            self.redis_client.hset(reference_key, mapping={
                's3_bucket': self.s3_bucket,
                's3_key': s3_key,
                'created_at': datetime.utcnow().isoformat(),
                'size_bytes': len(serialized_data.encode('utf-8'))
            })
            self.redis_client.expire(reference_key, 86400 * 7)  # 7 days

            return payload_id

        except Exception as e:
            raise RuntimeError(f"Failed to store payload externally: {e}")

    def _retrieve_payload_externally(self, payload_id):
        """Retrieve payload from external storage."""

        reference_key = f"payload_ref:{payload_id}"
        reference_data = self.redis_client.hgetall(reference_key)

        if not reference_data:
            raise ValueError(f"Payload reference not found: {payload_id}")

        try:
            # Retrieve from S3
            response = s3_client.get_object(
                Bucket=reference_data['s3_bucket'],
                Key=reference_data['s3_key']
            )

            payload_data = response['Body'].read().decode('utf-8')
            return json.loads(payload_data)

        except Exception as e:
            raise RuntimeError(f"Failed to retrieve payload from external storage: {e}")

    def cleanup_external_payload(self, payload_id):
        """Clean up external payload after processing."""

        reference_key = f"payload_ref:{payload_id}"
        reference_data = self.redis_client.hgetall(reference_key)

        if reference_data:
            try:
                # Delete from S3
                s3_client.delete_object(
                    Bucket=reference_data['s3_bucket'],
                    Key=reference_data['s3_key']
                )

                # Delete reference
                self.redis_client.delete(reference_key)

            except Exception as e:
                # Log error but don't fail the job
                print(f"Warning: Failed to clean up external payload {payload_id}: {e}")

def smart_payload_task(task_func):
    """
    Decorator for automatic payload management in tasks.
    """

    def wrapper(self, *args, **kwargs):
        payload_manager = PayloadManager()

        # Check if this is a payload-managed job
        if '_payload_info' in kwargs:
            payload_info = kwargs.pop('_payload_info')
            processed_payload = kwargs.pop('_processed_payload')

            # Retrieve original payload
            try:
                original_payload = payload_manager.retrieve_payload(processed_payload, payload_info)

                # Replace processed payload with original in kwargs
                kwargs['payload'] = original_payload

                # Execute task with original payload
                result = task_func(self, *args, **kwargs)

                # Clean up external payload if used
                if payload_info.get('external_storage') and payload_info.get('payload_id'):
                    payload_manager.cleanup_external_payload(payload_info['payload_id'])

                return result

            except Exception as e:
                # Clean up on failure too
                if payload_info.get('external_storage') and payload_info.get('payload_id'):
                    payload_manager.cleanup_external_payload(payload_info['payload_id'])
                raise
        else:
            # Regular task execution
            return task_func(self, *args, **kwargs)

    return wrapper

@app.task(bind=True)
@smart_payload_task
def process_bulk_data_with_payload_management(self, payload):
    """
    Process bulk data with automatic payload size management.
    """

    # The payload has been automatically retrieved by the decorator
    processed_records = 0
    failed_records = 0

    # Process data in chunks to manage memory usage
    for chunk in chunk_data(payload, chunk_size=1000):
        try:
            process_data_chunk(chunk)
            processed_records += len(chunk)
        except Exception as e:
            print(f"Failed to process chunk: {e}")
            failed_records += len(chunk)

    return {
        'processed_records': processed_records,
        'failed_records': failed_records,
        'total_records': len(payload) if isinstance(payload, list) else 1
    }

def queue_job_with_smart_payload(task_name, payload_data, **kwargs):
    """
    Queue a job with intelligent payload management.
    """

    payload_manager = PayloadManager()

    # Prepare payload based on size
    payload_info = payload_manager.prepare_payload(payload_data)

    # Log payload handling strategy
    print(f"Queueing job {task_name} with {payload_info['strategy']} strategy "
          f"(size: {payload_info['original_size_bytes']} bytes)")

    # Add payload info to job arguments
    kwargs.update({
        '_payload_info': payload_info,
        '_processed_payload': payload_info['processed_payload']
    })

    # Queue the job
    return app.send_task(task_name, kwargs=kwargs)

@app.task(bind=True)
def process_large_dataset_chunked(self, dataset_id, chunk_size=1000):
    """
    Process large datasets by breaking them into chunks.
    """

    # Instead of loading entire dataset into memory, process it in chunks
    dataset_info = get_dataset_metadata(dataset_id)
    total_records = dataset_info['record_count']

    if total_records <= chunk_size:
        # Small dataset - process directly
        return process_small_dataset(dataset_id)

    # Large dataset - create subtasks for each chunk
    chunk_tasks = []
    processed_chunks = 0

    for offset in range(0, total_records, chunk_size):
        # Create a subtask for each chunk
        chunk_task = process_dataset_chunk.delay(
            dataset_id=dataset_id,
            offset=offset,
            limit=chunk_size,
            parent_task_id=self.request.id
        )

        chunk_tasks.append(chunk_task.id)
        processed_chunks += 1

    # Monitor chunk processing
    return {
        'status': 'chunked',
        'total_records': total_records,
        'chunk_count': processed_chunks,
        'chunk_tasks': chunk_tasks,
        'dataset_id': dataset_id
    }

@app.task(bind=True)
def process_dataset_chunk(self, dataset_id, offset, limit, parent_task_id=None):
    """
    Process a single chunk of a large dataset.
    """

    try:
        # Load only the required chunk of data
        chunk_data = load_dataset_chunk(dataset_id, offset, limit)

        # Process the chunk
        results = []
        for record in chunk_data:
            result = process_single_record(record)
            results.append(result)

        # Update parent task progress if specified
        if parent_task_id:
            update_parent_task_progress(parent_task_id, offset, limit)

        return {
            'status': 'completed',
            'processed_count': len(results),
            'offset': offset,
            'limit': limit,
            'parent_task_id': parent_task_id
        }

    except Exception as e:
        # Record chunk failure
        if parent_task_id:
            record_chunk_failure(parent_task_id, offset, limit, str(e))
        raise

@app.task
def monitor_chunked_job_progress(parent_task_id):
    """
    Monitor progress of a chunked job and aggregate results.
    """

    # Get all chunks for this parent task
    chunk_results = get_chunk_results(parent_task_id)

    total_chunks = len(chunk_results)
    completed_chunks = sum(1 for r in chunk_results if r.get('status') == 'completed')
    failed_chunks = sum(1 for r in chunk_results if r.get('status') == 'failed')

    progress_report = {
        'parent_task_id': parent_task_id,
        'total_chunks': total_chunks,
        'completed_chunks': completed_chunks,
        'failed_chunks': failed_chunks,
        'progress_percent': (completed_chunks / total_chunks * 100) if total_chunks > 0 else 0,
        'status': 'in_progress' if completed_chunks < total_chunks else 'completed'
    }

    # If all chunks are done, aggregate final results
    if completed_chunks + failed_chunks == total_chunks:
        progress_report.update(aggregate_chunk_results(chunk_results))
        progress_report['status'] = 'completed'

    return progress_report

@app.task
def monitor_payload_health():
    """
    Monitor payload sizes and memory usage patterns.
    """

    health_report = {
        'timestamp': datetime.utcnow().isoformat(),
        'payload_statistics': {},
        'memory_usage': {},
        'alerts': [],
        'recommendations': []
    }

    # Analyze payload sizes from recent jobs
    payload_stats = analyze_recent_payload_sizes(hours=24)
    health_report['payload_statistics'] = payload_stats

    # Check Redis memory usage
    redis_info = redis_client.info('memory')
    memory_usage_gb = redis_info['used_memory'] / (1024**3)

    health_report['memory_usage'] = {
        'redis_memory_gb': memory_usage_gb,
        'redis_memory_peak_gb': redis_info['used_memory_peak'] / (1024**3),
        'memory_fragmentation_ratio': redis_info.get('mem_fragmentation_ratio', 1.0)
    }

    # Check for concerning patterns
    if payload_stats['avg_payload_size'] > 1024*1024:  # > 1MB average
        health_report['alerts'].append({
            'type': 'large_average_payload',
            'severity': 'high',
            'message': f"Average payload size is {payload_stats['avg_payload_size']/1024/1024:.2f}MB"
        })

    if payload_stats['max_payload_size'] > 10*1024*1024:  # > 10MB max
        health_report['alerts'].append({
            'type': 'extremely_large_payload',
            'severity': 'critical',
            'message': f"Maximum payload size is {payload_stats['max_payload_size']/1024/1024:.2f}MB"
        })

    if memory_usage_gb > 8.0:  # > 8GB Redis usage
        health_report['alerts'].append({
            'type': 'high_memory_usage',
            'severity': 'high',
            'message': f"Redis memory usage is {memory_usage_gb:.2f}GB"
        })

    # Generate recommendations
    if payload_stats['compression_candidates'] > 20:
        health_report['recommendations'].append(
            f"Consider enabling compression for {payload_stats['compression_candidates']} jobs with medium-sized payloads"
        )

    if payload_stats['external_storage_candidates'] > 5:
        health_report['recommendations'].append(
            f"Consider external storage for {payload_stats['external_storage_candidates']} jobs with large payloads"
        )

    return health_report

def analyze_recent_payload_sizes(hours=24):
    """Analyze payload sizes from recent job executions."""

    # This would analyze job execution logs or metrics
    # Implementation depends on your specific logging/monitoring setup

    return {
        'total_jobs': 1250,
        'avg_payload_size': 15680,  # bytes
        'max_payload_size': 2048000,  # bytes
        'min_payload_size': 128,
        'compression_candidates': 23,  # jobs that would benefit from compression
        'external_storage_candidates': 3,  # jobs that should use external storage
        'size_distribution': {
            'under_1kb': 892,
            '1kb_to_100kb': 287,
            '100kb_to_1mb': 65,
            '1mb_to_10mb': 5,
            'over_10mb': 1
        }
    }

from enum import Enum
from datetime import datetime, timedelta
import redis
import json
import requests
from celery import Celery
import random

app = Celery('circuit_breaker_jobs')

class CircuitState(Enum):
    CLOSED = "closed"           # Normal operation
    OPEN = "open"               # Circuit tripped, blocking requests
    HALF_OPEN = "half_open"     # Testing service recovery

class CircuitBreakerConfig:
    """Configuration for circuit breaker behavior."""

    def __init__(self,
                 failure_threshold=5,           # Failures to trip circuit
                 success_threshold=3,           # Successes to close circuit
                 timeout_seconds=300,           # Time before trying half-open
                 slow_call_threshold=10.0,      # Seconds to consider "slow"
                 slow_call_rate_threshold=0.5,  # % of slow calls to trip
                 minimum_calls=10):             # Minimum calls before evaluation

        self.failure_threshold = failure_threshold
        self.success_threshold = success_threshold
        self.timeout_seconds = timeout_seconds
        self.slow_call_threshold = slow_call_threshold
        self.slow_call_rate_threshold = slow_call_rate_threshold
        self.minimum_calls = minimum_calls

class CircuitBreaker:
    """
    Sophisticated circuit breaker with multiple failure modes and gradual recovery.
    """

    def __init__(self, service_name, config=None):
        self.service_name = service_name
        self.config = config or CircuitBreakerConfig()
        self.redis_client = redis.Redis()
        self.stats_key = f"circuit_breaker_stats:{service_name}"
        self.state_key = f"circuit_breaker_state:{service_name}"

    def call(self, func, *args, **kwargs):
        """
        Execute function with circuit breaker protection.
        """

        state = self.get_state()

        if state == CircuitState.OPEN:
            raise CircuitOpenError(f"Circuit breaker OPEN for {self.service_name}")

        elif state == CircuitState.HALF_OPEN:
            # In half-open state, only allow limited calls
            if not self._can_attempt_half_open_call():
                raise CircuitOpenError(f"Circuit breaker HALF-OPEN limiting calls for {self.service_name}")

        # Record call attempt
        start_time = datetime.utcnow()

        try:
            result = func(*args, **kwargs)

            # Record successful call
            execution_time = (datetime.utcnow() - start_time).total_seconds()
            self._record_success(execution_time)

            return result

        except Exception as e:
            # Record failed call
            execution_time = (datetime.utcnow() - start_time).total_seconds()
            self._record_failure(e, execution_time)

            # Re-raise the original exception
            raise

    def get_state(self):
        """Get current circuit breaker state."""

        state_data = self.redis_client.hgetall(self.state_key)

        if not state_data:
            return CircuitState.CLOSED

        current_state = CircuitState(state_data.get('state', 'closed'))

        # Check if OPEN circuit should transition to HALF_OPEN
        if current_state == CircuitState.OPEN:
            opened_at = state_data.get('opened_at')
            if opened_at:
                opened_time = datetime.fromisoformat(opened_at)
                if datetime.utcnow() - opened_time >= timedelta(seconds=self.config.timeout_seconds):
                    self._transition_to_half_open()
                    return CircuitState.HALF_OPEN

        return current_state

    def _record_success(self, execution_time):
        """Record a successful call."""

        now = datetime.utcnow()

        # Update statistics
        self.redis_client.hincrby(self.stats_key, 'total_calls', 1)
        self.redis_client.hincrby(self.stats_key, 'successful_calls', 1)

        # Track execution time for slow call detection
        if execution_time > self.config.slow_call_threshold:
            self.redis_client.hincrby(self.stats_key, 'slow_calls', 1)

        self.redis_client.hset(self.stats_key, 'last_success_at', now.isoformat())

        # Reset failure streak
        self.redis_client.hset(self.stats_key, 'consecutive_failures', 0)

        # Expire stats after 24 hours of inactivity
        self.redis_client.expire(self.stats_key, 86400)

        # Handle state transitions
        current_state = self.get_state()

        if current_state == CircuitState.HALF_OPEN:
            # Check if we've had enough successes to close the circuit
            consecutive_successes = int(self.redis_client.hget(self.stats_key, 'consecutive_successes') or 0) + 1
            self.redis_client.hset(self.stats_key, 'consecutive_successes', consecutive_successes)

            if consecutive_successes >= self.config.success_threshold:
                self._transition_to_closed()

    def _record_failure(self, exception, execution_time):
        """Record a failed call."""

        now = datetime.utcnow()

        # Update statistics
        self.redis_client.hincrby(self.stats_key, 'total_calls', 1)
        self.redis_client.hincrby(self.stats_key, 'failed_calls', 1)

        # Track consecutive failures
        consecutive_failures = int(self.redis_client.hget(self.stats_key, 'consecutive_failures') or 0) + 1
        self.redis_client.hset(self.stats_key, 'consecutive_failures', consecutive_failures)

        # Reset consecutive successes
        self.redis_client.hset(self.stats_key, 'consecutive_successes', 0)

        # Record failure details
        self.redis_client.hset(self.stats_key, mapping={
            'last_failure_at': now.isoformat(),
            'last_failure_type': type(exception).__name__,
            'last_failure_message': str(exception)
        })

        self.redis_client.expire(self.stats_key, 86400)

        # Check if we should trip the circuit
        self._evaluate_circuit_state()

    def _evaluate_circuit_state(self):
        """Evaluate whether circuit should be opened based on current stats."""

        stats = self.redis_client.hgetall(self.stats_key)

        total_calls = int(stats.get('total_calls', 0))
        failed_calls = int(stats.get('failed_calls', 0))
        slow_calls = int(stats.get('slow_calls', 0))
        consecutive_failures = int(stats.get('consecutive_failures', 0))

        # Don't evaluate until we have minimum number of calls
        if total_calls < self.config.minimum_calls:
            return

        should_open = False

        # Check failure rate
        if consecutive_failures >= self.config.failure_threshold:
            should_open = True

        # Check slow call rate
        if total_calls > 0:
            slow_call_rate = slow_calls / total_calls
            if slow_call_rate >= self.config.slow_call_rate_threshold:
                should_open = True

        if should_open and self.get_state() == CircuitState.CLOSED:
            self._transition_to_open()

    def _transition_to_open(self):
        """Transition circuit to OPEN state."""

        self.redis_client.hset(self.state_key, mapping={
            'state': CircuitState.OPEN.value,
            'opened_at': datetime.utcnow().isoformat()
        })

        self.redis_client.expire(self.state_key, 86400)

        # Log circuit opening
        print(f"Circuit breaker OPENED for {self.service_name}")

        # Send alert
        self._send_circuit_alert('opened')

    def _transition_to_half_open(self):
        """Transition circuit to HALF_OPEN state."""

        self.redis_client.hset(self.state_key, mapping={
            'state': CircuitState.HALF_OPEN.value,
            'half_opened_at': datetime.utcnow().isoformat()
        })

        # Reset consecutive success counter
        self.redis_client.hset(self.stats_key, 'consecutive_successes', 0)

        print(f"Circuit breaker HALF-OPEN for {self.service_name}")

    def _transition_to_closed(self):
        """Transition circuit to CLOSED state."""

        self.redis_client.delete(self.state_key)

        print(f"Circuit breaker CLOSED for {self.service_name}")

        # Send recovery alert
        self._send_circuit_alert('closed')

    def _can_attempt_half_open_call(self):
        """Determine if we can make a call in HALF_OPEN state."""

        # In half-open state, we limit the rate of calls to test service recovery
        # Use a simple token bucket approach

        bucket_key = f"half_open_bucket:{self.service_name}"

        # Allow 1 call per 10 seconds in half-open state
        current_token_count = self.redis_client.get(bucket_key)

        if current_token_count is None or int(current_token_count) > 0:
            # Set bucket to 0 and allow the call
            self.redis_client.set(bucket_key, 0, ex=10)  # Reset after 10 seconds
            return True

        return False

    def _send_circuit_alert(self, event_type):
        """Send alert when circuit state changes."""

        # Implementation would depend on your alerting system
        # This could send emails, Slack messages, PagerDuty alerts, etc.

        stats = self.redis_client.hgetall(self.stats_key)

        alert_data = {
            'service_name': self.service_name,
            'event_type': event_type,
            'timestamp': datetime.utcnow().isoformat(),
            'failure_count': stats.get('failed_calls', 0),
            'success_count': stats.get('successful_calls', 0),
            'consecutive_failures': stats.get('consecutive_failures', 0)
        }

        # Example: send to monitoring system
        send_monitoring_alert('circuit_breaker', alert_data)

    def get_health_metrics(self):
        """Get circuit breaker health metrics."""

        state = self.get_state()
        stats = self.redis_client.hgetall(self.stats_key)

        total_calls = int(stats.get('total_calls', 0))
        failed_calls = int(stats.get('failed_calls', 0))
        slow_calls = int(stats.get('slow_calls', 0))

        return {
            'service_name': self.service_name,
            'state': state.value,
            'total_calls': total_calls,
            'success_rate': (total_calls - failed_calls) / total_calls if total_calls > 0 else 1.0,
            'failure_rate': failed_calls / total_calls if total_calls > 0 else 0.0,
            'slow_call_rate': slow_calls / total_calls if total_calls > 0 else 0.0,
            'consecutive_failures': int(stats.get('consecutive_failures', 0)),
            'last_failure_at': stats.get('last_failure_at'),
            'last_success_at': stats.get('last_success_at')
        }

class CircuitOpenError(Exception):
    """Exception raised when circuit breaker is open."""
    pass

# Integration with background jobs
def with_circuit_breaker(service_name, config=None):
    """
    Decorator to add circuit breaker protection to external service calls.
    """

    def decorator(func):
        def wrapper(*args, **kwargs):
            circuit_breaker = CircuitBreaker(service_name, config)
            return circuit_breaker.call(func, *args, **kwargs)
        return wrapper
    return decorator

@app.task(bind=True, max_retries=3)
def external_api_call_protected(self, api_endpoint, data):
    """
    Example task with circuit breaker protection.
    """

    @with_circuit_breaker('external_api')
    def call_external_api():
        response = requests.post(api_endpoint, json=data, timeout=30)
        response.raise_for_status()
        return response.json()

    try:
        result = call_external_api()
        return {
            'status': 'success',
            'result': result
        }

    except CircuitOpenError as e:
        # Circuit is open - don't retry immediately
        # Schedule retry for later when circuit might be closed
        raise self.retry(countdown=300, max_retries=2)  # Retry in 5 minutes

    except Exception as e:
        # Other errors - normal retry logic
        raise self.retry(exc=e, countdown=60, max_retries=3)

class CircuitBreakerCoordinator:
    """
    Coordinate multiple circuit breakers and manage system-wide resilience.
    """

    def __init__(self):
        self.redis_client = redis.Redis()

        # Define service dependencies and criticality
        self.service_topology = {
            'payment_processor': {
                'criticality': 'critical',
                'dependencies': ['fraud_detection', 'user_verification'],
                'dependent_services': ['order_processing', 'subscription_management']
            },
            'email_service': {
                'criticality': 'high',
                'dependencies': ['template_service'],
                'dependent_services': []
            },
            'analytics_api': {
                'criticality': 'low',
                'dependencies': ['data_warehouse'],
                'dependent_services': ['reporting']
            },
            'inventory_service': {
                'criticality': 'critical',
                'dependencies': ['warehouse_api'],
                'dependent_services': ['order_processing', 'product_catalog']
            }
        }

    def assess_system_resilience(self):
        """
        Assess overall system resilience based on circuit breaker states.
        """

        resilience_report = {
            'timestamp': datetime.utcnow().isoformat(),
            'overall_health': 'healthy',
            'service_states': {},
            'critical_services_affected': [],
            'cascading_impact': {},
            'recommended_actions': []
        }

        # Check each service's circuit breaker state
        for service_name, topology in self.service_topology.items():
            circuit_breaker = CircuitBreaker(service_name)
            metrics = circuit_breaker.get_health_metrics()

            resilience_report['service_states'][service_name] = {
                'state': metrics['state'],
                'success_rate': metrics['success_rate'],
                'criticality': topology['criticality'],
                'impact': self._calculate_service_impact(service_name, metrics['state'])
            }

            # Track critical services that are down
            if (topology['criticality'] == 'critical' and
                metrics['state'] in ['open', 'half_open']):
                resilience_report['critical_services_affected'].append(service_name)

        # Analyze cascading impacts
        resilience_report['cascading_impact'] = self._analyze_cascading_impact()

        # Determine overall system health
        if resilience_report['critical_services_affected']:
            resilience_report['overall_health'] = 'critical'
        elif any(state['state'] == 'open' for state in resilience_report['service_states'].values()):
            resilience_report['overall_health'] = 'degraded'

        # Generate recommendations
        resilience_report['recommended_actions'] = self._generate_resilience_recommendations(
            resilience_report
        )

        return resilience_report

    def _calculate_service_impact(self, service_name, circuit_state):
        """Calculate the impact of a service's circuit breaker state."""

        if circuit_state == 'closed':
            return 'none'

        topology = self.service_topology.get(service_name, {})
        dependent_services = topology.get('dependent_services', [])

        impact_score = len(dependent_services)

        # Weight by criticality
        criticality_weights = {'critical': 3, 'high': 2, 'medium': 1, 'low': 0.5}
        criticality = topology.get('criticality', 'medium')
        impact_score *= criticality_weights.get(criticality, 1)

        if impact_score >= 6:
            return 'severe'
        elif impact_score >= 3:
            return 'moderate'
        else:
            return 'minor'

    def _analyze_cascading_impact(self):
        """Analyze potential cascading failures across services."""

        cascading_risks = {}

        for service_name, topology in self.service_topology.items():
            circuit_breaker = CircuitBreaker(service_name)
            metrics = circuit_breaker.get_health_metrics()

            if metrics['state'] in ['open', 'half_open']:
                # Service is down - check impact on dependent services
                dependent_services = topology.get('dependent_services', [])

                for dependent_service in dependent_services:
                    if dependent_service not in cascading_risks:
                        cascading_risks[dependent_service] = []

                    cascading_risks[dependent_service].append({
                        'failed_dependency': service_name,
                        'risk_level': topology.get('criticality', 'medium')
                    })

        return cascading_risks

    def _generate_resilience_recommendations(self, resilience_report):
        """Generate actionable recommendations based on resilience analysis."""

        recommendations = []

        # Critical service recommendations
        if resilience_report['critical_services_affected']:
            recommendations.append({
                'type': 'critical_service_failure',
                'priority': 'immediate',
                'message': f"Critical services affected: {', '.join(resilience_report['critical_services_affected'])}",
                'actions': ['escalate_to_oncall', 'enable_degraded_mode', 'prepare_rollback']
            })

        # Cascading failure prevention
        if resilience_report['cascading_impact']:
            at_risk_services = list(resilience_report['cascading_impact'].keys())
            recommendations.append({
                'type': 'cascading_failure_risk',
                'priority': 'high',
                'message': f"Services at risk of cascading failure: {', '.join(at_risk_services)}",
                'actions': ['increase_timeouts', 'reduce_retry_rates', 'enable_fallback_modes']
            })

        # Circuit breaker tuning recommendations
        degraded_services = [
            name for name, state in resilience_report['service_states'].items()
            if state['state'] in ['open', 'half_open'] and state['success_rate'] > 0.5
        ]

        if degraded_services:
            recommendations.append({
                'type': 'circuit_breaker_tuning',
                'priority': 'medium',
                'message': f"Services may need circuit breaker tuning: {', '.join(degraded_services)}",
                'actions': ['review_failure_thresholds', 'adjust_timeout_settings']
            })

        return recommendations

    def execute_coordinated_response(self, resilience_report):
        """Execute coordinated response actions based on system state."""

        actions_taken = []

        for recommendation in resilience_report['recommended_actions']:
            if recommendation['priority'] == 'immediate':

                if 'enable_degraded_mode' in recommendation['actions']:
                    self._enable_system_degraded_mode()
                    actions_taken.append('enabled_degraded_mode')

                if 'escalate_to_oncall' in recommendation['actions']:
                    self._escalate_to_oncall(resilience_report)
                    actions_taken.append('escalated_to_oncall')

            elif recommendation['priority'] == 'high':

                if 'increase_timeouts' in recommendation['actions']:
                    self._increase_system_timeouts()
                    actions_taken.append('increased_timeouts')

                if 'reduce_retry_rates' in recommendation['actions']:
                    self._reduce_system_retry_rates()
                    actions_taken.append('reduced_retry_rates')

        return actions_taken

    def _enable_system_degraded_mode(self):
        """Enable system-wide degraded mode."""

        degraded_config = {
            'enabled': True,
            'enabled_at': datetime.utcnow().isoformat(),
            'non_essential_services_disabled': True,
            'reduced_functionality_mode': True
        }

        self.redis_client.hset('system_config:degraded_mode', mapping=degraded_config)

        # Disable non-essential background jobs
        self._disable_non_essential_jobs()

    def _disable_non_essential_jobs(self):
        """Disable non-essential background jobs during degraded mode."""

        non_essential_queues = ['analytics', 'reporting', 'cleanup', 'optimization']

        for queue in non_essential_queues:
            self.redis_client.hset('queue_config', f'{queue}_disabled', 'true')

@app.task
def monitor_circuit_breaker_health():
    """
    Comprehensive circuit breaker health monitoring and coordination.
    """

    coordinator = CircuitBreakerCoordinator()
    resilience_report = coordinator.assess_system_resilience()

    # Take coordinated action if needed
    if resilience_report['overall_health'] in ['critical', 'degraded']:
        actions_taken = coordinator.execute_coordinated_response(resilience_report)
        resilience_report['automated_actions'] = actions_taken

    # Store report for historical analysis
    report_key = f"resilience_report:{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}"
    redis_client.set(report_key, json.dumps(resilience_report), ex=86400 * 7)  # Keep for 7 days

    return resilience_report

# Schedule regular circuit breaker monitoring
app.conf.beat_schedule = {
    'circuit-breaker-health': {
        'task': 'monitor_circuit_breaker_health',
        'schedule': timedelta(minutes=5),
    },
}

class AdaptiveCircuitBreaker(CircuitBreaker):
    """
    Circuit breaker that adapts its thresholds based on service behavior patterns.
    """

    def __init__(self, service_name, config=None):
        super().__init__(service_name, config)
        self.learning_key = f"circuit_breaker_learning:{service_name}"

    def _evaluate_circuit_state(self):
        """Enhanced evaluation with adaptive thresholds."""

        # Get current stats
        stats = self.redis_client.hgetall(self.stats_key)

        # Get historical performance data
        historical_data = self._get_historical_performance()

        # Adapt thresholds based on historical patterns
        adapted_config = self._adapt_thresholds(historical_data)

        # Use adapted thresholds for evaluation
        total_calls = int(stats.get('total_calls', 0))
        failed_calls = int(stats.get('failed_calls', 0))
        consecutive_failures = int(stats.get('consecutive_failures', 0))

        if total_calls < adapted_config['minimum_calls']:
            return

        # Check against adapted failure threshold
        if consecutive_failures >= adapted_config['failure_threshold']:
            if self.get_state() == CircuitState.CLOSED:
                self._transition_to_open()

    def _get_historical_performance(self):
        """Get historical performance data for learning."""

        learning_data = self.redis_client.hgetall(self.learning_key)

        return {
            'avg_success_rate': float(learning_data.get('avg_success_rate', 0.95)),
            'typical_failure_duration': float(learning_data.get('typical_failure_duration', 300)),
            'recovery_success_rate': float(learning_data.get('recovery_success_rate', 0.8)),
            'peak_traffic_failure_rate': float(learning_data.get('peak_traffic_failure_rate', 0.1))
        }

    def _adapt_thresholds(self, historical_data):
        """Adapt circuit breaker thresholds based on historical data."""

        adapted_config = {
            'failure_threshold': self.config.failure_threshold,
            'timeout_seconds': self.config.timeout_seconds,
            'minimum_calls': self.config.minimum_calls
        }

        # If service historically has low success rate, be more tolerant
        if historical_data['avg_success_rate'] < 0.9:
            adapted_config['failure_threshold'] = self.config.failure_threshold + 2

        # If service typically recovers quickly, use shorter timeout
        if historical_data['typical_failure_duration'] < 180:  # 3 minutes
            adapted_config['timeout_seconds'] = max(120, self.config.timeout_seconds * 0.7)

        # If we're in peak traffic period, be more conservative
        if self._is_peak_traffic_period():
            adapted_config['failure_threshold'] = max(2, self.config.failure_threshold - 1)

        return adapted_config

    def _is_peak_traffic_period(self):
        """Determine if we're currently in a peak traffic period."""

        # Simple implementation - peak hours are 9 AM to 5 PM
        current_hour = datetime.utcnow().hour
        return 9 <= current_hour <= 17

class BulkheadCircuitBreaker:
    """
    Circuit breaker that implements bulkhead pattern for resource isolation.
    """

    def __init__(self, service_name, resource_pools=None):
        self.service_name = service_name
        self.resource_pools = resource_pools or {
            'critical': {'size': 5, 'circuit_breaker': CircuitBreaker(f"{service_name}_critical")},
            'normal': {'size': 10, 'circuit_breaker': CircuitBreaker(f"{service_name}_normal")},
            'bulk': {'size': 20, 'circuit_breaker': CircuitBreaker(f"{service_name}_bulk")}
        }

    def call(self, func, priority='normal', *args, **kwargs):
        """
        Execute function with bulkhead isolation and appropriate circuit breaker.
        """

        if priority not in self.resource_pools:
            priority = 'normal'

        pool = self.resource_pools[priority]
        circuit_breaker = pool['circuit_breaker']

        # Check if pool has capacity
        if not self._acquire_pool_slot(priority):
            if priority != 'critical':
                # Try to downgrade to bulk pool
                if priority == 'normal' and self._acquire_pool_slot('bulk'):
                    return self.resource_pools['bulk']['circuit_breaker'].call(func, *args, **kwargs)

            raise ResourcePoolExhaustedError(f"No capacity in {priority} pool for {self.service_name}")

        try:
            return circuit_breaker.call(func, *args, **kwargs)
        finally:
            self._release_pool_slot(priority)

    def _acquire_pool_slot(self, priority):
        """Acquire a slot in the specified resource pool."""

        pool_key = f"resource_pool:{self.service_name}:{priority}"
        current_usage = self.redis_client.incr(pool_key)

        pool_size = self.resource_pools[priority]['size']

        if current_usage <= pool_size:
            # Set expiration to prevent leaks
            self.redis_client.expire(pool_key, 300)
            return True
        else:
            # Exceeded capacity - rollback increment
            self.redis_client.decr(pool_key)
            return False

    def _release_pool_slot(self, priority):
        """Release a slot in the specified resource pool."""

        pool_key = f"resource_pool:{self.service_name}:{priority}"
        self.redis_client.decr(pool_key)

class ResourcePoolExhaustedError(Exception):
    """Exception raised when resource pool is exhausted."""
    pass

@app.task(bind=True, max_retries=3)
def resilient_external_service_call(self, service_name, api_endpoint, data, priority='normal'):
    """
    Highly resilient external service call with multiple protection layers.
    """

    bulkhead_breaker = BulkheadCircuitBreaker(service_name)

    def make_api_call():
        response = requests.post(api_endpoint, json=data, timeout=30)
        response.raise_for_status()
        return response.json()

    try:
        result = bulkhead_breaker.call(make_api_call, priority=priority)

        return {
            'status': 'success',
            'result': result,
            'service': service_name,
            'priority_used': priority
        }

    except CircuitOpenError as e:
        # Circuit is open - implement graceful degradation
        fallback_result = self._attempt_fallback(service_name, data)

        if fallback_result:
            return {
                'status': 'fallback_success',
                'result': fallback_result,
                'service': service_name,
                'message': 'Used fallback due to circuit breaker'
            }
        else:
            # No fallback available - schedule retry when circuit might be closed
            raise self.retry(countdown=300, max_retries=1)

    except ResourcePoolExhaustedError as e:
        # Resource pool exhausted - try with lower priority or retry later
        if priority == 'normal':
            # Try again with bulk priority
            return resilient_external_service_call.apply(
                args=[service_name, api_endpoint, data, 'bulk']
            )
        else:
            # Already tried bulk or was critical - retry with delay
            raise self.retry(countdown=60, max_retries=2)

    except Exception as e:
        # Other errors - normal retry with exponential backoff
        countdown = 30 * (2 ** self.request.retries)
        raise self.retry(exc=e, countdown=countdown, max_retries=3)

    def _attempt_fallback(self, service_name, data):
        """Attempt fallback behavior when primary service is unavailable."""

        # Implementation depends on service type
        fallback_strategies = {
            'email_service': lambda d: {'message': 'Email queued for retry', 'queued': True},
            'analytics_api': lambda d: None,  # Analytics can be skipped
            'payment_processor': lambda d: None  # Payments can't use fallback
        }

        strategy = fallback_strategies.get(service_name)
        return strategy(data) if strategy else None