<?php
/**
 * Queues events and sends deliveries with retries, signing and a circuit breaker.
 *
 * @package Hookbird
 */

namespace Hookbird;

defined( 'ABSPATH' ) || exit;

/**
 * Dispatcher.
 */
class Dispatcher {

	const LOCK_OPTION = 'hookbird_queue_lock';
	const LOCK_TTL    = 300;

	/**
	 * Instance.
	 *
	 * @var Dispatcher|null
	 */
	private static $instance = null;

	/**
	 * Whether deliveries were queued in this request.
	 *
	 * @var bool
	 */
	private $queued = false;

	/**
	 * Singleton.
	 *
	 * @return Dispatcher
	 */
	public static function instance() {
		if ( null === self::$instance ) {
			self::$instance = new self();
		}
		return self::$instance;
	}

	/**
	 * Register hooks.
	 */
	public function init() {
		add_action( Installer::CRON_PROCESS, array( $this, 'process_queue' ) );
		add_action( Installer::CRON_SWEEP, array( $this, 'sweep' ) );
		add_action( Installer::CRON_CLEANUP, array( 'Hookbird\\Deliveries', 'cleanup' ) );
		add_action( 'shutdown', array( $this, 'kick' ), 100 );
		add_action( 'hookbird_dispatch', array( $this, 'dispatch_action' ), 10, 2 );
	}

	/**
	 * Handler for do_action( 'hookbird_dispatch', $event, $data ).
	 *
	 * @param string $event Event key.
	 * @param array  $data  Data.
	 */
	public function dispatch_action( $event, $data = array() ) {
		$this->dispatch( (string) $event, is_array( $data ) ? $data : array( 'value' => $data ) );
	}

	/**
	 * Build the standard event envelope.
	 *
	 * @param string $event Event key.
	 * @param array  $data  Data.
	 * @param bool   $test  Test event.
	 * @return array
	 */
	public static function envelope( $event, array $data, $test = false ) {
		$envelope = array(
			'id'         => 'evt_' . str_replace( '-', '', wp_generate_uuid4() ),
			'event'      => $event,
			'created_at' => gmdate( 'c' ),
			'site'       => array(
				'name' => Listeners::text( get_bloginfo( 'name' ) ),
				'url'  => home_url( '/' ),
			),
			'data'       => $data,
		);
		if ( $test ) {
			$envelope['test'] = true;
		}

		/**
		 * Filter the event envelope before it is queued.
		 *
		 * @param array  $envelope Envelope with id, event, created_at, site and data.
		 * @param string $event    Event key.
		 */
		return (array) apply_filters( 'hookbird_payload', $envelope, $event );
	}

	/**
	 * Queue an event for every subscribed endpoint whose conditions pass.
	 *
	 * @param string $event Event key.
	 * @param array  $data  Data.
	 * @return int Deliveries queued.
	 */
	public function dispatch( $event, array $data ) {
		$event = Endpoints::sanitize_event_key( $event );
		if ( '' === $event || false !== strpos( $event, '*' ) ) {
			return 0;
		}

		$endpoints = Endpoints::subscribed_to( $event );
		if ( empty( $endpoints ) ) {
			return 0;
		}

		/**
		 * Short-circuit an event. Return false to skip it.
		 *
		 * @param bool   $send  Whether to send.
		 * @param string $event Event key.
		 * @param array  $data  Data.
		 */
		if ( ! apply_filters( 'hookbird_should_dispatch', true, $event, $data ) ) {
			return 0;
		}

		$envelope = self::envelope( $event, $data );
		$queued   = 0;

		foreach ( $endpoints as $endpoint ) {
			if ( ! Conditions::passes( $endpoint['conditions'], $envelope ) ) {
				continue;
			}
			if ( Deliveries::create( $endpoint['id'], $envelope ) ) {
				++$queued;
			}
		}

		if ( $queued ) {
			$this->queued = true;
		}

		/**
		 * Fires after an event has been queued.
		 *
		 * @param array $envelope Envelope.
		 * @param int   $queued   Number of deliveries queued.
		 */
		do_action( 'hookbird_dispatched', $envelope, $queued );

		return $queued;
	}

