<?php
/**
 * Retention cleanup service.
 *
 * @package SubmissionInbox
 */

namespace SUBIN\Retention;

use SUBIN\Forms\SubmissionReceiptRepository;
use SUBIN\Leads\LeadFieldRepository;
use SUBIN\Leads\LeadRepository;
use SUBIN\MailLog\MailEventRepository;
use SUBIN\Notifications\NotificationJobRepository;
use SUBIN\Sync\SyncEventRepository;

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

// phpcs:disable WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- Retention uses bounded set-based deletes against plugin-owned tables.
// phpcs:disable PluginCheck.Security.DirectDB.UnescapedDBParameter, WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- Table names are generated by SUBIN_Database from the trusted WordPress prefix.

/**
 * Cleans expired lead records and bounded operational history.
 */
class RetentionService {
	const CRON_HOOK        = 'subin_daily_retention_cleanup';
	const CONTINUATION_HOOK = 'subin_retention_cleanup_continue';
	const LOCK_OPTION      = 'subin_retention_cleanup_lock';
	const POLICY_OPTION    = 'subin_retention_policy_state';
	const LOCK_TTL         = 600;

	const DEFAULT_BATCH_SIZE                      = 250;
	const DEFAULT_TIME_BUDGET                     = 20;
	const DEFAULT_MAIL_RETENTION_DAYS             = 90;
	const DEFAULT_SYNC_PAYLOAD_RETENTION_DAYS     = 1;
	const DEFAULT_SYNC_TERMINAL_RETENTION_DAYS    = 30;
	const DEFAULT_SYNC_FAILED_RETENTION_DAYS      = 90;
	const DEFAULT_NOTIFICATION_PAYLOAD_RETENTION_DAYS = 1;
	const DEFAULT_NOTIFICATION_RETENTION_DAYS     = 30;
	const DEFAULT_NOTIFICATION_FAILED_RETENTION_DAYS = 90;
	const DEFAULT_FORM_ANALYTICS_RETENTION_DAYS   = 90;

	/**
	 * Settings repository.
	 *
	 * @var \SUBIN_SettingsRepository
	 */
	private $settings;

	/**
	 * Lead repository.
	 *
	 * @var LeadRepository
	 */
	private $leads;

	/**
	 * Notification job repository.
	 *
	 * @var NotificationJobRepository
	 */
	private $notification_jobs;

	/**
	 * Submission receipt repository.
	 *
	 * @var SubmissionReceiptRepository
	 */
	private $receipts;

	/**
	 * Cloud sync outbox repository.
	 *
	 * @var SyncEventRepository
	 */
	private $sync_events;

	/**
	 * Constructor.
	 *
	 * The legacy repository arguments remain accepted to preserve the public
	 * composition signature while cleanup itself now uses set-based operations.
	 *
	 * @param \SUBIN_SettingsRepository $settings Settings repository.
	 * @param LeadRepository            $leads Lead repository.
	 * @param LeadFieldRepository       $fields Lead field repository.
	 * @param MailEventRepository       $mail_events Mail event repository.
	 * @param NotificationJobRepository|null $notification_jobs Notification jobs.
	 * @param SubmissionReceiptRepository|null $receipts Submission receipts.
	 * @param SyncEventRepository|null       $sync_events Cloud sync outbox.
	 */
	public function __construct(
		\SUBIN_SettingsRepository $settings,
		LeadRepository $leads,
		LeadFieldRepository $fields,
		MailEventRepository $mail_events,
		?NotificationJobRepository $notification_jobs = null,
		?SubmissionReceiptRepository $receipts = null,
		?SyncEventRepository $sync_events = null
	) {
		$this->settings          = $settings;
		$this->leads             = $leads;
		$this->notification_jobs = $notification_jobs ?: new NotificationJobRepository();
		$this->receipts          = $receipts ?: new SubmissionReceiptRepository();
		$this->sync_events       = $sync_events ?: new SyncEventRepository();

		unset( $fields, $mail_events );
	}

	/**
	 * Register cleanup hooks.
	 *
	 * @return void
	 */
	public function register() {
		add_action( self::CRON_HOOK, array( $this, 'cleanup_expired_leads' ) );
		add_action( self::CONTINUATION_HOOK, array( $this, 'cleanup_expired_leads' ) );
	}

