class-remote-source-node.php

<?php
/**
 * Remote_Source: one SSE connection to a spoke, carrying a durable reader per stream.
 *
 * @package Newspack_Nodes
 */

namespace Newspack_Nodes;

use Newspack_Nodes\Rest\SSE_Out_Node;

\defined( 'ABSPATH' ) || exit;

/**
 * An SSE-pull broker: one connection to a spoke carries every stream its
 * sources name, and each stamp it carries is read by its own
 * `Remote_Consumer_Node`.
 *
 * The command channel, the sources and the readers come from
 * `Remote_Broker_Node`. This class adds the stream: an `SSE_In` patron
 * (`<name>:sse-in`), the slot-keepalive heartbeat it mints every
 * ~HEARTBEAT_INTERVAL through the channel's HTTP_Out (whose reply self-routes
 * back into `fill()` for RTT bookkeeping), the reconnect, and the routing
 * between the stream and the readers. The tick connects, under `SSE_In`'s
 * backoff; a busy stream's scheduled close makes the housekeeping due on the
 * timer scan of the same drain iteration instead (`housekeep_now()`). `SSE_In` hands each raw
 * `msg` payload to `route()`, which reads its FROM stamp and hands it to the
 * reader the first matching source owns, building that reader on first sight.
 * A connect asks each live reader where its stream stands; a paused reader
 * drops out of the request and steps through `send_read()` over the HTTP_Out
 * instead. One valve meters the whole connection, on the bytes every reader
 * holds.
 *
 * Mirrors the JS RemoteLinkNode's channel, which the browser composes the
 * same way over its one page-wide stream.
 *
 * @phpstan-import-type Stream_Positions from SSE_In_Node
 * @phpstan-type Stream_Request array{0:list<string>,1:Stream_Positions}
 */
class Remote_Source_Node extends Remote_Broker_Node {

	/** `SSE_Slot_Pool` lease state for a slot released under us — expected, not an error. */
	public const RELEASED_SLOT = 'slot_released';

	/**
	 * SSE backpressure valve water marks, in raw blocks: disarm once four are
	 * held, re-arm once their drains bring it to two.
	 *
	 * The readers drain between arrivals, so their buffers accumulate in route():
	 * disarm once the sum crosses the high mark, re-arm once their drains bring
	 * it back to the low one. The hysteresis is what keeps the valve OPEN
	 * through normal flow, and a block arriving never crosses it alone.
	 */
	private const PUMP_ARM_BYTES    = 2 * Log_Sources::BLOCK_BYTES;
	private const PUMP_DISARM_BYTES = 4 * Log_Sources::BLOCK_BYTES;

	/** Patron SSE_In sibling (`<name>:sse-in`); null until the Vault entry resolves. */
	protected ?SSE_In_Node $sse_in = null;

	/** When the last heartbeat went out, fraction kept for its RTT; 0 before any. */
	private float $last_heartbeat_sent = 0.0;

	/** Wall-second of the last heartbeat reply; 0 until the first one comes back. */
	private int $last_heartbeat_response = 0;

	/** Reason the last heartbeat failed, published as `last_error`; null on success. */
	private ?string $last_heartbeat_error = null;

	/**
	 * The `[ slot, owner ]` lease the last heartbeat was sent for; null before
	 * any. At most one heartbeat is normally in flight: one goes out each
	 * `HEARTBEAT_INTERVAL`, and `HTTP_Out` gives up on its POST within
	 * `REQUEST_TIMEOUT`, the same 15 s, so one field suffices. A reply slower
	 * than that is judged against the newer heartbeat's lease.
	 *
	 * @var array{0:int,1:int}|null
	 */
	private ?array $heartbeat_lease = null;

	/** Mirror of the SSE_In valve state: true while armed. Only the readers' bytes flip it. */
	private bool $pump_armed = true;

	/**
	 * Each tick, after putting back the recurring cadence a `housekeep_now()`
	 * replaced. The re-arm comes first, before anything that can throw, so
	 * neither an early return nor a throw leaves the broker on its 0 ms slot.
	 *
	 * @api Dynamic entrypoint (Timer_Node::fire_cb).
	 */
	public function fire(): void {
		if ( self::TICK_INTERVAL_MS !== $this->interval_ms ) {
			$this->set_timer( self::TICK_INTERVAL_MS );
		}
		parent::fire();
	}

