<?php
/** Durable background delivery. Accepted events remain until the API confirms their UUID. */
if (!defined('ABSPATH')) {
  exit;
}

class AirefsTrackingQueue
{
  const HOOK = 'airefs_tracking_flush';
  const PREFIX = '_airefs_event_';
  const MAX_BYTES = 32768;
  const CAPACITY_BYTES = 67108864;
  const CAPACITY = 100000;
  const SCHEMA = '2';
  const SETUP_RETRY_SECONDS = 900;

  private static $continued = false;
  private static $storage_verified = array();

  private static function table()
  {
    global $wpdb;
    return $wpdb->base_prefix . 'airefs_event_queue';
  }

  private static function options_table()
  {
    global $wpdb;
    return $wpdb->base_prefix . 'options';
  }

  private static function network()
  {
    return is_multisite() ? get_current_network_id() : 1;
  }

  private static function prefix()
  {
    return self::PREFIX . 'n' . self::network() . '_b' . get_current_blog_id() . '_';
  }

  public static function binding($options)
  {
    return hash_hmac('sha256', $options['access_token'] . "\n" . rtrim($options['api_endpoint'], '/'), wp_salt('auth'));
  }

  /** Compare-and-delete leases prevent an expired owner from unlocking its successor. */
  private static function lock($name, $seconds)
  {
    global $wpdb;
    $table = self::options_table();
    $value = (time() + $seconds) . ':' . wp_generate_uuid4();
    $wpdb->query($wpdb->prepare("DELETE FROM $table WHERE option_name = %s AND CAST(SUBSTRING_INDEX(option_value, ':', 1) AS UNSIGNED) < %d", $name, time()));
    $added = $wpdb->query($wpdb->prepare("INSERT IGNORE INTO $table (option_name, option_value, autoload) VALUES (%s, %s, 'no')", $name, $value));
    return $added === 1 ? $value : false;
  }

  private static function unlock($name, $value)
  {
    global $wpdb;
    $table = self::options_table();
    return $wpdb->query($wpdb->prepare("DELETE FROM $table WHERE option_name = %s AND option_value = %s", $name, $value)) === 1;
  }

  private static function enabled_options()
  {
    $options = get_option(AIREFS_TRACKING_OPTIONS_NAME, array());
    if (empty($options['tracking_enabled']) || empty($options['access_token'])) {
      return false;
    }
    $options['api_endpoint'] = isset($options['api_endpoint']) ? rtrim($options['api_endpoint'], '/') : 'https://api.getairefs.com';
    return $options;
  }

  public static function init()
  {
    add_action(self::HOOK, array(__CLASS__, 'flush'));
    add_action('init', array(__CLASS__, 'ensure_scheduled'));
    add_action('update_option_' . AIREFS_TRACKING_OPTIONS_NAME, array(__CLASS__, 'settings_changed'), 10, 2);
    add_action('wp_ajax_airefs_queue_runner', array(__CLASS__, 'run_async'));
    add_action('wp_ajax_nopriv_airefs_queue_runner', array(__CLASS__, 'run_async'));
  }

  private static function storage_exists()
  {
    global $wpdb;
    $table = self::table();
    return $wpdb->get_var($wpdb->prepare('SHOW TABLES LIKE %s', $wpdb->esc_like($table))) === $table
      && $wpdb->get_var($wpdb->prepare("SHOW COLUMNS FROM $table LIKE %s", 'retry_individually')) === 'retry_individually';
  }