	/**
	 * Delete expired leads independently of whether storage of new leads is enabled.
	 *
	 * @return int Number of deleted lead rows.
	 */
	public function cleanup_expired_leads() {
		$lock_token = $this->acquire_lock();
		if ( '' === $lock_token ) {
			return 0;
		}

		$settings   = $this->settings->all();
		$batch_size = max( 25, min( 1000, absint( apply_filters( 'subin_retention_batch_size', self::DEFAULT_BATCH_SIZE ) ) ) );
		$time_budget = max( 5, min( 50, absint( apply_filters( 'subin_retention_time_budget', self::DEFAULT_TIME_BUDGET ) ) ) );
		$started_at  = microtime( true );
		$lead_deadline = $started_at + max( 3, floor( $time_budget * 0.65 ) );
		$deadline      = $started_at + $time_budget;
		$deleted       = 0;
		$needs_follow_up = false;

		try {
			$retention_days = max( 0, min( 3650, absint( $settings['retention_days'] ?? 90 ) ) );
			$policy_ready = $this->reconcile_expiration_policy( $retention_days, $batch_size, $lead_deadline );
			if ( is_wp_error( $policy_ready ) ) {
				do_action( 'subin_retention_cleanup_failed', $policy_ready );
				$needs_follow_up = true;
			} elseif ( ! $policy_ready ) {
				$needs_follow_up = true;
			} elseif ( $retention_days > 0 ) {
				do {
					$lead_ids = $this->leads->get_expired_lead_ids( $batch_size );
					if ( empty( $lead_ids ) ) {
						$needs_follow_up = false;
						break;
					}

					$batch_deleted = $this->delete_lead_batch( $lead_ids );
					if ( is_wp_error( $batch_deleted ) ) {
						do_action( 'subin_retention_cleanup_failed', $batch_deleted );
						$needs_follow_up = true;
						break;
					}

					$deleted += absint( $batch_deleted );
					$needs_follow_up = count( $lead_ids ) >= $batch_size;
				} while ( $needs_follow_up && microtime( true ) < $lead_deadline );
			} else {
				$needs_follow_up = false;
			}
			$operational_result = $this->purge_operational_history( $settings, $batch_size, $deadline );
			if ( is_wp_error( $operational_result ) ) {
				do_action( 'subin_retention_cleanup_failed', $operational_result );
				$needs_follow_up = true;
			} else {
				$needs_follow_up = $needs_follow_up || $operational_result;
			}

			if ( $needs_follow_up ) {
				self::schedule_continuation();
			}

			do_action(
				'subin_retention_cleanup_completed',
				array(
					'deleted_leads'  => $deleted,
					'needs_follow_up' => $needs_follow_up,
					'duration'       => microtime( true ) - $started_at,
				)
			);

			return $deleted;
		} finally {
			$this->release_lock( $lock_token );
		}
	}

