Allow model upgrades (#561)

Signed-off-by: Andrey Sobolev <haiodo@gmail.com>

dev/docker-compose.yaml

@@ -45,8 +45,7 @@ services:
       - mongodb
       - minio
       - elastic
-      - server
-      - upload
+      - transactor
     ports:
       - 8081:8080
     environment:

dev/tool/package.json

@@ -49,6 +49,7 @@
     "@anticrm/client-resources": "~0.6.4",
     "ws": "^8.2.0",
     "@anticrm/client": "~0.6.1",
-    "@anticrm/platform": "~0.6.5"
+    "@anticrm/platform": "~0.6.5",
+    "@anticrm/model": "~0.6.0"
   }
 }

dev/tool/src/connect.ts

@@ -1,24 +1,18 @@
 
 import client from '@anticrm/client'
 import clientResources from '@anticrm/client-resources'
-import core, { TxOperations } from '@anticrm/core'
+import { Client } from '@anticrm/core'
 import { setMetadata } from '@anticrm/platform'
 import { encode } from 'jwt-simple'
 
 // eslint-disable-next-line
 const WebSocket = require('ws')
 
-export async function connect (transactorUrl: string, workspace: string): Promise<{ connection: TxOperations, close: () => Promise<void>}> {
+export async function connect (transactorUrl: string, workspace: string): Promise<Client> {
   console.log('connecting to transactor...')
   const token = encode({ email: 'anticrm@hc.engineering', workspace }, 'secret')
 
   // We need to override default factory with 'ws' one.
   setMetadata(client.metadata.ClientSocketFactory, (url) => new WebSocket(url))
-  const connection = await (await clientResources()).function.GetClient(token, transactorUrl)
-  return {
-    connection: new TxOperations(connection, core.account.System),
-    close: async () => {
-      await connection.close()
-    }
-  }
+  return await (await clientResources()).function.GetClient(token, transactorUrl)
 }

dev/tool/src/index.ts

@@ -18,13 +18,13 @@ import {
   ACCOUNT_DB, assignWorkspace, createAccount, createWorkspace, dropAccount, dropWorkspace, getAccount, listWorkspaces
 } from '@anticrm/account'
 import contact, { combineName } from '@anticrm/contact'
-import core from '@anticrm/core'
+import core, { TxOperations } from '@anticrm/core'
 import { program } from 'commander'
 import { Client } from 'minio'
 import { Db, MongoClient } from 'mongodb'
 import { connect } from './connect'
 import { clearTelegramHistory } from './telegram'
-import { dumpWorkspace, initWorkspace, upgradeWorkspace } from './workspace'
+import { dumpWorkspace, initWorkspace, restoreWorkspace, upgradeWorkspace } from './workspace'
 
 const mongodbUri = process.env.MONGO_URL
 if (mongodbUri === undefined) {
@@ -103,24 +103,25 @@ program
       await assignWorkspace(db, email, workspace)
 
       console.log('connecting to transactor...')
-      const { connection, close } = await connect(transactorUrl, workspace)
+      const connection = await connect(transactorUrl, workspace)
+      const ops = new TxOperations(connection, core.account.System)
 
       const name = combineName(account.first, account.last)
 
       console.log('create user in target workspace...')
-      const employee = await connection.createDoc(contact.class.Employee, contact.space.Employee, {
+      const employee = await ops.createDoc(contact.class.Employee, contact.space.Employee, {
         name,
         city: 'Mountain View',
         channels: []
       })
 
       console.log('create account in target workspace...')
-      await connection.createDoc(contact.class.EmployeeAccount, core.space.Model, {
+      await ops.createDoc(contact.class.EmployeeAccount, core.space.Model, {
         email,
         employee,
         name
       })
-      await close()
+      await connection.close()
     })
   })
 
@@ -181,10 +182,17 @@ program
   })
 
 program
-  .command('dump-workspace <name> <fileName>')
+  .command('dump-workspace <workspace> <dirName>')
   .description('dump workspace transactions and minio resources')
