<?php
/**
 * Durable notification job storage.
 *
 * @package SubmissionInbox
 */

namespace SUBIN\Notifications;

if ( ! defined( 'ABSPATH' ) ) {
	exit;
}

/**
 * Stores and leases background email jobs.
 */
class NotificationJobRepository {
	const STATUS_PENDING    = 'pending';
	const STATUS_PROCESSING = 'processing';
	const STATUS_SENT       = 'sent';
	const STATUS_FAILED     = 'failed';

	const MAX_ATTEMPTS = 6;
	const MAX_EXHAUSTED_RECOVERY_BATCH = 50;
	const MAX_REDACTION_BATCH = 10;
	const REDACTED_PAYLOAD = '{}';

	/**
	 * Enqueue a notification once.
	 *
	 * @param string              $idempotency_key Stable unique key.
	 * @param int                 $lead_id Lead id, if saved.
	 * @param string              $email_type Email type.
	 * @param array<string,mixed> $payload Optional payload.
	 * @return int|\WP_Error
	 */
	public function enqueue( $idempotency_key, $lead_id, $email_type, array $payload = array() ) {
		return $this->enqueue_at( $idempotency_key, $lead_id, $email_type, $payload, '', 0 );
	}

	/**
	 * Reactivate the latest durable job for a saved lead without creating duplicates.
	 *
	 * Pending and processing jobs are returned as-is. A failed job is reset only
	 * when it is still the latest job for the requested lead and email type.
	 *
	 * @param int    $lead_id Saved lead id.
	 * @param string $email_type Notification type.
	 * @return int|\WP_Error
	 */
	public function retry_saved_lead( $lead_id, $email_type ) {
		global $wpdb;

		$lead_id    = absint( $lead_id );
		$email_type = sanitize_key( (string) $email_type );
		if ( $lead_id <= 0 || '' === $email_type ) {
			return new \WP_Error( 'subin_notification_retry_invalid', __( 'The notification retry request is invalid.', 'submission-inbox' ) );
		}

		$table = $this->table_name();
		$job   = $wpdb->get_row(
			$wpdb->prepare(
				"SELECT id, status FROM {$table} WHERE lead_id = %d AND email_type = %s ORDER BY id DESC LIMIT 1",
				$lead_id,
				$email_type
			)
		); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.InterpolatedNotPrepared

		if ( ! $job ) {
			return $this->enqueue( 'manual_retry_lead_' . $lead_id . '_' . $email_type, $lead_id, $email_type, array() );
		}

		$job_id = absint( $job->id ?? 0 );
		$status = sanitize_key( $job->status ?? '' );
		if ( in_array( $status, array( self::STATUS_PENDING, self::STATUS_PROCESSING ), true ) ) {
			return $job_id;
		}
		if ( self::STATUS_SENT === $status ) {
			return new \WP_Error( 'subin_notification_already_sent', __( 'The latest notification has already been sent.', 'submission-inbox' ) );
		}
		if ( self::STATUS_FAILED !== $status ) {
			return new \WP_Error( 'subin_notification_retry_unavailable', __( 'The latest notification cannot be retried.', 'submission-inbox' ) );
		}

		$updated = $wpdb->update(
			$table,
			array(
				'status'       => self::STATUS_PENDING,
				'attempts'     => 0,
				'available_at' => current_time( 'mysql' ),
				'retry_of_event_id' => null,
				'lease_token'  => null,
				'lease_until'  => null,
				'last_error'   => '',
				'updated_at'   => current_time( 'mysql' ),
			),
			array(
				'id'     => $job_id,
				'status' => self::STATUS_FAILED,
			),
			array( '%s', '%d', '%s', '%s', '%s', '%s', '%s', '%s' ),
			array( '%d', '%s' )
		); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching

		if ( false === $updated ) {
			return new \WP_Error( 'subin_notification_retry_failed', __( 'The notification could not be returned to the delivery queue.', 'submission-inbox' ) );
		}
		if ( 0 === $updated ) {
			$current_status = $wpdb->get_var(
				$wpdb->prepare( "SELECT status FROM {$table} WHERE id = %d", $job_id )
			); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.InterpolatedNotPrepared
			if ( ! in_array( sanitize_key( (string) $current_status ), array( self::STATUS_PENDING, self::STATUS_PROCESSING ), true ) ) {
				return new \WP_Error( 'subin_notification_retry_conflict', __( 'The notification changed before it could be retried.', 'submission-inbox' ) );
			}
		}

		return $job_id;
	}