	/**
	 * Recalculate expiration timestamps when the configured policy changes.
	 *
	 * Deletion is paused until this cursor completes so switching to "Forever"
	 * can never delete rows using stale expiration timestamps.
	 *
	 * @param int   $retention_days Retention days, zero for forever.
	 * @param int   $batch_size Batch size.
	 * @param float $deadline Unix microtime deadline.
	 * @return bool|\WP_Error
	 */
	private function reconcile_expiration_policy( $retention_days, $batch_size, $deadline ) {
		global $wpdb;

		$retention_days = max( 0, min( 3650, absint( $retention_days ) ) );
		$state = get_option( self::POLICY_OPTION, array() );
		$state_days = is_array( $state ) && isset( $state['days'] ) ? absint( $state['days'] ) : null;
		if ( ! is_array( $state ) || null === $state_days || $state_days !== $retention_days ) {
			$state = array(
				'days'       => $retention_days,
				'last_id'    => 0,
				'complete'   => false,
				'started_at' => current_time( 'mysql' ),
			);
			update_option( self::POLICY_OPTION, $state, false );
		}

		if ( ! empty( $state['complete'] ) ) {
			return true;
		}

		$table = \SUBIN_Database::get_table_names()['leads'];
		do {
			$wpdb->last_error = '';
			$last_id = absint( $state['last_id'] ?? 0 );
			$ids = array_map(
				'absint',
				(array) $wpdb->get_col(
					$wpdb->prepare(
						"SELECT id FROM {$table} WHERE id > %d ORDER BY id ASC LIMIT %d",
						$last_id,
						$batch_size
					) // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
				)
			);
			if ( '' !== (string) $wpdb->last_error ) {
				return new \WP_Error(
					'subin_retention_policy_read_failed',
					__( 'Submission Inbox could not read leads while applying the retention policy.', 'submission-inbox' ),
					array( 'database_error' => sanitize_text_field( (string) $wpdb->last_error ) )
				);
			}

			$ids = array_values( array_filter( $ids ) );
			if ( empty( $ids ) ) {
				$state['complete']     = true;
				$state['completed_at'] = current_time( 'mysql' );
				update_option( self::POLICY_OPTION, $state, false );

				return true;
			}

			$placeholders = implode( ',', array_fill( 0, count( $ids ), '%d' ) );
			if ( $retention_days > 0 ) {
				$params = array_merge( array( $retention_days ), $ids );
				$query = $wpdb->prepare(
					"UPDATE {$table}
					SET expires_at = DATE_ADD(COALESCE(submitted_at, created_at), INTERVAL %d DAY)
					WHERE id IN ({$placeholders})",
					...$params
				); // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
			} else {
				$query = $wpdb->prepare(
					"UPDATE {$table} SET expires_at = NULL WHERE id IN ({$placeholders})",
					...$ids
				); // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
			}

			if ( false === $wpdb->query( $query ) ) { // phpcs:ignore WordPress.DB.PreparedSQL.NotPrepared
				return new \WP_Error(
					'subin_retention_policy_reconcile_failed',
					__( 'Submission Inbox could not apply the updated retention policy.', 'submission-inbox' ),
					array( 'database_error' => sanitize_text_field( (string) $wpdb->last_error ) )
				);
			}

			$state['last_id'] = max( $ids );
			update_option( self::POLICY_OPTION, $state, false );

			if ( count( $ids ) < $batch_size ) {
				$state['complete']     = true;
				$state['completed_at'] = current_time( 'mysql' );
				update_option( self::POLICY_OPTION, $state, false );

				return true;
			}
		} while ( microtime( true ) < $deadline );

		return false;
	}

