@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
Copy

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();
Copy

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,
});
Copy

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

Subpath exports

Subpath	Description
@playlive/realtime-pipeline	Default barrel — re-exports every subpath below plus PACKAGE_NAME / KNOWN_URLS.
@playlive/realtime-pipeline/protocol	Wire-format types only (zero runtime code except CharityTypes).
@playlive/realtime-pipeline/reducer	Pure protect*Amounts + merge* helpers.
@playlive/realtime-pipeline/store	createPipelineStore + PipelineStore / PipelineStoreApi.
@playlive/realtime-pipeline/connection	createPipelineConnection, processMessage, identify builders, session, Emitter.
@playlive/realtime-pipeline/config	initializeDataLayer, getConfig, isInitialized, resetConfig.
@playlive/realtime-pipeline/demo	Demo fixtures + isDemoMode for offline overlay previews.

API reference

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

Connection (./connection)

Export	Kind	Notes
createPipelineConnection	function	(options: PipelineConnectionOptions) => PipelineConnection. Native-WebSocket state machine + identify lifecycle.
PipelineConnection	interface	Handle: connect, disconnect, send, identify, refresh, on, off, readyState, destroy.
PipelineConnectionOptions	interface	store (required) + url, webSocketCtor, maxReconnectAttempts (100), reconnectIntervalMs (1500), identifyDebounceMs (50), identifyDedupeMs (500), keepAliveIntervalMs (60 000), autoConnect (false), onCommandReload, log.
PipelineConnectionEvents	interface	Event map: open, close, error, reconnect-stop, message, identify.
WebSocketCtor / WebSocketLike	interface	Minimal DOM-free WebSocket shapes, for injecting ws or a mock.
processMessage	function	(store, rawMessage: string, options?: ProcessMessageOptions) => void. Pure dispatcher — never throws.
ProcessMessageOptions	interface	onCommandReload, initialDataDelayMs (250), log.
Emitter<Events>	class	Tiny typed event emitter (on returns an unsubscribe fn). No external dep.
buildIdentifyPayload	function	(store: PipelineStoreApi) => IdentifyPayload. Pure.
IdentifyPayload	interface	The action: "identify" frame — see Protocol below.
isIdentifyReady	function	(store) => boolean. `(hasSetOverlayInfo
buildRefreshPayload	function	(filters?: RefreshCommandFilters) => Record<string, unknown>. Pure.
stableEqual	function	JSON.stringify-based deep equality used to suppress spurious re-identifies.
resolveOverlaySession	function	() => OverlaySession. Per-tab identity stamped onto every identify.
resetOverlaySession	function	Drops the memoised identity. For tests.
OverlaySession	interface	{ id, label?, viewport? } — label comes from ?session=, capped at 24 chars.

Store (./store)

Export	Kind	Notes
createPipelineStore	function	(initialProps?: InitialProps) => PipelineStoreApi. Vanilla Zustand store factory.
PipelineStore	interface	Full state + action surface exposed via store.getState().
PipelineStoreApi	type	ReturnType<typeof createPipelineStore> — what every helper here accepts.

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)

Export	Signature
protectCampaignAmounts	(incoming: TiltifyCampaign[], existing: TiltifyCampaign[]) => TiltifyCampaign[]
protectTeamCampaignAmounts	(incoming: TiltifyTeamCampaign[], existing: TiltifyTeamCampaign[]) => TiltifyTeamCampaign[]
protectFundraisingEventAmounts	(incoming: TiltifyFundraisingEvent[], existing: TiltifyFundraisingEvent[]) => TiltifyFundraisingEvent[]
mergePolls	(existing, incoming, authoritative?: boolean) => Record<string, TiltifyPoll[]>
mergeRewards	same shape, TiltifyReward[]
mergeMilestones	same shape, TiltifyMilestone[]
mergeTargets	same shape, TiltifyTarget[]
mergeLeaderboardEntries	same shape, TiltifyLeaderboardEntry[]
mergeFundraisingEventSupportingCampaigns	(existing, incoming) => Record<string, TiltifyCampaign[]> — no authoritative flag; delegates to protectCampaignAmounts.

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)

Export	Kind	Notes
initializeDataLayer	function	(config: DataLayerConfig) => void. Idempotent; replaces any previous config.
getConfig	function	() => Required<DataLayerConfig>. Throws if called before initialisation.
isInitialized	function	() => boolean.
resetConfig	function	Resets to uninitialised. For tests.
DataLayerConfig	interface	tiltifyProxyUrl (required) + twitchServiceUrl, websocketUrl, causeId, defaultPollingInterval.
DEFAULT_CONFIG	const	Defaults applied over every override.
DEFAULT_WEBSOCKET_URL	const	"wss://main.playlive.ws.api.experience.stjude.org".

Demo (./demo)

Export	Kind	Notes
isDemoMode	function	(userOrTeamSlug, campaignSlug) => boolean. Two args, both nullable.
isDemoCampaignId	function	(id) => boolean — matches the two all-zero demo UUIDs.
DEMO_USER_SLUG / DEMO_TEAM_SLUG	const	"playliveDemoUser" / "playliveDemoTeam".
DEMO_CAMPAIGN_SLUG / DEMO_TEAM_CAMPAIGN_SLUG	const	Campaign slugs paired with the above.
DEMO_CAMPAIGN_ID / DEMO_TEAM_CAMPAIGN_ID / DEMO_FUNDRAISING_EVENT_ID / DEMO_CAUSE_ID	const	Stable fixture IDs.
DEMO_USER, DEMO_TEAM, DEMO_CAMPAIGN, DEMO_TEAM_CAMPAIGN, DEMO_FUNDRAISING_EVENT, DEMO_DONATIONS, DEMO_MILESTONES, DEMO_REWARDS, DEMO_POLL, DEMO_POLL_OPTIONS, DEMO_LEADERBOARD, DEMO_SCHEDULE	const	Fixture data.
getDemoCampaign(isTeam), getDemoUser(), getDemoTeam(), getDemoFundraisingEvent(), getDemoDonations(), getDemoMilestones(), getDemoRewards(), getDemoPoll(), getDemoLeaderboard(), getDemoSchedule(params?), getDemoScheduleBlockRaised(params?), getDemoDonorSpotlight(params)	function	Accessors returning fresh copies.
buildDemoSchedule, getDemoBlockRaised	function	Lower-level schedule builders (BuildDemoScheduleParams, GetDemoBlockRaisedParams).
DEFAULT_DEMO_SCHEDULE_INTERVAL_MS / MIN_DEMO_SCHEDULE_INTERVAL_MS	const	60_000 / 500.

Package metadata (./)

Export	Kind	Notes
PACKAGE_NAME	const	Identifier for runtime version-pinning.
KNOWN_URLS	const	Twitch Extension URL disclosure list.

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}}
Copy

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

Frame	Builder	Notes
{ action: "identify", … }	buildIdentifyPayload	IdentifyPayload: campaignIDs, fundraisingEventIDs, causeIDs, teamCampaignIDs, type, tiltify, overlayInfo (incl. session), subathonTimer, donationTrains, donorSpotlight, optional adminApiKey / twitchUsername.
{ action: "request", command: "refresh", … }	buildRefreshPayload	Optional campaignID / overlayName / overlayPath filters (RefreshCommandFilters). Admin-only; no-op while disconnected.

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.

type	Exported interface	Store effect
connection	ConnectionMessage	Normalised settings echo + connectionType + webhook subscriptions.
tick	TiltifyTickMessage	Bulk aggregate update (TiltifyTickMessageData); flips hasReceivedInitialData after 250 ms.
public:direct:donation_updated	DonationMessage	Appends to donations[campaignID] — or testDonations when sourceType === "test".
public:direct:fact_updated	CampaignMessage	Merges a campaign / team campaign. sourceType === "reset" opts into allowDecrease.
updateDonationTrain	DonationTrainUpdatedMessage	Upserts donationTrains[campaignID].
deleteDonationTrain	DonationTrainDeletedMessage	Removes the train.
updateSubathonTimer	SubathonTimerUpdatedMessage	Upserts subathonTimers[campaignID].
updateDonorSpotlight	DonorSpotlightUpdatedMessage	Replaces donorSpotlights[campaignID] (ComputedSpotlight).
updateDonorSpotlightSettings	DonorSpotlightSettingsUpdatedMessage	Replaces donorSpotlightSettings[campaignID].
updatePoll	UpdatePollMessage	Upserts one poll in polls[campaignID] + records lastPollDeltas[poll.id] (PollOptionDelta[]; the field is omitted on a poll's first push, and the store then leaves the prior deltas untouched).
updateLeaderboard	UpdateLeaderboardMessage	Replaces leaderboardEntries[campaignID] wholesale.
updateLeaderboardExclusions	UpdateLeaderboardExclusionsMessage	Replaces leaderboardExclusions[campaignID] wholesale.
updateSchedule	UpdateScheduleMessage	Replaces schedules[campaignID] with ScheduleTransitionBlock[].
request:refresh	RefreshResponseMessage	Stores lastRefreshResponse ({ refreshed: number }).
admin:webhook:firehose	AdminWebhookFirehoseMessage	Union of TiltifyFirehoseMessage | TwitchFirehoseMessage. Admin connections only.
twitch:chat:message	TwitchChatMessage	Appends to twitchChatMessages.
command:reload	(no interface — bare { type })	Invokes options.onCommandReload; defaults to window.location.reload() in a browser.

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"]
Copy

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();
}
Copy

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");
Copy

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
}
Copy

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();
Copy

License

MIT © St. Jude Children's Research Hospital