<?php
/**
 * Sends the outbox to the hub.
 *
 * @package Rahkar\WP
 */

declare( strict_types = 1 );

namespace Rahkar\WP\Modules\HubConnector;

use Rahkar\WP\Queue\Queue;
use Rahkar\WP\Support\Hooks;
use Rahkar\WP\Support\Logger;

defined( 'ABSPATH' ) || exit;

/**
 * The flush job: batches from the outbox, signed, and what to do with each answer.
 *
 * Every status code of EVENTS-v1.md §3-5 has a rule here. The ones that cannot
 * fix themselves (a revoked key, a copied site, an old contract) stop sending and
 * say so on the settings screen; the rest retry on their own.
 */
final class Flusher {

	/**
	 * Batches per run.
	 */
	private const MAX_BATCHES = 5;

	/**
	 * Wall-clock budget per run, in seconds.
	 */
	private const BUDGET = 20.0;

	/**
	 * Longest back-off after repeated failures, in seconds.
	 */
	private const MAX_BACKOFF = 3600;

	/**
	 * Hub client.
	 *
	 * @var HubClient
	 */
	private HubClient $client;

	/**
	 * Build.
	 *
	 * @param HubClient|null $client Client; injectable for tests.
	 */
	public function __construct( ?HubClient $client = null ) {
		$this->client = $client ?? new HubClient();
	}

	/**
	 * Run one flush.
	 *
	 * @internal Hooked to {@see Hooks::HUB_FLUSH}.
	 */
	public function run(): void {
		$credentials = Credentials::load();

		if ( null === $credentials ) {
			return;
		}

		$state = new State();

		if ( ! $credentials->matchesSite() ) {
			$state->update(
				[
					'status'     => State::SITE_MISMATCH,
					'last_error' => __( 'آدرس سایت با زمان اتصال فرق دارد؛ احتمالاً این یک کپی (staging) است. برای ارسال از این سایت، دوباره وصل شوید.', 'rahkar-toolkit' ),
				]
			);

			return;
		}

		if ( in_array( $state->status(), [ State::AUTH_FAILED, State::SITE_MISMATCH, State::UPGRADE_REQUIRED ], true ) ) {
			return;
		}

		if ( $state->waitUntil() > time() ) {
			$this->again( $state->waitUntil() - time() );

			return;
		}

		$started   = microtime( true );
		$batchSize = (int) $state->get( 'batch_size', (int) $state->get( 'max_events', 500 ) );

		for ( $batches = 0; $batches < self::MAX_BATCHES; $batches++ ) {
			if ( microtime( true ) - $started >= self::BUDGET ) {
				break;
			}

			$rows = Outbox::pending( max( 1, $batchSize ) );

			if ( [] === $rows ) {
				$state->update( [ 'batch_size' => (int) $state->get( 'max_events', 500 ) ] );
				Outbox::purgeDead();

				return;
			}

			$outcome = $this->sendBatch( $rows, $credentials, $state );

			if ( 'stop' === $outcome ) {
				return;
			}

			if ( 'shrink' === $outcome ) {
				if ( 1 === count( $rows ) ) {
					Outbox::bury( [ (int) $rows[0]['id'] ], (string) $state->get( 'last_error', '' ) );
				} else {
					$batchSize = max( 1, intdiv( count( $rows ), 2 ) );
					$state->update( [ 'batch_size' => $batchSize ] );
				}
			}
		}

		// More may be waiting: come back soon rather than after the usual delay.
		$this->again( 5 );
	}

