<?php
/**
 * Bring your history with you (#21).
 *
 * @package ActivityLog404
 */

namespace ActivityLog404\Import;

use ActivityLog404\ActivityLog;
use ActivityLog404\Database;
use ActivityLog404\EventLogger;
use ActivityLog404\Exclusions;
use ActivityLog404\Network\NetworkLog;
use ActivityLog404\Support\IpAnonymizer;

defined( 'ABSPATH' ) || exit;

/**
 * Imports another activity log's history into this one, in resumable
 * batches.
 *
 * A run is a cursor over the source's ids: `start()` snapshots the source's
 * highest id, and every `batch()` reads the next rows after the cursor, up
 * to that snapshot, and moves the cursor past them. So a run always ends --
 * a source still logging while it runs does not make it chase a moving
 * target -- and it can stop anywhere (a timeout, a closed tab, WP-Cron off)
 * and pick up from one integer.
 *
 * A NEW run always reads the source from its first row, and relies on the
 * row identity below to skip what is already here. It does not start after
 * the previous run's cursor: a source's ids are not stable across its own
 * life -- clearing Simple History's log, or deleting and reinstalling
 * either plugin, starts its AUTO_INCREMENT again, and a cursor carried over
 * would silently skip every new row below it (found by the e2e spec, whose
 * fixture tables are recreated on every run). Re-reading rows that are
 * already here costs a read, not a write.
 *
 * A row's identity is `import_source` + `import_id` + its timestamp, under
 * a unique key (Database::SCHEMA_VERSION 2): the id alone would call a
 * recycled id's new event a duplicate of the old one. A batch reads which
 * of its rows are already here and writes only the rest, so a re-run, a
 * repeated batch or two batches racing each other never write a row twice.
 *
 * What an imported row is held to -- the same rules as a row the live
 * logger writes, applied at import time:
 *
 * - retention: a row older than the retention period is not imported (the
 *   daily purge would delete it the next night anyway), unless retention
 *   is unlimited (ActivityLog::retention_unlimited(), the Free retention setting);
 * - exclusions: a row whose actor is excluded now (Exclusions::excludes())
 *   is not imported -- the owner's choice of whose clicks are noise does
 *   not depend on which plugin recorded them;
 * - IP addresses: truncated unless the site stores full addresses, like
 *   every live row;
 * - secrets: the sources withhold values by key before a row reaches here
 *   (SourceTables::is_withheld_key(), which is the live logger's own
 *   EventLogger::is_secret_setting() test).
 *
 * Imported rows are written directly, NOT through EventLogger::log(): that
 * would forward every historic row to the Pro add-on's Slack/webhook
 * forwarding and fire the free high-risk email alerts for events that
 * happened months ago. What IS logged through it is one `history_imported`
 * row when a run finishes, naming who ran it.
 *
 * Skipped rows are counted by reason and shown, never dropped silently.
 */
final class Importer {

	/** Option holding each source's run state (not autoloaded). */
	public const STATE_OPTION = 'ops_activity_log_import_state';

	/** Rows read per batch: bounds memory whatever the source's size. */
	public const BATCH_SIZE = 500;

	/** The event type of the row written when a run finishes. */
	public const FINISHED_EVENT = 'history_imported';

	/** The event type of a source row with no equivalent here. */
	public const GENERIC_EVENT = 'imported_event';

	public const IDLE      = 'idle';
	public const RUNNING   = 'running';
	public const DONE      = 'done';
	public const CANCELLED = 'cancelled';

	/** Reasons a row is skipped, each counted. */
	public const SKIP_RETENTION  = 'retention';
	public const SKIP_EXCLUDED   = 'excluded';
	public const SKIP_DUPLICATE  = 'already_imported';
	public const SKIP_UNREADABLE = 'unreadable';
	public const SKIP_FAILED     = 'write_failed';

	/**
	 * @return array<string, string> Event type => label for the log's filter.
	 */
	public static function event_types(): array {
		return array(
			self::FINISHED_EVENT => 'History Imported',
			self::GENERIC_EVENT  => 'Imported Event',
		);
	}

