Realtime providers

A provider adapter connects Interact to an external delivery service. Choose an adapter once; application islands use the same subscriptions, publisher methods, and template directives with every provider.

Choose an adapter

Adapter	Setup
Included Mercure integration	Install the optional CakePHP package and configure a Mercure hub. See Mercure.
Application-owned WebSocket integration	Connect your gateway through PHP provider contracts and a browser transport. See Custom providers.

Mercure is optional. The custom adapter contracts can connect other realtime services too; Interact does not require their messages to arrive over SSE. Action streaming uses the action's own HTTP response and needs no realtime provider.

Mercure

Install the optional CakePHP 6 integration and run a Mercure 1.x hub:

composer require josbeir/cakephp-mercure:dev-cake6

Supply the provider configuration before loading Interact in application bootstrap. This example configures Mercure explicitly instead of loading its default config file:

use Cake\Core\Configure;
use Interact\InteractPlugin;
use Mercure\MercurePlugin;
use function Cake\Core\env;

Configure::write('Mercure', [
    'url' => env('MERCURE_URL'),
    'public_url' => env('MERCURE_PUBLIC_URL'),
    'jwt' => [
        'secret' => env('MERCURE_JWT_SECRET'),
        'issuer' => env('MERCURE_JWT_ISSUER'),
        'audience' => env('MERCURE_JWT_AUDIENCE'),
    ],
]);
Configure::write('Interact.realtime.provider', 'mercure');
$this->addPlugin(MercurePlugin::class, ['bootstrap' => false]);
$this->addPlugin(InteractPlugin::class);

Configure the hub with the same issuer, signing key, and resource identifier as the token audience. url is the server-side publication endpoint; public_url is the browser subscription endpoint. Different origins require hub CORS configuration allowing the application origin and bearer authorization. Allow the hub in the application's CSP connect-src when that directive is restricted.

The bundled browser adapter uses native fetch and subscriber-only bearer tokens. It leaves application Mercure cookies untouched. Provider credentials remain in memory, outside $server.state and persistent browser storage.

Custom providers

Implement two PHP contracts and register a browser transport. The following WebSocket example supplies all three parts; applications keep the ordinary realtime APIs.

Boundary	Contract	Responsibility
PHP publication	BroadcasterInterface::publish()	Deliver a Message, preserving its ID, topic, and privacy
PHP authorization	SubscriptionAuthorizerInterface::authorize()	Issue subscribe-only credentials for authorized topics until the supplied expiry
Browser connection	RealtimeTransport	Authenticate, receive messages, report status, reconnect, and clean up

PublisherInterface and its shortcuts are supplied by Interact. Registering a SubscriptionAuthorizerInterface enables lease-based authorization; a custom provider does not require the Mercure package or a provider-name configuration flag.

WebSocket example

This example connects an application-owned WebSocket gateway with an HTTP management API. It defines a protocol for that gateway; existing WebSocket services have their own APIs and need corresponding changes to the adapter. Interact does not start or implement the gateway server.

Gateway protocol

The gateway exposes these management endpoints to PHP. Both require a server-only management key in Authorization: Bearer …:

Endpoint	JSON request	Required behavior
POST /api/subscriptions	{ "topics": ["articles/1"], "expires": 1800000000 }	Return { "token": "opaque-subscription-token" } with exact subscribe-only permissions and the requested expiry
POST /api/publish	{ "private": true, "message": { … } }	Deliver the normalized message to authorized subscribers and acknowledge with a 2xx response

The browser connects to wss://gateway.example.test/socket and sends:

{
  "type": "subscribe",
  "token": "opaque-subscription-token",
  "topics": ["articles/1"]
}

The gateway must validate the token, restrict requested topics to its exact permissions, and enforce its expiry for the lifetime of the connection. Only after authorization succeeds does it send { "type": "subscribed" } and start delivery. It must prevent unauthenticated publication and unauthorized private-topic access. The management key must never be accepted as a browser subscription token.

For a notification, forward the envelope returned by Message::encode($topic):

{
  "version": 1,
  "id": "message-42",
  "topic": "articles/1",
  "event": "article-updated",
  "detail": { "id": 1 }
}

Text and HTML messages also contain display; forward that field unchanged. Preserve JSON records such as an empty detail: {}, message IDs, and topic names. Use the gateway's own SDK or HTTP API when adapting this example to another server.

PHP adapter

Create src/Realtime/WebSocketProvider.php in your application. The tested source below uses the fixture namespace; replace it with your application's namespace, for example App\Realtime.

<?php
declare(strict_types=1);

namespace Interact\Test\Fixture\Realtime\Providers;

use Cake\Http\Client;
use Cake\Http\Client\Response;
use Interact\Realtime\BroadcasterInterface;
use Interact\Realtime\Grant;
use Interact\Realtime\Message;
use Interact\Realtime\SubscriptionAuthorizerInterface;
use RuntimeException;
use SensitiveParameter;

/**
 * Bridges an application-owned WebSocket gateway to Interact's provider contracts.
 *
 * The gateway's HTTP API owns subscription tokens and socket delivery. Only its
 * subscriber token reaches the browser; its management API key stays in PHP.
 */
