replicache/src/replicache.ts at v0.5.0 · hmmhmmhm/replicache

Repository navigation

import type {JSONValue, ToJSON} from './json.js';
import type {ScanOptions} from './scan-options.js';
  OpenTransactionRequest,
  REPMWasmInvoker,
} from './repm-invoker.js';
import {ReadTransactionImpl, WriteTransactionImpl} from './transactions.js';
import {ScanResult} from './scan-iterator.js';
import type {ReadTransaction, WriteTransaction} from './transactions.js';
export type Mutator<Return extends JSONValue | void, Args extends JSONValue> = (
  args: Args,
) => Promise<Return | void>;
type MutatorImpl<Return extends JSONValue | void, Args extends JSONValue> = (
  tx: WriteTransaction,
  args: Args,
) => Promise<Return>;
type BeginSyncResult = {
  syncID: string;
  syncHead: string;
export const httpStatusUnauthorized = 401;
type MaybePromise<T> = T | Promise<T>;
/** The key name to use in localStorage when synchronizing changes. */
const storageKeyName = (name: string) => `/replicache/root/${name}`;
export default class Replicache implements ReadTransaction {
  private readonly _batchURL: string;
  private _dataLayerAuth: string;
  private readonly _diffServerAuth: string;
  private readonly _diffServerURL: string;
  private readonly _name: string;
  private readonly _repmInvoker: Invoker;
  private _closed = false;
  private _online = true;
  protected _opened: Promise<unknown> | null = null;
  private _root: Promise<string | undefined> = Promise.resolve(undefined);
  private readonly _mutatorRegistry = new Map<
    string,
    MutatorImpl<JSONValue, JSONValue>
  private _syncPromise: Promise<void> | null = null;
  private readonly _subscriptions = new Set<Subscription<unknown, unknown>>();
  private _syncInterval: number | null;
  // NodeJS has a non standard setTimeout function :'(
  protected _timerId: ReturnType<typeof setTimeout> | 0 = 0;
   * The delay between when a change is made to Replicache and when Replicache
   * attempts to push that change.
  public pushDelay: number;
  onSync: ((syncing: boolean) => void) | null = null;
   * This gets called when we get an HTTP unauthorized from the client view or
   * the batch endpoint. Set this to a function that will ask your user to
   * reauthenticate.
  getDataLayerAuth:
    | (() => MaybePromise<string | null | undefined>)
    | undefined = null;
  constructor({
    batchURL = '',
    dataLayerAuth = '',
    diffServerAuth = '',
    diffServerURL,
    name = 'default',
    repmInvoker = new REPMWasmInvoker(),
    syncInterval = 60_000,
    pushDelay = 300,
    batchURL?: string;
    dataLayerAuth?: string;
    diffServerAuth?: string;
    diffServerURL: string;
    name?: string;
    repmInvoker?: Invoker;
    syncInterval?: number | null;
    pushDelay?: number;
    this._batchURL = batchURL;
    this._dataLayerAuth = dataLayerAuth;
    this._diffServerAuth = diffServerAuth;
    this._diffServerURL = diffServerURL;
    this._name = name;
    this._repmInvoker = repmInvoker;
    this._syncInterval = syncInterval;
    this.pushDelay = pushDelay;
    this._open();
  private async _open(): Promise<void> {
    this._opened = this._repmInvoker.invoke(this._name, 'open');
    this._setRoot(this._getRoot());
    await this._root;
    if (this._syncInterval !== null) {
      await this.sync();
    window.addEventListener('storage', this._onStorage);
  get online(): boolean {
    return this._online;
  get closed(): boolean {
    return this._closed;
   * The duration between each `sync`. Set this to `null` to prevent syncing in
   * the background.
  get syncInterval(): number | null {
    return this._syncInterval;
  set syncInterval(duration: number | null) {
    this._clearTimer();
    this._syncInterval = duration;
    this._scheduleSync(this._syncInterval);
  private _scheduleSync(interval: number | null): void {
    if (interval) {
      this._timerId = setTimeout(() => this.sync(), interval);
  private _clearTimer() {
    if (this._timerId !== 0) {
      clearTimeout(this._timerId);
      this._timerId = 0;
  async close(): Promise<void> {
    this._closed = true;
    const p = this._invoke('close');
    this._clearTimer();
    window.removeEventListener('storage', this._onStorage);
    // Clear subscriptions
    for (const subscription of this._subscriptions) {
      subscription.onDone?.();
    this._subscriptions.clear();
    await p;
  private async _getRoot(): Promise<string | undefined> {
    if (this._closed) {
      return undefined;
    const res = await this._invoke('getRoot');
    return res.root;
  private _setRoot(root: Promise<string | undefined>) {
    this._root = root;
    this._setStorage(root);
  private async _setStorage(root: Promise<string | undefined>) {
    // Also set an item in localStorage so that we can synchronize multiple
    // windows/tabs.
    localStorage[storageKeyName(this._name)] = await root;
  // Callback for when window.onstorage fires which happens when a different tab
  // changes the db.
  private _onStorage = (e: StorageEvent): void => {
    if (e.key === storageKeyName(this._name)) {
      this._checkChange(e.newValue as string);
  private async _checkChange(root: string | undefined): Promise<void> {
    const currentRoot = await this._root; // instantaneous except maybe first time
    if (root !== undefined && root !== currentRoot) {
      this._setRoot(Promise.resolve(root));
      await this._fireOnChange();
  private _invoke: Invoke = async (
    rpc: string,
    args?: JSONValue | ToJSON,
  ): Promise<JSONValue> => {
    await this._opened;
    return await this._repmInvoker.invoke(this._name, rpc, args);
  /** Get a single value from the database. */
  get(key: string): Promise<JSONValue | undefined> {
    return this.query(tx => tx.get(key));
  /** Determines if a single key is present in the database. */
  has(key: string): Promise<boolean> {
    return this.query(tx => tx.has(key));
   * Gets many values from the database. This returns a `Result` which
   * implements `AsyncIterable`. It also has methods to iterate over the `keys`
   * and `entries`.
  scan({prefix = '', start}: ScanOptions = {}): ScanResult {
    let tx: ReadTransactionImpl;
    return new ScanResult(
      prefix,
      start,
      this._invoke,
      async () => {
        if (tx) {
          return tx;
        tx = new ReadTransactionImpl(this._invoke);
        await tx.open({});
        return tx;
      true,
   * Convenience form of scan() which returns all the results as an array.
  async scanAll(options: ScanOptions = {}): Promise<[string, JSONValue][]> {
    const tx = new ReadTransactionImpl(this._invoke);
      await tx.open({});
      return await tx.scanAll(options);
    } finally {
      tx.close();
  private async _sync(): Promise<void> {
      const beginSyncResult = await this._beginSync();
      // TODO(repc-switchover)
      // replicache-client sends all zeros for null sync,
      // repc sends empty string.
      if (beginSyncResult.syncHead.replace(/0/g, '') !== '') {
        await this._maybeEndSync(beginSyncResult);
    } catch (e) {
      // The error paths of beginSync and maybeEndSync need to be reworked.
      // We want to distinguish between:
      // a) network requests failed -- we're offline basically
      // b) sync was aborted because one's already in progress
      // c) oh noes - something unexpected happened
      // Right now, all of these come out as errors. We distinguish (b) with a
      // hacky string search. (a) and (c) are not distinguishable currently
      // because repc doesn't provide sufficient information, so we treat all
      // errors that aren't (b) as (a).
      if (e.toString().includes('JSLogInfo')) {
      console.info(`Sync returned: ${e}`);
    this._online = online;
  protected async _beginSync(): Promise<BeginSyncResult> {
    const beginSyncResult = await this._invoke('beginSync', {
      batchPushURL: this._batchURL,
      diffServerURL: this._diffServerURL,
      dataLayerAuth: this._dataLayerAuth,
      diffServerAuth: this._diffServerAuth,
    const {syncInfo} = beginSyncResult;
    let reauth = false;
    function checkStatus(
      data: {httpStatusCode?: number; errorMessage?: string},
      serverName: string,
      const {httpStatusCode, errorMessage} = data;
      if (errorMessage !== '') {
        console.error(
          `Got error response from ${serverName} server: ${httpStatusCode}: ${errorMessage}`,
      if (httpStatusCode === httpStatusUnauthorized) {
        reauth = true;
    const {batchPushInfo, clientViewInfo} = syncInfo;
    if (batchPushInfo) {
      checkStatus(batchPushInfo, 'batch');
      const mutationInfos = batchPushInfo.batchPushResponse?.mutationInfos;
      if (mutationInfos != null) {
        for (const mutationInfo of mutationInfos) {
          console.error(
            `MutationInfo: ID: ${mutationInfo.id}, Error: ${mutationInfo.error}`,
    if (clientViewInfo) {
      checkStatus(clientViewInfo, 'client view');
    if (reauth && this.getDataLayerAuth) {
      const dataLayerAuth = await this.getDataLayerAuth();
      if (dataLayerAuth != null) {
        this._dataLayerAuth = dataLayerAuth;
        // Try again now instead of waiting for another 5 seconds.
        return await this._beginSync();
    const syncHead = beginSyncResult.syncHead;
    const {syncID} = syncInfo;
    return {syncID, syncHead};
  protected async _maybeEndSync(
    beginSyncResult: BeginSyncResult,
  ): Promise<void> {
    if (this._closed) {
      return;
    let {syncHead} = beginSyncResult;
    const {replayMutations} = await this._invoke(
      'maybeEndSync',
      beginSyncResult,
    if (!replayMutations || replayMutations.length === 0) {
      // All done.
      await this._checkChange(syncHead);
      return;
    // Replay.
    for (const mutation of replayMutations) {
      const {original} = mutation;
      console.log('parsing', mutation.args);
      syncHead = await this._replay(
        syncHead,
        original,
        mutation.name,
        JSON.parse(mutation.args),
    const {syncID} = beginSyncResult;
    await this._maybeEndSync({syncID, syncHead});
  private async _replay<A extends JSONValue>(
    basis: string,
    original: string,
    name: string,
    args: A,
  ): Promise<string> {
    const mutatorImpl = this._mutatorRegistry.get(name);
    if (!mutatorImpl) {
      console.error(`Unknown mutator ${name}`);
      return basis;
    const res = await this._mutate(name, mutatorImpl, args, {
      invokeArgs: {
        rebaseOpts: {basis, original},
      shouldCheckChange: false,
    return res.ref;
   * Synchronizes this cache with the server. New local mutations are sent to
   * the server, and the latest server state is applied to the cache. Any local
   * mutations not included in the new server state are replayed. See the
   * Replicache design document for more information on sync:
   * https://github.com/rocicorp/replicache/blob/master/design.md
  async sync(): Promise<void> {
    if (this._closed) {
      return;
    if (this._syncPromise !== null) {
      await this._syncPromise;
      await this.sync();
      return;
    this._clearTimer();
    this._fireOnSync(true);
      this._syncPromise = this._sync();
      await this._syncPromise;
    } finally {
      this._syncPromise = null;
      this._fireOnSync(false);
      this._scheduleSync(this._syncInterval);
  private _fireOnSync(syncing: boolean): void {
    queueMicrotask(() => this.onSync?.(syncing));
  private async _fireOnChange(): Promise<void> {
    const subscriptions = [...this._subscriptions];
    const results = await this.query(async tx => {
      const promises = subscriptions.map(async s => {
        // Tag the result so we can deal with success vs error below.
        try {
          return {ok: true, value: await s.body(tx)};
        } catch (ex) {
          return {ok: false, error: ex};
      return await Promise.all(promises);
    for (let i = 0; i < subscriptions.length; i++) {
      const result = results[i];
      if (result.ok) {
        subscriptions[i].onData(result.value);
      } else {
        subscriptions[i].onError?.(result.error);
   * Subcribe to changes to the underlying data. Every time the underlying data
   * changes `onData` is called. The function is also called once the first time
   * the subscription is added. There is currently no guarantee that the res
E8B9
ult
   * of this subscription changes and it might get called with the same value
   * over and over.
   * This returns a function that can be used to cancel the subscription.
   * If an error occurs in the `body` the `onError` function is called if
   * present. Otherwise, the error is thrown.
  subscribe<R, E>(
    body: (tx: ReadTransaction) => Promise<R>,
      onData,
      onError,
      onDone,
      onData: (result: R) => void;
      onError?: (error: E) => void;
      onDone?: () => void;
  ): () => void {
    const s = {body, onData, onError, onDone} as Subscription<unknown, unknown>;
    this._subscriptions.add(s);
    (async () => {
      try {
        const res = await this.query(s.body);
        s.onData(res);
      } catch (ex) {
        if (s.onError) {
          s.onError(ex);
        } else {
          throw ex;
    return (): void => {
      this._subscriptions.delete(s);
   * Query is used for read transactions. It is recommended to use transactions
   * to ensure you get a consistent view across multiple calls to `get`, `has`
   * and `scan`.
  async query<R>(body: (tx: ReadTransaction) => Promise<R> | R): Promise<R> {
    const tx = new ReadTransactionImpl(this._invoke);
    await tx.open({});
      return await body(tx);
    } finally {
      // No need to await the response.
      tx.close();
   * Registers a *mutator*, which is used to make changes to the data.
   * ## Replays
   * Mutators run once when they are initially invoked, but they might also be
   * *replayed* multiple times during sync. As such mutators should not modify
   * application state directly. Also, it is important that the set of
   * registered mutator names only grows over time. If Replicache syncs and
   * needed mutator is not registered, it will substitute a no-op mutator, but
   * this might be a poor user experience.
   * ## Server application
   * During sync, a description of each mutation is sent to the server's [batch
   * endpoint](https://github.com/rocicorp/replicache/blob/master/README.md#step-5-upstream-sync)
   * where it is applied. Once the mutation has been applied successfully, as
   * indicated by the [client
   * view](https://github.com/rocicorp/replicache/blob/master/README.md#step-2-downstream-sync)'s
   * `lastMutationId` field, the local version of the mutation is removed. See
   * the [design
   * doc](https://github.com/rocicorp/replicache/blob/master/design.md) for
   * additional details on the sync protocol.
   * ## Transactionality
   * Mutators are atomic: all their changes are applied together, or none are.
   * Throwing an exception aborts the transaction. Otherwise, it is committed.
   * As with [query] and [subscribe] all reads will see a consistent view of
   * the cache while they run.
  register<Return extends JSONValue | void, Args extends JSONValue>(
    name: string,
    mutatorImpl: MutatorImpl<Return, Args>,
  ): Mutator<Return, Args> {
    this._mutatorRegistry.set(
      name,
      (mutatorImpl as unknown) as MutatorImpl<JSONValue, JSONValue>,
    return async (args: Args): Promise<Return> =>
      (await this._mutate(name, mutatorImpl, args, {shouldCheckChange: true}))
        .result;
  private async _mutate<R extends JSONValue | void, A extends JSONValue>(
    name: string,
    mutatorImpl: MutatorImpl<R, A>,
    args: A,
      invokeArgs,
      shouldCheckChange,
    }: {invokeArgs?: OpenTransactionRequest; shouldCheckChange: boolean},
  ): Promise<{result: R; ref: string}> {
    let actualInvokeArgs: OpenTransactionRequest = {
      args: JSON.stringify(args),
      name,
    if (invokeArgs !== undefined) {
      actualInvokeArgs = {...actualInvokeArgs, ...invokeArgs};
    let result: R;
    const tx = new WriteTransactionImpl(this._invoke);
    await tx.open(actualInvokeArgs);
      result = await mutatorImpl(tx, args);
    } catch (ex) {
      // No need to await the response.
      tx.close();
      throw ex;
    const commitRes = await tx.commit();
    if (commitRes.retryCommit) {
      return await this._mutate(name, mutatorImpl, args, {
        invokeArgs,
        shouldCheckChange,
    const {ref} = commitRes;
    if (shouldCheckChange) {
      await this._checkChange(ref);
    this._clearTimer();
    this._scheduleSync(this.pushDelay);
    return {result, ref};
export class ReplicacheTest extends Replicache {
  static async new({
    batchURL,
    dataLayerAuth,
    diffServerAuth,
    diffServerURL,
    name = '',
    repmInvoker,
    diffServerURL: string;
    batchURL?: string;
    dataLayerAuth?: string;
    diffServerAuth?: string;
    name?: string;
    repmInvoker: Invoker;
  }): Promise<ReplicacheTest> {
    const rep = new ReplicacheTest({
      batchURL,
      dataLayerAuth,
      diffServerAuth,
      diffServerURL,
      name,
      repmInvoker,
      syncInterval: null,
    await rep._opened;
    return rep;
  beginSync(): Promise<BeginSyncResult> {
    return super._beginSync();
  maybeEndSync(beginSyncResult: BeginSyncResult): Promise<void> {
    return super._maybeEndSync(beginSyncResult);
type Subscription<R, E> = {
  body: (tx: ReadTransaction) => Promise<R>;
  onData: (r: R) => void;
  onError?: (e: E) => void;
  onDone?: () => void;
FilesExpand file tree

replicache.ts

Latest commit

History

replicache.ts

File metadata and controls
