<?php
declare(strict_types=1);

/**
 * CakePHP(tm) : Rapid Development Framework (https://cakephp.org)
 * Copyright (c) Cake Software Foundation, Inc. (https://cakefoundation.org/)
 *
 * Licensed under The MIT License
 * For full copyright and license information, please see the LICENSE.txt
 * Redistributions of files must retain the above copyright notice.
 *
 * @copyright     Copyright (c) Cake Software Foundation, Inc. (https://cakefoundation.org/)
 * @link          https://cakephp.org CakePHP(tm) Project
 * @since         2.2.0
 * @license       https://opensource.org/licenses/MIT MIT License
 */
namespace Cake\Queue\Queue;

use Cake\Core\ContainerInterface;
use Cake\Queue\Job\Message;
use Interop\Queue\Message as QueueMessage;
use Psr\Log\LoggerInterface;
use RuntimeException;

/**
 * Subprocess processor that executes jobs in isolated PHP processes.
 *
 * This processor spawns a new PHP process for each job, providing complete isolation
 * between jobs. This is useful for development environments where code changes need
 * to be reloaded without restarting the worker.
 *
 * Configuration options:
 * - `command`: Full command to execute (default: 'php bin/cake.php queue subprocess_runner')
 * - `timeout`: Maximum execution time in seconds (default: 300)
 * - `maxOutputSize`: Maximum output size in bytes (default: 1048576 = 1MB)
 *
 * Example configuration:
 * ```
 * 'Queue' => [
 *     'default' => [
 *         'subprocess' => [
 *             'command' => 'php bin/cake.php queue subprocess_runner',
 *             'timeout' => 60,
 *             'maxOutputSize' => 2097152, // 2MB
 *         ],
 *     ],
 * ],
 * ```
 *
 * Extends Processor to reuse event handling and processing logic (DRY principle).
 */
class SubprocessProcessor extends Processor
{
    /**
     * @param \Psr\Log\LoggerInterface $logger Logger instance
     * @param array<string, mixed> $config Subprocess configuration options
     * @param \Cake\Core\ContainerInterface|null $container DI container instance
     */
    public function __construct(
        LoggerInterface $logger,
        protected readonly array $config = [],
        ?ContainerInterface $container = null,
    ) {
        parent::__construct($logger, $container);
    }

    /**
     * Execute the job in a subprocess.
     *
     * @param \Cake\Queue\Job\Message $jobMessage Job message wrapper
     * @param \Interop\Queue\Message $queueMessage Original queue message
     * @return object|string with __toString method implemented
     */
    protected function executeJob(Message $jobMessage, QueueMessage $queueMessage): string|object
    {
        $jobData = $this->prepareJobData($queueMessage);
        $subprocessResult = $this->executeInSubprocess($jobData);

        return $this->handleSubprocessResult($subprocessResult, $queueMessage);
    }

    /**
     * Handle subprocess result and return appropriate response.
     *
     * @param array<string, mixed> $result Subprocess result
     * @param \Interop\Queue\Message $message Original message
     * @return string
     * @throws \RuntimeException
     */
    protected function handleSubprocessResult(array $result, QueueMessage $message): string
    {
        if ($result['success']) {
            return $result['result'];
        }

        if (isset($result['exception'])) {
            $exception = $this->reconstructException($result['exception']);
            $message->setProperty('jobException', $exception);

            throw $exception;
        }

        throw new RuntimeException($result['error'] ?? 'Subprocess execution failed');
    }

    /**
     * Prepare job data for subprocess execution.
     *
     * @param \Interop\Queue\Message $message Message
     * @return array<string, mixed>
     */
    protected function prepareJobData(QueueMessage $message): array
    {
        $body = json_decode($message->getBody(), true);
        if (json_last_error() !== JSON_ERROR_NONE) {
            throw new RuntimeException('Invalid JSON in message body');
        }

        $properties = $message->getProperties();

        return [
            'messageClass' => get_class($message),
            'body' => $body,
            'properties' => $properties,
            'logger' => $this->config['logger'] ?? 'stderr',
        ];
    }