	/**
	 * Describe whether one exact immutable job can be retried from Email Log.
	 *
	 * Unlike retry_saved_lead(), this method never falls back to rebuilding a
	 * message from current lead or form settings. That distinction is important
	 * for an audit log: a retry offered beside a historical event must use the
	 * frozen payload which produced that event.
	 *
	 * @param int    $job_id Exact notification job id stored on the mail event.
	 * @param int    $lead_id Expected saved lead id.
	 * @param string $email_type Expected immutable envelope type.
	 * @param int    $event_id Exact failed mail event which initiated the retry.
	 * @return array{retryable:bool,queued:bool,reason:string,job:object|null,event:object|null}
	 */
	public function get_event_retry_state( $job_id, $lead_id, $email_type, $event_id ) {
		global $wpdb;

		$job_id     = absint( $job_id );
		$lead_id    = absint( $lead_id );
		$event_id   = absint( $event_id );
		$email_type = sanitize_key( (string) $email_type );
		$empty      = array(
			'retryable' => false,
			'queued'    => false,
			'reason'    => 'invalid',
			'job'       => null,
			'event'     => null,
		);
		if ( $job_id <= 0 || $lead_id <= 0 || $event_id <= 0 || '' === $email_type ) {
			return $empty;
		}

		$table = $this->table_name();
		$job   = $wpdb->get_row(
			$wpdb->prepare(
				"SELECT id, lead_id, email_type, status, payload_json, payload_version, payload_hash, payload_redacted_at
				FROM {$table}
				WHERE id = %d AND lead_id = %d AND email_type = %s
				LIMIT 1",
				$job_id,
				$lead_id,
				$email_type
			)
		); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.InterpolatedNotPrepared

		if ( ! $job ) {
			$empty['reason'] = 'not_found';
			return $empty;
		}

		$event = $wpdb->get_row(
			$wpdb->prepare(
				"SELECT id, job_id, lead_id, status, provider_status, provider_message_id, provider_evidence_source, provider_evidence_at
				FROM {$wpdb->prefix}subin_mail_events
				WHERE id = %d AND job_id = %d AND lead_id = %d AND status = %s
				LIMIT 1",
				$event_id,
				$job_id,
				$lead_id,
				'failed'
			)
		); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.InterpolatedNotPrepared
		if ( ! $event ) {
			$empty['reason'] = 'event_mismatch';
			$empty['job']    = $job;
			return $empty;
		}
		$empty['event'] = $event;
		if ( $this->event_has_provider_evidence( $event ) ) {
			$empty['reason'] = 'provider_evidence_present';
			$empty['job']    = $job;
			return $empty;
		}

		$status = sanitize_key( (string) ( $job->status ?? '' ) );
		if ( in_array( $status, array( self::STATUS_PENDING, self::STATUS_PROCESSING ), true ) ) {
			return array(
				'retryable' => false,
				'queued'    => true,
				'reason'    => 'already_queued',
				'job'       => $job,
				'event'     => $event,
			);
		}
		if ( self::STATUS_FAILED !== $status ) {
			$empty['reason'] = self::STATUS_SENT === $status ? 'already_sent' : 'not_failed';
			$empty['job']    = $job;
			return $empty;
		}
		if (
			! empty( $job->payload_redacted_at )
			|| absint( $job->payload_version ?? 0 ) <= 0
			|| ! is_string( $job->payload_json ?? null )
			|| '' === trim( (string) $job->payload_json )
			|| self::REDACTED_PAYLOAD === trim( (string) $job->payload_json )
			|| ! preg_match( '/^[a-f0-9]{64}$/', strtolower( trim( (string) ( $job->payload_hash ?? '' ) ) ) )
		) {
			$empty['reason'] = 'payload_unavailable';
			$empty['job']    = $job;
			return $empty;
		}

		return array(
			'retryable' => true,
			'queued'    => false,
			'reason'    => '',
			'job'       => $job,
			'event'     => $event,
		);
	}

