<?php
/**
 * ClickHouse storage: an append-only event log written over the HTTP(S) interface (no PHP extension needed),
 * and views that rebuild the visits, hits, goals and visitors tables the reports read.
 * Works with ClickHouse Cloud, Yandex Managed Service for ClickHouse, Altinity, DoubleCloud and self-hosted servers.
 *
 * @package Visitrix
 */

defined( 'ABSPATH' ) || exit;

// Copying reads the plugin's own tables of the WordPress database: direct queries are expected here.
// phpcs:disable WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching

/**
 * ClickHouse connection with the part of the wpdb interface the reports use.
 * Queries are prepared with $wpdb->prepare() and the few MySQL-only expressions are rewritten.
 */
class Visitrix_ClickHouse_DB {

	/**
	 * Table prefix.
	 *
	 * @var string
	 */
	public $prefix = 'wp_';

	/**
	 * Not used by ClickHouse (kept for the wpdb interface).
	 *
	 * @var int
	 */
	public $insert_id = 0;

	/**
	 * Not reported by ClickHouse (kept for the wpdb interface).
	 *
	 * @var int
	 */
	public $rows_affected = 0;

	/**
	 * Last error.
	 *
	 * @var string
	 */
	public $last_error = '';

	/**
	 * Settings.
	 *
	 * @var array
	 */
	private $s;

	/**
	 * Connection check from the settings page: ignores the back-off after a failure.
	 *
	 * @var bool
	 */
	private $probe = false;

	/**
	 * Constructor.
	 *
	 * @param array $s     Settings with the plain password.
	 * @param bool  $probe Connection check.
	 */
	public function __construct( $s, $probe = false ) {
		$this->s      = $s;
		$this->prefix = (string) $s['prefix'];
		$this->probe  = (bool) $probe;
	}

	/**
	 * Database name.
	 *
	 * @return string
	 */
	public function database() {
		return '' !== (string) $this->s['name'] ? (string) $this->s['name'] : 'default';
	}

	/**
	 * Sends one statement.
	 *
	 * @param string $sql    Statement.
	 * @param string $body   Data for INSERT … FORMAT JSONEachRow (appended after the statement).
	 * @param int    $timeout Seconds.
	 * @return string|false Response body.
	 */
	public function request( $sql, $body = '', $timeout = 0 ) {
		if ( ! $this->probe && get_transient( 'visitrix_storage_down' ) ) {
			$this->last_error = 'connection paused after an error';
			return false;
		}
		$params = array(
			'database'                                => $this->database(),
			// MySQL-like results: NULL for aggregates over no rows, plain numbers, NULLs from outer joins.
			'aggregate_functions_null_for_empty'      => 1,
			'output_format_json_quote_64bit_integers' => 0,
			'join_use_nulls'                          => 1,
			'date_time_input_format'                  => 'best_effort',
			// "SUM(pageviews) AS pageviews": a column name means the column, as in MySQL, not the alias.
			'prefer_column_name_to_alias'             => 1,
		);
		if ( '' !== $body ) {
			$params['async_insert'] = 1;
			// Beacons do not wait for the server to flush its insert buffer; copying data does.
			$params['wait_for_async_insert'] = isset( $_GET['visitrix_collect'] ) ? 0 : 1; // phpcs:ignore WordPress.Security.NonceVerification.Recommended -- only picks the insert mode.
		}
		$url  = add_query_arg( $params, untrailingslashit( (string) $this->s['url'] ) . '/' );
		$args = array(
			// Beacons must stay fast; reports and settings may wait for a slow TLS handshake.
			'timeout'   => $timeout ? $timeout : ( isset( $_GET['visitrix_collect'] ) ? 5 : 30 ), // phpcs:ignore WordPress.Security.NonceVerification.Recommended -- only picks the timeout.
			'headers'   => array(
				'X-ClickHouse-User' => (string) $this->s['user'],
				'X-ClickHouse-Key'  => (string) $this->s['pass'],
				'Content-Type'      => 'text/plain; charset=utf-8',
			),
			'body'      => '' !== $body ? $sql . "\n" . $body : $sql,
			'sslverify' => true,
		);
		// A dropped connection (a cloud service waking up, a proxy resetting TLS) is retried: once for beacons,
		// twice for reports and settings.
		$tries = isset( $_GET['visitrix_collect'] ) ? 2 : 3; // phpcs:ignore WordPress.Security.NonceVerification.Recommended -- only picks the number of attempts.
		for ( $i = 1; $i <= $tries; $i++ ) {
			$response = self::post( $url, $args );
			if ( ! is_wp_error( $response ) ) {
				break;
			}
			if ( $i < $tries ) {
				usleep( 300000 );
			}
		}
		if ( is_wp_error( $response ) ) {
			$this->last_error = $response->get_error_message();
			$this->down();
			return false;
		}
		$code = (int) $response['code'];
		$text = (string) $response['body'];
		if ( $code < 200 || $code >= 300 ) {
			$this->last_error = trim( substr( $text, 0, 500 ) );
			// Authentication failures and an unavailable server pause new attempts. ClickHouse answers 500 for
			// query errors too, so those are told apart by the response.
			$auth = 401 === $code || 403 === $code || false !== strpos( $text, 'Code: 516' ) || false !== stripos( $text, 'AUTHENTICATION_FAILED' );
			$gone = in_array( $code, array( 502, 503, 504 ), true ) || ( $code >= 500 && false === strpos( $text, 'Code: ' ) );
			if ( $auth || $gone ) {
				$this->down();
			}
			return false;
		}
		$this->last_error = '';
		return $text;
	}