	/**
	 * Delete one lead batch and every plugin-owned row that contains its private data.
	 *
	 * @param array<int,int> $lead_ids Lead ids.
	 * @return int|\WP_Error
	 */
	private function delete_lead_batch( array $lead_ids ) {
		global $wpdb;

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

		$tables = \SUBIN_Database::get_table_names();
		if ( false === $wpdb->query( 'START TRANSACTION' ) ) {
			return $this->operational_database_error(
				'subin_retention_transaction_failed',
				__( 'Submission Inbox could not start an expired-data transaction.', 'submission-inbox' ),
				'lead_transaction_start'
			);
		}

		$transaction_open = true;
		$stage            = 'lead_selection';
		try {
			$lock_placeholders = implode( ',', array_fill( 0, count( $lead_ids ), '%d' ) );
			$selection_args    = array_merge( $lead_ids, array( current_time( 'mysql' ) ) );
			$wpdb->last_error  = '';
			$lead_ids          = array_values(
				array_filter(
					array_map(
						'absint',
						(array) $wpdb->get_col(
							$wpdb->prepare(
								"SELECT id FROM {$tables['leads']}
								WHERE id IN ({$lock_placeholders})
									AND expires_at IS NOT NULL
									AND expires_at <> ''
									AND expires_at < %s
								ORDER BY id ASC FOR UPDATE",
								...$selection_args
							) // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
						)
					)
				)
			);
			if ( '' !== (string) $wpdb->last_error ) {
				throw new \RuntimeException( (string) $wpdb->last_error );
			}
			if ( empty( $lead_ids ) ) {
				$stage = 'lead_empty_commit';
				if ( false === $wpdb->query( 'COMMIT' ) ) {
					throw new \RuntimeException( (string) $wpdb->last_error );
				}
				$transaction_open = false;

				return 0;
			}

			$placeholders = implode( ',', array_fill( 0, count( $lead_ids ), '%d' ) );
			$stage        = 'related_mail_selection';
			$wpdb->last_error = '';
			$mail_event_ids = array_values(
				array_filter(
					array_map(
						'absint',
						(array) $wpdb->get_col(
							$wpdb->prepare(
								"SELECT id FROM {$tables['mail_events']}
								WHERE lead_id IN ({$placeholders})
								ORDER BY id ASC FOR UPDATE",
								...$lead_ids
							)
						)
					)
			)
			);
			if ( '' !== (string) $wpdb->last_error ) {
				throw new \RuntimeException( (string) $wpdb->last_error );
			}

			$stage  = 'sync_history_lock';
			$locked = $this->sync_events->lock_deletion_events_in_transaction( $lead_ids, $mail_event_ids );
			if ( is_wp_error( $locked ) ) {
				return $this->rollback_retention_error( $locked, 'lead_sync_history_lock' );
			}

			$settings                = $this->settings->all();
			$enqueue_without_history = ! empty( $settings['cloud_sync_enabled'] );
			$deleted_at              = current_time( 'mysql' );
			foreach ( $mail_event_ids as $event_id ) {
				$stage  = 'related_mail_tombstone';
				$staged = $this->sync_events->stage_mail_event_delete_in_transaction( $event_id, $deleted_at, 'parent_deleted', $enqueue_without_history );
				if ( is_wp_error( $staged ) ) {
					return $this->rollback_retention_error( $staged, $stage );
				}
			}
			foreach ( $lead_ids as $lead_id ) {
				$stage  = 'lead_tombstone';
				$staged = $this->sync_events->stage_lead_delete_in_transaction( $lead_id, $deleted_at, 'retention', $enqueue_without_history );
				if ( is_wp_error( $staged ) ) {
					return $this->rollback_retention_error( $staged, $stage );
				}
			}

			$queries = array();

			if ( \SUBIN_Database::table_exists( $tables['notification_jobs'] ) ) {
				$queries[] = array(
					"DELETE FROM {$tables['notification_jobs']} WHERE lead_id IN ({$placeholders})",
					$lead_ids,
				);
			}

			$queries[] = array(
				"UPDATE {$tables['mail_events']} events
				LEFT JOIN {$tables['leads']} leads ON leads.id = events.lead_id
					SET events.provider = COALESCE(NULLIF(events.provider, ''), NULLIF(leads.provider, ''), NULLIF(events.source, '')),
						events.form_id = COALESCE(NULLIF(events.form_id, ''), NULLIF(leads.form_id, '')),
						events.form_name = COALESCE(NULLIF(events.form_name, ''), NULLIF(leads.form_name, '')),
						events.lead_id = NULL
					WHERE events.lead_id IN ({$placeholders})",
					$lead_ids,
			);
			$queries[] = array( "DELETE FROM {$tables['lead_fields']} WHERE lead_id IN ({$placeholders})", $lead_ids );
			$queries[] = array( "DELETE FROM {$tables['lead_notes']} WHERE lead_id IN ({$placeholders})", $lead_ids );
			$queries[] = array( "DELETE FROM {$tables['lead_activity']} WHERE lead_id IN ({$placeholders})", $lead_ids );

			foreach ( $queries as $query ) {
				$stage  = 'lead_child_cleanup';
				$result = $wpdb->query( $wpdb->prepare( $query[0], ...$query[1] ) ); // phpcs:ignore WordPress.DB.PreparedSQL.NotPrepared
				if ( false === $result ) {
					throw new \RuntimeException( (string) $wpdb->last_error );
				}
			}

			$stage   = 'lead_parent_delete';
			$deleted = $wpdb->query(
				$wpdb->prepare(
					"DELETE FROM {$tables['leads']} WHERE id IN ({$placeholders})",
					...$lead_ids
				) // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
			);
			if ( false === $deleted ) {
				throw new \RuntimeException( (string) $wpdb->last_error );
			}

			$stage = 'lead_transaction_commit';
			if ( false === $wpdb->query( 'COMMIT' ) ) {
				throw new \RuntimeException( (string) $wpdb->last_error );
			}
			$transaction_open = false;

			return absint( $deleted );
		} catch ( \Throwable $throwable ) {
			return $this->rollback_retention_error(
				new \WP_Error(
					'subin_retention_batch_failed',
					__( 'Submission Inbox could not delete an expired data batch.', 'submission-inbox' ),
					array(
						'operation'      => sanitize_key( $stage ),
						'database_error' => sanitize_text_field( $throwable->getMessage() ),
						'retryable'      => true,
					)
				),
				$stage,
				$transaction_open
			);
		}
	}

