<?php
namespace EasyChat\Library\Queue;

use EasyChat\Library\Queue\Jobs\CmsIndexJob;
use EasyChat\Library\Queue\Jobs\CatalogUrlVerifyJob;
use EasyChat\Library\Queue\Jobs\EmbeddingJob;
use EasyChat\Library\Queue\Jobs\ImageVariantJob;
use EasyChat\Library\Queue\Jobs\MessengerAvitoHistorySyncJob;
use EasyChat\Library\Queue\Jobs\MessengerAvitoItemContextEnrichJob;
use EasyChat\Library\Queue\Jobs\MessengerYoulaItemContextEnrichJob;
use EasyChat\Library\Queue\Jobs\MessengerAiTickJob;
use EasyChat\Library\Queue\Jobs\MessengerAvatarJob;
use EasyChat\Library\Queue\Jobs\MessengerVkCallbackRepairJob;
use EasyChat\Library\Queue\Jobs\MessengerVkInboundJob;
use EasyChat\Library\Queue\Jobs\MessengerOutboundJob;
use EasyChat\Library\Queue\Jobs\MessengerMarketplaceSyncJob;
use EasyChat\Library\Queue\Jobs\MessengerMarketplaceAttachmentRetryJob;
use EasyChat\Library\Queue\Jobs\MessengerMarketplaceReadReceiptJob;
use EasyChat\Library\Queue\Jobs\ResponseTimeMetricsBackfillJob;
use EasyChat\Library\Queue\Jobs\SchedulerProbeJob;
use EasyChat\Library\Queue\Jobs\DirectSmsDispatchRecoveryJob;
use EasyChat\Library\Queue\Jobs\ScheduledMessageReleaseJob;
use EasyChat\Library\Queue\Jobs\SmsInboundPushJob;
use EasyChat\Library\Queue\Jobs\SmsInboundFilterRefreshJob;
use EasyChat\Library\Queue\Jobs\ChatAttachmentCleanupJob;
use EasyChat\Library\Queue\Jobs\VoiceMediaFetchJob;
use EasyChat\Library\Queue\Jobs\VoiceTranscriptionJob;
use EasyChat\Library\Queue\Jobs\ChatAssignmentPushJob;
use EasyChat\Library\Queue\Jobs\CrmConversationSyncJob;
use EasyChat\Library\Queue\Jobs\EmailSyncJob;
use EasyChat\Library\Messenger\ContactAvatarService;
use EasyChat\Library\Messenger\SmsDeliveryProjector;
use EasyChat\Library\Catalog\CatalogIndexScheduler;
use EasyChat\Library\Messenger\Marketplace\MarketplaceSyncService;
use EasyChat\Library\Messenger\ScheduledMessageService;
use EasyChat\Library\Messenger\AvitoHistorySyncService;
use EasyChat\Library\Email\CmsMailBridgeHealth;
use EasyChat\Library\Email\CmsOutboundMailBridgeInterface;

class QueueWorker {

    private $registry;

    public function __construct($registry) {
        $this->registry = $registry;
    }

    






    public function run(array $options = []): array {
        $health = new SchedulerHealthService($this->registry);
        $run_id = $health->started((string)($options['runner'] ?? 'standalone'));
        try {
            $result = $this->runBatch($options);
            $health->succeeded($run_id, $result);
            return $result;
        } catch (\Throwable $e) {
            $health->failed($run_id, $e);
            throw $e;
        }
    }