  public static function install($force = false)
  {
    global $wpdb;
    $suppressed = $wpdb->suppress_errors(true);
    try {
      $network = self::network();
      $schema = get_site_option('airefs_queue_schema');
      if (!$force && $schema === self::SCHEMA && isset(self::$storage_verified[$network]) && self::$storage_verified[$network] === $schema) {
        return true;
      }
      unset(self::$storage_verified[$network]);
      if ($schema !== self::SCHEMA || !self::storage_exists()) {
        if (!$force && (int) get_site_option('airefs_queue_storage_retry_at', 0) > time()) {
          return false;
        }
        update_site_option('airefs_queue_storage_error', 'WordPress could not create or update the local tracking queue.');
        update_site_option('airefs_queue_storage_retry_at', time() + self::SETUP_RETRY_SECONDS);
        require_once ABSPATH . 'wp-admin/includes/upgrade.php';
        $table = self::table();
        $collate = $wpdb->get_charset_collate();
        dbDelta("CREATE TABLE $table (
      id bigint(20) unsigned NOT NULL AUTO_INCREMENT,
      event_id varchar(36) NOT NULL,
      network_id bigint(20) unsigned NOT NULL,
      blog_id bigint(20) unsigned NOT NULL,
      binding varchar(64) NOT NULL,
      endpoint varchar(2048) NOT NULL DEFAULT '',
      site_id varchar(191) NOT NULL DEFAULT '',
      payload longtext NOT NULL,
      payload_bytes int(10) unsigned NOT NULL,
      created_at bigint(20) unsigned NOT NULL,
      next_attempt bigint(20) unsigned NOT NULL,
      attempts int(10) unsigned NOT NULL DEFAULT 0,
      claim varchar(36) NOT NULL DEFAULT '',
      claim_until bigint(20) unsigned NOT NULL DEFAULT 0,
      blocked tinyint(1) unsigned NOT NULL DEFAULT 0,
      retry_individually tinyint(1) unsigned NOT NULL DEFAULT 0,
      last_error varchar(255) NOT NULL DEFAULT '',
      PRIMARY KEY  (id),
      UNIQUE KEY event (network_id,blog_id,event_id),
      KEY ready (network_id,blog_id,next_attempt),
      KEY binding (network_id,blog_id,binding)
    ) $collate;");
        if (!self::storage_exists()) {
          return false;
        }
        update_site_option('airefs_queue_schema', self::SCHEMA);
      }
      delete_site_option('airefs_queue_storage_error');
      delete_site_option('airefs_queue_storage_retry_at');
      if (get_option('airefs_queue_error', '') === 'Queue storage could not be created.') {
        self::status('error', '');
      }
      self::$storage_verified[$network] = self::SCHEMA;
      return true;
    } finally {
      $wpdb->suppress_errors($suppressed);
    }
  }

  public static function retry_setup()
  {
    if (!self::install(true)) {
      return false;
    }
    self::ensure_scheduled();
    return true;
  }

  private static function status($key, $value)
  {
    update_option('airefs_queue_' . $key, $value, false);
  }

  private static function reject($code, $message)
  {
    global $wpdb;
    // Atomic increments keep concurrent capacity rejections visible.
    $name = '_airefs_queue_rejected_n' . self::network() . '_b' . get_current_blog_id();
    $table = self::options_table();
    $wpdb->query($wpdb->prepare("INSERT INTO $table (option_name, option_value, autoload) VALUES (%s, '1', 'no') ON DUPLICATE KEY UPDATE option_value = CAST(option_value AS UNSIGNED) + 1", $name));
    self::status('error', $message);
    self::status('last_rejected', time());
    self::status('rejected_reason', $message);
    return new WP_Error($code, $message, array('status' => 503));
  }

  public static function settings_changed($old, $new)
  {
    // Old rows retain their original destination identity across disablement and key changes.
    foreach (array('tracking_enabled', 'access_token', 'api_endpoint') as $key) {
      if ((isset($old[$key]) ? $old[$key] : null) !== (isset($new[$key]) ? $new[$key] : null)) {
        self::status('identity_retry', 0);
        self::ensure_scheduled();
        break;
      }
    }
  }

  /** Copy legacy rows before deleting them; deterministic UUIDs make interrupted migration safe. */
  private static function migrate_legacy()
  {
    global $wpdb;
    if ((int) get_option('airefs_queue_legacy_retry', 0) > time()) {
      return;
    }
    $options_table = self::options_table();
    $table = self::table();
    $cursor = (int) get_option('airefs_queue_legacy_cursor', 0);
    $rows = $wpdb->get_results($wpdb->prepare("SELECT option_id, option_name, option_value FROM $options_table WHERE option_name LIKE %s AND option_id > %d ORDER BY option_id LIMIT 100", $wpdb->esc_like(self::prefix()) . '%', $cursor));
    $options = self::enabled_options();
    foreach ($rows as $row) {
      $cursor = (int) $row->option_id;
      $event = json_decode($row->option_value, true);
      $parts = explode('_', $row->option_name);
      $uuid = end($parts);
      $created = (int) $parts[count($parts) - 2];
      if (!is_array($event) || empty($event['binding']) || !is_string($event['binding']) || !isset($event['payload']) || !is_array($event['payload']) || !preg_match('/^[a-f0-9-]{36}$/i', $uuid)) {
        self::status('error', 'A legacy event could not be migrated; its original data is retained.');
        continue;
      }
      $payload = wp_json_encode($event['payload']);
      if (!is_string($payload)) {
        continue;
      }
      $endpoint = $options && hash_equals(self::binding($options), $event['binding']) ? $options['api_endpoint'] : '';
      $added = $wpdb->query($wpdb->prepare("INSERT IGNORE INTO $table (event_id, network_id, blog_id, binding, endpoint, payload, payload_bytes, created_at, next_attempt) VALUES (%s, %d, %d, %s, %s, %s, %d, %d, %d)", $uuid, self::network(), get_current_blog_id(), $event['binding'], $endpoint, $payload, strlen($payload) + strlen($endpoint) + 512, $created, time()));
      if ($added === 1 || ($added === 0 && $wpdb->get_var($wpdb->prepare("SELECT id FROM $table WHERE network_id = %d AND blog_id = %d AND event_id = %s", self::network(), get_current_blog_id(), $uuid)))) {
        $wpdb->query($wpdb->prepare("DELETE FROM $options_table WHERE option_name = %s AND option_value = %s", $row->option_name, $row->option_value));
      }
    }
    self::status('legacy_cursor', count($rows) < 100 ? 0 : $cursor);
    if (count($rows) < 100) {
      // A full scan has reached its end; retained malformed rows must not cause a busy loop.
      $remaining = $wpdb->get_var($wpdb->prepare("SELECT option_id FROM $options_table WHERE option_name LIKE %s LIMIT 1", $wpdb->esc_like(self::prefix()) . '%'));
      self::status('legacy_retry', $remaining ? time() + 3600 : 0);
    }
  }

