<?php

namespace FluentCrm\App\Hooks\Handlers;

use FluentCrm\App\Models\Campaign;
use FluentCrm\App\Models\CampaignEmail;
use FluentCrm\App\Services\CampaignProcessor;
use FluentCrm\App\Services\ExternalIntegrations\Maintenance;
use FluentCrm\App\Services\Helper;
use FluentCrm\App\Services\Libs\FileSystem;
use FluentCrm\App\Services\Libs\Mailer\Handler;
use FluentCrm\App\Services\Libs\Mailer\MultiThreadHandler;

/**
 *  Scheduler Class
 *
 * @package FluentCrm\App\Hooks
 *
 * @version 1.0.0
 */
class Scheduler
{

    public static function register()
    {
        /*
         * Migrating from CRON to Action Scheduler for Every Minutes Tasks
         */
        add_action('fluentcrm_scheduled_minute_tasks', function () {
            // Auto-migration: ensure the Action Scheduler recurring action exists.
            if (!as_has_scheduled_action('fluentcrm_scheduled_every_minute_tasks', [], 'fluent-crm')) {
                Helper::debugLog('Migrating Every Minute CRON to Action Scheduler for FluentCRM');
                as_schedule_recurring_action(time(), 60, 'fluentcrm_scheduled_every_minute_tasks', [], 'fluent-crm');
                return;
            }

            // WP-Cron is a TRUE fallback: only take over when Action Scheduler
            // has actually stalled. _fcrm_last_scheduler is written by
            // Scheduler::process() on every successful AS-driven minute tick;
            // if it's fresh, AS owns this minute and we no-op here.
            $lastScheduler = fluentCrmGetOptionCache('_fcrm_last_scheduler');
            if ($lastScheduler && (time() - $lastScheduler) <= 70) {
                return;
            }

            // AS appears stalled (or has never run on this site yet) — take
            // over via the same locked entry point AS uses. The atomic lock
            // inside process() prevents two concurrent WP-Cron runners both
            // deciding to take over from racing each other.
            self::process();
        });

        // This is required to instantly send emails for regular email handler
        add_action('wp_ajax_nopriv_fluentcrm-post-campaigns-send-now', function () {
            if (!get_option('fluentcrm_is_sending_emails')) {
                $nextCron = as_next_scheduled_action('fluentcrm_scheduled_every_minute_tasks');
                $willRun = !$nextCron || $nextCron == 1 || ($nextCron - time()) >= 3 || ($nextCron - time()) < -70;

                if (!$willRun) {
                    $lastCalled = (int)fluentcrm_get_option('fluentcrm_is_sending_emails_last_called');
                    if ($lastCalled && (time() - $lastCalled) < 52) {
                        $willRun = true;
                    }
                }

                if ($willRun) {
                    Helper::debugLog('AJAX: post-campaigns-send-now', 'Timing: ' . ($nextCron - time()), 'extended');
                    $mailer = new \FluentCrm\App\Services\Libs\Mailer\Handler();
                    $mailer->handle();
                }
            }

            nocache_headers();
            wp_send_json_success([
                'message'   => 'success',
                'timestamp' => time()
            ]);
        });

        // For Multi Threaded Emails Internal Ajax
        add_action('wp_ajax_nopriv_fluentcrm-post-multi-thread-send-now', function () {

            if (!get_option('fluentcrm_is_sending_multi_emails')) {
                $nextCron = as_next_scheduled_action('fluent_crm_send_multi_thread_emails');
                $willRun = !$nextCron || $nextCron == 1 || ($nextCron - time()) >= 3 || ($nextCron - time()) < -70;

                if (!$willRun) {
                    $lastCalled = (int)fluentcrm_get_option('fluentcrm_is_sending_multi_emails_last_called');
                    if ($lastCalled && (time() - $lastCalled) < 52) {
                        $willRun = true;
                    }
                }

                if ($willRun) {
                    if (Helper::isExperimentalEnabled('multi_threading_emails')) {
                        (new MultiThreadHandler())->handle();
                    }
                }
            }

            nocache_headers();
            wp_send_json_success([
                'message'   => 'success',
                'timestamp' => time()
            ]);
        });

        add_action('fluentcrm_scheduled_every_minute_tasks', array(__CLASS__, 'process'));
        add_action('fluentcrm_scheduled_hourly_tasks', array(__CLASS__, 'processHourly'));
        add_action('fluentcrm_scheduled_five_minute_tasks', array(__CLASS__, 'processFiveMinutes'));
        add_action('fluentcrm_process_contact_jobs', array(__CLASS__, 'processForSubscriber'), 999, 1);
        add_action('fluentcrm_scheduled_weekly_tasks', array(__CLASS__, 'processWeekly'));
        add_action('fluent_crm_send_multi_thread_emails', array(__CLASS__, 'processMultiThreadEmails'), 10);

        add_action('fluent_crm_cancel_multi_thread_mailing', function () {
            as_unschedule_all_actions('fluent_crm_send_multi_thread_emails');
            return true;
        });

        /*
         *  Clean up schedule that means removing from database-  tasks by action scheduler
         * Clean up before last 7 days logs generated by action scheduler
         * this action will be triggered daily and will remove all the logs generated before 7 days
         */
        add_action('fluent_crm_ascheduler_runs_daily', function () {
            Cleanup::maybeRemoveOldScheuledActionLogs();
        });

    }