	/**
	 * Send one batch and apply the hub's answer.
	 *
	 * @param list<array<string, mixed>> $rows        Outbox rows.
	 * @param Credentials                $credentials Store id and key.
	 * @param State                      $state       Connector state.
	 *
	 * @return string `sent`, `shrink` (send fewer next time), or `stop`.
	 */
	private function sendBatch( array $rows, Credentials $credentials, State $state ): string {
		$json     = $this->encode( $rows );
		$response = $this->client->signed( 'POST', 'events', $json, $credentials, $state->clockOffset() );

		if ( 401 === $response->status && 'clock_skew' === $response->code() && null !== $response->hubTime() ) {
			// A wrong server clock is common on cheap hosting and must not stop the
			// sending: learn the difference and sign with it from now on.
			$state->update( [ 'clock_offset' => $response->hubTime() - time() ] );
			$response = $this->client->signed( 'POST', 'events', $json, $credentials, $state->clockOffset() );
		}

		$ids = array_map( static fn( array $row ): int => (int) $row['id'], $rows );

		switch ( true ) {
			case 200 === $response->status:
				Outbox::delete( $ids );
				$this->logQuarantine( $response );
				Directives::apply( $response->body['directives'] ?? [] );
				$state->update(
					[
						'status'      => State::OK,
						'failures'    => 0,
						'wait_until'  => 0,
						'last_sent'   => time(),
						'last_error'  => '',
						'last_result' => count( $ids ),
					]
				);

				return 'sent';

			case 400 === $response->status:
			case 413 === $response->status:
				$state->update( [ 'last_error' => $response->message() ] );

				return 'shrink';

			case 401 === $response->status:
				return $this->halt( $state, State::AUTH_FAILED, $response->message() );

			case 403 === $response->status:
				$state->update(
					[
						'status'     => State::PAUSED,
						'wait_until' => time() + 6 * HOUR_IN_SECONDS,
						'last_error' => $response->message(),
					]
				);
				$this->again( 6 * HOUR_IN_SECONDS );

				return 'stop';

			case 409 === $response->status:
				return $this->halt( $state, State::SITE_MISMATCH, $response->message() );

			case 410 === $response->status:
				// The store is gone from the hub: stop for good and forget the key.
				Connector::forgetLocally();
				Logger::warning( 'hub', 'The hub removed this store; the connection was cleared.', [] );

				return 'stop';

			case 426 === $response->status:
				return $this->halt( $state, State::UPGRADE_REQUIRED, $response->message() );

			case 429 === $response->status:
				$wait = max( 30, $response->retryAfter );
				$state->update( [ 'wait_until' => time() + $wait ] );
				$this->again( $wait );

				return 'stop';

			default:
				$failures = (int) $state->get( 'failures', 0 ) + 1;
				$wait     = min( self::MAX_BACKOFF, 60 * ( 2 ** min( 6, $failures - 1 ) ) ) + wp_rand( 0, 30 );

				$state->update(
					[
						'failures'   => $failures,
						'wait_until' => time() + $wait,
						'last_error' => $response->message(),
					]
				);
				Logger::warning( 'hub', 'Sending to the hub failed; will retry.', [ 'error' => $response->message(), 'retry_in' => $wait ] );
				$this->again( $wait );

				return 'stop';
		}
	}

	/**
	 * Build the batch body in contract shape.
	 *
	 * @param list<array<string, mixed>> $rows Outbox rows.
	 */
	private function encode( array $rows ): string {
		$origins = [
			Outbox::ORIGIN_LIVE     => 'live',
			Outbox::ORIGIN_BACKFILL => 'backfill',
			Outbox::ORIGIN_RESYNC   => 'resync',
		];
		$events  = [];

		foreach ( $rows as $row ) {
			$data = json_decode( (string) $row['payload'], true );

			$events[] = [
				'id'          => (string) $row['event_id'],
				'type'        => (string) $row['type'],
				'occurred_at' => (string) Contract::time( (int) strtotime( (string) $row['occurred_at'] . ' UTC' ) ),
				'seq'         => (int) $row['id'],
				'origin'      => $origins[ (int) $row['origin'] ] ?? 'live',
				// An empty PHP array encodes as [], but the contract wants an object.
				'data'        => is_array( $data ) && [] !== $data ? $data : new \stdClass(),
			];
		}

		return (string) wp_json_encode(
			[
				'batch_id' => wp_generate_uuid4(),
				'contract' => Contract::VERSION,
				'sent_at'  => (string) Contract::time( time() ),
				'source'   => [
					'site_url'         => home_url(),
					'platform'         => 'woocommerce',
					'platform_version' => defined( 'WC_VERSION' ) ? (string) WC_VERSION : 'unknown',
					'plugin_version'   => RAHKAR_WP_VERSION,
					'pro_version'      => defined( 'RAHKAR_WP_PRO_VERSION' ) ? (string) RAHKAR_WP_PRO_VERSION : null,
				],
				'events'   => $events,
			],
			JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES
		);
	}

	/**
	 * Stop sending until a person acts.
	 *
	 * @param State  $state  Connector state.
	 * @param string $status Stop reason.
	 * @param string $error  Message.
	 */
	private function halt( State $state, string $status, string $error ): string {
		$state->update(
			[
				'status'     => $status,
				'last_error' => $error,
			]
		);
		Logger::error( 'hub', 'Sending to the hub stopped.', [ 'status' => $status, 'error' => $error ] );

		return 'stop';
	}

	/**
	 * Record events the hub quarantined, for the health screen.
	 *
	 * @param HubResponse $response Answer.
	 */
	private function logQuarantine( HubResponse $response ): void {
		$quarantined = is_array( $response->body['quarantined'] ?? null ) ? $response->body['quarantined'] : [];

		if ( [] !== $quarantined ) {
			Logger::warning( 'hub', 'The hub quarantined events that did not match the contract.', [ 'events' => array_slice( $quarantined, 0, 5 ) ] );
		}
	}

	/**
	 * Schedule the next run.
	 *
	 * `push()`, not `pushUnique()`: this job is reserved while it runs, and
	 * `pushUnique()` would count it as the duplicate and drop the follow-up.
	 *
	 * @param int $delay Seconds.
	 */
	private function again( int $delay ): void {
		Queue::push( Hooks::HUB_FLUSH, [], max( 1, $delay ), 20, 'hub:flush' );
	}
}