	/**
	 * One tick's housekeeping: drive the passive SSE_In, keep the slot
	 * alive, and publish the status snapshot. Idempotent and cheap.
	 * `should_connect()` gates whether a tick initiates or keeps the
	 * connection.
	 */
	protected function housekeep(): void {
		if ( ! $this->should_connect() ) {
			$this->publish_connection();
			return;
		}
		$sse = $this->ensure_patrons();
		if ( null === $sse ) {
			return;
		}
		$sse->check_stale();
		$sse->maybe_connect();
		$this->maybe_send_heartbeat();
		$this->publish_connection();
	}

	/**
	 * Build the channel, then the SSE_In patron configured from the Vault
	 * entry, and wire its seams to the readers: the handshake and skip frames
	 * hand each stamp's cursor to the reader that owns it, and each raw `msg`
	 * payload goes through `route()`. The seal-grace flag is seeded onto it,
	 * because an aggregator configures its spokes before anything connects.
	 *
	 * @return SSE_In_Node|null The SSE_In patron once configured, else null.
	 */
	protected function ensure_patrons(): ?SSE_In_Node {
		if ( null !== $this->sse_in ) {
			return $this->sse_in;
		}
		$entry = $this->spoke_entry();
		if ( null === $entry || null === $this->ensure_channel() ) {
			return null;
		}
		$sse = new SSE_In_Node();
		$sse->patron( $this );
		$sse->on_connecting = function () use ( $sse ): void {
			[ $subscribe, $positions ] = $this->stream_request();
			$sse->streams( $subscribe, $positions );
		};
		$sse->configure(
			Vault::url_of( $entry ),
			Vault::credential_header( $entry ),
			Vault::tls_opts(),
			Vault::require_ssl()
		);
		$sse->set_multi_writer( $this->multi_writer );
		$sse->on_connected = function ( array $starts ): void {
			/** @var array<string,array{segment?:int,offset:int,distance?:int}> $starts */
			foreach ( $starts as $stamp => $start ) {
				$this->consumer_for( $stamp )?->adopt_stream_start( $start );
			}
		};
		$sse->on_message = function ( string $raw ): void {
			$this->route( $raw );
		};
		$sse->on_reopen_due = $this->housekeep_now( ... );
		$this->sse_in = $sse;
		$this->publish_patron( self::SSE_IN_KIND, $sse );
		return $sse;
	}

	/**
	 * Make the housekeeping due on the timer scan that follows the cURL
	 * completion asking for it, later in the same drain iteration: a recurring
	 * 0 ms slot of its own in place of the Router hitchhike, which `fire()`
	 * swaps back. `SSE_In` asks from inside that completion, where connecting
	 * is not allowed. A stopped broker stays stopped.
	 */
	private function housekeep_now(): void {
		if ( ! $this->timer_is_active() ) {
			return;
		}
		$this->set_timer( 0 );
	}

	/**
	 * What the next connect asks for: each exact source whose reader is live,
	 * from that reader's own position, and each glob, with every reader it
	 * already owns stated — a paused one as `SSE_Out_Node::SKIP`, so the glob
	 * keeps finding new dirs while that one stays out. A new stream opens with
	 * its valve armed.
	 *
	 * @return Stream_Request Subscriptions, then per-stamp positions.
	 */
	private function stream_request(): array {
		$this->pump_armed = true;
		$subscribe        = [];
		$positions        = [];
		foreach ( $this->sources as $source ) {
			$subscription = $source['source'];
			if ( ! Log_Discovery::is_glob( $subscription ) ) {
				$child = $this->consumer_for( $subscription );
				if ( null !== $child && $child->is_live() ) {
					$subscribe[]                = $subscription;
					$positions[ $subscription ] = $child->connect_position();
				}
				continue;
			}
			$subscribe[] = $subscription;
			foreach ( $this->readers() as $child_stamp => $child ) {
				if ( $this->source_for( $child_stamp ) === $source ) {
					$positions[ $child_stamp ] = $child->is_live() ? $child->connect_position() : SSE_Out_Node::SKIP;
				}
			}
		}
		return [ $subscribe, $positions ];
	}