  public static function enqueue($payload)
  {
    global $wpdb;
    $options = self::enabled_options();
    if (!$options) {
      return new WP_Error('tracking_disabled', 'Tracking is not configured.', array('status' => 403));
    }
    $payload['url_query_allowlist'] = airefs_tracking_query_allowlist(isset($options['url_query_allowlist']) ? $options['url_query_allowlist'] : array());
    $payload['url'] = isset($payload['url']) ? airefs_tracking_event_url($payload['url'], $payload['url_query_allowlist']) : null;
    if ($payload['url'] === null) {
      return new WP_Error('invalid_event_url', 'Event URL must be an HTTP or HTTPS URL.', array('status' => 400));
    }
    $event = wp_json_encode($payload);
    if (!is_string($event) || strlen($event) > self::MAX_BYTES) {
      return new WP_Error('event_too_large', 'Tracking event is too large.', array('status' => 413));
    }
    if (!self::install()) {
      return self::reject('queue_unavailable', 'Tracking queue storage is unavailable.');
    }
    $table = self::table();
    $lock_name = '_airefs_queue_capacity_n' . self::network();
    $lease = self::lock($lock_name, 30);
    $deadline = microtime(true) + 0.1;
    while (!$lease && microtime(true) < $deadline) {
      usleep(5000);
      $lease = self::lock($lock_name, 30);
    }
    if (!$lease) {
      return self::reject('queue_busy', 'Tracking queue is busy; this new event was not accepted.');
    }
    $options_table = self::options_table();
    $legacy = $wpdb->get_row($wpdb->prepare("SELECT COUNT(*) AS count, COALESCE(SUM(OCTET_LENGTH(option_value) + 512), 0) AS bytes FROM $options_table WHERE option_name LIKE %s", $wpdb->esc_like(self::PREFIX . 'n' . self::network() . '_') . '%'));
    $usage = $wpdb->get_row($wpdb->prepare("SELECT COUNT(*) AS count, COALESCE(SUM(payload_bytes), 0) AS bytes FROM $table WHERE network_id = %d", self::network()));
    $bytes = strlen($event) + strlen($options['api_endpoint']) + 512;
    $capacity = self::capacity_bytes();
    if (!$usage || (int) $usage->count + ($legacy ? (int) $legacy->count : 0) >= self::CAPACITY || (int) $usage->bytes + ($legacy ? (int) $legacy->bytes : 0) + $bytes > $capacity) {
      self::unlock($lock_name, $lease);
      return self::reject('queue_full', 'Tracking queue capacity reached; this new event was not accepted. Saved events are retained.');
    }
    $identity = get_option('airefs_queue_identity', array());
    $site_id = isset($identity['binding'], $identity['endpoint'], $identity['site_id']) && hash_equals($identity['binding'], self::binding($options)) && $identity['endpoint'] === $options['api_endpoint'] ? $identity['site_id'] : '';
    $added = $wpdb->insert($table, array(
      'event_id' => wp_generate_uuid4(), 'network_id' => self::network(), 'blog_id' => get_current_blog_id(),
      'binding' => self::binding($options), 'endpoint' => $options['api_endpoint'], 'site_id' => $site_id, 'payload' => $event,
      'payload_bytes' => $bytes, 'created_at' => time(), 'next_attempt' => time(),
    ), array('%s', '%d', '%d', '%s', '%s', '%s', '%s', '%d', '%d', '%d'));
    self::unlock($lock_name, $lease);
    if ($added !== 1) {
      return self::reject('queue_unavailable', 'Tracking queue could not save this event.');
    }
    self::ensure_scheduled();
    return true;
  }