	/**
	 * Purge old operational rows and redact bounded notification payloads.
	 *
	 * @param array<string,mixed> $settings Settings.
	 * @param int                 $batch_size Batch size.
	 * @param float               $deadline Unix microtime deadline.
	 * @return bool|\WP_Error Whether another bounded pass should run soon, or a retryable failure.
	 */
	private function purge_operational_history( array $settings, $batch_size, $deadline ) {
		global $wpdb;

		$tables = \SUBIN_Database::get_table_names();
		$mail_days = \SUBIN_SettingsRepository::resolve_mail_log_retention_days( $settings );
		$payload_days = $this->retention_days( $settings, 'sync_payload_retention_days', self::DEFAULT_SYNC_PAYLOAD_RETENTION_DAYS, 'subin_sync_payload_retention_days' );
		$terminal_days = $this->retention_days( $settings, 'sync_event_retention_days', self::DEFAULT_SYNC_TERMINAL_RETENTION_DAYS, 'subin_sync_event_retention_days' );
		$failed_days = $this->retention_days( $settings, 'sync_failed_retention_days', self::DEFAULT_SYNC_FAILED_RETENTION_DAYS, 'subin_sync_failed_retention_days' );
		$notification_payload_days = min(
			30,
			max(
				1,
				$this->retention_days(
					$settings,
					'notification_payload_retention_days',
					self::DEFAULT_NOTIFICATION_PAYLOAD_RETENTION_DAYS,
					'subin_notification_payload_retention_days'
				)
			)
		);
		$notification_days = $this->retention_days( $settings, 'notification_job_retention_days', self::DEFAULT_NOTIFICATION_RETENTION_DAYS, 'subin_notification_job_retention_days' );
		$notification_failed_days = $this->retention_days( $settings, 'notification_failed_retention_days', self::DEFAULT_NOTIFICATION_FAILED_RETENTION_DAYS, 'subin_notification_failed_retention_days' );
		$analytics_days = $this->retention_days( $settings, 'form_analytics_retention_days', self::DEFAULT_FORM_ANALYTICS_RETENTION_DAYS, 'subin_form_analytics_retention_days' );
		$notification_payload_batch_size = min( $batch_size, NotificationJobRepository::MAX_REDACTION_BATCH );
		$has_rate_limit_table = \SUBIN_Database::table_exists( $tables['rate_limits'] );
		$has_form_analytics_table = ! empty( $tables['form_analytics'] ) && \SUBIN_Database::table_exists( $tables['form_analytics'] );

		do {
			$more = false;
			if ( $has_rate_limit_table && microtime( true ) < $deadline ) {
				$result = $this->run_operational_query(
					$wpdb->prepare(
						"DELETE FROM {$tables['rate_limits']}
						WHERE expires_at < %s
						ORDER BY expires_at ASC
						LIMIT %d",
						current_time( 'mysql' ),
						$batch_size
					), // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
					'rate_limit_expiration'
				);
				if ( is_wp_error( $result ) ) {
					return $result;
				}
				$more = $more || $batch_size === absint( $result );
			}

			if ( $has_form_analytics_table && $analytics_days > 0 && microtime( true ) < $deadline ) {
				$result = $this->run_operational_query(
					$wpdb->prepare(
						"DELETE FROM {$tables['form_analytics']}
						WHERE occurred_on < %s
						ORDER BY occurred_on ASC, id ASC
						LIMIT %d",
						substr( $this->cutoff( $analytics_days ), 0, 10 ),
						$batch_size
					), // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
					'form_analytics_expiration'
				);
				if ( is_wp_error( $result ) ) {
					return $result;
				}
				$more = $more || $batch_size === absint( $result );
			}

			if ( microtime( true ) < $deadline ) {
				$redacted = $this->notification_jobs->redact_terminal_payloads(
					$this->cutoff( $notification_payload_days ),
					$notification_payload_batch_size
				);
				if ( is_wp_error( $redacted ) ) {
					return $redacted;
				}
				$more = $more || $notification_payload_batch_size === absint( $redacted );
			}

			if ( microtime( true ) < $deadline ) {
				$receipts_deleted = $this->receipts->cleanup_expired( $batch_size );
				if ( is_wp_error( $receipts_deleted ) ) {
					return $receipts_deleted;
				}
				$more = $more || $batch_size === absint( $receipts_deleted );
			}


			if ( $payload_days > 0 && microtime( true ) < $deadline ) {
				$cutoff = $this->cutoff( $payload_days );
				$result = $this->run_operational_query(
					$wpdb->prepare(
						"UPDATE {$tables['sync_events']}
						SET payload_json = NULL
						WHERE status IN ('sent', 'skipped')
							AND updated_at < %s
							AND payload_json IS NOT NULL
						ORDER BY id ASC
						LIMIT %d",
						$cutoff,
						$batch_size
					), // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
					'sync_payload_redaction'
				);
				if ( is_wp_error( $result ) ) {
					return $result;
				}
				$more = $more || $batch_size === absint( $result );
			}

			if ( $terminal_days > 0 && microtime( true ) < $deadline ) {
				$result = $this->run_operational_query(
					$wpdb->prepare(
						"DELETE FROM {$tables['sync_events']}
						WHERE status IN ('sent', 'skipped')
							AND updated_at < %s
							AND NOT (
								event_type = 'email.deleted'
								AND EXISTS (
									SELECT 1 FROM {$tables['mail_events']} retained_mail
									WHERE CAST(retained_mail.id AS CHAR) = {$tables['sync_events']}.object_id
								)
							)
						ORDER BY id ASC
						LIMIT %d",
						$this->cutoff( $terminal_days ),
						$batch_size
					), // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
					'sync_terminal_expiration'
				);
				if ( is_wp_error( $result ) ) {
					return $result;
				}
				$more = $more || $batch_size === absint( $result );
			}

			if ( $failed_days > 0 && microtime( true ) < $deadline ) {
				$result = $this->run_operational_query(
					$wpdb->prepare(
						"DELETE FROM {$tables['sync_events']}
						WHERE status = 'failed'
							AND attempts >= 6
							AND event_type NOT LIKE %s
							AND updated_at < %s
						ORDER BY id ASC
						LIMIT %d",
						'%.deleted',
						$this->cutoff( $failed_days ),
						$batch_size
					), // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
					'sync_failed_expiration'
				);
				if ( is_wp_error( $result ) ) {
					return $result;
				}
				$more = $more || $batch_size === absint( $result );
			}

			if ( $mail_days > 0 && microtime( true ) < $deadline ) {
				$mail_purged = $this->purge_mail_event_batch( $tables, $this->cutoff( $mail_days ), $batch_size );
				if ( is_wp_error( $mail_purged ) ) {
					return $mail_purged;
				}
				$more = $more || $batch_size === absint( $mail_purged );
			}

			if ( microtime( true ) < $deadline ) {
				if ( $notification_days > 0 ) {
					$result = $this->run_operational_query(
						$wpdb->prepare(
							"DELETE FROM {$tables['notification_jobs']}
							WHERE status IN ('sent', 'skipped')
								AND updated_at < %s
							ORDER BY id ASC
							LIMIT %d",
							$this->cutoff( $notification_days ),
							$batch_size
						), // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
						'notification_terminal_expiration'
					);
					if ( is_wp_error( $result ) ) {
						return $result;
					}
					$more = $more || $batch_size === absint( $result );
				}

				if ( $notification_failed_days > 0 && microtime( true ) < $deadline ) {
					$result = $this->run_operational_query(
						$wpdb->prepare(
							"DELETE FROM {$tables['notification_jobs']}
							WHERE status = 'failed'
								AND attempts >= 6
								AND updated_at < %s
							ORDER BY id ASC
							LIMIT %d",
							$this->cutoff( $notification_failed_days ),
							$batch_size
						), // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
						'notification_failed_expiration'
					);
					if ( is_wp_error( $result ) ) {
						return $result;
					}
					$more = $more || $batch_size === absint( $result );
				}
			}
		} while ( $more && microtime( true ) < $deadline );

		return $more || microtime( true ) >= $deadline;
	}

