<?php
/**
 * Background CSV export.
 *
 * @package Recast\WCS
 */

namespace Recast\WCS\Export;

use Recast\WCS\Cohort\CohortSpec;
use Recast\WCS\Cohort\InvalidSpecException;
use Recast\WCS\Pipeline\Scheduler;
use Recast\WCS\Support\Logger;

defined( 'ABSPATH' ) || exit;

/**
 * Writes an export file across chained Action Scheduler actions with a persisted cursor.
 */
final class BackgroundExporter {

	/**
	 * Registers the handler.
	 *
	 * @return void
	 */
	public static function register() {
		add_action( Scheduler::HOOK_EXPORT_CHUNK, array( __CLASS__, 'handle' ), 10, 2 );
		add_action( 'recast_watchdog', array( __CLASS__, 'recover' ) );
	}

	/**
	 * Restarts exports whose background job stopped without finishing (for example a PHP
	 * worker that was killed). An export with no queued or running job is started again
	 * from the first row in a new file. A second stop marks it failed.
	 *
	 * @return void
	 */
	public static function recover() {
		if ( ! function_exists( 'as_get_scheduled_actions' ) ) {
			return;
		}
		$repo = new ExportRepository();
		$live = array();
		foreach ( array( 'pending', 'in-progress' ) as $status ) {
			$actions = as_get_scheduled_actions(
				array(
					'hook'     => Scheduler::HOOK_EXPORT_CHUNK,
					'status'   => $status,
					'group'    => Scheduler::GROUP,
					'per_page' => 100,
				)
			);
			foreach ( $actions as $action ) {
				$args = $action->get_args();
				if ( isset( $args[0] ) ) {
					$live[ (int) $args[0] ] = true;
				}
			}
		}
		foreach ( $repo->unfinished() as $export ) {
			if ( isset( $live[ $export['id'] ] ) ) {
				continue;
			}
			if ( $export['file_name'] ) {
				Storage::delete( $export['file_name'] );
			}
			if ( ! empty( $export['error'] ) ) {
				$repo->update(
					$export['id'],
					array(
						'status' => ExportRepository::STATUS_FAILED,
						'error'  => __( 'The background job stopped twice. Start the export again.', 'recast-subscriptions' ),
					)
				);
				continue;
			}
			$repo->update(
				$export['id'],
				array(
					'status'        => ExportRepository::STATUS_PENDING,
					'file_name'     => null,
					'rows_written'  => 0,
					'resume_cursor' => null,
					'error'         => __( 'The background job stopped. Recast started the export again.', 'recast-subscriptions' ),
				)
			);
			self::start( $export['id'] );
		}
	}

	/**
	 * Starts an export.
	 *
	 * @param int $export_id Export id.
	 * @return void
	 */
	public static function start( $export_id ) {
		Scheduler::enqueue( Scheduler::HOOK_EXPORT_CHUNK, array( (int) $export_id, 0 ) );
	}

	/**
	 * Processes one chunk. Hooked to recast/export_chunk [export_id, chunk].
	 *
	 * @param int $export_id Export.
	 * @param int $chunk     Chunk number.
	 * @return void
	 */
	public static function handle( $export_id, $chunk = 0 ) {
		$repo   = new ExportRepository();
		$export = $repo->get( (int) $export_id );
		if ( ! $export || ! in_array( $export['status'], array( ExportRepository::STATUS_PENDING, ExportRepository::STATUS_RUNNING ), true ) ) {
			return;
		}
		try {
			$spec = CohortSpec::from_array( $export['spec'] );
		} catch ( InvalidSpecException $e ) {
			$repo->update(
				$export['id'],
				array(
					'status' => ExportRepository::STATUS_FAILED,
					'error'  => $e->getMessage(),
				)
			);
			return;
		}

		$file_name = $export['file_name'] ? $export['file_name'] : Storage::new_file_name( $export['id'] );
		$first     = empty( $export['file_name'] );
		if ( $first ) {
			$repo->update(
				$export['id'],
				array(
					'file_name' => $file_name,
					'status'    => ExportRepository::STATUS_RUNNING,
				)
			);
		}
		$path   = Storage::path( $file_name );
		$handle = fopen( $path, $first ? 'w' : 'a' ); // phpcs:ignore WordPress.WP.AlternativeFunctions
		if ( ! $handle ) {
			$repo->update(
				$export['id'],
				array(
					'status' => ExportRepository::STATUS_FAILED,
					'error'  => 'Could not open the export file for writing.',
				)
			);
			return;
		}
		$cursor = $export['cursor'];
		// The job runs on a request nobody waits for; keep going if that connection closes.
		ignore_user_abort( true );
		try {
			$result = StreamExporter::write( $handle, $spec, $first, $cursor, Scheduler::time_budget() );
		} catch ( \Throwable $e ) {
			fclose( $handle ); // phpcs:ignore WordPress.WP.AlternativeFunctions
			Logger::error(
				'Export failed',
				array(
					'export' => $export['id'],
					'error'  => $e->getMessage(),
				)
			);
			$repo->update(
				$export['id'],
				array(
					'status' => ExportRepository::STATUS_FAILED,
					'error'  => $e->getMessage(),
				)
			);
			return;
		}
		fclose( $handle ); // phpcs:ignore WordPress.WP.AlternativeFunctions

		$rows = (int) $export['rows_written'] + (int) $result['rows'];
		if ( $result['done'] ) {
			$repo->update(
				$export['id'],
				array(
					'status'        => ExportRepository::STATUS_READY,
					'rows_written'  => $rows,
					'file_size'     => (int) filesize( $path ),
					'finished_at'   => current_time( 'mysql', true ),
					'resume_cursor' => array(),
				)
			);
			return;
		}
		$repo->update(
			$export['id'],
			array(
				'rows_written'  => $rows,
				'resume_cursor' => $cursor,
			)
		);
		Scheduler::enqueue( Scheduler::HOOK_EXPORT_CHUNK, array( (int) $export_id, (int) $chunk + 1 ) );
	}
}