Queue System

// Import queue system (all organized modules included)
import 'package:khadem/src/core/queue/index.dart';

// Or import specific modules
import 'package:khadem/src/core/queue/queue_manager.dart';
import 'package:khadem/src/core/queue/drivers/in_memory_driver.dart';
import 'package:khadem/src/contracts/queue/queue_job.dart';

// Initialize queue manager with config
final config = Config(); // Your config implementation
final queueManager = QueueManager(config);

// Load configuration and register drivers
queueManager.loadFromConfig();

// Or manually register a driver
final memoryDriver = InMemoryDriver(config: DriverConfig(name: 'memory'));
queueManager.registerDriver('memory', memoryDriver);

// Create and dispatch a simple job
class WelcomeEmailJob extends QueueJob {
  final String userEmail;
  final String userName;

  WelcomeEmailJob(this.userEmail, this.userName);

  @override
  Future<void> handle() async {
    print('Sending welcome email to $userName at $userEmail');
    // Send email logic here
    await Future.delayed(Duration(seconds: 1)); // Simulate work
    print('✅ Welcome email sent!');
  }

  @override
  String get displayName => 'Send Welcome Email';

  @override
  int get maxRetries => 3;
}

// Dispatch the job
final job = WelcomeEmailJob('user@example.com', 'John Doe');
await queueManager.dispatch(job);

// Start a worker to process jobs
await queueManager.startWorker(
  delay: Duration(seconds: 2),
  onJobComplete: (job, result) => print('Job completed: ${job.displayName}'),
  onJobError: (job, error, stack) => print('Job failed: ${job.displayName} - $error'),
);

// Import required classes
import 'package:khadem/src/contracts/queue/queue_job.dart';
import 'package:khadem/src/core/queue/serialization/index.dart';

// Simple job class
class ProcessPaymentJob extends QueueJob {
  final String orderId;
  final double amount;

  ProcessPaymentJob(this.orderId, this.amount);

  @override
  Future<void> handle() async {
    print('Processing payment for order $orderId: $$${amount}');

    // Simulate payment processing
    await Future.delayed(Duration(seconds: 2));

    if (Random().nextBool()) {
      throw Exception('Payment gateway timeout');
    }

    print('✅ Payment processed successfully');
  }

  @override
  String get displayName => 'Process Payment';

  @override
  int get maxRetries => 5;
}

// Job with custom serialization (for persistent queues)
class SendNotificationJob extends QueueJob with SerializableJob {
  final int userId;
  final String message;
  final NotificationType type;

  SendNotificationJob(this.userId, this.message, this.type);

  @override
  Future<void> handle() async {
    print('Sending $type notification to user $userId: $message');

    // Send notification logic
    await Future.delayed(Duration(milliseconds: 500));

    print('✅ Notification sent');
  }

  @override
  String get displayName => 'Send ${type.name} Notification';

  @override
  int get maxRetries => 3;

  // Custom serialization for complex objects
  @override
  Map<String, dynamic> toJson() => {
    'userId': userId,
    'message': message,
    'type': type.name,
  };

  static SendNotificationJob fromJson(Map<String, dynamic> json) {
    return SendNotificationJob(
      json['userId'],
      json['message'],
      NotificationType.values.firstWhere((e) => e.name == json['type']),
    );
  }
}

enum NotificationType { email, sms, push }

// Configuration file structure (config/queue.yaml or JSON)
final config = {
  'queue': {
    'default': 'memory', // Default driver to use
    'drivers': {
      'memory': {
        'driver': 'memory',
        'track_metrics': true,
        'use_dlq': true,
        'max_retries': 3,
      },
      'file': {
        'driver': 'file',
        'path': './storage/queue',
        'track_metrics': true,
        'use_dlq': true,
      },
      'sync': {
        'driver': 'sync',
      },
      'redis': {
        'driver': 'redis',
        'host': 'localhost',
        'port': 6379,
        'password': null,
        'queue_name': 'default',
        'track_metrics': true,
        'use_dlq': true,
      },
    },
  },
};

// Initialize with configuration
final queueManager = QueueManager(config);

// Load config and auto-register all drivers
queueManager.loadFromConfig();

// Or manually register drivers
final memoryDriver = InMemoryDriver(
  config: DriverConfig(
    name: 'memory',
    trackMetrics: true,
    useDLQ: true,
    maxRetries: 3,
    defaultJobTimeout: Duration(minutes: 5),
  ),
);
queueManager.registerDriver('memory', memoryDriver);

// Switch default driver
queueManager.setDefaultDriver('memory');

