<?php
/**
 * Data Access Object for the DataStream ingestion table.
 *
 * @package DataStream
 */

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

// Custom table DAO. WP_Query cannot address wp_ds_ingestion_table, so every
// method talks to $wpdb. Table names cannot be %s placeholders.
// phpcs:disable WordPress.DB.DirectDatabaseQuery.DirectQuery
// phpcs:disable WordPress.DB.DirectDatabaseQuery.NoCaching
// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared

/**
 * Class DataStream_DAO
 */
class DataStream_DAO {

	/**
	 * Fully qualified table name.
	 *
	 * @var string
	 */
	private $table_name;

	/**
	 * Constructor.
	 */
	public function __construct() {
		global $wpdb;
		$this->table_name = $wpdb->prefix . DATASTREAM_TABLE;
	}

	/**
	 * Whether the ingestion table exists.
	 *
	 * @return bool
	 */
	public function table_exists() {
		global $wpdb;

		$like = $wpdb->esc_like( $this->table_name );

		return $this->table_name === $wpdb->get_var(
			$wpdb->prepare( 'SHOW TABLES LIKE %s', $like )
		);
	}

	/**
	 * Upsert a single ingested record.
	 *
	 * @param string $external_id Sanitized external identifier.
	 * @param string $payload     JSON-encoded payload.
	 * @param string $stream_type Stream classification.
	 * @param string $source      Origin identifier.
	 * @return bool True on success.
	 */
	public function upsert( $external_id, $payload, $stream_type = 'generic', $source = '' ) {
		global $wpdb;

		$external_id = sanitize_text_field( $external_id );
		$stream_type = sanitize_key( $stream_type );
		$source      = sanitize_text_field( $source );
		$max_id_len  = absint( DataStream_Config::get( 'max_external_id_len' ) );
		$max_payload = absint( DataStream_Config::get( 'max_payload_bytes' ) );

		if ( '' === $external_id || '' === $payload ) {
			return false;
		}

		if ( $max_id_len > 0 && strlen( $external_id ) > $max_id_len ) {
			$external_id = substr( $external_id, 0, $max_id_len );
		}

		if ( $max_payload > 0 && strlen( $payload ) > $max_payload ) {
			DataStream_Logger::error(
				'Payload exceeds maximum size and was rejected.',
				array(
					'external_id' => $external_id,
					'size'        => strlen( $payload ),
					'max'         => $max_payload,
				)
			);
			return false;
		}

		if ( ! $this->table_exists() ) {
			DataStream_Schema::create_schema();
		}

		$existing_id = $wpdb->get_var(
			$wpdb->prepare(
				"SELECT id FROM {$this->table_name} WHERE external_id = %s",
				$external_id
			)
		);

		$row = array(
			'external_id'  => $external_id,
			'stream_type'  => $stream_type,
			'source'       => $source,
			'data_payload' => $payload,
			'updated_at'   => current_time( 'mysql' ),
		);

		$formats = array( '%s', '%s', '%s', '%s', '%s' );

		if ( $existing_id ) {
			$result = $wpdb->update(
				$this->table_name,
				$row,
				array( 'external_id' => $external_id ),
				$formats,
				array( '%s' )
			);

			return false !== $result;
		}

		$result = $wpdb->insert( $this->table_name, $row, $formats );

		return false !== $result;
	}

	/**
	 * Persist multiple records inside a transaction when supported.
	 *
	 * @param array<int, array{external_id: string, payload: string, stream_type?: string, source?: string}> $records Records.
	 * @return array{success: bool, processed: int, failed: int, skipped: int}
	 */
	public function batch_upsert( $records ) {
		global $wpdb;

		$processed = 0;
		$skipped   = 0;
		$records   = $this->dedupe_records( $records );

		if ( empty( $records ) ) {
			return array(
				'success'   => true,
				'processed' => 0,
				'failed'    => 0,
				'skipped'   => 0,
			);
		}

		$use_transaction = $this->supports_transactions();
		$started           = false;

		if ( $use_transaction ) {
			// wpdb::query() returns 0 (num_rows) for START TRANSACTION, so a
			// bool cast reads success as failure — the transaction would open
			// but never COMMIT, and InnoDB would silently discard every row
			// at disconnect. Only an explicit false means it failed to start.
			$started = ( false !== $wpdb->query( 'START TRANSACTION' ) );
		}

		/**
		 * When true, one failed record rolls back the whole batch (pre-1.4
		 * behavior). Default is best-effort: bad records are skipped and
		 * counted, good records persist — fickle APIs routinely mix one
		 * malformed record into an otherwise fine page.
		 *
		 * @param bool $all_or_nothing Roll back everything on any failure.
		 */
		$all_or_nothing = (bool) apply_filters( 'datastream_batch_all_or_nothing', false );
		$failed         = 0;

		foreach ( $records as $record ) {
			if ( empty( $record['external_id'] ) || empty( $record['payload'] ) ) {
				++$skipped;
				continue;
			}

			$result = $this->upsert(
				$record['external_id'],
				$record['payload'],
				$record['stream_type'] ?? 'generic',
				$record['source'] ?? ''
			);

			if ( ! $result ) {
				++$failed;

				DataStream_Logger::error(
					'Record upsert failed.',
					array(
						'external_id' => $record['external_id'] ?? '',
						'processed'   => $processed,
					)
				);

				if ( $all_or_nothing ) {
					if ( $started ) {
						$wpdb->query( 'ROLLBACK' );
					}

					return array(
						'success'   => false,
						'processed' => $processed,
						'failed'    => $failed,
						'skipped'   => $skipped,
					);
				}

				continue;
			}

			++$processed;
		}

		if ( $started ) {
			$wpdb->query( 'COMMIT' );
		}

		return array(
			'success'   => $processed > 0 || 0 === $failed,
			'processed' => $processed,
			'failed'    => $failed,
			'skipped'   => $skipped,
		);
	}

