<?php
/**
 * The import itself: one resumable job, whatever the source.
 *
 * @package AiRankSuite
 */

declare(strict_types=1);

namespace AiRankSuite\Modules\ImportExport\Services;

use AiRankSuite\Support\Database;
use AiRankSuite\Support\Job;
use WP_Error;

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

/**
 * An import on a real site is not a button that runs for two seconds. A shop
 * with 40,000 products and a decade of Yoast data is minutes of work, on a
 * host that will cut the request off after thirty seconds, from a browser tab
 * the customer will close. So this is a job like the others: a small state
 * option, a keyset cursor per phase, as much work per tick as the time budget
 * allows, and a cron resumer for the tab that went away.
 *
 * Phases run in order and each is skipped when the run did not ask for it:
 *
 *   settings → posts → terms → redirects → finish
 *
 * plus two of its own: `file` for a CSV or JSON upload, and `undo`, which
 * walks the journal backwards and puts everything back.
 *
 * The safety rules are in the writer (never overwrite by default, journal
 * every write, never touch the source plugin). What lives here is the part
 * that makes them hold at scale: one record failing is one logged error and
 * the run continues, a tick that dies is retried from a cursor that never
 * moved past unwritten work, and a second run cannot start on top of a first.
 */
final class ImportRun {

	public const JOB  = 'airanksuite_import_job';
	public const FILE = 'airanksuite_import_file';

	private const POST_BATCH  = 40;
	private const TERM_BATCH  = 40;
	private const RED_BATCH   = 50;
	private const UNDO_BATCH  = 200;

	private ImportWriter $writer;

	public function __construct( ?ImportWriter $writer = null ) {
		$this->writer = $writer ?? new ImportWriter();
	}

	/**
	 * Start a run (or an undo). Returns the job state, or a WP_Error when the
	 * request cannot be honoured — an unknown source, a run already going, an
	 * undo of something that was already undone.
	 *
	 * @param array<string, mixed> $args source|undo, mode, dry_run, parts, file.
	 * @return array<string, mixed>|WP_Error
	 */
	public function start( array $args ) {
		$current = get_option( self::JOB );
		if ( is_array( $current ) && 'running' === ( $current['state'] ?? '' ) ) {
			return array_merge( $current, array( 'busy' => true ) );
		}

		$undo_of = (string) ( $args['undo'] ?? '' );
		if ( '' !== $undo_of ) {
			$record = ImportLog::run( $undo_of );
			if ( null === $record ) {
				return new WP_Error( 'airanksuite_import_unknown', __( 'That import is no longer on record.', 'ai-rank-suite' ), array( 'status' => 404 ) );
			}
			if ( ! empty( $record['undone'] ) ) {
				return new WP_Error( 'airanksuite_import_undone', __( 'That import has already been undone.', 'ai-rank-suite' ), array( 'status' => 409 ) );
			}
			if ( empty( $record['undoable'] ) ) {
				return new WP_Error( 'airanksuite_import_not_undoable', __( 'That import cannot be undone — it was a preview, or its journal has been cleared.', 'ai-rank-suite' ), array( 'status' => 409 ) );
			}
			$job = array(
				'state'     => 'running',
				'import_id' => $undo_of,
				'phase'     => 'undo',
				'undo'      => true,
				'cursor'    => 0,
				'counts'    => array( 'restored' => 0 ),
				'started'   => time(),
				'ended'     => 0,
			);
			update_option( self::JOB, $job, false );
			return $job;
		}

		$source = (string) ( $args['source'] ?? '' );
		if ( 'file' !== $source && null === ImportRegistry::get( $source ) ) {
			return new WP_Error( 'airanksuite_import_source', __( 'Unknown import source.', 'ai-rank-suite' ), array( 'status' => 400 ) );
		}
		$parts = array_values( array_intersect( array( 'settings', 'posts', 'terms', 'redirects' ), (array) ( $args['parts'] ?? array( 'settings', 'posts', 'terms', 'redirects' ) ) ) );
		if ( 'file' === $source ) {
			$file = (array) ( $args['file'] ?? array() );
			$rows = FileImport::rows( (string) ( $file['content'] ?? '' ), (string) ( $file['format'] ?? 'csv' ) );
			if ( empty( $rows ) ) {
				return new WP_Error( 'airanksuite_import_file', __( 'That file holds no rows this import can read.', 'ai-rank-suite' ), array( 'status' => 400 ) );
			}
			update_option(
				self::FILE,
				array(
					'rows'  => $rows,
					'match' => (string) ( $file['match'] ?? 'id' ),
					'map'   => (array) ( $file['map'] ?? array() ),
				),
				false
			);
			$parts = array( 'file' );
		}

		$import_id = substr( md5( uniqid( 'airanksuite-import', true ) ), 0, 12 );
		$job       = array(
			'state'     => 'running',
			'import_id' => $import_id,
			'source'    => $source,
			'mode'      => 'overwrite' === ( $args['mode'] ?? 'fill' ) ? 'overwrite' : 'fill',
			'dry_run'   => ! empty( $args['dry_run'] ),
			'parts'     => $parts,
			'phase'     => $parts[0] ?? 'finish',
			'cursor'    => 0,
			'counts'    => array( 'posts' => 0, 'terms' => 0, 'redirects' => 0, 'settings' => 0, 'errors' => 0, 'unmatched' => 0, 'seen' => 0, 'kept' => 0 ),
			// How much there is to do, so the screen can draw a bar that means
			// something rather than a stripe that spins.
			'total'     => self::total_for( $source, $parts ),
			'fields'    => array(),
			'started'   => time(),
			'ended'     => 0,
		);
		update_option( self::JOB, $job, false );
		ImportLog::save(
			$import_id,
			array(
				'source'   => $source,
				'mode'     => $job['mode'],
				'dry_run'  => $job['dry_run'],
				'parts'    => $parts,
				'state'    => 'running',
				'started'  => time(),
				'ended'    => 0,
				'counts'   => $job['counts'],
				'fields'   => array(),
				'errors'   => array(),
				'undoable' => ! $job['dry_run'],
				'undone'   => 0,
			)
		);
		ImportLog::prune();
		return $job;
	}