	/**
	 * Return an exact failed immutable job to the queue using a CAS boundary.
	 *
	 * The original mail event remains immutable. A worker run creates another
	 * event carrying the same job id, which gives Email Log an explicit attempt
	 * history without pretending the first failure never happened.
	 *
	 * @param int    $job_id Exact notification job id.
	 * @param int    $lead_id Expected saved lead id.
	 * @param string $email_type Expected immutable envelope type.
	 * @param int    $event_id Exact failed mail event which initiated the retry.
	 * @return int|\WP_Error
	 */
	public function retry_event_job( $job_id, $lead_id, $email_type, $event_id ) {
		global $wpdb;

		$event_id = absint( $event_id );
		$state    = $this->get_event_retry_state( $job_id, $lead_id, $email_type, $event_id );
		if ( ! empty( $state['queued'] ) && ! empty( $state['job']->id ) ) {
			return absint( $state['job']->id );
		}
		if ( empty( $state['retryable'] ) || empty( $state['job']->id ) ) {
			$reason = sanitize_key( (string) ( $state['reason'] ?? 'unavailable' ) );
			return new \WP_Error(
				'subin_notification_retry_' . ( '' !== $reason ? $reason : 'unavailable' ),
				'payload_unavailable' === $reason
					? __( 'The immutable email snapshot has expired and cannot be retried safely.', 'submission-inbox' )
					: __( 'This email attempt cannot be retried safely.', 'submission-inbox' )
			);
		}

		$job        = $state['job'];
		$job_id     = absint( $job->id );
		$lead_id    = absint( $lead_id );
		$email_type = sanitize_key( (string) $email_type );
		$payload_version = absint( $job->payload_version ?? 0 );
		$payload_hash    = strtolower( trim( (string) ( $job->payload_hash ?? '' ) ) );
		$payload_json    = (string) ( $job->payload_json ?? '' );
		$updated    = $wpdb->update(
			$this->table_name(),
			array(
				'status'       => self::STATUS_PENDING,
				'attempts'     => 0,
				'available_at' => current_time( 'mysql' ),
				'retry_of_event_id' => $event_id,
				'lease_token'  => null,
				'lease_until'  => null,
				'last_error'   => '',
				'updated_at'   => current_time( 'mysql' ),
			),
			array(
				'id'         => $job_id,
				'lead_id'    => $lead_id,
				'email_type' => $email_type,
				'status'     => self::STATUS_FAILED,
				'payload_version' => $payload_version,
				'payload_hash' => $payload_hash,
				'payload_json' => $payload_json,
				'payload_redacted_at' => null,
			),
			array( '%s', '%d', '%s', '%d', '%s', '%s', '%s', '%s' ),
			array( '%d', '%d', '%s', '%s', '%d', '%s', '%s', '%s' )
		); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching

		if ( false === $updated ) {
			return new \WP_Error( 'subin_notification_retry_failed', __( 'The email could not be returned to the delivery queue.', 'submission-inbox' ) );
		}
		if ( 0 === $updated ) {
			$current = $wpdb->get_row(
				$wpdb->prepare(
					"SELECT status, payload_json, payload_version, payload_hash, payload_redacted_at, retry_of_event_id
					FROM {$this->table_name()} WHERE id = %d LIMIT 1",
					$job_id
				)
			); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.InterpolatedNotPrepared
			$current_status = sanitize_key( (string) ( $current->status ?? '' ) );
			if (
				in_array( $current_status, array( self::STATUS_PENDING, self::STATUS_PROCESSING ), true )
				&& $event_id === absint( $current->retry_of_event_id ?? 0 )
			) {
				return $job_id;
			}
			if (
				! empty( $current->payload_redacted_at )
				|| self::REDACTED_PAYLOAD === trim( (string) ( $current->payload_json ?? '' ) )
				|| $payload_version !== absint( $current->payload_version ?? 0 )
				|| ! hash_equals( $payload_hash, strtolower( trim( (string) ( $current->payload_hash ?? '' ) ) ) )
				|| ! hash_equals( $payload_json, (string) ( $current->payload_json ?? '' ) )
			) {
				return new \WP_Error( 'subin_notification_retry_payload_unavailable', __( 'The immutable email snapshot changed or expired before it could be retried safely.', 'submission-inbox' ) );
			}

			return new \WP_Error( 'subin_notification_retry_conflict', __( 'The email attempt changed before it could be retried.', 'submission-inbox' ) );
		}

		do_action( 'subin_notification_event_retry_queued', $job_id, $lead_id, $email_type, $event_id );

		return $job_id;
	}

	/**
	 * Whether an authenticated provider has already reported this message.
	 *
	 * Any provider evidence makes an apparently failed local handoff ambiguous;
	 * retrying it could send a duplicate or repeat a confirmed bounce/complaint.
	 *
	 * @param object $event Mail event row.
	 * @return bool
	 */
	private function event_has_provider_evidence( $event ) {
		$status     = sanitize_key( (string) ( $event->provider_status ?? '' ) );
		$message_id = sanitize_text_field( (string) ( $event->provider_message_id ?? '' ) );
		$source     = sanitize_key( (string) ( $event->provider_evidence_source ?? '' ) );
		$occurred   = sanitize_text_field( (string) ( $event->provider_evidence_at ?? '' ) );

		return in_array( $status, array( 'accepted', 'queued', 'delivered', 'bounced', 'deferred', 'dropped', 'complained', 'opened' ), true )
			&& '' !== $message_id
			&& '' !== $source
			&& 1 === preg_match( '/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$/', $occurred );
	}