	/**
	 * At the end of a request that queued deliveries, start background processing.
	 */
	public function kick() {
		if ( ! $this->queued ) {
			return;
		}
		$this->queued = false;
		$this->schedule_run( time() );
	}

	/**
	 * Make sure a queue run is scheduled no later than $when and try to start WP-Cron now.
	 *
	 * @param int $when Unix timestamp.
	 */
	public function schedule_run( $when ) {
		$next = wp_next_scheduled( Installer::CRON_PROCESS );
		if ( $next && $next > $when ) {
			wp_unschedule_event( $next, Installer::CRON_PROCESS );
			$next = false;
		}
		if ( ! $next ) {
			wp_schedule_single_event( $when, Installer::CRON_PROCESS );
		}
		if ( $when <= time() && ! wp_doing_cron() && ! ( defined( 'DISABLE_WP_CRON' ) && DISABLE_WP_CRON ) && function_exists( 'spawn_cron' ) ) {
			spawn_cron();
		}
	}

	/**
	 * Safety net every 15 minutes: recover stuck deliveries and process anything due.
	 */
	public function sweep() {
		Deliveries::recover_stuck();
		$next = Deliveries::next_due_time();
		if ( $next && $next <= time() ) {
			$this->process_queue();
		}
	}

	/**
	 * Take the queue lock.
	 *
	 * @return bool
	 */
	private function lock() {
		if ( add_option( self::LOCK_OPTION, time(), '', false ) ) {
			return true;
		}
		$locked_at = (int) get_option( self::LOCK_OPTION );
		if ( $locked_at && $locked_at < time() - self::LOCK_TTL ) {
			delete_option( self::LOCK_OPTION );
			return add_option( self::LOCK_OPTION, time(), '', false );
		}
		return false;
	}

	/**
	 * Whether another request is sending deliveries right now.
	 *
	 * @return bool
	 */
	public static function is_running() {
		$locked_at = (int) get_option( self::LOCK_OPTION );
		return $locked_at > 0 && $locked_at >= time() - self::LOCK_TTL;
	}

	/**
	 * Release the queue lock.
	 */
	private function unlock() {
		delete_option( self::LOCK_OPTION );
	}

	/**
	 * Send due deliveries within a time budget.
	 *
	 * @return int Deliveries attempted.
	 */
	public function process_queue() {
		if ( ! $this->lock() ) {
			return 0;
		}

		/**
		 * Filter how many seconds one queue run may spend sending.
		 *
		 * @param int $seconds Seconds.
		 */
		$budget  = (int) apply_filters( 'hookbird_queue_time_budget', 20 );
		$started = microtime( true );
		$sent    = 0;
		$batch   = max( 1, (int) Settings::get( 'batch_size' ) );
		// Time kept in reserve for one more request, so a run does not overshoot the budget.
		$reserve = min( max( 2, (int) Settings::get( 'timeout' ) ), (int) floor( $budget / 2 ) );

		try {
			do {
				$due = Deliveries::due( $batch );
				foreach ( $due as $delivery ) {
					// Never start a request that could run past the budget (and the PHP time limit).
					if ( $sent && microtime( true ) - $started + $reserve > $budget ) {
						break 2;
					}
					$endpoint = Endpoints::get( $delivery['endpoint_id'] );
					if ( ! $endpoint || ! Endpoints::is_live( $endpoint ) ) {
						continue;
					}
					if ( ! Deliveries::claim( $delivery['id'] ) ) {
						continue;
					}
					$this->send( $delivery, $endpoint );
					++$sent;
					update_option( self::LOCK_OPTION, time(), false );
					if ( microtime( true ) - $started > $budget ) {
						break 2;
					}
				}
				$fetched = count( $due );
			} while ( $fetched === $batch && microtime( true ) - $started <= $budget );
		} finally {
			$this->unlock();
			update_option( 'hookbird_last_queue_run', time(), false );
		}

		$next = Deliveries::next_due_time();
		if ( $next ) {
			$this->schedule_run( max( time(), $next ) );
		}

		return $sent;
	}