    public static function process()
    {
        wp_raise_memory_limit('admin');

        // In-process re-entrance guard (cheap; complements the cross-process lock below).
        if (did_action('fluentcrm_process_scheduled_tasks_init')) {
            return false;
        }

        // Atomic cross-process mutex. Prevents concurrent AS + WP-Cron + AJAX
        // runners from all reaching Handler->handle() at the same time. The
        // downstream BaseHandler also has its own lock — this outer guard
        // avoids wasted PHP bootstraps for the loser of the race.
        if (!self::acquireLock('minute_scheduler', 90)) {
            return false;
        }

        try {
            // _fcrm_last_scheduler stays as the success-timestamp signal used
            // by the WP-Cron fallback in register() to detect a stalled Action
            // Scheduler. It is no longer the gate that prevents re-entry —
            // that role belongs to the atomic lock above.
            fluentCrmSetOptionCache('_fcrm_last_scheduler', time(), 50);
            do_action('fluentcrm_process_scheduled_tasks_init');

            (new Handler)->handle();
        } finally {
            self::releaseLock('minute_scheduler');
        }

        return true;
    }

    public static function processForSubscriber($subscriber)
    {
        if (!is_object($subscriber) || empty($subscriber->id)) {
            return false;
        }

        if (!defined('FLUENTCRM_DOING_BULK_IMPORT')) {
            // @todo: Implement this immediately
            (new Handler)->processSubscriberEmail($subscriber->id);
        }

        return true;
    }

    public static function processHourly()
    {
        // Atomic mutex. Closes the duplicate-event leak in markArchiveCampaigns():
        // without this, two concurrent hourly runners both pass the SELECT,
        // both UPDATE rows to 'archived' (idempotent), and both fire
        // fluent_crm/campaign_archived for the same campaign — causing
        // listeners (webhooks, metrics, notifications) to fire twice.
        if (!self::acquireLock('hourly_scheduler', 300)) {
            return;
        }

        try {
            self::markArchiveCampaigns();
            self::maybeCleanupCsvFiles();
            do_action('fluent_crm_process_automation');
        } finally {
            self::releaseLock('hourly_scheduler');
        }
    }


    public static function markArchiveCampaigns()
    {
        // get the scheduled or working  campaigns where scheduled_at is five minutes ago
        $campaigns = Campaign::whereIn('status', ['working', 'scheduled'])
            ->whereDoesntHave('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);
        }
    }
}
