@playlive/realtime-pipeline

Framework-agnostic native-WebSocket client + vanilla Zustand store for the Play Live unified data pipeline (UDP). Owns the wire format every Play Live overlay speaks: identify → connection echo → tick + incremental push messages. Ported from playlive-overlay-data-layer with the React glue stripped out and react-use-websocket-lite replaced by a built-in reconnect state machine.

Install

bun add @playlive/realtime-pipeline zustand

zustand (^5.0.0) is the only peer dependency — jose-style, the consumer brings their own copy, and it must be a real runtime dependency, not a dev one. @playlive/tiltify-core is a regular dependency (Tiltify wire types only) and installs transitively; nothing extra to do.

Native WebSocket + native fetch only. Runs in the browser, Bun, Node ≥ 22, and any framework. For React, use @playlive/react-pipeline instead of wiring this by hand.

Quick start

Vanilla — Node / Bun / browser script

import {
  createPipelineStore,
  createPipelineConnection,
} from "@playlive/realtime-pipeline";

const store = createPipelineStore({
  initialCampaignIDs: ["abc-123"],
  initialOverlayName: "donation-bar",
  initialOverlayPath: "/overlays/donation-bar",
  initialTeamUserSlug: "@playliver",
});

// Client connections don't identify until overlay info is set
// (`isIdentifyReady`); admin connections identify on an admin key alone.
store.getState().setOverlayInfo({
  name: "donation-bar",
  teamUserSlug: "@playliver",
  config: {},
  path: "/overlays/donation-bar",
});

const conn = createPipelineConnection({
  store,
  url: "wss://main.playlive.ws.api.experience.stjude.org",
  autoConnect: true,
});

conn.on("open", () => console.log("pipeline connected"));
conn.on("identify", (payload) => console.log("SENT identify:", payload));
conn.on("message", (raw) => console.log("RECV:", raw));
conn.on("close", (info) => console.log("disconnected", info));
conn.on("reconnect-stop", (attempts) => console.warn(`gave up after ${attempts}`));

store.subscribe((s) => {
  console.log(`raised: ${s.campaigns[0]?.amount_raised?.value}`);
});

// Later, on teardown — removes listeners, timers, and the store subscription.
conn.destroy();

If url is omitted the connection uses store.getState().wsURL, which defaults to the websocketUrl from initializeDataLayer() when the config singleton has been initialised, and to DEFAULT_WEBSOCKET_URL otherwise.

Older Node without a global WebSocket

import WebSocketCtor from "ws";
import { createPipelineConnection } from "@playlive/realtime-pipeline/connection";

const conn = createPipelineConnection({
  store,
  webSocketCtor: WebSocketCtor as unknown as typeof WebSocket,
});

createPipelineConnection throws immediately when no constructor is available and none was supplied.

Subpath exports

API reference

Full generated API documentation: https://packages.playlive.experience.stjude.org/p/@playlive/realtime-pipeline/docs/

Connection (./connection)

Store (./store)

State slices worth knowing: campaigns, teamCampaigns, fundraisingEvents, causes, donations / testDonations, polls, rewards, milestones, targets, leaderboardEntries, leaderboardExclusions, donationTrains, subathonTimers, donorSpotlights, donorSpotlightSettings, schedules, lastScheduleTransitions, lastPollDeltas (keyed by poll ID, not campaign ID), auctionHouses, twitchChatMessages, plus connected / connecting / hasReceivedInitialData.

Fresh-read getters (getDonations, getPolls, getRewards, getMilestones, getTargets, getLeaderboardEntries, getLeaderboardExclusions, getDonationTrains, getSubathonTimers, getDonorSpotlight, getSchedule, getLastScheduleTransition, getFirstCampaign, …) all take a campaign ID and never return undefined. Actions prefixed _ are wired by createPipelineConnection and are not part of the consumer contract.

Reducers (./reducer)

The protect* helpers guarantee amount_raised / total_amount_raised never decrease when a stale snapshot lands after a newer one. The merge* helpers accumulate by row id across ticks; pass authoritative: true (driven by tick.data.authoritative) to replace the named campaigns wholesale so rows deleted upstream actually disappear.

Config (./config)

Demo (./demo)

Package metadata (./)

Upstream spec

The "upstream spec" for this package is the Play Live realtime WebSocket wire protocol itself — there is no external OpenAPI document to drift against. The tables and diagram in this section are the authoritative public description of that protocol: every frame the client sends, every message the server can push, and the store effect each one has.

@playlive/realtime-pipeline/protocol is a types-only subpath, deliberately split out so both ends of the socket — overlays and the pipeline server — can import the same wire-format declarations. Payload bodies reuse the Tiltify v5 shapes from @playlive/tiltify-core.

Unknown type values are dropped rather than treated as errors, so the server can roll out new message types ahead of client upgrades.

Connect / identify / dispatch flow