	/**
	 * Every source this importer knows, by key.
	 *
	 * @return array<string, Source>
	 */
	public static function sources(): array {
		$sources = array( new SimpleHistory(), new Wsal(), new Stream() );

		// Stream's network-admin rows (#8), on the main site only -- the
		// network log's rows follow the main site's settings -- and only
		// while the network log exists to hold them.
		if ( NetworkLog::is_active() && is_main_site() ) {
			$sources[] = new Stream( true );
		}

		$keyed = array();
		foreach ( $sources as $source ) {
			$keyed[ $source->key() ] = $source;
		}

		return $keyed;
	}

	/**
	 * @param string $key A source key.
	 */
	public static function source( string $key ): ?Source {
		return self::sources()[ $key ] ?? null;
	}

	/**
	 * Whether a source key imports into the network log (#8) rather than
	 * this site's.
	 *
	 * @param string $key Source key.
	 */
	public static function targets_network( string $key ): bool {
		return Stream::NETWORK_KEY === $key;
	}

	/**
	 * The capability that starts, steps or stops a source: a site's history
	 * is its administrators'; the network's is its super admins'.
	 *
	 * @param string $key Source key.
	 */
	public static function capability( string $key ): string {
		return self::targets_network( $key ) ? 'manage_network' : 'manage_options';
	}

	/**
	 * The table a source's rows are written to.
	 *
	 * @param string $key Source key.
	 */
	private static function table_for( string $key ): string {
		return self::targets_network( $key ) ? NetworkLog::table() : Database::get_table_name();
	}

	/**
	 * One source's run state.
	 *
	 * @param string $key Source key.
	 * @return array<string, mixed>
	 */
	public static function state( string $key ): array {
		$all   = get_option( self::STATE_OPTION, array() );
		$state = is_array( $all ) && isset( $all[ $key ] ) && is_array( $all[ $key ] ) ? $all[ $key ] : array();

		return array_merge( self::empty_state(), $state );
	}

	/**
	 * @return array<string, mixed>
	 */
	private static function empty_state(): array {
		return array(
			'status'      => self::IDLE,
			'cursor'      => 0,
			'max_id'      => 0,
			'total'       => 0,
			'processed'   => 0,
			'imported'    => 0,
			'generic'     => 0,
			'skipped'     => array(),
			'started_at'  => 0,
			'finished_at' => 0,
			'started_by'  => 0,
		);
	}

	/**
	 * @param string               $key   Source key.
	 * @param array<string, mixed> $state Its new state.
	 */
	private static function save_state( string $key, array $state ): void {
		$all = get_option( self::STATE_OPTION, array() );
		$all = is_array( $all ) ? $all : array();

		$all[ $key ] = $state;
		update_option( self::STATE_OPTION, $all, false );
	}

	/**
	 * Begin (or begin again) importing a source. A source already running is
	 * left alone. A new run reads the whole source (see the class docblock
	 * for why never from an old cursor); rows already here are counted, not
	 * rewritten.
	 *
	 * @param Source $source  The source.
	 * @param int    $user_id Who started it.
	 * @return string Result code: started, running, unsupported or empty.
	 */
	public static function start( Source $source, int $user_id ): string {
		if ( ! $source->detect()->is_supported() ) {
			return 'unsupported';
		}

		$state = self::state( $source->key() );
		if ( self::RUNNING === $state['status'] ) {
			return 'running';
		}

		$preview = $source->preview();
		$cursor  = 0;
		$total   = $preview['count'];

		if ( 0 === $total ) {
			return 'empty';
		}

		self::save_state(
			$source->key(),
			array_merge(
				self::empty_state(),
				array(
					'status'     => self::RUNNING,
					'cursor'     => $cursor,
					'max_id'     => $preview['max_id'],
					'total'      => $total,
					'started_at' => time(),
					'started_by' => $user_id,
				)
			)
		);

		return 'started';
	}

