model: add simple mutation registration

src/Model.test.ts

@@ -227,5 +227,33 @@ describe('Model', () => {
     });
   });
 
+  it<ModelTestContext>('executes a registered mutation', async ({ streams }) => {
+    streams.s1.subscribe = vi.fn();
+    streams.s2.subscribe = vi.fn();
+    const model = new Model<string>('test', { streams, sync: async () => ({ version: 1, data: 'foobar' }) });
+    await modelStatePromise(model, ModelState.READY);
+
+    const mutation = vi.fn();
+    model.registerMutation('foo', mutation);
+    expect(mutation).toHaveBeenCalledTimes(0);
+    await model.mutate<[string, number], void>('foo', 'bar', 123);
+    expect(mutation).toHaveBeenCalledTimes(1);
+    expect(mutation).toHaveBeenCalledWith('bar', 123);
+  });
+
+  it<ModelTestContext>('fails to register a duplicate mutation', async ({ streams }) => {
+    streams.s1.subscribe = vi.fn();
+    streams.s2.subscribe = vi.fn();
+    const model = new Model<string>('test', { streams, sync: async () => ({ version: 1, data: 'foobar' }) });
+    await modelStatePromise(model, ModelState.READY);
+
+    const mutation = vi.fn();
+    model.registerMutation('foo', mutation);
+    expect(mutation).toHaveBeenCalledTimes(0);
+    expect(() => model.registerMutation('foo', mutation)).toThrowError(
+      `mutation with name 'foo' already registered on model 'test'`,
+    );
+  });
+
   // TODO disposes of the model on stream failed
 });

src/Model.ts

@@ -32,8 +32,15 @@ export enum ModelState {
 }
 
 export type SyncFunc<T> = () => Promise<Versioned<T>>;
+
 export type UpdateFunc<T> = (state: T, event: Types.Message) => Promise<T>;
 
+export type MutationResult<R> = {
+  result: R;
+  expected: Partial<Types.Message>;
+};
+export type MutationFunc<T extends any[] = any[], R = any> = (...args: T) => Promise<MutationResult<R>>;
+
 export type Streams = {
   [name: string]: Stream;
 };
@@ -65,9 +72,12 @@ class Model<T> extends EventEmitter<Record<ModelState, ModelStateChange>> {
   private currentData: Versioned<T>;
 
   private sync: SyncFunc<T>;
+
   private streams: Streams;
   private updateFuncs: UpdateFuncs<Versioned<T>> = {};
 
+  private mutations: Record<string, MutationFunc> = {};
+
   private subscriptions = new Subject<T>();
   private subscriptionMap: Map<StandardCallback<T>, Subscription> = new Map();
 
@@ -124,6 +134,25 @@ class Model<T> extends EventEmitter<Record<ModelState, ModelStateChange>> {
     this.updateFuncs[stream][event].push(update);
   }
 
+  public registerMutation(name: string, mutation: MutationFunc) {
+    if (this.mutations[name]) {
+      throw new Error(`mutation with name '${name}' already registered on model '${this.name}'`);
+    }
+    this.mutations[name] = mutation;
+  }
+
+  public async mutate<TArgs extends any[], R>(name: string, ...args: TArgs): Promise<MutationResult<R>> {
+    const mutation = this.mutations[name];
+    if (!mutation) {
+      throw new Error(`mutation with name '${name}' not registered on model '${this.name}'`);
+    }
+    const result = await mutation(...args);
+
+    // TODO: process the expected result optimistically
+
+    return result;
+  }
+
   public subscribe(callback: StandardCallback<T>) {
     const subscription = this.subscriptions.subscribe({
       next: (message) => callback(null, message),