// Get a specific driver
final fileDriver = queueManager.getDriver('file');

// Check available drivers
print(queueManager.driverNames); // ['memory', 'file', 'sync', 'redis']
print(queueManager.hasDriver('redis')); // true

// Immediate job execution
await queueManager.dispatch(UrgentJob());

// Delayed job execution
await queueManager.dispatch(
  ScheduledJob(),
  delay: Duration(minutes: 5),
);

// Various delay examples
final delays = [
  Duration(seconds: 30),     // 30 seconds
  Duration(minutes: 5),      // 5 minutes
  Duration(hours: 1),        // 1 hour
  Duration(days: 1),         // 1 day
  Duration(hours: 24),       // 24 hours
];

// Schedule newsletter
await queueManager.dispatch(
  SendNewsletterJob(),
  delay: Duration(hours: 9), // Send at 9 AM tomorrow
);

// Retry with backoff
class RetryWithBackoffJob extends QueueJob {
  final int attempt;

  RetryWithBackoffJob([this.attempt = 0]);

  @override
  Future<void> handle() async {
    try {
      await riskyOperation();
      print('✅ Operation succeeded on attempt ${attempt + 1}');
    } catch (e) {
      if (attempt < maxRetries) {
        // Exponential backoff: 30s, 1m, 2m, 4m, 8m...
        final delay = Duration(seconds: 30 * (1 << attempt));
        await queueManager.dispatch(
          RetryWithBackoffJob(attempt + 1),
          delay: delay,
        );
        print('⏰ Retrying in ${delay.inSeconds} seconds...');
      } else {
        print('❌ Operation failed permanently after ${attempt + 1} attempts');
      }
    }
  }

  @override
  String get displayName => 'Retry Operation (Attempt ${attempt + 1})';

  @override
  int get maxRetries => 5;
}

// Basic worker
await queueManager.startWorker();

// Worker with custom configuration
await queueManager.startWorker(
  maxJobs: 100,                    // Process max 100 jobs
  delay: Duration(seconds: 5),     // 5 second delay between jobs
  timeout: Duration(hours: 1),     // Stop after 1 hour
  runInBackground: true,           // Run in background
);

// Worker with event callbacks
await queueManager.startWorker(
  onJobStart: (job) {
    print('🚀 Starting job: ${job.displayName}');
  },
  onJobComplete: (job, result) {
    print('✅ Completed job: ${job.displayName}');
  },
  onJobError: (job, error, stack) {
    print('❌ Failed job: ${job.displayName} - $error');
    // Send alert, log to external service, etc.
  },
  onError: (error, stack) {
    print('💥 Worker error: $error');
    // Handle worker-level errors
  },
);

// Long-running worker for production
await queueManager.startWorker(
  delay: Duration(milliseconds: 500),  // Fast processing
  maxJobs: null,                       // Run indefinitely
  timeout: null,                       // No timeout
  runInBackground: true,
);

// Development worker with verbose logging
await queueManager.startWorker(
  delay: Duration(seconds: 2),
  onJobStart: (job) => print('[DEV] Processing: ${job.displayName}'),
  onJobComplete: (job, result) => print('[DEV] Completed: ${job.displayName}'),
  onJobError: (job, error, stack) => print('[DEV] Error in ${job.displayName}: $error'),
);

// Process multiple jobs at once
final jobs = [
  SendEmailJob('user1@example.com', 'Welcome User 1'),
  SendEmailJob('user2@example.com', 'Welcome User 2'),
  SendEmailJob('user3@example.com', 'Welcome User 3'),
  ProcessPaymentJob('order_123', 99.99),
  UpdateInventoryJob('product_456', -5),
];

// Dispatch all at once
await queueManager.dispatchBatch(jobs);

// Dispatch with individual delays
await queueManager.dispatchBatch(jobs, delay: Duration(seconds: 10));

// Batch processing job
class BulkEmailJob extends QueueJob {
  final List<String> emails;
  final String subject;
  final String template;

  BulkEmailJob(this.emails, this.subject, this.template);

  @override
  Future<void> handle() async {
    print('Sending ${emails.length} emails...');

    for (final email in emails) {
      try {
        final personalizedContent = template.replaceAll('{{email}}', email);
        await sendEmail(email, subject, personalizedContent);
        print('✅ Sent to $email');
      } catch (e) {
        print('❌ Failed to send to $email: $e');
        // Continue with other emails
      }
    }

    print('✅ Bulk email job completed');
  }

  @override
  String get displayName => 'Send Bulk Email (${emails.length} recipients)';

  @override
  int get maxRetries => 2;
}