	/**
	 * Whether an HTTP status code should be retried.
	 *
	 * @param int $code Status code (0 for network errors).
	 * @return bool
	 */
	public static function is_retryable( $code ) {
		$code = (int) $code;
		return 0 === $code || 408 === $code || 409 === $code || 425 === $code || 429 === $code || $code >= 500;
	}

	/**
	 * Send one delivery and record the result.
	 *
	 * @param array $delivery Delivery.
	 * @param array $endpoint Endpoint.
	 * @param bool  $retry    Whether failures may be retried (false for tests).
	 * @return array Result with success, code, error, duration_ms, body.
	 */
	public function send( array $delivery, array $endpoint, $retry = true ) {
		$envelope = $delivery['payload'];
		$body     = (string) wp_json_encode( Formatter::body( $endpoint, $envelope ), JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION );
		$headers  = array_merge(
			$endpoint['headers'],
			array(
				'Content-Type'     => 'application/json; charset=utf-8',
				'User-Agent'       => 'Hookbird/' . HOOKBIRD_VERSION . ' (+' . home_url( '/' ) . ')',
				'X-Hookbird-Event' => $delivery['event'],
			),
			Signer::headers( $endpoint['secret'], $delivery['uuid'], $body )
		);

		$args = array(
			'method'      => 'POST',
			'timeout'     => max( 2, (int) Settings::get( 'timeout' ) ),
			'redirection' => 0,
			'headers'     => $headers,
			'body'        => $body,
			'data_format' => 'body',
		);

		/**
		 * Filter the HTTP request arguments.
		 *
		 * @param array $args     wp_remote_post() arguments.
		 * @param array $delivery Delivery.
		 * @param array $endpoint Endpoint.
		 */
		$args = (array) apply_filters( 'hookbird_request_args', $args, $delivery, $endpoint );

		$start    = microtime( true );
		$response = Settings::allow_local_urls() ? wp_remote_post( $endpoint['url'], $args ) : wp_safe_remote_post( $endpoint['url'], $args );
		$duration = (int) round( ( microtime( true ) - $start ) * 1000 );

		$attempts = $delivery['attempts'] + 1;
		$code     = 0;
		$error    = '';
		$resp     = '';

		if ( is_wp_error( $response ) ) {
			$error = $response->get_error_message();
		} else {
			$code = (int) wp_remote_retrieve_response_code( $response );
			$resp = self::clean_response( wp_remote_retrieve_body( $response ) );
			if ( $code >= 300 && $code < 400 ) {
				$location = (string) wp_remote_retrieve_header( $response, 'location' );
				/* translators: 1: HTTP status code, 2: redirect target. */
				$error = sprintf( __( 'Endpoint redirected (HTTP %1$d) to %2$s. Use the final URL; redirects are not followed so the signed request is never sent somewhere else.', 'hookbird' ), $code, $location ? Endpoints::mask_url( $location ) : __( 'another address', 'hookbird' ) );
			} elseif ( $code < 200 || $code >= 300 ) {
				/* translators: %d: HTTP status code. */
				$error = sprintf( __( 'Endpoint responded with HTTP %d.', 'hookbird' ), $code );
			}
		}

		$success = '' === $error;
		$fields  = array(
			'attempts'      => $attempts,
			'response_code' => $code ? $code : null,
			'response_body' => $resp,
			'error'         => $error,
			'duration_ms'   => $duration,
		);

		if ( $success ) {
			$fields['status']          = 'success';
			$fields['next_attempt_at'] = null;
		} elseif ( $retry && self::is_retryable( $code ) && $attempts < (int) Settings::get( 'max_attempts' ) ) {
			$fields['status']          = 'pending';
			$fields['next_attempt_at'] = Deliveries::now( self::retry_delay( $attempts, $response ) );
		} else {
			$fields['status']          = 'failed';
			$fields['next_attempt_at'] = null;
		}

		Deliveries::update( $delivery['id'], $fields );

		if ( ! $delivery['is_test'] ) {
			$this->record_endpoint_result( $endpoint, $success, $error );
		}

		$result = array(
			'success'     => $success,
			'status'      => $fields['status'],
			'code'        => $code,
			'error'       => $error,
			'duration_ms' => $duration,
			'body'        => $resp,
			'attempts'    => $attempts,
		);

		/**
		 * Fires after a delivery attempt.
		 *
		 * @param array $result   Result.
		 * @param array $delivery Delivery (before the attempt).
		 * @param array $endpoint Endpoint.
		 */
		do_action( 'hookbird_delivery_attempted', $result, $delivery, $endpoint );

		return $result;
	}

