<?php

namespace App\Console\Commands;

use App\Jobs\SendCampaignEmailJob;
use App\Jobs\SendCampaignSmsJob;
use App\Models\Campaign;
use App\Models\CampaignRecipient;
use App\Services\Queue\QueueHealthService;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;

/**
 * Re-dispatches jobs for campaign recipients that ARE saved as `pending`
 * but have no matching row in the `jobs` table.
 *
 * Two scenarios this covers (both reported by customers who ended up
 * writing their own scripts to unstick the inbox):
 *
 *  1. The "withOperationLock silently failed" bug, pre-fix — recipients
 *     were created by `resolveAudience()` but the per-recipient job
 *     dispatch loop never ran because something earlier in the
 *     transaction caught the lock failure and bailed. Result: a
 *     campaign in status='sending' with N recipients all stuck at
 *     'pending' and zero entries in the `jobs` table.
 *
 *  2. The worker hard-crashed mid-dispatch (OOM, supervisor restart)
 *     between two `chunk()` calls, so some recipients got their jobs
 *     queued and some didn't. The half that didn't are forever stuck.
 *
 * Detection logic: find every CampaignRecipient with `status='pending'`
 * whose parent campaign is in `status='sending'` and has been so for
 * more than $thresholdMinutes. For each, check that no live job exists
 * in the `jobs` table referencing that recipient. Re-dispatch only the
 * orphans. Sends already-queued recipients are untouched.
 *
 * Scheduled in routes/console.php (every 5 minutes, withoutOverlapping).
 *
 * Idempotent: each SendCampaignEmailJob / SendCampaignSmsJob claims its
 * recipient with an atomic `WHERE status='pending'` UPDATE, so even if
 * we double-dispatch by accident the duplicate exits cleanly.
 */
class ProcessPendingCampaignRecipientsCommand extends Command
{
    protected $signature = 'campaigns:process-pending
                            {--threshold-minutes=5 : Only consider recipients pending for at least this long.}
                            {--limit=500 : Max recipients to re-dispatch per run.}
                            {--dry-run : Print what would change without dispatching.}';

    protected $description = 'Re-dispatch jobs for campaign recipients stuck at "pending" with no matching job in the queue.';