-  .action(async (workspace, fileName, cmd) => {
-    return await dumpWorkspace(mongodbUri, workspace, fileName, minio)
+  .action(async (workspace, dirName, cmd) => {
+    return await dumpWorkspace(mongodbUri, workspace, dirName, minio)
+  })
+
+program
+  .command('restore-workspace <workspace> <dirName>')
+  .description('restore workspace transactions and minio resources from previous dump.')
+  .action(async (workspace, dirName, cmd) => {
+    return await restoreWorkspace(mongodbUri, workspace, dirName, minio)
   })
 
 program

dev/tool/src/upgrade.ts

@@ -0,0 +1,72 @@
+import {
+  Doc,
+  DocumentQuery,
+  Domain,
+  FindOptions,
+  isOperator, SortingOrder
+} from '@anticrm/core'
+import { MigrationClient, MigrateUpdate, MigrationResult } from '@anticrm/model'
+import { Db, Document, Filter, Sort, UpdateFilter } from 'mongodb'
+
+/**
+ * Upgrade client implementation.
+ */
+export class MigrateClientImpl implements MigrationClient {
+  constructor (readonly db: Db) {
+  }
+
+  private translateQuery<T extends Doc>(query: DocumentQuery<T>): Filter<Document> {
+    const translated: any = {}
+    for (const key in query) {
+      const value = (query as any)[key]
+      if (value !== null && typeof value === 'object') {
+        const keys = Object.keys(value)
+        if (keys[0] === '$like') {
+          const pattern = value.$like as string
+          translated[key] = {
+            $regex: `^${pattern.split('%').join('.*')}$`,
+            $options: 'i'
+          }
+          continue
+        }
+      }
+      translated[key] = value
+    }
+    return translated
+  }
+
+  async find<T extends Doc>(
+    domain: Domain,
+    query: DocumentQuery<T>,
+    options?: FindOptions<T> | undefined
+  ): Promise<T[]> {
+    let cursor = this.db.collection(domain).find<T>(this.translateQuery(query))
+    if (options?.limit !== undefined) {
+      cursor = cursor.limit(options.limit)
+    }
+    if (options !== null && options !== undefined) {
+      if (options.sort !== undefined) {
+        const sort: Sort = {}
+        for (const key in options.sort) {
+          const order = options.sort[key] === SortingOrder.Ascending ? 1 : -1
+          sort[key] = order
+        }
+        cursor = cursor.sort(sort)
+      }
+    }
+    return await cursor.toArray()
+  }
+
+  async update<T extends Doc>(domain: Domain, query: DocumentQuery<T>, operations: MigrateUpdate<T>): Promise<MigrationResult> {
+    if (isOperator(operations)) {
+      const result = await this.db
+        .collection(domain)
+        .updateMany(this.translateQuery(query), { ...operations } as unknown as UpdateFilter<any>)
+
+      return { matched: result.matchedCount, updated: result.modifiedCount }
+    } else {
+      const result = await this.db.collection(domain).updateMany(this.translateQuery(query), { $set: operations })
+      return { matched: result.matchedCount, updated: result.modifiedCount }
+    }
+  }
+}

dev/tool/src/workspace.ts

@@ -16,13 +16,14 @@
 
 import contact from '@anticrm/contact'
 import core, { DOMAIN_TX, Tx } from '@anticrm/core'
-import builder from '@anticrm/model-all'
+import builder, { migrateOperations } from '@anticrm/model-all'
 import { existsSync } from 'fs'
-import { mkdir, writeFile } from 'fs/promises'
-import { BucketItem, Client } from 'minio'
+import { mkdir, readFile, writeFile } from 'fs/promises'
+import { BucketItem, Client, ItemBucketMetadata } from 'minio'
 import { Document, MongoClient } from 'mongodb'
 import { join } from 'path'
 import { connect } from './connect'
+import { MigrateClientImpl } from './upgrade'
 
 const txes = JSON.parse(JSON.stringify(builder.getTxes())) as Tx[]
 
@@ -46,11 +47,11 @@ export async function initWorkspace (mongoUrl: string, dbName: string, transacto
     console.log('creating data...')
     const data = txes.filter((tx) => tx.objectSpace !== core.space.Model)
 
-    const { connection, close } = await connect(transactorUrl, dbName)
+    const connection = await connect(transactorUrl, dbName)
     for (const tx of data) {
       await connection.tx(tx)
     }
-    await close()
+    await connection.close()
 
     console.log('create minio bucket')
     if (!(await minio.bucketExists(dbName))) {
@@ -88,11 +89,38 @@ export async function upgradeWorkspace (
     const model = txes.filter((tx) => tx.objectSpace === core.space.Model)
     const insert = await db.collection(DOMAIN_TX).insertMany(model as Document[])
     console.log(`${insert.insertedCount} model transactions inserted.`)
+
+    const migrateClient = new MigrateClientImpl(db)
+    for (const op of migrateOperations) {
+      await op.migrate(migrateClient)
+    }
+
+    console.log('Apply upgrade operations')
+
+    const connection = await connect(transactorUrl, dbName)
+    for (const op of migrateOperations) {
+      await op.upgrade(connection)
+    }
+
+    await connection.close()
   } finally {
     await client.close()
   }
 }
 
+interface CollectionInfo {
+  name: string
+  file: string
+}
+
+type MinioWorkspaceItem = BucketItem & { metaData: ItemBucketMetadata }
+
+interface WorkspaceInfo {
+  version: string
+  collections: CollectionInfo[]
+  minioData: MinioWorkspaceItem[]
+}
+
 /**
  * @public
  */
@@ -104,29 +132,34 @@ export async function dumpWorkspace (mongoUrl: string, dbName: string, fileName:
 
     console.log('dumping transactions...')
 
-    const dbTxes = await db.collection<Tx>(DOMAIN_TX).find().toArray()
-    await writeFile(fileName + '.tx.json', JSON.stringify(dbTxes, undefined, 2))
+    if (!existsSync(fileName)) {
+      await mkdir(fileName, { recursive: true })
+    }
+
+    const workspaceInfo: WorkspaceInfo = {
+      version: '0.6.0',
+      collections: [],
+      minioData: []
+    }
+    const collections = await db.collections()
+    for (const c of collections) {
+      const docs = await c.find().toArray()
+      workspaceInfo.collections.push({ name: c.collectionName, file: c.collectionName + '.json' })
+      await writeFile(fileName + c.collectionName + '.json', JSON.stringify(docs, undefined, 2))
+    }
 
     console.log('Dump minio objects')
     if (await minio.bucketExists(dbName)) {
-      const minioData: BucketItem[] = []
-      const list = await minio.listObjects(dbName, undefined, true)
-      await new Promise((resolve) => {
-        list.on('data', (data) => {
-          minioData.push(data)
-        })
-        list.on('end', () => {
-          resolve(null)
-        })
-      })
+      workspaceInfo.minioData.push(...(await listMinioObjects(minio, dbName)))
       const minioDbLocation = fileName + '.minio'
       if (!existsSync(minioDbLocation)) {
         await mkdir(minioDbLocation)
       }
-      await writeFile(fileName + '.minio.json', JSON.stringify(minioData, undefined, 2))
-      for (const d of minioData) {
+      for (const d of workspaceInfo.minioData) {
+        const stat = await minio.statObject(dbName, d.name)
         const data = await minio.getObject(dbName, d.name)
         const allData = []
+        d.metaData = stat.metaData
         let chunk
         while ((chunk = data.read()) !== null) {
           allData.push(chunk)
@@ -134,6 +167,64 @@ export async function dumpWorkspace (mongoUrl: string, dbName: string, fileName:
         await writeFile(join(minioDbLocation, d.name), allData.join(''))
       }
     }
+
+    await writeFile(fileName + '.workspace.json', JSON.stringify(workspaceInfo, undefined, 2))
+  } finally {
+    await client.close()
+  }
+}
+
+async function listMinioObjects (minio: Client, dbName: string): Promise<MinioWorkspaceItem[]> {
+  const items: MinioWorkspaceItem[] = []
+  const list = await minio.listObjects(dbName, undefined, true)
+  await new Promise((resolve) => {
+    list.on('data', (data) => {
+      items.push({ ...data, metaData: {} })
+    })
+    list.on('end', () => {
+      resolve(null)
+    })
+  })
+  return items
+}
+
+export async function restoreWorkspace (mongoUrl: string, dbName: string, fileName: string, minio: Client): Promise<void> {
+  const client = new MongoClient(mongoUrl)
+  try {
+    await client.connect()
+    const db = client.db(dbName)
+
+    console.log('dumping transactions...')
+
+    const workspaceInfo = JSON.parse((await readFile(fileName + '.workspace.json')).toString()) as WorkspaceInfo
+
+    // Drop existing collections
+
+    const cols = await db.collections()
+    for (const c of cols) {
+      await db.dropCollection(c.collectionName)
+    }
+    // Restore collections.
+    for (const c of workspaceInfo.collections) {
+      const collection = db.collection(c.name)
+      await collection.deleteMany({})
+      const data = JSON.parse((await readFile(fileName + c.name + '.json')).toString()) as Document[]
+      await collection.insertMany(data)
+    }
+
+    console.log('Restore minio objects')
+    if (await minio.bucketExists(dbName)) {
+      const objectNames = (await listMinioObjects(minio, dbName)).map(i => i.name)
+      await minio.removeObjects(dbName, objectNames)
+      await minio.removeBucket(dbName)
+    }
+    await minio.makeBucket(dbName, 'k8s')
+
+    const minioDbLocation = fileName + '.minio'
+    for (const d of workspaceInfo.minioData) {
+      const data = await readFile(join(minioDbLocation, d.name))
+      await minio.putObject(dbName, d.name, data, d.size, d.metaData)
+    }
   } finally {
     await client.close()
   }

models/all/package.json

@@ -45,7 +45,7 @@
     "@anticrm/model-server-recruit": "~0.6.0",
     "@anticrm/model-server-view": "~0.6.0",
     "@anticrm/model-activity": "~0.6.0",
-    "@anticrm/model-attachment": "~0.6.0"
-
+    "@anticrm/model-attachment": "~0.6.0",
+    "@anticrm/core": "~0.6.13"
   }
 }

models/all/src/index.ts

@@ -58,3 +58,6 @@ serverViewModel(builder)
 createDemo(builder)
 
 export default builder
+
+// Export upgrade procedures
+export { migrateOperations } from './migration'

models/all/src/migration.ts

@@ -0,0 +1,22 @@
+//
+// Copyright © 2020 Anticrm Platform Contributors.
+//
+// Licensed under the Eclipse Public License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License. You may
+// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+import { MigrateOperation } from '@anticrm/model'
+
+// Import migrate operations.
+import { taskOperation } from '@anticrm/model-task'
+import { attachmentOperation } from '@anticrm/model-attachment'
+
+export const migrateOperations: MigrateOperation[] = [taskOperation, attachmentOperation]

models/attachment/src/index.ts

@@ -23,6 +23,8 @@ import activity from '@anticrm/activity'
 import view from '@anticrm/model-view'
 import attachment from './plugin'
 
+export { attachmentOperation } from './migration'
+
 export const DOMAIN_ATTACHMENT = 'attachment' as Domain
 
 @Model(attachment.class.Attachment, core.class.AttachedDoc, DOMAIN_ATTACHMENT)

models/attachment/src/migration.ts

@@ -0,0 +1,45 @@
+//
+// Copyright © 2020, 2021 Anticrm Platform Contributors.
+//
+// Licensed under the Eclipse Public License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License. You may
+// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+import { Class, Doc, DOMAIN_TX, Ref, TxCUD } from '@anticrm/core'
+import { MigrateOperation, MigrationClient, MigrationUpgradeClient } from '@anticrm/model'
+import { DOMAIN_ATTACHMENT } from './index'
+import attachment from './plugin'
+
+export const attachmentOperation: MigrateOperation = {
+  async migrate (client: MigrationClient): Promise<void> {
+    console.log('Attachments: Upgrading attachments.')
+    // Replace attachment class
+    const attachResult = await client.update(DOMAIN_ATTACHMENT, { _class: 'chunter:class:Attachment' as Ref<Class<Doc>> }, { _class: attachment.class.Attachment as Ref<Class<Doc>> })
+    if (attachResult.updated > 0) {
+      console.log(`Attachments: Update ${attachResult.updated} Attachment objects`)
+    }
+
+    // Update transactions.
+    const txResult = await client.update<TxCUD<Doc>>(DOMAIN_TX, { objectClass: 'chunter:class:Attachment' as Ref<Class<Doc>> }, { objectClass: attachment.class.Attachment as Ref<Class<Doc>> })
+
+    if (txResult.updated > 0) {
+      console.log(`Attachments: Update ${txResult.updated} Transactions`)
+    }
+
+    const txResult2 = await client.update<TxCUD<Doc>>(DOMAIN_TX, { 'tx.objectClass': 'chunter:class:Attachment' as Ref<Class<Doc>> }, { 'tx.objectClass': attachment.class.Attachment as Ref<Class<Doc>> })
+
+    if (txResult2.updated > 0) {
+      console.log(`Attachments: Update ${txResult.updated} Transactions`)
+    }
+  },
+  async upgrade (client: MigrationUpgradeClient): Promise<void> {
+  }
+}

models/task/src/index.ts

@@ -144,3 +144,5 @@ export function createModel (builder: Builder): void {
     return await Promise.resolve()
   }).catch((err) => console.error(err))
 }
+
+export { taskOperation } from './migration'

models/task/src/migration.ts

@@ -0,0 +1,55 @@
+//
+// Copyright © 2020, 2021 Anticrm Platform Contributors.
+//
+// Licensed under the Eclipse Public License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License. You may
+// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+import { Doc, TxOperations } from '@anticrm/core'
+import { MigrateOperation, MigrationClient, MigrationUpgradeClient } from '@anticrm/model'
+import core from '@anticrm/model-core'
+import view from '@anticrm/model-view'
+import { createProjectKanban } from '@anticrm/task'
+import task from './plugin'
+
+export const taskOperation: MigrateOperation = {
+  async migrate (client: MigrationClient): Promise<void> {
+
+  },
+  async upgrade (client: MigrationUpgradeClient): Promise<void> {
+    console.log('Task: Performing model upgrades')
+
+    const ops = new TxOperations(client, core.account.System)
+    if (await client.findOne(view.class.Sequence, { attachedTo: task.class.Task }) === undefined) {
+      console.info('Create sequence for default task project.')
+      // We need to create sequence
+      await ops.createDoc(view.class.Sequence, view.space.Sequence, {
+        attachedTo: task.class.Task,
+        sequence: 0
+      })
+    } else {
+      console.log('Task: => sequence is ok')
+    }
+    if (await client.findOne(view.class.Kanban, { attachedTo: task.space.TasksPublic }) === undefined) {
+      console.info('Create kanban for default task project.')
+      await createProjectKanban(task.space.TasksPublic, async (_class, space, data, id) => {
+        const doc = await ops.findOne<Doc>(_class, { _id: id })
+        if (doc === undefined) {
+          await ops.createDoc(_class, space, data, id)
+        } else {
+          await ops.updateDoc(_class, space, id, data)
+        }
+      }).catch((err) => console.error(err))
+    } else {
+      console.log('Task: => public project Kanban is ok')
+    }
+  }
+}

packages/model/src/index.ts

@@ -14,3 +14,4 @@
 //
 
 export * from './dsl'
+export * from './migration'

packages/model/src/migration.ts

@@ -0,0 +1,46 @@
+import { Client, Doc, DocumentQuery, Domain, FindOptions, IncOptions, PushOptions } from '@anticrm/core'
+
+/**
+ * @public
+ */
+export type MigrateUpdate<T extends Doc> = Partial<T> & Omit<PushOptions<T>, '$move'> & IncOptions<T> & {
+  // For any other mongo stuff
+  [key: string]: any
+}
+
+/**
+ * @public
+ */
+export interface MigrationResult {
+  matched: number
+  updated: number
+}
+
+/**
+ * @public
+ * Client to perform model upgrades
+ */
+export interface MigrationClient {
+  // Raw collection operations
+
+  // Raw FIND, allow to find documents inside domain.
+  find: <T extends Doc>(domain: Domain, query: DocumentQuery<T>, options?: Omit<FindOptions<T>, 'lookup'>) => Promise<T[]>
+
+  // Allow to raw update documents inside domain.
+  update: <T extends Doc>(domain: Domain, query: DocumentQuery<T>, operations: MigrateUpdate<T>) => Promise<MigrationResult>
+}
+
+/**
+ * @public
+ */
+export type MigrationUpgradeClient = Client
+
+/**
+ * @public
+ */
+export interface MigrateOperation {
+  // Perform low level migration
+  migrate: (client: MigrationClient) => Promise<void>
+  // Perform high level upgrade operations.
+  upgrade: (client: MigrationUpgradeClient) => Promise<void>
+}