<?php
/**
 * Sync event repository.
 *
 * @package SubmissionInbox
 */

namespace SUBIN\Sync;

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

// phpcs:disable WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- Repository stores local sync queue rows.

/**
 * Stores local events for future Cloud synchronization.
 */
class SyncEventRepository {
	const STATUS_PENDING             = 'pending';
	const STATUS_PROCESSING          = 'processing';
	const STATUS_SENT                = 'sent';
	const STATUS_FAILED              = 'failed';
	const STATUS_SKIPPED             = 'skipped';
	const DELETION_CONTRACT_VERSION = 1;

	const MAX_ATTEMPTS         = 6;
	const DEFAULT_LEASE_SECONDS = 300;

	/**
	 * Enqueue a sync event.
	 *
	 * @param string              $event_type Event type.
	 * @param string              $object_type Object type.
	 * @param int|string          $object_id Object id.
	 * @param array<string,mixed> $payload Payload.
	 * @return int
	 */
	public function enqueue( $event_type, $object_type, $object_id, array $payload ) {
		$event_type  = $this->sanitize_event_type( $event_type );
		$object_type = sanitize_key( (string) $object_type );
		if ( '.deleted' === substr( $event_type, -8 ) ) {
			return 0;
		}
		if ( 'lead' === $object_type ) {
			return $this->enqueue_existing_object_event( $event_type, $object_type, $object_id, $payload );
		}
		if ( 'mail_event' === $object_type ) {
			return $this->enqueue_existing_object_event( $event_type, $object_type, $object_id, $payload );
		}

		return $this->insert_event( $event_type, $object_type, $object_id, $payload );
	}

	/**
	 * Insert an already validated event, including repository-owned tombstones.
	 *
	 * @param string              $event_type Event type.
	 * @param string              $object_type Object type.
	 * @param int|string          $object_id Object id.
	 * @param array<string,mixed> $payload Payload.
	 * @return int
	 */
	private function insert_event( $event_type, $object_type, $object_id, array $payload ) {
		global $wpdb;

		$encoded = wp_json_encode( $payload );
		if ( ! is_string( $encoded ) ) {
			return 0;
		}

		$now      = current_time( 'mysql' );
		$inserted = $wpdb->insert(
			$wpdb->prefix . 'subin_sync_events',
			array(
				'event_type'      => $this->sanitize_event_type( $event_type ),
				'object_type'     => sanitize_key( (string) $object_type ),
				'object_id'       => sanitize_text_field( (string) $object_id ),
				'payload_json'    => $encoded,
				'status'          => self::STATUS_PENDING,
				'attempts'        => 0,
				'last_error'      => '',
				'claim_token'     => '',
				'lease_until'     => null,
				'next_attempt_at' => $now,
				'created_at'      => $now,
				'updated_at'      => $now,
				'sent_at'         => null,
			),
			array( '%s', '%s', '%s', '%s', '%s', '%d', '%s', '%s', '%s', '%s', '%s', '%s', '%s' )
		);

		return $inserted ? absint( $wpdb->insert_id ) : 0;
	}

	/**
	 * Queue a lead event while sharing the parent lock used by note writes and retention.
	 *
	 * A recorder that wakes after permanent deletion waits for the deletion transaction,
	 * then observes that the lead no longer exists instead of placing a stale event after
	 * its tombstone.
	 *
	 * @param string              $event_type Event type other than lead.deleted.
	 * @param int                 $lead_id Lead id.
	 * @param array<string,mixed> $payload Payload already reduced to the configured consent mode.
	 * @return int
	 */
	public function enqueue_lead_event( $event_type, $lead_id, array $payload ) {
		return $this->enqueue_existing_object_event( $event_type, 'lead', $lead_id, $payload );
	}

	/**
	 * Queue an email event while sharing the mail-row lock used by retention.
	 *
	 * @param string              $event_type Event type other than email.deleted.
	 * @param int                 $event_id Mail event id.
	 * @param array<string,mixed> $payload Payload already reduced to the configured consent mode.
	 * @return int
	 */
	public function enqueue_mail_event( $event_type, $event_id, array $payload ) {
		return $this->enqueue_existing_object_event( $event_type, 'mail_event', $event_id, $payload );
	}

	/**
	 * Atomically replace retained lead sync history with one PII-free tombstone.
	 *
	 * The caller must have an open transaction and hold the lead row FOR UPDATE.
	 * This method intentionally does not start, commit, or roll back the transaction.
	 *
	 * @param int    $lead_id Lead id.
	 * @param string $deleted_at Deletion time in WordPress mysql format.
	 * @param string $reason Stable machine-readable deletion reason.
	 * @param bool   $enqueue_without_history Whether an active Cloud connection can have snapshot history.
	 * @return int|true|\WP_Error Tombstone event id, safe no-op, or failure.
	 */
	public function stage_lead_delete_in_transaction( $lead_id, $deleted_at = '', $reason = 'retention', $enqueue_without_history = true ) {
		return $this->stage_object_delete_in_transaction( 'lead.deleted', 'lead', $lead_id, $deleted_at, $reason, $enqueue_without_history );
	}

	/**
	 * Atomically replace retained email sync history with one PII-free tombstone.
	 *
	 * The caller must have an open transaction and hold the mail-event row FOR
	 * UPDATE. This method intentionally does not manage the transaction boundary.
	 *
	 * @param int    $event_id Mail event id.
	 * @param string $deleted_at Deletion time in WordPress mysql format.
	 * @param string $reason Stable machine-readable deletion reason.
	 * @param bool   $enqueue_without_history Whether an active Cloud connection can have snapshot history.
	 * @return int|true|\WP_Error Tombstone event id, safe no-op, or failure.
	 */
	public function stage_mail_event_delete_in_transaction( $event_id, $deleted_at = '', $reason = 'retention', $enqueue_without_history = true ) {
		return $this->stage_object_delete_in_transaction( 'email.deleted', 'mail_event', $event_id, $deleted_at, $reason, $enqueue_without_history );
	}