    /**
     * Execute job in subprocess.
     *
     * @param array<string, mixed> $jobData Job data
     * @return array<string, mixed>
     */
    protected function executeInSubprocess(array $jobData): array
    {
        $command = $this->config['command'] ?? 'php bin/cake.php queue subprocess_runner';
        $timeout = $this->config['timeout'] ?? 300;

        $descriptors = [
            0 => ['pipe', 'r'],
            1 => ['pipe', 'w'],
            2 => ['pipe', 'w'],
        ];

        $process = proc_open($command, $descriptors, $pipes);

        if (!is_resource($process)) {
            throw new RuntimeException('Failed to create subprocess');
        }

        try {
            $jobDataJson = json_encode($jobData);
            if ($jobDataJson !== false) {
                fwrite($pipes[0], $jobDataJson);
            }

            fclose($pipes[0]);

            $output = '';
            $errorOutput = '';
            $startTime = time();
            $maxOutputSize = $this->config['maxOutputSize'] ?? 1048576; // 1MB default

            stream_set_blocking($pipes[1], false);
            stream_set_blocking($pipes[2], false);

            while (true) {
                if ($timeout > 0 && (time() - $startTime) > $timeout) {
                    proc_terminate($process, 9);

                    return [
                        'success' => false,
                        'error' => sprintf('Subprocess execution timeout after %d seconds', $timeout),
                    ];
                }

                $read = [$pipes[1], $pipes[2]];
                $write = null;
                $except = null;
                $selectResult = stream_select($read, $write, $except, 1);

                if ($selectResult === false) {
                    return [
                        'success' => false,
                        'error' => 'Stream select failed',
                    ];
                }

                if (in_array($pipes[1], $read)) {
                    $chunk = fread($pipes[1], 8192);
                    if ($chunk !== false) {
                        if (strlen($output) + strlen($chunk) > $maxOutputSize) {
                            proc_terminate($process, 9);

                            return [
                            'success' => false,
                            'error' => sprintf('Subprocess output exceeded maximum size of %d bytes', $maxOutputSize),
                            ];
                        }

                        $output .= $chunk;
                    }
                }

                if (in_array($pipes[2], $read)) {
                    $chunk = fread($pipes[2], 8192);
                    if ($chunk !== false) {
                        if (strlen($errorOutput) + strlen($chunk) > $maxOutputSize) {
                            proc_terminate($process, 9);

                            return [
                                'success' => false,
                                'error' => sprintf(
                                    'Subprocess error output exceeded maximum size of %d bytes',
                                    $maxOutputSize,
                                ),
                            ];
                        }

                        $errorOutput .= $chunk;
                        // Stream subprocess logs to parent's stderr in real-time
                        // Skip in PHPUnit test context to avoid test framework issues
                        if (!defined('PHPUNIT_COMPOSER_INSTALL') && !defined('__PHPUNIT_PHAR__')) {
                            fwrite(STDERR, $chunk);
                        }
                    }
                }

                if (feof($pipes[1]) && feof($pipes[2])) {
                    break;
                }
            }
        } finally {
            // Always cleanup resources
            if (is_resource($pipes[1])) {
                fclose($pipes[1]);
            }

            if (is_resource($pipes[2])) {
                fclose($pipes[2]);
            }
        }

        $exitCode = proc_close($process);

        if ($exitCode !== 0 && empty($output)) {
            return [
                'success' => false,
                'error' => sprintf('Subprocess exited with code %d. Error: %s', $exitCode, $errorOutput),
            ];
        }

        $result = json_decode($output, true);
        if (json_last_error() !== JSON_ERROR_NONE) {
            return [
                'success' => false,
                'error' => 'Invalid JSON output from subprocess: ' . $output,
            ];
        }

        return $result;
    }

    /**
     * Reconstruct exception from array data.
     *
     * @param array<string, mixed> $exceptionData Exception data
     * @return \RuntimeException
     */
    protected function reconstructException(array $exceptionData): RuntimeException
    {
        $message = sprintf(
            '%s: %s in %s:%d',
            $exceptionData['class'] ?? 'Exception',
            $exceptionData['message'] ?? 'Unknown error',
            $exceptionData['file'] ?? 'unknown',
            $exceptionData['line'] ?? 0,
        );

        return new RuntimeException($message, (int)($exceptionData['code'] ?? 0));
    }
}