	/**
	 * What a run has ahead of it: the records the source holds for the parts
	 * that were asked for. An estimate by nature — a source can gain a post
	 * while the run walks it — but an honest one, and enough for a bar.
	 *
	 * @param array<int, string> $parts
	 */
	private static function total_for( string $source, array $parts ): int {
		if ( 'file' === $source ) {
			$file = get_option( self::FILE );
			return is_array( $file ) ? count( (array) ( $file['rows'] ?? array() ) ) : 0;
		}
		$adapter = ImportRegistry::get( $source );
		if ( ! $adapter ) {
			return 0;
		}
		$counts = $adapter->counts();
		$total  = 0;
		if ( in_array( 'posts', $parts, true ) ) {
			$total += (int) ( $counts['posts'] ?? 0 );
		}
		if ( in_array( 'terms', $parts, true ) ) {
			$total += (int) ( $counts['terms'] ?? 0 );
		}
		if ( in_array( 'redirects', $parts, true ) ) {
			$total += (int) ( $counts['redirects'] ?? 0 );
		}
		return $total;
	}

	/**
	 * @return array<string, mixed>
	 */
	public function state(): array {
		$job = get_option( self::JOB );
		return is_array( $job ) ? $job : array( 'state' => 'idle' );
	}

	/**
	 * Stop a run. What it already wrote stays — and stays undoable, which is
	 * the honest outcome: a half-applied import that can be taken back in one
	 * click is safer than one that rolls itself back while the customer
	 * watches a progress bar go backwards.
	 *
	 * @return array<string, mixed>
	 */
	public function cancel(): array {
		$job = get_option( self::JOB );
		if ( ! is_array( $job ) || 'running' !== ( $job['state'] ?? '' ) ) {
			return is_array( $job ) ? $job : array( 'state' => 'idle' );
		}
		$job['state'] = 'cancelled';
		$job['ended'] = time();
		update_option( self::JOB, $job, false );
		delete_option( self::FILE );
		if ( empty( $job['undo'] ) ) {
			ImportLog::save(
				(string) $job['import_id'],
				array(
					'state'    => 'cancelled',
					'ended'    => time(),
					'counts'   => $job['counts'],
					'fields'   => $job['fields'] ?? array(),
					'undoable' => empty( $job['dry_run'] ),
				)
			);
		}
		return $job;
	}