	/**
	 * Seconds until the next attempt, honouring Retry-After (header or JSON body) on 429 and 503 responses.
	 *
	 * @param int             $attempts Attempts made so far.
	 * @param array|\WP_Error $response Response.
	 * @return int
	 */
	public static function retry_delay( $attempts, $response = null ) {
		$schedule = Settings::backoff_schedule();
		$delay    = $schedule ? $schedule[ min( count( $schedule ) - 1, max( 0, $attempts - 1 ) ) ] : 60;

		if ( $response && ! is_wp_error( $response ) ) {
			$retry_after = wp_remote_retrieve_header( $response, 'retry-after' );
			if ( is_array( $retry_after ) ) {
				$retry_after = reset( $retry_after );
			}
			if ( ! $retry_after ) {
				// Telegram and Discord also say how long to wait in the JSON body.
				$json = json_decode( (string) wp_remote_retrieve_body( $response ), true );
				if ( isset( $json['parameters']['retry_after'] ) ) {
					$retry_after = $json['parameters']['retry_after'];
				} elseif ( isset( $json['retry_after'] ) ) {
					$retry_after = $json['retry_after'];
				}
			}
			if ( is_numeric( $retry_after ) ) {
				$delay = max( $delay, (int) ceil( (float) $retry_after ) );
			} elseif ( $retry_after && strtotime( (string) $retry_after ) ) {
				$delay = max( $delay, strtotime( (string) $retry_after ) - time() );
			}
		}

		// Up to 10% jitter so many failed deliveries do not retry at the same second.
		$delay += wp_rand( 0, (int) floor( $delay / 10 ) );
		return min( DAY_IN_SECONDS, max( 1, $delay ) );
	}

	/**
	 * Keep response bodies small and valid UTF-8 for storage.
	 *
	 * @param string $body Body.
	 * @return string
	 */
	private static function clean_response( $body ) {
		$body = wp_check_invalid_utf8( (string) $body, true );
		return mb_substr( $body, 0, 2000 );
	}

	/**
	 * Update endpoint health and trip the circuit breaker after too many consecutive failures.
	 *
	 * @param array  $endpoint Endpoint.
	 * @param bool   $success  Success.
	 * @param string $error    Error message.
	 */
	private function record_endpoint_result( array $endpoint, $success, $error ) {
		if ( $success ) {
			Endpoints::update_state(
				$endpoint['id'],
				array(
					'consecutive_failures' => 0,
					'last_success_at'      => Deliveries::now(),
				)
			);
			return;
		}

		$failures  = $endpoint['consecutive_failures'] + 1;
		$threshold = (int) Settings::get( 'circuit_threshold' );
		$fields    = array(
			'consecutive_failures' => $failures,
			'last_failure_at'      => Deliveries::now(),
		);
		$trip      = $threshold > 0 && $failures >= $threshold && null === $endpoint['paused_at'];
		if ( $trip ) {
			$fields['paused_at'] = Deliveries::now();
		}
		Endpoints::update_state( $endpoint['id'], $fields );

		if ( $trip ) {
			$this->notify_paused( $endpoint, $failures, $error );

			/**
			 * Fires when the circuit breaker pauses an endpoint.
			 *
			 * @param array  $endpoint Endpoint.
			 * @param int    $failures Consecutive failures.
			 * @param string $error    Last error.
			 */
			do_action( 'hookbird_endpoint_paused', $endpoint, $failures, $error );
		}
	}

