Rework derived state (#7290)

* Remove derived state from state classes

* Create provider for derived state

Derived state is automatically stored to memory storage, but can be derived from any observable.

* Fixup state provider method definitions

* Test `DefaultDerivedState`

* remove implementation notes

* Write docs for derived state

* fixup derived state provider types

* Implement buffered delayUntil operator

* Move state types to a common module

* Move mock ports to centra location

* Alias DerivedStateDependency type

* Add dependencies to browser

* Prefer internal rxjs operators for ref counting

* WIP

* Ensure complete on subjects

* Foreground/background messaging for browser

Defers work for browser to the background

* Test foreground port behaviors

* Inject foreground and background derived state services

* remove unnecessary class field

* Adhere to required options

* Add dderived state to CLI

* Prefer type definition in type parameters to options

* Prefer instance method

* Implements factory methods for common uses

* Remove nothing test

* Remove share subject reference

Share manages connector subjects internally and will reuse them until
refcount is 0 and the cleanup time has passed. Saving our own reference
just risks memory leaks without real testability benefits.

* Fix interaction state

libs/common/src/platform/state/implementations/default-active-user-state.ts

@@ -18,12 +18,10 @@ import {
   AbstractStorageService,
   ObservableStorageService,
 } from "../../abstractions/storage.service";
-import { DerivedUserState } from "../derived-user-state";
 import { KeyDefinition, userKeyBuilder } from "../key-definition";
 import { StateUpdateOptions, populateOptionsWithDefault } from "../state-update-options";
-import { Converter, ActiveUserState, activeMarker } from "../user-state";
+import { ActiveUserState, activeMarker } from "../user-state";
 
-import { DefaultDerivedUserState } from "./default-derived-state";
 import { getStoredValue } from "./util";
 
 const FAKE_DEFAULT = Symbol("fakeDefault");
@@ -91,10 +89,6 @@ export class DefaultActiveUserState<T> implements ActiveUserState<T> {
     return await getStoredValue(key, this.chosenStorageLocation, this.keyDefinition.deserializer);
   }
 
-  createDerived<TTo>(converter: Converter<T, TTo>): DerivedUserState<TTo> {
-    return new DefaultDerivedUserState<T, TTo>(converter, this.encryptService, this);
-  }
-
   private async internalUpdate<TCombine>(
     configureState: (state: T, dependency: TCombine) => T,
     options: StateUpdateOptions<T, TCombine>,

libs/common/src/platform/state/implementations/default-derived-state.provider.ts

@@ -0,0 +1,49 @@
+import { Observable } from "rxjs";
+
+import { DerivedStateDependencies, ShapeToInstances } from "../../../types/state";
+import {
+  AbstractStorageService,
+  ObservableStorageService,
+} from "../../abstractions/storage.service";
+import { DeriveDefinition } from "../derive-definition";
+import { DerivedState } from "../derived-state";
+import { DerivedStateProvider } from "../derived-state.provider";
+
+import { DefaultDerivedState } from "./default-derived-state";
+
+export class DefaultDerivedStateProvider implements DerivedStateProvider {
+  private cache: Record<string, DerivedState<unknown>> = {};
+
+  constructor(protected memoryStorage: AbstractStorageService & ObservableStorageService) {}
+
+  get<TFrom, TTo, TDeps extends DerivedStateDependencies>(
+    parentState$: Observable<TFrom>,
+    deriveDefinition: DeriveDefinition<TFrom, TTo, TDeps>,
+    dependencies: ShapeToInstances<TDeps>,
+  ): DerivedState<TTo> {
+    const cacheKey = deriveDefinition.buildCacheKey();
+    const existingDerivedState = this.cache[cacheKey];
+    if (existingDerivedState != null) {
+      // I have to cast out of the unknown generic but this should be safe if rules
+      // around domain token are made
+      return existingDerivedState as DefaultDerivedState<TFrom, TTo, TDeps>;
+    }
+
+    const newDerivedState = this.buildDerivedState(parentState$, deriveDefinition, dependencies);
+    this.cache[cacheKey] = newDerivedState;
+    return newDerivedState;
+  }
+
+  protected buildDerivedState<TFrom, TTo, TDeps extends DerivedStateDependencies>(
+    parentState$: Observable<TFrom>,
+    deriveDefinition: DeriveDefinition<TFrom, TTo, TDeps>,
+    dependencies: ShapeToInstances<TDeps>,
+  ): DerivedState<TTo> {
+    return new DefaultDerivedState<TFrom, TTo, TDeps>(
+      parentState$,
+      deriveDefinition,
+      this.memoryStorage,
+      dependencies,
+    );
+  }
+}

libs/common/src/platform/state/implementations/default-derived-state.spec.ts

@@ -0,0 +1,222 @@
+/**
+ * need to update test environment so trackEmissions works appropriately
+ * @jest-environment ../shared/test.environment.ts
+ */
+import { Subject, firstValueFrom } from "rxjs";
+
+import { awaitAsync, trackEmissions } from "../../../../spec";
+import { FakeStorageService } from "../../../../spec/fake-storage.service";
+import { DeriveDefinition } from "../derive-definition";
+import { StateDefinition } from "../state-definition";
+
+import { DefaultDerivedState } from "./default-derived-state";
+
+let callCount = 0;
+const cleanupDelayMs = 10;
+const stateDefinition = new StateDefinition("test", "memory");
+const deriveDefinition = new DeriveDefinition<string, Date, { date: typeof Date }>(
+  stateDefinition,
+  "test",
+  {
+    derive: (dateString: string) => {
+      callCount++;
+      return new Date(dateString);
+    },
+    deserializer: (dateString: string) => new Date(dateString),
+    cleanupDelayMs,
+  },
+);
+
+describe("DefaultDerivedState", () => {
+  let parentState$: Subject<string>;
+  let memoryStorage: FakeStorageService;
+  let sut: DefaultDerivedState<string, Date, { date: typeof Date }>;
+  const deps = {
+    date: new Date(),
+  };
+
+  beforeEach(() => {
+    callCount = 0;
+    parentState$ = new Subject();
+    memoryStorage = new FakeStorageService();
+    sut = new DefaultDerivedState(parentState$, deriveDefinition, memoryStorage, deps);
+  });
+
+  afterEach(() => {
+    parentState$.complete();
+    jest.resetAllMocks();
+  });
+
+  it("should derive the state", async () => {
+    const dateString = "2020-01-01";
+    const emissions = trackEmissions(sut.state$);
+
+    parentState$.next(dateString);
+    await awaitAsync();
+
+    expect(emissions).toEqual([new Date(dateString)]);
+  });
+
+  it("should derive the state once", async () => {
+    const dateString = "2020-01-01";
+    trackEmissions(sut.state$);
+
+    parentState$.next(dateString);
+
+    expect(callCount).toBe(1);
+  });
+
+  it("should store the derived state in memory", async () => {
+    const dateString = "2020-01-01";
+    trackEmissions(sut.state$);
+    parentState$.next(dateString);
+    await awaitAsync();
+
+    expect(memoryStorage.internalStore[deriveDefinition.buildCacheKey()]).toEqual(
+      new Date(dateString),
+    );
+    const calls = memoryStorage.mock.save.mock.calls;
+    expect(calls.length).toBe(1);
+    expect(calls[0][0]).toBe(deriveDefinition.buildCacheKey());
+    expect(calls[0][1]).toEqual(new Date(dateString));
+  });
+
+  describe("forceValue", () => {
+    const initialParentValue = "2020-01-01";
+    const forced = new Date("2020-02-02");
+    let emissions: Date[];
+
+    describe("without observers", () => {
+      beforeEach(async () => {
+        parentState$.next(initialParentValue);
+        await awaitAsync();
+      });
+
+      it("should store the forced value", async () => {
+        await sut.forceValue(forced);
+        expect(memoryStorage.internalStore[deriveDefinition.buildCacheKey()]).toEqual(forced);
+      });
+    });
+
+    describe("with observers", () => {
+      beforeEach(async () => {
+        emissions = trackEmissions(sut.state$);
+        parentState$.next(initialParentValue);
+        await awaitAsync();
+      });
+
+      it("should store the forced value", async () => {
+        await sut.forceValue(forced);
+        expect(memoryStorage.internalStore[deriveDefinition.buildCacheKey()]).toEqual(forced);
+      });
+
+      it("should force the value", async () => {
+        await sut.forceValue(forced);
+        expect(emissions).toEqual([new Date(initialParentValue), forced]);
+      });
+
+      it("should only force the value once", async () => {
+        await sut.forceValue(forced);
+
+        parentState$.next(initialParentValue);
+        await awaitAsync();
+
+        expect(emissions).toEqual([
+          new Date(initialParentValue),
+          forced,
+          new Date(initialParentValue),
+        ]);
+      });
+    });
+  });
+
+  describe("cleanup", () => {
+    const newDate = "2020-02-02";
+
+    it("should cleanup after last subscriber", async () => {
+      const subscription = sut.state$.subscribe();
+      await awaitAsync();
+
+      subscription.unsubscribe();
+      // Wait for cleanup
+      await awaitAsync(cleanupDelayMs * 2);
+
+      expect(parentState$.observed).toBe(false);
+    });
+
+    it("should not cleanup if there are still subscribers", async () => {
+      const subscription1 = sut.state$.subscribe();
+      const sub2Emissions: Date[] = [];
+      const subscription2 = sut.state$.subscribe((v) => sub2Emissions.push(v));
+      await awaitAsync();
+
+      subscription1.unsubscribe();
+
+      // Wait for cleanup
+      await awaitAsync(cleanupDelayMs * 2);
+
+      // Still be listening to parent updates
+      parentState$.next(newDate);
+      await awaitAsync();
+      expect(sub2Emissions).toEqual([new Date(newDate)]);
+
+      subscription2.unsubscribe();
+      // Wait for cleanup
+      await awaitAsync(cleanupDelayMs * 2);
+
+      expect(parentState$.observed).toBe(false);
+    });
+
+    it("can re-initialize after cleanup", async () => {
+      const subscription = sut.state$.subscribe();
+      await awaitAsync();
+
+      subscription.unsubscribe();
+      // Wait for cleanup
+      await awaitAsync(cleanupDelayMs * 2);
+
+      const emissions = trackEmissions(sut.state$);
+      await awaitAsync();
+
+      parentState$.next(newDate);
+      await awaitAsync();
+
+      expect(emissions).toEqual([new Date(newDate)]);
+    });
+
+    it("should not cleanup if a subscriber joins during the cleanup delay", async () => {
+      const subscription = sut.state$.subscribe();
+      await awaitAsync();
+
+      await parentState$.next(newDate);
+      await awaitAsync();
+
+      subscription.unsubscribe();
+      // Do not wait long enough for cleanup
+      await awaitAsync(cleanupDelayMs / 2);
+
+      expect(parentState$.observed).toBe(true); // still listening to parent
+
+      const emissions = trackEmissions(sut.state$);
+      expect(emissions).toEqual([new Date(newDate)]); // we didn't lose our buffered value
+    });
+
+    it("state$ observables are durable to cleanup", async () => {
+      const observable = sut.state$;
+      let subscription = observable.subscribe();
+
+      await parentState$.next(newDate);
+      await awaitAsync();
+
+      subscription.unsubscribe();
+      // Wait for cleanup
+      await awaitAsync(cleanupDelayMs * 2);
+
+      subscription = observable.subscribe();
+      await parentState$.next(newDate);
+      await awaitAsync();
+
+      expect(await firstValueFrom(observable)).toEqual(new Date(newDate));
+    });
+  });
+});

libs/common/src/platform/state/implementations/default-derived-state.ts

@@ -1,23 +1,57 @@
-import { Observable, switchMap } from "rxjs";
+import { Observable, ReplaySubject, Subject, concatMap, merge, share, timer } from "rxjs";
 
-import { EncryptService } from "../../abstractions/encrypt.service";
-import { DerivedUserState } from "../derived-user-state";
-import { Converter, DeriveContext, UserState } from "../user-state";
+import { ShapeToInstances, DerivedStateDependencies } from "../../../types/state";
+import {
+  AbstractStorageService,
+  ObservableStorageService,
+} from "../../abstractions/storage.service";
+import { DeriveDefinition } from "../derive-definition";
+import { DerivedState } from "../derived-state";
+
+/**
+ * Default derived state
+ */
+export class DefaultDerivedState<TFrom, TTo, TDeps extends DerivedStateDependencies>
+  implements DerivedState<TTo>
+{
+  private readonly storageKey: string;
+  private forcedValueSubject = new Subject<TTo>();
 
-export class DefaultDerivedUserState<TFrom, TTo> implements DerivedUserState<TTo> {
   state$: Observable<TTo>;
 
   constructor(
-    private converter: Converter<TFrom, TTo>,
-    private encryptService: EncryptService,
-    private userState: UserState<TFrom>,
+    private parentState$: Observable<TFrom>,
+    protected deriveDefinition: DeriveDefinition<TFrom, TTo, TDeps>,
+    private memoryStorage: AbstractStorageService & ObservableStorageService,
+    private dependencies: ShapeToInstances<TDeps>,
   ) {
-    this.state$ = userState.state$.pipe(
-      switchMap(async (from) => {
-        // TODO: How do I get the key?
-        const convertedData = await this.converter(from, new DeriveContext(null, encryptService));
-        return convertedData;
+    this.storageKey = deriveDefinition.storageKey;
+
+    const derivedState$ = this.parentState$.pipe(
+      concatMap(async (state) => {
+        let derivedStateOrPromise = this.deriveDefinition.derive(state, this.dependencies);
+        if (derivedStateOrPromise instanceof Promise) {
+          derivedStateOrPromise = await derivedStateOrPromise;
+        }
+        const derivedState = derivedStateOrPromise;
+        await this.memoryStorage.save(this.storageKey, derivedState);
+        return derivedState;
+      }),
+    );
+
+    this.state$ = merge(this.forcedValueSubject, derivedState$).pipe(
+      share({
+        connector: () => {
+          return new ReplaySubject<TTo>(1);
+        },
+        resetOnRefCountZero: () => timer(this.deriveDefinition.cleanupDelayMs),
       }),
     );
   }
+
+  async forceValue(value: TTo) {
+    await this.memoryStorage.save(this.storageKey, value);
+    this.forcedValueSubject.next(value);
+    return value;
+  }
 }

libs/common/src/platform/state/implementations/default-single-user-state.ts

@@ -14,12 +14,10 @@ import {
   AbstractStorageService,
   ObservableStorageService,
 } from "../../abstractions/storage.service";
-import { DerivedUserState } from "../derived-user-state";
 import { KeyDefinition, userKeyBuilder } from "../key-definition";
 import { StateUpdateOptions, populateOptionsWithDefault } from "../state-update-options";
-import { Converter, SingleUserState } from "../user-state";
+import { SingleUserState } from "../user-state";
 
-import { DefaultDerivedUserState } from "./default-derived-state";
 import { getStoredValue } from "./util";
 
 const FAKE_DEFAULT = Symbol("fakeDefault");
@@ -68,10 +66,6 @@ export class DefaultSingleUserState<T> implements SingleUserState<T> {
     }
   }
 
-  createDerived<TTo>(converter: Converter<T, TTo>): DerivedUserState<TTo> {
-    return new DefaultDerivedUserState<T, TTo>(converter, this.encryptService, this);
-  }
-
   private async internalUpdate<TCombine>(
     configureState: (state: T, dependency: TCombine) => T,
     options: StateUpdateOptions<T, TCombine>,

libs/common/src/platform/state/implementations/default-state.provider.spec.ts

@@ -1,9 +1,13 @@
+import { of } from "rxjs";
+
 import {
   FakeActiveUserStateProvider,
+  FakeDerivedStateProvider,
   FakeGlobalStateProvider,
   FakeSingleUserStateProvider,
 } from "../../../../spec/fake-state-provider";
 import { UserId } from "../../../types/guid";
+import { DeriveDefinition } from "../derive-definition";
 import { KeyDefinition } from "../key-definition";
 import { StateDefinition } from "../state-definition";
 
@@ -14,15 +18,18 @@ describe("DefaultStateProvider", () => {
   let activeUserStateProvider: FakeActiveUserStateProvider;
   let singleUserStateProvider: FakeSingleUserStateProvider;
   let globalStateProvider: FakeGlobalStateProvider;
+  let derivedStateProvider: FakeDerivedStateProvider;
 
   beforeEach(() => {
     activeUserStateProvider = new FakeActiveUserStateProvider();
     singleUserStateProvider = new FakeSingleUserStateProvider();
     globalStateProvider = new FakeGlobalStateProvider();
+    derivedStateProvider = new FakeDerivedStateProvider();
     sut = new DefaultStateProvider(
       activeUserStateProvider,
       singleUserStateProvider,
       globalStateProvider,
+      derivedStateProvider,
     );
   });
 
@@ -53,4 +60,15 @@ describe("DefaultStateProvider", () => {
     const actual = sut.getGlobal(keyDefinition);
     expect(actual).toBe(existing);
   });
+
+  it("should bind the derivedStateProvider", () => {
+    const derivedDefinition = new DeriveDefinition(new StateDefinition("test", "disk"), "test", {
+      derive: () => null,
+      deserializer: () => null,
+    });
+    const parentState$ = of(null);
+    const existing = derivedStateProvider.get(parentState$, derivedDefinition, {});
+    const actual = sut.getDerived(parentState$, derivedDefinition, {});
+    expect(actual).toBe(existing);
+  });
 });

libs/common/src/platform/state/implementations/default-state.provider.ts

@@ -1,3 +1,9 @@
+import { Observable } from "rxjs";
+
+import { ShapeToInstances, DerivedStateDependencies } from "../../../types/state";
+import { DeriveDefinition } from "../derive-definition";
+import { DerivedState } from "../derived-state";
+import { DerivedStateProvider } from "../derived-state.provider";
 import { GlobalStateProvider } from "../global-state.provider";
 import { StateProvider } from "../state.provider";
 import { ActiveUserStateProvider, SingleUserStateProvider } from "../user-state.provider";
@@ -7,6 +13,7 @@ export class DefaultStateProvider implements StateProvider {
     private readonly activeUserStateProvider: ActiveUserStateProvider,
     private readonly singleUserStateProvider: SingleUserStateProvider,
     private readonly globalStateProvider: GlobalStateProvider,
+    private readonly derivedStateProvider: DerivedStateProvider,
   ) {}
 
   getActive: InstanceType<typeof ActiveUserStateProvider>["get"] =
@@ -16,4 +23,9 @@ export class DefaultStateProvider implements StateProvider {
   getGlobal: InstanceType<typeof GlobalStateProvider>["get"] = this.globalStateProvider.get.bind(
     this.globalStateProvider,
   );
+  getDerived: <TFrom, TTo, TDeps extends DerivedStateDependencies>(
+    parentState$: Observable<TFrom>,
+    deriveDefinition: DeriveDefinition<unknown, TTo, TDeps>,
+    dependencies: ShapeToInstances<TDeps>,
+  ) => DerivedState<TTo> = this.derivedStateProvider.get.bind(this.derivedStateProvider);
 }