	/**
	 * Purge one mail-event batch and any outbox rows referring to those events.
	 *
	 * @param array<string,string> $tables Table names.
	 * @param string               $cutoff Cutoff datetime.
	 * @param int                  $batch_size Batch size.
	 * @return int|\WP_Error
	 */
	private function purge_mail_event_batch( array $tables, $cutoff, $batch_size ) {
		global $wpdb;

		$batch_size = max( 1, absint( $batch_size ) );
		if ( false === $wpdb->query( 'START TRANSACTION' ) ) {
			return $this->operational_database_error(
				'subin_mail_retention_failed',
				__( 'Submission Inbox could not start an email-log retention transaction.', 'submission-inbox' ),
				'mail_transaction_start'
			);
		}

		$transaction_open = true;
		$stage            = 'mail_event_selection';
		try {
			$wpdb->last_error = '';
			$event_ids = array_values(
				array_filter(
					array_map(
						'absint',
						(array) $wpdb->get_col(
							$wpdb->prepare(
								"SELECT id FROM {$tables['mail_events']}
								WHERE created_at < %s
								ORDER BY id ASC
								LIMIT %d
								FOR UPDATE",
								$cutoff,
								$batch_size
							) // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
						)
					)
				)
			);
			if ( '' !== (string) $wpdb->last_error ) {
				throw new \RuntimeException( (string) $wpdb->last_error );
			}

			if ( empty( $event_ids ) ) {
				$stage = 'mail_empty_commit';
				if ( false === $wpdb->query( 'COMMIT' ) ) {
					throw new \RuntimeException( (string) $wpdb->last_error );
				}
				$transaction_open = false;

				return 0;
			}

			$placeholders = implode( ',', array_fill( 0, count( $event_ids ), '%d' ) );
			$stage        = 'mail_sync_history_lock';
			$locked       = $this->sync_events->lock_deletion_events_in_transaction( array(), $event_ids );
			if ( is_wp_error( $locked ) ) {
				return $this->rollback_retention_error( $locked, $stage );
			}

			$settings                = $this->settings->all();
			$enqueue_without_history = ! empty( $settings['cloud_sync_enabled'] );
			$deleted_at              = current_time( 'mysql' );
			foreach ( $event_ids as $event_id ) {
				$stage  = 'mail_tombstone';
				$staged = $this->sync_events->stage_mail_event_delete_in_transaction( $event_id, $deleted_at, 'retention', $enqueue_without_history );
				if ( is_wp_error( $staged ) ) {
					return $this->rollback_retention_error( $staged, $stage );
				}
			}

			$stage = 'mail_event_expiration';
			$mail_deleted = $wpdb->query(
				$wpdb->prepare(
					"DELETE FROM {$tables['mail_events']} WHERE id IN ({$placeholders})",
					...$event_ids
				) // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
			);
			if ( false === $mail_deleted ) {
				throw new \RuntimeException( (string) $wpdb->last_error );
			}

			$stage = 'mail_transaction_commit';
			if ( false === $wpdb->query( 'COMMIT' ) ) {
				throw new \RuntimeException( (string) $wpdb->last_error );
			}
			$transaction_open = false;

			return absint( $mail_deleted );
		} catch ( \Throwable $throwable ) {
			return $this->rollback_retention_error(
				new \WP_Error(
					'subin_mail_retention_failed',
					__( 'Submission Inbox could not purge an old email-log batch.', 'submission-inbox' ),
					array(
						'operation'      => sanitize_key( $stage ),
						'database_error' => sanitize_text_field( $throwable->getMessage() ),
						'retryable'      => true,
					)
				),
				$stage,
				$transaction_open
			);
		}
	}