    private function runBatch(array $options = []): array {
        $batch_size = isset($options['batch_size'])
            ? max(1, (int)$options['batch_size'])
            : QueueManager::BATCH_SIZE;
        $time_limit = isset($options['time_limit'])
            ? max(1, (int)$options['time_limit'])
            : 50;

        $result = [
            'processed' => 0,
            'reset' => 0,
            'failed' => 0,
            'rescheduled' => 0,
            'elapsed' => 0.0,
            'stopped_by_time_limit' => false,
            'maintenance' => [],
            'errors' => [],
        ];

        $queue = new QueueManager($this->registry);
        $start_time = microtime(true);

        $reset = $queue->cleanup();
        $result['reset'] = $reset;
        if ($reset > 0) {
            $queue->log("Cleanup: reset $reset stale job(s) to pending");
        }

        $result['maintenance'] = $this->runMaintenance($queue);

        $handlers = $this->handlers();

        for ($i = 0; $i < $batch_size; $i++) {
            if ((microtime(true) - $start_time) >= $time_limit) {
                $result['stopped_by_time_limit'] = true;
                $queue->log('Time limit reached, stopping worker early');
                break;
            }

            $job = $queue->pop();
            if ($job === null) {
                break;
            }

            $job_id = (int)$job['id'];
            $job_type = (string)$job['job_type'];
            $lock_token = (string)($job['lock_token'] ?? '');
            $payload = json_decode((string)$job['payload'], true);

            if (!is_array($payload)) {
                $this->failJob($queue, $result, $job_id, $job_type, 'Invalid JSON in payload column', $lock_token);
                continue;
            }

            if (!isset($handlers[$job_type])) {
                $this->failJob($queue, $result, $job_id, $job_type, "Unknown job type: $job_type", $lock_token);
                continue;
            }

            $class_name = $handlers[$job_type];
            if (!class_exists($class_name)) {
                $this->failJob($queue, $result, $job_id, $job_type, "Handler class is missing: $class_name", $lock_token);
                continue;
            }

            $attempt = (int)$job['attempts'] + 1;

            try {
                $queue->log("Starting job #$job_id (type: $job_type, attempt: $attempt/" . $job['max_attempts'] . ")");
                $handler = new $class_name($this->registry);
                $handler->handle($payload);
                $queue->markDone($job_id, $lock_token);
                $queue->log("Job #$job_id completed successfully");
                $result['processed']++;
            } catch (RequeueWithoutAttemptException $e) {
                $error = $e->getMessage();
                $delay = $e->getDelaySeconds();
                $replacement_payload = $e->getReplacementPayload();
                if (is_array($replacement_payload)) {
                    $queue->rescheduleWithPayload($job_id, $replacement_payload, $delay, $error, $lock_token);
                } else {
                    $queue->reschedule($job_id, $delay, $error, $lock_token);
                }
                $expected_continuation = $e->isExpectedContinuation();
                $queue->log(
                    $expected_continuation
                        ? "Job #$job_id continued with next batch, delay: {$delay}s"
                        : "Job #$job_id rescheduled without attempt increment: $error, delay: {$delay}s"
                );
                $result['rescheduled']++;
                if (!$expected_continuation) {
                    $result['errors'][] = $this->formatError($job_id, $job_type, $error);
                }
            } catch (\Throwable $e) {
                $error = $e->getMessage();
                $queue->markFailed($job_id, $error, $lock_token);
                $queue->log("Job #$job_id failed: $error");
                $result['failed']++;
                $result['errors'][] = $this->formatError($job_id, $job_type, $error);
            }
        }

        $result['elapsed'] = round(microtime(true) - $start_time, 2);

        if ($result['processed'] > 0 || $result['reset'] > 0 || $result['failed'] > 0 || $result['rescheduled'] > 0 || $result['stopped_by_time_limit']) {
            $queue->log(
                'Worker done - processed: ' . $result['processed']
                . ', failed: ' . $result['failed']
                . ', rescheduled: ' . $result['rescheduled']
                . ', reset: ' . $result['reset']
                . ', elapsed: ' . $result['elapsed'] . 's'
            );
        }

        return $result;
    }