	/**
	 * Stop a run where it is. What was imported stays; starting again reads
	 * the source again and skips it.
	 *
	 * @param string $key Source key.
	 */
	public static function cancel( string $key ): bool {
		$state = self::state( $key );
		if ( self::RUNNING !== $state['status'] ) {
			return false;
		}

		$state['status']      = self::CANCELLED;
		$state['finished_at'] = time();
		self::save_state( $key, $state );

		return true;
	}

	/**
	 * Run batches for up to `$seconds`, or until the run is done.
	 *
	 * Serialised by a MySQL named lock, so the browser's polling and a
	 * WP-Cron event never run two batches of one source at once. A caller
	 * that cannot take the lock returns the state as it stands; the other
	 * one is making progress. The lock is the connection's, so a request
	 * that dies mid-batch releases it by dying.
	 *
	 * @param string $key     Source key.
	 * @param float  $seconds Time budget.
	 * @return array<string, mixed> The state after.
	 */
	public static function run( string $key, float $seconds ): array {
		$source = self::source( $key );
		if ( null === $source || self::RUNNING !== self::state( $key )['status'] ) {
			return self::state( $key );
		}

		if ( ! self::lock( $key ) ) {
			return self::state( $key );
		}

		$deadline = microtime( true ) + max( 0.0, $seconds );
		try {
			do {
				$state = self::batch( $source );
			} while ( self::RUNNING === $state['status'] && microtime( true ) < $deadline );
		} finally {
			self::unlock( $key );
		}

		return $state;
	}

	/**
	 * One batch: read, map, filter, write, advance.
	 *
	 * @param Source $source Source.
	 * @return array<string, mixed> The state after.
	 */
	public static function batch( Source $source ): array {
		$key   = $source->key();
		$state = self::state( $key );
		if ( self::RUNNING !== $state['status'] ) {
			return $state;
		}

		$rows = $source->rows( (int) $state['cursor'], (int) $state['max_id'], self::BATCH_SIZE );

		if ( array() === $rows ) {
			return self::finish( $source, $state );
		}

		$already = self::already_imported( $key, $rows );
		$cutoff  = self::retention_cutoff();
		$insert  = array();

		foreach ( $rows as $row ) {
			$reason = self::skip_reason( $row, $already, $cutoff );
			if ( null !== $reason ) {
				$state['skipped'][ $reason ] = (int) ( $state['skipped'][ $reason ] ?? 0 ) + 1;
				continue;
			}
			$insert[] = $row;
		}

		$written = self::insert( $key, $insert );
		$refused = array_diff_key( $insert, $written );
		$landed  = array() !== $refused ? self::already_imported( $key, $refused ) : array();

		foreach ( $insert as $i => $row ) {
			if ( ! isset( $written[ $i ] ) ) {
				// Refused. Either another writer of the same row won a race and
				// the unique key turned this one away -- it is here -- or the
				// write failed, which is counted as that, not as a duplicate.
				$reason                      = isset( $landed[ self::identity( $row['source_id'], $row['timestamp'] ) ] ) ? self::SKIP_DUPLICATE : self::SKIP_FAILED;
				$state['skipped'][ $reason ] = (int) ( $state['skipped'][ $reason ] ?? 0 ) + 1;
				continue;
			}
			++$state['imported'];
			if ( self::GENERIC_EVENT === $row['event_type'] ) {
				++$state['generic'];
			}
		}

		$state['processed'] = (int) $state['processed'] + count( $rows );
		$state['cursor']    = max( array_merge( array( (int) $state['cursor'] ), array_map( 'intval', array_column( $rows, 'source_id' ) ) ) );

		// Stopped from another request while this batch ran: keep the stop,
		// and the progress this batch made.
		wp_cache_delete( self::STATE_OPTION, 'options' );
		if ( self::CANCELLED === self::state( $key )['status'] ) {
			$state['status']      = self::CANCELLED;
			$state['finished_at'] = time();
			self::save_state( $key, $state );
			return $state;
		}

		self::save_state( $key, $state );

		if ( $state['cursor'] >= (int) $state['max_id'] ) {
			return self::finish( $source, $state );
		}

		return $state;
	}