  private static function capacity_bytes()
  {
    return max(self::MAX_BYTES, (int) apply_filters('airefs_tracking_queue_capacity_bytes', self::CAPACITY_BYTES, self::network()));
  }

  public static function ensure_scheduled()
  {
    global $wpdb;
    if (!self::install()) {
      return false;
    }
    if (!self::enabled_options()) {
      return true;
    }
    $table = self::table();
    $next = $wpdb->get_var($wpdb->prepare("SELECT MIN(GREATEST(next_attempt, claim_until)) FROM $table WHERE network_id = %d AND blog_id = %d", self::network(), get_current_blog_id()));
    $options_table = self::options_table();
    $legacy = $wpdb->get_var($wpdb->prepare("SELECT option_id FROM $options_table WHERE option_name LIKE %s LIMIT 1", $wpdb->esc_like(self::prefix()) . '%'));
    if ($next === null && !$legacy) {
      return true;
    }
    $legacy_next = max(time(), (int) get_option('airefs_queue_legacy_retry', 0));
    $next = $legacy ? ($next === null ? $legacy_next : min((int) $next, $legacy_next)) : (int) $next;
    $next = max($next, (int) get_option('airefs_queue_identity_retry', 0));
    if ($next <= time()) {
      add_action('shutdown', array(__CLASS__, 'dispatch'));
    }
    $scheduled = wp_next_scheduled(self::HOOK);
    if ($scheduled && $scheduled > max(time() + 60, $next)) {
      wp_unschedule_event($scheduled, self::HOOK);
      $scheduled = false;
    }
    if ($scheduled) {
      return true;
    }
    $result = wp_schedule_single_event(max(time() + 5, $next), self::HOOK);
    if ($result === false || is_wp_error($result)) {
      self::status('scheduler_error', 'WordPress could not schedule delivery. Saved events are retained; the signed background runner will also retry.');
      return false;
    }
    self::status('scheduler_error', '');
    return true;
  }

  /** A one-use secret authenticates the fallback runner; public requests cannot drain the queue. */
  public static function dispatch()
  {
    global $wpdb;
    if (!self::enabled_options() || (int) get_option('airefs_queue_identity_retry', 0) > time()) {
      return;
    }
    $table = self::table();
    $options_table = self::options_table();
    $due = $wpdb->get_var($wpdb->prepare("SELECT id FROM $table WHERE network_id = %d AND blog_id = %d AND next_attempt <= %d AND claim_until <= %d LIMIT 1", self::network(), get_current_blog_id(), time(), time()));
    $legacy = (int) get_option('airefs_queue_legacy_retry', 0) <= time() && $wpdb->get_var($wpdb->prepare("SELECT option_id FROM $options_table WHERE option_name LIKE %s LIMIT 1", $wpdb->esc_like(self::prefix()) . '%'));
    $worker = $wpdb->get_var($wpdb->prepare("SELECT option_value FROM $options_table WHERE option_name = %s", '_airefs_worker_' . get_current_blog_id()));
    if ((!$due && !$legacy) || ($worker && (int) $worker >= time())) {
      return;
    }
    if (!self::$continued && !self::lock('_airefs_queue_wake_b' . get_current_blog_id(), 10)) {
      return;
    }
    $name = '_airefs_queue_dispatch_b' . get_current_blog_id();
    $lease = self::lock($name, 60);
    if (!$lease) {
      return;
    }
    $response = wp_remote_post(admin_url('admin-ajax.php'), array(
      'timeout' => 0.01, 'blocking' => false, 'redirection' => 0,
      'body' => array('action' => 'airefs_queue_runner', 'ticket' => $lease),
    ));
    if (is_wp_error($response)) {
      self::unlock($name, $lease);
      self::status('scheduler_error', 'Background loopback could not start. WordPress cron will retry; saved events are retained.');
    }
  }

  public static function run_async()
  {
    global $wpdb;
    $ticket = isset($_POST['ticket']) && is_string($_POST['ticket']) ? wp_unslash($_POST['ticket']) : '';
    $table = self::options_table();
    $name = '_airefs_queue_dispatch_b' . get_current_blog_id();
    $stored = $wpdb->get_var($wpdb->prepare("SELECT option_value FROM $table WHERE option_name = %s", $name));
    if (!$stored || !hash_equals($stored, $ticket) || (int) $stored < time() || !self::unlock($name, $stored)) {
      wp_die('', '', array('response' => 403));
    }
    self::flush();
    wp_die();
  }

