move client from separate repo

client/lib/action-serializer.ts

@@ -0,0 +1,44 @@
+import { Subject, BehaviorSubject, of, Observable, Observer, throwError } from 'rxjs'
+import { concatMap, map, catchError, filter, take } from 'rxjs/operators'
+
+export type Action<T> = {
+  action: () => T,
+  notify: BehaviorSubject<undefined | T>
+}
+
+export class ActionSerializer {
+  private readonly sequentialActions = new Subject<Action<any>>()
+
+  constructor () {
+    this.sequentialActions.pipe(
+      concatMap(({ action, notify }) => fromSync$(action).pipe(
+        catchError(e => of(notify.next({ error: e }))),
+        map(result => notify.next({ result })),
+      )),
+      catchError(e => of(console.error(`Action Serializer Exception`, e))),
+    ).subscribe()
+  }
+
+  run$<T> (action: () => T): Observable<T> {
+    const notify = new BehaviorSubject(undefined) as BehaviorSubject<T | undefined>
+    this.sequentialActions.next({ action, notify })
+    return (notify as BehaviorSubject<T>).pipe(
+      filter(res => res !== undefined),
+      take(1),
+      concatMap((res: any) => res.error ? throwError(res.error) : of(res.result)),
+    )
+  }
+}
+
+function fromSync$<S, T> (action: (s: S) => T, s: S): Observable<T>
+function fromSync$<T> (action: () => T): Observable<T>
+function fromSync$<S, T> (action: (s: S) => T, s?: S): Observable<T> {
+  return new Observable((subscriber: Observer<T>) => {
+    try {
+      subscriber.next(action(s as S))
+      subscriber.complete()
+    } catch (e) {
+      subscriber.error(e)
+    }
+  })
+}

client/lib/patch-db.ts

