DbAdapters finalization (#65)

Signed-off-by: Andrey Platov <andrey@hardcoreeng.com>

dev/client-resources/src/connection.ts

@@ -14,8 +14,10 @@
 //
 
 import type { Tx, Storage, Ref, Doc, Class, DocumentQuery, FindResult, FindOptions, TxHander, ServerStorage } from '@anticrm/core'
-import { createInMemoryAdapter } from '@anticrm/dev-storage'
+import { DOMAIN_TX } from '@anticrm/core'
+import { createInMemoryAdapter, createInMemoryTxAdapter } from '@anticrm/dev-storage'
 import { createServerStorage } from '@anticrm/server-core'
+import type { DbConfiguration } from '@anticrm/server-core'
 
 class ServerStorageWrapper implements Storage {
   constructor (private readonly storage: ServerStorage, private readonly handler: TxHander) {}
@@ -31,6 +33,23 @@ class ServerStorageWrapper implements Storage {
 }
 
 export async function connect (handler: (tx: Tx) => void): Promise<Storage> {
-  const serverStorage = await createServerStorage(createInMemoryAdapter, '', '')
+  const conf: DbConfiguration = {
+    domains: {
+      [DOMAIN_TX]: 'InMemoryTx'
+    },
+    defaultAdapter: 'InMemory',
+    adapters: {
+      InMemoryTx: {
+        factory: createInMemoryTxAdapter,
+        url: ''
+      },
+      InMemory: {
+        factory: createInMemoryAdapter,
+        url: ''
+      }
+    },
+    workspace: ''
+  }
+  const serverStorage = await createServerStorage(conf)
   return new ServerStorageWrapper(serverStorage, handler)
 }

dev/server/src/server.ts

@@ -14,9 +14,11 @@
 // limitations under the License.
 //
 
+import { DOMAIN_TX } from '@anticrm/core'
 import { start as startJsonRpc } from '@anticrm/server-ws'
-import { createInMemoryAdapter } from '@anticrm/dev-storage'
+import { createInMemoryAdapter, createInMemoryTxAdapter } from '@anticrm/dev-storage'
 import { createServerStorage } from '@anticrm/server-core'
+import type { DbConfiguration } from '@anticrm/server-core'
 
 import { addLocation } from '@anticrm/platform'
 import { serverChunterId } from '@anticrm/server-chunter'
@@ -27,5 +29,24 @@ import { serverChunterId } from '@anticrm/server-chunter'
 export async function start (port: number, host?: string): Promise<void> {
   addLocation(serverChunterId, () => import('@anticrm/server-chunter-resources'))
 
-  startJsonRpc(() => createServerStorage(createInMemoryAdapter, '', ''), port, host)
+  startJsonRpc(() => {
+    const conf: DbConfiguration = {
+      domains: {
+        [DOMAIN_TX]: 'InMemoryTx'
+      },
+      defaultAdapter: 'InMemory',
+      adapters: {
+        InMemoryTx: {
+          factory: createInMemoryTxAdapter,
+          url: ''
+        },
+        InMemory: {
+          factory: createInMemoryAdapter,
+          url: ''
+        }
+      },
+      workspace: ''
+    }
+    return createServerStorage(conf)
+  }, port, host)
 }

dev/storage/src/index.ts

@@ -14,4 +14,4 @@
 // limitations under the License.
 //
 
-export { createInMemoryAdapter } from './storage'
+export * from './storage'

dev/storage/src/storage.ts

@@ -14,37 +14,54 @@
 //
 
 import type { Tx, Ref, Doc, Class, DocumentQuery, FindResult, FindOptions } from '@anticrm/core'
-import { ModelDb, TxDb, Hierarchy, DOMAIN_TX } from '@anticrm/core'
-import type { DbAdapter } from '@anticrm/server-core'
+import { ModelDb, TxDb, Hierarchy } from '@anticrm/core'
+import type { DbAdapter, TxAdapter } from '@anticrm/server-core'
 
 import * as txJson from './model.tx.json'
 
-const txes = txJson as unknown as Tx[]
+class InMemoryTxAdapter implements TxAdapter {
+  private readonly txdb: TxDb
+
+  constructor (hierarchy: Hierarchy) {
+    this.txdb = new TxDb(hierarchy)
+  }
+
+  async findAll<T extends Doc> (_class: Ref<Class<T>>, query: DocumentQuery<T>, options?: FindOptions<T>): Promise<FindResult<T>> {
+    return await this.txdb.findAll(_class, query, options)
+  }
+
+  tx (tx: Tx): Promise<void> {
+    return this.txdb.tx(tx)
+  }
+
+  async init (model: Tx[]): Promise<void> {
+    for (const tx of model) {
+      await this.txdb.tx(tx)
+    }
+  }
+
+  async getModel (): Promise<Tx[]> {
+    return txJson as unknown as Tx[]
+  }
+}
 
 class InMemoryAdapter implements DbAdapter {
-  constructor (
-    private readonly hierarchy: Hierarchy,
-    private readonly txdb: TxDb,
-    private readonly modeldb: ModelDb
-  ) {}
+  private readonly modeldb: TxDb
 
-  async findAll<T extends Doc> (
-    _class: Ref<Class<T>>,
-    query: DocumentQuery<T>,
-    options?: FindOptions<T>
-  ): Promise<FindResult<T>> {
-    const domain = this.hierarchy.getDomain(_class)
-    if (domain === DOMAIN_TX) return await this.txdb.findAll(_class, query, options)
+  constructor (hierarchy: Hierarchy) {
+    this.modeldb = new ModelDb(hierarchy)
+  }
+
+  async findAll<T extends Doc> (_class: Ref<Class<T>>, query: DocumentQuery<T>, options?: FindOptions<T>): Promise<FindResult<T>> {
     return await this.modeldb.findAll(_class, query, options)
   }
 
   async tx (tx: Tx): Promise<void> {
-    await Promise.all([this.modeldb.tx(tx), this.txdb.tx(tx)])
+    return await this.modeldb.tx(tx)
   }
 
-  async init (): Promise<void> {
-    for (const tx of txes) {
-      await this.txdb.tx(tx)
+  async init (model: Tx[]): Promise<void> {
+    for (const tx of model) {
       await this.modeldb.tx(tx)
     }
   }
@@ -53,8 +70,13 @@ class InMemoryAdapter implements DbAdapter {
 /**
  * @public
  */
-export async function createInMemoryAdapter (hierarchy: Hierarchy, url: string, db: string): Promise<[DbAdapter, Tx[]]> {
-  const txdb = new TxDb(hierarchy)
-  const modeldb = new ModelDb(hierarchy)
-  return [new InMemoryAdapter(hierarchy, txdb, modeldb), txes]
+export async function createInMemoryTxAdapter (hierarchy: Hierarchy, url: string, workspace: string): Promise<TxAdapter> {
+  return new InMemoryTxAdapter(hierarchy)
+}
+
+/**
+ * @public
+ */
+export async function createInMemoryAdapter (hierarchy: Hierarchy, url: string, db: string): Promise<DbAdapter> {
+  return new InMemoryAdapter(hierarchy)
 }

server/core/src/index.ts

@@ -14,8 +14,8 @@
 // limitations under the License.
 //
 
-import type { Doc, Tx, TxCreateDoc, Ref, Class, ServerStorage, DocumentQuery, FindOptions, FindResult, Storage, Account } from '@anticrm/core'
-import core, { Hierarchy, TxFactory, ModelDb, DOMAIN_MODEL } from '@anticrm/core'
+import type { Doc, Tx, TxCreateDoc, Ref, Class, ServerStorage, DocumentQuery, FindOptions, FindResult, Storage, Account, Domain, TxCUD } from '@anticrm/core'
+import core, { Hierarchy, TxFactory, DOMAIN_TX } from '@anticrm/core'
 import type { Resource, Plugin } from '@anticrm/platform'
 import { getResource, plugin } from '@anticrm/platform'
 
@@ -59,73 +59,130 @@ export class Triggers {
  * @public
  */
 export interface DbAdapter extends Storage {
-  init: () => Promise<void>
+  /**
+   * Method called after hierarchy is ready to use.
+   */
+  init: (model: Tx[]) => Promise<void>
 }
 
 /**
  * @public
  */
-export type DbAdapterFactory = (hierarchy: Hierarchy, url: string, db: string) => Promise<[DbAdapter, Tx[]]>
+export interface TxAdapter extends DbAdapter {
+  getModel: () => Promise<Tx[]>
+}
+
+/**
+ * @public
+ */
+export type DbAdapterFactory = (hierarchy: Hierarchy, url: string, db: string) => Promise<DbAdapter>
+
+/**
+ * @public
+ */
+export interface DbAdapterConfiguration {
+  factory: DbAdapterFactory
+  url: string
+}
+
+/**
+ * @public
+ */
+export interface DbConfiguration {
+  adapters: Record<string, DbAdapterConfiguration>
+  domains: Record<string, string>
+  defaultAdapter: string
+  workspace: string
+}
 
 class TServerStorage implements ServerStorage {
   constructor (
-    private readonly dbAdapter: Storage,
+    private readonly domains: Record<string, string>,
+    private readonly defaultAdapter: string,
+    private readonly adapters: Map<string, DbAdapter>,
     private readonly hierarchy: Hierarchy,
-    private readonly triggers: Triggers,
-    private readonly modeldb: ModelDb
+    private readonly triggers: Triggers
   ) {
   }
 
+  private getAdapter (domain: Domain): DbAdapter {
+    const name = this.domains[domain] ?? this.defaultAdapter
+    const adapter = this.adapters.get(name)
+    if (adapter === undefined) {
+      throw new Error('adapter not provided: ' + name)
+    }
+    return adapter
+  }
+
+  private routeTx (tx: Tx): Promise<void> {
+    if (this.hierarchy.isDerived(tx._class, core.class.TxCUD)) {
+      const txCUD = tx as TxCUD<Doc>
+      const domain = this.hierarchy.getDomain(txCUD.objectClass)
+      return this.getAdapter(domain).tx(txCUD)
+    } else {
+      throw new Error('not implemented (not derived from TxCUD)')
+    }
+  }
+
   async findAll<T extends Doc> (
     clazz: Ref<Class<T>>,
     query: DocumentQuery<T>,
     options?: FindOptions<T>
   ): Promise<FindResult<T>> {
     const domain = this.hierarchy.getDomain(clazz)
-    console.log('findAll', clazz, domain, query)
-    if (domain === DOMAIN_MODEL) return await this.modeldb.findAll(clazz, query, options)
-    return await this.dbAdapter.findAll(clazz, query, options)
+    return await this.getAdapter(domain).findAll(clazz, query, options)
   }
 
   async tx (tx: Tx): Promise<Tx[]> {
+    // maintain hiearachy and triggers
     if (tx.objectSpace === core.space.Model) {
       this.hierarchy.tx(tx)
-      await this.modeldb.tx(tx)
       await this.triggers.tx(tx)
-      return [] // we do not apply triggers on model changes?
-    } else {
-      await this.dbAdapter.tx(tx)
+    }
+
+    // store tx
+    await this.getAdapter(DOMAIN_TX).tx(tx)
+
+    // store object
+    await this.routeTx(tx)
     const derived = await this.triggers.apply(tx.modifiedBy, tx)
     for (const tx of derived) {
-        await this.dbAdapter.tx(tx) // triggers does not generate changes to model objects?
+      await this.routeTx(tx)
     }
     return derived
   }
 }
-}
 
 /**
  * @public
  */
-export async function createServerStorage (factory: DbAdapterFactory, url: string, db: string): Promise<ServerStorage> {
+export async function createServerStorage (conf: DbConfiguration): Promise<ServerStorage> {
   const hierarchy = new Hierarchy()
-  const model = new ModelDb(hierarchy)
   const triggers = new Triggers()
+  const adapters = new Map<string, DbAdapter>()
 
-  const [dbAdapter, txes] = await factory(hierarchy, url, db)
-
-  for (const tx of txes) {
-    hierarchy.tx(tx)
+  for (const key in conf.adapters) {
+    const adapterConf = conf.adapters[key]
+    adapters.set(key, await adapterConf.factory(hierarchy, adapterConf.url, conf.workspace))
   }
 
-  for (const tx of txes) {
-    await model.tx(tx)
+  const txAdapter = adapters.get(conf.domains[DOMAIN_TX]) as TxAdapter
+  if (txAdapter === undefined) {
+    console.log('no txadapter found')
+  }
+
+  const model = await txAdapter.getModel()
+
+  for (const tx of model) {
+    hierarchy.tx(tx)
     await triggers.tx(tx)
   }
 
-  await dbAdapter.init()
+  for (const [, adapter] of adapters) {
+    await adapter.init(model)
+  }
 
-  return new TServerStorage(dbAdapter, hierarchy, triggers, model)
+  return new TServerStorage(conf.domains, conf.defaultAdapter, adapters, hierarchy, triggers)
 }
 
 /**

server/mongo/src/index.ts

@@ -14,4 +14,4 @@
 // limitations under the License.
 //
 
-export { createMongoAdapter } from './storage'
+export * from './storage'

server/mongo/src/storage.ts

@@ -15,7 +15,7 @@
 
 import type { Tx, Ref, Doc, Class, DocumentQuery, FindResult, FindOptions, TxCreateDoc } from '@anticrm/core'
 import core, { TxProcessor, Hierarchy, DOMAIN_TX, SortingOrder } from '@anticrm/core'
-import type { DbAdapter } from '@anticrm/server-core'
+import type { DbAdapter, TxAdapter } from '@anticrm/server-core'
 
 import { MongoClient, Db, Filter, Document, Sort } from 'mongodb'
 
@@ -27,28 +27,16 @@ function translateDoc (doc: Doc): Document {
   return doc as Document
 }
 
-class MongoAdapter extends TxProcessor implements DbAdapter {
+abstract class MongoAdapterBase extends TxProcessor {
   constructor (
-    private readonly db: Db,
-    private readonly hierarchy: Hierarchy
+    protected readonly db: Db,
+    protected readonly hierarchy: Hierarchy
   ) {
     super()
   }
 
   async init (): Promise<void> {}
 
-  override async tx (tx: Tx): Promise<void> {
-    const p1 = this.db.collection(DOMAIN_TX).insertOne(translateDoc(tx))
-    const p2 = super.tx(tx)
-    await Promise.all([p1, p2])
-  }
-
-  protected override async txCreateDoc (tx: TxCreateDoc<Doc>): Promise<void> {
-    const doc = TxProcessor.createDoc2Doc(tx)
-    const domain = this.hierarchy.getDomain(doc._class)
-    await this.db.collection(domain).insertOne(translateDoc(doc))
-  }
-
   async findAll<T extends Doc> (
     _class: Ref<Class<T>>,
     query: DocumentQuery<T>,
@@ -70,13 +58,42 @@ class MongoAdapter extends TxProcessor implements DbAdapter {
   }
 }
 
+class MongoAdapter extends MongoAdapterBase {
+  protected override async txCreateDoc (tx: TxCreateDoc<Doc>): Promise<void> {
+    const doc = TxProcessor.createDoc2Doc(tx)
+    const domain = this.hierarchy.getDomain(doc._class)
+    console.log('mongo', domain, doc)
+    await this.db.collection(domain).insertOne(translateDoc(doc))
+  }
+}
+
+class MongoTxAdapter extends MongoAdapterBase implements TxAdapter {
+  override async tx (tx: Tx): Promise<void> {
+    console.log('mongotx', tx)
+    await this.db.collection(DOMAIN_TX).insertOne(translateDoc(tx))
+  }
+
+  async getModel (): Promise<Tx[]> {
+    return await this.db.collection(DOMAIN_TX).find<Tx>({ objectSpace: core.space.Model }).sort({ _id: 1 }).toArray()
+  }
+}
+
 /**
  * @public
  */
-export async function createMongoAdapter (hierarchy: Hierarchy, url: string, dbName: string): Promise<[DbAdapter, Tx[]]> {
+export async function createMongoAdapter (hierarchy: Hierarchy, url: string, dbName: string): Promise<DbAdapter> {
   const client = new MongoClient(url)
   await client.connect()
   const db = client.db(dbName)
-  const txes = await db.collection(DOMAIN_TX).find<Tx>({ objectSpace: core.space.Model }).sort({ _id: 1 }).toArray()
-  return [new MongoAdapter(db, hierarchy), txes]
+  return new MongoAdapter(db, hierarchy)
+}
+
+/**
+ * @public
+ */
+export async function createMongoTxAdapter (hierarchy: Hierarchy, url: string, dbName: string): Promise<TxAdapter> {
+  const client = new MongoClient(url)
+  await client.connect()
+  const db = client.db(dbName)
+  return new MongoTxAdapter(db, hierarchy)
 }

server/server/src/server.ts

@@ -14,9 +14,11 @@
 // limitations under the License.
 //
 
+import { DOMAIN_TX } from '@anticrm/core'
 import { start as startJsonRpc } from '@anticrm/server-ws'
-import { createMongoAdapter } from '@anticrm/mongo'
+import { createMongoAdapter, createMongoTxAdapter } from '@anticrm/mongo'
 import { createServerStorage } from '@anticrm/server-core'
+import type { DbConfiguration } from '@anticrm/server-core'
 
 import { addLocation } from '@anticrm/platform'
 import { serverChunterId } from '@anticrm/server-chunter'
@@ -27,5 +29,24 @@ import { serverChunterId } from '@anticrm/server-chunter'
 export async function start (dbUrl: string, port: number, host?: string): Promise<void> {
   addLocation(serverChunterId, () => import('@anticrm/server-chunter-resources'))
 
-  startJsonRpc((workspace: string) => createServerStorage(createMongoAdapter, dbUrl, workspace), port, host)
+  startJsonRpc((workspace: string) => {
+    const conf: DbConfiguration = {
+      domains: {
+        [DOMAIN_TX]: 'MongoTx'
+      },
+      defaultAdapter: 'Mongo',
+      adapters: {
+        MongoTx: {
+          factory: createMongoTxAdapter,
+          url: dbUrl
+        },
+        Mongo: {
+          factory: createMongoAdapter,
+          url: dbUrl
+        }
+      },
+      workspace
+    }
+    return createServerStorage(conf)
+  }, port, host)
 }