	/**
	 * POST over one kept-alive connection per PHP request: a report runs a couple of dozen queries, and a new
	 * TLS handshake for each of them is slow and, through some proxies, unreliable. Without the cURL extension
	 * the WordPress HTTP API is used.
	 *
	 * @param string $url  URL.
	 * @param array  $args wp_remote_post() arguments.
	 * @return array|WP_Error {code, body}
	 */
	private static function post( $url, $args ) {
		static $handle = null;
		if ( ! function_exists( 'curl_init' ) ) {
			$response = wp_remote_post( $url, $args );
			return is_wp_error( $response ) ? $response : array(
				'code' => (int) wp_remote_retrieve_response_code( $response ),
				'body' => (string) wp_remote_retrieve_body( $response ),
			);
		}
		// phpcs:disable WordPress.WP.AlternativeFunctions -- connection reuse is not available through the HTTP API.
		if ( null === $handle ) {
			$handle = curl_init();
		} else {
			curl_reset( $handle ); // Keeps the open connection.
		}
		$headers = array();
		foreach ( $args['headers'] as $name => $value ) {
			$headers[] = $name . ': ' . $value;
		}
		$ca = ABSPATH . WPINC . '/certificates/ca-bundle.crt';
		curl_setopt_array(
			$handle,
			array(
				CURLOPT_URL            => $url,
				CURLOPT_POST           => true,
				CURLOPT_POSTFIELDS     => $args['body'],
				CURLOPT_HTTPHEADER     => $headers,
				CURLOPT_RETURNTRANSFER => true,
				CURLOPT_TIMEOUT        => (int) $args['timeout'],
				CURLOPT_CONNECTTIMEOUT => min( 10, (int) $args['timeout'] ),
				CURLOPT_SSL_VERIFYPEER => true,
				CURLOPT_SSL_VERIFYHOST => 2,
				CURLOPT_USERAGENT      => 'Visitrix/' . VISITRIX_VERSION . '; ' . home_url( '/' ),
				CURLOPT_ENCODING       => '',
			) + ( is_readable( $ca ) ? array( CURLOPT_CAINFO => $ca ) : array() )
		);
		$body = curl_exec( $handle );
		if ( false === $body ) {
			$error = curl_error( $handle );
			curl_close( $handle );
			$handle = null; // A broken connection is not reused.
			return new WP_Error( 'visitrix_clickhouse', 'cURL: ' . $error );
		}
		$code = (int) curl_getinfo( $handle, CURLINFO_RESPONSE_CODE );
		// phpcs:enable
		return array(
			'code' => $code,
			'body' => (string) $body,
		);
	}

	/**
	 * Records a failed connection and pauses new attempts for a minute.
	 */
	private function down() {
		if ( ! $this->probe ) {
			set_transient( 'visitrix_storage_down', 1, MINUTE_IN_SECONDS );
			Visitrix_Storage::fail( $this->last_error );
		}
	}

	/**
	 * Prepares a query exactly like $wpdb->prepare().
	 *
	 * @param string $query Query.
	 * @param mixed  ...$args Values.
	 * @return string
	 */
	public function prepare( $query, ...$args ) {
		global $wpdb;
		return $wpdb->prepare( $query, ...$args ); // phpcs:ignore WordPress.DB.PreparedSQL.NotPrepared -- passes the caller's query and values through.
	}