	/**
	 * Route one raw block, or its header, to the reader its FROM stamp names,
	 * which splits the block into records. A frame that will not unpack has
	 * nothing to route by, and one that is not a raw block does not belong to
	 * any stream this broker reads: both are dropped, rate-limited.
	 * `consumer_for()` refuses the rest.
	 *
	 * @param string $raw One packed Message.
	 */
	private function route( string $raw ): void {
		try {
			$message = Message::unpacked( $raw );
		} catch ( \InvalidArgumentException $e ) {
			$this->print_less_often( 'dropping unparseable SSE frame' );
			return;
		}
		$from = Core::as_string( $message[ Message::FROM ] );
		if ( ! ( Core::int( $message[ Message::TYPE ] ) & Message::TM_BATCH ) ) {
			$this->print_less_often( 'dropping a frame that is not a raw block: ', $from );
			return;
		}
		$child = $this->consumer_for( Log_Discovery::dir_from_stamp( $from ) );
		if ( null === $child ) {
			return;
		}
		$child->fill( $message );
		$this->pump_maybe_disarm();
	}

	/** Close the valve once the readers' backlog crosses the high-water mark. */
	private function pump_maybe_disarm(): void {
		if ( $this->pump_armed && $this->buffered_bytes() >= self::PUMP_DISARM_BYTES ) {
			$this->sse_in?->disarm();
			$this->pump_armed = false;
		}
	}

	/**
	 * Connect while any exact source's reader is live, or any glob may yet
	 * find a dir. The tick builds each exact source's reader before
	 * housekeeping, so this only reads them.
	 */
	protected function should_connect(): bool {
		$connect = false;
		foreach ( $this->sources as [ 'source' => $source ] ) {
			$live    = Log_Discovery::is_glob( $source ) || ( $this->reader( $source )?->is_live() ?? false );
			$connect = $connect || $live;
		}
		return $connect;
	}

	/**
	 * Every ~HEARTBEAT_INTERVAL seconds, mint a `workers.heartbeat` TM_COMMAND
	 * (FROM=<this node>, TO=workers, args `<slot> <owner>`) and fill it into the
	 * patron HTTP_Out. Skips until SSE_In reports the complete lease. The slot
	 * pool keys on (user, ip, slot, owner) — no partition.
	 *
	 * Signing needs a session with the spoke, so `mint()` mints nothing on a
	 * tick that finds none, asks for one on the broker's own throttle, and
	 * leaves the heartbeat clock where it was.
	 */
	private function maybe_send_heartbeat(): void {
		if ( null === $this->sse_in || null === $this->http_out ) {
			return;
		}
		$slot  = $this->sse_in->slot();
		$owner = $this->sse_in->owner();
		if ( null === $slot || $slot < 0 || null === $owner || $owner <= 0 ) {
			return;
		}
		$now = (int) Core::$now;
		// The lease exists from this tick; the session offset counts from it.
		if ( 0 === $this->lease_epoch ) {
			$this->lease_epoch = $now;
		}
		// Silence is not a refusal — HTTP_Out drops the session on a 401.
		if ( '' === $this->http_out->vault_id() || $now - (int) $this->last_heartbeat_sent < self::HEARTBEAT_INTERVAL ) {
			return;
		}
		$message = $this->mint( $this->name, 'workers', 'heartbeat', [ (string) $slot, (string) $owner ] );
		if ( null === $message ) {
			return;
		}
		$this->last_heartbeat_sent = Core::$now;
		$this->heartbeat_lease     = [ $slot, $owner ];
		++$this->counter;
		$this->http_out->fill( $message );
	}