  public static function retry_now()
  {
    global $wpdb;
    if (!self::install()) {
      return false;
    }
    $table = self::table();
    $wpdb->query($wpdb->prepare("UPDATE $table SET next_attempt = %d WHERE network_id = %d AND blog_id = %d", time(), self::network(), get_current_blog_id()));
    self::status('identity_retry', 0);
    self::status('legacy_retry', 0);
    return self::ensure_scheduled();
  }

  public static function resume($network = false)
  {
    self::install();
    if ($network && is_multisite()) {
      $offset = 0;
      do {
        $sites = get_sites(array('network_id' => self::network(), 'fields' => 'ids', 'number' => 100, 'offset' => $offset));
        foreach ($sites as $blog_id) {
          switch_to_blog($blog_id);
          self::ensure_scheduled();
          restore_current_blog();
        }
        $offset += count($sites);
      } while (count($sites) === 100);
    } else {
      self::ensure_scheduled();
    }
  }

  public static function pause($network = false)
  {
    // Deactivation leaves accepted rows, identities and health intact for reactivation.
    if ($network && is_multisite()) {
      $offset = 0;
      do {
        $sites = get_sites(array('network_id' => self::network(), 'fields' => 'ids', 'number' => 100, 'offset' => $offset));
        foreach ($sites as $blog_id) {
          switch_to_blog($blog_id);
          wp_clear_scheduled_hook(self::HOOK);
          restore_current_blog();
        }
        $offset += count($sites);
      } while (count($sites) === 100);
    } else {
      wp_clear_scheduled_hook(self::HOOK);
    }
    airefs_tracking_clear_rate_limits($network);
  }

  private static function identity($options)
  {
    $binding = self::binding($options);
    if ((int) get_option('airefs_queue_identity_retry', 0) > time()) {
      return false;
    }
    self::status('last_attempt', time());
    $response = wp_remote_get($options['api_endpoint'] . '/health', array(
      'timeout' => 5, 'redirection' => 0, 'limit_response_size' => 2048,
      'headers' => array('Authorization' => 'Bearer ' . $options['access_token'], 'User-Agent' => AIREFS_TRACKING_UA),
    ));
    $body = is_wp_error($response) ? null : json_decode(wp_remote_retrieve_body($response), true);
    if (is_wp_error($response) || wp_remote_retrieve_response_code($response) !== 200 || !is_array($body) || !isset($body['delivery_protocol'], $body['site_id']) || $body['delivery_protocol'] !== 1 || !is_string($body['site_id']) || $body['site_id'] === '' || strlen($body['site_id']) > 191) {
      self::status('error', 'Destination identity could not be verified. Saved events are retained.');
      self::status('identity_retry', time() + 60);
      return false;
    }
    $identity = array('binding' => $binding, 'site_id' => $body['site_id'], 'endpoint' => $options['api_endpoint']);
    if (isset($body['batch_delivery']) && is_array($body['batch_delivery'])) {
      $batch = $body['batch_delivery'];
      if (isset($batch['protocol'], $batch['max_events'], $batch['max_bytes']) && $batch['protocol'] === 1 && is_int($batch['max_events']) && $batch['max_events'] > 0 && is_int($batch['max_bytes']) && $batch['max_bytes'] > 0) {
        $identity['batch_delivery'] = array('max_events' => min(100, $batch['max_events']), 'max_bytes' => min(524288, $batch['max_bytes']));
      }
    }
    self::status('identity', $identity);
    self::status('identity_retry', 0);
    return $identity;
  }

  private static function retry($row, $claim, $error, $blocked, $retry_after = 0, $retry_individually = false)
  {
    global $wpdb;
    $attempts = (int) $row->attempts + 1;
    $delay = $blocked ? 3600 : min(3600, 5 * pow(2, min($attempts - 1, 10)));
    $delay = max(min(3600, $delay + wp_rand(0, max(1, (int) ($delay / 4)))), min(86400, $retry_after));
    $values = array('attempts' => $attempts, 'next_attempt' => time() + $delay, 'claim' => '', 'claim_until' => 0, 'blocked' => $blocked ? 1 : 0, 'last_error' => $error);
    if ($retry_individually) {
      $values['retry_individually'] = 1;
    }
    $wpdb->update(self::table(), $values, array('id' => $row->id, 'claim' => $claim));
    self::status('error', $error);
  }

  private static function release($rows, $claim)
  {
    global $wpdb;
    foreach ($rows as $row) {
      $wpdb->update(self::table(), array('claim' => '', 'claim_until' => 0), array('id' => $row->id, 'claim' => $claim));
    }
  }