	/**
	 * One tick: work until the time budget is up, then commit — but only if
	 * this run is still the one that is running.
	 *
	 * @return array<string, mixed>
	 */
	public function tick(): array {
		$job = get_option( self::JOB );
		if ( ! is_array( $job ) || 'running' !== ( $job['state'] ?? '' ) ) {
			return is_array( $job ) ? $job : array( 'state' => 'idle' );
		}
		$deadline = time() + Job::budget_seconds();
		$this->writer->begin( (string) $job['import_id'], ! empty( $job['dry_run'] ) );

		$suspended = wp_suspend_cache_addition();
		wp_suspend_cache_addition( true );
		try {
			do {
				$phase = (string) $job['phase'];
				if ( 'finish' === $phase ) {
					break;
				}
				$method = 'phase_' . $phase;
				if ( ! method_exists( $this, $method ) ) {
					$job['phase'] = $this->next_phase( $job );
					continue;
				}
				$this->{$method}( $job, $deadline );
			} while ( time() < $deadline && ! Job::memory_tight() && 'finish' !== $job['phase'] );
		} finally {
			wp_suspend_cache_addition( $suspended );
		}

		// Field counters are per tick; the run keeps the sum.
		$fields = (array) ( $job['fields'] ?? array() );
		foreach ( $this->writer->fields() as $field => $n ) {
			if ( 'kept' === $field ) {
				// Not a field that was written — a value that was left alone.
				$job['counts']['kept'] = (int) ( $job['counts']['kept'] ?? 0 ) + (int) $n;
				continue;
			}
			$fields[ $field ] = (int) ( $fields[ $field ] ?? 0 ) + (int) $n;
		}
		$job['fields'] = $fields;

		if ( 'finish' === $job['phase'] ) {
			$job['state'] = 'done';
			$job['ended'] = time();
		}
		return $this->commit( $job );
	}

	/**
	 * Compare-and-set: a cancel (or a newer run) that landed while this tick
	 * was working wins, and this tick's state is dropped rather than written
	 * over it.
	 *
	 * @param array<string, mixed> $job
	 * @return array<string, mixed>
	 */
	private function commit( array $job ): array {
		global $wpdb;
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching
		$raw  = $wpdb->get_var( $wpdb->prepare( "SELECT option_value FROM {$wpdb->options} WHERE option_name = %s LIMIT 1", self::JOB ) );
		$live = null === $raw ? false : maybe_unserialize( $raw );
		if ( ! is_array( $live ) || 'running' !== ( $live['state'] ?? '' ) || ( $live['import_id'] ?? '' ) !== ( $job['import_id'] ?? '-' ) ) {
			wp_cache_delete( self::JOB, 'options' );
			return is_array( $live ) ? $live : array( 'state' => 'idle' );
		}
		update_option( self::JOB, $job, false );

		if ( ! empty( $job['undo'] ) ) {
			if ( 'done' === ( $job['state'] ?? '' ) ) {
				ImportLog::save( (string) $job['import_id'], array( 'undone' => time(), 'undoable' => false, 'restored' => (int) ( $job['counts']['restored'] ?? 0 ) ) );
			}
			return $job;
		}
		ImportLog::save(
			(string) $job['import_id'],
			array(
				'state'  => (string) $job['state'],
				'ended'  => (int) $job['ended'],
				'counts' => (array) $job['counts'],
				'fields' => (array) $job['fields'],
			)
		);
		if ( 'done' === ( $job['state'] ?? '' ) ) {
			delete_option( self::FILE );
			delete_transient( \AiRankSuite\Modules\LlmsTxt\LlmsTxtModule::CACHE_KEY );
			delete_transient( \AiRankSuite\Modules\LlmsTxt\LlmsTxtModule::CACHE_KEY_FULL );
			\AiRankSuite\Support\Inventory::flush();
			/*
			 * And the library counts.
			 *
			 * An import writes post meta directly — it never calls
			 * wp_update_post, so `save_post` does not fire and the hook that
			 * flushes these counts never runs. The five-minute cache then
			 * outlived the import, and the screen kept saying "160 pages with
			 * no keyword" on a site that had just imported 151 of them. The
			 * customer's conclusion was the obvious one: the import does
			 * nothing.
			 */
			\AiRankSuite\Modules\ContentAnalyzer\Services\Library::flush();
			delete_transient( 'airanksuite_kpis' );
			delete_transient( 'airanksuite_dashboard_trends' );
		}
		return $job;
	}

