<?php
/**
 * Outgoing event queue.
 *
 * @package Rahkar\WP
 */

declare( strict_types = 1 );

namespace Rahkar\WP\Modules\HubConnector;

use Rahkar\WP\Database\Schema;
use Rahkar\WP\Queue\Queue;
use Rahkar\WP\Support\Hooks;

defined( 'ABSPATH' ) || exit;

/**
 * Events waiting to reach the hub.
 *
 * The row id is the contract's `seq`, so it must only ever grow. Events of one
 * request are buffered and written with a single multi-row INSERT at shutdown;
 * the send itself is a queue job a minute later, so a minute's events travel in
 * one batch and nothing is sent on a path a visitor waits on (EVENTS-v1.md §3-7).
 */
final class Outbox {

	public const PENDING = 0;
	public const DEAD    = 3;

	public const ORIGIN_LIVE     = 0;
	public const ORIGIN_BACKFILL = 1;
	public const ORIGIN_RESYNC   = 2;

	/**
	 * Seconds between the first event of a quiet period and the send.
	 */
	public const FLUSH_DELAY = 60;

	/**
	 * Events of this request not yet written.
	 *
	 * @var list<array{type: string, data: array<string, mixed>, origin: int, occurred_at: int}>
	 */
	private static array $buffer = [];

	/**
	 * Whether the shutdown writer is hooked for this request.
	 *
	 * @var bool
	 */
	private static bool $hooked = false;

	/**
	 * Queue an event for this request; written at shutdown.
	 *
	 * @param string               $type       Contract event type.
	 * @param array<string, mixed> $data       Event body, already in contract shape.
	 * @param int                  $origin     One of the ORIGIN_ constants.
	 * @param int|null             $occurredAt Unix time it happened; now when null.
	 */
	public static function add( string $type, array $data, int $origin = self::ORIGIN_LIVE, ?int $occurredAt = null ): void {
		self::$buffer[] = [
			'type'        => $type,
			'data'        => $data,
			'origin'      => $origin,
			'occurred_at' => $occurredAt ?? time(),
		];

		if ( ! self::$hooked ) {
			self::$hooked = true;
			// Priority 20: after the queue's own shutdown drain has closed the
			// response, so the visitor is not kept waiting on this insert.
			add_action( 'shutdown', [ self::class, 'write' ], 20 );
		}
	}

	/**
	 * Write the buffer and make sure a send is scheduled.
	 *
	 * @internal Hooked to `shutdown`; also called directly by background jobs.
	 */
	public static function write(): void {
		if ( [] === self::$buffer ) {
			return;
		}

		global $wpdb;

		$rows         = self::$buffer;
		self::$buffer = [];
		$now          = gmdate( 'Y-m-d H:i:s' );
		$tuples       = [];

		foreach ( $rows as $row ) {
			$payload = wp_json_encode( $row['data'], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES );

			if ( is_string( $payload ) ) {
				$tuples[] = [ Uuid7::generate(), $row['type'], gmdate( 'Y-m-d H:i:s', $row['occurred_at'] ), $row['origin'], $payload, self::PENDING, 0, $now ];
			}
		}

		if ( [] === $tuples ) {
			return;
		}

		$params = array_merge( [ Schema::table( 'hub_outbox' ) ], ...$tuples );

		// One multi-row INSERT for the whole request. The value list repeats a fixed
		// eight-placeholder tuple once per row, so the replacement count is 1 + 8n.
		// phpcs:disable WordPress.DB.DirectDatabaseQuery, WordPress.DB.PreparedSQLPlaceholders.ReplacementsWrongNumber -- See above: the count depends on the number of rows.
		$wpdb->query(
			$wpdb->prepare(
				'INSERT INTO %i (event_id, type, occurred_at, origin, payload, status, attempts, created_at) VALUES '
				. implode( ', ', array_fill( 0, count( $tuples ), '(%s, %s, %s, %d, %s, %d, %d, %s)' ) ),
				$params
			)
		);
		// phpcs:enable WordPress.DB.DirectDatabaseQuery, WordPress.DB.PreparedSQLPlaceholders.ReplacementsWrongNumber

		Queue::pushUnique( Hooks::HUB_FLUSH, [], 'hub:flush', max( self::FLUSH_DELAY, ( new State() )->minInterval() ) );
	}

	/**
	 * The oldest pending rows.
	 *
	 * @param int $limit Maximum rows.
	 *
	 * @return list<array<string, mixed>>
	 */
	public static function pending( int $limit ): array {
		global $wpdb;

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery
		$rows = $wpdb->get_results(
			$wpdb->prepare( 'SELECT * FROM %i WHERE status = %d ORDER BY id ASC LIMIT %d', Schema::table( 'hub_outbox' ), self::PENDING, max( 1, $limit ) ),
			ARRAY_A
		);

		return is_array( $rows ) ? $rows : [];
	}

	/**
	 * Remove delivered rows.
	 *
	 * @param list<int> $ids Row ids.
	 */
	public static function delete( array $ids ): void {
		if ( [] === $ids ) {
			return;
		}

		global $wpdb;

		$params = array_merge( [ Schema::table( 'hub_outbox' ) ], $ids );

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery
		$wpdb->query( $wpdb->prepare( 'DELETE FROM %i WHERE id IN (' . implode( ',', array_fill( 0, count( $ids ), '%d' ) ) . ')', $params ) );
	}

	/**
	 * Park rows the hub will never accept, so they stop blocking the queue.
	 *
	 * @param list<int> $ids   Row ids.
	 * @param string    $error Why.
	 */
	public static function bury( array $ids, string $error ): void {
		if ( [] === $ids ) {
			return;
		}

		global $wpdb;

		$params = array_merge( [ Schema::table( 'hub_outbox' ), self::DEAD, mb_substr( $error, 0, 250 ) ], $ids );

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery
		$wpdb->query( $wpdb->prepare( 'UPDATE %i SET status = %d, last_error = %s WHERE id IN (' . implode( ',', array_fill( 0, count( $ids ), '%d' ) ) . ')', $params ) );
	}

	/**
	 * Counts for the settings screen and the heartbeat.
	 *
	 * @return array{pending: int, dead: int}
	 */
	public static function counts(): array {
		global $wpdb;

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery
		$row = $wpdb->get_row(
			$wpdb->prepare( 'SELECT SUM(status = %d) AS pending, SUM(status = %d) AS dead FROM %i', self::PENDING, self::DEAD, Schema::table( 'hub_outbox' ) ),
			ARRAY_A
		);

		return [
			'pending' => (int) ( $row['pending'] ?? 0 ),
			'dead'    => (int) ( $row['dead'] ?? 0 ),
		];
	}

	/**
	 * Empty the queue. Disconnecting must leave nothing behind to be sent later.
	 */
	public static function clear(): void {
		global $wpdb;

		self::$buffer = [];

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery
		$wpdb->query( $wpdb->prepare( 'DELETE FROM %i', Schema::table( 'hub_outbox' ) ) );
	}

	/**
	 * Dead rows older than 30 days are only noise on the health screen.
	 */
	public static function purgeDead(): void {
		global $wpdb;

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery
		$wpdb->query(
			$wpdb->prepare( 'DELETE FROM %i WHERE status = %d AND created_at < %s LIMIT 500', Schema::table( 'hub_outbox' ), self::DEAD, gmdate( 'Y-m-d H:i:s', time() - 30 * DAY_IN_SECONDS ) )
		);
	}
}