	/**
	 * Email the site admin when an endpoint is paused.
	 *
	 * @param array  $endpoint Endpoint.
	 * @param int    $failures Failures.
	 * @param string $error    Last error.
	 */
	private function notify_paused( array $endpoint, $failures, $error ) {
		/**
		 * Filter whether to email an alert when an endpoint is paused.
		 *
		 * @param bool  $send     Send email.
		 * @param array $endpoint Endpoint.
		 */
		if ( ! apply_filters( 'hookbird_send_pause_email', true, $endpoint ) ) {
			return;
		}

		$site = Listeners::text( get_bloginfo( 'name' ) );
		/* translators: 1: site name, 2: endpoint name. */
		$subject = sprintf( __( '[%1$s] Hookbird paused the endpoint "%2$s"', 'hookbird' ), $site, $endpoint['name'] );
		$message = sprintf(
			/* translators: 1: endpoint name, 2: number of failures, 3: last error, 4: admin URL. */
			__( "Hookbird paused the endpoint \"%1\$s\" after %2\$d failed delivery attempts in a row.\n\nLast error: %3\$s\n\nNew events are still queued and will be delivered once you resume the endpoint. Check the endpoint and resume it here:\n%4\$s", 'hookbird' ),
			$endpoint['name'],
			$failures,
			$error ? $error : __( 'unknown', 'hookbird' ),
			admin_url( 'admin.php?page=hookbird-edit&id=' . $endpoint['id'] )
		);

		wp_mail( Settings::alert_email(), $subject, $message );
	}

	/**
	 * Send a sample event to an endpoint right away (ignores active state, conditions and the circuit breaker).
	 *
	 * @param int    $endpoint_id Endpoint ID.
	 * @param string $event       Event key for the sample.
	 * @return array|\WP_Error Result.
	 */
	public function send_test( $endpoint_id, $event = '' ) {
		$endpoint = Endpoints::get( $endpoint_id );
		if ( ! $endpoint ) {
			return new \WP_Error( 'hookbird_not_found', __( 'Endpoint not found.', 'hookbird' ) );
		}
		$event = Endpoints::sanitize_event_key( $event );
		if ( '' === $event || false !== strpos( $event, '*' ) ) {
			$event = 'user.registered';
			foreach ( $endpoint['events'] as $subscribed ) {
				if ( false === strpos( $subscribed, '*' ) ) {
					$event = $subscribed;
					break;
				}
			}
		}

		$envelope    = self::envelope( $event, Registry::sample( $event ), true );
		$delivery_id = Deliveries::create( $endpoint['id'], $envelope, true );
		if ( ! $delivery_id || ! Deliveries::claim( $delivery_id ) ) {
			return new \WP_Error( 'hookbird_db', __( 'Could not create the test delivery.', 'hookbird' ) );
		}
		$result                = $this->send( Deliveries::get( $delivery_id ), $endpoint, false );
		$result['delivery_id'] = $delivery_id;
		$result['event']       = $event;
		return $result;
	}

	/**
	 * Resend a delivery as a new delivery, immediately.
	 *
	 * @param int $delivery_id Delivery ID.
	 * @return array|\WP_Error Result.
	 */
	public function resend( $delivery_id ) {
		$original = Deliveries::get( $delivery_id );
		if ( ! $original ) {
			return new \WP_Error( 'hookbird_not_found', __( 'Delivery not found.', 'hookbird' ) );
		}
		$endpoint = Endpoints::get( $original['endpoint_id'] );
		if ( ! $endpoint ) {
			return new \WP_Error( 'hookbird_not_found', __( 'The endpoint for this delivery no longer exists.', 'hookbird' ) );
		}

		if ( 'pending' === $original['status'] && Deliveries::claim( $original['id'] ) ) {
			// Still queued: just send it now instead of creating a copy.
			$result                = $this->send( Deliveries::get( $original['id'] ), $endpoint );
			$result['delivery_id'] = $original['id'];
			return $result;
		}

		$new_id = Deliveries::create( $endpoint['id'], $original['payload'], $original['is_test'] );
		if ( ! $new_id || ! Deliveries::claim( $new_id ) ) {
			return new \WP_Error( 'hookbird_db', __( 'Could not create the delivery.', 'hookbird' ) );
		}
		$result                = $this->send( Deliveries::get( $new_id ), $endpoint );
		$result['delivery_id'] = $new_id;
		return $result;
	}
}