@@ -0,0 +1,136 @@
+import { EMPTY, from, merge, Observable, of, Subject, timer } from 'rxjs'
+import { catchError, concatMap, delay, finalize, map, skip, take, takeUntil, tap, throttleTime } from 'rxjs/operators'
+import { Source } from './source/source'
+import { Dump, SequenceStore, Result, Revision } from './sequence-store'
+import { Store } from './store'
+export { Operation } from 'fast-json-patch'
+
+export class PatchDB<T extends object> {
+  private readonly cancelStashTimeout = new Subject()
+
+  private constructor (
+    private readonly sources: Source<T>[],
+    private readonly http: Http<T>,
+    private readonly sequenceStore: SequenceStore<T>,
+    private readonly timeoutForMissingRevision: number = 5000,
+  ) { }
+
+  get store (): Store<T> { return this.sequenceStore.store }
+
+  static async init<T extends object> (conf: PatchDbConfig<T>): Promise<PatchDB<T>> {
+    console.log('PATCHDB - init(): ', conf)
+    const { sources, http, bootstrapper, timeoutForMissingRevision } = conf
+
+    let sequence: number = 0
+    let data: T = { } as T
+    if (bootstrapper) {
+      try {
+        const cache = await bootstrapper.init()
+        console.log('FROM CACHE', cache)
+        sequence = cache.sequence
+        data = cache.data
+      } catch (e) {
+        console.error('bootstrapper failed: ', e)
+      }
+    }
+
+    const store = new Store(data)
+
+    const sequenceStore = new SequenceStore(store, sequence)
+
+    // update cache when sequenceStore emits, throttled
+    if (bootstrapper) {
+      sequenceStore.watch$().pipe(throttleTime(500), delay(500), skip(1)).subscribe(({ data, sequence }) => {
+        console.log('PATCHDB - update cache(): ', sequence, data)
+        bootstrapper.update({ sequence, data }).catch(e => {
+          console.error('Exception in updateCache: ', e)
+        })
+      })
+    }
+
+    return new PatchDB(sources, http, sequenceStore, timeoutForMissingRevision)
+  }
+
+  sync$ (): Observable<void> {
+    console.log('PATCHDB - sync$()')
+    const sequence$ = this.sequenceStore.watch$().pipe(map(cache => cache.sequence))
+    // nested concatMaps, as it is written, ensure sync is not run for update2 until handleSyncResult is complete for update1.
+    // flat concatMaps would allow many syncs to run while handleSyncResult was hanging. We can consider such an idea if performance requires it.
+    return merge(...this.sources.map(s => s.watch$(sequence$))).pipe(
+      tap(update => console.log('PATCHDB - sources updated:', update)),
+      concatMap(update =>
+        this.sequenceStore.update$(update).pipe(
+          concatMap(res => this.handleSyncResult$(res)),
+        ),
+      ),
+      finalize(() => {
+        console.log('FINALIZING')
+        this.sequenceStore.sequence = 0
+      }),
+    )
+  }
+
+  private handleSyncResult$ (res: Result): Observable<void> {
+    console.log('PATCHDB - handleSyncResult$(): ', res)
+    switch (res) {
+      case Result.DUMPED: return of(this.cancelStashTimeout.next('')) // cancel stash timeout
+      case Result.REVISED: return of(this.cancelStashTimeout.next('')) // cancel stash timeout
+      case Result.STASHED: return this.handleStashTimeout$() // call error after timeout
+      case Result.ERROR: return this.handlePatchError$() // call error immediately
+      default: return EMPTY
+    }
+  }
+
+  private handleStashTimeout$ (): Observable<void> {
+    console.log('PATCHDB - handleStashTimeout$()')
+    return timer(this.timeoutForMissingRevision).pipe(
+      tap(time => console.log('PATCHDB - timeout for missing patch:', time)),
+      takeUntil(this.cancelStashTimeout),
+      take(1),
+      concatMap(() => this.handlePatchError$()),
+    )
+  }
+
+  // Here flattened concatMaps are functionally equivalent to nested because the source observable emits at most once.
+  private handlePatchError$ (): Observable<void> {
+    return from(this.http.getDump()).pipe(
+      concatMap(dump => this.sequenceStore.update$(dump)),
+      // note the above is a "dump" update, which will always return DUMPED (it can't error)
+      // handleSyncResult will therefore never re-call handlePatchError()
+      concatMap(res => this.handleSyncResult$(res)),
+      catchError(e => {
+        console.error(e)
+        return EMPTY
+      }),
+    )
+  }
+}
+
+export type PatchDbConfig<T> = {
+  http: Http<T>
+  sources: Source<T>[]
+  bootstrapper?: Bootstrapper<T>
+  timeoutForMissingRevision?: number
+}
+
+export enum PatchOp {
+  ADD = 'add',
+  REMOVE = 'remove',
+  REPLACE = 'replace',
+}
+
+export interface Http<T> {
+  getRevisions (since: number): Promise<Revision[] | Dump<T>>
+  getDump (): Promise<Dump<T>>
+}
+
+export interface Bootstrapper<T> {
+  init (): Promise<DBCache<T>>
+  update (cache: DBCache<T>): Promise<void>
+  clear (): Promise<void>
+}
+
+export interface DBCache<T>{
+  sequence: number,
+  data: T
+}

client/lib/sequence-store.ts