	/**
	 * Run one bounded operational cleanup query without hiding database failures.
	 *
	 * @param string $query Prepared SQL query.
	 * @param string $operation Stable operation identifier.
	 * @return int|\WP_Error
	 */
	private function run_operational_query( $query, $operation ) {
		global $wpdb;

		$result = $wpdb->query( $query ); // phpcs:ignore WordPress.DB.PreparedSQL.NotPrepared -- Query is prepared by the caller.
		if ( false !== $result ) {
			return absint( $result );
		}

		return $this->operational_database_error(
			'subin_operational_retention_failed',
			__( 'Submission Inbox could not purge an operational-history batch.', 'submission-inbox' ),
			$operation
		);
	}

	/**
	 * Build an observable database failure that the caller must retry.
	 *
	 * @param string $code Error code.
	 * @param string $message User-safe error message.
	 * @param string $operation Stable operation identifier.
	 * @return \WP_Error
	 */
	private function operational_database_error( $code, $message, $operation ) {
		global $wpdb;

		return new \WP_Error(
			sanitize_key( $code ),
			$message,
			array(
				'operation'      => sanitize_key( $operation ),
				'database_error' => sanitize_text_field( (string) $wpdb->last_error ),
				'retryable'      => true,
			)
		);
	}

	/**
	 * Roll back an open retention transaction without losing the original cause.
	 *
	 * @param \WP_Error $error Original failure.
	 * @param string    $operation Stable operation identifier.
	 * @param bool      $transaction_open Whether rollback is still required.
	 * @return \WP_Error
	 */
	private function rollback_retention_error( \WP_Error $error, $operation, $transaction_open = true ) {
		global $wpdb;

		$data = $error->get_error_data();
		$data = is_array( $data ) ? $data : array();
		$data['operation'] = sanitize_key( (string) ( $data['operation'] ?? $operation ) );
		$data['retryable'] = true;
		if ( $transaction_open && false === $wpdb->query( 'ROLLBACK' ) ) {
			$data['rollback_error'] = sanitize_text_field( (string) $wpdb->last_error );
		}

		return new \WP_Error(
			(string) $error->get_error_code(),
			(string) $error->get_error_message(),
			$data
		);
	}