	/**
	 * Lock every affected outbox row in global id order before staging a delete graph.
	 *
	 * The caller must already hold parent lead and mail-event locks. Pre-locking
	 * the union prevents a multi-object retention batch from deadlocking with the
	 * worker's global id-ordered claim UPDATE.
	 *
	 * @param array<int,int> $lead_ids Lead ids.
	 * @param array<int,int> $event_ids Mail event ids.
	 * @return int|\WP_Error Number of locked rows or a retryable failure.
	 */
	public function lock_deletion_events_in_transaction( array $lead_ids, array $event_ids ) {
		global $wpdb;

		$lead_ids  = array_values( array_unique( array_filter( array_map( 'absint', $lead_ids ) ) ) );
		$event_ids = array_values( array_unique( array_filter( array_map( 'absint', $event_ids ) ) ) );
		sort( $lead_ids, SORT_NUMERIC );
		sort( $event_ids, SORT_NUMERIC );
		if ( empty( $lead_ids ) && empty( $event_ids ) ) {
			return 0;
		}

		$table      = $wpdb->prefix . 'subin_sync_events';
		$where      = array();
		$arguments  = array();
		if ( ! empty( $lead_ids ) ) {
			$lead_strings = array_map( 'strval', $lead_ids );
			$where[]      = "(object_type = 'lead' AND object_id IN (" . implode( ',', array_fill( 0, count( $lead_strings ), '%s' ) ) . '))';
			$arguments    = array_merge( $arguments, $lead_strings );
		}
		if ( ! empty( $event_ids ) ) {
			$event_strings = array_map( 'strval', $event_ids );
			$where[]       = "(object_type = 'mail_event' AND object_id IN (" . implode( ',', array_fill( 0, count( $event_strings ), '%s' ) ) . '))';
			$arguments     = array_merge( $arguments, $event_strings );
		}
		$wpdb->last_error = '';
		$sql = "SELECT id, object_type, object_id, status FROM {$table} WHERE "
			. implode( ' OR ', $where )
			. ' ORDER BY id ASC FOR UPDATE';
		$rows = $wpdb->get_results(
			// phpcs:ignore WordPress.DB.PreparedSQL.NotPrepared -- Dynamic clauses contain generated placeholders only; the identifier uses the trusted WordPress prefix.
			$wpdb->prepare( $sql, ...$arguments )
		); // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.DB.PreparedSQL.NotPrepared
		if ( ! is_array( $rows ) || '' !== (string) $wpdb->last_error ) {
			return new \WP_Error( 'subin_delete_sync_lock_failed', __( 'Cloud sync history could not be locked for safe deletion.', 'submission-inbox' ) );
		}
		foreach ( $rows as $row ) {
			if ( self::STATUS_PROCESSING === sanitize_key( (string) ( $row->status ?? '' ) ) ) {
				return new \WP_Error(
					'subin_delete_sync_in_flight',
					__( 'Cloud is still processing this item. Deletion will be retried safely.', 'submission-inbox' ),
					array(
						'object_type' => sanitize_key( (string) ( $row->object_type ?? '' ) ),
						'object_id'   => absint( $row->object_id ?? 0 ),
						'retryable'   => true,
					)
				);
			}
		}

		return count( $rows );
	}

	/**
	 * Queue an event only while its local parent still exists.
	 *
	 * @param string              $event_type Event type.
	 * @param string              $object_type Object type.
	 * @param int                 $object_id Object id.
	 * @param array<string,mixed> $payload Event payload.
	 * @return int
	 */
	private function enqueue_existing_object_event( $event_type, $object_type, $object_id, array $payload ) {
		global $wpdb;

		$event_type   = $this->sanitize_event_type( $event_type );
		$object_type  = sanitize_key( (string) $object_type );
		$object_id    = absint( $object_id );
		$object_table = 'lead' === $object_type
			? $wpdb->prefix . 'subin_leads'
			: $wpdb->prefix . 'subin_mail_events';
		$delete_type = 'lead' === $object_type ? 'lead.deleted' : 'email.deleted';
		if ( $object_id <= 0 || $delete_type === $event_type ) {
			return 0;
		}
		if ( false === $wpdb->query( 'START TRANSACTION' ) ) {
			return 0;
		}

		$locked_id = absint(
			$wpdb->get_var(
				$wpdb->prepare(
					"SELECT id FROM {$object_table} WHERE id = %d LIMIT 1 FOR UPDATE",
					$object_id
				)
			)
		);
		if ( $object_id !== $locked_id ) {
			$wpdb->query( 'ROLLBACK' );

			return 0;
		}

		$tombstone_id = absint(
			$wpdb->get_var(
				$wpdb->prepare(
					"SELECT id FROM {$wpdb->prefix}subin_sync_events
					WHERE object_type = %s AND object_id = %s AND event_type = %s
					ORDER BY id ASC LIMIT 1 FOR UPDATE",
					$object_type,
					(string) $object_id,
					$delete_type
				)
			)
		);
		if ( $tombstone_id > 0 ) {
			$wpdb->query( 'ROLLBACK' );

			return 0;
		}

		$event_id = $this->insert_event( $event_type, $object_type, $object_id, $payload );
		if ( $event_id <= 0 || false === $wpdb->query( 'COMMIT' ) ) {
			$wpdb->query( 'ROLLBACK' );

			return 0;
		}

		return $event_id;
	}