	/**
	 * @param array<string, mixed> $job
	 */
	private function next_phase( array $job ): string {
		$order = array( 'settings', 'posts', 'terms', 'redirects', 'file' );
		$parts = (array) ( $job['parts'] ?? array() );
		$seen  = false;
		foreach ( $order as $phase ) {
			if ( $phase === $job['phase'] ) {
				$seen = true;
				continue;
			}
			if ( $seen && in_array( $phase, $parts, true ) ) {
				return $phase;
			}
		}
		return 'finish';
	}

	/**
	 * @param array<string, mixed> $job
	 */
	private function advance( array &$job ): void {
		$job['cursor'] = 0;
		$job['phase']  = $this->next_phase( $job );
	}

	// ------------------------------------------------------------- phases --

	/**
	 * @param array<string, mixed> $job
	 */
	private function phase_settings( array &$job, int $deadline ): void {
		$adapter = ImportRegistry::get( (string) $job['source'] );
		if ( $adapter ) {
			try {
				$applied                    = $this->writer->settings( $adapter->settings(), (string) $job['mode'] );
				$job['counts']['settings'] += $applied;
			} catch ( \Throwable $e ) {
				ImportLog::error( (string) $job['import_id'], 'settings', $e->getMessage() );
				++$job['counts']['errors'];
			}
		}
		$this->advance( $job );
	}

	/**
	 * @param array<string, mixed> $job
	 */
	private function phase_posts( array &$job, int $deadline ): void {
		$adapter = ImportRegistry::get( (string) $job['source'] );
		if ( ! $adapter ) {
			$this->advance( $job );
			return;
		}
		$ids = $adapter->post_ids( (int) $job['cursor'], self::POST_BATCH );
		if ( empty( $ids ) ) {
			$this->advance( $job );
			return;
		}
		foreach ( $ids as $post_id ) {
			$job['cursor'] = max( (int) $job['cursor'], (int) $post_id );
			++$job['counts']['seen'];
			try {
				$record = $adapter->post_record( (int) $post_id );
				/**
				 * Filter one import record before it is written.
				 *
				 * @param array<string, mixed> $record
				 * @param int                  $object_id
				 * @param string               $source
				 */
				$record = (array) apply_filters( 'airanksuite/import/record', $record, (int) $post_id, (string) $job['source'] );
				if ( ! empty( $record ) && $this->writer->post( (int) $post_id, $record, (string) $job['mode'] ) ) {
					++$job['counts']['posts'];
				}
			} catch ( \Throwable $e ) {
				ImportLog::error( (string) $job['import_id'], 'post ' . $post_id, $e->getMessage() );
				++$job['counts']['errors'];
			}
			if ( time() >= $deadline || Job::memory_tight() ) {
				return;
			}
		}
	}

