class-settings-sync-node.php

<?php
/**
 * Settings_Sync: the hub end of the settings-sync control plane.
 *
 * `Settings_Event_Writer` appends an option NAME to `settings.p0` on every
 * watched change, and a worker Consumer tails that log into this node. The value
 * a spoke receives is therefore the one read at consume time, not the one the
 * admin request saw. `add_setting()` declares which local option reaches which
 * spoke under which remote name, `remove_setting()` withdraws one, and
 * `dump_config()` replays the declarations left as TSL.
 *
 * @package Newspack_Nodes
 */

namespace Newspack_Nodes;

\defined( 'ABSPATH' ) || exit;

/**
 * Pushes every registered WP option to the connected spokes as a signed `set`.
 *
 * The node fans out itself rather than sinking into a Tee. `send_signed()` mints
 * one command per live target and signs it under that spoke's session key, and
 * the key chosen IS the destination binding
 * ([ADR-15](docs/architecture-decisions.md#adr-15-command-authorization-local-taint--the-minter-signs)),
 * so a Tee re-addressing one signed command to N spokes would produce a command
 * that verifies nowhere.
 *
 * Two things drive a push. An event from the Consumer pushes the one option that
 * changed; the recurring sweep pushes every registered option, which is how a
 * spoke connecting mid-stream converges without an event of its own.
 */
class Settings_Sync_Node extends Timer_Node {
	use Schema_Reflection;
	use Fanout_Targets;

	/** Sweep cadence in seconds when the make_node line carries no interval token. */
	private const DEFAULT_INTERVAL_SECONDS = 300;

	/** Re-push sweep cadence in seconds; positional 0, floored at one second by cadence_ms(). */
	protected int $interval_seconds = self::DEFAULT_INTERVAL_SECONDS;

	/**
	 * Local option name to the mappings it pushes under: `to` is the path below
	 * every spoke target, `remote` the option name to set there.
	 *
	 * One local option maps to several. A hub `remote_*` setting seeds both the
	 * spoke's stripped option, which is that spoke's own config, and the spoke's
	 * `remote_*` copy, which the spoke propagates onward to ITS spokes.
	 *
	 * @var array<string,list<array{to:string,remote:string}>>
	 */
	protected array $registry = [];

	/** No-arg constructor (Tachikoma parity); wires the sibling `:config` interpreter from node_schema()['commands']. */
	public function __construct() {
		parent::__construct();
		$this->auto_wire_interpreter();
	}

	/**
	 * Read the argument tokens, or apply them and arm the recurring sweep.
	 *
	 * `Timer_Node::arguments()` arms only a bare `Timer`, so a subclass arms
	 * itself: drop the `set_timer()` call and the node parses a cadence it then
	 * never fires on. `parse_schema_args()` assigns `interval_seconds` from the
	 * schema declaration, and that declaration is the whole parse
	 * ([ADR-11](docs/architecture-decisions.md#adr-11-make_node-construction-sequence)).
	 *
	 * @param list<string>|null $args Interval in seconds at token 0, blank for the schema default, or null to read back.
	 * @return list<string> Last-set argument tokens.
	 * @throws \InvalidArgumentException When the interval token is not a whole number.
	 */
	public function arguments( ?array $args = null ): array {
		if ( null === $args ) {
			return $this->arguments;
		}
		$this->parse_schema_args( $args );
		$this->set_timer( $this->cadence_ms( $this->interval_seconds ) );
		return $this->arguments;
	}

