PotentialStorage/src/document-storage/event-sourced-document-storage.ts

import type { BaseEventStorage, StoredEvent } from '../event-storage/base-event-storage.js'
import type { Doc } from '../utils/document-path.js'
import type { DocumentQuery, FindResult } from '../utils/document-query.js'
import { DocumentUpdate, type DocChange } from '../utils/document-update.js'
import type { AnySpec, ChildOf, KeyOf, Spec, ValueAt } from '../utils/schema.js'
import { BaseDocumentStorage } from './base-document-storage.js'

// Types only: children of an event-sourced store are event-sourced stores too (child() creates
// them with Object.create, so they share this class's prototype).
export interface EventSourcedDocumentStorage<S extends Spec = AnySpec> {
  child<K extends KeyOf<S>>(key: K): EventSourcedDocumentStorage<ChildOf<S, K>>
}

/**
 * Documents as event streams: each change to a document is appended to its own stream, and the
 * current state is kept in a projection store (any `BaseDocumentStorage`), which serves all reads
 * and queries.
 *
 * A document's state is its events folded in order. Changes are stored as resolved values
 * (`$inc` becomes `set: { n: 6 }`), so two devices editing the same field concurrently end with
 * the later event's value; edits to different fields both survive.
 *
 * The projection is updated by this store's own writes. Events written elsewhere (another device,
 * arriving through sync) are folded in by `catchUp(key)`, which rebuilds the document from its
 * whole stream; call it when such events arrive.
 */
export class EventSourcedDocumentStorage<S extends Spec = AnySpec> extends BaseDocumentStorage<S> {
  constructor(
    private readonly events: BaseEventStorage,
    private readonly projections: BaseDocumentStorage,
  ) {
    super()
  }

  readMany(paths: (readonly string[])[]): Promise<(Doc | undefined)[]> {
    return this.projections.readMany(paths)
  }

  query(parent: readonly string[], query: DocumentQuery): AsyncIterable<FindResult> {
    return this.projections.query(parent, query)
  }

  async mutate(path: readonly string[], change: (current: Doc | undefined) => DocChange | null): Promise<Doc | undefined> {
    const [current] = await this.projections.readMany([path])
    const resolved = change(current)
    if (!resolved) return current
    await this.events.appendTo(path, [resolved])
    // Apply the change itself (not the state computed above), so concurrent writes in this
    // process each land on the latest projection.
    return this.projections.mutate(path, () => resolved)
  }

  /** Events of this document (or a child), oldest first. */
  history(key?: KeyOf<S>): AsyncIterable<StoredEvent<DocChange>> {
    return this.events.readFrom(this.pathOf(arguments.length > 0, key), {}) as AsyncIterable<StoredEvent<DocChange>>
  }

  /** State right after event `eventId`: `getAt(eventId)` for this document, `getAt(key, eventId)` for a child. */
  async getAt<K extends KeyOf<S>>(...args: [eventId: string] | [key: K, eventId: string]): Promise<ValueAt<S, K> | undefined> {
    const until = args[args.length - 1] as string
    let state: Doc | undefined
    for await (const event of this.events.readFrom(this.pathOf(args.length > 1, args[0]), {})) {
      if (event.id > until) break
      state = DocumentUpdate.apply(state, event.data as DocChange)
    }
    return state as ValueAt<S, K> | undefined
  }

  /** Rebuilds the projection of this document (or a child) from all its events. Returns the state. */
  async catchUp(key?: KeyOf<S>): Promise<Doc | undefined> {
    const path = this.pathOf(arguments.length > 0, key)
    let state: Doc | undefined
    for await (const event of this.events.readFrom(path, {})) state = DocumentUpdate.apply(state, event.data as DocChange)
    await this.projections.mutate(path, () => (state ? { op: 'set', doc: state } : { op: 'delete' }))
    return state
  }
}