<?php
defined( 'ABSPATH' ) || exit;

/**
 * Durable retry queue for server-side events.
 *
 * Design goals, in order:
 *   1. A purchase is never lost because the API had a bad thirty seconds.
 *   2. Checkout is never blocked on a network call. The customer sees their
 *      order confirmation whether or not mcp.advisorppc.com answered.
 *   3. A broken endpoint can never fill the customer's database.
 *
 * Storage is a custom table because options are a single serialised blob and
 * two concurrent checkouts would clobber each other's writes.
 *
 * @package AdvisorPPC
 */
class AdvisorPPC_Queue {

	/**
	 * Schema version. Bump to trigger a dbDelta upgrade.
	 */
	const DB_VERSION = 2;

	/**
	 * Own option row, not part of lane A's settings array, and not autoloaded.
	 */
	const OPTION_DB_VERSION = 'advisorppc_queue_db_version';

	/**
	 * Cron hook and custom schedule.
	 */
	// Name matches the hook lane A already unschedules in advisorppc_deactivate()
	// and uninstall.php. Do not rename without changing both of those.
	const CRON_HOOK     = 'advisorppc_flush_queue';
	const CRON_SCHEDULE = 'advisorppc_five_minutes';

	/**
	 * Bounded retry policy for transport failures and `retry: true`.
	 *
	 * MAX_ATTEMPTS bounds non-critical events only. A purchase or refund is
	 * never deleted for running out of attempts: it keeps retrying at
	 * BACKOFF_CAP until MAX_AGE, because "the API had a bad hour" is exactly
	 * the failure a durable queue exists to survive, and a sale deleted after
	 * six tries in half an hour is a sale nobody can get back.
	 */
	const MAX_ATTEMPTS   = 6;
	const BACKOFF_BASE   = 60;
	const BACKOFF_CAP    = 3600;

	/**
	 * Backoff for a plan gate, which is not a failure and spends no attempt.
	 *
	 * A gate lasts until someone upgrades, which is hours or days, so a gated
	 * row is re-sent on a schedule that grows with its age: never sooner than
	 * GATE_BACKOFF_MIN, never later than GATE_BACKOFF_CAP, a quarter of the
	 * row's age in between (see gate_delay()). Over MAX_AGE that is a few
	 * dozen signed POSTs per gated event instead of one every five minutes.
	 */
	const GATE_BACKOFF_MIN = 900;   // 15 minutes.
	const GATE_BACKOFF_CAP = 21600; // 6 hours.

	/**
	 * Hard caps. A broken endpoint stops at MAX_ROWS and rows older than
	 * MAX_AGE are dropped whatever their attempt count.
	 */
	const MAX_ROWS = 5000;
	const MAX_AGE  = 604800; // 7 days.

	/**
	 * Rows sent per drain run, and the HTTP timeout in seconds.
	 *
	 * The ingest fans each event out to its destinations BEFORE it answers,
	 * because the response's `plan_gated` and `retry` flags are derived from
	 * those outcomes. Seven sequential platform calls take longer than the
	 * five seconds this used to allow, and a request the plugin abandons is a
	 * request the server is still working on: the retry then meets a held
	 * claim. The timeout therefore has to cover a full fan-out. It runs on
	 * cron, never in a customer's request, so waiting is cheap. Filterable via
	 * `advisorppc_queue_timeout` between 5 and 60 seconds.
	 *
	 * DRAIN_BUDGET caps the wall time one drain() spends on POSTs, so a slow
	 * ingest cannot push a cron run past the host's execution limit; whatever
	 * is left waits for the next run.
	 */
	const BATCH_SIZE   = 25;
	const MAX_BATCHES  = 4;
	const TIMEOUT      = 20;
	const DRAIN_BUDGET = 50;

	/**
	 * Concurrency lock. Longer than DRAIN_BUDGET plus one TIMEOUT.
	 */
	const LOCK_KEY = 'advisorppc_queue_lock';
	const LOCK_TTL = 120;

	/**
	 * Event names that are never evicted to make room.
	 */
	const CRITICAL = array( 'purchase', 'refund' );

	/**
	 * Guard against double wiring.
	 *
	 * @var bool
	 */
	private static $booted = false;