	/**
	 * Stage one canonical deletion event without retaining superseded payloads.
	 *
	 * @param string $event_type Deletion event type.
	 * @param string $object_type Sync object type.
	 * @param int    $object_id Local object id.
	 * @param string $deleted_at Deletion time.
	 * @param string $reason Deletion reason.
	 * @param bool   $enqueue_without_history Whether an active Cloud connection can have snapshot history.
	 * @return int|true|\WP_Error
	 */
	private function stage_object_delete_in_transaction( $event_type, $object_type, $object_id, $deleted_at, $reason, $enqueue_without_history ) {
		global $wpdb;

		$event_type  = $this->sanitize_event_type( $event_type );
		$object_type = sanitize_key( (string) $object_type );
		$object_id   = absint( $object_id );
		$expected    = 'lead' === $object_type ? 'lead.deleted' : 'email.deleted';
		if ( $object_id <= 0 || $expected !== $event_type ) {
			return new \WP_Error( 'subin_delete_sync_object_invalid', __( 'The deleted item could not prepare its privacy event.', 'submission-inbox' ) );
		}

		$table = $wpdb->prefix . 'subin_sync_events';
		$wpdb->last_error = '';
		$rows = $wpdb->get_results(
			$wpdb->prepare(
				"SELECT id, event_type, payload_json, status FROM {$table}
				WHERE object_type = %s AND object_id = %s
				ORDER BY id ASC FOR UPDATE",
				$object_type,
				(string) $object_id
			)
		); // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
		if ( ! is_array( $rows ) || '' !== (string) $wpdb->last_error ) {
			return new \WP_Error( 'subin_delete_sync_read_failed', __( 'The deleted item could not inspect local sync history.', 'submission-inbox' ) );
		}
		foreach ( $rows as $row ) {
			if ( self::STATUS_PROCESSING === sanitize_key( (string) ( $row->status ?? '' ) ) ) {
				return new \WP_Error(
					'subin_delete_sync_in_flight',
					__( 'Cloud is still processing this item. Deletion will be retried safely.', 'submission-inbox' ),
					array(
						'object_type' => $object_type,
						'object_id'   => $object_id,
						'retryable'   => true,
					)
				);
			}
		}

		$had_history      = ! empty( $rows );
		$tombstone_id     = 0;
		$tombstone_status = '';
		$existing_payload = array();
		foreach ( $rows as $row ) {
			if ( 0 === $tombstone_id && $event_type === (string) ( $row->event_type ?? '' ) ) {
				$tombstone_id     = absint( $row->id ?? 0 );
				$tombstone_status = sanitize_key( (string) ( $row->status ?? '' ) );
				$decoded          = json_decode( (string) ( $row->payload_json ?? '' ), true );
				$existing_payload = is_array( $decoded ) ? $decoded : array();
			}
		}

		$payload = $this->deletion_payload( $object_type, $object_id, $deleted_at, $reason, $existing_payload );
		$encoded = wp_json_encode( $payload );
		if ( ! is_string( $encoded ) ) {
			return new \WP_Error( 'subin_delete_sync_encode_failed', __( 'The deletion privacy event could not be encoded.', 'submission-inbox' ) );
		}

		if ( $tombstone_id > 0 ) {
			$update = array(
				'payload_json' => $encoded,
				'updated_at'   => current_time( 'mysql' ),
			);
			$formats = array( '%s', '%s' );
			if ( self::STATUS_SENT !== $tombstone_status ) {
				$update['status']          = self::STATUS_PENDING;
				$update['attempts']        = 0;
				$update['last_error']      = '';
				$update['claim_token']     = '';
				$update['lease_until']     = null;
				$update['next_attempt_at'] = current_time( 'mysql' );
				$update['sent_at']         = null;
				$formats = array( '%s', '%s', '%s', '%d', '%s', '%s', '%s', '%s', '%s' );
			}
			$updated = $wpdb->update(
				$table,
				$update,
				array( 'id' => $tombstone_id ),
				$formats,
				array( '%d' )
			);
			if ( false === $updated ) {
				return new \WP_Error( 'subin_delete_sync_update_failed', __( 'The deletion privacy event could not be normalized.', 'submission-inbox' ) );
			}
		}

		$delete_sql  = "DELETE FROM {$table} WHERE object_type = %s AND object_id = %s";
		$delete_args = array( $object_type, (string) $object_id );
		if ( $tombstone_id > 0 ) {
			$delete_sql   .= ' AND id <> %d';
			$delete_args[] = $tombstone_id;
		}
		$removed = $wpdb->query(
			// phpcs:ignore WordPress.DB.PreparedSQL.NotPrepared -- The optional condition uses a generated integer placeholder.
			$wpdb->prepare( $delete_sql, ...$delete_args ) // phpcs:ignore WordPress.DB.PreparedSQL.NotPrepared,WordPress.DB.PreparedSQL.InterpolatedNotPrepared
		);
		if ( false === $removed ) {
			return new \WP_Error( 'subin_delete_sync_scrub_failed', __( 'Superseded sync payloads could not be removed safely.', 'submission-inbox' ) );
		}

		if ( $tombstone_id > 0 ) {
			return $tombstone_id;
		}
		if ( ! $enqueue_without_history && ! $had_history ) {
			return true;
		}

		$tombstone_id = $this->insert_event( $event_type, $object_type, $object_id, $payload );
		if ( $tombstone_id <= 0 ) {
			return new \WP_Error( 'subin_delete_sync_enqueue_failed', __( 'The deletion privacy event could not be staged.', 'submission-inbox' ) );
		}

		return $tombstone_id;
	}

	/**
	 * Build the complete allowlisted deletion payload.
	 *
	 * @param string              $object_type Sync object type.
	 * @param int                 $object_id Local object id.
	 * @param string              $deleted_at Requested deletion time.
	 * @param string              $reason Requested deletion reason.
	 * @param array<string,mixed> $existing Existing tombstone payload, when idempotently restaged.
	 * @return array<string,mixed>
	 */
	private function deletion_payload( $object_type, $object_id, $deleted_at, $reason, array $existing = array() ) {
		$id_key              = 'lead' === $object_type ? 'external_lead_id' : 'external_event_id';
		$existing_deleted_at = $this->normalize_deletion_datetime( $existing['deleted_at'] ?? '' );
		$existing_reason     = $this->normalize_deletion_reason( $existing['reason'] ?? '' );
		$requested_deleted_at = $this->normalize_deletion_datetime( $deleted_at );
		if ( '' === $requested_deleted_at ) {
			$requested_deleted_at = $this->normalize_deletion_datetime( current_time( 'mysql' ) );
		}

		return array(
			'contract_version' => self::DELETION_CONTRACT_VERSION,
			$id_key            => (string) absint( $object_id ),
			'deleted'          => true,
			'deleted_at'       => '' !== $existing_deleted_at ? $existing_deleted_at : $requested_deleted_at,
			'reason'           => '' !== $existing_reason ? $existing_reason : $this->normalize_deletion_reason( $reason, 'retention' ),
		);
	}