  /** Reject the whole receipt if its IDs or result shapes cannot be trusted. */
  private static function batch_results($ack, $rows, $site_id)
  {
    if (!is_object($ack) || !isset($ack->success, $ack->delivery_protocol, $ack->site_id, $ack->results) || $ack->success !== true || $ack->delivery_protocol !== 1 || $ack->site_id !== $site_id || !is_array($ack->results)) {
      return false;
    }
    $expected = array();
    foreach ($rows as $row) {
      $expected[$row->event_id] = true;
    }
    $results = array();
    foreach ($ack->results as $result) {
      if (!is_object($result) || !isset($result->event_id, $result->success) || !is_string($result->event_id) || !isset($expected[$result->event_id]) || isset($results[$result->event_id]) || !is_bool($result->success)) {
        return false;
      }
      if (property_exists($result, 'retry_individually') && !is_bool($result->retry_individually)) {
        return false;
      }
      if ($result->success) {
        if (property_exists($result, 'error') || property_exists($result, 'retryable') || property_exists($result, 'retry_individually')) {
          return false;
        }
      } elseif (!isset($result->error, $result->retryable) || !is_string($result->error) || $result->error === '' || !is_bool($result->retryable)) {
        return false;
      }
      $results[$result->event_id] = $result;
    }
    return $results;
  }

  private static function deliver($rows, $claim, $body, $batch, $options, $site_id, $started)
  {
    global $wpdb;
    self::status('last_attempt', time());
    $response = wp_remote_post($options['api_endpoint'] . ($batch ? '/v1/events/batch' : '/v1/events'), array(
      'timeout' => min(15, 20 - (microtime(true) - $started)), 'redirection' => 0, 'blocking' => true, 'limit_response_size' => $batch ? 65536 : 2048,
      'headers' => array('Authorization' => 'Bearer ' . $options['access_token'], 'Content-Type' => 'application/json', 'User-Agent' => AIREFS_TRACKING_UA),
      'body' => $body, 'data_format' => 'body',
    ));
    $code = is_wp_error($response) ? 0 : wp_remote_retrieve_response_code($response);
    if ($batch && in_array($code, array(404, 405), true)) {
      self::release($rows, $claim);
      return false;
    }
    $ack = is_wp_error($response) ? null : json_decode(wp_remote_retrieve_body($response), !$batch);
    $results = $batch && $code === 200 ? self::batch_results($ack, $rows, $site_id) : false;
    $error = $code === 0 ? 'Delivery connection failed; retry pending.' : (($code >= 200 && $code < 300) ? 'API did not confirm this event UUID and delivery protocol; retry pending.' : 'Delivery returned HTTP ' . $code . '; saved event retained.');
    $retry_header = is_wp_error($response) ? '' : wp_remote_retrieve_header($response, 'retry-after');
    $retry_after = is_numeric($retry_header) ? (int) $retry_header : max(0, (int) strtotime($retry_header) - time());
    foreach ($rows as $row) {
      $result = $results !== false && isset($results[$row->event_id]) ? $results[$row->event_id] : null;
      $success = $batch ? ($result && $result->success === true) : ($code === 200 && is_array($ack) && isset($ack['success'], $ack['delivery_protocol'], $ack['event_id'], $ack['site_id']) && $ack['success'] === true && $ack['delivery_protocol'] === 1 && $ack['event_id'] === $row->event_id && $ack['site_id'] === $site_id);
      if ($success) {
        if ($wpdb->delete(self::table(), array('id' => $row->id, 'claim' => $claim)) === 1) {
          self::status('last_success', time());
          self::status('error', '');
        }
      } elseif ($result) {
        self::retry($row, $claim, wp_html_excerpt('API rejected this event: ' . sanitize_text_field($result->error), 255, ''), !$result->retryable, $retry_after, !empty($result->retry_individually));
      } else {
        self::retry($row, $claim, $error, $code >= 400 && $code < 500 && !in_array($code, array(408, 425, 429), true), $retry_after);
      }
    }
    return true;
  }

