nhance_partner_be/app/Controllers/JobWorker.php

<?php

namespace App\Controllers;

use App\Models\JobModel;
use App\Models\FileModel;
class JobWorker extends BaseController
{
    const STATUS_DONE = 'done';
    const STATUS_QUEUED = 'queued';
    const STATUS_RUNNING = 'running';
    const STATUS_FAILED = 'failed';
    /**
     * Constructs the class
     */

    private static $event_class_mapping = 
    [
        'readFileAndCalculateCommission' => [
            'type' => 'CC', // Handler Category (Possible values: HC, CC, HF)
            'handler' => 'App\Controllers\PolicyController',
        ],
    ];

    

    public function __construct()
    {
       
    }

    public static function streamOutput($data)
    {
        ob_implicit_flush(true);
        // try { ob_end_flush(); } catch(Exception $e) { echo $e->getMessage(); }
        echo $data;
        flush();
    }

    public static function  processJobs(array $jobdata = [])
    {
        $db = \Config\Database::connect();

        $runningJob = $db->query(
            "SELECT id FROM partner_jobs WHERE status=? LIMIT 1",
            [self::STATUS_RUNNING]
        )->getRow();

        if ($runningJob !== null) {
            SELF::streamOutput("Another job is already running (id: {$runningJob->id}). Skipping.\n");
            return false;
        }

        $query = "
            SELECT id, name, payload, uuid
            FROM partner_jobs
            WHERE status=?
            ORDER BY created_dt ASC
            LIMIT 1";
        $jobs = $db->query($query, [self::STATUS_QUEUED])->getResult();

        if (count($jobs)) {
            $job = $jobs[0];
            return SELF::processJob(['id' => $job->id, 'uuid' => $job->uuid]);
        }

        return false;
    }
    /**
     * process jobs
     */
    public static function  processJob(array $jobdata = [])
    {

        // print_r($jobdata);
        // echo 'listen';
        // die();

        $db = \Config\Database::connect();
        $db->transStart();

        // Do not start a new job while another job is already running.
        $runningJob = $db->query(
            "SELECT id FROM partner_jobs WHERE status=? LIMIT 1",
            [self::STATUS_RUNNING]
        )->getRow();

        if ($runningJob !== null) {
            $db->transComplete();
            SELF::streamOutput("Another job is already running (id: {$runningJob->id}). Skipping.\n");
            return false;
        }

        if (isset($jobdata) && count($jobdata)) {
            $query = "
            SELECT id, name, payload, uuid
            FROM partner_jobs
            WHERE status=? AND id=? AND uuid=?
            ORDER BY created_dt ASC
            LIMIT 1 FOR UPDATE";
            $where_condition = [self::STATUS_QUEUED, $jobdata['id'], $jobdata['uuid']];
        } else {
            $query = "
            SELECT id, name, payload, uuid
            FROM partner_jobs
            WHERE status=?
            ORDER BY created_dt ASC
            LIMIT 1 FOR UPDATE";
            $where_condition = [self::STATUS_QUEUED];
        }

        $job = $db->query($query, $where_condition)->getResult();
        $db->transComplete();
        // print_r($job);die();
        if ($job !== [])
        {
            $job = $job[0];

           

            // echo "\nProcessing job id - " . $job->id . "\n";
            SELF::streamOutput("\nProcessing job id - " . $job->id . "\n");
            // sleep(5);
            // echo "Job name - " . $job->name . "\n";
            SELF::streamOutput("Job name - " . $job->name . "\n");
            // sleep(5);

            // print_r(SELF::$event_class_mapping);
            // echo array_key_exists($job->name,SELF::$event_class_mapping) ? 'mapped' : 'notmapped';
            // die();
            try
            {
                $start = microtime(true);
                $runtime = null;
                $job_status = self::STATUS_RUNNING;
                $db->query("UPDATE partner_jobs SET status=? WHERE id=? AND uuid=?", [$job_status, $job->id,$job->uuid]);
                
                if (!array_key_exists($job->name,SELF::$event_class_mapping))
                {
                    throw new \RuntimeException('Job ' . $job->name . ' handler not registered');
                }

                $handler = SELF::$event_class_mapping[$job->name];
                $handleInstance = null;

                if($handler['type'] == 'CC' || $handler['type'] == 'HC')
                {
                    // echo 'CLASS - ' . $handler['type'].' - ' . $handler['handler'];
                    $handlerClass = $handler['handler'];
                    if(class_exists($handlerClass))
                    {
                        $handleInstance = new $handlerClass();
                    }
                    else
                    {
                        throw new \RuntimeException('Job ' . $job->name . ' or handler class not found');
                        // echo $e->getMessage();
                    }

                    if (method_exists($handleInstance, $job->name))
                    {
                        $jobHandler = [$handleInstance, $job->name];
                        //throw new \RuntimeException('Job ' . $job->name . ' not found');
                    }
                    else if(method_exists($handleInstance, 'handle'))
                    {
                        $jobHandler = [$handleInstance, 'handle'];
                    }
                    else
                    {
                        throw new \RuntimeException('Job ' . $job->name . ' or handler not found');
                    }
                }
                else if($handler['type'] == 'HF')
                {
                    // echo 'HF - ' . $handler['type'];
                    $jobHandler = $handleInstance = $handler['handler'];
                    // echo $jobHandler;

                }
                else{
                    throw new \RuntimeException('Job ' . $job->name . ' Invalid job type');
                }
                
                $payload = json_decode($job->payload, true);
                if (!is_array($payload))
                {
                    throw new \InvalidArgumentException('Invalid payload format here');
                }
                
                $payload = is_array($payload) ? $payload : [];
                try
                {
                    $response = $jobHandler($payload,$job->id);
                    $job_status = self::STATUS_DONE;
                }
                catch(\Exception $e)
                {
                    $job_status = self::STATUS_FAILED;
                    $runtime = $runtime === null ?  microtime(true) - $start : $runtime;
                    $response = ['file_name' => $e->getFile(),'error' => $e->getMessage(),'line_no' => $e->getLine(),'info' => $e->getTraceAsString(),'scope' => 'task failed'];
                }

                $runtime = microtime(true) - $start;
               
            }
            catch (\Exception $e)
            {
                //die();
                $job_status = self::STATUS_FAILED;
                $runtime = $runtime === null ?  microtime(true) - $start : $runtime;
                $response = ['file_name' => $e->getFile(),'error' => $e->getMessage(),'line_no' => $e->getLine(),'info' => $e->getTraceAsString(),'scope' => 'worker failed'];
            }

            $db->query("UPDATE partner_jobs SET status=?, run_time=?, response=? WHERE id=? AND uuid=?", [
                $job_status,
                $runtime,
                json_encode($response),
                $job->id,$job->uuid
            ]);

            //update file status if this job is directly linked with a file id

            if($job_status == self::STATUS_FAILED)
            {
                //get file from job payload
                // if(array_key_exists('file_id', $payload) && $payload['file_id'] != NULL && $payload['file_id'] != "" &&  is_numeric($payload['file_id']) && count($payload) == 1)
                // {
                //      $fileModel = new FileModel();
                //      $fileModel->where('id', $payload['file_id'])
                //                      ->set(['status' => 'failed','reason' => json_encode(['error_type' => 0,'error_summary' => [0],'error_data' => 'System Error, please contact Admin/Support team']) ])
                //                      ->update();
                // }

            }

           // echo "Job $job->id  $job_status. Response - $response \n";
            // echo "Job $job->id  $job_status \n";
            SELF::streamOutput("Job $job->id  $job_status \n");
            return true;
        }
        else
        {
            echo 'no job found in queue';
        }
    }



    
    

}