	/**
	 * Normalize a deletion timestamp without accepting arbitrary retained text.
	 *
	 * @param mixed $value Candidate timestamp.
	 * @return string
	 */
	private function normalize_deletion_datetime( $value ) {
		$value = trim( (string) $value );
		if ( ! preg_match( '/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$/', $value ) ) {
			return '';
		}
		$parsed = \DateTimeImmutable::createFromFormat( '!Y-m-d H:i:s', $value, new \DateTimeZone( 'UTC' ) );
		if ( ! $parsed || $parsed->format( 'Y-m-d H:i:s' ) !== $value ) {
			return '';
		}

		return $value;
	}

	/**
	 * Normalize deletion reason to a non-PII allowlist.
	 *
	 * @param mixed  $value Candidate reason.
	 * @param string $fallback Fallback when candidate is invalid.
	 * @return string
	 */
	private function normalize_deletion_reason( $value, $fallback = '' ) {
		$value   = sanitize_key( (string) $value );
		$allowed = array( 'retention', 'manual', 'privacy_request', 'parent_deleted' );

		return in_array( $value, $allowed, true ) ? $value : $fallback;
	}

	/**
	 * Serialize one versioned note mutation with retention and prior tombstones.
	 *
	 * The parent-lead lock is shared with note writes and retention. A delete
	 * redacts every local revision before its tombstone is queued, while an
	 * out-of-order add/update cannot resurrect a note after a newer delete.
	 *
	 * @param string              $event_type Versioned note event type.
	 * @param int                 $lead_id Lead id.
	 * @param array<string,mixed> $payload Immutable event payload.
	 * @param bool                $enqueue_without_history Whether current consent allows a new outbound tombstone.
	 * @return int
	 */
	public function enqueue_note_event( $event_type, $lead_id, array $payload, $enqueue_without_history = true ) {
		global $wpdb;

		$event_type = $this->sanitize_event_type( $event_type );
		$lead_id    = absint( $lead_id );
		$note       = is_array( $payload['note'] ?? null ) ? $payload['note'] : array();
		$note_id    = absint( $note['id'] ?? 0 );
		$version    = absint( $note['note_version'] ?? 0 );
		$deleted    = 'lead.note_deleted' === $event_type && ! empty( $note['deleted'] );
		$had_sync_footprint = ! empty( $note['sync_footprint'] );
		if (
			$lead_id <= 0
			|| $note_id <= 0
			|| $version <= 0
			|| ! in_array( $event_type, array( 'lead.note_added', 'lead.note_updated', 'lead.note_deleted' ), true )
			|| ( 'lead.note_deleted' === $event_type ) !== $deleted
		) {
			return 0;
		}
		if ( false === $wpdb->query( 'START TRANSACTION' ) ) {
			return 0;
		}

		$locked_lead = $wpdb->get_var(
			$wpdb->prepare(
				"SELECT id FROM {$wpdb->prefix}subin_leads WHERE id = %d LIMIT 1 FOR UPDATE",
				$lead_id
			)
		);
		if ( $lead_id !== absint( $locked_lead ) ) {
			$wpdb->query( 'ROLLBACK' );

			return 0;
		}
		if ( $deleted ) {
			$staged = $this->stage_note_delete_in_transaction( $lead_id, $payload, $enqueue_without_history );
			if ( is_wp_error( $staged ) || false === $wpdb->query( 'COMMIT' ) ) {
				$wpdb->query( 'ROLLBACK' );

				return 0;
			}

			return is_int( $staged ) ? $staged : 0;
		}
		if ( ! $deleted ) {
			$current_note_version = absint(
				$wpdb->get_var(
					$wpdb->prepare(
						"SELECT note_version FROM {$wpdb->prefix}subin_lead_notes
						WHERE id = %d AND lead_id = %d LIMIT 1",
						$note_id,
						$lead_id
					)
				)
			);
			if ( $current_note_version < $version ) {
				$wpdb->query( 'ROLLBACK' );

				return 0;
			}
		}

		$table = $wpdb->prefix . 'subin_sync_events';
		$wpdb->last_error = '';
		$rows  = $wpdb->get_results(
			$wpdb->prepare(
				"SELECT id, event_type, payload_json FROM {$table}
				WHERE object_type = %s AND object_id = %s
				ORDER BY id ASC FOR UPDATE",
				'lead',
				(string) $lead_id
			)
		); // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
		if ( ! is_array( $rows ) || '' !== (string) $wpdb->last_error ) {
			$wpdb->query( 'ROLLBACK' );

			return 0;
		}

		$existing_tombstone_id = 0;
		$has_note_history      = false;
		foreach ( $rows as $row ) {
			$stored = json_decode( (string) ( $row->payload_json ?? '' ), true );
			if ( ! is_array( $stored ) ) {
				continue;
			}
			if ( $note_id === $this->payload_note_id( $stored ) ) {
				$has_note_history = true;
				$stored_note    = is_array( $stored['note'] ?? null ) ? $stored['note'] : array();
				$stored_version = absint( $stored_note['note_version'] ?? 0 );
				if ( 'lead.note_deleted' === (string) ( $row->event_type ?? '' ) && $stored_version >= $version ) {
					$existing_tombstone_id = absint( $row->id ?? 0 );
				}
			}
			if ( $this->payload_contains_note( $stored, $note_id ) ) {
				$has_note_history = true;
			}
			if ( ! $deleted ) {
				continue;
			}

			$redacted = $this->redact_note_from_payload( $stored, $note_id );
			$encoded  = wp_json_encode( $redacted );
			if ( ! is_string( $encoded ) ) {
				$wpdb->query( 'ROLLBACK' );

				return 0;
			}
			if ( $encoded !== (string) ( $row->payload_json ?? '' ) ) {
				$updated = $wpdb->update(
					$table,
					array(
						'payload_json' => $encoded,
						'updated_at'   => current_time( 'mysql' ),
					),
					array( 'id' => absint( $row->id ?? 0 ) ),
					array( '%s', '%s' ),
					array( '%d' )
				);
				if ( false === $updated ) {
					$wpdb->query( 'ROLLBACK' );

					return 0;
				}
			}
		}

		if ( $existing_tombstone_id > 0 ) {
			if ( false === $wpdb->query( 'COMMIT' ) ) {
				$wpdb->query( 'ROLLBACK' );

				return 0;
			}

			return $deleted ? $existing_tombstone_id : 0;
		}
		if ( $deleted && ! $enqueue_without_history && ! $has_note_history && ! $had_sync_footprint ) {
			if ( false === $wpdb->query( 'COMMIT' ) ) {
				$wpdb->query( 'ROLLBACK' );
			}

			return 0;
		}

		$event_id = $this->insert_event( $event_type, 'lead', $lead_id, $payload );
		if ( $event_id <= 0 ) {
			$wpdb->query( 'ROLLBACK' );

			return 0;
		}
		if ( ! $deleted ) {
			$marked = $wpdb->update(
				$wpdb->prefix . 'subin_lead_notes',
				array( 'sync_footprint' => 1 ),
				array(
					'id'      => $note_id,
					'lead_id' => $lead_id,
				),
				array( '%d' ),
				array( '%d', '%d' )
			);
			if ( false === $marked ) {
				$wpdb->query( 'ROLLBACK' );

				return 0;
			}
		}
		if ( false === $wpdb->query( 'COMMIT' ) ) {
			$wpdb->query( 'ROLLBACK' );

			return 0;
		}

		return $event_id;
	}