	/**
	 * Register hooks. Idempotent, safe to call more than once.
	 *
	 * @return void
	 */
	public static function init() {
		if ( self::$booted ) {
			return;
		}

		self::$booted = true;

		add_filter( 'cron_schedules', array( __CLASS__, 'add_cron_schedule' ) ); // phpcs:ignore WordPress.WP.CronInterval.ChangeDetected -- five minutes, drains at most 100 queued events.
		add_action( self::CRON_HOOK, array( __CLASS__, 'drain' ) );
		// The table is created when a workspace is connected, not on every
		// site that merely activates the plugin. add() and drain() still call
		// maybe_install() first, so a site migrated without the table heals.
		add_action( 'advisorppc_connected', array( __CLASS__, 'maybe_install' ) );
		add_action( 'init', array( __CLASS__, 'maybe_schedule' ), 20 );
		add_action( 'advisorppc_deactivate', array( __CLASS__, 'deactivate' ) );
	}

	/**
	 * Fully qualified table name.
	 *
	 * @return string
	 */
	public static function table() {
		global $wpdb;

		return $wpdb->prefix . 'advisorppc_queue';
	}

	/**
	 * Add the five minute cron schedule.
	 *
	 * @param array $schedules Existing schedules.
	 * @return array
	 */
	public static function add_cron_schedule( $schedules ) {
		if ( ! is_array( $schedules ) ) {
			$schedules = array();
		}

		$schedules[ self::CRON_SCHEDULE ] = array(
			'interval' => 300,
			'display'  => __( 'Every five minutes (AdvisorPPC)', 'advisorppc' ),
		);

		return $schedules;
	}

	/**
	 * Schedule or unschedule the recurring drain, following the connected state.
	 *
	 * @return void
	 */
	public static function maybe_schedule() {
		// Deliberately cheaper than AdvisorPPC_Events::is_enabled(): this runs on
		// every request, and reading the site secret would add a database query
		// to every page load. The full check still runs inside drain().
		$enabled = false;

		if ( class_exists( 'AdvisorPPC_Options' ) ) {
			$enabled = (bool) AdvisorPPC_Options::get( 'connected', false );

			if ( $enabled && method_exists( 'AdvisorPPC_Options', 'module_enabled' ) ) {
				$enabled = (bool) AdvisorPPC_Options::module_enabled( 'server_events' );
			}
		}

		$scheduled = wp_next_scheduled( self::CRON_HOOK );

		if ( $enabled && ! $scheduled ) {
			wp_schedule_event( time() + 60, self::CRON_SCHEDULE, self::CRON_HOOK );
			return;
		}

		if ( ! $enabled && $scheduled && 0 === self::pending_count() ) {
			wp_clear_scheduled_hook( self::CRON_HOOK );
		}
	}

	/**
	 * Clear scheduled work. Lane A may call this on deactivation.
	 *
	 * @return void
	 */
	public static function deactivate() {
		wp_clear_scheduled_hook( self::CRON_HOOK );
	}

	/**
	 * Create or upgrade the table when the stored schema version is stale.
	 *
	 * Runs on connect, and again before every add() and drain(), so the queue
	 * exists once a site can use it, including a site that was migrated by
	 * copying files and the database. A site that never connects never gets
	 * the table.
	 *
	 * @return void
	 */
	public static function maybe_install() {
		if ( (int) get_option( self::OPTION_DB_VERSION, 0 ) === self::DB_VERSION ) {
			return;
		}

		self::install();
	}

	/**
	 * Create the table.
	 *
	 * @return void
	 */
	public static function install() {
		global $wpdb;

		$table           = self::table();
		$charset_collate = $wpdb->get_charset_collate();

		$sql = "CREATE TABLE {$table} (
	id bigint(20) unsigned NOT NULL AUTO_INCREMENT,
	event_id varchar(64) NOT NULL DEFAULT '',
	event_name varchar(32) NOT NULL DEFAULT '',
	payload longtext NOT NULL,
	attempts smallint(5) unsigned NOT NULL DEFAULT 0,
	next_attempt_at bigint(20) unsigned NOT NULL DEFAULT 0,
	created_at bigint(20) unsigned NOT NULL DEFAULT 0,
	last_error varchar(191) NOT NULL DEFAULT '',
	browser_half varchar(16) NOT NULL DEFAULT '',
	PRIMARY KEY  (id),
	KEY next_attempt_at (next_attempt_at),
	KEY event_id (event_id),
	KEY created_at (created_at)
) {$charset_collate};";