@@ -0,0 +1,211 @@
+import { BehaviorSubject, Observable } from 'rxjs'
+import { filter } from 'rxjs/operators'
+import { Store } from './store'
+import { DBCache } from './patch-db'
+import { patchDocument } from './store'
+import { ActionSerializer } from './action-serializer'
+import { Operation } from 'fast-json-patch'
+import BTree from 'sorted-btree'
+
+export class SequenceStore<T extends object> {
+  private readonly lastState$: BehaviorSubject<DBCache<T>> = new BehaviorSubject(undefined as any)
+  private readonly actionSerializer = new ActionSerializer()
+  private preTemps: T
+  private stashed = new BTree<number, Revision>()
+  private temps: UpdateTemp[] = []
+  private sequence$ = new BehaviorSubject(0)
+
+  constructor (
+    readonly store: Store<T>,
+    initialSequence: number,
+  ) {
+    const data = store.peek
+    this.preTemps = data
+    this.commit({ data, sequence: initialSequence }, [])
+  }
+
+  get sequence (): number { return this.sequence$.getValue() }
+  set sequence (seq: number) { this.sequence$.next(seq) }
+
+  // subscribe to watch$ to get sequence + T feed, e.g. for caching and bootstrapping from a cache
+  watch$ (): Observable<DBCache<T>> {
+    return this.lastState$.pipe(filter(a => !!a))
+  }
+
+  update$ (update: Update<T>): Observable<Result> {
+    return this.actionSerializer.run$(() => {
+      if (isTemp(update)) {
+        return this.updateTemp(update)
+      } else {
+        return this.updateReal(update)
+      }
+    })
+  }
+
+  viewRevisions (): Revision[] {
+    // return this.revisions.filter(a => !!a)
+    return this.stashed.valuesArray()
+  }
+
+  private updateReal (update: UpdateReal<T>): Result {
+    if (update.expireId) { this.temps = this.temps.filter(temp => temp.expiredBy !== update.expireId) }
+    if (update.id <= this.sequence) return Result.NOOP
+
+    const { result, dbCache, revisionsToDelete } = isDump(update) ?
+      this.dump(update) :
+      this.revise(update)
+
+    this.preTemps = dbCache.data
+    const afterTemps = this.stageSeqTemps(dbCache)
+    this.commit(afterTemps, revisionsToDelete)
+    return result
+  }
+
+  private updateTemp (update: UpdateTemp): Result {
+    this.temps.push(update)
+    const data = patchDocument(update.patch, this.store.peek)
+    const res = {
+      data,
+      sequence: this.sequence,
+    }
+    this.commit(res, [])
+    return Result.TEMP
+  }
+
+  private commit (res: DBCache<T>, revisionsToDelete: number[]): void {
+    const { data, sequence } = res
+    this.stashed.deleteKeys(revisionsToDelete)
+    this.sequence$.next(sequence)
+    this.store.set(data)
+    this.lastState$.next({ data, sequence })
+  }
+
+  private dump (dump: Dump<T>): { result: Result, dbCache: DBCache<T>, revisionsToDelete: number[] } {
+    try {
+      const oldRevisions = this.stashed.filter((key, _) => key < dump.id).keysArray()
+      const { dbCache, revisionsToDelete } = this.processRevisions(dump.value, dump.id)
+      return {
+        result: Result.DUMPED,
+        dbCache,
+        revisionsToDelete: oldRevisions.concat(revisionsToDelete),
+      }
+    } catch (e) {
+      console.error(`Dump error for ${JSON.stringify(dump)}: `, e)
+      return {
+        result: Result.ERROR,
+        dbCache: {
+          data: this.preTemps,
+          sequence: this.sequence,
+        },
+        revisionsToDelete: [],
+      }
+    }
+  }
+
+  private revise (revision: Revision): { result: Result, dbCache: DBCache<T>, revisionsToDelete: number[] } {
+    this.stashed.set(revision.id, revision)
+    try {
+      return this.processRevisions(this.preTemps, this.sequence)
+    } catch (e) {
+      console.error(`Revise error for ${JSON.stringify(revision)}: `, e)
+      return {
+        result: Result.ERROR,
+        dbCache: {
+          data: this.preTemps,
+          sequence: this.sequence,
+        },
+        revisionsToDelete: [],
+      }
+    }
+  }
+
+  private stageSeqTemps<S extends DBCache<T>> (resultSoFar: S): S {
+    return this.temps.reduce(({ data, ...rest }, nextTemp ) => {
+      try {
+        const nextContents = patchDocument(nextTemp.patch, data)
+        return { data: nextContents, ...rest } as S
+      } catch (e) {
+        console.error(`Skipping temporary patch ${JSON.stringify(nextTemp)} due to exception: `, e)
+        return { data, ...rest } as S
+      }
+    }, resultSoFar)
+  }
+
+  private processRevisions (data: T, sequence: number): { result: Result, dbCache: DBCache<T>, revisionsToDelete: number[] } {
+    const applicableRevisions = this.applicableRevisions(sequence)
+
+    console.log('APPLICABLE: ', applicableRevisions)
+
+    if (!applicableRevisions.length) {
+      return {
+        result: Result.STASHED,
+        dbCache: {
+          data,
+          sequence,
+        },
+        revisionsToDelete: [],
+      }
+    }
+
+    const revisionsToDelete: number[] = []
+    const toReturn = applicableRevisions.reduce(({ data, sequence }, revision) => {
+      const nextContents = patchDocument(revision.patch, data)
+      const nextSequence = sequence + 1
+      revisionsToDelete.push(revision.id) // @TODO original was `revisionsToDelete.concat([seqPatch.id])`, why?
+      return { data: nextContents, sequence: nextSequence }
+    }, { data, sequence })
+
+    return {
+      result: Result.REVISED,
+      dbCache: toReturn,
+      revisionsToDelete,
+    }
+  }
+
+  private applicableRevisions (sequence: number): Revision[] {
+    const toReturn = [] as Revision[]
+
+    let i = sequence
+    while (true) {
+      i++
+      const next = this.stashed.get(i)
+      if (next) {
+        toReturn.push(next)
+      } else {
+        break
+      }
+    }
+
+    return toReturn
+  }
+}
+
+export enum Result {
+  DUMPED = 'DUMPED', // store was dumped/replaced
+  REVISED = 'REVISED', // store was revised
+  TEMP = 'TEMP', // store was revised temporarily
+  STASHED = 'STASHED', // attempted to revise store but sequence too high. revision stashed for later
+  ERROR = 'ERROR', // attempted to revise/dump store, but failed
+  NOOP = 'NOOP', // sequence too low, update ignored
+}
+
+// revise a collection of nodes.
+export type Revision = { id: number, patch: Operation[], expireId: string | null }
+// dump/replace the entire store with T
+export type Dump<T> = { id: number, value: T, expireId: string | null }
+
+export type Update<T> = UpdateReal<T> | UpdateTemp
+export type UpdateReal<T> = Revision | Dump<T>
+export type UpdateTemp = Omit<Revision, 'id' | 'expireId'> & { expiredBy : string }
+
+function isTemp<T> (s: Update<T>): s is UpdateTemp {
+  return !!(s as any).expiredBy
+}
+
+function isRevision<T> (s: Update<T>): s is Revision {
+  return !isTemp(s) && !!(s as any).patch
+}
+
+function isDump<T> (s: UpdateReal<T>): s is Dump<T> {
+  return !isTemp(s) && !!(s as any).value
+}