	/**
	 * Atomically redact note history and stage its tombstone in an open transaction.
	 *
	 * The caller must already hold the parent lead row lock. Returning true means
	 * local redaction succeeded and no outbound tombstone was required.
	 *
	 * @param int                 $lead_id Lead id.
	 * @param array<string,mixed> $payload Immutable delete payload.
	 * @param bool                $enqueue_without_history Whether current consent allows a new tombstone.
	 * @return int|true|\WP_Error
	 */
	public function stage_note_delete_in_transaction( $lead_id, array $payload, $enqueue_without_history = true ) {
		global $wpdb;

		$lead_id = absint( $lead_id );
		$note    = is_array( $payload['note'] ?? null ) ? $payload['note'] : array();
		$note_id = absint( $note['id'] ?? 0 );
		$version = absint( $note['note_version'] ?? 0 );
		if ( $lead_id <= 0 || $note_id <= 0 || $version <= 0 || empty( $note['deleted'] ) ) {
			return new \WP_Error( 'subin_note_delete_sync_payload_invalid', __( 'The note deletion could not prepare its privacy event.', 'submission-inbox' ) );
		}

		$table = $wpdb->prefix . 'subin_sync_events';
		$wpdb->last_error = '';
		$rows = $wpdb->get_results(
			$wpdb->prepare(
				"SELECT id, event_type, payload_json FROM {$table}
				WHERE object_type = %s AND object_id = %s
				ORDER BY id ASC FOR UPDATE",
				'lead',
				(string) $lead_id
			)
		); // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
		if ( ! is_array( $rows ) || '' !== (string) $wpdb->last_error ) {
			return new \WP_Error( 'subin_note_delete_sync_read_failed', __( 'The note deletion could not inspect local sync history.', 'submission-inbox' ) );
		}

		$existing_tombstone_id = 0;
		$has_note_history      = ! empty( $note['sync_footprint'] );
		foreach ( $rows as $row ) {
			$stored = json_decode( (string) ( $row->payload_json ?? '' ), true );
			if ( ! is_array( $stored ) ) {
				continue;
			}
			if ( $note_id === $this->payload_note_id( $stored ) ) {
				$has_note_history = true;
				$stored_note      = is_array( $stored['note'] ?? null ) ? $stored['note'] : array();
				if ( 'lead.note_deleted' === (string) ( $row->event_type ?? '' ) && absint( $stored_note['note_version'] ?? 0 ) >= $version ) {
					$existing_tombstone_id = absint( $row->id ?? 0 );
				}
			}
			if ( $this->payload_contains_note( $stored, $note_id ) ) {
				$has_note_history = true;
			}

			$redacted = $this->redact_note_from_payload( $stored, $note_id );
			$encoded  = wp_json_encode( $redacted );
			if ( ! is_string( $encoded ) ) {
				return new \WP_Error( 'subin_note_delete_sync_encode_failed', __( 'The note deletion could not sanitize local sync history.', 'submission-inbox' ) );
			}
			if ( $encoded === (string) ( $row->payload_json ?? '' ) ) {
				continue;
			}
			$updated = $wpdb->update(
				$table,
				array(
					'payload_json' => $encoded,
					'updated_at'   => current_time( 'mysql' ),
				),
				array( 'id' => absint( $row->id ?? 0 ) ),
				array( '%s', '%s' ),
				array( '%d' )
			);
			if ( false === $updated ) {
				return new \WP_Error( 'subin_note_delete_sync_redaction_failed', __( 'The note deletion could not redact local sync history.', 'submission-inbox' ) );
			}
		}

		if ( $existing_tombstone_id > 0 ) {
			return $existing_tombstone_id;
		}
		if ( ! $enqueue_without_history && ! $has_note_history ) {
			return true;
		}

		$event_id = $this->insert_event( 'lead.note_deleted', 'lead', $lead_id, $payload );
		if ( $event_id <= 0 ) {
			return new \WP_Error( 'subin_note_delete_sync_enqueue_failed', __( 'The note deletion could not stage its sync tombstone.', 'submission-inbox' ) );
		}

		return $event_id;
	}

	/**
	 * Resolve a note id from current or pre-versioning payload shapes.
	 *
	 * @param array<string,mixed> $payload Stored payload.
	 * @return int
	 */
	private function payload_note_id( array $payload ) {
		$note = is_array( $payload['note'] ?? null ) ? $payload['note'] : array();

		return absint( $note['id'] ?? $payload['note_id'] ?? 0 );
	}