  public static function flush()
  {
    global $wpdb;
    if (!self::install() || !self::enabled_options()) {
      return;
    }
    $lock_name = '_airefs_worker_' . get_current_blog_id();
    $lease = self::lock($lock_name, 60);
    if (!$lease) {
      return;
    }
    self::status('scheduler_error', '');
    $started = microtime(true);
    self::migrate_legacy();
    $options = self::enabled_options();
    $identity = $options ? self::identity($options) : false;
    if (!$identity) {
      self::unlock($lock_name, $lease);
      self::ensure_scheduled();
      return;
    }
    $site_id = $identity['site_id'];
    $batch_limits = isset($identity['batch_delivery']) ? $identity['batch_delivery'] : false;
    $table = self::table();
    $options_table = self::options_table();
    $binding = self::binding($options);
    // Only the original credential may establish identity for previously unverified rows.
    $wpdb->query($wpdb->prepare("UPDATE $table SET site_id = %s, endpoint = %s WHERE network_id = %d AND blog_id = %d AND binding = %s AND site_id = ''", $site_id, $options['api_endpoint'], self::network(), get_current_blog_id(), $binding));
    $wpdb->query($wpdb->prepare("UPDATE $table SET blocked = 1, last_error = %s, next_attempt = %d WHERE network_id = %d AND blog_id = %d AND (site_id <> %s OR endpoint <> %s)", 'Saved event belongs to a different or unverified destination. Restore its original connection to resume it.', time() + 3600, self::network(), get_current_blog_id(), $site_id, $options['api_endpoint']));
    $wpdb->query($wpdb->prepare("UPDATE $table SET blocked = 0, next_attempt = %d, last_error = '' WHERE network_id = %d AND blog_id = %d AND site_id = %s AND endpoint = %s AND blocked = 1 AND last_error LIKE %s", time(), self::network(), get_current_blog_id(), $site_id, $options['api_endpoint'], 'Saved event belongs%'));
    $sent = 0;
    while ($sent < 200 && microtime(true) - $started <= 10) {
      $first = $wpdb->get_row($wpdb->prepare("SELECT * FROM $table WHERE network_id = %d AND blog_id = %d AND site_id = %s AND endpoint = %s AND next_attempt <= %d AND claim_until <= %d ORDER BY next_attempt, id LIMIT 1", self::network(), get_current_blog_id(), $site_id, $options['api_endpoint'], time(), time()));
      if (!$first) {
        break;
      }
      $batch = $batch_limits && !$first->retry_individually;
      $candidates = $batch ? $wpdb->get_results($wpdb->prepare("SELECT * FROM $table WHERE network_id = %d AND blog_id = %d AND site_id = %s AND endpoint = %s AND retry_individually = 0 AND next_attempt <= %d AND claim_until <= %d ORDER BY next_attempt, id LIMIT %d", self::network(), get_current_blog_id(), $site_id, $options['api_endpoint'], time(), time(), min($batch_limits['max_events'], 200 - $sent))) : array($first);
      $claim = wp_generate_uuid4();
      $rows = array();
      $events = array();
      $body = '';
      foreach ($candidates as $row) {
        $payload = json_decode($row->payload, true);
        $valid_uuid = preg_match('/^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i', $row->event_id);
        if (!is_array($payload) || !$valid_uuid) {
          $claimed = $wpdb->query($wpdb->prepare("UPDATE $table SET claim = %s, claim_until = %d WHERE id = %d AND claim_until <= %d", $claim, time() + 60, $row->id, time()));
          if ($claimed === 1) {
            self::retry($row, $claim, $valid_uuid ? 'Stored event is unreadable and has been retained for recovery.' : 'Stored event UUID is invalid and has been retained for recovery.', true);
            $sent++;
          }
          continue;
        }
        $payload['event_id'] = $row->event_id;
        $payload['delivery_protocol'] = 1;
        $payload['expected_site_id'] = $site_id;
        $proposed_events = array_merge($events, array($payload));
        $proposed_body = wp_json_encode($batch ? array('delivery_protocol' => 1, 'expected_site_id' => $site_id, 'events' => $proposed_events) : $payload);
        if ($batch && is_string($proposed_body) && strlen($proposed_body) > $batch_limits['max_bytes']) {
          if ($rows) {
            break;
          }
          // An accepted event may exceed a receiver's smaller advertised batch budget.
          $batch = false;
          $proposed_body = wp_json_encode($payload);
        }
        $claimed = $wpdb->query($wpdb->prepare("UPDATE $table SET claim = %s, claim_until = %d WHERE id = %d AND claim_until <= %d", $claim, time() + 60, $row->id, time()));
        if ($claimed !== 1) {
          continue;
        }
        $sent++;
        if (!is_string($proposed_body)) {
          self::retry($row, $claim, 'Stored event could not be encoded and has been retained for recovery.', true);
          continue;
        }
        $rows[] = $row;
        $events[] = $payload;
        $body = $proposed_body;
        if (!$batch) {
          break;
        }
      }
      if (!$rows) {
        continue;
      }
      $current = self::enabled_options();
      $worker = $wpdb->get_var($wpdb->prepare("SELECT option_value FROM $options_table WHERE option_name = %s", $lock_name));
      if (!$current || !hash_equals(self::binding($current), $binding) || $worker !== $lease || (int) $lease < time() || microtime(true) - $started >= 20) {
        self::release($rows, $claim);
        break;
      }
      if (!self::deliver($rows, $claim, $body, $batch, $options, $site_id, $started)) {
        $batch_limits = false;
      }
    }
    self::$continued = true;
    self::unlock($lock_name, $lease);
    self::ensure_scheduled();
  }