// Usage
await queueManager.dispatch(BulkEmailJob(
  ['user1@example.com', 'user2@example.com', 'user3@example.com'],
  'Monthly Newsletter',
  'Hello {{email}}, here is your newsletter...',
));

// Import metrics (if accessing directly)
import 'package:khadem/src/core/queue/metrics/index.dart';

// Get driver statistics
final stats = await queueManager.driver.getStats();
print('Queue Statistics:');
print('  Driver: ${stats['driver_name']}');
print('  Queue Depth: ${stats['queue_depth']}');
print('  Processing: ${stats['processing_count']}');
print('  Pending: ${stats['pending_count']}');
print('  Ready Jobs: ${stats['ready_jobs']}');
print('  Delayed Jobs: ${stats['delayed_jobs']}');

// Metrics tracking (if enabled)
if (stats.containsKey('metrics')) {
  final metrics = stats['metrics'] as Map<String, dynamic>;
  print('Metrics:');
  print('  Total Queued: ${metrics['jobs_queued']}');
  print('  Total Completed: ${metrics['jobs_completed']}');
  print('  Total Failed: ${metrics['jobs_failed']}');
  print('  Success Rate: ${(metrics['success_rate'] * 100).toStringAsFixed(1)}%');
  
  // Jobs by type
  final byType = metrics['jobs_by_type'] as Map<String, dynamic>;
  print('Jobs by Type:');
  byType.forEach((type, count) {
    print('  $type: $count');
  });
}

// Dead Letter Queue statistics (if enabled)
if (stats.containsKey('dlq')) {
  final dlq = stats['dlq'] as Map<String, dynamic>;
  print('Dead Letter Queue:');
  print('  Total: ${dlq['total_jobs']}');
  print('  By Job Type:');
  (dlq['jobs_by_type'] as Map<String, dynamic>).forEach((type, count) {
    print('    $type: $count');
  });
}

// Custom monitoring with periodic checks
class QueueMonitor {
  final QueueManager _queueManager;

  QueueMonitor(this._queueManager);

  void startMonitoring() {
    Timer.periodic(Duration(minutes: 5), (timer) async {
      final stats = await _queueManager.driver.getStats();
      final metrics = stats['metrics'] as Map<String, dynamic>?;

      if (metrics == null) {
        print('⚠️ Metrics not enabled');
        return;
      }

      // Check for issues
      final successRate = metrics['success_rate'] as double;
      if (successRate < 0.9) { // < 90% success rate
        print('⚠️ Low success rate: ${(successRate * 100).toStringAsFixed(1)}%');
        // Send alert
      }

      final queueDepth = stats['queue_depth'] as int;
      if (queueDepth > 100) {
        print('⚠️ Large queue backlog: $queueDepth jobs');
        // Scale up workers
      }

      // Log metrics
      print('📊 Queue Health Check:');
      print('   Success Rate: ${(successRate * 100).toStringAsFixed(1)}%');
      print('   Processing: ${stats['processing_count']} jobs');
      print('   Queue Depth: $queueDepth jobs');
    });
  }
}

// Reset metrics (for testing)
// Note: This requires direct access to driver's metrics object
import 'package:khadem/src/core/queue/drivers/base_driver.dart';

final driver = queueManager.driver as BaseQueueDriver;
if (driver.metrics != null) {
  driver.metrics!.reset();
}

// Job with error handling
class ProcessOrderJob extends QueueJob {
  final String orderId;

  ProcessOrderJob(this.orderId);

  @override
  Future<void> handle() async {
    try {
      // Step 1: Validate order
      final order = await Order.find(orderId);
      if (order == null) {
        throw Exception('Order not found: $orderId');
      }

      // Step 2: Process payment
      await PaymentService.charge(order);

      // Step 3: Update inventory
      for (final item in order.items) {
        await Inventory.decrement(item.productId, item.quantity);
      }

      // Step 4: Send confirmation
      await EmailService.sendOrderConfirmation(order);

      print('✅ Order $orderId processed successfully');

    } catch (e) {
      print('❌ Failed to process order $orderId: $e');

      // Log detailed error information
      await ErrorLogger.log('order_processing_failed', {
        'orderId': orderId,
        'error': e.toString(),
        'timestamp': DateTime.now().toIso8601String(),
        'attempt': attempt,
      });

      // Don't rethrow - let queue handle retry logic
      // The job will be retried based on maxRetries
    }
  }

  @override
  String get displayName => 'Process Order $orderId';

  @override
  int get maxRetries => 3;
}