	/**
	 * Determine whether any current or aggregate payload contains one note.
	 *
	 * @param array<string,mixed> $payload Stored payload.
	 * @param int                 $note_id Note id.
	 * @return bool
	 */
	private function payload_contains_note( array $payload, $note_id ) {
		if ( absint( $this->payload_note_id( $payload ) ) === absint( $note_id ) ) {
			return true;
		}
		$collections = array(
			$payload['lead']['notes'] ?? null,
			$payload['notes'] ?? null,
		);
		foreach ( $collections as $notes ) {
			if ( ! is_array( $notes ) ) {
				continue;
			}
			foreach ( $notes as $stored_note ) {
				if ( is_array( $stored_note ) && absint( $stored_note['id'] ?? 0 ) === absint( $note_id ) ) {
					return true;
				}
			}
		}

		return false;
	}

	/**
	 * Remove one deleted note body from a locally retained sync payload.
	 *
	 * @param array<string,mixed> $payload Stored payload.
	 * @param int                 $note_id Deleted note id.
	 * @return array<string,mixed>
	 */
	private function redact_note_from_payload( array $payload, $note_id ) {
		if ( is_array( $payload['note'] ?? null ) && absint( $payload['note']['id'] ?? 0 ) === absint( $note_id ) ) {
			unset( $payload['note']['body'], $payload['note']['note'] );
		}
		foreach ( array( array( 'lead', 'notes' ), array( 'notes' ) ) as $path ) {
			$notes = 2 === count( $path )
				? ( $payload[ $path[0] ][ $path[1] ] ?? null )
				: ( $payload[ $path[0] ] ?? null );
			if ( ! is_array( $notes ) ) {
				continue;
			}
			$filtered = array_values(
				array_filter(
					$notes,
					static function ( $stored_note ) use ( $note_id ) {
						return ! is_array( $stored_note ) || absint( $stored_note['id'] ?? 0 ) !== absint( $note_id );
					}
				)
			);
			if ( 2 === count( $path ) ) {
				$payload[ $path[0] ][ $path[1] ] = $filtered;
			} else {
				$payload[ $path[0] ] = $filtered;
			}
		}

		return $payload;
	}

	/**
	 * Sanitize event type while preserving dot-separated names.
	 *
	 * @param string $event_type Event type.
	 * @return string
	 */
	private function sanitize_event_type( $event_type ) {
		return preg_replace( '/[^a-z0-9_.-]/', '', strtolower( (string) $event_type ) );
	}

	/**
	 * Get pending events eligible for a sync attempt without claiming them.
	 *
	 * @param int $limit Limit.
	 * @return array<int,object>
	 */
	public function get_pending_batch( $limit = 25 ) {
		global $wpdb;

		$table = $wpdb->prefix . 'subin_sync_events';
		$limit = max( 1, min( 100, absint( $limit ) ) );
		$now   = current_time( 'mysql' );

		return $wpdb->get_results(
			$wpdb->prepare(
				"SELECT * FROM {$table}
				WHERE status IN (%s, %s)
					AND (attempts < %d OR event_type LIKE %s)
					AND (next_attempt_at IS NULL OR next_attempt_at <= %s)
				ORDER BY id ASC
				LIMIT %d",
				self::STATUS_PENDING,
				self::STATUS_FAILED,
				self::MAX_ATTEMPTS,
				'%.deleted',
				$now,
				$limit
			) // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
		);
	}

	/**
	 * Atomically claim a disjoint batch for one worker.
	 *
	 * The single UPDATE with ORDER BY/LIMIT changes eligibility while rows are
	 * locked, so concurrent workers cannot receive the same processing rows.
	 *
	 * @param int $limit Batch size.
	 * @param int $lease_seconds Lease duration.
	 * @return array{claim_token:string,events:array<int,object>,lease_until:string}
	 */
	public function claim_pending_batch( $limit = 25, $lease_seconds = self::DEFAULT_LEASE_SECONDS ) {
		global $wpdb;

		$this->release_expired_claims();

		$table         = $wpdb->prefix . 'subin_sync_events';
		$limit         = max( 1, min( 100, absint( $limit ) ) );
		$lease_seconds = max( 60, min( HOUR_IN_SECONDS, absint( $lease_seconds ) ) );
		$now           = current_time( 'mysql' );
		$lease_until   = gmdate( 'Y-m-d H:i:s', current_time( 'timestamp' ) + $lease_seconds );
		$claim_token   = $this->new_claim_token();

		if ( '' === $claim_token ) {
			return $this->empty_claim();
		}

		$claimed = $wpdb->query(
			$wpdb->prepare(
				"UPDATE {$table}
				SET status = %s, claim_token = %s, lease_until = %s, updated_at = %s
				WHERE status IN (%s, %s)
					AND (attempts < %d OR event_type LIKE %s)
					AND (next_attempt_at IS NULL OR next_attempt_at <= %s)
				ORDER BY id ASC
				LIMIT %d",
				self::STATUS_PROCESSING,
				$claim_token,
				$lease_until,
				$now,
				self::STATUS_PENDING,
				self::STATUS_FAILED,
				self::MAX_ATTEMPTS,
				'%.deleted',
				$now,
				$limit
			) // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
		);

		if ( ! $claimed ) {
			return $this->empty_claim();
		}

		return array(
			'claim_token' => $claim_token,
			'events'      => $this->get_by_claim_token( $claim_token, array( self::STATUS_PROCESSING ) ),
			'lease_until' => $lease_until,
		);
	}