	/**
	 * Resolve a bounded retention setting with a safe default and filter.
	 *
	 * A value of zero explicitly disables that purge policy.
	 *
	 * @param array<string,mixed> $settings Settings.
	 * @param string              $key Setting key.
	 * @param int                 $default Default days.
	 * @param string              $filter Filter name.
	 * @return int
	 */
	private function retention_days( array $settings, $key, $default, $filter ) {
		$value = array_key_exists( $key, $settings ) ? absint( $settings[ $key ] ) : absint( $default );

		return max( 0, min( 3650, absint( apply_filters( $filter, $value ) ) ) );
	}

	/**
	 * Build a local-site datetime cutoff.
	 *
	 * @param int $days Retention days.
	 * @return string
	 */
	private function cutoff( $days ) {
		return wp_date( 'Y-m-d H:i:s', time() - ( absint( $days ) * DAY_IN_SECONDS ), wp_timezone() );
	}

	/**
	 * Acquire an atomic cleanup lock.
	 *
	 * @return string Empty when another worker owns a valid lock.
	 */
	private function acquire_lock() {
		$current = get_option( self::LOCK_OPTION, false );
		if ( is_array( $current ) && ! empty( $current['acquired_at'] ) ) {
			if ( ( time() - absint( $current['acquired_at'] ) ) < self::LOCK_TTL ) {
				return '';
			}
			delete_option( self::LOCK_OPTION );
		} elseif ( false !== $current ) {
			delete_option( self::LOCK_OPTION );
		}

		$token = wp_generate_uuid4();
		$locked = add_option(
			self::LOCK_OPTION,
			array(
				'token'       => $token,
				'acquired_at' => time(),
			),
			'',
			false
		);

		return $locked ? $token : '';
	}

	/**
	 * Release the cleanup lock when owned by this worker.
	 *
	 * @param string $token Lock token.
	 * @return void
	 */
	private function release_lock( $token ) {
		$current = get_option( self::LOCK_OPTION, array() );
		$current_token = is_array( $current ) ? (string) ( $current['token'] ?? '' ) : '';

		if ( '' !== $current_token && hash_equals( $current_token, (string) $token ) ) {
			delete_option( self::LOCK_OPTION );
		}
	}

	/**
	 * Schedule daily cleanup.
	 *
	 * @return void
	 */
	public static function schedule() {
		if ( ! wp_next_scheduled( self::CRON_HOOK ) ) {
			wp_schedule_event( time() + HOUR_IN_SECONDS, 'daily', self::CRON_HOOK );
		}
	}

	/**
	 * Schedule a near-term continuation without duplicating pending continuations.
	 *
	 * @return void
	 */
	private static function schedule_continuation() {
		if ( ! wp_next_scheduled( self::CONTINUATION_HOOK ) ) {
			wp_schedule_single_event( time() + MINUTE_IN_SECONDS, self::CONTINUATION_HOOK );
		}
	}

	/**
	 * Clear scheduled cleanup and its lock.
	 *
	 * @return void
	 */
	public static function unschedule() {
		wp_clear_scheduled_hook( self::CRON_HOOK );
		wp_clear_scheduled_hook( self::CONTINUATION_HOOK );
		delete_option( self::LOCK_OPTION );
	}
}