	/**
	 * Rewrites the MySQL-only expressions used by the reports.
	 *
	 * @param string $sql MySQL query.
	 * @return string ClickHouse query.
	 */
	public static function translate( $sql ) {
		$sql = (string) preg_replace( '/\bLEFT\(\s*([a-z_.]+)\s*,/i', 'left(toString($1),', $sql );
		$sql = (string) preg_replace( '/\bCAST\(\s*(.+?)\s+AS\s+SIGNED\s*\)/i', 'toInt64($1)', $sql );
		$sql = (string) preg_replace( '/\bCAST\(\s*(.+?)\s+AS\s+UNSIGNED\s*\)/i', 'toUInt64($1)', $sql );
		// CONCAT_WS(sep, a, NULLIF(b, ''), …) skips empty parts.
		$sql = (string) preg_replace_callback(
			"/CONCAT_WS\\(\\s*('[^']*')\\s*,\\s*([^()]+?)\\s*,\\s*NULLIF\\(\\s*([a-z_.]+)\\s*,\\s*''\\s*\\)\\s*,\\s*NULLIF\\(\\s*([a-z_.]+)\\s*,\\s*''\\s*\\)\\s*\\)/i",
			static function ( $m ) {
				return "arrayStringConcat(arrayFilter(x -> x != '', [toString({$m[2]}), toString({$m[3]}), toString({$m[4]})]), {$m[1]})";
			},
			$sql
		);
		return $sql;
	}

	/**
	 * Runs a statement.
	 *
	 * @param string $sql  Statement.
	 * @param bool   $rows Return the result rows.
	 * @return int|bool|array
	 */
	public function query( $sql, $rows = false ) {
		global $wpdb;
		$sql    = self::translate( $wpdb->remove_placeholder_escape( (string) $sql ) );
		$select = (bool) preg_match( '/^\s*(SELECT|WITH|SHOW|DESCRIBE|EXISTS)\b/i', $sql );
		if ( $select && ! preg_match( '/\bFORMAT\s+\w+\s*$/i', $sql ) ) {
			$sql .= ' FORMAT JSON';
		}
		$text = $this->request( $sql );
		if ( false === $text ) {
			return false;
		}
		if ( ! $select ) {
			return $rows ? array() : true;
		}
		$json = json_decode( $text, true );
		$out  = is_array( $json ) && isset( $json['data'] ) && is_array( $json['data'] ) ? $json['data'] : array();
		return $rows ? $out : count( $out );
	}

	/**
	 * Result rows.
	 *
	 * @param string $sql    Query.
	 * @param string $output OBJECT, OBJECT_K, ARRAY_A or ARRAY_N.
	 * @return array|null
	 */
	public function get_results( $sql, $output = OBJECT ) {
		$rows = $this->query( $sql, true );
		if ( false === $rows ) {
			return null;
		}
		if ( ARRAY_A === $output ) {
			return $rows;
		}
		if ( ARRAY_N === $output ) {
			return array_map( 'array_values', $rows );
		}
		if ( OBJECT_K === $output ) {
			$out = array();
			foreach ( $rows as $r ) {
				$key = reset( $r );
				if ( ! isset( $out[ $key ] ) ) {
					$out[ $key ] = (object) $r;
				}
			}
			return $out;
		}
		return array_map(
			static function ( $r ) {
				return (object) $r;
			},
			$rows
		);
	}

	/**
	 * One row.
	 *
	 * @param string $sql    Query.
	 * @param string $output Output type.
	 * @return mixed
	 */
	public function get_row( $sql, $output = OBJECT ) {
		$rows = $this->get_results( $sql, $output );
		return $rows ? reset( $rows ) : null;
	}

	/**
	 * One value.
	 *
	 * @param string $sql Query.
	 * @return mixed
	 */
	public function get_var( $sql ) {
		$row = $this->get_row( $sql, ARRAY_N );
		return $row ? $row[0] : null;
	}

	/**
	 * First column.
	 *
	 * @param string $sql Query.
	 * @return array
	 */
	public function get_col( $sql ) {
		return array_map(
			static function ( $r ) {
				return $r[0];
			},
			(array) $this->get_results( $sql, ARRAY_N )
		);
	}

	/**
	 * Inserts rows (JSONEachRow, asynchronous insert on the server).
	 *
	 * @param string $table Table.
	 * @param array  $rows  Rows.
	 * @return bool
	 */
	public function insert_rows( $table, $rows ) {
		if ( ! $rows ) {
			return true;
		}
		$lines = array();
		foreach ( $rows as $row ) {
			$lines[] = wp_json_encode( $row, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES );
		}
		return false !== $this->request( 'INSERT INTO `' . str_replace( '`', '', $table ) . '` FORMAT JSONEachRow', implode( "\n", $lines ) );
	}

	/**
	 * Charset clause (not used by ClickHouse).
	 *
	 * @return string
	 */
	public function get_charset_collate() {
		return '';
	}

	/**
	 * Nothing to close: every request is separate.
	 */
	public function close() {}
}