  public static function health()
  {
    global $wpdb;
    $table = self::table();
    $storage_ready = self::install();
    $row = $storage_ready ? $wpdb->get_row($wpdb->prepare("SELECT COUNT(*) AS count, COALESCE(SUM(payload_bytes), 0) AS bytes, MIN(created_at) AS oldest, MIN(next_attempt) AS next_attempt, COALESCE(SUM(blocked), 0) AS blocked, COALESCE(SUM(next_attempt <= %d), 0) AS due FROM $table WHERE network_id = %d AND blog_id = %d", time(), self::network(), get_current_blog_id())) : null;
    $pending_error = $row ? $wpdb->get_var($wpdb->prepare("SELECT last_error FROM $table WHERE network_id = %d AND blog_id = %d AND last_error <> '' ORDER BY blocked DESC, id LIMIT 1", self::network(), get_current_blog_id())) : '';
    $options_table = self::options_table();
    $rejected = $wpdb->get_var($wpdb->prepare("SELECT option_value FROM $options_table WHERE option_name = %s", '_airefs_queue_rejected_n' . self::network() . '_b' . get_current_blog_id()));
    $legacy = $wpdb->get_row($wpdb->prepare("SELECT COUNT(*) AS count, COALESCE(SUM(OCTET_LENGTH(option_value)), 0) AS bytes, MIN(CAST(SUBSTRING_INDEX(SUBSTRING_INDEX(option_name, '_', -2), '_', 1) AS UNSIGNED)) AS oldest FROM $options_table WHERE option_name LIKE %s", $wpdb->esc_like(self::prefix()) . '%'));
    $legacy_blocked = $legacy && (int) get_option('airefs_queue_legacy_retry', 0) > time() ? (int) $legacy->count : 0;
    if ($legacy_blocked && !$pending_error) {
      $pending_error = 'Legacy events could not be migrated; their original data is retained for recovery.';
    }
    $network_bytes = $row ? (int) $wpdb->get_var($wpdb->prepare("SELECT COALESCE(SUM(payload_bytes), 0) FROM $table WHERE network_id = %d", self::network())) : 0;
    $network_legacy_bytes = (int) $wpdb->get_var($wpdb->prepare("SELECT COALESCE(SUM(OCTET_LENGTH(option_value) + 512), 0) FROM $options_table WHERE option_name LIKE %s", $wpdb->esc_like(self::PREFIX . 'n' . self::network() . '_') . '%'));
    $oldest = array_filter(array($row ? (int) $row->oldest : 0, $legacy ? (int) $legacy->oldest : 0));
    return array(
      'storage_ready' => $storage_ready,
      'storage_error' => get_site_option('airefs_queue_storage_error', ''),
      'storage_retry_at' => (int) get_site_option('airefs_queue_storage_retry_at', 0),
      'count' => ($row ? (int) $row->count : 0) + ($legacy ? (int) $legacy->count : 0),
      'bytes' => $network_bytes + $network_legacy_bytes,
      'site_bytes' => ($row ? (int) $row->bytes : 0) + ($legacy ? (int) $legacy->bytes + (int) $legacy->count * 512 : 0),
      'oldest' => $oldest ? min($oldest) : 0, 'next_attempt' => $row ? (int) $row->next_attempt : 0,
      'due' => $row ? (int) $row->due : 0, 'blocked' => ($row ? (int) $row->blocked : 0) + $legacy_blocked,
      'last_attempt' => (int) get_option('airefs_queue_last_attempt', 0), 'last_success' => (int) get_option('airefs_queue_last_success', 0),
      'error' => $pending_error ? $pending_error : get_option('airefs_queue_error', ''), 'scheduler_error' => get_option('airefs_queue_scheduler_error', ''),
      'last_rejected' => (int) get_option('airefs_queue_last_rejected', 0), 'rejected_reason' => get_option('airefs_queue_rejected_reason', ''),
      'rejected' => (int) $rejected, 'capacity_bytes' => self::capacity_bytes(),
      'paused' => !self::enabled_options(), 'legacy_count' => $legacy ? (int) $legacy->count : 0,
    );
  }
}

AirefsTrackingQueue::init();