    public function handle(QueueHealthService $health): int
    {
        $threshold = max(1, (int) $this->option('threshold-minutes'));
        $limit     = max(1, (int) $this->option('limit'));
        $dryRun    = (bool) $this->option('dry-run');

        $cutoff = now()->subMinutes($threshold);

        // Pull pending recipients whose parent campaign is mid-send.
        // Join the campaign so we know the channel (email vs sms) for
        // the dispatch and so we can scope to in-flight sends only.
        $candidates = CampaignRecipient::query()
            ->where('campaign_recipients.status', 'pending')
            ->where('campaign_recipients.updated_at', '<', $cutoff)
            ->whereHas('campaign', function ($q) {
                $q->where('status', 'sending');
            })
            ->with(['campaign:id,channel,workspace_id,name'])
            ->limit($limit)
            ->get();

        if ($candidates->isEmpty()) {
            $this->info('No stuck pending recipients found.');
            return self::SUCCESS;
        }

        $queueStatus = $health->status('campaigns');
        $this->line(sprintf(
            'Queue status: %s | driver=%s | pending=%d | stale=%d',
            $queueStatus['alive'] ? 'ALIVE' : 'DOWN',
            $queueStatus['driver'],
            $queueStatus['pending_jobs'],
            $queueStatus['stale_jobs']
        ));

        if (! $queueStatus['alive']) {
            // No point dispatching more jobs onto a dead queue — they'd
            // just join the existing pile. Surface the situation and
            // bail so the operator runs `queue:work` first.
            $this->warn('Queue worker is DOWN. Skipping dispatch — start the worker first: php artisan queue:work --queue=campaigns,default');
            return self::SUCCESS;
        }

        $this->newLine();

        // Build the set of recipient ids that already have a live job
        // in the queue, in one shot. This is the cheap version — we
        // search the serialized payload for the recipient id. On the
        // database queue driver `jobs.payload` is the serialized job;
        // SendCampaignEmailJob carries the CampaignRecipient model with
        // its id. A LIKE on `"id":N,` is cheap because we limit by id.
        $queuedRecipientIds = $this->findRecipientIdsInJobsTable(
            $candidates->pluck('id')->all()
        );

        $reDispatched = 0;
        $alreadyQueued = 0;
        $errors = 0;

        foreach ($candidates as $recipient) {
            if (in_array((int) $recipient->id, $queuedRecipientIds, true)) {
                $alreadyQueued++;
                continue;
            }

            $channel = $recipient->campaign?->channel ?: 'email';
            $jobClass = $channel === 'sms' ? SendCampaignSmsJob::class : SendCampaignEmailJob::class;

            $this->line(sprintf(
                '%s recipient #%d (campaign #%d "%s", channel=%s)',
                $dryRun ? '[DRY-RUN]' : 'DISPATCH',
                $recipient->id,
                $recipient->campaign?->id ?? 0,
                $recipient->campaign?->name ?? '?',
                $channel
            ));

            if ($dryRun) {
                continue;
            }

            try {
                $jobClass::dispatch($recipient)->onQueue('campaigns');
                $reDispatched++;
            } catch (\Throwable $e) {
                $errors++;
                Log::error('campaigns:process-pending — failed to dispatch', [
                    'recipient_id' => $recipient->id,
                    'campaign_id'  => $recipient->campaign?->id,
                    'error'        => $e->getMessage(),
                ]);
                $this->error("  ↳ dispatch failed: {$e->getMessage()}");
            }
        }

        $this->newLine();
        $this->info(sprintf(
            '%s %d recipient(s) | already queued: %d | errors: %d (total candidates: %d)',
            $dryRun ? 'Would re-dispatch' : 'Re-dispatched',
            $reDispatched,
            $alreadyQueued,
            $errors,
            $candidates->count()
        ));

        if ($reDispatched > 0) {
            Log::warning('campaigns:process-pending re-dispatched stuck recipients', [
                'count'   => $reDispatched,
                'errors'  => $errors,
                'reaper'  => 'process-pending',
            ]);
        }

        return self::SUCCESS;
    }

    /**
     * Find which of the given recipient ids already have a live job
     * sitting in the `jobs` table — so we don't double-dispatch.
     *
     * Best-effort: only the database queue driver exposes the jobs
     * table. On redis/sqs we have no cheap way to inspect; return [] so
     * every candidate gets re-dispatched. That's safe because the jobs
     * themselves are idempotent (atomic claim).
     */
    private function findRecipientIdsInJobsTable(array $ids): array
    {
        if (empty($ids) || config('queue.default') !== 'database') {
            return [];
        }

        try {
            $found = [];
            // Search in chunks to keep the LIKE list bounded.
            foreach (array_chunk($ids, 100) as $chunk) {
                $query = DB::table('jobs')->where('queue', 'campaigns');
                $query->where(function ($q) use ($chunk) {
                    foreach ($chunk as $id) {
                        // CampaignRecipient is serialized as
                        // {"id":N,"connection":"...","class":"..."} so
                        // an `"id":N,` substring is a reliable match.
                        $q->orWhere('payload', 'like', '%"id":'.(int) $id.',%');
                    }
                });
                $payloads = $query->pluck('payload');
                foreach ($payloads as $payload) {
                    foreach ($chunk as $id) {
                        if (str_contains($payload, '"id":'.(int) $id.',')) {
                            $found[] = (int) $id;
                        }
                    }
                }
            }
            return array_values(array_unique($found));
        } catch (\Throwable $e) {
            Log::warning('process-pending: jobs-table scan failed; assuming no duplicates', [
                'error' => $e->getMessage(),
            ]);
            return [];
        }
    }
}