	/**
	 * Get events belonging to a claim token.
	 *
	 * @param string            $claim_token Claim token.
	 * @param array<int,string> $statuses Optional statuses.
	 * @return array<int,object>
	 */
	public function get_by_claim_token( $claim_token, array $statuses = array() ) {
		global $wpdb;

		$claim_token = $this->sanitize_claim_token( $claim_token );
		if ( '' === $claim_token ) {
			return array();
		}

		$table  = $wpdb->prefix . 'subin_sync_events';
		$params = array( $claim_token );
		$sql    = "SELECT * FROM {$table} WHERE claim_token = %s";

		if ( ! empty( $statuses ) ) {
			$statuses     = array_values( array_unique( array_filter( array_map( 'sanitize_key', $statuses ) ) ) );
			$placeholders = implode( ',', array_fill( 0, count( $statuses ), '%s' ) );
			$sql         .= " AND status IN ({$placeholders})";
			$params       = array_merge( $params, $statuses );
		}

		$sql .= ' ORDER BY id ASC';

		return $wpdb->get_results(
			$wpdb->prepare( $sql, ...$params ) // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.DB.PreparedSQL.NotPrepared
		);
	}

	/**
	 * Idempotently acknowledge claimed events as sent.
	 *
	 * Terminal rows retain their claim token, allowing a repeated ACK to produce
	 * the same successful result without mutating a newer claim.
	 *
	 * @param string         $claim_token Claim token.
	 * @param array<int,int> $event_ids Optional event ids.
	 * @return int Number of matching claimed or already acknowledged rows.
	 */
	public function acknowledge_claim( $claim_token, array $event_ids = array() ) {
		$claim_token = $this->sanitize_claim_token( $claim_token );
		if ( '' === $claim_token ) {
			return 0;
		}

		$rows = $this->get_by_claim_token(
			$claim_token,
			array( self::STATUS_PROCESSING, self::STATUS_SENT )
		);
		if ( empty( $rows ) ) {
			return 0;
		}

		$claimed_ids = array_map( 'absint', wp_list_pluck( $rows, 'id' ) );
		$event_ids   = array_values( array_filter( array_map( 'absint', $event_ids ) ) );
		$event_ids   = empty( $event_ids ) ? $claimed_ids : array_values( array_intersect( $claimed_ids, $event_ids ) );

		if ( empty( $event_ids ) ) {
			return 0;
		}

		$this->mark_sent( $event_ids, $claim_token );

		return count( $event_ids );
	}

	/**
	 * Release a failed claim and schedule it with exponential backoff.
	 *
	 * @param string $claim_token Claim token.
	 * @param string $error Error message.
	 * @return int Number of released rows.
	 */
	public function fail_claim( $claim_token, $error ) {
		$claim_token = $this->sanitize_claim_token( $claim_token );
		if ( '' === $claim_token ) {
			return 0;
		}

		$rows = $this->get_by_claim_token( $claim_token, array( self::STATUS_PROCESSING ) );
		$ids  = array_map( 'absint', wp_list_pluck( $rows, 'id' ) );
		if ( empty( $ids ) ) {
			return 0;
		}

		$this->mark_failed( $ids, $error, $claim_token );

		return count( $ids );
	}

	/**
	 * Release expired processing leases for a future retry.
	 *
	 * @return int Number of released rows.
	 */
	public function release_expired_claims() {
		global $wpdb;

		$table = $wpdb->prefix . 'subin_sync_events';
		$now   = current_time( 'mysql' );

		return absint(
			$wpdb->query(
				$wpdb->prepare(
					"UPDATE {$table}
					SET status = %s,
						next_attempt_at = TIMESTAMPADD(SECOND, LEAST(3600, POW(2, LEAST(10, attempts + 1)) * 60), %s),
						attempts = attempts + 1,
						last_error = %s,
						claim_token = '',
						lease_until = NULL,
						updated_at = %s
					WHERE status = %s
						AND lease_until IS NOT NULL
						AND lease_until <= %s",
					self::STATUS_FAILED,
					$now,
					__( 'Cloud sync lease expired before acknowledgement.', 'submission-inbox' ),
					$now,
					self::STATUS_PROCESSING,
					$now
				) // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
			)
		);
	}

	/**
	 * Mark events as sent.
	 *
	 * @param array<int,int> $event_ids Event ids.
	 * @param string         $claim_token Optional claim token.
	 * @return void
	 */
	public function mark_sent( array $event_ids, $claim_token = '' ) {
		global $wpdb;

		$event_ids = array_values( array_filter( array_map( 'absint', $event_ids ) ) );
		if ( empty( $event_ids ) ) {
			return;
		}

		$table        = $wpdb->prefix . 'subin_sync_events';
		$placeholders = implode( ',', array_fill( 0, count( $event_ids ), '%d' ) );
		$params       = array_merge(
			array(
				self::STATUS_SENT,
				current_time( 'mysql' ),
				current_time( 'mysql' ),
			),
			$event_ids
		);
		$claim_token = $this->sanitize_claim_token( $claim_token );
		$where_claim = '';

		if ( '' !== $claim_token ) {
			$where_claim = ' AND claim_token = %s AND status = %s';
			$params[]    = $claim_token;
			$params[]    = self::STATUS_PROCESSING;
		}

		$wpdb->query(
			$wpdb->prepare(
				"UPDATE {$table}
				SET status = %s, sent_at = %s, updated_at = %s, last_error = '', lease_until = NULL, next_attempt_at = NULL
				WHERE id IN ({$placeholders}){$where_claim}",
				...$params
			) // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
		);
	}

	/**
	 * Mark events as failed.
	 *
	 * @param array<int,int> $event_ids Event ids.
	 * @param string         $error Error message.
	 * @param string         $claim_token Optional claim token.
	 * @return void
	 */
	public function mark_failed( array $event_ids, $error, $claim_token = '' ) {
		global $wpdb;

		$event_ids = array_values( array_filter( array_map( 'absint', $event_ids ) ) );
		if ( empty( $event_ids ) ) {
			return;
		}

		$table        = $wpdb->prefix . 'subin_sync_events';
		$placeholders = implode( ',', array_fill( 0, count( $event_ids ), '%d' ) );
		$now          = current_time( 'mysql' );
		$params       = array_merge(
			array(
				self::STATUS_FAILED,
				$now,
				sanitize_textarea_field( (string) $error ),
				$now,
			),
			$event_ids
		);
		$claim_token = $this->sanitize_claim_token( $claim_token );
		$where_claim = '';

		if ( '' !== $claim_token ) {
			$where_claim = ' AND claim_token = %s AND status = %s';
			$params[]    = $claim_token;
			$params[]    = self::STATUS_PROCESSING;
		}

		$wpdb->query(
			$wpdb->prepare(
				"UPDATE {$table}
				SET status = %s,
					next_attempt_at = TIMESTAMPADD(SECOND, LEAST(3600, POW(2, LEAST(10, attempts + 1)) * 60), %s),
					attempts = attempts + 1,
					last_error = %s,
					updated_at = %s,
					claim_token = '',
					lease_until = NULL
				WHERE id IN ({$placeholders}){$where_claim}",
				...$params
			) // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
		);
	}

