optimistic-meteor

This package is designed to allow for Meteor's optimistic UI with method calls executed over HTTP to a (potentially) separate server than the one handling DDP calls.

Meteor's optimistic UX can (broadly) be reduced to four steps:

1. Run a stub of the method on the client - glossing over the details of saving and restoring originals, tracking invocations etc
2. Triggering the same functionality on the server, with the same random seed and receiving a response
3. Receiving all subscribed changes caused by the call
4. Then and only then, receiving an updated message.

It turns out none of this needs to be done over DDP, though some of it does make sense (e.g., the subscription and the updated method). In this package we utilise redis and reuse the redis-oplog package for reduced overhead, to communicate state between servers.

The following two diagrams broadly explains how this package works at a high level - for the two different usecases.

sequenceDiagram

title Client - Server(s) interactions

participant Client
participant HTTP server(s)
participant DDP Server(s)


Client->>DDP Server(s):Subscribes
Client->>HTTP server(s):Calls
HTTP server(s)-->>Client:Returns
HTTP server(s)->>DDP Server(s):Flushes
DDP Server(s)->>Client:Changes
DDP Server(s)->>Client:Updated

sequenceDiagram

title Client - Front End - Back End interactions

participant Client
participant Front End (DDP)
participant Back End (anything)


Client->>Front End (DDP):Subscribes
Client->>Front End (DDP):Calls
Front End (DDP)->>Back End (anything):Calls
Back End (anything)-->>Front End (DDP): Returns
Front End (DDP)->>Client:Changes
Front End (DDP)->>Client:Updated
Back End (anything)->>Front End (DDP):Flushes
Front End (DDP)->>Client: Returns

Details

This package can be imported into both a meteor and non-meteor server. The pubsub side expects to be on a meteor server, but the HTTP side doesn't need to be. You're also not limited to HTTP - any transport can be used to trigger any functionality on the server.

On the DDP server, you'll call:

import { Config } from "meteor/cultofcoders:redis-oplog";
import { getPubSubFromRedisOplog, pipeUpdatedOnFlush } from "optimistic-meteor";

const pubSubManager = getPubSubFromRedisOplog(Config.pubSubManager);
Meteor.server.onConnection(({
  id: sessionId,
  onClose
}) => {
  const handler = pipeUpdatedOnFlush(
    pubSubManager,
    sessionId,
    Meteor.server
  );
  onClose(() => handler.stop());
});

On the HTTP server you'll need to trigger some functionality that causes updates. If the HTTP server is also a meteor server, you might want to expose a method over HTTP.

import { getHttpHandler, init } from "optimistic-meteor";
import { Config } from "meteor/cultofcoders:redis-oplog";

init(Meteor.EnvironmentVariable); // optional. if you're *NOT* using a meteor server, you'd pass in `AsyncLocalStorage`
const methodName = "someMethodIveDefined";
const handler = getHttpHandler({
  server: Meteor.server,
  publisher: Config.pubSubManager.pusher
  methodName
});

WebApp.rawConnectHandlers.use(methodName, handler);

This will expose the meteor method named someMethodIveDefined, exactly as defined, but over HTTP. If (as is probably the case) you need to know which user triggerd the behaviour, provide a getUserId function

const handler = getHttpHandler({
  server: Meteor.server,
  publisher: Config.pubSubManager.pusher
  methodName,
  getUserId: async({ token }) => {
    // check it's a string here...
    return Meteor.users.findOne({ "services.resume.loginTokens.token": token }, { fields: { _id: 1 } })?._id
  }
});

The argument to getUserId is a user object - which you can control from the client side call - by default it will use the login token, pulled from local storage. However, you could easily amend this to take a signed JWT.

From the meteor client, you would call:

import { applyOverHTTPAndRunStubs } from "optimistic-meteor";

applyOverHTTPAndRunStubs(
  Meteor.connection,
  "someMethodIveDefined",
  [{ some: "arguments" }]
);

This will run the client stub of someMethodIveDefined then call over HTTP.

The final part of this is the HTTP server must determine which updates have been made. To do so, we must gather the redis-oplog channels that must be flushed. The easiest way to do this is with meteor's collection hooks. Or, if you're not using a meteor server, you can use mongo-collection-hooks.

import { gatherChannels } from "optimistic-meteor";

MyCollection.after.update((_userId, _doc, _fieldNames, _modifier, options) => {
  if (options.channels) {
    // use these instead
  }
  else {
    gatherChannels(["my-collection", `my-collection::${doc._id}`]);
  }
});
// same for after.insert and after.remove

gatherChannels knows which session and method call is in use and associates the channels with the session and method. When the HTTP call completes, it will "flush" all the channels written to, and send one additional message to the __session::{sessionId} channel with a list of channels to wait for notification from. Because redis ensures delivery of messages on the same channel in the correct order, when all flushes have been received, all the updates have also been observed and it's safe to send the updated message to the client.

Proxying requests to a backend server

In some cases you may not want to have totally separate stacks and instead have a front end server proxy work to a backend - but you still need to ensure the changes propagate before the method returns

import { Config } from "meteor/cultofcoders:redis-oplog";
import { getPubSubFromRedisOplog, optimisticUICallServer } from 'optimistic-meteor';

const pubSubManager = getPubSubFromRedisOplog(Config.pubSubManager);

Meteor.methods({
  myMethod(...args) {
    const methodId = Random.id();
    return optimisticUICallServer(() => {
      return (await fetch("http://someResourceThatUnderstands/myMethod", {
        body: JSON.stringify({
          sessionId: this.connection.id,
          methodId,
          userId: this.userId,
          args
        });
      })).json()
    }, {
      pubsub: pubSubManager,
      sessionId: this.sessionId,
      methodId
    });
  }
})