Have('emails', function ($query) {
                $query->whereIn('status', ['scheduling', 'pending', 'scheduled', 'processing', 'draft']);
                return $query;
            })
            ->withoutGlobalScope('type')
            ->whereIn('type', fluentCrmAutoProcessCampaignTypes())
            ->where('scheduled_at', '<', gmdate('Y-m-d H:i:s', current_time('timestamp') - 300))
            ->get();

        if (!$campaigns->isEmpty()) {

            Campaign::whereIn('id', array_unique($campaigns->pluck('id')->toArray()))
                ->withoutGlobalScope('type')
                ->update([
                    'status' => 'archived'
                ]);

            foreach ($campaigns as $campaign) {
                do_action('fluent_crm/campaign_archived', $campaign);
            }

            return true;
        }

        return false;
    }

    /**
     * @return void
     */
    public static function processWeekly()
    {
        (new Maintenance())->maybeProcessData();

        // Clear email_body from historical 'sent' rows to reclaim disk space.
        // Loop a LIMIT-bounded UPDATE so each statement's row-lock footprint
        // stays small (an unbounded UPDATE on a multi-million-row table holds
        // locks for minutes and stalls report/dashboard SELECTs) while still
        // draining the full backlog in this tick. Going direct to $wpdb skips
        // ORM overhead on what is effectively the same repeated statement.
        try {
            global $wpdb;
            $table         = $wpdb->prefix . 'fc_campaign_emails';
            $chunkSize     = 50000;
            $maxIterations = 100; // safety cap — up to ~5M rows per weekly tick

            for ($i = 0; $i < $maxIterations; $i++) {
                $affected = (int) $wpdb->query(
                    "UPDATE {$table} SET email_body = '' WHERE status = 'sent' AND email_body != '' LIMIT {$chunkSize}"
                );

                if ($affected < $chunkSize || fluentCrmIsMemoryExceeded()) {
                    break;
                }
            }
        } catch (\Exception $e) {
            Helper::debugLog('processWeekly', 'email_body cleanup deferred: ' . $e->getMessage(), 'extended');
        }
    }

    /**
     * Discover and process pending campaigns.
     *
     * Called by cron/Action Scheduler. Handles housekeeping (stale email reset),
     * finds campaigns ready to process, and kicks off processing. For continuous
     * processing, use processCampaignById() via the AJAX handler.
     *
     * @return bool
     */
    public static function processFiveMinutes()
    {
        // Cheap time-based pre-check — skips the lock-acquire round trip when
        // the function is called more frequently than the work needs to run.
        $lastRun = fluentCrmGetOptionCache('_fcrm_last_five_minutes_run', 30);
        if ($lastRun && (time() - $lastRun) < 60) {
            return false;
        }

        // Atomic mutex. The throttle above is non-atomic so two near-simultaneous
        // callers can both pass it; the lock guarantees that only one actually
        // proceeds into discovery + processing.
        if (!self::acquireLock('five_minute_scheduler', 180)) {
            return false;
        }

        try {
            fluentCrmSetOptionCache('_fcrm_last_five_minutes_run', time(), 60);

            self::resetStaleProcessingEmails(100, 'processFiveMinutes');

            $cutOutTime = gmdate('Y-m-d H:i:s', current_time('timestamp') + 360);

            $campaigns = Campaign::whereIn('status', ['pending-scheduled', 'processing'])
                ->withoutGlobalScope('type')
                ->whereIn('type', fluentCrmAutoProcessCampaignTypes())
                ->orderBy('scheduled_at', 'ASC')
                ->where('scheduled_at', '<=', $cutOutTime)
                ->limit(2)
                ->get();

            if ($campaigns->isEmpty()) {
                do_action('fluent_crm_process_automation');
                do_action('fluentcrm_scheduled_hourly_tasks');
                return false;
            }

            $firstCampaign = $campaigns->first();

            if ($firstCampaign->status == 'pending-scheduled') {
                $firstCampaign->status = 'processing';
                $firstCampaign->save();
            }

            $result = self::processCampaignById($firstCampaign->id);

            // If first campaign is done and there are more queued, chain the next one.
            // Skip if memory is low (aborted) to avoid cascading failures.
            if (!$result && count($campaigns) > 1 && !fluentCrmIsMemoryExceeded()) {
                // Verify first campaign actually finished (not just aborted)
                $firstCampaign = Campaign::withoutGlobalScope('type')->find($firstCampaign->id);
                if ($firstCampaign && $firstCampaign->status != 'processing') {
                    $nextCampaign = $campaigns->last();
                    if ($nextCampaign->status == 'pending-scheduled') {
                        $nextCampaign->status = 'processing';
                        $nextCampaign->save();
                    }
                    self::fireCampaignProcessingChain($nextCampaign->id);
                }
            }

            return $result;
        } finally {
            self::releaseLock('five_minute_scheduler');
        }
    }

    /**
     * Reset rows stuck in 'processing' back to 'pending' so they get re-claimed.
     *
     * An unbounded mass UPDATE on (status='processing' AND updated_at < cutoff)
     * locks a wide range and deadlocks against the row-level SELECT ... FOR
     * UPDATE claims that the mailer Handler / MultiThreadHandler hold while
     * sending. We instead drain in bounded chunks by primary key.
     *
     * We deliberately do NOT order the SELECT: ORDER BY id would push MySQL
     * onto PRIMARY (full id-walk looking for sparse matches on a multi-million
     * row table) instead of the (status, scheduled_at) index, which contains
     * only the small currently-'processing' slice. Each chunk drains rows out
     * of the predicate, so the next iteration naturally finds different rows
     * without an explicit order.
     *
     * Any deadlock that still slips through is harmless — remaining rows will
     * be picked up on the next caller's tick.
     *
     * @param int    $maxAgeSeconds Rows older than this (in 'processing') get reset.
     * @param string $callerContext Used in the deferred-log message.
     * @return void
     */
    public static function resetStaleProcessingEmails($maxAgeSeconds = 100, $callerContext = '')
    {
        try {
            $staleCutoff = gmdate('Y-m-d H:i:s', current_time('timestamp') - (int) $maxAgeSeconds);
            $chunkSize   = 200;
            $maxChunks   = 50; // up to 10k rows per call; subsequent calls drain the rest

            for ($i = 0; $i < $maxChunks; $i++) {
                $staleIds = CampaignEmail::where('status', 'processing')
                    ->where('updated_at', '<', $staleCutoff)
                    ->limit($chunkSize)
                    ->pluck('id')
                    ->toArray();

                if (empty($staleIds)) {
                    break;
                }

                CampaignEmail::whereIn('id', $staleIds)
                    ->where('status', 'processing')
                    ->update([
                        'status' => 'pending'
                    ]);

                if (count($staleIds) < $chunkSize || fluentCrmIsMemoryExceeded()) {
                    break;
                }
            }
        } catch (\Exception $e) {
            Helper::debugLog($callerContext ?: 'resetStaleProcessingEmails', 'Stale email reset deferred: ' . $e->getMessage(), 'extended');
        }
    }

    /**
     * Process a specific campaign by ID.
     *
     * Can be called directly from the AJAX handler for continuous chaining
     * without re-discovering campaigns or running housekeeping.
     *
     * @param int $campaignId
     * @return bool True if more processing is needed, false if done.
     */
    public static function processCampaignById($campaignId)
    {
        // Per-campaign scheduler lock. processCampaignById has two entry points
        // — the AJAX self-trigger fluentcrm-post-campaigns-emails-processing
        // (which bypasses processFiveMinutes' scheduler-level lock entirely)
        // and processFiveMinutes itself (which holds five_minute_scheduler).
        // Without this guard, fireCampaignProcessingChain could pile up
        // overlapping AJAX requests for the same campaign that all reach
        // CampaignProcessor and bail at its per-campaign lock — wasted PHP
        // bootstraps. Lock name is per-campaign so different campaigns still
        // process in parallel. TTL matches the set_time_limit(120) below.
        $lockName = 'campaign_chain_' . (int)$campaignId;
        if (!self::acquireLock($lockName, 120)) {
            return false;
        }

        try {
            if (function_exists('set_time_limit')) {
                @set_time_limit(120);
            }

            $campaign = Campaign::withoutGlobalScope('type')->find($campaignId);
            if (!$campaign) {
                return false;
            }

            $campaignProcessingChunk = (int)apply_filters('fluent_crm/five_minute_campaign_processing_chunk', 20, $campaign);
            if ($campaignProcessingChunk < 1) {
                $campaignProcessingChunk = 1;
            }

            $runTime = fluentCrmMaxRunTime() - 5;
            $campaign = (new CampaignProcessor($campaignId))->processEmails($campaignProcessingChunk, $runTime);

            if (fluentCrmIsMemoryExceeded()) {
                return false;
            }

            if ($campaign && $campaign->status == 'processing') {
                self::fireCampaignProcessingChain($campaignId);
                return true;
            }

            return false;
        } finally {
            self::releaseLock($lockName);
        }
    }

    /**
     * Fire a background AJAX request to continue processing a specific campaign.
     *
     * @param int $campaignId
     */
    private static function fireCampaignProcessingChain($campaignId)
    {
        $url = add_query_arg([
            'action'      => 'fluentcrm-post-campaigns-emails-processing',
            'campaign_id' => $campaignId,
            'time'        => time()
        ], admin_url('admin-ajax.php'));

        \FluentCrm\App\Services\Libs\Mailer\Handler::fireNonBlockingRequest($url, [
            'retry' => 1
        ]);
    }

    public static function maybeCleanupCsvFiles()
    {
        $dir = FileSystem::getDir();

        // loop through files in directory
        foreach (glob($dir . '/fluentcrm-*.csv') as $filename) {
            // check if file was created before last 30 minutes
            if (time() - filectime($filename) >= 1800) {
                wp_delete_file($filename); // delete file
            }
        }
    }

    public static function processMultiThreadEmails()
    {
        (new MultiThreadHandler())->handle();
        return true;
    }

    /**
     * Atomically claim a scheduler-level lock so two runners can't enter the
     * same critical section concurrently (e.g. Action Scheduler + WP-Cron
     * minute ticks landing in the same second).
     *
     * Uses wp_cache_add() when an external object cache is available, otherwise
     * a conditional UPDATE on wp_options keyed off a timestamp. The UPDATE
     * succeeds only if the row is unclaimed or its stored timestamp is older
     * than $ttl, so a crashed runner's lock self-recovers after the TTL.
     *
     * Mirrors BaseHandler::acquireLock() / FunnelHandler::acquireFunnelProcessorLock().
     * Kept local instead of extracted to a shared helper to limit blast radius.
     *
     * @param string $name Lock identifier appended to the option key.
     * @param int    $ttl  Seconds before a held lock is considered abandoned.
     * @return bool True if the lock was acquired by this process.
     */
    private static function acquireLock($name, $ttl)
    {
        $key = '_fluentcrm_lock_' . $name;
        $now = time();

        if (wp_using_ext_object_cache()) {
            if (wp_cache_add($key, $now, 'fc_instant_options', $ttl)) {
                return true;
            }

            $existing = wp_cache_get($key, 'fc_instant_options');
            if ($existing && ($now - (int)$existing) > $ttl) {
                wp_cache_delete($key, 'fc_instant_options');
                if (wp_cache_add($key, $now, 'fc_instant_options', $ttl)) {
                    return true;
                }
            }

            return false;
        }

        global $wpdb;

        $wpdb->query($wpdb->prepare(
            "INSERT IGNORE INTO {$wpdb->options} (option_name, option_value, autoload) VALUES (%s, %s, %s)",
            $key, '', 'no'
        ));

        $affected = $wpdb->query($wpdb->prepare(
            "UPDATE {$wpdb->options} SET option_value = %s WHERE option_name = %s AND (option_value = '' OR option_value < %d)",
            (string)$now, $key, $now - $ttl
        ));

        if ($affected > 0) {
            wp_cache_delete($key, 'options');
            return true;
        }

        return false;
    }

    /**
     * Release a scheduler-level lock previously acquired by acquireLock().
     * Safe to call even if the lock was not held by this process — the worst
     * case is freeing the slot a tick early.
     */
    private static function releaseLock($name)
    {
        $key = '_fluentcrm_lock_' . $name;

        if (wp_using_ext_object_cache()) {
            wp_cache_delete($key, 'fc_instant_options');
        } else {
            update_option($key, '', false);
        }
    }
}