	/**
	 * Enqueue a versioned notification with explicit queue eligibility.
	 *
	 * @param string              $idempotency_key Stable unique key.
	 * @param int                 $lead_id Lead id, if saved.
	 * @param string              $email_type Email type.
	 * @param array<string,mixed> $payload Immutable payload.
	 * @param string              $available_at Earliest local delivery datetime.
	 * @param int                 $payload_version Immutable payload codec version.
	 * @return int|\WP_Error
	 */
	public function enqueue_at( $idempotency_key, $lead_id, $email_type, array $payload, $available_at, $payload_version ) {
		global $wpdb;

		$idempotency_key = sanitize_text_field( (string) $idempotency_key );
		$email_type      = sanitize_key( (string) $email_type );
		$encoded         = wp_json_encode( $payload );
		$payload_hash    = is_string( $encoded ) ? $this->payload_hash( $encoded ) : '';
		$payload_version = absint( $payload_version );
		$available_at    = $this->normalize_available_at( $available_at );

		if ( '' === $idempotency_key || '' === $email_type || ! is_string( $encoded ) || '' === $available_at || $payload_version > 65535 ) {
			return new \WP_Error(
				'subin_invalid_notification_job',
				__( 'The notification job could not be created.', 'submission-inbox' )
			);
		}

		$table = $this->table_name();
		$now   = current_time( 'mysql' );
		$query = $wpdb->prepare(
			"INSERT INTO {$table}
				(idempotency_key, lead_id, email_type, payload_json, payload_version, payload_hash, status, attempts, available_at, lease_token, lease_until, last_error, created_at, updated_at, sent_at)
			VALUES (%s, %d, %s, %s, %d, %s, %s, 0, %s, NULL, NULL, '', %s, %s, NULL)
			ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id)",
			$idempotency_key,
			absint( $lead_id ),
			$email_type,
			$encoded,
			$payload_version,
			$payload_hash,
			self::STATUS_PENDING,
			$available_at,
			$now,
			$now
		);

		$inserted = $wpdb->query( $query ); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.NotPrepared
		if ( false === $inserted ) {
			return new \WP_Error(
				'subin_notification_job_insert_failed',
				__( 'The notification job could not be stored.', 'submission-inbox' ),
				array( 'db_error' => sanitize_text_field( (string) $wpdb->last_error ) )
			);
		}

		$wpdb->last_error = '';
		$row = $wpdb->get_row(
			$wpdb->prepare(
				"SELECT id, idempotency_key, lead_id, email_type, payload_json, payload_version, payload_hash, payload_redacted_at, available_at
				FROM {$table}
				WHERE idempotency_key = %s
				LIMIT 1",
				$idempotency_key
			)
		); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.InterpolatedNotPrepared
		if ( '' !== (string) $wpdb->last_error ) {
			return new \WP_Error(
				'subin_notification_job_read_failed',
				__( 'The notification job could not be verified.', 'submission-inbox' ),
				array( 'db_error' => sanitize_text_field( (string) $wpdb->last_error ) )
			);
		}
		if (
			! $row
			|| absint( $row->id ?? 0 ) <= 0
			|| $idempotency_key !== (string) ( $row->idempotency_key ?? '' )
			|| absint( $lead_id ) !== absint( $row->lead_id ?? 0 )
			|| $email_type !== (string) ( $row->email_type ?? '' )
			|| $payload_version !== absint( $row->payload_version ?? 0 )
			|| ( $payload_version > 0 && $available_at !== (string) ( $row->available_at ?? '' ) )
			|| ( $payload_version > 0 && ! $this->payload_matches( $row, $encoded, $payload_hash ) )
		) {
			return new \WP_Error(
				'subin_notification_job_conflict',
				__( 'The notification job conflicts with an existing delivery.', 'submission-inbox' )
			);
		}