    private function handlers(): array {
        return [
            'cms_index' => CmsIndexJob::class,
            'catalog_url_verify' => CatalogUrlVerifyJob::class,
            'embedding' => EmbeddingJob::class,
            'image_variant_generate' => ImageVariantJob::class,
            'messenger_avito_history_sync' => MessengerAvitoHistorySyncJob::class,
            'messenger_avito_item_context_enrich' => MessengerAvitoItemContextEnrichJob::class,
            'messenger_youla_item_context_enrich' => MessengerYoulaItemContextEnrichJob::class,
            'messenger_avatar_fetch' => MessengerAvatarJob::class,
            'messenger_outbound' => MessengerOutboundJob::class,
            'messenger_marketplace_sync' => MessengerMarketplaceSyncJob::class,
            'messenger_marketplace_attachment_retry' => MessengerMarketplaceAttachmentRetryJob::class,
            'messenger_marketplace_read_receipt' => MessengerMarketplaceReadReceiptJob::class,
            'messenger_ai_tick' => MessengerAiTickJob::class,
            'messenger_vk_callback_repair' => MessengerVkCallbackRepairJob::class,
            'messenger_vk_inbound' => MessengerVkInboundJob::class,
            'response_time_metrics_backfill' => ResponseTimeMetricsBackfillJob::class,
            'scheduler_probe' => SchedulerProbeJob::class,
            'direct_sms_dispatch_recovery' => DirectSmsDispatchRecoveryJob::class,
            'scheduled_message_release' => ScheduledMessageReleaseJob::class,
            'sms_inbound_push' => SmsInboundPushJob::class,
            'sms_inbound_filter_refresh' => SmsInboundFilterRefreshJob::class,
            'chat_attachment_cleanup' => ChatAttachmentCleanupJob::class,
            'voice_media_fetch' => VoiceMediaFetchJob::class,
            'voice_transcribe' => VoiceTranscriptionJob::class,
            'chat_assignment_push' => ChatAssignmentPushJob::class,
            'crm_conversation_sync' => CrmConversationSyncJob::class,
            'email_sync' => EmailSyncJob::class,
        ];
    }