	/**
	 * On a settings-change event, push the named option to the spokes, and
	 * every registered option whose substrate key follows its key, since a
	 * blank follower reads the changed value.
	 *
	 * The Consumer feeds a TM_STRUCT carrying only the option NAME
	 * (`VALUE = ['option' => $name]`); the effective value is read here at
	 * consume time, after both caches are dropped, because this worker lives
	 * ~595 s. Anything that isn't a TM_STRUCT with an 'option' key is dropped,
	 * and so is an option this node would not push, before it pays the
	 * flush: `Settings_Event_Writer` records every `newspack_*` option. Every
	 * option is attempted, and what any of them threw is raised after the last.
	 *
	 * @param array<int,mixed> $message The 7-field positional message array.
	 * @throws \Throwable Every option's failure, combined, raised after the last.
	 */
	public function fill( array $message ): void {
		++$this->counter;
		$type = Core::int( $message[ Message::TYPE ] );
		if ( ! ( $type & Message::TM_STRUCT ) ) {
			return;
		}
		$value = $message[ Message::VALUE ];
		if ( ! \is_array( $value ) || ! isset( $value['option'] ) ) {
			return;
		}
		$locals = $this->registered_with_followers( Core::as_string( $value['option'] ) );
		if ( [] === $locals || [] === $this->live_targets() ) {
			return;
		}
		Config::invalidate_options_cache();
		Config::forget_effective();
		$caught = Worker_Should_Stop::attempt_each( $locals, fn ( string $local ) => $this->push( $local ) );
		Worker_Should_Stop::raise( $caught );
	}

	/**
	 * Periodic re-push: emit one `set` command for EVERY registered option in a
	 * single tick, so the batching `HTTP_Out` downstream coalesces them into one
	 * POST per spoke. It replaces the base Timer heartbeat rather than adding to
	 * it — a spoke wants the settings, not a tick. One cache flush serves the
	 * whole sweep, and a sweep with nothing to push pays none. Every option is
	 * attempted, and what any of them threw is raised after the last.
	 *
	 * @throws \Throwable Every option's failure, combined, raised after the sweep.
	 */
	public function fire(): void {
		if ( [] === $this->registry || [] === $this->live_targets() ) {
			return;
		}
		Config::invalidate_options_cache();
		Config::forget_effective();
		$caught = Worker_Should_Stop::attempt_each(
			\array_keys( $this->registry ),
			fn ( string $local ) => $this->push( $local )
		);
		Worker_Should_Stop::raise( $caught );
	}

	/**
	 * Read one registered local option and fan its current value out, one `set`
	 * per mapping. The caller has already checked that the option is registered
	 * and a spoke is connected, and flushed the options cache. A value that
	 * will not encode sends nothing and throws, since an empty token would
	 * wipe the option at every spoke. Every mapping is attempted, and what any
	 * of them threw is raised after the last.
	 *
	 * @param string $local Registered local WP-option name.
	 * @throws \RuntimeException When the value will not encode.
	 * @throws \Throwable Every mapping's failure, combined, raised after the last.
	 */
	private function push( string $local ): void {
		// App-overridable: a plugin transforms its own options on the way out.
		$value  = \apply_filters( 'newspack_nodes/settings_sync/value', self::local_value( $local ), $local );
		$scalar = self::scalarize( $value )
			?? throw new \RuntimeException( \esc_html( "settings_sync: cannot encode value for {$local}" ) );
		// One `set` per mapping — a local may target several spoke options.
		$caught = Worker_Should_Stop::attempt_each(
			$this->registry[ $local ],
			fn ( array $spec ) => $this->send_signed( $spec['to'], 'set', [ $spec['remote'], $scalar ] )
		);
		Worker_Should_Stop::raise( $caught );
	}

	/**
	 * $local when it is registered, then every registered option whose
	 * substrate key follows $local's (`Schema::followers()`) and has no stored
	 * row: a stored follower did not move. A follower whose row another
	 * process just deleted pushes on its own delete event.
	 *
	 * @param string $local A changed WP-option name.
	 * @return list<string>
	 */
	private function registered_with_followers( string $local ): array {
		$locals = [ $local ];
		$key    = self::short_key_of( $local );
		if ( null !== $key ) {
			$prefix = Settings_Schema::get()->prefix();
			foreach ( Settings_Schema::get()->followers( $key ) as $follower ) {
				if ( false === \get_option( $prefix . $follower ) ) {
					$locals[] = $prefix . $follower;
				}
			}
		}
		return \array_values( \array_filter( $locals, fn ( string $l ): bool => isset( $this->registry[ $l ] ) ) );
	}