	/**
	 * Why a row is not imported, or null to import it.
	 *
	 * @param array<string, mixed> $row     A mapped source row.
	 * @param array<string, true>  $already Identities already imported (self::identity()).
	 * @param string|null          $cutoff  Oldest timestamp kept, or null for unlimited.
	 */
	private static function skip_reason( array $row, array $already, ?string $cutoff ): ?string {
		if ( isset( $already[ self::identity( $row['source_id'], $row['timestamp'] ) ] ) ) {
			return self::SKIP_DUPLICATE;
		}
		// A row with no readable date cannot be placed in the log, or held
		// to retention; it is counted rather than written with a made-up one.
		if ( 1 !== preg_match( '/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$/D', (string) $row['timestamp'] ) || str_starts_with( (string) $row['timestamp'], '0000' ) ) {
			return self::SKIP_UNREADABLE;
		}
		if ( null !== $cutoff && (string) $row['timestamp'] < $cutoff ) {
			return self::SKIP_RETENTION;
		}
		if ( null !== $row['user_id'] && Exclusions::excludes( (int) $row['user_id'] ) ) {
			return self::SKIP_EXCLUDED;
		}

		return null;
	}

	/**
	 * The oldest timestamp the site keeps (UTC), or null when retention is
	 * unlimited.
	 */
	public static function retention_cutoff(): ?string {
		if ( ActivityLog::retention_unlimited() ) {
			return null;
		}

		$days = ActivityLog::retention_days_or_default( get_option( 'ops_activity_log_retention_days', ActivityLog::DEFAULT_RETENTION_DAYS ) );

		return gmdate( 'Y-m-d H:i:s', time() - ( $days * DAY_IN_SECONDS ) );
	}

	/**
	 * Which of these rows this source has already imported, by identity.
	 *
	 * @param string                           $key  Source key.
	 * @param array<int, array<string, mixed>> $rows Mapped rows.
	 * @return array<string, true>
	 */
	private static function already_imported( string $key, array $rows ): array {
		global $wpdb;

		$ids = array_values( array_unique( array_map( 'intval', array_column( $rows, 'source_id' ) ) ) );
		if ( array() === $ids ) {
			return array();
		}

		$table = self::table_for( $key );
		$scope = self::targets_network( $key ) ? ' AND blog_id = 0 AND network_id = ' . (int) get_current_network_id() : '';
		$query = 'SELECT import_id, timestamp FROM ' . $table . ' WHERE import_source = %s' . $scope . ' AND import_id IN (' . implode( ',', array_fill( 0, count( $ids ), '%d' ) ) . ')';

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery,WordPress.DB.PreparedSQL.NotPrepared -- placeholders built above, one per id.
		$found = $wpdb->get_results( $wpdb->prepare( $query, array_merge( array( $key ), $ids ) ), ARRAY_A );

		$map = array();
		foreach ( is_array( $found ) ? $found : array() as $row ) {
			$map[ self::identity( $row['import_id'], $row['timestamp'] ) ] = true;
		}

		return $map;
	}

	/**
	 * A source row's identity: its id AND its time, so a recycled id is a
	 * new row (see the class docblock).
	 *
	 * @param mixed $source_id Source id.
	 * @param mixed $timestamp UTC datetime.
	 */
	private static function identity( $source_id, $timestamp ): string {
		return (int) $source_id . '|' . (string) $timestamp;
	}