	/**
	 * Deduplicate records by external_id (last wins).
	 *
	 * @param array<int, array<string, mixed>> $records Input records.
	 * @return array<int, array<string, mixed>>
	 */
	private function dedupe_records( $records ) {
		$map = array();

		foreach ( $records as $record ) {
			if ( empty( $record['external_id'] ) ) {
				continue;
			}

			$map[ $record['external_id'] ] = $record;
		}

		return array_values( $map );
	}

	/**
	 * Detect InnoDB transaction support.
	 *
	 * @return bool
	 */
	private function supports_transactions() {
		global $wpdb;

		if ( ! $this->table_exists() ) {
			return false;
		}

		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- table name is trusted.
		$engine = $wpdb->get_var(
			$wpdb->prepare(
				'SELECT ENGINE FROM information_schema.TABLES WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s',
				DB_NAME,
				$this->table_name
			)
		);

		return 'InnoDB' === $engine;
	}

	/**
	 * Retrieve payload for a single external ID.
	 *
	 * @param string $external_id External identifier.
	 * @return string|null
	 */
	public function get_payload_by_external_id( $external_id ) {
		global $wpdb;

		if ( ! $this->table_exists() ) {
			return null;
		}

		$result = $wpdb->get_var(
			$wpdb->prepare(
				"SELECT data_payload FROM {$this->table_name} WHERE external_id = %s",
				sanitize_text_field( $external_id )
			)
		);

		return $result ? $result : null;
	}

	/**
	 * Retrieve recent records.
	 *
	 * @param int    $limit       Maximum rows.
	 * @param string $stream_type Optional filter.
	 * @return array<int, array<string, mixed>>
	 */
	public function get_recent_records( $limit = 50, $stream_type = '', $offset = 0 ) {
		global $wpdb;

		if ( ! $this->table_exists() ) {
			return array();
		}

		$limit  = max( 1, min( 100, absint( $limit ) ) );
		$offset = absint( $offset );

		if ( '' !== $stream_type ) {
			// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- table name is trusted.
			return $wpdb->get_results(
				$wpdb->prepare(
					"SELECT id, external_id, stream_type, source, data_payload, updated_at FROM {$this->table_name} WHERE stream_type = %s ORDER BY updated_at DESC LIMIT %d OFFSET %d",
					sanitize_key( $stream_type ),
					$limit,
					$offset
				),
				ARRAY_A
			) ?: array();
		}

		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- table name is trusted.
		return $wpdb->get_results(
			$wpdb->prepare(
				"SELECT id, external_id, stream_type, source, data_payload, updated_at FROM {$this->table_name} ORDER BY updated_at DESC LIMIT %d OFFSET %d",
				$limit,
				$offset
			),
			ARRAY_A
		) ?: array();
	}

	/**
	 * Count records, optionally for one stream.
	 *
	 * @param string $stream_type Stream filter ('' = all).
	 * @return int
	 */
	public function count_records( $stream_type = '' ) {
		global $wpdb;

		if ( ! $this->table_exists() ) {
			return 0;
		}

		if ( '' === $stream_type ) {
			return $this->get_record_count();
		}

		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- table name is trusted.
		return (int) $wpdb->get_var(
			$wpdb->prepare(
				"SELECT COUNT(*) FROM {$this->table_name} WHERE stream_type = %s",
				sanitize_key( $stream_type )
			)
		);
	}

	/**
	 * Count all ingested records.
	 *
	 * @return int
	 */
	public function get_record_count() {
		global $wpdb;

		if ( ! $this->table_exists() ) {
			return 0;
		}

		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- table name is trusted.
		return (int) $wpdb->get_var( "SELECT COUNT(*) FROM {$this->table_name}" );
	}

	/**
	 * Count records grouped by stream type.
	 *
	 * @return array<string, int>
	 */
	public function get_stream_counts() {
		global $wpdb;

		if ( ! $this->table_exists() ) {
			return array();
		}

		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- table name is trusted.
		$rows = $wpdb->get_results(
			"SELECT stream_type, COUNT(*) AS total FROM {$this->table_name} GROUP BY stream_type",
			ARRAY_A
		);

		$counts = array();

		foreach ( $rows as $row ) {
			$counts[ $row['stream_type'] ] = (int) $row['total'];
		}

		return $counts;
	}

	/**
	 * Delete records older than N days.
	 *
	 * @param int $days Age threshold.
	 * @return int Rows deleted.
	 */
	public function purge_stale_records( $days ) {
		global $wpdb;

		$days = absint( $days );

		if ( $days < 1 || ! $this->table_exists() ) {
			return 0;
		}

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

		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- table name is trusted.
		return (int) $wpdb->query(
			$wpdb->prepare(
				"DELETE FROM {$this->table_name} WHERE updated_at < %s",
				$threshold
			)
		);
	}

	/**
	 * Delete all records for a stream type.
	 *
	 * @param string $stream_type Stream slug.
	 * @return int
	 */
	public function purge_by_stream_type( $stream_type ) {
		global $wpdb;

		if ( ! $this->table_exists() ) {
			return 0;
		}

		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- table name is trusted.
		return (int) $wpdb->query(
			$wpdb->prepare(
				"DELETE FROM {$this->table_name} WHERE stream_type = %s",
				sanitize_key( $stream_type )
			)
		);
	}

	/**
	 * Delete all ingested records.
	 *
	 * @return int
	 */
	public function purge_all() {
		global $wpdb;

		if ( ! $this->table_exists() ) {
			return 0;
		}

		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- table name is trusted.
		return (int) $wpdb->query( "TRUNCATE TABLE {$this->table_name}" );
	}
}