	/**
	 * @param array<string, mixed> $job
	 */
	private function phase_terms( array &$job, int $deadline ): void {
		$adapter = ImportRegistry::get( (string) $job['source'] );
		if ( ! $adapter ) {
			$this->advance( $job );
			return;
		}
		$ids = $adapter->term_ids( (int) $job['cursor'], self::TERM_BATCH );
		if ( empty( $ids ) ) {
			$this->advance( $job );
			return;
		}
		foreach ( $ids as $term_id ) {
			$job['cursor'] = max( (int) $job['cursor'], (int) $term_id );
			++$job['counts']['seen'];
			try {
				$record = $adapter->term_record( (int) $term_id );
				$record = (array) apply_filters( 'airanksuite/import/record', $record, (int) $term_id, (string) $job['source'] );
				if ( ! empty( $record ) && $this->writer->term( (int) $term_id, $record, (string) $job['mode'] ) ) {
					++$job['counts']['terms'];
				}
			} catch ( \Throwable $e ) {
				ImportLog::error( (string) $job['import_id'], 'term ' . $term_id, $e->getMessage() );
				++$job['counts']['errors'];
			}
			if ( time() >= $deadline || Job::memory_tight() ) {
				return;
			}
		}
	}

	/**
	 * @param array<string, mixed> $job
	 */
	private function phase_redirects( array &$job, int $deadline ): void {
		$adapter = ImportRegistry::get( (string) $job['source'] );
		if ( ! $adapter ) {
			$this->advance( $job );
			return;
		}
		$rows = $adapter->redirects( (int) $job['cursor'], self::RED_BATCH );
		if ( empty( $rows ) ) {
			$this->advance( $job );
			return;
		}
		foreach ( $rows as $row ) {
			$job['cursor'] = max( (int) $job['cursor'], (int) ( $row['id'] ?? 0 ) );
			try {
				if ( $this->writer->redirect( (string) ( $row['source'] ?? '' ), (string) ( $row['target'] ?? '' ), (int) ( $row['type'] ?? 301 ) ) ) {
					++$job['counts']['redirects'];
				}
			} catch ( \Throwable $e ) {
				ImportLog::error( (string) $job['import_id'], 'redirect ' . ( $row['source'] ?? '' ), $e->getMessage() );
				++$job['counts']['errors'];
			}
			if ( time() >= $deadline || Job::memory_tight() ) {
				return;
			}
		}
	}

	/**
	 * A CSV or JSON upload: the rows are held in one option for the run and
	 * walked by index, so a 20,000-row spreadsheet is the same resumable job
	 * as a plugin with 20,000 posts.
	 *
	 * @param array<string, mixed> $job
	 */
	private function phase_file( array &$job, int $deadline ): void {
		$file = get_option( self::FILE );
		$rows = is_array( $file ) ? (array) ( $file['rows'] ?? array() ) : array();
		if ( empty( $rows ) ) {
			$this->advance( $job );
			return;
		}
		$match = (string) ( $file['match'] ?? 'id' );
		$map   = (array) ( $file['map'] ?? array() );
		$index = (int) $job['cursor'];
		$total = count( $rows );
		while ( $index < $total ) {
			$row = (array) $rows[ $index ];
			++$index;
			$job['cursor'] = $index;
			++$job['counts']['seen'];
			try {
				$post_id = FileImport::match( $row, $match );
				if ( $post_id <= 0 ) {
					++$job['counts']['unmatched'];
				} else {
					$record = FileImport::record( $row, $map );
					$record = (array) apply_filters( 'airanksuite/import/record', $record, $post_id, 'file' );
					if ( ! empty( $record ) && $this->writer->post( $post_id, $record, (string) $job['mode'] ) ) {
						++$job['counts']['posts'];
					}
				}
			} catch ( \Throwable $e ) {
				ImportLog::error( (string) $job['import_id'], 'row ' . $index, $e->getMessage() );
				++$job['counts']['errors'];
			}
			if ( time() >= $deadline || Job::memory_tight() ) {
				return;
			}
		}
		$this->advance( $job );
	}