	/**
	 * The value to push for $local: its stored option, except that a
	 * substrate-declared option absent or blank pushes what a blank there
	 * yields (`Config::blank_value()`), since a spoke's typed `set` refuses ''.
	 * Any other option, which is not declared by any substrate Field, passes
	 * through.
	 *
	 * @param string $local Registered local WP-option name.
	 * @throws \RuntimeException When a blank declared option resolves to null, which encodes as nothing.
	 */
	private static function local_value( string $local ): mixed {
		$stored = \get_option( $local );
		$key    = self::short_key_of( $local );
		if ( null === $key || ( false !== $stored && '' !== $stored ) ) {
			return $stored;
		}
		return Config::blank_value( $key )
			?? throw new \RuntimeException( \esc_html( "settings_sync: cannot encode value for {$local}" ) );
	}

	/**
	 * The substrate config key $local stores, or null when $local is not a
	 * `newspack_nodes_` option a substrate Field declares.
	 *
	 * @param string $local A WP-option name.
	 */
	private static function short_key_of( string $local ): ?string {
		$schema = Settings_Schema::get();
		if ( ! \str_starts_with( $local, $schema->prefix() ) ) {
			return null;
		}
		$key = \substr( $local, \strlen( $schema->prefix() ) );
		return '' !== $key && null !== $schema->field_for_short( $key ) ? $key : null;
	}

	/**
	 * Flatten a value to one command token: an array encodes as JSON, a scalar
	 * stringifies. JSON rather than a join because an option's array KEYS are
	 * data — `custom_events` is `event_name => true` — and a join drops them.
	 *
	 * @param mixed $v Raw option value.
	 * @return string|null The token, or null when the value will not encode.
	 */
	private static function scalarize( mixed $v ): ?string {
		if ( \is_array( $v ) ) {
			$json = \wp_json_encode( $v, \JSON_UNESCAPED_SLASHES );
			return false === $json ? null : $json;
		}
		return Core::as_string( $v );
	}

	/**
	 * Register one mapping from a local option to a spoke option.
	 *
	 * Repeatable per local option: a hub `remote_*` setting is registered twice,
	 * once against the spoke's stripped option and once against the spoke's own
	 * `remote_*` copy. An exact duplicate is ignored, so re-running the topology
	 * does not double the fan-out.
	 *
	 * @param string $local  Local option name.
	 * @param string $to     Path the push is addressed TO.
	 * @param string $remote Option name on the spoke.
	 * @return string "ok\n".
	 */
	public function add_setting( string $local, string $to, string $remote ): string {
		$spec = [
			'to'     => $to,
			'remote' => $remote,
		];
		$this->registry[ $local ] ??= [];
		if ( ! \in_array( $spec, $this->registry[ $local ], true ) ) {
			$this->registry[ $local ][] = $spec;
		}
		return "ok\n";
	}

	/**
	 * Undo the `add_setting` that registered this mapping: neither an event
	 * nor the sweep pushes it again, and a local option left with no mapping
	 * leaves the registry, so its events are dropped before any cache flush.
	 *
	 * @param string $local  Local option name.
	 * @param string $to     Path the push is addressed TO.
	 * @param string $remote Option name on the spoke.
	 * @return string "ok\n".
	 * @throws \InvalidArgumentException When that mapping is not registered.
	 */
	public function remove_setting( string $local, string $to, string $remote ): string {
		$spec = [
			'to'     => $to,
			'remote' => $remote,
		];
		if ( ! \in_array( $spec, $this->registry[ $local ] ?? [], true ) ) {
			throw Command_Args::undeclared( 'remove_setting', [ $local, $to, $remote ], 'setting', 'node' );
		}
		$this->registry[ $local ] = Core::without( $this->registry[ $local ], $spec );
		if ( [] === $this->registry[ $local ] ) {
			unset( $this->registry[ $local ] );
		}
		return "ok\n";
	}