sequenceDiagram
    participant App as Overlay
    participant Conn as createPipelineConnection
    participant Store as createPipelineStore
    participant UDP as Pipeline server

    App->>Store: setOverlayInfo / addCampaignID
    App->>Conn: connect()
    Conn->>UDP: WebSocket open
    UDP-->>Conn: onopen
    Conn->>Store: _setConnectionState(true, false)
    Conn->>UDP: {"action":"identify", campaignIDs, overlayInfo, ...}
    UDP-->>Conn: {"type":"connection", connection, subscriptions}
    Conn->>Store: _updateSettingsFromServer / _setLastIdentityMessage
    UDP-->>Conn: {"type":"tick", data:{...}}
    Conn->>Store: _updateTickData -> hasReceivedInitialData after 250ms
    UDP-->>Conn: {"type":"updatePoll"|"updateLeaderboard"|...}
    Conn->>Store: per-message action
    App->>Conn: refresh({campaignID})
    Conn->>UDP: {"action":"request","command":"refresh",...}
    UDP-->>Conn: {"type":"request:refresh", data:{refreshed}}

Any state change that affects the identify payload (campaign IDs, overlay info, tiltify settings, admin key, feature toggles, twitch username) re-triggers a debounced identify automatically via a store subscription. A 60 s keep-alive re-identify runs on top of that. Identical payloads sent within 500 ms are suppressed.

Client → server frames

Server → client messages

Every typed message extends WebSocketMessage ({ type, id, sourceType }). This is the complete set processMessage dispatches — anything else is silently dropped, which is what makes new message types safe to roll out ahead of client upgrades.

WebSocketErrorMessage ({ message: "Internal server error", connectionId, requestId }) has no type field — processMessage logs it via options.log and drops it.

Supporting payload types exported from the same module: CharityTypes / CharityType, TiltifyWebSocketSettings, TiltifyWebhookSubscription, WebSocketConnection / TiltifyWebSocketConnection / TwitchWebSocketConnection, ConnectionMessageSubscriptions, DonationTrain

• DonationTrainRuleset / PartialDonationTrainRuleset / TrainDonation / DonationTrainStatus, SubathonTimer + SubathonTimerDonation / SubathonTimerEffect / SubathonTimerStatus, the donor-spotlight family (ComputedSpotlight, DonorOfTheHour, BiggestDonation, CommunityHero, DonorSpotlightSettings, PartialDonorSpotlightSettings, DonorSpotlightDonation and their Raw* counterparts), PollOptionDelta, LeaderboardExclusion, AuctionHouse, TiltifyDonationWithTestFlag, TiltifyAuthoritativeSlice, TiltifyTickMessageData, ScheduleTransitionBlock, ScheduleTransitionPayload, RefreshCommandFilters, InitialProps, and OverlayInfo.

Twitch Extension URL disclosure

The KNOWN_URLS export enumerates every absolute URL or host this package can connect to, ready to paste into an Extension's manifest URL allowlist.

import { KNOWN_URLS } from "@playlive/realtime-pipeline";
console.log(KNOWN_URLS);
// ["wss://main.playlive.ws.api.experience.stjude.org"]

If your consumer overrides the WebSocket URL via initializeDataLayer({ websocketUrl: "wss://other.example/ws" }) — or via createPipelineConnection({ url }) — add the override to your own Extension URL disclosure too: the auditor walks KNOWN_URLS from every dep, but it can't see runtime overrides.

Migration from playlive-overlay-data-layer

@playlive/realtime-pipeline is a one-for-one replacement for the framework-agnostic half of playlive-overlay-data-layer. The React-specific parts (useUnifiedDataPipeline, UnifiedDataPipelineProvider, useWebSocketManager, the useAddPipeline* family) live in @playlive/react-pipeline. Every type, every reducer, and the entire store action surface are preserved verbatim — only:

• createUDPStore → createPipelineStore
• UDPStore interface → PipelineStore
• UDPStoreApi → PipelineStoreApi
• react-use-websocket-lite → built-in createPipelineConnection state machine (same defaults: 100 attempts, 1500 ms base, jittered).

Every other symbol keeps its name, so most migrations are a package rename plus those four renames.

Examples

Donation-train overlay, end to end

Connect, wait for the first tick, then render the active train and react to every subsequent push. This is the full shape of a real overlay minus the DOM.

import {
  createPipelineConnection,
  createPipelineStore,
  initializeDataLayer,
  type DonationTrain,
  type PipelineStoreApi,
} from "@playlive/realtime-pipeline";

const CAMPAIGN_ID = "abc-123";

initializeDataLayer({
  tiltifyProxyUrl: "https://tiltify-proxy.prod.experience.stjude.org",
  websocketUrl: "wss://main.playlive.ws.api.experience.stjude.org",
});

const store: PipelineStoreApi = createPipelineStore({
  initialCampaignIDs: [CAMPAIGN_ID],
  initialOverlayName: "donation-train",
  initialOverlayPath: "/overlays/donation-train",
  initialTeamUserSlug: "@playliver",
  initialDonationTrainsEnabled: true,
});

store.getState().setOverlayInfo({
  name: "donation-train",
  teamUserSlug: "@playliver",
  config: { theme: "dark" },
  path: "/overlays/donation-train",
});

