Client example

/**
 * A zero-dependency client for the Predict WebSocket API.
 * Designed to be used in the browser.
 */

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

function neOf<T>(item: T): NonEmptyArray<T> {
  return [item] as NonEmptyArray<T>;
}

function isNonEmpty<T>(items: T[]): items is NonEmptyArray<T> {
  return items.length > 0;
}

/*
 * Realtime Client
 */

export class RealtimeClient {
  constructor(
    private ws: WebSocket,
    private readonly options: {
      maxConnAttempts: number;
      maxRetryInterval: number;
    },
  ) {
    this.bindAll();
  }

  /** Topic of each in-flight subscribe or unsubscribe request, cleared on its response. */
  private topicSubReqIdMap: Map<number, TopicName> = new Map();

  /** Active topics this connection is subscribed to. */
  private subscriptions: Map<TopicName, NonEmptyArray<EventCallback>> = new Map();

  private connectionAttempts = 0;
  private requestId = 0;
  /** Set by close(), so the client doesn't reconnect after a close it asked for. */
  private closedByUser = false;
  /** Set by a reconnect notice, so the next connection opens at once. */
  private reconnectNow = false;

  private bindAll(): void {
    this.ws.onerror = this.onError;
    this.ws.onmessage = this.onMessage;
    this.ws.onclose = this.onClose;
    this.ws.onopen = this.onOpen;
  }

  private async reconnect() {
    // a reconnect notice moves to a new connection at once, and isn't a failed attempt.
    if (this.reconnectNow) {
      this.reconnectNow = false;
      return this.open();
    }

    const attempt = this.connectionAttempts++;
    if (attempt < this.options.maxConnAttempts) {
      // reconnect at once the first time. while reconnecting keeps failing, wait
      // 1, 2, 4 and 8 seconds and so on, up to maxRetryInterval. up to half a
      // second of jitter keeps clients from all reconnecting at the same moment.
      const wait =
        attempt === 0 ? 0 : Math.min(1000 * 2 ** (attempt - 1), this.options.maxRetryInterval);
      await sleep(wait + Math.random() * 500);
      this.open();
    } else {
      const allHandlers = Array.from(this.subscriptions.values()).flat();
      for (const handler of allHandlers) {
        handler({
          err: { code: 'ws_disconnected', message: 'Max connection attempts reached' },
        });
      }
    }
  }

  /** Opens a new connection to the same URL, unless close() was called. */
  private open() {
    if (this.closedByUser) return;
    this.ws = new WebSocket(
      this.ws.url,
      this.ws.protocol !== '' ? this.ws.protocol : undefined,
    );
    this.bindAll();
  }

  private onOpen = () => {
    this.connectionAttempts = 0;
    // re-subscribe every active topic after a reconnect.
    for (const topic of this.subscriptions.keys()) {
      this.subUnsub('subscribe', topic);
    }
  };

  private onClose = () => {
    // reconnect after every close the client didn't ask for, clean or not. the
    // server's own closes, such as 1001 when it shuts down, are clean ones.
    if (!this.closedByUser) {
      return this.reconnect();
    }
  };

  private onMessage = (event: MessageEvent) => {
    const parsed = JSON.parse(event.data as unknown as string) as Response;

    if (parsed.type === 'M') {
      const topic = parsed.topic;

      if (topic === 'heartbeat') {
        // echo the server timestamp back exactly.
        this.send({ method: 'heartbeat', data: parsed.data });
      } else if (parsed.topic === 'system') {
        if (parsed.data.event === 'reconnect') {
          // the server is shutting down, so move to a new connection right away.
          this.reconnectNow = true;
          this.ws.close();
        }
      } else {
        const handlers = this.subscriptions.get(parsed.topic);
        for (const handler of handlers || []) {
          handler({ data: parsed.data });
        }
      }
    } else if (parsed.type === 'R') {
      const requestIdTopic = this.topicSubReqIdMap.get(parsed.requestId);

      if (requestIdTopic) {
        if (!parsed.success) {
          const handlers = this.subscriptions.get(requestIdTopic);
          if (handlers) {
            for (const handler of handlers) {
              handler({ err: parsed.error });
            }
          }
          this.subscriptions.delete(requestIdTopic);
        }
        this.topicSubReqIdMap.delete(parsed.requestId);
      } else {
        console.warn(`Unknown response received`, JSON.stringify(parsed, null, 2));
      }
    }
  };

  private onError = (event: Event) => {
    const allHandlers = Array.from(this.subscriptions.values()).flat();
    for (const handler of allHandlers) {
      handler({ err: { code: 'ws_error_terminated', message: 'Websocket error' } });
    }
    console.error('RealtimeClientSocketError', event);
  };