class WebSocketProvider implements BroadcasterInterface, SubscriptionAuthorizerInterface
{
    /**
     * Inject CakePHP HTTP behavior and application-supplied gateway configuration.
     */
    public function __construct(
        protected readonly Client $http,
        protected readonly string $apiUrl,
        protected readonly string $socketUrl,
        #[SensitiveParameter]
        protected readonly string $apiKey,
    ) {}

    /**
     * Forward the normalized envelope and delivery privacy through the management API.
     */
    public function publish(string $topic, Message $message, bool $private = true): void
    {
        $this->post('/publish', [
            'private' => $private,
            'message' => json_decode($message->encode($topic), flags: JSON_THROW_ON_ERROR),
        ]);
    }

    /**
     * Request an expiring subscribe-only token for the already authorized topic set.
     *
     * @param list<string> $topics Exact topics selected by authorized island declarations.
     */
    public function authorize(array $topics, int $expires): Grant
    {
        $response = $this->post('/subscriptions', ['topics' => $topics, 'expires' => $expires]);
        /** @var mixed $data */
        $data = $response->getJson();
        if (!is_array($data) || !is_string($data['token'] ?? null) || $data['token'] === '') {
            throw new RuntimeException('The WebSocket gateway returned an invalid subscription token.');
        }

        return new Grant(
            'websocket',
            [
                'url' => $this->socketUrl,
                'token' => $data['token'],
                'topics' => $topics,
            ],
            $expires,
        );
    }

    /**
     * Make one authenticated management request without redirects or publication retries.
     *
     * @param array<string, mixed> $data Gateway request data, never browser-chosen permissions.
     */
    protected function post(string $path, array $data): Response
    {
        $response = $this->http->post(
            rtrim($this->apiUrl, '/') . $path,
            json_encode($data, JSON_THROW_ON_ERROR),
            [
                'type' => 'json',
                'headers' => ['Authorization' => 'Bearer ' . $this->apiKey],
                'redirect' => false,
                'timeout' => 5,
            ],
        );
        if ($response->getStatusCode() < 200 || $response->getStatusCode() >= 300) {
            throw new RuntimeException('The WebSocket gateway request failed.');
        }

        return $response;
    }
}

authorize() receives topics already selected by fresh, authorized islands. The browser cannot add permissions to that set. The gateway issues the opaque token; PHP passes only the socket URL, subscriber token, and authorized topics in the Grant. Its websocket driver name must match the browser registration.

The adapter makes one management request per operation and throws on a non-2xx response or an invalid token. Publishing uses the shared Message envelope, so notification, refresh, text, and HTML modes all use the same provider.

Register container services

For an application using the App namespace, add these bindings to Application::services(). The gateway URLs and management key are supplied by your environment. Load Interact through its normal plugin lifecycle.

use App\Realtime\WebSocketProvider;
use Cake\Container\ContainerInterface;
use Cake\Http\Client;
use Interact\Realtime\BroadcasterInterface;
use Interact\Realtime\SubscriptionAuthorizerInterface;
use function Cake\Core\env;

public function services(ContainerInterface $container): void
{
    $container->addShared(WebSocketProvider::class, static fn(): WebSocketProvider =>
        new WebSocketProvider(
            new Client(),
            (string)env('WEBSOCKET_API_URL'),
            (string)env('WEBSOCKET_URL'),
            (string)env('WEBSOCKET_API_KEY'),
        )
    );
    foreach ([BroadcasterInterface::class, SubscriptionAuthorizerInterface::class] as $contract) {
        $container->addShared($contract, static fn(): WebSocketProvider =>
            $container->get(WebSocketProvider::class)
        );
    }
}

Example URLs are WEBSOCKET_API_URL=https://gateway.example.test/api and WEBSOCKET_URL=wss://gateway.example.test/socket. The management key belongs in server configuration. Application code continues to inject Interact\Realtime\PublisherInterface and call publishEvent(), publishText(), or publishHtml(); it does not construct this adapter on each call.

Browser transport

Save this external script as webroot/js/websocket-feed.js and load it before Interact starts. It registers the websocket transport during the synchronous interact:configure event. This startup event works with both the Composer bundle and an application-owned Alpine installation.