	/**
	 * Write rows, one statement per row so a unique-key refusal costs one
	 * row, not the batch. Returns which of the given indexes were written.
	 *
	 * @param string                           $key  Source key.
	 * @param array<int, array<string, mixed>> $rows Mapped rows.
	 * @return array<int, true>
	 */
	private static function insert( string $key, array $rows ): array {
		global $wpdb;

		$written    = array();
		$network    = self::targets_network( $key );
		$table      = self::table_for( $key );
		$suppressed = $wpdb->suppress_errors( true );

		// A network row (#8) names its network and no site (blog_id 0);
		// the formats below lead with the two columns only then.
		$scope   = $network ? array(
			'network_id' => (int) get_current_network_id(),
			'blog_id'    => 0,
		) : array();
		$leading = $network ? array( '%d', '%d' ) : array();

		foreach ( $rows as $i => $row ) {
			$details = is_array( $row['details'] ) ? $row['details'] : array();

			$ok = $wpdb->insert(
				$table,
				$scope + array(
					'timestamp'     => (string) $row['timestamp'],
					'user_id'       => $row['user_id'],
					'user_email'    => $row['user_email'],
					'ip_address'    => self::stored_ip( $row['ip'] ),
					'event_type'    => (string) $row['event_type'],
					'object_type'   => $row['object_type'],
					'object_id'     => $row['object_id'],
					'summary'       => (string) $row['summary'],
					'details'       => wp_json_encode( $details ),
					'import_source' => $key,
					'import_id'     => (int) $row['source_id'],
				),
				array_merge( $leading, array( '%s', '%d', '%s', '%s', '%s', '%s', '%d', '%s', '%s', '%s', '%d' ) )
			);

			if ( false !== $ok ) {
				$written[ $i ] = true;
				// A site row's copy in the network log (#8), as the live
				// logger makes one.
				if ( ! $network ) {
					NetworkLog::mirror_site_row( (int) $wpdb->insert_id );
				}
			}
		}

		$wpdb->suppress_errors( $suppressed );

		return $written;
	}

	/**
	 * An address as this site stores addresses: in full only when it opted
	 * in, truncated otherwise -- the live logger's rule (EventLogger).
	 *
	 * @param mixed $ip A valid address or null.
	 */
	public static function stored_ip( $ip ): ?string {
		if ( ! is_string( $ip ) || '' === $ip ) {
			return null;
		}
		if ( (bool) get_option( EventLogger::OPTION_STORE_FULL_IPS, false ) ) {
			return $ip;
		}

		$truncated = IpAnonymizer::truncate( $ip );

		return '' !== $truncated ? $truncated : null;
	}

	/**
	 * Mark a run done and record it in the log.
	 *
	 * @param Source               $source Source.
	 * @param array<string, mixed> $state  Its state.
	 * @return array<string, mixed>
	 */
	private static function finish( Source $source, array $state ): array {
		$state['status']      = self::DONE;
		$state['finished_at'] = time();
		self::save_state( $source->key(), $state );

		$skipped = array_sum( array_map( 'intval', (array) $state['skipped'] ) );
		$summary = sprintf( 'History imported from %1$s: %2$d events imported, %3$d skipped', $source->label(), (int) $state['imported'], $skipped );
		$details = array(
			'source'   => $source->key(),
			'imported' => (int) $state['imported'],
			'generic'  => (int) $state['generic'],
			'skipped'  => (array) $state['skipped'],
		);
		$actor   = (int) $state['started_by'] > 0 ? (int) $state['started_by'] : null;

		// The record of an import lands where its rows did.
		if ( self::targets_network( $source->key() ) ) {
			NetworkLog::log( self::FINISHED_EVENT, $summary, 'import', null, $details, 0, $actor );
		} else {
			EventLogger::log( self::FINISHED_EVENT, $summary, 'import', null, $details, $actor );
		}

		return $state;
	}

	/**
	 * @param string $key Source key.
	 */
	private static function lock( string $key ): bool {
		global $wpdb;

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- a named lock has no WordPress API.
		return '1' === (string) $wpdb->get_var( $wpdb->prepare( 'SELECT GET_LOCK( %s, 0 )', self::lock_name( $key ) ) );
	}

	/**
	 * @param string $key Source key.
	 */
	private static function unlock( string $key ): void {
		global $wpdb;

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- a named lock has no WordPress API.
		$wpdb->get_var( $wpdb->prepare( 'SELECT RELEASE_LOCK( %s )', self::lock_name( $key ) ) );
	}

	/**
	 * Named per table (so per site) and per source; MySQL caps the name at 64.
	 *
	 * @param string $key Source key.
	 */
	private static function lock_name( string $key ): string {
		return substr( 'ops404al_import_' . md5( self::table_for( $key ) . '|' . $key ), 0, 64 );
	}
}