	/**
	 * Mark events as skipped.
	 *
	 * @param array<int,int> $event_ids Event ids.
	 * @param string         $reason Reason.
	 * @param string         $claim_token Optional claim token.
	 * @return void
	 */
	public function mark_skipped( array $event_ids, $reason, $claim_token = '' ) {
		global $wpdb;

		$event_ids = array_values( array_filter( array_map( 'absint', $event_ids ) ) );
		if ( empty( $event_ids ) ) {
			return;
		}

		$table        = $wpdb->prefix . 'subin_sync_events';
		$placeholders = implode( ',', array_fill( 0, count( $event_ids ), '%d' ) );
		$params       = array_merge(
			array(
				self::STATUS_SKIPPED,
				sanitize_textarea_field( (string) $reason ),
				current_time( 'mysql' ),
			),
			$event_ids
		);
		$claim_token = $this->sanitize_claim_token( $claim_token );
		$where_claim = '';

		if ( '' !== $claim_token ) {
			$where_claim = ' AND claim_token = %s AND status = %s';
			$params[]    = $claim_token;
			$params[]    = self::STATUS_PROCESSING;
		}

		$wpdb->query(
			$wpdb->prepare(
				"UPDATE {$table}
				SET status = %s, last_error = %s, updated_at = %s, lease_until = NULL, next_attempt_at = NULL
				WHERE id IN ({$placeholders}){$where_claim}",
				...$params
			) // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
		);
	}

	/**
	 * Count queued events by status.
	 *
	 * @param string $status Status.
	 * @return int
	 */
	public function count_by_status( $status = self::STATUS_PENDING ) {
		global $wpdb;

		$table = $wpdb->prefix . 'subin_sync_events';

		return absint(
			$wpdb->get_var(
				$wpdb->prepare(
					"SELECT COUNT(*) FROM {$table} WHERE status = %s",
					sanitize_key( (string) $status )
				) // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
			)
		);
	}

	/**
	 * Count events grouped by status.
	 *
	 * @return array<string,int>
	 */
	public function count_all_by_statuses() {
		global $wpdb;

		$table  = $wpdb->prefix . 'subin_sync_events';
		$counts = array(
			self::STATUS_PENDING    => 0,
			self::STATUS_PROCESSING => 0,
			self::STATUS_FAILED     => 0,
			self::STATUS_SENT       => 0,
			self::STATUS_SKIPPED    => 0,
		);
		$rows   = $wpdb->get_results( "SELECT status, COUNT(*) AS total FROM {$table} GROUP BY status", ARRAY_A ); // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared

		foreach ( $rows as $row ) {
			$status = sanitize_key( (string) ( $row['status'] ?? '' ) );
			if ( array_key_exists( $status, $counts ) ) {
				$counts[ $status ] = absint( $row['total'] ?? 0 );
			}
		}

		return $counts;
	}

	/**
	 * Get recent sync events.
	 *
	 * @param int $limit Limit.
	 * @return array<int,object>
	 */
	public function recent( $limit = 10 ) {
		global $wpdb;

		$table = $wpdb->prefix . 'subin_sync_events';
		$limit = max( 1, min( 50, absint( $limit ) ) );

		return $wpdb->get_results(
			$wpdb->prepare(
				"SELECT id, event_type, object_type, object_id, status, attempts, last_error, created_at, updated_at, sent_at, next_attempt_at, lease_until
				FROM {$table}
				ORDER BY id DESC
				LIMIT %d",
				$limit
			) // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
		);
	}

	/**
	 * Move failed events back to pending.
	 *
	 * @return int|false Number of updated rows, or false on failure.
	 */
	public function retry_failed() {
		global $wpdb;

		return $wpdb->update(
			$wpdb->prefix . 'subin_sync_events',
			array(
				'status'          => self::STATUS_PENDING,
				'attempts'        => 0,
				'last_error'      => '',
				'claim_token'     => '',
				'lease_until'     => null,
				'next_attempt_at' => current_time( 'mysql' ),
				'updated_at'      => current_time( 'mysql' ),
			),
			array( 'status' => self::STATUS_FAILED ),
			array( '%s', '%d', '%s', '%s', '%s', '%s', '%s' ),
			array( '%s' )
		);
	}

	/**
	 * Get the highest local event id.
	 *
	 * @return int
	 */
	public function latest_id() {
		global $wpdb;

		$table = $wpdb->prefix . 'subin_sync_events';

		return absint( $wpdb->get_var( "SELECT MAX(id) FROM {$table}" ) ); // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
	}

	/**
	 * Build an empty claim response.
	 *
	 * @return array{claim_token:string,events:array<int,object>,lease_until:string}
	 */
	private function empty_claim() {
		return array(
			'claim_token' => '',
			'events'      => array(),
			'lease_until' => '',
		);
	}

	/**
	 * Generate a storage-safe claim token.
	 *
	 * @return string
	 */
	private function new_claim_token() {
		return $this->sanitize_claim_token( wp_generate_password( 48, false, false ) );
	}

	/**
	 * Normalize a claim token for storage and lookup.
	 *
	 * @param string $claim_token Claim token.
	 * @return string
	 */
	private function sanitize_claim_token( $claim_token ) {
		return substr( (string) preg_replace( '/[^A-Za-z0-9]/', '', (string) $claim_token ), 0, 64 );
	}
}