// Worker with error handling
await queueManager.startWorker(
  onJobError: (job, error, stack) async {
    // Log to external service
    await ExternalLogger.error('Queue job failed', {
      'job': job.displayName,
      'error': error.toString(),
      'stack': stack.toString(),
      'timestamp': DateTime.now().toIso8601String(),
    });

    // Send alert for critical jobs
    if (job.displayName.contains('Payment') || job.displayName.contains('Order')) {
      await AlertService.send('Critical job failed: ${job.displayName}');
    }
  },
  onError: (error, stack) async {
    // Worker-level error
    await ExternalLogger.critical('Queue worker error', {
      'error': error.toString(),
      'stack': stack.toString(),
    });

    // Restart worker or send alert
    await AlertService.send('Queue worker crashed - manual intervention required');
  },
);

// Graceful error recovery
class ResilientJob extends QueueJob {
  @override
  Future<void> handle() async {
    const maxRetries = 3;
    var attempt = 0;

    while (attempt < maxRetries) {
      try {
        await performRiskyOperation();
        print('✅ Operation succeeded');
        return;
      } catch (e) {
        attempt++;
        print('⚠️  Attempt $attempt failed: $e');

        if (attempt < maxRetries) {
          // Wait before retry
          await Future.delayed(Duration(seconds: attempt * 5));
        }
      }
    }

    throw Exception('Operation failed after $maxRetries attempts');
  }

  @override
  String get displayName => 'Resilient Operation';

  @override
  int get maxRetries => 0; // Handle retries internally
}

// Import required contracts and base classes
import 'package:khadem/src/contracts/queue/queue_driver.dart';
import 'package:khadem/src/contracts/queue/queue_job.dart';
import 'package:khadem/src/core/queue/drivers/base_driver.dart';

// Custom database queue driver
class DatabaseQueueDriver extends BaseQueueDriver {
  final DatabaseConnection _db;

  DatabaseQueueDriver(this._db);

  @override
  Future<void> push(QueueJob job, {Duration? delay}) async {
    final scheduledAt = DateTime.now().add(delay ?? Duration.zero);

    await _db.table('jobs').insert({
      'id': DateTime.now().millisecondsSinceEpoch.toString(),
      'type': job.runtimeType.toString(),
      'payload': jsonEncode(job.toJson()),
      'scheduled_at': scheduledAt.toIso8601String(),
      'created_at': DateTime.now().toIso8601String(),
      'attempts': 0,
      'max_retries': job.maxRetries,
      'status': 'pending',
    });

    print('🗄️ Job queued to database: ${job.displayName}');
  }

  @override
  Future<void> process() async {
    // Find next pending job
    final jobRecord = await _db.table('jobs')
      .where('status', 'pending')
      .where('scheduled_at', '<=', DateTime.now().toIso8601String())
      .orderBy('created_at')
      .first();

    if (jobRecord == null) return;

    // Mark as processing
    await _db.table('jobs')
      .where('id', jobRecord['id'])
      .update({'status': 'processing'});

    try {
      // Create job instance and execute
      final job = _createJobFromRecord(jobRecord);
      await job.handle();

      // Mark as completed
      await _db.table('jobs')
        .where('id', jobRecord['id'])
        .update({'status': 'completed'});

      print('🗄️ Job completed: ${job.displayName}');

    } catch (e) {
      // Handle failure and retry logic
      final attempts = jobRecord['attempts'] + 1;
      final maxRetries = jobRecord['max_retries'];

      if (attempts >= maxRetries) {
        await _db.table('jobs')
          .where('id', jobRecord['id'])
          .update({
            'status': 'failed',
            'error': e.toString(),
          });
        print('🗄️ Job failed permanently: ${jobRecord['type']}');
      } else {
        // Schedule retry
        final retryDelay = Duration(seconds: 30 * attempts);
        final nextAttempt = DateTime.now().add(retryDelay);

        await _db.table('jobs')
          .where('id', jobRecord['id'])
          .update({
            'status': 'pending',
            'attempts': attempts,
            'scheduled_at': nextAttempt.toIso8601String(),
          });
        print('🗄️ Job retry scheduled: ${jobRecord['type']}');
      }
    }
  }

  QueueJob _createJobFromRecord(Map<String, dynamic> record) {
    // This would need a job registry/factory
    // Simplified version for demonstration
    final payload = jsonDecode(record['payload']);
    return GenericQueueJob.fromPayload(record['type'], payload);
  }
}

// Register custom driver
QueueFactory.registerDriver('database', DatabaseQueueDriver(database));

// Use custom driver
final config = {
  'queue': {
    'driver': 'database',
  },
};
final queueManager = QueueManager(config);