  private send(data: Requests) {
    if (this.ws.readyState === this.ws.OPEN) {
      if (data.method !== 'heartbeat') {
        this.topicSubReqIdMap.set(data.requestId, data.params[0]);
      }
      return this.ws.send(JSON.stringify(data));
    } else {
      console.warn(`Not connected to WS. Ignoring.`);
    }
  }

  private subUnsub(method: 'subscribe' | 'unsubscribe', topic: string) {
    this.send({ requestId: this.requestId++, method, params: neOf(topic) });
  }

  private getTopicStringFor(channel: Channel): string {
    switch (channel.name) {
      case 'predictOrderbook':
        return `predictOrderbook/${channel.marketId}`;
      case 'predictTradingStatus':
        return `predictTradingStatus/${channel.marketId}`;
      case 'predictMarketStatus':
        return `predictMarketStatus/${channel.marketId}`;
      case 'predictMarketChanged':
        return `predictMarketChanged/${channel.marketId}`;
      case 'predictCategoryChanged':
        return `predictCategoryChanged/${channel.categoryId}`;
      case 'predictWalletEvents':
        return `predictWalletEvents/${channel.jwt}`;
    }
  }

  subscribe(channel: Channel, callback: EventCallback): { unsubscribe: () => void } {
    const topicName = this.getTopicStringFor(channel);
    const existing = this.subscriptions.get(topicName);

    if (existing) {
      existing.push(callback);
    } else {
      this.subscriptions.set(topicName, [callback]);
      this.subUnsub('subscribe', topicName);
    }

    return {
      unsubscribe: (): void => {
        const item = this.subscriptions.get(topicName);
        if (!item) {
          throw new Error('InconsistentState: No subscriptions for this topic');
        }
        const cbRemoved = item.filter((x) => x !== callback);
        if (isNonEmpty(cbRemoved)) {
          this.subscriptions.set(topicName, cbRemoved);
        } else {
          this.subscriptions.delete(topicName);
          this.subUnsub('unsubscribe', topicName);
        }
      },
    };
  }

  close(): void {
    this.closedByUser = true;
    this.ws.close();
  }
}

/*
 * Types
 */

export type NonEmptyArray<T> = [T, ...T[]];
export type Pretty<T> = { [K in keyof T]: T[K] } extends infer U ? U : never;

/* Payloads */

export type MarketTradingStatusUpdate = {
  kind: 'tradingStatusUpdate';
  tsMs: number;
  marketId: number;
  tradingStatus: 'OPEN' | 'MATCHING_NOT_ENABLED' | 'CANCEL_ONLY' | 'CLOSED';
};

export type MarketStatusOutcomeEntry = {
  label: string;
  price: string;
  indexSet: number;
  onChainId?: string;
};

export type MarketStatusUpdatePayload = {
  kind: 'marketStatusUpdate';
  tsMs: number;
  marketId: number;
  status: string;
  marketOutcomes?: MarketStatusOutcomeEntry[];
};

export type DepthLevel = [number, number];

export type OrderbookLastSettled = {
  id: string;
  price: string;
  kind: string;
  marketId: number;
  side: 'Ask' | 'Bid';
  outcome: 'Yes' | 'No';
};

export type PredictOrderbookUpdate = {
  version: 1;
  marketId: number;
  updateTimestampMs: number;
  orderCount: number;
  asks: DepthLevel[];
  bids: DepthLevel[];
  lastOrderSettled: OrderbookLastSettled | null;
  settlementsPending: { asks: DepthLevel[]; bids: DepthLevel[] };
};

export type MarketChangedUpdate = {
  kind: 'marketChanged';
  patchKind: string | null;
  tsMs: number;
  marketId: number;
  patch?: Record<string, unknown>;
};

export type CategoryChangedUpdate = {
  kind: 'categoryChanged';
  patchKind: string | null;
  tsMs: number;
  categoryId: number;
  slug: string;
  patch?: Record<string, unknown>;
};

/** A change with no detail. Refetch the category. */
export type CategoryChangedNotice = {
  categoryId: number;
  tsMs: number;
};

export type SystemEvent = {
  event: string;
  reason: string;
  graceMs: number;
};

/* Wallet events */

type BaseEvent = {
  orderId: string;
  orderHash: string;
  timestamp: number;
  walletAddress?: string;
  details: {
    marketId: number;
    outcomeIndex: number;
    marketQuestion: string;
    outcome: 'YES' | 'NO';
    outcomeName: string;
    quoteType: 'ASK' | 'BID';
    quantity: string;
    quantityFilled: string;
    price: string;
    value: string;
    valueFilled: string;
    strategyType: 'MARKET' | 'LIMIT';
    categorySlug: string;
  };
};

type Fill = {
  executedPriceWei: string;
  executedSizeWei: string;
  executedValueWei: string;
};