// Opt into the server-side slices this overlay actually needs. The second
// argument re-sends the settings to the server on the next identify.
store.getState().updateTiltifySettings({ milestones: true, polls: false }, true);

const conn = createPipelineConnection({
  store,
  autoConnect: true, // url falls back to store.wsURL (from initializeDataLayer)
  log: (msg, payload) => console.debug("[pipeline]", msg, payload),
});

conn.on("error", (event) => console.error("socket error", event));
conn.on("reconnect-stop", (attempts) =>
  console.error(`pipeline unreachable after ${attempts} attempts`),
);

function render(train: DonationTrain | undefined) {
  if (!train || !train.trainVisible) return;
  console.log(
    `${train.trainStatus}: ${train.donationCount} donations, ${train.donationValue}`,
  );
}

let lastTrainId: string | null = null;
store.subscribe((s) => {
  if (!s.hasReceivedInitialData) return;

  const [train] = s.getDonationTrains(CAMPAIGN_ID, "ACTIVE");
  if (train && train.id !== lastTrainId) {
    lastTrainId = train.id;
    console.log("new train started", train.trainStart);
  }
  render(train);
});

// Operator-triggered resync (admin connections only — no-op otherwise).
export function forceResync() {
  conn.refresh({ campaignID: CAMPAIGN_ID });
}

export function teardown() {
  conn.destroy();
}

Driving the store without a socket

processMessage is the same pure dispatcher the connection uses, so tests, replays, and simulators can feed the store directly:

import { createPipelineStore } from "@playlive/realtime-pipeline/store";
import { processMessage } from "@playlive/realtime-pipeline/connection";

const store = createPipelineStore({ initialCampaignIDs: ["c1"] });

processMessage(
  store,
  JSON.stringify({
    type: "public:direct:donation_updated",
    id: "msg-1",
    sourceType: "live",
    data: {
      id: "donation-1",
      campaign_id: "c1",
      fundraising_event_id: null,
      donor_name: "Viewer1",
      amount: { value: "25.00", currency: "USD" },
      completed_at: "2025-01-01T00:00:00Z",
    },
  }),
);

console.log(store.getState().getDonations("c1").length); // 1

// Unknown types are dropped, invalid JSON is dropped — never throws.
processMessage(store, '{"type":"someFutureMessage"}');
processMessage(store, "not json");

Offline demo mode

Overlays render demo content when the URL carries the demo slugs, with no network access at all:

import {
  getDemoCampaign,
  getDemoDonations,
  isDemoMode,
} from "@playlive/realtime-pipeline/demo";

// /overlays/donation-bar/tiltify/@playliveDemoUser/playliveDemoCampaign
const [userOrTeamSlug, campaignSlug] = ["playliveDemoUser", "playliveDemoCampaign"];

if (isDemoMode(userOrTeamSlug, campaignSlug)) {
  const campaign = getDemoCampaign(false); // true → team campaign
  const donations = getDemoDonations();
  console.log(campaign.name, donations.length);
} else {
  // …boot the real pipeline connection
}

The demo slugs are exported as DEMO_USER_SLUG / DEMO_TEAM_SLUG and DEMO_CAMPAIGN_SLUG / DEMO_TEAM_CAMPAIGN_SLUG; isDemoCampaignId covers the matching all-zero fixture UUIDs.

Injecting a mock socket in tests

WebSocketLike / WebSocketCtor are exported precisely so tests never touch the network — the package's own suite drives the state machine this way:

import {
  createPipelineConnection,
  createPipelineStore,
  type WebSocketCtor,
  type WebSocketLike,
} from "@playlive/realtime-pipeline";

class MockSocket implements WebSocketLike {
  static instances: MockSocket[] = [];
  static readonly CONNECTING = 0;
  static readonly OPEN = 1;
  static readonly CLOSING = 2;
  static readonly CLOSED = 3;

  readyState = 0;
  onopen: ((ev: unknown) => void) | null = null;
  onclose: ((ev: unknown) => void) | null = null;
  onerror: ((ev: unknown) => void) | null = null;
  onmessage: ((ev: { data: unknown }) => void) | null = null;
  sent: string[] = [];

  constructor(public readonly url: string) {
    MockSocket.instances.push(this);
  }
  send(data: string) {
    this.sent.push(data);
  }
  close() {
    this.readyState = 3;
    this.onclose?.({ code: 1000, reason: "", wasClean: true });
  }
}

const store = createPipelineStore({
  initialURL: "ws://test.local",
  initialCampaignIDs: ["c1"],
});
store.getState().setOverlayInfo({
  name: "donation-bar",
  teamUserSlug: "@user",
  config: {},
  path: "/overlays/donation-bar",
});

const conn = createPipelineConnection({
  store,
  webSocketCtor: MockSocket as unknown as WebSocketCtor,
  autoConnect: true,
});

const socket = MockSocket.instances[0]!;
socket.readyState = 1;
socket.onopen?.({});

const identify = JSON.parse(socket.sent[0]!);
console.log(identify.action, identify.campaignIDs); // "identify" ["c1"]

conn.destroy();

License

MIT © St. Jude Children's Research Hospital