	/**
	 * Publish the connection-state snapshot from SSE_In::connection(). Ages out the heartbeat
	 * round-trip so the dashboard's Status badge can't latch 'success' on a stale timestamp: the
	 * response is "live" only while connected AND seen within the node's HEARTBEAT_INTERVAL*4
	 * window, and is nulled otherwise. While it IS live the two keys are left out of the
	 * write, so what record_heartbeat_reply() merged stands.
	 */
	private function publish_connection(): void {
		$conn = null !== $this->sse_in
			? $this->sse_in->connection()
			: [ 'connected' => false, 'connecting' => false, 'last_http_code' => null, 'last_error' => null, 'current_backoff' => SSE_In_Node::INITIAL_BACKOFF, 'last_sse_heartbeat' => null, 'last_attempt' => null, 'scheduled_reconnect_at' => null ];
		$data = [
			'last_connection_attempt' => $conn['last_attempt'],
			'connected'               => $conn['connected'],
			// A socket mid-open: neither up nor a failure to rail red.
			'connecting'              => $conn['connecting'],
			'last_http_code'          => $conn['last_http_code'],
			'last_error'              => $conn['last_error'] ?? $this->last_heartbeat_error,
			'current_backoff'         => $conn['current_backoff'],
			'last_sse_heartbeat'      => $conn['last_sse_heartbeat'],
			// The dashboard's idle reading: closed on purpose, back at T.
			'scheduled_reconnect_at'  => $conn['scheduled_reconnect_at'],
		];
		// Live only while connected AND the response is within slot-TTL window.
		$hb_live = $conn['connected']
			&& $this->last_heartbeat_response > 0
			&& ( (int) Core::$now - $this->last_heartbeat_response ) <= self::HEARTBEAT_INTERVAL * 4;
		if ( ! $hb_live ) {
			$data['last_response'] = null;
			$data['last_rtt']      = null;
		}
		$this->publish_status( $data );
	}

	/**
	 * The heartbeat's reply: a success records the round-trip, and a failure
	 * is said and published unless it is a race: a released slot, or a lease
	 * the stream no longer holds, which a reconnect retired and the spoke's
	 * pool may have handed a new owner while the heartbeat was on the wire.
	 *
	 * @param array<int,mixed> $message The reply, TM_COMMAND|TM_RESPONSE or |TM_ERROR.
	 */
	protected function settle_reply( array $message ): void {
		$failure = self::heartbeat_failure( $message );
		if ( null === $failure ) {
			$this->record_heartbeat_reply();
		} elseif ( ! \str_contains( $failure, self::RELEASED_SLOT ) && $this->holds_heartbeat_lease() ) {
			$this->stderr( 'ERROR: client heartbeat failed - ' . $failure );
			$this->record_heartbeat_failure( $failure );
		}
	}

	/** Whether the stream still holds the lease the heartbeat in flight was sent for. */
	private function holds_heartbeat_lease(): bool {
		return null !== $this->sse_in && null !== $this->heartbeat_lease && [ $this->sse_in->slot(), $this->sse_in->owner() ] === $this->heartbeat_lease;
	}

	/**
	 * Return null only for an explicit successful heartbeat response.
	 *
	 * @param array<int,mixed> $message Command response/error envelope.
	 * @return string|null The failure reason, or null on success.
	 */
	private static function heartbeat_failure( array $message ): ?string {
		$type    = Core::int( $message[ Message::TYPE ] );
		$value   = $message[ Message::VALUE ];
		$payload = \is_array( $value ) && \array_key_exists( 'payload', $value )
			? $value['payload']
			: $value;

		if ( $type & Message::TM_ERROR ) {
			return self::failure_reason( $payload, 'heartbeat command failed' );
		}
		if (
			! \is_array( $payload )
			|| ! \array_key_exists( 'success', $payload )
			|| true !== $payload['success']
		) {
			return self::failure_reason( $payload, 'heartbeat response was not successful' );
		}
		return null;
	}

	/**
	 * Record a heartbeat reply's round-trip into the status snapshot and clear the stored
	 * failure. `last_error` is republished from SSE_In's connection rather than blanked,
	 * because the stream can be down while the command channel still answers, and the
	 * dashboard badge has to say so.
	 */
	private function record_heartbeat_reply(): void {
		if ( 0.0 === $this->last_heartbeat_sent ) {
			return;
		}
		$now                           = (int) Core::$now;
		$this->last_heartbeat_response = $now;
		$this->last_heartbeat_error    = null;
		$connection_error              = null !== $this->sse_in
			? $this->sse_in->connection()['last_error']
			: null;
		$this->publish_status( [
			'last_response'           => $now,
			'last_rtt'                => self::round_trip_ms( Core::$now - $this->last_heartbeat_sent ),
			'last_error'              => $connection_error,
		] );
	}