client/lib/source/poll-source.ts

@@ -0,0 +1,56 @@
+import { BehaviorSubject, concat, from, Observable, of } from 'rxjs'
+import { catchError, concatMap, delay, skip, switchMap, take, tap } from 'rxjs/operators'
+import { Http } from '../patch-db'
+import { UpdateReal } from '../sequence-store'
+import { Source } from './source'
+
+export type PollConfig = {
+  cooldown: number
+}
+
+export class PollSource<T> implements Source<T> {
+
+  constructor (
+    private readonly pollConfig: PollConfig,
+    private readonly http: Http<T>,
+  ) { }
+
+  watch$ (sequence$: Observable<number>): Observable<UpdateReal<T>> {
+    console.log('POLL_SOURCE - watch$()')
+
+    const polling$ = new BehaviorSubject('')
+
+    const updates$ = of('').pipe(
+      concatMap(_ => sequence$),
+      take(1),
+      tap(_ => console.log('making request')),
+      concatMap(seq => this.http.getRevisions(seq)),
+      catchError(e => {
+        console.error(e)
+        return of([])
+      }),
+      tap(_ => console.log('request complete')),
+    )
+
+    const delay$ = of([]).pipe(
+      tap(_ => console.log('starting cooldown')),
+      delay(this.pollConfig.cooldown),
+      tap(_ => console.log('cooldown finished')),
+      tap(_ => polling$.next('')),
+      skip(1),
+    )
+
+    const poll$ = concat(updates$, delay$)
+
+    return polling$.pipe(
+       switchMap(_ => poll$),
+       concatMap(res => {
+        if (Array.isArray(res)) {
+          return from(res) // takes Revision[] and converts it into Observable<Revision>
+        } else {
+          return of(res) // takes Dump<T> and converts it into Observable<Dump<T>>
+        }
+       }),
+    )
+  }
+}

client/lib/source/source.ts