	/**
	 * The base graph snippet plus one `command_node {name}:config add_setting …`
	 * line per registry mapping, so a dump replays the whole mapping table.
	 *
	 * @return string Newline-terminated TSL lines.
	 */
	public function dump_config(): string {
		$out = parent::dump_config();
		foreach ( $this->registry as $local => $specs ) {
			foreach ( $specs as $spec ) {
				$out .= $this->config_line( 'add_setting', $local, $spec['to'], $spec['remote'] );
			}
		}
		return $out;
	}

	/**
	 * The `add_setting` command handler: register the mapping on the patron node.
	 *
	 * The command table links this method instead of carrying the body inline, so
	 * `node_schema()` stays a declaration.
	 *
	 * @param Command_Interpreter_Node $interpreter The sibling `:config` interpreter.
	 * @param array<array-key,mixed>   $args        Bound command arguments: local_option, to, remote_option.
	 * @return string Result line.
	 */
	public static function cmd_add_setting( Command_Interpreter_Node $interpreter, array $args ): string {
		/** @var self $patron */
		$patron = $interpreter->patron();
		return $patron->add_setting( Core::as_string( $args['local_option'] ), Core::as_string( $args['to'] ), Core::as_string( $args['remote_option'] ) );
	}

	/**
	 * The `remove_setting` command handler: drop the mapping from the patron node.
	 *
	 * @param Command_Interpreter_Node $interpreter The sibling `:config` interpreter.
	 * @param array<array-key,mixed>   $args        Bound command arguments: local_option, to, remote_option.
	 * @return string Result line.
	 * @throws \InvalidArgumentException What `remove_setting()` refused.
	 */
	public static function cmd_remove_setting( Command_Interpreter_Node $interpreter, array $args ): string {
		/** @var self $patron */
		$patron = $interpreter->patron();
		return $patron->remove_setting( Core::as_string( $args['local_option'] ), Core::as_string( $args['to'] ), Core::as_string( $args['remote_option'] ) );
	}

	/**
	 * Topology-console manifest: the palette entry, the constructor argument and
	 * the `add_setting` and `remove_setting` command forms, merged over Timer_Node's.
	 *
	 * @return array<string,mixed>
	 */
	public static function node_schema(): array {
		$setting_args = [
			[ 'name' => 'local_option',  'type' => 'string', 'required' => true ],
			[ 'name' => 'to',            'type' => 'string', 'required' => true ],
			[ 'name' => 'remote_option', 'type' => 'string', 'required' => true ],
		];
		return \array_merge( parent::node_schema(), [
			'category'    => 'Control',
			'description' => 'Pushes registered WP-option changes to connected spokes.',
			'arguments'   => [
				[ 'name' => 'interval_seconds', 'type' => 'int', 'default' => self::DEFAULT_INTERVAL_SECONDS, 'description' => 'Re-push sweep cadence in seconds — how often every registered option is re-sent to spokes (digits only; default 300, floored at 1).' ],
			],
			'commands'    => [
				[
					'name'        => 'add_setting',
					'description' => 'Register a local-option → spoke mapping: <local_option> <TO> <remote_option>.',
					'args'        => $setting_args,
					'handler'     => static fn ( Command_Interpreter_Node $interpreter, array $args ): string => self::cmd_add_setting( $interpreter, $args ),
					'multiple'    => true,
				],
				[
					'name'        => 'remove_setting',
					'description' => 'Undo one add_setting: <local_option> <TO> <remote_option>, as add_setting took them. Refused unless one matches.',
					'args'        => $setting_args,
					'handler'     => static fn ( Command_Interpreter_Node $interpreter, array $args ): string => self::cmd_remove_setting( $interpreter, $args ),
					'action'      => true,
				],
			],
			'has_target'  => true,
		] );
	}
}