	/** Clear a prior success immediately and retain the spoke's safe failure reason. */
	private function record_heartbeat_failure( string $reason ): void {
		$this->last_heartbeat_response = 0;
		$this->last_heartbeat_error    = 'Client heartbeat failed: ' . $reason;
		$this->publish_status( [
			'last_response'           => null,
			'last_rtt'                => null,
			'last_error'              => $this->last_heartbeat_error,
		] );
	}

	/**
	 * Re-open the valve once the readers have drained below the low-water mark.
	 *
	 * @param Remote_Consumer_Node $child The reader that drained.
	 */
	public function refill( Remote_Consumer_Node $child ): void {
		if ( ! $this->pump_armed && $this->buffered_bytes() <= self::PUMP_ARM_BYTES ) {
			$this->sse_in?->arm();
			$this->pump_armed = true;
		}
	}

	/** Bytes buffered across every reader: one connection, one valve. */
	private function buffered_bytes(): int {
		$bytes = 0;
		foreach ( $this->reader_walk() as $child ) {
			$bytes += $child->buffered_bytes();
		}
		return $bytes;
	}

	/**
	 * The seal-grace rides a connect-time query parameter, so a CHANGE
	 * restreams, and the tick reconnects past what is buffered.
	 *
	 * @param bool $flag Whether the spoke should apply the seal-grace.
	 */
	public function set_multi_writer( bool $flag ): void {
		$changed = $flag !== $this->multi_writer();
		parent::set_multi_writer( $flag );
		if ( null !== $this->sse_in && $changed ) {
			$this->sse_in->set_multi_writer( $flag );
			$this->restream();
		}
	}

	/**
	 * Drop the live stream so the next request states the live set again;
	 * what it left buffered still drains. The next tick reconnects, so
	 * readers that pause in one tick share one.
	 */
	public function restream(): void {
		$this->sse_in?->disconnect();
	}

	/**
	 * The spoke's last heartbeat while the stream is up. Its reader pushes
	 * every byte it reads, so a heartbeat says nothing more waits to be sent.
	 * A shut valve is the hub holding the spoke quiet, which says it now.
	 */
	public function heard_at(): ?float {
		if ( ! $this->pump_armed ) {
			return Core::$now;
		}
		$conn = $this->sse_in?->connection();
		return null !== $conn && $conn['connected'] && null !== $conn['last_sse_heartbeat'] ? (float) $conn['last_sse_heartbeat'] : null;
	}

	/** Tear down the SSE_In patron, then the channel. */
	protected function drop_patrons(): void {
		$this->sse_in = null;
		$this->retract_sibling( self::SSE_IN_KIND );
		parent::drop_patrons();
	}

	/** Composite stat delegation: report the children's tallies, not zeros. */
	public function counter(): int {
		return null !== $this->sse_in ? $this->sse_in->counter() : parent::counter();
	}

	/** SSE_In's tally: the source holds no socket of its own to read. */
	public function bytes_read(): int {
		return $this->sse_in?->bytes_read() ?? 0;
	}

	/** The largest SSE frame SSE_In has delivered. */
	public function largest_msg_sent(): int {
		return $this->sse_in?->largest_msg_sent() ?? 0;
	}

	/**
	 * Palette entry: the broker's form and commands, in the I/O category.
	 *
	 * @api Dynamic entrypoint.
	 * @return array<string,mixed>
	 */
	public static function node_schema(): array {
		return \array_merge( parent::node_schema(), [
			// The base hides itself; this subclass belongs in the palette.
			'category'    => 'I/O',
			'description' => 'SSE-pull broker: one connection to a spoke carrying several streams, a durable reader per source `add_source` names (Vault-resolved).',
		] );
	}
}