/**
 * Schema, writes, maintenance and data copy for the ClickHouse storage.
 */
class Visitrix_ClickHouse {

	/** Minimal server version: window functions and JOIN filters. */
	const MIN_VERSION = '23.3';

	/**
	 * Events table.
	 *
	 * @param Visitrix_ClickHouse_DB $db Connection.
	 * @return string
	 */
	public static function events( $db ) {
		return $db->prefix . 'visitrix_events';
	}

	/**
	 * Visitors carried over from the MySQL tables (first visit and visits that are no longer in the events).
	 *
	 * @param Visitrix_ClickHouse_DB $db Connection.
	 * @return string
	 */
	public static function seed( $db ) {
		return $db->prefix . 'visitrix_seed';
	}

	/**
	 * Retention period in days.
	 *
	 * @return int
	 */
	private static function days() {
		return max( 1, (int) Visitrix_Settings::get()['retention_days'] );
	}

	/**
	 * Creates the tables and the views.
	 *
	 * @param Visitrix_ClickHouse_DB $db Connection.
	 * @return bool
	 */
	public static function install( $db ) {
		$p    = $db->prefix . 'visitrix_';
		$days = self::days();
		$ev   = '`' . $p . 'ev`';
		$base = '`' . $p . 'visits_base`';

		$sql = array(
			"CREATE TABLE IF NOT EXISTS `{$p}events` (
				created_at DateTime('UTC'),
				ts UInt32,
				ts_us UInt64,
				kind LowCardinality(String),
				visit_key String,
				visitor_key String,
				id_type LowCardinality(String),
				source_type LowCardinality(String),
				source_name String,
				referrer String,
				landing String,
				utm_source String,
				utm_medium String,
				utm_campaign String,
				device LowCardinality(String),
				browser LowCardinality(String),
				os LowCardinality(String),
				path String,
				title String,
				post_id UInt64,
				scroll UInt8,
				goal_type LowCardinality(String),
				goal_target String
			) ENGINE = MergeTree ORDER BY (visit_key, ts_us) TTL created_at + INTERVAL {$days} DAY",
			"CREATE TABLE IF NOT EXISTS `{$p}seed` (
				visitor_key String,
				first_seen DateTime('UTC'),
				last_seen DateTime('UTC'),
				extra UInt32
			) ENGINE = ReplacingMergeTree ORDER BY visitor_key TTL last_seen + INTERVAL {$days} DAY",
			// Events with their pageview number in the visit and a flag for activity after a 30-minute pause.
			"CREATE OR REPLACE VIEW {$ev} AS
			SELECT *, max(toUInt8(toInt64(ts) - toInt64(prev_ts) > " . Visitrix_Tracker::VISIT_TIMEOUT . ")) OVER (PARTITION BY visit_key ORDER BY ts_us ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS broke
			FROM (
				SELECT *,
					sum(toUInt32(kind = 'pv')) OVER (PARTITION BY visit_key ORDER BY ts_us ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS hit_no,
					lagInFrame(ts, 1, ts) OVER (PARTITION BY visit_key ORDER BY ts_us ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS prev_ts
				FROM `{$p}events`
			)",
			// One row per visit, the rules of the MySQL tracker: the first pageview starts the visit, pageviews and pings
			// extend it while it is active, a second pageview or an interaction makes it engaged.
			"CREATE OR REPLACE VIEW {$base} AS
			SELECT
				toInt64(bitShiftRight(cityHash64(visit_key), 1)) AS id,
				visit_key,
				argMinIf(visitor_key, ts_us, kind = 'pv') AS visitor_key,
				minIf(created_at, kind = 'pv') AS started_at,
				minIf(ts, kind = 'pv') AS ts_start,
				maxIf(ts, broke = 0 AND kind IN ('pv', 'ping')) AS ts_last,
				least(toUInt32(greatest(toInt64(maxIf(ts, broke = 0 AND kind IN ('pv', 'ping'))) - toInt64(minIf(ts, kind = 'pv')), 0)), " . Visitrix_Tracker::MAX_DURATION . ") AS duration,
				least(ifNull(countIf(kind = 'pv' AND broke = 0), 0), 65000) AS pageviews,
				toUInt8(ifNull(countIf(kind = 'engage'), 0) > 0 OR ifNull(countIf(kind = 'pv'), 0) >= 2) AS engaged,
				minIf(ts_us, kind = 'engage' OR (kind = 'pv' AND hit_no >= 2)) AS engaged_us,
				argMinIf(id_type, ts_us, kind = 'pv') AS id_type,
				argMinIf(source_type, ts_us, kind = 'pv') AS source_type,
				argMinIf(source_name, ts_us, kind = 'pv') AS source_name,
				argMinIf(referrer, ts_us, kind = 'pv') AS referrer,
				argMinIf(landing, ts_us, kind = 'pv') AS landing,
				argMinIf(utm_source, ts_us, kind = 'pv') AS utm_source,
				argMinIf(utm_medium, ts_us, kind = 'pv') AS utm_medium,
				argMinIf(utm_campaign, ts_us, kind = 'pv') AS utm_campaign,
				argMinIf(device, ts_us, kind = 'pv') AS device,
				argMinIf(browser, ts_us, kind = 'pv') AS browser,
				argMinIf(os, ts_us, kind = 'pv') AS os
			FROM {$ev}
			GROUP BY visit_key
			HAVING countIf(kind = 'pv') > 0",
			// A visit is new when it is the visitor's first engaged visit and the visitor was not seen before the data was copied.
			"CREATE OR REPLACE VIEW `{$p}visits` AS
			SELECT b.id AS id, b.visit_key AS visit_key, b.visitor_key AS visitor_key, b.started_at AS started_at, b.ts_start AS ts_start,
				b.ts_last AS ts_last, b.duration AS duration, b.pageviews AS pageviews, b.engaged AS engaged, b.id_type AS id_type,
				b.source_type AS source_type, b.source_name AS source_name, b.referrer AS referrer, b.landing AS landing,
				b.utm_source AS utm_source, b.utm_medium AS utm_medium, b.utm_campaign AS utm_campaign,
				b.device AS device, b.browser AS browser, b.os AS os,
				toUInt8(b.engaged = 1 AND b.id_type != 'a' AND f.first_key = b.visit_key AND (s.visitor_key IS NULL OR s.first_seen >= b.started_at)) AS is_new
			FROM {$base} AS b
			LEFT JOIN (
				SELECT visitor_key, argMin(visit_key, engaged_us) AS first_key FROM {$base} WHERE engaged = 1 AND id_type != 'a' GROUP BY visitor_key
			) AS f ON f.visitor_key = b.visitor_key
			LEFT JOIN (SELECT visitor_key, min(first_seen) AS first_seen FROM `{$p}seed` GROUP BY visitor_key) AS s ON s.visitor_key = b.visitor_key",
			"CREATE OR REPLACE VIEW `{$p}hits` AS
			SELECT
				toInt64(bitShiftRight(cityHash64(visit_key, hit_no), 1)) AS id,
				toInt64(bitShiftRight(cityHash64(visit_key), 1)) AS visit_id,
				anyIf(created_at, kind = 'pv') AS created_at,
				anyIf(path, kind = 'pv') AS path,
				anyIf(title, kind = 'pv') AS title,
				anyIf(post_id, kind = 'pv') AS post_id,
				ifNull(maxIf(scroll, kind = 'ping'), 0) AS scroll
			FROM {$ev}
			WHERE hit_no > 0
			GROUP BY visit_key, hit_no
			HAVING countIf(kind = 'pv') > 0",
			"CREATE OR REPLACE VIEW `{$p}goals` AS
			SELECT
				toInt64(bitShiftRight(cityHash64(g.visit_key, g.ts_us, g.goal_type, g.goal_target), 1)) AS id,
				toInt64(bitShiftRight(cityHash64(g.visit_key), 1)) AS visit_id,
				g.created_at AS created_at,
				g.goal_type AS type,
				g.goal_target AS target,
				ifNull(h.path, '') AS path
			FROM (SELECT * FROM {$ev} WHERE kind = 'goal') AS g
			LEFT JOIN (SELECT visit_key, hit_no, any(path) AS path FROM {$ev} WHERE kind = 'pv' GROUP BY visit_key, hit_no) AS h
				ON h.visit_key = g.visit_key AND h.hit_no = g.hit_no",
			"CREATE OR REPLACE VIEW `{$p}visitors` AS
			SELECT
				ifNull(a.visitor_key, s.visitor_key) AS visitor_key,
				if(a.visitor_key IS NULL, s.first_seen, if(s.visitor_key IS NULL, a.first_seen, least(a.first_seen, s.first_seen))) AS first_seen,
				if(a.visitor_key IS NULL, s.last_seen, if(s.visitor_key IS NULL, a.last_seen, greatest(a.last_seen, s.last_seen))) AS last_seen,
				ifNull(a.visits, 0) + ifNull(s.extra, 0) AS visits
			FROM (
				SELECT visitor_key, min(started_at) AS first_seen, max(started_at) AS last_seen, count() AS visits
				FROM {$base} WHERE engaged = 1 AND id_type != 'a' GROUP BY visitor_key
			) AS a
			FULL OUTER JOIN (
				SELECT visitor_key, min(first_seen) AS first_seen, max(last_seen) AS last_seen, max(extra) AS extra FROM `{$p}seed` GROUP BY visitor_key
			) AS s ON s.visitor_key = a.visitor_key",
		);
		foreach ( $sql as $statement ) {
			if ( false === $db->request( $statement ) ) {
				return false;
			}
		}
		update_option( 'visitrix_ch_ttl', $days, false );
		return true;
	}

	/**
	 * Checks the server: version, rights to create, write, read and drop a table.
	 *
	 * @param array $s Settings with the plain password.
	 * @return true|WP_Error
	 */
	public static function test( $s ) {
		if ( ! preg_match( '#^https?://#i', (string) $s['url'] ) ) {
			return new WP_Error( 'visitrix_storage', __( 'Enter the ClickHouse HTTPS address, e.g. https://abc123.eu-central-1.aws.clickhouse.cloud:8443', 'visitrix' ) );
		}
		$db      = new Visitrix_ClickHouse_DB( $s, true );
		$version = $db->request( 'SELECT version()' );
		if ( false === $version ) {
			return new WP_Error( 'visitrix_storage', $db->last_error );
		}
		$version = trim( $version );
		if ( version_compare( $version, self::MIN_VERSION, '<' ) ) {
			/* translators: 1: server version, 2: required version. */
			return new WP_Error( 'visitrix_storage', sprintf( __( 'ClickHouse %1$s is too old: version %2$s or newer is needed.', 'visitrix' ), $version, self::MIN_VERSION ) );
		}
		$t  = '`' . $db->prefix . 'visitrix_check`';
		$ok = false !== $db->request( "CREATE TABLE IF NOT EXISTS {$t} (v String) ENGINE = MergeTree ORDER BY v" )
			&& false !== $db->request( "INSERT INTO {$t} FORMAT JSONEachRow", '{"v":"ok"}' )
			&& 'ok' === trim( (string) $db->request( "SELECT v FROM {$t} LIMIT 1 FORMAT TSV" ) )
			&& false !== $db->request( "DROP TABLE IF EXISTS {$t}" );
		return $ok ? true : new WP_Error(
			'visitrix_storage',
			/* translators: %s: database error. */
			sprintf( __( 'Connected, but the user cannot create or write tables: %s', 'visitrix' ), $db->last_error )
		);
	}

	/**
	 * Keeps the TTL in line with the retention setting (daily cron).
	 *
	 * @param Visitrix_ClickHouse_DB $db Connection.
	 */
	public static function maintain( $db ) {
		$days = self::days();
		if ( (int) get_option( 'visitrix_ch_ttl' ) === $days ) {
			return;
		}
		$ok = false !== $db->request( 'ALTER TABLE `' . self::events( $db ) . "` MODIFY TTL created_at + INTERVAL {$days} DAY" )
			&& false !== $db->request( 'ALTER TABLE `' . self::seed( $db ) . "` MODIFY TTL last_seen + INTERVAL {$days} DAY" );
		if ( $ok ) {
			update_option( 'visitrix_ch_ttl', $days, false );
		}
	}

	/**
	 * Removes all statistics.
	 *
	 * @param Visitrix_ClickHouse_DB $db Connection.
	 */
	public static function truncate( $db ) {
		$db->request( 'TRUNCATE TABLE IF EXISTS `' . self::events( $db ) . '`' );
		$db->request( 'TRUNCATE TABLE IF EXISTS `' . self::seed( $db ) . '`' );
	}

	/**
	 * Latest still-active visit of an anonymous visitor.
	 *
	 * @param Visitrix_ClickHouse_DB $db          Connection.
	 * @param string                 $visitor_key Visitor hash.
	 * @param int                    $since       Unix time.
	 * @return array|null {visit_key, referrer}
	 */
	public static function recent_visit( $db, $visitor_key, $since ) {
		$row = $db->get_row(
			$db->prepare(
				'SELECT visit_key, argMinIf(referrer, ts_us, kind = %s) AS referrer FROM %i
				WHERE visit_key IN (SELECT visit_key FROM %i WHERE visitor_key = %s AND ts >= %d)
				GROUP BY visit_key ORDER BY max(ts_us) DESC LIMIT 1',
				'pv',
				self::events( $db ),
				self::events( $db ),
				$visitor_key,
				$since
			),
			ARRAY_A
		);
		return $row ? $row : null;
	}

	/**
	 * Event row with every column set.
	 *
	 * @param array $values Values.
	 * @return array
	 */
	public static function row( $values ) {
		static $seq = 0;
		$now = microtime( true );
		return array_merge(
			array(
				'created_at'   => current_time( 'mysql' ),
				'ts'           => (int) $now,
				'ts_us'        => (int) round( $now * 1000000 ) + ( $seq++ % 1000 ),
				'kind'         => '',
				'visit_key'    => '',
				'visitor_key'  => '',
				'id_type'      => 's',
				'source_type'  => '',
				'source_name'  => '',
				'referrer'     => '',
				'landing'      => '',
				'utm_source'   => '',
				'utm_medium'   => '',
				'utm_campaign' => '',
				'device'       => '',
				'browser'      => '',
				'os'           => '',
				'path'         => '',
				'title'        => '',
				'post_id'      => 0,
				'scroll'       => 0,
				'goal_type'    => '',
				'goal_target'  => '',
			),
			$values
		);
	}

	/* ---------- Copy from the WordPress database ---------- */

	/**
	 * Copies a chunk of visits (with their pageviews and goals) or visitors from the WordPress tables.
	 *
	 * @param Visitrix_ClickHouse_DB $db    Connection.
	 * @param string                 $table visits|visitors (hits and goals are copied with their visits).
	 * @param string                 $after Last copied key.
	 * @return array|WP_Error {copied, next, done}
	 */
	public static function copy_chunk( $db, $table, $after ) {
		global $wpdb;
		$p = $wpdb->prefix . 'visitrix_';
		if ( 'hits' === $table || 'goals' === $table ) {
			return array( 'copied' => 0, 'next' => '', 'done' => true );
		}
		if ( 'visitors' === $table ) {
			return self::copy_visitors( $db, $after );
		}
		$visits = $wpdb->get_results( $wpdb->prepare( 'SELECT * FROM %i WHERE id > %d ORDER BY id LIMIT 500', $p . 'visits', (int) $after ), ARRAY_A ); // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- the plugin's own tables, read once while copying.
		if ( ! $visits ) {
			return array( 'copied' => 0, 'next' => '', 'done' => true );
		}
		$ids  = array_map( 'intval', wp_list_pluck( $visits, 'id' ) );
		$in   = implode( ',', array_fill( 0, count( $ids ), '%d' ) );
		// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.PreparedSQLPlaceholders.UnfinishedPrepare, WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- $in contains only %d placeholders; the plugin's own tables.
		$hits  = $wpdb->get_results( $wpdb->prepare( "SELECT * FROM %i WHERE visit_id IN ($in) ORDER BY id", array_merge( array( $p . 'hits' ), $ids ) ), ARRAY_A );
		$goals = $wpdb->get_results( $wpdb->prepare( "SELECT * FROM %i WHERE visit_id IN ($in) ORDER BY id", array_merge( array( $p . 'goals' ), $ids ) ), ARRAY_A );
		// phpcs:enable
		$by_visit = array();
		foreach ( (array) $hits as $h ) {
			$by_visit[ (int) $h['visit_id'] ]['hits'][] = $h;
		}
		foreach ( (array) $goals as $g ) {
			$by_visit[ (int) $g['visit_id'] ]['goals'][] = $g;
		}
		$tz   = wp_timezone();
		$unix = static function ( $local ) use ( $tz ) {
			try {
				return ( new DateTimeImmutable( (string) $local, $tz ) )->getTimestamp();
			} catch ( Exception $e ) {
				return 0;
			}
		};
		$rows = array();
		foreach ( $visits as $v ) {
			$common = array(
				'visit_key'    => $v['visit_key'],
				'visitor_key'  => $v['visitor_key'],
				'id_type'      => $v['id_type'],
				'source_type'  => $v['source_type'],
				'source_name'  => $v['source_name'],
				'referrer'     => $v['referrer'],
				'landing'      => $v['landing'],
				'utm_source'   => $v['utm_source'],
				'utm_medium'   => $v['utm_medium'],
				'utm_campaign' => $v['utm_campaign'],
				'device'       => $v['device'],
				'browser'      => $v['browser'],
				'os'           => $v['os'],
			);
			$ts0  = (int) $v['ts_start'];
			$us   = $ts0 * 1000000;
			$list = $by_visit[ (int) $v['id'] ]['hits'] ?? array();
			if ( ! $list ) {
				// A visit without stored pageviews still needs its first pageview.
				$list = array( array( 'created_at' => $v['started_at'], 'path' => $v['landing'], 'title' => '', 'post_id' => 0, 'scroll' => 0 ) );
			}
			foreach ( $list as $i => $h ) {
				$ts = 0 === $i ? $ts0 : max( $ts0, $unix( $h['created_at'] ) );
				$us = max( $us + 1, $ts * 1000000 );
				$rows[] = self::row( $common + array( 'created_at' => 0 === $i ? $v['started_at'] : $h['created_at'], 'ts' => $ts, 'ts_us' => $us, 'kind' => 'pv', 'path' => $h['path'], 'title' => $h['title'], 'post_id' => (int) $h['post_id'] ) );
				if ( (int) $h['scroll'] ) {
					$us     = $us + 1;
					$rows[] = self::row( array( 'created_at' => $h['created_at'], 'ts' => $ts, 'ts_us' => $us, 'kind' => 'ping', 'visit_key' => $v['visit_key'], 'visitor_key' => $v['visitor_key'], 'scroll' => (int) $h['scroll'] ) );
				}
				foreach ( $by_visit[ (int) $v['id'] ]['goals'] ?? array() as $g ) {
					// Goals belong to the pageview they were reached on.
					if ( $g['path'] === $h['path'] && empty( $g['_done'] ) ) {
						$us     = $us + 1;
						$rows[] = self::row( array( 'created_at' => $g['created_at'], 'ts' => $ts, 'ts_us' => $us, 'kind' => 'goal', 'visit_key' => $v['visit_key'], 'visitor_key' => $v['visitor_key'], 'goal_type' => $g['type'], 'goal_target' => $g['target'] ) );
					}
				}
				foreach ( $by_visit[ (int) $v['id'] ]['goals'] ?? array() as $k => $g ) {
					if ( $g['path'] === $h['path'] ) {
						$by_visit[ (int) $v['id'] ]['goals'][ $k ]['_done'] = 1;
					}
				}
			}
			// The last activity of the visit keeps its duration.
			$end = $ts0 + (int) $v['duration'];
			if ( $end * 1000000 > $us ) {
				$rows[] = self::row( array( 'created_at' => wp_date( 'Y-m-d H:i:s', $end ), 'ts' => $end, 'ts_us' => $end * 1000000, 'kind' => 'ping', 'visit_key' => $v['visit_key'], 'visitor_key' => $v['visitor_key'] ) );
			}
			if ( (int) $v['engaged'] && count( $list ) < 2 ) {
				$rows[] = self::row( array( 'created_at' => $v['started_at'], 'ts' => $ts0, 'ts_us' => $ts0 * 1000000 + 999, 'kind' => 'engage', 'visit_key' => $v['visit_key'], 'visitor_key' => $v['visitor_key'] ) );
			}
		}
		if ( ! $db->insert_rows( self::events( $db ), $rows ) ) {
			return new WP_Error( 'visitrix_storage', $db->last_error );
		}
		return array(
			'copied' => count( $visits ) + count( (array) $hits ) + count( (array) $goals ),
			'next'   => (string) end( $ids ),
			'done'   => count( $visits ) < 500,
		);
	}

	/**
	 * Visitors of the WordPress tables: first visit and the visits that are not among the copied ones.
	 *
	 * @param Visitrix_ClickHouse_DB $db    Connection.
	 * @param string                 $after Last copied visitor key.
	 * @return array|WP_Error
	 */
	private static function copy_visitors( $db, $after ) {
		global $wpdb;
		$p    = $wpdb->prefix . 'visitrix_';
		$rows = $wpdb->get_results( // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- the plugin's own tables, read once while copying.
			$wpdb->prepare(
				'SELECT r.visitor_key, r.first_seen, r.last_seen, r.visits, COUNT(v.id) AS kept
				FROM %i r LEFT JOIN %i v ON v.visitor_key = r.visitor_key AND v.engaged = 1 AND v.id_type <> %s
				WHERE r.visitor_key > %s GROUP BY r.visitor_key, r.first_seen, r.last_seen, r.visits ORDER BY r.visitor_key LIMIT 2000',
				$p . 'visitors',
				$p . 'visits',
				'a',
				(string) $after
			),
			ARRAY_A
		);
		if ( ! $rows ) {
			return array( 'copied' => 0, 'next' => '', 'done' => true );
		}
		$seed = array();
		foreach ( $rows as $r ) {
			$seed[] = array(
				'visitor_key' => $r['visitor_key'],
				'first_seen'  => $r['first_seen'],
				'last_seen'   => $r['last_seen'],
				'extra'       => max( 0, (int) $r['visits'] - (int) $r['kept'] ),
			);
		}
		if ( ! $db->insert_rows( self::seed( $db ), $seed ) ) {
			return new WP_Error( 'visitrix_storage', $db->last_error );
		}
		$last = end( $rows );
		return array(
			'copied' => count( $rows ),
			'next'   => (string) $last['visitor_key'],
			'done'   => count( $rows ) < 2000,
		);
	}
}