type OrderAccepted = BaseEvent & { type: 'orderAccepted' };
type OrderNotAccepted = BaseEvent & {
  type: 'orderNotAccepted';
  reason: 'noMarketMatch' | 'rejectedDuplicate' | 'rejectedPostOnly';
  settlementId?: string | null;
};
type OrderMarketMatchPlan = BaseEvent & {
  type: 'orderMarketMatchPlan';
  amount: string;
  amountFilled: string;
  plannedFill: string;
  remainderCancelled: string;
  settlementIds: string[];
};
type OrderExpired = BaseEvent & { type: 'orderExpired' };
type OrderCancelled = BaseEvent & {
  type: 'orderCancelled';
  reason?: 'selfTradePrevention';
};
type OrderTransactionSubmitted = BaseEvent & {
  type: 'orderTransactionSubmitted';
  kind: 'SALE' | 'PURCHASE';
  isMaker: boolean;
  settlementId: string;
  fill: Fill;
};
type OrderTransactionSuccess = BaseEvent & {
  type: 'orderTransactionSuccess';
  settlementId: string;
  fill: Fill;
  isMaker?: boolean;
  fee?: { amountWei: string; type: 'COLLATERAL' | 'SHARES' };
};
type OrderTransactionFailed = BaseEvent & {
  type: 'orderTransactionFailed';
  settlementId: string;
  fill: Fill;
};

export type PredictWalletEvents =
  | OrderAccepted
  | OrderNotAccepted
  | OrderMarketMatchPlan
  | OrderExpired
  | OrderCancelled
  | OrderTransactionSubmitted
  | OrderTransactionSuccess
  | OrderTransactionFailed;

/* Channels */

export interface PredictOrderbookChannel { name: 'predictOrderbook'; marketId: number }
export interface PredictTradingStatusChannel { name: 'predictTradingStatus'; marketId: number }
export interface PredictMarketStatusChannel { name: 'predictMarketStatus'; marketId: number }
export interface PredictMarketChangedChannel { name: 'predictMarketChanged'; marketId: number }
export interface PredictCategoryChangedChannel { name: 'predictCategoryChanged'; categoryId: number }
export interface PredictWalletEventsChannel { name: 'predictWalletEvents'; jwt: string }

export type Channel = Pretty<
  | PredictOrderbookChannel
  | PredictTradingStatusChannel
  | PredictMarketStatusChannel
  | PredictMarketChangedChannel
  | PredictCategoryChangedChannel
  | PredictWalletEventsChannel
>;

/* Requests */

type WithRequestId<T> = T & { requestId: number };
export type SubscribeRequest = WithRequestId<{ method: 'subscribe'; params: NonEmptyArray<string> }>;
export type UnsubscribeRequest = WithRequestId<{ method: 'unsubscribe'; params: NonEmptyArray<string> }>;
export type HeartbeatRequest = { method: 'heartbeat'; data: unknown };
export type Requests = Pretty<HeartbeatRequest | SubscribeRequest | UnsubscribeRequest>;

/* Errors */

type InvalidJson = { code: 'invalid_json'; message?: string };
export type InvalidTopic = { code: 'invalid_topic'; message?: string };
export type InvalidParams = { code: 'invalid_params'; message?: string };
export type InvalidCredentials = { code: 'invalid_credentials'; message?: string };
export type InternalServerError = { code: 'internal_server_error'; message?: string };

export type ResponseError = Pretty<
  InvalidJson | InvalidTopic | InvalidParams | InvalidCredentials | InternalServerError
>;

/* Responses */

export type RequestResponse = {
  type: 'R';
  requestId: number;
} & ({ success: true } | { success: false; error: ResponseError });

export type MessageResponse<Topic extends string, Data> = {
  type: 'M';
  topic: Topic;
  data: Data;
};

export type MessageResponses =
  | MessageResponse<'heartbeat', number>
  | MessageResponse<`predictOrderbook${string}`, PredictOrderbookUpdate>
  | MessageResponse<`predictTradingStatus${string}`, MarketTradingStatusUpdate>
  | MessageResponse<`predictMarketStatus${string}`, MarketStatusUpdatePayload>
  | MessageResponse<`predictMarketChanged${string}`, MarketChangedUpdate>
  | MessageResponse<`predictCategoryChanged${string}`, CategoryChangedUpdate | CategoryChangedNotice>
  | MessageResponse<`predictWalletEvents${string}`, PredictWalletEvents>;

export type Response = Pretty<
  MessageResponses | MessageResponse<'system', SystemEvent> | RequestResponse
>;

/* Internal */

export type WSError =
  | ResponseError
  | { code: 'ws_disconnected'; message?: string }
  | { code: 'ws_error_terminated'; message?: string };

export type TopicName = string;

export type EventCallback = (
  arg: { err?: null; data: MessageResponses['data'] } | { err: WSError; data?: null },
) => void;