// Import test framework and queue classes
import 'package:test/test.dart';
import 'package:khadem/src/core/queue/index.dart';
import 'package:khadem/src/contracts/queue/queue_job.dart';

// Unit tests for queue jobs
void main() {
  group('QueueJob Tests', () {
    late QueueManager queueManager;
    late InMemoryDriver driver;

    setUp(() {
      // Use memory driver for testing
      driver = InMemoryDriver(
        config: DriverConfig(name: 'test'),
      );
      
      final config = Config(); // Mock config
      queueManager = QueueManager(config, driver: driver);
    });

    test('Simple job executes successfully', () async {
      var executed = false;
      
      final job = SimpleJob(() => executed = true);
      await queueManager.dispatch(job);

      // Process the job
      await queueManager.process();

      expect(executed, isTrue);
      expect(driver.isEmpty, isTrue);
    });

    test('Job with retry logic', () async {
      var executionCount = 0;
      
      final job = RetryableJob(() {
        executionCount++;
        if (executionCount < 3) throw Exception('Retry me');
      });
      
      await queueManager.dispatch(job);

      // Process multiple times - automatic retry happens
      await queueManager.process(); // Fail 1, auto-retry
      await Future.delayed(Duration(milliseconds: 100));
      await queueManager.process(); // Fail 2, auto-retry
      await Future.delayed(Duration(milliseconds: 100));
      await queueManager.process(); // Success

      expect(executionCount, equals(3));
    });

    test('Delayed job execution', () async {
      var executed = false;
      final job = SimpleJob(() => executed = true);
      final delay = Duration(milliseconds: 500);

      await queueManager.dispatch(job, delay: delay);

      // Job should not execute immediately
      await queueManager.process();
      expect(executed, isFalse);

      // Wait for delay and process again
      await Future.delayed(delay + Duration(milliseconds: 100));
      await queueManager.process();
      expect(executed, isTrue);
    });

    test('Batch job processing', () async {
      final executed = <int>[];
      final jobs = List.generate(
        3, 
        (i) => SimpleJob(() => executed.add(i)),
      );

      await queueManager.dispatchBatch(jobs);

      // Process all jobs
      for (var i = 0; i < jobs.length; i++) {
        await queueManager.process();
      }

      expect(executed.length, equals(3));
      expect(driver.isEmpty, isTrue);
    });

    test('Worker processes jobs correctly', () async {
      final executed = <int>[];
      final jobs = List.generate(
        5, 
        (i) => SimpleJob(() => executed.add(i)),
      );
      
      for (final job in jobs) {
        await queueManager.dispatch(job);
      }

      // Start worker with short delay
      await queueManager.startWorker(
        maxJobs: 5,
        delay: Duration(milliseconds: 10),
      );

      // Wait for all jobs to complete
      await Future.delayed(Duration(milliseconds: 200));

      expect(executed.length, equals(5));
      expect(driver.isEmpty, isTrue);
    });

    test('Queue statistics', () async {
      final job = SimpleJob(() {});
      await queueManager.dispatch(job);

      final stats = await driver.getStats();
      
      expect(stats['queue_depth'], equals(1));
      expect(stats['ready_jobs'], equals(1));
      expect(stats['delayed_jobs'], equals(0));
    });

    test('Metrics tracking', () async {
      // Enable metrics
      final metricsDriver = InMemoryDriver(
        config: DriverConfig(
          name: 'test',
          trackMetrics: true,
        ),
        metrics: QueueMetrics(),
      );
      
      final qm = QueueManager(Config(), driver: metricsDriver);

      await qm.dispatch(SimpleJob(() {}));
      await qm.process();

      final stats = await metricsDriver.getStats();
      final metrics = stats['metrics'] as Map<String, dynamic>;
      
      expect(metrics['jobs_queued'], equals(1));
      expect(metrics['jobs_completed'], equals(1));
      expect(metrics['success_rate'], equals(1.0));
    });
  });
}

// Test job implementations
class SimpleJob extends QueueJob {
  final void Function() action;

  SimpleJob(this.action);

  @override
  Future<void> handle() async {
    await Future.delayed(Duration(milliseconds: 10)); // Simulate work
    action();
  }

  @override
  String get displayName => 'Simple Test Job';

  @override
  int get maxRetries => 3;
}

class RetryableJob extends QueueJob {
  final void Function() action;

  RetryableJob(this.action);

  @override
  Future<void> handle() async {
    await Future.delayed(Duration(milliseconds: 10));
    action(); // May throw exception
  }

  @override
  String get displayName => 'Retryable Job';

  @override
  int get maxRetries => 5;
}