	/**
	 * Undo: the journal, backwards. Newest rows first, so a value written
	 * twice in one run comes back to what it was before the run — and each
	 * restored row is deleted as it is applied, which is what makes the undo
	 * itself resumable after a timeout.
	 *
	 * @param array<string, mixed> $job
	 */
	private function phase_undo( array &$job, int $deadline ): void {
		global $wpdb;
		$table = Database::import_undo();
		do {
			// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching
			$rows = $wpdb->get_results( $wpdb->prepare( "SELECT id, kind, object_id, name, prev_value, existed FROM {$table} WHERE import_id = %s ORDER BY id DESC LIMIT %d", (string) $job['import_id'], self::UNDO_BATCH ), ARRAY_A );
			if ( empty( $rows ) ) {
				$job['phase'] = 'finish';
				return;
			}
			$ids = array();
			foreach ( $rows as $row ) {
				try {
					$this->restore( $row );
					$job['counts']['restored'] = (int) ( $job['counts']['restored'] ?? 0 ) + 1;
				} catch ( \Throwable $e ) {
					ImportLog::error( (string) $job['import_id'], 'undo ' . ( $row['name'] ?? '' ), $e->getMessage() );
				}
				$ids[] = (int) $row['id'];
			}
			$in = implode( ',', array_fill( 0, count( $ids ), '%d' ) );
			// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.InterpolatedNotPrepared
			$deleted = $wpdb->query( $wpdb->prepare( "DELETE FROM {$table} WHERE id IN ({$in})", ...$ids ) );
			if ( false === $deleted ) {
				throw new \RuntimeException( 'could not clear the undo journal: ' . esc_html( mb_substr( (string) $wpdb->last_error, 0, 200 ) ) );
			}
		} while ( time() < $deadline && ! Job::memory_tight() );
	}

	/**
	 * Put one journalled value back.
	 *
	 * @param array<string, mixed> $row
	 */
	private function restore( array $row ): void {
		global $wpdb;
		$kind    = (string) $row['kind'];
		$id      = (int) $row['object_id'];
		$name    = (string) $row['name'];
		$existed = ! empty( $row['existed'] );
		$value   = maybe_unserialize( (string) $row['prev_value'] );

		if ( 'post_meta' === $kind ) {
			if ( $existed ) {
				update_post_meta( $id, $name, $value );
			} else {
				delete_post_meta( $id, $name );
			}
			return;
		}
		if ( 'term_meta' === $kind ) {
			if ( $existed ) {
				update_term_meta( $id, $name, $value );
			} else {
				delete_term_meta( $id, $name );
			}
			return;
		}
		if ( 'option' === $kind ) {
			if ( $existed ) {
				update_option( $name, $value, false );
			} else {
				delete_option( $name );
			}
			return;
		}
		if ( 'redirect' === $kind ) {
			$table = Database::redirects();
			if ( ! $existed ) {
				// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching
				$wpdb->delete( $table, array( 'source' => $name ), array( '%s' ) );
				return;
			}
			$previous = json_decode( (string) $row['prev_value'], true );
			if ( is_array( $previous ) ) {
				unset( $previous['id'] );
				// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching
				$wpdb->replace( $table, $previous );
			}
		}
	}

	// ------------------------------------------------------------- reports --

	/**
	 * Which SEO plugins are running alongside this one, and which of them is
	 * still writing meta tags into the page.
	 *
	 * An import does not deactivate anything — that is the customer's call,
	 * and a plugin deactivated by another plugin's importer is how a site ends
	 * up with no meta tags at all. What we can do is say plainly what is
	 * happening, so the duplicate tags in the page source are not a mystery.
	 *
	 * @return array{active: array<int, array<string, string>>, emitting: string, ours: bool}
	 */
	public function conflicts(): array {
		$active = array();
		foreach ( ImportRegistry::all() as $id => $adapter ) {
			if ( $adapter->active() && ! in_array( $id, array( 'woocommerce', 'redirection' ), true ) ) {
				$active[] = array( 'id' => $id, 'label' => $adapter->label() );
			}
		}
		$conflict = \AiRankSuite\Modules\Seo\SeoModule::conflict();
		return array(
			'active'   => $active,
			'emitting' => is_string( $conflict ) ? $conflict : '',
			'ours'     => null === $conflict,
		);
	}
}