		return absint( $row->id );
	}

	/**
	 * Enqueue one transaction-bound notification for a submission receipt.
	 *
	 * The receipt/type pair is a database-enforced idempotency boundary. An
	 * existing row is returned only when every immutable input still matches.
	 *
	 * @param int                 $receipt_id Submission receipt id.
	 * @param int                 $lead_id Lead id, if saved.
	 * @param string              $email_type Email type.
	 * @param array<string,mixed> $payload Notification payload.
	 * @return int|\WP_Error
	 */
	public function enqueue_for_receipt( $receipt_id, $lead_id, $email_type, array $payload = array() ) {
		global $wpdb;

		$receipt_id      = absint( $receipt_id );
		$lead_id         = absint( $lead_id );
		$email_type      = substr( sanitize_key( (string) $email_type ), 0, 50 );
		$encoded         = wp_json_encode( $payload );
		$idempotency_key = 'receipt_' . $receipt_id . '_' . $email_type;
		$payload_version = 0;
		$payload_hash    = is_string( $encoded ) ? $this->payload_hash( $encoded ) : '';
		$available_at    = is_array( $payload['envelope']['meta'] ?? null )
			? (string) ( $payload['envelope']['meta']['scheduled_at'] ?? '' )
			: '';
		$available_at    = $this->normalize_available_at( $available_at );

		if ( $receipt_id <= 0 || '' === $email_type || ! is_string( $encoded ) || '' === $available_at ) {
			return new \WP_Error(
				'subin_invalid_notification_job',
				__( 'The notification job could not be created.', 'submission-inbox' )
			);
		}
		if ( array_key_exists( 'envelope', $payload ) ) {
			$validated = NotificationEnvelope::decode( $payload, $email_type, $receipt_id, $lead_id );
			if ( is_wp_error( $validated ) ) {
				return $validated;
			}
			$payload_version = NotificationEnvelope::VERSION;
		}

		$table = $this->table_name();
		$now   = current_time( 'mysql' );
		$query = $wpdb->prepare(
			"INSERT INTO {$table}
				(receipt_id, idempotency_key, lead_id, email_type, payload_json, payload_version, payload_hash, status, attempts, available_at, lease_token, lease_until, last_error, created_at, updated_at, sent_at)
			VALUES (%d, %s, NULLIF(%d, 0), %s, %s, %d, %s, %s, 0, %s, NULL, NULL, '', %s, %s, NULL)
			ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id)",
			$receipt_id,
			$idempotency_key,
			$lead_id,
			$email_type,
			$encoded,
			$payload_version,
			$payload_hash,
			self::STATUS_PENDING,
			$available_at,
			$now,
			$now
		);

		$inserted = $wpdb->query( $query ); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.NotPrepared
		if ( false === $inserted ) {
			return new \WP_Error(
				'subin_notification_job_insert_failed',
				__( 'The notification job could not be stored.', 'submission-inbox' ),
				array( 'db_error' => sanitize_text_field( (string) $wpdb->last_error ) )
			);
		}

		$wpdb->last_error = '';
		$row = $wpdb->get_row(
			$wpdb->prepare(
				"SELECT id, receipt_id, idempotency_key, lead_id, email_type, payload_json, payload_version, payload_hash, payload_redacted_at, available_at
				FROM {$table}
				WHERE receipt_id = %d AND email_type = %s
				LIMIT 1",
				$receipt_id,
				$email_type
			)
		); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.InterpolatedNotPrepared

		if ( '' !== (string) $wpdb->last_error ) {
			return new \WP_Error(
				'subin_notification_job_read_failed',
				__( 'The notification job could not be verified.', 'submission-inbox' ),
				array( 'db_error' => sanitize_text_field( (string) $wpdb->last_error ) )
			);
		}

		if (
			! $row
			|| absint( $row->id ?? 0 ) <= 0
			|| $idempotency_key !== (string) ( $row->idempotency_key ?? '' )
			|| $lead_id !== absint( $row->lead_id ?? 0 )
			|| $payload_version !== absint( $row->payload_version ?? 0 )
			|| ( $payload_version > 0 && $available_at !== (string) ( $row->available_at ?? '' ) )
			|| ! $this->payload_matches( $row, $encoded, $payload_hash )
		) {
			return new \WP_Error(
				'subin_notification_job_conflict',
				__( 'The notification job conflicts with an existing receipt notification.', 'submission-inbox' )
			);
		}

		$stored_hash = strtolower( trim( (string) ( $row->payload_hash ?? '' ) ) );
		if (
			empty( $row->payload_redacted_at )
			&& ! hash_equals( $stored_hash, $payload_hash )
		) {
			$backfilled = $wpdb->query(
				$wpdb->prepare(
					"UPDATE {$table}
					SET payload_hash = %s
					WHERE id = %d
						AND payload_redacted_at IS NULL
						AND payload_json = %s",
					$payload_hash,
					absint( $row->id ),
					$encoded
				)
			); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.InterpolatedNotPrepared
			if ( false === $backfilled ) {
				return new \WP_Error(
					'subin_notification_job_hash_backfill_failed',
					__( 'The notification job could not be verified.', 'submission-inbox' ),
					array( 'db_error' => sanitize_text_field( (string) $wpdb->last_error ) )
				);
			}
		}

		return absint( $row->id );
	}

	/**
	 * Atomically lease jobs for one worker.
	 *
	 * @param int $limit Maximum jobs.
	 * @param int $lease_seconds Lease duration.
	 * @return array{token:string,jobs:array<int,object>,exhausted_jobs:array<int,object>}|\WP_Error
	 */
	public function claim_batch( $limit = 10, $lease_seconds = 300 ) {
		global $wpdb;

		$table         = $this->table_name();
		$limit         = max( 1, min( 50, absint( $limit ) ) );
		$lease_seconds = max( 30, min( HOUR_IN_SECONDS, absint( $lease_seconds ) ) );
		$token         = str_replace( '-', '', wp_generate_uuid4() );
		$now           = current_time( 'mysql' );
		$lease_until   = gmdate( 'Y-m-d H:i:s', current_time( 'timestamp' ) + $lease_seconds );

		$exhausted_jobs = $this->recover_exhausted_jobs( $table, $now );
		if ( is_wp_error( $exhausted_jobs ) ) {
			return $exhausted_jobs;
		}

		$updated = $wpdb->query(
			$wpdb->prepare(
				"UPDATE {$table}
				SET status = %s, attempts = attempts + 1, lease_token = %s, lease_until = %s, updated_at = %s
				WHERE attempts < %d
					AND available_at <= %s
					AND (
						status IN (%s, %s)
						OR (status = %s AND (lease_until IS NULL OR lease_until <= %s))
					)
				ORDER BY id ASC
				LIMIT %d",
				self::STATUS_PROCESSING,
				$token,
				$lease_until,
				$now,
				self::MAX_ATTEMPTS,
				$now,
				self::STATUS_PENDING,
				self::STATUS_FAILED,
				self::STATUS_PROCESSING,
				$now,
				$limit
			)
		); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.InterpolatedNotPrepared

		if ( false === $updated ) {
			return new \WP_Error(
				'subin_notification_claim_failed',
				__( 'Notification jobs could not be claimed.', 'submission-inbox' ),
				array( 'db_error' => sanitize_text_field( (string) $wpdb->last_error ) )
			);
		}

		$wpdb->last_error = '';
		$jobs = $wpdb->get_results(
			$wpdb->prepare(
				"SELECT * FROM {$table} WHERE lease_token = %s AND status = %s ORDER BY id ASC",
				$token,
				self::STATUS_PROCESSING
			)
		); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.InterpolatedNotPrepared
		if ( '' !== (string) $wpdb->last_error ) {
			return new \WP_Error(
				'subin_notification_claim_read_failed',
				__( 'Claimed notification jobs could not be read.', 'submission-inbox' ),
				array( 'db_error' => sanitize_text_field( (string) $wpdb->last_error ) )
			);
		}

		return array(
			'token'          => $token,
			'jobs'           => is_array( $jobs ) ? $jobs : array(),
			'exhausted_jobs' => $exhausted_jobs,
		);
	}

	/**
	 * Atomically terminalize expired final leases and return their lead channels.
	 *
	 * @param string $table Notification table.
	 * @param string $now Current database time.
	 * @return array<int,object>|\WP_Error
	 */
	private function recover_exhausted_jobs( $table, $now ) {
		global $wpdb;

		if ( false === $wpdb->query( 'START TRANSACTION' ) ) {
			return new \WP_Error(
				'subin_notification_exhausted_transaction_failed',
				__( 'Exhausted notification jobs could not be recovered.', 'submission-inbox' ),
				array( 'db_error' => sanitize_text_field( (string) $wpdb->last_error ) )
			);
		}

		$wpdb->last_error = '';
		$jobs = $wpdb->get_results(
			$wpdb->prepare(
				"SELECT id, lead_id, email_type, attempts
				FROM {$table}
				WHERE status = %s
					AND attempts >= %d
					AND (lease_until IS NULL OR lease_until <= %s)
				ORDER BY id ASC
				LIMIT %d
				FOR UPDATE",
				self::STATUS_PROCESSING,
				self::MAX_ATTEMPTS,
				$now,
				self::MAX_EXHAUSTED_RECOVERY_BATCH
			)
		); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.InterpolatedNotPrepared
		if ( ! is_array( $jobs ) || '' !== (string) $wpdb->last_error ) {
			$error = new \WP_Error(
				'subin_notification_exhausted_read_failed',
				__( 'Exhausted notification jobs could not be recovered.', 'submission-inbox' ),
				array( 'db_error' => sanitize_text_field( (string) $wpdb->last_error ) )
			);
			$wpdb->query( 'ROLLBACK' );
			return $error;
		}
		if ( empty( $jobs ) ) {
			if ( false === $wpdb->query( 'COMMIT' ) ) {
				$error = new \WP_Error(
					'subin_notification_exhausted_commit_failed',
					__( 'Exhausted notification jobs could not be recovered.', 'submission-inbox' ),
					array( 'db_error' => sanitize_text_field( (string) $wpdb->last_error ) )
				);
				$wpdb->query( 'ROLLBACK' );
				return $error;
			}

			return array();
		}

		$job_ids      = array_values( array_filter( array_map( 'absint', wp_list_pluck( $jobs, 'id' ) ) ) );
		$placeholders = implode( ',', array_fill( 0, count( $job_ids ), '%d' ) );
		$params       = array_merge(
			array(
				self::STATUS_FAILED,
				__( 'The notification worker stopped before acknowledging the final delivery attempt.', 'submission-inbox' ),
				$now,
			),
			$job_ids,
			array( self::STATUS_PROCESSING, self::MAX_ATTEMPTS, $now )
		);
		$updated = $wpdb->query(
			$wpdb->prepare(
				"UPDATE {$table}
				SET status = %s, lease_token = NULL, lease_until = NULL, last_error = %s, updated_at = %s
				WHERE id IN ({$placeholders})
					AND status = %s
					AND attempts >= %d
					AND (lease_until IS NULL OR lease_until <= %s)",
				$params
			)
		); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.DB.PreparedSQLPlaceholders.UnfinishedPrepare
		if ( count( $job_ids ) !== $updated ) {
			$error = new \WP_Error(
				'subin_notification_exhausted_recovery_failed',
				__( 'Exhausted notification jobs could not be recovered.', 'submission-inbox' ),
				array( 'db_error' => sanitize_text_field( (string) $wpdb->last_error ) )
			);
			$wpdb->query( 'ROLLBACK' );
			return $error;
		}
		if ( false === $wpdb->query( 'COMMIT' ) ) {
			$error = new \WP_Error(
				'subin_notification_exhausted_commit_failed',
				__( 'Exhausted notification jobs could not be recovered.', 'submission-inbox' ),
				array( 'db_error' => sanitize_text_field( (string) $wpdb->last_error ) )
			);
			$wpdb->query( 'ROLLBACK' );
			return $error;
		}

		return $jobs;
	}

	/**
	 * Mark a leased job as sent.
	 *
	 * @param int    $job_id Job id.
	 * @param string $lease_token Worker lease token.
	 * @return bool
	 */
	public function mark_sent( $job_id, $lease_token ) {
		global $wpdb;

		$updated = $wpdb->update(
			$this->table_name(),
			array(
				'status'      => self::STATUS_SENT,
				'lease_token' => null,
				'lease_until' => null,
				'last_error'  => '',
				'updated_at'  => current_time( 'mysql' ),
				'sent_at'     => current_time( 'mysql' ),
			),
			array(
				'id'          => absint( $job_id ),
				'lease_token' => sanitize_text_field( (string) $lease_token ),
				'status'      => self::STATUS_PROCESSING,
			),
			array( '%s', '%s', '%s', '%s', '%s', '%s' ),
			array( '%d', '%s', '%s' )
		);

		return false !== $updated && $updated > 0;
	}

	/**
	 * Release a failed job with exponential backoff.
	 *
	 * @param int    $job_id Job id.
	 * @param string $lease_token Worker lease token.
	 * @param string $error Safe error.
	 * @param bool   $retryable Whether another delivery attempt can succeed.
	 * @return bool
	 */
	public function mark_failed( $job_id, $lease_token, $error, $retryable = true ) {
		global $wpdb;

		$table = $this->table_name();
		$job   = $wpdb->get_row(
			$wpdb->prepare(
				"SELECT attempts FROM {$table} WHERE id = %d AND lease_token = %s AND status = %s",
				absint( $job_id ),
				sanitize_text_field( (string) $lease_token ),
				self::STATUS_PROCESSING
			)
		); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.InterpolatedNotPrepared

		if ( ! $job ) {
			return false;
		}

		$attempts     = $retryable ? max( 1, absint( $job->attempts ) ) : self::MAX_ATTEMPTS;
		$delay        = min( HOUR_IN_SECONDS, (int) pow( 2, $attempts ) * MINUTE_IN_SECONDS );
		$available_at = $retryable
			? gmdate( 'Y-m-d H:i:s', current_time( 'timestamp' ) + $delay )
			: current_time( 'mysql' );

		$updated = $wpdb->update(
			$table,
			array(
				'status'       => self::STATUS_FAILED,
				'attempts'     => $attempts,
				'available_at' => $available_at,
				'lease_token'  => null,
				'lease_until'  => null,
				'last_error'   => sanitize_textarea_field( (string) $error ),
				'updated_at'   => current_time( 'mysql' ),
			),
			array(
				'id'          => absint( $job_id ),
				'lease_token' => sanitize_text_field( (string) $lease_token ),
				'status'      => self::STATUS_PROCESSING,
			),
			array( '%s', '%d', '%s', '%s', '%s', '%s', '%s' ),
			array( '%d', '%s', '%s' )
		);

		return false !== $updated && $updated > 0;
	}

	/**
	 * Redact rendered content from terminal jobs while preserving audit metadata.
	 *
	 * Active and retryable rows must keep their immutable payload because a
	 * later worker may still need it for delivery or lost-ack recovery.
	 *
	 * @param string $cutoff Redact terminal rows older than this local datetime.
	 * @param int    $limit Maximum rows.
	 * @return int|\WP_Error
	 */
	public function redact_terminal_payloads( $cutoff, $limit = 250 ) {
		global $wpdb;

		$table  = $this->table_name();
		$cutoff = sanitize_text_field( (string) $cutoff );
		$limit  = max( 1, min( self::MAX_REDACTION_BATCH, absint( $limit ) ) );
		$wpdb->last_error = '';
		$rows = $wpdb->get_results(
			$wpdb->prepare(
				"SELECT id, payload_json, payload_hash
				FROM {$table}
				WHERE payload_redacted_at IS NULL
					AND updated_at < %s
					AND (
						status IN (%s, %s)
						OR (status = %s AND attempts >= %d)
					)
				ORDER BY id ASC
				LIMIT %d",
				$cutoff,
				self::STATUS_SENT,
				'skipped',
				self::STATUS_FAILED,
				self::MAX_ATTEMPTS,
				$limit
			)
		); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.InterpolatedNotPrepared
		if ( '' !== (string) $wpdb->last_error || ! is_array( $rows ) ) {
			return new \WP_Error(
				'subin_notification_payload_retention_read_failed',
				__( 'Terminal notification payloads could not be selected for redaction.', 'submission-inbox' ),
				array( 'db_error' => sanitize_text_field( (string) $wpdb->last_error ) )
			);
		}

		$redacted = 0;
		$now      = current_time( 'mysql' );
		foreach ( $rows as $row ) {
			$encoded = (string) ( $row->payload_json ?? '' );
			$stored_hash = strtolower( trim( (string) ( $row->payload_hash ?? '' ) ) );
			$retained_hash = preg_match( '/^[a-f0-9]{64}$/', $stored_hash )
				? $stored_hash
				: $this->payload_hash( $encoded );
			$updated = $wpdb->query(
				$wpdb->prepare(
					"UPDATE {$table}
					SET payload_json = %s,
						payload_hash = %s,
						payload_redacted_at = %s,
						last_error = ''
					WHERE id = %d
						AND payload_redacted_at IS NULL
						AND updated_at < %s
						AND (
							status IN (%s, %s)
							OR (status = %s AND attempts >= %d)
					)",
					self::REDACTED_PAYLOAD,
					$retained_hash,
					$now,
					absint( $row->id ?? 0 ),
					$cutoff,
					self::STATUS_SENT,
					'skipped',
					self::STATUS_FAILED,
					self::MAX_ATTEMPTS
				)
			); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.InterpolatedNotPrepared
			if ( false === $updated ) {
				return new \WP_Error(
					'subin_notification_payload_retention_update_failed',
					__( 'A terminal notification payload could not be redacted.', 'submission-inbox' ),
					array( 'db_error' => sanitize_text_field( (string) $wpdb->last_error ) )
				);
			}

			$redacted += absint( $updated );
		}

		return $redacted;
	}

	/**
	 * Decode a job payload.
	 *
	 * @param object $job Job row.
	 * @return array<string,mixed>
	 */
	public function payload( $job ) {
		$payload = json_decode( (string) ( $job->payload_json ?? '' ), true );

		return is_array( $payload ) ? $payload : array();
	}

	/**
	 * Strictly decode a queued payload without converting corruption into an empty legacy payload.
	 *
	 * @param object $job Job row.
	 * @return array<string,mixed>|\WP_Error
	 */
	public function decode_payload( $job ) {
		$encoded = (string) ( $job->payload_json ?? '' );
		$payload = json_decode( $encoded, true );

		if ( JSON_ERROR_NONE !== json_last_error() || ! is_array( $payload ) ) {
			return new \WP_Error(
				'subin_notification_payload_invalid',
				__( 'The queued notification payload is invalid.', 'submission-inbox' ),
				array( 'retryable' => false )
			);
		}

		return $payload;
	}

	/**
	 * Compare an immutable payload with a stored row, including redacted rows.
	 *
	 * Legacy rows without a hash can be verified only while their original
	 * payload is still present.
	 *
	 * @param object $row Stored row.
	 * @param string $encoded Expected JSON.
	 * @param string $expected_hash Expected privacy-preserving hash.
	 * @return bool
	 */
	private function payload_matches( $row, $encoded, $expected_hash ) {
		$stored_hash = strtolower( trim( (string) ( $row->payload_hash ?? '' ) ) );
		if ( ! empty( $row->payload_redacted_at ) ) {
			return preg_match( '/^[a-f0-9]{64}$/', $stored_hash )
				&& hash_equals( $stored_hash, $expected_hash );
		}

		return hash_equals( (string) ( $row->payload_json ?? '' ), $encoded );
	}

	/**
	 * Build a stable keyed hash without retaining recoverable payload values.
	 *
	 * @param string $encoded Encoded payload.
	 * @return string
	 */
	private function payload_hash( $encoded ) {
		return hash_hmac(
			'sha256',
			'notification-payload:v1|' . (string) $encoded,
			wp_salt( 'nonce' )
		);
	}

	/**
	 * Normalize an optional local MySQL datetime used by queue eligibility.
	 *
	 * @param mixed $available_at Raw datetime; empty means now.
	 * @return string Empty when invalid.
	 */
	private function normalize_available_at( $available_at ) {
		$available_at = is_scalar( $available_at ) ? trim( (string) $available_at ) : '';
		if ( '' === $available_at ) {
			return current_time( 'mysql' );
		}
		if ( 1 !== preg_match( '/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$/', $available_at ) ) {
			return '';
		}

		$date = \DateTime::createFromFormat( '!Y-m-d H:i:s', $available_at );

		return $date instanceof \DateTime && $date->format( 'Y-m-d H:i:s' ) === $available_at
			? $available_at
			: '';
	}

	/**
	 * Get table name.
	 *
	 * @return string
	 */
	private function table_name() {
		global $wpdb;

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