		if ( ! function_exists( 'dbDelta' ) ) {
			require_once ABSPATH . 'wp-admin/includes/upgrade.php';
		}

		dbDelta( $sql );

		// Autoloaded on purpose: it is a small integer that is read on every
		// request, so a non-autoloaded row would cost an extra query per page.
		update_option( self::OPTION_DB_VERSION, self::DB_VERSION );
	}

	/**
	 * Drop the table. Lane A's uninstall.php may call this.
	 *
	 * @return void
	 */
	public static function drop() {
		global $wpdb;

		$table = self::table();

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.SchemaChange, WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- dropping this plugin's own table.
		$wpdb->query( $wpdb->prepare( 'DROP TABLE IF EXISTS %i', $table ) );

		delete_option( self::OPTION_DB_VERSION );
	}

	/**
	 * Number of rows waiting.
	 *
	 * @return int
	 */
	public static function pending_count() {
		global $wpdb;

		$table = self::table();

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- a cached count would be wrong immediately.
		$count = $wpdb->get_var( $wpdb->prepare( 'SELECT COUNT(*) FROM %i', $table ) );

		return null === $count ? 0 : (int) $count;
	}

	/**
	 * Queue one built event.
	 *
	 * @param array  $event        Contract event from AdvisorPPC_Events::build().
	 * @param string $browser_half Local note, 'sent' or 'skipped', recording
	 *                             whether a browser pixel was produced for
	 *                             this conversion. Never sent to the service:
	 *                             CONTRACT.md owns the envelope, and this is
	 *                             bookkeeping for the site that produced it.
	 * @return bool
	 */
	public static function add( $event, $browser_half = '' ) {
		global $wpdb;

		if ( ! is_array( $event ) || empty( $event['event_id'] ) || empty( $event['event_name'] ) ) {
			return false;
		}

		self::maybe_install();

		$table      = self::table();
		$event_id   = substr( (string) $event['event_id'], 0, 64 );
		$event_name = substr( (string) $event['event_name'], 0, 32 );

		// Already waiting to go out. Deduping here keeps a hook that fires twice
		// in one request (thankyou plus payment_complete) from queueing twice.
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- queue state, must not be cached.
		$existing = $wpdb->get_var(
			$wpdb->prepare(
				'SELECT id FROM %i WHERE event_id = %s LIMIT 1',
				$table,
				$event_id
			)
		);

		if ( $existing ) {
			return true;
		}

		if ( ! self::make_room( $event_name ) ) {
			return false;
		}

		$payload = wp_json_encode( $event );

		if ( ! is_string( $payload ) ) {
			return false;
		}

		$now = time();

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery -- custom table, $wpdb->insert prepares internally.
		$inserted = $wpdb->insert(
			$table,
			array(
				'event_id'        => $event_id,
				'event_name'      => $event_name,
				'payload'         => $payload,
				'attempts'        => 0,
				'next_attempt_at' => $now,
				'created_at'      => $now,
				'last_error'      => '',
				'browser_half'    => in_array( $browser_half, array( 'sent', 'skipped' ), true ) ? $browser_half : '',
			),
			array( '%s', '%s', '%s', '%d', '%d', '%d', '%s', '%s' )
		);

		if ( ! $inserted ) {
			return false;
		}

		// Ask WordPress to run cron on the next request. This does not block the
		// current one, so the customer still gets their confirmation page now.
		if ( ! wp_next_scheduled( self::CRON_HOOK ) || in_array( $event_name, self::CRITICAL, true ) ) {
			wp_schedule_single_event( $now + 10, self::CRON_HOOK );
		}

		return true;
	}

	/**
	 * Enforce the row cap before an insert.
	 *
	 * Non-critical events are evicted oldest first. A purchase or refund will
	 * evict the oldest row of any kind rather than be dropped itself.
	 *
	 * @param string $event_name Incoming event name.
	 * @return bool              False when the incoming event must be dropped.
	 */
	private static function make_room( $event_name ) {
		global $wpdb;

		if ( self::pending_count() < self::MAX_ROWS ) {
			return true;
		}

		$table = self::table();

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching
		$wpdb->query(
			$wpdb->prepare(
				"DELETE FROM %i WHERE event_name NOT IN ( 'purchase', 'refund' ) ORDER BY id ASC LIMIT %d",
				$table,
				200
			)
		);

		if ( self::pending_count() < self::MAX_ROWS ) {
			return true;
		}

		if ( ! in_array( $event_name, self::CRITICAL, true ) ) {
			return false;
		}

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching
		$wpdb->query(
			$wpdb->prepare(
				'DELETE FROM %i ORDER BY id ASC LIMIT %d',
				$table,
				1
			)
		);

		return self::pending_count() < self::MAX_ROWS;
	}

	/**
	 * Drain due rows. Runs on cron only.
	 *
	 * @return int Number of events accepted by the API.
	 */
	public static function drain() {
		global $wpdb;

		if ( ! class_exists( 'AdvisorPPC_Events' ) || ! class_exists( 'AdvisorPPC_Signer' ) ) {
			return 0;
		}

		self::maybe_install();
		self::purge_expired();

		if ( ! AdvisorPPC_Events::is_enabled() ) {
			// Not connected, or the module is off. Rows stay put and age out.
			return 0;
		}

		if ( get_transient( self::LOCK_KEY ) ) {
			return 0;
		}

		set_transient( self::LOCK_KEY, 1, self::LOCK_TTL );

		$table   = self::table();
		$sent    = 0;
		$started = time();

		for ( $batch = 0; $batch < self::MAX_BATCHES; $batch++ ) {
			// Stop before a POST that could not finish inside the budget. The
			// rows are still due and the next run picks them up.
			if ( $batch > 0 && ( time() - $started ) + self::timeout() > self::DRAIN_BUDGET ) {
				break;
			}

			// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching
			$rows = $wpdb->get_results(
				$wpdb->prepare(
					'SELECT id, payload, attempts FROM %i WHERE next_attempt_at <= %d ORDER BY id ASC LIMIT %d',
					$table,
					time(),
					self::BATCH_SIZE
				)
			);

			if ( empty( $rows ) ) {
				break;
			}

			$ids    = array();
			$events = array();

			foreach ( $rows as $row ) {
				$decoded = json_decode( $row->payload, true );

				if ( ! is_array( $decoded ) || empty( $decoded['event_id'] ) ) {
					// Undecodable payload will never succeed. Drop it.
					self::delete_ids( array( (int) $row->id ) );
					continue;
				}

				$ids[]    = (int) $row->id;
				$events[] = $decoded;
			}

			if ( empty( $events ) ) {
				continue;
			}

			$result = self::post( $events );

			if ( 'ok' === $result['status'] ) {
				self::delete_ids( $ids );
				$sent += count( $events );
				continue;
			}

			if ( 'permanent' === $result['status'] ) {
				self::log( 'dropping ' . count( $ids ) . ' event(s): ' . $result['error'] );
				self::delete_ids( $ids );
				continue;
			}

			// Every other outcome keeps the rows. A transport failure, a 5xx,
			// and a 200 carrying `retry: true` all spend an attempt; a plan
			// gate does not (see classify_response() and defer()).
			self::defer( $ids, $result['error'], empty( $result['gated'] ) );
			break;
		}

		delete_transient( self::LOCK_KEY );

		return $sent;
	}

	/**
	 * POST a batch of events.
	 *
	 * @param array $events Events.
	 * @return array        { status: ok|retry|permanent, error: string, gated: bool }
	 */
	private static function post( $events ) {
		// The queue POSTs directly rather than through AdvisorPPC_Http, so it
		// carries the same build-level gate. Rows stay put and age out.
		if ( ! AdvisorPPC_Http::workspace_enabled() ) {
			return array(
				'status' => 'retry',
				'error'  => 'workspace features are off in this version',
			);
		}

		$body = wp_json_encode( AdvisorPPC_Events::envelope( $events ) );

		if ( ! is_string( $body ) ) {
			return array(
				'status' => 'permanent',
				'error'  => 'payload could not be encoded',
			);
		}

		$timestamp = time();
		$headers   = AdvisorPPC_Signer::headers( $body, $timestamp );

		if ( empty( $headers ) ) {
			return array(
				'status' => 'retry',
				'error'  => 'not connected',
			);
		}

		$version = defined( 'ADVISORPPC_VERSION' ) ? ADVISORPPC_VERSION : '0.0.0';

		$response = wp_remote_post(
			self::endpoint(),
			array(
				'timeout'     => self::timeout(),
				'redirection' => 0,
				'blocking'    => true,
				'sslverify'   => true,
				'headers'     => $headers,
				'body'        => $body,
				'user-agent'  => 'AdvisorPPC-WP/' . $version . '; ' . AdvisorPPC_Events::site_url(),
			)
		);

		if ( is_wp_error( $response ) ) {
			return array(
				'status' => 'retry',
				'error'  => $response->get_error_message(),
			);
		}

		$code = (int) wp_remote_retrieve_response_code( $response );
		$raw  = wp_remote_retrieve_body( $response );

		$result = self::classify_response( $code, $raw );

		if ( 'ok' === $result['status'] ) {
			// A 200 does NOT mean every event in the batch was stored. The
			// contract response carries per-event `rejected` and `errors`, and
			// a rejected event is dropped here along with the rest of the
			// batch. Retrying would not help: a rejection is a malformed body,
			// which will be malformed again. So it is logged loudly instead of
			// disappearing, which is the difference between a bug someone can
			// find and revenue that quietly never arrives.
			self::log_rejections( $raw );
		}

		return $result;
	}

	/**
	 * The HTTP timeout for one batch POST, in seconds.
	 *
	 * @return int Between 5 and 60.
	 */
	public static function timeout() {
		/**
		 * Filter the ingest POST timeout.
		 *
		 * The ingest fans out to every destination before it answers, so this
		 * has to cover a full fan-out or the plugin abandons requests the
		 * server is still completing.
		 *
		 * @since 0.1.0
		 * @param int $timeout Seconds. Default 20.
		 */
		$timeout = (int) apply_filters( 'advisorppc_queue_timeout', self::TIMEOUT );

		return max( 5, min( 60, $timeout ) );
	}

	/**
	 * Turn an HTTP status and response body into a queue decision.
	 *
	 * Pure: no network, no database, no WordPress. Kept separate from post()
	 * so the retry rules can be checked standalone (tests/test-queue-response.php).
	 *
	 * Contract response, always 200 unless auth failed:
	 *
	 *   { accepted, rejected, errors, plan_gated, retry }
	 *
	 * `plan_gated: true`  the batch, or a fan-out destination for one of its
	 *                     events, was refused on plan grounds. Nothing was
	 *                     delivered there. The rows are kept and re-sent later
	 *                     WITHOUT spending an attempt: a gate can last a week
	 *                     and MAX_AGE, not MAX_ATTEMPTS, bounds it.
	 * `retry: true`       for at least one event no destination reached
	 *                     "sent" and at least one returned a transient error.
	 *                     Treated exactly like a transport failure: a counted
	 *                     attempt with backoff. Events whose destinations were
	 *                     all skipped or suppressed (consent, no pixel id) do
	 *                     not set this and are not re-sent.
	 *
	 * When both are set the gate wins, because the gated rows are the ones
	 * that must survive the longest. Re-sending is safe on every path: the
	 * commerce ledger is idempotent and fan-out is claim-guarded per
	 * (site, event_id), so a replay of an already-sent event is accepted,
	 * counted, and not dispatched again.
	 *
	 * A 2xx counts as delivered ONLY when its body is the contract response:
	 * a JSON object carrying an integer `accepted`. An empty body, a 204, an
	 * HTML page or anything else that is not that object is a proxy, a CDN
	 * interstitial or a load balancer answering in place of the ingest, and
	 * a row deleted on the strength of that answer is the plugin's only copy
	 * of a sale gone on a page that never saw it. Those are retried, with a
	 * counted attempt, like any other transport failure.
	 *
	 * @param int    $code HTTP status code.
	 * @param string $raw  Raw response body.
	 * @return array       { status: ok|retry|permanent, error: string, gated: bool }
	 */
	public static function classify_response( $code, $raw ) {
		$code = (int) $code;

		if ( $code >= 200 && $code < 300 ) {
			$decoded = ( is_string( $raw ) && '' !== $raw ) ? json_decode( $raw, true ) : null;

			if ( ! is_array( $decoded ) || ! isset( $decoded['accepted'] ) || ! is_numeric( $decoded['accepted'] ) ) {
				return array(
					'status' => 'retry',
					'error'  => 'HTTP ' . $code . ' without a contract response body',
					'gated'  => false,
				);
			}

			// A plan-gated batch was NOT delivered anywhere. Treating it as
			// success deletes the rows, so a customer who upgrades the next day
			// has permanently lost the conversions that arrived while they were
			// gated. `accepted` cannot tell the two apart: it counts events
			// that PARSED, not events that were delivered.
			if ( ! empty( $decoded['plan_gated'] ) ) {
				return array(
					'status' => 'retry',
					'error'  => 'plan gated',
					'gated'  => true,
				);
			}

			// The server accepted the batch but could not deliver at least one
			// event anywhere because a destination failed transiently. That is
			// a transport failure that happened one hop further on, and it is
			// retried the same way: counted, with backoff, bounded by
			// MAX_ATTEMPTS. The rows are NOT deleted on this 200.
			if ( ! empty( $decoded['retry'] ) ) {
				return array(
					'status' => 'retry',
					'error'  => 'destination asked for a retry',
					'gated'  => false,
				);
			}

			return array(
				'status' => 'ok',
				'error'  => '',
				'gated'  => false,
			);
		}

		// Auth failures and rate limits are worth retrying: a key can be
		// repaired, a limit expires.
		if ( 401 === $code || 403 === $code || 408 === $code || 429 === $code || $code >= 500 ) {
			return array(
				'status' => 'retry',
				'error'  => 'HTTP ' . $code,
				'gated'  => false,
			);
		}

		// Any other 4xx means the body itself is wrong. Retrying forever would
		// just fill the table.
		return array(
			'status' => 'permanent',
			'error'  => 'HTTP ' . $code,
			'gated'  => false,
		);
	}

	/**
	 * Ingest endpoint URL.
	 *
	 * @return string
	 */
	public static function endpoint() {
		// CONTRACT rule 10: the service directory the server returned at
		// exchange (and refreshes through /wp/v1/health) is the host authority.
		// AdvisorPPC_Http::api_base() reads its `tracking` entry first, applies
		// the allowlist, falls back to the compiled-in base, and only then runs
		// the documented `advisorppc_api_base` filter. Going through it means a
		// host move, a staging override and a proxy all apply to the live event
		// path, which is the one request that carries money.
		if ( class_exists( 'AdvisorPPC_Http' ) && method_exists( 'AdvisorPPC_Http', 'url' ) ) {
			$url = AdvisorPPC_Http::url( '/wp/v1/events' );

			// Events carry hashed customer identifiers and order values. A
			// filter that downgrades the transport is refused rather than
			// honoured.
			if ( is_string( $url ) && 0 === strpos( $url, 'https://' ) ) {
				return $url;
			}
		}

		// Without the HTTP shell: the directory still comes first, under the
		// same allowlist, and the constant is the fallback for a site that has
		// never connected.
		$base = '';

		if ( class_exists( 'AdvisorPPC_Options' ) && method_exists( 'AdvisorPPC_Options', 'service_url' ) ) {
			$base = (string) AdvisorPPC_Options::service_url( 'tracking' );

			if ( '' !== $base
				&& class_exists( 'AdvisorPPC_Connect' )
				&& method_exists( 'AdvisorPPC_Connect', 'is_allowed_service_url' )
				&& ! AdvisorPPC_Connect::is_allowed_service_url( $base ) ) {
				$base = '';
			}
		}

		if ( '' === $base ) {
			$base = defined( 'ADVISORPPC_API_BASE' ) ? ADVISORPPC_API_BASE : 'https://mcp.advisorppc.com';
		}

		return untrailingslashit( $base ) . '/wp/v1/events';
	}

	/**
	 * Delete rows by id.
	 *
	 * @param array $ids Row ids.
	 * @return void
	 */
	private static function delete_ids( $ids ) {
		global $wpdb;

		$ids = array_values( array_filter( array_map( 'intval', (array) $ids ) ) );

		if ( empty( $ids ) ) {
			return;
		}

		$table        = self::table();
		$placeholders = implode( ', ', array_fill( 0, count( $ids ), '%d' ) );

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching
		$wpdb->query(
			$wpdb->prepare(
				// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.PreparedSQLPlaceholders.UnfinishedPrepare -- $placeholders is a generated run of %d only, one per id.
				"DELETE FROM %i WHERE id IN ( {$placeholders} )",
				array_merge( array( $table ), $ids )
			)
		);
	}

	/**
	 * Record a failed attempt and push the rows into the future.
	 *
	 * Transport failure or `retry: true`: exponential backoff, 60s, 120s,
	 * 240s, 480s, 960s, capped at one hour, plus jitter so a fleet of sites
	 * does not retry in lockstep. A non-critical row that exhausts
	 * MAX_ATTEMPTS is deleted; a purchase or refund keeps retrying at the cap
	 * until MAX_AGE.
	 *
	 * Plan gate: no attempt is spent, the attempt counter is RESET, and the
	 * delay comes from gate_delay(). The reset matters when a response
	 * alternates between gated and a transient error: without it the error
	 * path keeps incrementing and MAX_ATTEMPTS deletes a row whose real
	 * reason for not delivering is the gate. A gate is evidence that the
	 * server is up and answering, so the transient-failure count starts over.
	 *
	 * @param array  $ids           Row ids.
	 * @param string $error         Error text.
	 * @param bool   $count_attempt False for a plan gate, which is not a
	 *                              failure and must not spend a try.
	 * @return void
	 */
	private static function defer( $ids, $error, $count_attempt = true ) {
		global $wpdb;

		$ids = array_values( array_filter( array_map( 'intval', (array) $ids ) ) );

		if ( empty( $ids ) ) {
			return;
		}

		$table = self::table();
		$error = substr( sanitize_text_field( (string) $error ), 0, 191 );
		$now   = time();

		foreach ( $ids as $id ) {
			// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching
			$row = $wpdb->get_row(
				$wpdb->prepare(
					'SELECT attempts, event_name, created_at FROM %i WHERE id = %d',
					$table,
					$id
				),
				ARRAY_A
			);

			if ( ! is_array( $row ) ) {
				continue;
			}

			$attempts   = (int) $row['attempts'];
			$critical   = in_array( (string) $row['event_name'], self::CRITICAL, true );
			$created_at = (int) $row['created_at'];

			if ( ! $count_attempt ) {
				// A plan gate is not a transport failure. Counting it as an
				// attempt spent the row's six tries in about half an hour, so a
				// customer who upgraded the next morning had already lost the
				// conversions that arrived overnight. Gated rows are bounded by
				// MAX_AGE (purge_expired(), which every drain() calls), and
				// re-sent on the age-based schedule in gate_delay().
				$attempts = 0;
				$delay    = self::gate_delay( $created_at, $now );
			} else {
				$attempts++;

				if ( $attempts >= self::MAX_ATTEMPTS && ! $critical ) {
					self::log( 'giving up on queue row ' . $id . ' after ' . $attempts . ' attempts: ' . $error );
					self::delete_ids( array( $id ) );
					continue;
				}

				$delay = self::retry_delay( $attempts );
			}

			$delay += wp_rand( 0, 30 );

			// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- queue state, must not be cached.
			$wpdb->update(
				$table,
				array(
					'attempts'        => $attempts,
					'next_attempt_at' => $now + (int) $delay,
					'last_error'      => $error,
				),
				array( 'id' => $id ),
				array( '%d', '%d', '%s' ),
				array( '%d' )
			);
		}
	}

	/**
	 * Backoff after a counted failure, before jitter.
	 *
	 * Pure, so the schedule can be checked standalone.
	 *
	 * @param int $attempts Attempts made so far, including this one.
	 * @return int          Seconds.
	 */
	public static function retry_delay( $attempts ) {
		$attempts = max( 1, (int) $attempts );
		$delay    = self::BACKOFF_BASE * pow( 2, $attempts - 1 );

		return (int) min( self::BACKOFF_CAP, $delay );
	}

	/**
	 * Backoff after a plan gate, before jitter.
	 *
	 * A quarter of the row's age, clamped to [GATE_BACKOFF_MIN, GATE_BACKOFF_CAP]:
	 * 15 minutes for the first hour, an hour by four hours of age, six hours
	 * from a day old onwards. Pure, so the schedule can be checked standalone.
	 *
	 * @param int $created_at Row creation time, unix seconds.
	 * @param int $now        Current time, unix seconds.
	 * @return int            Seconds.
	 */
	public static function gate_delay( $created_at, $now ) {
		$age   = max( 0, (int) $now - (int) $created_at );
		$delay = (int) floor( $age / 4 );

		return (int) max( self::GATE_BACKOFF_MIN, min( self::GATE_BACKOFF_CAP, $delay ) );
	}

	/**
	 * Delete rows older than MAX_AGE whatever their state.
	 *
	 * @return void
	 */
	public static function purge_expired() {
		global $wpdb;

		$table = self::table();

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching
		$wpdb->query(
			$wpdb->prepare(
				'DELETE FROM %i WHERE created_at > 0 AND created_at < %d',
				$table,
				time() - self::MAX_AGE
			)
		);
	}

	/**
	 * Queue health, for lane A's admin screen.
	 *
	 * @return array
	 */
	public static function stats() {
		global $wpdb;

		$table = self::table();

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching
		$row = $wpdb->get_row(
			$wpdb->prepare( 'SELECT COUNT(*) AS pending, MAX(attempts) AS worst, MIN(created_at) AS oldest FROM %i', $table ),
			ARRAY_A
		);

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching
		$last_error = $wpdb->get_var(
			$wpdb->prepare( "SELECT last_error FROM %i WHERE last_error <> '' ORDER BY id DESC LIMIT 1", $table )
		);

		return array(
			'pending'    => isset( $row['pending'] ) ? (int) $row['pending'] : 0,
			'attempts'   => isset( $row['worst'] ) ? (int) $row['worst'] : 0,
			'oldest'     => isset( $row['oldest'] ) ? (int) $row['oldest'] : 0,
			'last_error' => is_string( $last_error ) ? $last_error : '',
			'next_run'   => (int) wp_next_scheduled( self::CRON_HOOK ),
		);
	}

	/**
	 * Debug logging. Silent unless WP_DEBUG is on, per wordpress.org guidance.
	 *
	 * @param string $message Message. Never contains a secret or raw PII.
	 * @return void
	 */
	private static function log( $message ) {
		if ( class_exists( 'AdvisorPPC_Logger' ) && method_exists( 'AdvisorPPC_Logger', 'warning' ) ) {
			call_user_func( array( 'AdvisorPPC_Logger', 'warning' ), 'queue: ' . $message );
			return;
		}

		if ( defined( 'WP_DEBUG' ) && WP_DEBUG ) {
			// phpcs:ignore WordPress.PHP.DevelopmentFunctions.error_log_error_log -- debug builds only.
			error_log( 'AdvisorPPC queue: ' . $message );
		}
	}
	/**
	 * Log any per-event rejection the server reported inside a 200.
	 *
	 * @param string $raw Raw response body.
	 * @return int        Number of rejected events.
	 */
	private static function log_rejections( $raw ) {
		if ( ! is_string( $raw ) || '' === $raw ) {
			return 0;
		}

		$decoded = json_decode( $raw, true );

		if ( ! is_array( $decoded ) ) {
			return 0;
		}

		$rejected = isset( $decoded['rejected'] ) ? (int) $decoded['rejected'] : 0;

		if ( $rejected < 1 ) {
			return 0;
		}

		$detail = '';

		if ( isset( $decoded['errors'] ) && is_array( $decoded['errors'] ) ) {
			$parts = array();

			foreach ( $decoded['errors'] as $error ) {
				if ( is_array( $error ) && isset( $error['error'] ) ) {
					$parts[] = (string) $error['error'];
				}
			}

			$detail = implode( '; ', array_slice( $parts, 0, 5 ) );
		}

		if ( class_exists( 'AdvisorPPC_Logger' ) ) {
			AdvisorPPC_Logger::error(
				sprintf(
					/* translators: 1: number of rejected events, 2: server error detail. */
					'AdvisorPPC: %1$d event(s) were refused by the server and have been discarded. %2$s',
					$rejected,
					$detail
				)
			);
		}

		/**
		 * Fires when the server refused one or more events inside a 200.
		 *
		 * @since 0.1.0
		 * @param int    $rejected Number of refused events.
		 * @param string $detail   Server-supplied reasons, truncated.
		 * @param array  $decoded  Full decoded response.
		 */
		do_action( 'advisorppc_events_rejected', $rejected, $detail, $decoded );

		return $rejected;
	}

}

add_action( 'advisorppc_loaded', array( 'AdvisorPPC_Queue', 'init' ) );