@@ -0,0 +1,6 @@
+import { Observable } from 'rxjs'
+import { Update } from '../sequence-store'
+
+export interface Source<T> {
+  watch$ (sequence$?: Observable<number>): Observable<Update<T>>
+}

client/lib/source/ws-source.ts

@@ -0,0 +1,28 @@
+import { Observable } from 'rxjs'
+import { webSocket, WebSocketSubject, WebSocketSubjectConfig } from 'rxjs/webSocket'
+import { UpdateReal } from '../sequence-store'
+import { Source } from './source'
+
+export class WebsocketSource<T> implements Source<T> {
+  private websocket$: WebSocketSubject<UpdateReal<T>>
+
+  constructor (
+    readonly url: string,
+  ) {
+    const fullConfig: WebSocketSubjectConfig<UpdateReal<T>> = {
+      url,
+      openObserver: {
+        next: () => console.log('WebSocket connection open'),
+      },
+      closeObserver: {
+        next: () => console.log('WebSocket connection closed'),
+      },
+      closingObserver: {
+        next: () => console.log('Websocket subscription cancelled, websocket closing'),
+      },
+    }
+    this.websocket$ = webSocket(fullConfig)
+  }
+
+  watch$ (): Observable<UpdateReal<T>> { return this.websocket$.asObservable() }
+}

client/lib/store.ts

@@ -0,0 +1,48 @@
+import { from, Observable } from 'rxjs'
+import { applyPatch, Operation } from 'fast-json-patch'
+import { observable, runInAction } from 'mobx'
+import { toStream } from 'mobx-utils'
+
+export class Store<T extends object> {
+  private o: { data: T }
+
+  constructor (data: T) {
+    this.o = observable({ data })
+  }
+
+  get peek (): T { return this.o.data }
+
+  watch$ (): Observable<T>
+  watch$<P1 extends keyof T> (p1: P1): Observable<T[P1]>
+  watch$<P1 extends keyof T, P2 extends keyof T[P1]> (p1: P1, p2: P2): Observable<T[P1][P2]>
+  watch$<P1 extends keyof T, P2 extends keyof T[P1], P3 extends keyof T[P1][P2]> (p1: P1, p2: P2, p3: P3): Observable<T[P1][P2][P3]>
+  watch$<P1 extends keyof T, P2 extends keyof T[P1], P3 extends keyof T[P1][P2], P4 extends keyof T[P1][P2][P3]> (p1: P1, p2: P2, p3: P3, p4: P4): Observable<T[P1][P2][P3][P4]>
+  watch$<P1 extends keyof T, P2 extends keyof T[P1], P3 extends keyof T[P1][P2], P4 extends keyof T[P1][P2][P3], P5 extends keyof T[P1][P2][P3][P4]> (p1: P1, p2: P2, p3: P3, p4: P4, p5: P5): Observable<T[P1][P2][P3][P4][P5]>
+  watch$<P1 extends keyof T, P2 extends keyof T[P1], P3 extends keyof T[P1][P2], P4 extends keyof T[P1][P2][P3], P5 extends keyof T[P1][P2][P3][P4], P6 extends keyof T[P1][P2][P3][P4][P5]> (p1: P1, p2: P2, p3: P3, p4: P4, p5: P5, p6: P6): Observable<T[P1][P2][P3][P4][P5][P6]>
+  watch$ (...args: (string | number)[]): Observable<any> {
+    return from(toStream(() => this.peekAccess(...args), true))
+  }
+
+  set (data: T): void {
+    runInAction(() => this.o.data = data)
+  }
+
+  applyPatchDocument (patch: Operation[]): { oldDocument: T, newDocument: T } {
+    const oldDocument = this.o.data
+    const newDocument = patchDocument(patch, oldDocument)
+    this.set(newDocument)
+    return { oldDocument, newDocument }
+  }
+
+  private peekAccess (...args: (string | number)[]): any {
+    try {
+      return args.reduce((acc, next) => (acc as any)[`${next}`], this.o.data)
+    } catch (e) {
+      return undefined
+    }
+  }
+}
+
+export function patchDocument<T> (patch: Operation[], doc: T): T {
+  return applyPatch(doc, patch, true, false).newDocument
+}