    private function runMaintenance(QueueManager $queue): array {
        $maintenance = [
            'sms_delivery' => [
                'ran' => false,
                'expired' => 0,
            ],
            'avatar_backfill' => [
                'ran' => false,
                'queued' => 0,
                'skipped' => false,
                'reason' => '',
            ],
            'catalog_index' => [
                'ran' => false,
                'queued' => 0,
                'reason' => '',
            ],
            'marketplace_sync' => [
                'ran' => false,
                'queued' => 0,
                'reason' => '',
            ],
            'scheduled_messages' => [
                'ran' => false,
                'queued' => 0,
                'recovered' => 0,
            ],
            'email_sync' => ['ran'=>false,'queued'=>0,'reason'=>''],
            'cms_mail_bridge' => ['ran'=>false,'ready'=>false],
            'avito_access' => ['ran'=>false,'queued'=>0],
        ];

        try {
            $maintenance['sms_delivery'] = (new SmsDeliveryProjector($this->registry))->healExpired(200);
            if (!empty($maintenance['sms_delivery']['expired'])) {
                $queue->log('Expired direct SMS projected: ' . (int)$maintenance['sms_delivery']['expired']);
            }
        } catch (\Throwable $e) {
            $message = mb_substr($e->getMessage(), 0, 500);
            $maintenance['sms_delivery'] = [
                'ran' => false,
                'expired' => 0,
                'error' => $message,
            ];
            $queue->log('Direct SMS expiry projection failed: ' . $message);
        }

        try {
            $maintenance['avatar_backfill'] = (new ContactAvatarService($this->registry))->enqueueScheduledBackfill();
            if (!empty($maintenance['avatar_backfill']['ran'])) {
                $queue->log('Avatar scheduled backfill queued: ' . (int)($maintenance['avatar_backfill']['queued'] ?? 0));
            }
        } catch (\Throwable $e) {
            $message = mb_substr($e->getMessage(), 0, 500);
            $maintenance['avatar_backfill'] = [
                'ran' => false,
                'queued' => 0,
                'skipped' => true,
                'reason' => 'error',
                'error' => $message,
            ];
            $queue->log('Avatar scheduled backfill failed: ' . $message);
        }

        try {
            $maintenance['catalog_index'] = (new CatalogIndexScheduler($this->registry))->enqueueIfDue();
            if (!empty($maintenance['catalog_index']['ran'])) {
                $queue->log('Catalog semantic index scheduled: job #' . (int)($maintenance['catalog_index']['job_id'] ?? 0));
            }
        } catch (\Throwable $e) {
            $message = mb_substr($e->getMessage(), 0, 500);
            $maintenance['catalog_index'] = [
                'ran' => false,
                'queued' => 0,
                'reason' => 'error',
                'error' => $message,
            ];
            $queue->log('Catalog semantic index scheduling failed: ' . $message);
        }

        try {
            $maintenance['marketplace_sync'] = (new MarketplaceSyncService($this->registry))->enqueueDueStreams();
            if (!empty($maintenance['marketplace_sync']['queued'])) {
                $queue->log('Marketplace sync queued: ' . (int)$maintenance['marketplace_sync']['queued']);
            }
        } catch (\Throwable $e) {
            $message = mb_substr($e->getMessage(), 0, 500);
            $maintenance['marketplace_sync'] = [
                'ran' => false,
                'queued' => 0,
                'reason' => 'error',
                'error' => $message,
            ];
            $queue->log('Marketplace sync scheduling failed: ' . $message);
        }

        try {
            $maintenance['scheduled_messages'] = (new ScheduledMessageService($this->registry))->enqueueDue(100);
            if (!empty($maintenance['scheduled_messages']['queued']) || !empty($maintenance['scheduled_messages']['recovered'])) {
                $queue->log(
                    'Scheduled messages queued: ' . (int)($maintenance['scheduled_messages']['queued'] ?? 0)
                    . ', recovered: ' . (int)($maintenance['scheduled_messages']['recovered'] ?? 0)
                );
            }
        } catch (\Throwable $e) {
            $message = mb_substr($e->getMessage(), 0, 500);
            $maintenance['scheduled_messages'] = [
                'ran' => false,
                'queued' => 0,
                'recovered' => 0,
                'error' => $message,
            ];
            $queue->log('Scheduled message maintenance failed: ' . $message);
        }

        try {
            $bridge = $this->registry->get('cms_outbound_mail_bridge');
            if ($bridge instanceof CmsOutboundMailBridgeInterface) {
                $snapshot = (new CmsMailBridgeHealth($this->registry))->record($bridge);
                $maintenance['cms_mail_bridge'] = array(
                    'ran' => true,
                    'ready' => CmsMailBridgeHealth::supportsRequiredHeaders((array)$snapshot['capabilities']),
                    'bridge' => (string)$snapshot['bridge'],
                );
            }
        } catch (\Throwable $e) {
            $maintenance['cms_mail_bridge'] = ['ran'=>false,'ready'=>false,'reason'=>'error'];
            $queue->log('CMS mail bridge health failed: '.mb_substr($e->getMessage(),0,200));
        }

        try {
            $maintenance['avito_access'] = [
                'ran' => true,
                'queued' => (new AvitoHistorySyncService($this->registry))->enqueueAccessChecks(8),
            ];
        } catch (\Throwable $e) {
            $maintenance['avito_access'] = ['ran'=>false,'queued'=>0,'reason'=>'error'];
            $queue->log('Avito access monitoring failed: ' . mb_substr($e->getMessage(), 0, 200));
        }

        try {
            $db=$this->registry->get('db');$prefix=easychat_db_prefix();
            $result=$db->query("SELECT `channel_id` FROM `{$prefix}messenger_channels` WHERE `provider`='email' AND `status`='active' ORDER BY `channel_id` ASC LIMIT 10");
            foreach((array)($result->rows??[]) as $row){$id=(int)$row['channel_id'];$queue->pushUnique('email_sync',['channel_id'=>$id],'email_sync_'.$id.'_'.gmdate('YmdHi'),0,8);$maintenance['email_sync']['queued']++;}
            $maintenance['email_sync']['ran']=true;
        } catch (\Throwable $e) {
            $maintenance['email_sync']=['ran'=>false,'queued'=>0,'reason'=>'error'];
            $queue->log('Email sync scheduling failed: '.mb_substr($e->getMessage(),0,200));
        }

        return $maintenance;
    }

    private function failJob(QueueManager $queue, array &$result, int $job_id, string $job_type, string $error, string $lock_token): void {
        $queue->markFailed($job_id, $error, $lock_token);
        $queue->log("Job #$job_id failed: $error");
        $result['failed']++;
        $result['errors'][] = $this->formatError($job_id, $job_type, $error);
    }

    private function formatError(int $job_id, string $job_type, string $error): array {
        return [
            'job_id' => $job_id,
            'job_type' => $job_type,
            'error' => mb_substr($error, 0, 500),
        ];
    }
}