// Application-owned protocol: subscribe with a token, receive an acknowledgement,
// then receive normalized Interact PushMessage envelopes from the authorized server.
document.addEventListener('interact:configure', (event) => {
  event.detail.registerTransport('websocket', (grant, callbacks) => {
    const url = new window.URL(grant.data.url);
    if (!['ws:', 'wss:'].includes(url.protocol)) {
      throw new Error('Invalid WebSocket provider URL.');
    }
    let socket;
    let cleanup;
    let timer;
    let stopped = false;
    let failures = 0;

    const connect = () => {
      if (stopped) return;
      socket = new window.WebSocket(url);
      const current = socket;
      const listeners = new window.AbortController();
      cleanup = () => {
        listeners.abort();
        current.close();
      };
      let authorized = false;
      current.addEventListener(
        'open',
        () => {
          current.send(
            JSON.stringify({
              type: 'subscribe',
              token: grant.data.token,
              topics: grant.data.topics,
            }),
          );
        },
        { signal: listeners.signal },
      );
      current.addEventListener(
        'message',
        (message) => {
          if (stopped || current !== socket) return;
          try {
            const value = JSON.parse(message.data);
            if (value.type === 'subscribed') {
              authorized = true;
              failures = 0;
              callbacks.status('connected');
            } else if (authorized) {
              callbacks.receive(value);
            } else {
              throw new Error('Subscription acknowledgement required.');
            }
          } catch {
            stopped = true;
            callbacks.status('error', 'WebSocket message rejected.');
            cleanup();
          }
        },
        { signal: listeners.signal },
      );
      current.addEventListener(
        'close',
        () => {
          listeners.abort();
          if (stopped) return;
          callbacks.status('reconnecting');
          timer = window.setTimeout(
            connect,
            Math.min(30000, 500 * 2 ** Math.min(failures++, 6)) *
              (0.75 + Math.random() * 0.5),
          );
        },
        { signal: listeners.signal },
      );
    };

    connect();
    return () => {
      stopped = true;
      window.clearTimeout(timer);
      cleanup?.();
    };
  });
});

The adapter authenticates in the first socket message. It reports connected only after the gateway acknowledges authorization, then passes incoming envelopes to callbacks.receive(). Interact validates the envelopes and routes them by exact topic and event. Invalid input or a missing acknowledgement ends that connection; closures retry with backoff. Cleanup cancels retries, removes listeners, and closes the socket.

For application-owned modules, the equivalent registration is createInteract({ realtime: { transports: { websocket: transport } } }), where transport is your factory. The npm package exports RealtimeTransport, RealtimeGrant, TransportCallbacks, and PushMessage for typed integrations.

Allow the gateway's wss: origin in CSP connect-src when your application restricts that directive. Provider credentials stay in memory, outside island state and browser storage.

Lifecycle and testing

The coordinator batches island leases and shares connections across topics in the same tab and authorization scope, with up to 64 islands and 1 MiB per authorization request. It renews grants before expiry and replaces the transport when permissions change. Reauthorization failures and grant expiry end access; immediate permission revocation remains provider-specific.

Transports own provider reconnect and cursor rules. A connected status triggers refresh subscriptions to reconcile current application data. Notifications and display messages are best effort; durable history and application publication retries belong to your storage or outbox workflow.

The example's PHP tests verify management authentication, exact-topic grants, container bindings, message encoding, invalid responses, and failed publication without retries. Its browser transport tests use a controlled WebSocket boundary for acknowledgement, delivery, reconnects, and cleanup. Test your actual gateway with two independent identities, forbidden topics, expired tokens, and disconnects.

Direct display adapters

For managed subscriptions, use a provider adapter above: Interact handles island authorization, shared connections, and grant renewal. A direct display adapter is an advanced alternative when the application owns a separate feed and its entire connection lifecycle.

An adapter connects an application-owned feed, such as Mercure or a WebSocket, to an island's display targets. Map your provider's messages to $server.receiveStream(); the normal action POST keeps its built-in SSE reader.

For example, suppose an authorized SSE feed sends {"text":"Next result\n"}. Register this external script before Interact starts:

// webroot/js/text-feed.js; register before Interact's runtime starts.
document.addEventListener('alpine:init', () => {
  window.Alpine.data('textFeed', () => {
    let feed;
    return {
      init() {
        const id = this.$el.closest('[data-interact-id]').dataset.interactId;
        feed = new window.EventSource(this.$el.dataset.feedUrl);
        feed.onmessage = (event) => {
          try {
            this.$server.receiveStream({
              version: 1,
              id,
              type: 'target',
              to: 'output',
              content: JSON.parse(event.data).text,
              replace: false,
              html: false,
            });
          } catch (error) {
            window.console.error('Rejected feed message', error);
          }
        };
      },
      destroy() {
        feed?.close();
      },
    };
  });
});

Inside the receiving island, supply an authorized subscription URL as $feedUrl through viewData() and attach the adapter:

<?php
declare(strict_types=1);
use function Cake\Core\h;
/** @var string $feedUrl */
?>
<div x-data="textFeed" data-feed-url="<?= (string)h($feedUrl) ?>">
    <pre x-interact:stream="output"></pre>
</div>

Each message appends literal text to this island's output target immediately. The adapter supplies the current island ID, catches malformed messages, and closes its connection when Alpine destroys the wrapper. The connection stays outside reactive state. Use your provider's client and cleanup method in the same lifecycle; authentication, authorized topics, and reconnection belong to that integration.

The receiver also accepts replace: true, HTML for explicit .html targets, and type: 'event' messages with name and JSON record detail. Event messages reach x-interact:on:event listeners in the island. The raw interact:stream event remains available to advanced observers. The npm package exports StreamMessage and IslandBridge for typed adapters; see the message contract.

These are provisional display updates. Signed state, validation, request activity, and action results stay with normal requests; a subsequent render is authoritative. Invalid input throws to the adapter, while delivery after island disposal is ignored.