Space security (#2694)

Signed-off-by: Denis Bykhov <bykhov.denis@gmail.com>

dev/server/src/server.ts

@@ -42,9 +42,10 @@ async function createNullContentTextAdapter (): Promise<ContentTextAdapter> {
  * @public
  */
 export async function start (port: number, host?: string): Promise<void> {
+  const ctx = new MeasureMetricsContext('server', {})
   startJsonRpc(
-    new MeasureMetricsContext('server', {}),
+    ctx,
-    () => {
+    (ctx) => {
       const conf: DbConfiguration = {
         domains: {
           [DOMAIN_TX]: 'InMemoryTx'
@@ -72,7 +73,7 @@ export async function start (port: number, host?: string): Promise<void> {
         },
         workspace: getWorkspaceId('')
       }
-      return createPipeline(conf, [], false, () => {})
+      return createPipeline(ctx, conf, [], false, () => {})
     },
     (token, pipeline, broadcast) => new ClientSession(broadcast, token, pipeline),
     port,

models/board/src/index.ts

@@ -228,17 +228,6 @@ export function createModel (builder: Builder): void {
     inlineEditor: board.component.CardCoverEditor
   })
 
-  builder.createDoc(
-    task.class.KanbanTemplateSpace,
-    core.space.Model,
-    {
-      name: board.string.Boards,
-      description: board.string.ManageBoardStatuses,
-      icon: board.component.TemplatesIcon
-    },
-    board.space.BoardTemplates
-  )
-
   builder.createDoc(
     view.class.ViewletDescriptor,
     core.space.Model,

models/board/src/migration.ts

@@ -21,7 +21,6 @@ import {
   DOMAIN_TX,
   generateId,
   Ref,
-  Space,
   TxCollectionCUD,
   TxCreateDoc,
   TxCUD,
@@ -55,6 +54,24 @@ async function createSpace (tx: TxOperations): Promise<void> {
       board.space.DefaultBoard
     )
   }
+  const currentTemplate = await tx.findOne(core.class.Space, {
+    _id: board.space.BoardTemplates
+  })
+  if (currentTemplate === undefined) {
+    await tx.createDoc(
+      task.class.KanbanTemplateSpace,
+      core.space.Space,
+      {
+        name: board.string.Boards,
+        description: board.string.ManageBoardStatuses,
+        icon: board.component.TemplatesIcon,
+        private: false,
+        archived: false,
+        members: []
+      },
+      board.space.BoardTemplates
+    )
+  }
 }
 
 async function createDefaultKanbanTemplate (tx: TxOperations): Promise<Ref<KanbanTemplate>> {
@@ -71,7 +88,7 @@ async function createDefaultKanbanTemplate (tx: TxOperations): Promise<Ref<Kanba
 
   return await createKanbanTemplate(tx, {
     kanbanId: board.template.DefaultBoard,
-    space: board.space.BoardTemplates as Ref<Doc> as Ref<Space>,
+    space: board.space.BoardTemplates,
     title: 'Default board',
     states: defaultKanban.states,
     doneStates: defaultKanban.doneStates

models/core/src/migration.ts

@@ -13,9 +13,28 @@
 // limitations under the License.
 //
 
+import core, { AccountRole, Client, TxOperations } from '@hcengineering/core'
 import { MigrateOperation, MigrationClient, MigrationUpgradeClient } from '@hcengineering/model'
 
 export const coreOperation: MigrateOperation = {
   async migrate (client: MigrationClient): Promise<void> {},
-  async upgrade (client: MigrationUpgradeClient): Promise<void> {}
+  async upgrade (client: MigrationUpgradeClient): Promise<void> {
+    await createSystemAccount(client)
+  }
+}
+
+async function createSystemAccount (client: Client): Promise<void> {
+  const current = await client.findOne(core.class.Account, { _id: core.account.System })
+  if (current === undefined) {
+    const txop = new TxOperations(client, core.account.System)
+    await txop.createDoc(
+      core.class.Account,
+      core.space.Model,
+      {
+        email: 'anticrm@hc.engineering',
+        role: AccountRole.Owner
+      },
+      core.account.System
+    )
+  }
 }

models/lead/src/index.ts

@@ -308,17 +308,6 @@ export function createModel (builder: Builder): void {
     filters: ['_class']
   })
 
-  builder.createDoc(
-    task.class.KanbanTemplateSpace,
-    core.space.Model,
-    {
-      name: lead.string.Funnels,
-      description: lead.string.ManageFunnelStatuses,
-      icon: lead.component.TemplatesIcon
-    },
-    lead.space.FunnelTemplates
-  )
-
   createAction(builder, {
     action: workbench.actionImpl.Navigate,
     actionProps: {

models/lead/src/migration.ts

@@ -13,12 +13,12 @@
 // limitations under the License.
 //
 
-import { Doc, DOMAIN_TX, Ref, Space, TxCreateDoc, TxOperations } from '@hcengineering/core'
+import { DOMAIN_TX, Ref, TxCreateDoc, TxOperations } from '@hcengineering/core'
 import { Funnel } from '@hcengineering/lead'
 import { MigrateOperation, MigrationClient, MigrationUpgradeClient } from '@hcengineering/model'
 import core, { DOMAIN_SPACE } from '@hcengineering/model-core'
 import { createKanbanTemplate, createSequence } from '@hcengineering/model-task'
-import task, { KanbanTemplate, createKanban } from '@hcengineering/task'
+import task, { createKanban, KanbanTemplate } from '@hcengineering/task'
 import lead from './plugin'
 
 async function createSpace (tx: TxOperations): Promise<void> {
@@ -39,6 +39,25 @@ async function createSpace (tx: TxOperations): Promise<void> {
       lead.space.DefaultFunnel
     )
   }
+
+  const currentTemplate = await tx.findOne(core.class.Space, {
+    _id: lead.space.FunnelTemplates
+  })
+  if (currentTemplate === undefined) {
+    await tx.createDoc(
+      task.class.KanbanTemplateSpace,
+      core.space.Space,
+      {
+        name: lead.string.Funnels,
+        description: lead.string.ManageFunnelStatuses,
+        icon: lead.component.TemplatesIcon,
+        private: false,
+        members: [],
+        archived: false
+      },
+      lead.space.FunnelTemplates
+    )
+  }
 }
 
 async function createDefaultKanbanTemplate (tx: TxOperations): Promise<Ref<KanbanTemplate>> {
@@ -59,7 +78,7 @@ async function createDefaultKanbanTemplate (tx: TxOperations): Promise<Ref<Kanba
 
   return await createKanbanTemplate(tx, {
     kanbanId: lead.template.DefaultFunnel,
-    space: lead.space.FunnelTemplates as Ref<Doc> as Ref<Space>,
+    space: lead.space.FunnelTemplates,
     title: 'Default funnel',
     states: defaultKanban.states,
     doneStates: defaultKanban.doneStates

models/notification/src/migration.ts

@@ -58,11 +58,33 @@ async function fillNotificationType (client: MigrationUpgradeClient): Promise<vo
   await Promise.all(promises)
 }
 
+async function createSpace (client: MigrationUpgradeClient): Promise<void> {
+  const txop = new TxOperations(client, core.account.System)
+  const currentTemplate = await txop.findOne(core.class.Space, {
+    _id: notification.space.Notifications
+  })
+  if (currentTemplate === undefined) {
+    await txop.createDoc(
+      core.class.Space,
+      core.space.Space,
+      {
+        name: 'Notification space',
+        description: 'Notification space',
+        private: false,
+        archived: false,
+        members: []
+      },
+      notification.space.Notifications
+    )
+  }
+}
+
 export const notificationOperation: MigrateOperation = {
   async migrate (client: MigrationClient): Promise<void> {
     await fillNotificationText(client)
   },
   async upgrade (client: MigrationUpgradeClient): Promise<void> {
+    await createSpace(client)
     await fillNotificationType(client)
   }
 }

models/recruit/src/index.ts

@@ -708,18 +708,6 @@ export function createModel (builder: Builder): void {
     recruit.action.CreateGlobalApplication
   )
 
-  builder.createDoc(
-    task.class.KanbanTemplateSpace,
-    core.space.Model,
-    {
-      name: recruit.string.Vacancies,
-      description: recruit.string.ManageVacancyStatuses,
-      icon: recruit.component.TemplatesIcon,
-      editor: recruit.component.VacancyTemplateEditor
-    },
-    recruit.space.VacancyTemplates
-  )
-
   builder.createDoc(
     presentation.class.ObjectSearchCategory,
     core.space.Model,

models/recruit/src/migration.ts

@@ -23,7 +23,7 @@ import { DOMAIN_SPACE } from '@hcengineering/model-core'
 import tags, { TagCategory } from '@hcengineering/model-tags'
 import { createKanbanTemplate, createSequence, DOMAIN_TASK } from '@hcengineering/model-task'
 import { Applicant, Candidate, Vacancy } from '@hcengineering/recruit'
-import { KanbanTemplate } from '@hcengineering/task'
+import task, { KanbanTemplate } from '@hcengineering/task'
 import recruit from './plugin'
 
 async function fixImportedTitle (client: MigrationClient): Promise<void> {
@@ -253,11 +253,33 @@ async function createSpaces (tx: TxOperations): Promise<void> {
       {
         name: 'Reviews',
         description: 'Public reviews',
-        private: true,
+        private: false,
         members: [],
         archived: false
       },
       recruit.space.Reviews
     )
+  } else if (currentReviews.private) {
+    await tx.update(currentReviews, { private: false })
+  }
+
+  const currentTemplate = await tx.findOne(core.class.Space, {
+    _id: recruit.space.VacancyTemplates
+  })
+  if (currentTemplate === undefined) {
+    await tx.createDoc(
+      task.class.KanbanTemplateSpace,
+      core.space.Space,
+      {
+        name: recruit.string.Vacancies,
+        description: recruit.string.ManageVacancyStatuses,
+        icon: recruit.component.TemplatesIcon,
+        editor: recruit.component.VacancyTemplateEditor,
+        private: false,
+        members: [],
+        archived: false
+      },
+      recruit.space.VacancyTemplates
+    )
   }
 }

models/tags/src/migration.ts

@@ -58,12 +58,14 @@ export const tagsOperation: MigrateOperation = {
         {
           name: 'Tags',
           description: 'Space for all tags',
-          private: true,
+          private: false,
           archived: false,
           members: []
         },
         tags.space.Tags
       )
+    } else if (current.private) {
+      await tx.update(current, { private: false })
     }
   }
 }

models/task/src/index.ts

@@ -15,18 +15,7 @@
 
 import type { Employee } from '@hcengineering/contact'
 import contact from '@hcengineering/contact'
-import {
+import { Arr, Class, Doc, Domain, FindOptions, IndexKind, Ref, Space, Timestamp } from '@hcengineering/core'
-  Arr,
-  Class,
-  Doc,
-  Domain,
-  DOMAIN_MODEL,
-  FindOptions,
-  IndexKind,
-  Ref,
-  Space,
-  Timestamp
-} from '@hcengineering/core'
 import {
   Builder,
   Collection,
@@ -208,8 +197,8 @@ export class TKanban extends TDoc implements Kanban {
   attachedTo!: Ref<Space>
 }
 
-@Model(task.class.KanbanTemplateSpace, core.class.Doc, DOMAIN_MODEL)
+@Model(task.class.KanbanTemplateSpace, core.class.Space)
-export class TKanbanTemplateSpace extends TDoc implements KanbanTemplateSpace {
+export class TKanbanTemplateSpace extends TSpace implements KanbanTemplateSpace {
   name!: IntlString
   description!: IntlString
   icon!: AnyComponent
@@ -417,17 +406,6 @@ export function createModel (builder: Builder): void {
     card: task.component.KanbanCard
   })
 
-  builder.createDoc(
-    task.class.KanbanTemplateSpace,
-    core.space.Model,
-    {
-      name: task.string.Projects,
-      description: task.string.ManageProjectStatues,
-      icon: task.component.TemplatesIcon
-    },
-    task.space.ProjectTemplates
-  )
-
   builder.createDoc(
     view.class.ActionCategory,
     core.space.Model,

models/task/src/migration.ts

@@ -175,7 +175,29 @@ async function createDefaultKanban (tx: TxOperations): Promise<void> {
   await createKanban(tx, task.space.TasksPublic, defaultTmpl)
 }
 
+async function createSpace (tx: TxOperations): Promise<void> {
+  const currentTemplate = await tx.findOne(core.class.Space, {
+    _id: task.space.ProjectTemplates
+  })
+  if (currentTemplate === undefined) {
+    await tx.createDoc(
+      task.class.KanbanTemplateSpace,
+      core.space.Space,
+      {
+        name: task.string.Projects,
+        description: task.string.ManageProjectStatues,
+        icon: task.component.TemplatesIcon,
+        private: false,
+        members: [],
+        archived: false
+      },
+      task.space.ProjectTemplates
+    )
+  }
+}
+
 async function createDefaults (tx: TxOperations): Promise<void> {
+  await createSpace(tx)
   await createDefaultSequence(tx)
   await createDefaultProject(tx)
   await createSequence(tx, task.class.Issue)

models/templates/src/migration.ts

@@ -40,6 +40,8 @@ export const templatesOperation: MigrateOperation = {
         },
         templates.space.Templates
       )
+    } else if (current.private) {
+      await tx.update(current, { private: false })
     }
   }
 }

packages/core/src/tx.ts

@@ -125,7 +125,7 @@ export interface TxMixin<D extends Doc, M extends D> extends TxCUD<D> {
  * @public
  */
 export type ArrayAsElement<T> = {
-  [P in keyof T]: T[P] extends Arr<infer X> ? Partial<X> | PullArray<X> : never
+  [P in keyof T]: T[P] extends Arr<infer X> ? Partial<X> | PullArray<X> | X : never
 }
 
 /**

plugins/client-resources/src/connection.ts

@@ -127,6 +127,7 @@ class Connection implements ClientConnection {
           }
           this.requests.delete(resp.id)
           if (resp.error !== undefined) {
+            console.log('ERROR', resp.id)
             promise.reject(new PlatformError(resp.error))
           } else {
             promise.resolve(resp.result)

plugins/task/src/index.ts

@@ -176,7 +176,7 @@ export interface KanbanTemplate extends Doc {
 /**
  * @public
  */
-export interface KanbanTemplateSpace extends Doc {
+export interface KanbanTemplateSpace extends Space {
   name: IntlString
   description: IntlString
   icon: AnyComponent

plugins/tracker-resources/src/components/myissues/MyIssues.svelte

@@ -52,12 +52,12 @@
   const subscribedQuery = createQuery()
   $: subscribedQuery.query(
     notification.class.LastView,
-    { user: getCurrentAccount()._id, attachedToClass: tracker.class.Issue, lastView: { $gte: 0 } },
+    { user: getCurrentAccount()._id, attachedToClass: tracker.class.Issue, lastView: { $ne: -1 } },
     (result) => {
       const newSub = result.map(({ attachedTo }) => attachedTo as Ref<Issue>)
       const curSub = subscribed._id.$in
       if (curSub.length !== newSub.length || curSub.some((id, i) => newSub[i] !== id)) {
-        subscribed = { ...subscribed, _id: { $in: newSub } }
+        subscribed = { _id: { $in: newSub } }
       }
     },
     { sort: { _id: 1 } }

pods/server/src/server.ts

@@ -25,7 +25,12 @@ import {
   WorkspaceId
 } from '@hcengineering/core'
 import { createElasticAdapter, createElasticBackupDataAdapter } from '@hcengineering/elastic'
-import { ConfigurationMiddleware, ModifiedMiddleware, PrivateMiddleware } from '@hcengineering/middleware'
+import {
+  ConfigurationMiddleware,
+  ModifiedMiddleware,
+  PrivateMiddleware,
+  SpaceSecurityMiddleware
+} from '@hcengineering/middleware'
 import { MinioService } from '@hcengineering/minio'
 import { createMongoAdapter, createMongoTxAdapter } from '@hcengineering/mongo'
 import { OpenAIEmbeddingsStage, openAIId, openAIPluginImpl } from '@hcengineering/openai'
@@ -192,6 +197,7 @@ export function start (
   const middlewares: MiddlewareCreator[] = [
     ModifiedMiddleware.create,
     PrivateMiddleware.create,
+    SpaceSecurityMiddleware.create,
     ConfigurationMiddleware.create
   ]
 
@@ -248,7 +254,7 @@ export function start (
 
   return startJsonRpc(
     getMetricsContext(),
-    (workspace, upgrade, broadcast) => {
+    (ctx, workspace, upgrade, broadcast) => {
       const conf: DbConfiguration = {
         domains: {
           [DOMAIN_TX]: 'MongoTx',
@@ -310,7 +316,7 @@ export function start (
           }),
         workspace
       }
-      return createPipeline(conf, middlewares, upgrade, broadcast)
+      return createPipeline(ctx, conf, middlewares, upgrade, broadcast)
     },
     (token: Token, pipeline: Pipeline, broadcast: BroadcastCall) => {
       if (token.extra?.mode === 'backup') {

server/account/src/index.ts

@@ -351,6 +351,11 @@ export async function createAccount (
   const salt = randomBytes(32)
   const hash = hashWithSalt(password, salt)
 
+  const systemEmails = ['anticrm@hc.engineering']
+  if (systemEmails.includes(email)) {
+    throw new PlatformError(new Status(Severity.ERROR, accountPlugin.status.AccountAlreadyExists, { account: email }))
+  }
+
   const account = await getAccount(db, email)
   if (account !== null) {
     throw new PlatformError(new Status(Severity.ERROR, accountPlugin.status.AccountAlreadyExists, { account: email }))
@@ -588,10 +593,8 @@ async function createEmployeeAccount (account: Account, productId: string, works
     const ops = new TxOperations(connection, core.account.System)
 
     const name = combineName(account.first, account.last)
-
     // Check if EmployeeAccoun is not exists
     const existingAccount = await ops.findOne(contact.class.EmployeeAccount, { email: account.email })
-
     if (existingAccount === undefined) {
       const employee = await createEmployee(ops, name, account.email)
 

server/core/src/pipeline.ts

@@ -20,6 +20,7 @@ import {
   Domain,
   FindOptions,
   FindResult,
+  MeasureContext,
   ModelDb,
   Ref,
   ServerStorage,
@@ -34,6 +35,7 @@ import { Middleware, MiddlewareCreator, Pipeline, SessionContext } from './types
  * @public
  */
 export async function createPipeline (
+  ctx: MeasureContext,
   conf: DbConfiguration,
   constructors: MiddlewareCreator[],
   upgrade: boolean,
@@ -43,22 +45,32 @@ export async function createPipeline (
     upgrade,
     broadcast
   })
-  return new TPipeline(storage, constructors)
+  const pipeline = PipelineImpl.create(ctx, storage, constructors)
+  return await pipeline
 }
 
-class TPipeline implements Pipeline {
+class PipelineImpl implements Pipeline {
-  private readonly head: Middleware | undefined
+  private head: Middleware | undefined
   readonly modelDb: ModelDb
-  constructor (readonly storage: ServerStorage, constructors: MiddlewareCreator[]) {
+  private constructor (readonly storage: ServerStorage) {
-    this.head = this.buildChain(constructors)
     this.modelDb = storage.modelDb
   }
 
-  private buildChain (constructors: MiddlewareCreator[]): Middleware | undefined {
+  static async create (
+    ctx: MeasureContext,
+    storage: ServerStorage,
+    constructors: MiddlewareCreator[]
+  ): Promise<PipelineImpl> {
+    const pipeline = new PipelineImpl(storage)
+    pipeline.head = await pipeline.buildChain(ctx, constructors)
+    return pipeline
+  }
+
+  private async buildChain (ctx: MeasureContext, constructors: MiddlewareCreator[]): Promise<Middleware | undefined> {
     let current: Middleware | undefined
     for (let index = constructors.length - 1; index >= 0; index--) {
       const element = constructors[index]
-      current = element(this.storage, current)
+      current = await element(ctx, this.storage, current)
     }
     return current
   }
@@ -69,15 +81,18 @@ class TPipeline implements Pipeline {
     query: DocumentQuery<T>,
     options?: FindOptions<T>
   ): Promise<FindResult<T>> {
-    const [session, resClass, resQuery, resOptions] =
+    return this.head !== undefined
-      this.head === undefined ? [ctx, _class, query, options] : await this.head.findAll(ctx, _class, query, options)
+      ? await this.head.findAll(ctx, _class, query, options)
-    return await this.storage.findAll(session, resClass, resQuery, resOptions)
+      : await this.storage.findAll(ctx, _class, query, options)
   }
 
-  async tx (ctx: SessionContext, tx: Tx): Promise<[TxResult, Tx[], string | undefined]> {
+  async tx (ctx: SessionContext, tx: Tx): Promise<[TxResult, Tx[], string[] | undefined]> {
-    const [session, resTx, target] = this.head === undefined ? [ctx, tx] : await this.head.tx(ctx, tx)
+    if (this.head === undefined) {
-    const res = await this.storage.tx(session, resTx)
+      const res = await this.storage.tx(ctx, tx)
-    return [res[0], res[1], target]
+      return [...res, undefined]
+    } else {
+      return await this.head.tx(ctx, tx)
+    }
   }
 
   async close (): Promise<void> {

server/core/src/types.ts

@@ -58,28 +58,18 @@ export interface Middleware {
     _class: Ref<Class<T>>,
     query: DocumentQuery<T>,
     options?: FindOptions<T>
-  ) => Promise<FindAllMiddlewareResult<T>>
+  ) => Promise<FindResult<T>>
 }
 
 /**
  * @public
  */
-export type MiddlewareCreator = (storage: ServerStorage, next?: Middleware) => Middleware
+export type MiddlewareCreator = (ctx: MeasureContext, storage: ServerStorage, next?: Middleware) => Promise<Middleware>
 
 /**
  * @public
  */
-export type TxMiddlewareResult = [SessionContext, Tx, string | undefined]
+export type TxMiddlewareResult = [TxResult, Tx[], string[] | undefined]
-
-/**
- * @public
- */
-export type FindAllMiddlewareResult<T extends Doc> = [
-  SessionContext,
-  Ref<Class<T>>,
-  DocumentQuery<T>,
-  FindOptions<T> | undefined
-]
 
 /**
  * @public
@@ -93,7 +83,7 @@ export interface Pipeline extends LowLevelStorage {
     query: DocumentQuery<T>,
     options?: FindOptions<T>
   ) => Promise<FindResult<T>>
-  tx: (ctx: SessionContext, tx: Tx) => Promise<[TxResult, Tx[], string | undefined]>
+  tx: (ctx: SessionContext, tx: Tx) => Promise<[TxResult, Tx[], string[] | undefined]>
   close: () => Promise<void>
 }
 

server/middleware/src/base.ts

@@ -13,8 +13,8 @@
 // limitations under the License.
 //
 
-import { Class, Doc, DocumentQuery, FindOptions, Ref, ServerStorage, Tx } from '@hcengineering/core'
+import { Class, Doc, DocumentQuery, FindOptions, FindResult, Ref, ServerStorage, Tx } from '@hcengineering/core'
-import { FindAllMiddlewareResult, Middleware, SessionContext, TxMiddlewareResult } from '@hcengineering/server-core'
+import { Middleware, SessionContext, TxMiddlewareResult } from '@hcengineering/server-core'
 
 /**
  * @public
@@ -27,7 +27,7 @@ export abstract class BaseMiddleware {
     _class: Ref<Class<T>>,
     query: DocumentQuery<T>,
     options?: FindOptions<T>
-  ): Promise<FindAllMiddlewareResult<T>> {
+  ): Promise<FindResult<T>> {
     return await this.provideFindAll(ctx, _class, query, options)
   }
 
@@ -35,7 +35,8 @@ export abstract class BaseMiddleware {
     if (this.next !== undefined) {
       return await this.next.tx(ctx, tx)
     }
-    return [ctx, tx, undefined]
+    const res = await this.storage.tx(ctx, tx)
+    return [res[0], res[1], undefined]
   }
 
   protected async provideFindAll<T extends Doc>(
@@ -43,10 +44,10 @@ export abstract class BaseMiddleware {
     _class: Ref<Class<T>>,
     query: DocumentQuery<T>,
     options?: FindOptions<T>
-  ): Promise<FindAllMiddlewareResult<T>> {
+  ): Promise<FindResult<T>> {
     if (this.next !== undefined) {
       return await this.next.findAll(ctx, _class, query, options)
     }
-    return [ctx, _class, query, options]
+    return await this.storage.findAll(ctx, _class, query, options)
   }
 }

server/middleware/src/configuration.ts

@@ -21,13 +21,15 @@ import core, {
   DocumentQuery,
   DOMAIN_CONFIGURATION,
   FindOptions,
+  FindResult,
+  MeasureContext,
   Ref,
   ServerStorage,
   Tx,
   TxCUD
 } from '@hcengineering/core'
 import platform, { PlatformError, Severity, Status } from '@hcengineering/platform'
-import { FindAllMiddlewareResult, Middleware, SessionContext, TxMiddlewareResult } from '@hcengineering/server-core'
+import { Middleware, SessionContext, TxMiddlewareResult } from '@hcengineering/server-core'
 import { BaseMiddleware } from './base'
 
 const configurationAccountEmail = '#configurator@hc.engineering'
@@ -41,7 +43,11 @@ export class ConfigurationMiddleware extends BaseMiddleware implements Middlewar
     super(storage, next)
   }
 
-  static create (storage: ServerStorage, next?: Middleware): ConfigurationMiddleware {
+  static async create (
+    ctx: MeasureContext,
+    storage: ServerStorage,
+    next?: Middleware
+  ): Promise<ConfigurationMiddleware> {
     return new ConfigurationMiddleware(storage, next)
   }
 
@@ -66,7 +72,7 @@ export class ConfigurationMiddleware extends BaseMiddleware implements Middlewar
     _class: Ref<Class<T>>,
     query: DocumentQuery<T>,
     options?: FindOptions<T>
-  ): Promise<FindAllMiddlewareResult<T>> {
+  ): Promise<FindResult<T>> {
     const domain = this.storage.hierarchy.getDomain(_class)
     if (this.targetDomains.includes(domain)) {
       if (ctx.userEmail !== configurationAccountEmail) {

server/middleware/src/index.ts

@@ -17,3 +17,4 @@ export * from './base'
 export * from './modified'
 export * from './private'
 export * from './configuration'
+export * from './spaceSecurity'

server/middleware/src/modified.ts

@@ -13,7 +13,16 @@
 // limitations under the License.
 //
 
-import core, { AttachedDoc, Doc, ServerStorage, Timestamp, Tx, TxCollectionCUD, TxCreateDoc } from '@hcengineering/core'
+import core, {
+  AttachedDoc,
+  Doc,
+  MeasureContext,
+  ServerStorage,
+  Timestamp,
+  Tx,
+  TxCollectionCUD,
+  TxCreateDoc
+} from '@hcengineering/core'
 import { Middleware, SessionContext, TxMiddlewareResult } from '@hcengineering/server-core'
 import { BaseMiddleware } from './base'
 
@@ -25,7 +34,7 @@ export class ModifiedMiddleware extends BaseMiddleware implements Middleware {
     super(storage, next)
   }
 
-  static create (storage: ServerStorage, next?: Middleware): ModifiedMiddleware {
+  static async create (ctx: MeasureContext, storage: ServerStorage, next?: Middleware): Promise<ModifiedMiddleware> {
     return new ModifiedMiddleware(storage, next)
   }
 

server/middleware/src/private.ts

@@ -14,20 +14,24 @@
 //
 
 import core, {
-  Tx,
+  AttachedDoc,
-  Doc,
-  Ref,
   Class,
+  Doc,
   DocumentQuery,
   FindOptions,
+  FindResult,
+  LookupData,
+  MeasureContext,
+  Ref,
   ServerStorage,
-  Account,
+  Tx,
   TxCUD
 } from '@hcengineering/core'
 import platform, { PlatformError, Severity, Status } from '@hcengineering/platform'
-import { Middleware, SessionContext, TxMiddlewareResult, FindAllMiddlewareResult } from '@hcengineering/server-core'
+import { Middleware, SessionContext, TxMiddlewareResult } from '@hcengineering/server-core'
 import { DOMAIN_PREFERENCE } from '@hcengineering/server-preference'
 import { BaseMiddleware } from './base'
+import { getUser, mergeTargets } from './utils'
 
 /**
  * @public
@@ -39,25 +43,25 @@ export class PrivateMiddleware extends BaseMiddleware implements Middleware {
     super(storage, next)
   }
 
-  static create (storage: ServerStorage, next?: Middleware): PrivateMiddleware {
+  static async create (ctx: MeasureContext, storage: ServerStorage, next?: Middleware): Promise<PrivateMiddleware> {
     return new PrivateMiddleware(storage, next)
   }
 
   async tx (ctx: SessionContext, tx: Tx): Promise<TxMiddlewareResult> {
-    let target: string | undefined
+    let target: string[] | undefined
     if (this.storage.hierarchy.isDerived(tx._class, core.class.TxCUD)) {
       const txCUD = tx as TxCUD<Doc>
       const domain = this.storage.hierarchy.getDomain(txCUD.objectClass)
       if (this.targetDomains.includes(domain)) {
-        const account = await this.getUser(ctx)
+        const account = await getUser(this.storage, ctx)
-        if (account !== tx.modifiedBy) {
+        if (account !== tx.modifiedBy && account !== core.account.System) {
           throw new PlatformError(new Status(Severity.ERROR, platform.status.Forbidden, {}))
         }
-        target = ctx.userEmail
+        target = [ctx.userEmail]
       }
     }
     const res = await this.provideTx(ctx, tx)
-    return [res[0], res[1], res[2] ?? target]
+    return [res[0], res[1], mergeTargets(target, res[2])]
   }
 
   override async findAll<T extends Doc>(
@@ -65,27 +69,52 @@ export class PrivateMiddleware extends BaseMiddleware implements Middleware {
     _class: Ref<Class<T>>,
     query: DocumentQuery<T>,
     options?: FindOptions<T>
-  ): Promise<FindAllMiddlewareResult<T>> {
+  ): Promise<FindResult<T>> {
     let newQuery = query
     const domain = this.storage.hierarchy.getDomain(_class)
     if (this.targetDomains.includes(domain)) {
-      const account = await this.getUser(ctx)
+      const account = await getUser(this.storage, ctx)
-      newQuery = {
+      if (account !== core.account.System) {
-        ...query,
+        newQuery = {
-        modifiedBy: account
+          ...query,
+          modifiedBy: account
+        }
       }
     }
-    return await this.provideFindAll(ctx, _class, newQuery, options)
+    const findResult = await this.provideFindAll(ctx, _class, newQuery, options)
+    if (options?.lookup !== undefined) {
+      for (const object of findResult) {
+        if (object.$lookup !== undefined) {
+          await this.filterLookup(ctx, object.$lookup)
+        }
+      }
+    }
+    return findResult
   }
 
-  private async getUser (ctx: SessionContext): Promise<Ref<Account>> {
+  async isAvailable (ctx: SessionContext, doc: Doc): Promise<boolean> {
-    if (ctx.userEmail === undefined) {
+    const domain = this.storage.hierarchy.getDomain(doc._class)
-      throw new PlatformError(new Status(Severity.ERROR, platform.status.Forbidden, {}))
+    if (!this.targetDomains.includes(domain)) return true
+    const account = await getUser(this.storage, ctx)
+    return doc.modifiedBy === account || account === core.account.System
+  }
+
+  async filterLookup<T extends Doc>(ctx: SessionContext, lookup: LookupData<T>): Promise<void> {
+    for (const key in lookup) {
+      const val = lookup[key]
+      if (Array.isArray(val)) {
+        const arr: AttachedDoc[] = []
+        for (const value of val) {
+          if (await this.isAvailable(ctx, value)) {
+            arr.push(value)
+          }
+        }
+        lookup[key] = arr as any
+      } else if (val !== undefined) {
+        if (!(await this.isAvailable(ctx, val))) {
+          lookup[key] = undefined
+        }
+      }
     }
-    const account = (await this.storage.modelDb.findAll(core.class.Account, { email: ctx.userEmail }))[0]
-    if (account === undefined) {
-      throw new PlatformError(new Status(Severity.ERROR, platform.status.Forbidden, {}))
-    }
-    return account._id
   }
 }

server/middleware/src/spaceSecurity.ts

@@ -0,0 +1,335 @@
+//
+// Copyright © 2023 Hardcore Engineering Inc.
+//
+// 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 core, {
+  Account,
+  AttachedDoc,
+  Class,
+  Doc,
+  DocumentQuery,
+  FindOptions,
+  FindResult,
+  LookupData,
+  MeasureContext,
+  ObjQueryType,
+  Position,
+  PullArray,
+  Ref,
+  ServerStorage,
+  Space,
+  Tx,
+  TxCreateDoc,
+  TxCUD,
+  TxProcessor,
+  TxRemoveDoc,
+  TxUpdateDoc
+} from '@hcengineering/core'
+import platform, { PlatformError, Severity, Status } from '@hcengineering/platform'
+import { Middleware, SessionContext, TxMiddlewareResult } from '@hcengineering/server-core'
+import { BaseMiddleware } from './base'
+import { getUser, mergeTargets } from './utils'
+
+/**
+ * @public
+ */
+export class SpaceSecurityMiddleware extends BaseMiddleware implements Middleware {
+  private allowedSpaces: Record<Ref<Account>, Ref<Space>[]> = {}
+  private privateSpaces: Record<Ref<Space>, Space | undefined> = {}
+  private publicSpaces: Ref<Space>[] = []
+  private readonly systemSpaces = [
+    core.space.Configuration,
+    core.space.DerivedTx,
+    core.space.Model,
+    core.space.Space,
+    core.space.Tx
+  ]
+
+  private constructor (storage: ServerStorage, next?: Middleware) {
+    super(storage, next)
+  }
+
+  static async create (
+    ctx: MeasureContext,
+    storage: ServerStorage,
+    next?: Middleware
+  ): Promise<SpaceSecurityMiddleware> {
+    const res = new SpaceSecurityMiddleware(storage, next)
+    await res.init(ctx)
+    return res
+  }
+
+  private addMemberSpace (member: Ref<Account>, space: Ref<Space>): void {
+    const arr = this.allowedSpaces[member] ?? []
+    arr.push(space)
+    this.allowedSpaces[member] = arr
+  }
+
+  private addSpace (space: Space): void {
+    this.privateSpaces[space._id] = space
+    for (const member of space.members) {
+      this.addMemberSpace(member, space._id)
+    }
+  }
+
+  private async init (ctx: MeasureContext): Promise<void> {
+    const spaces = await this.storage.findAll(ctx, core.class.Space, { private: true })
+    for (const space of spaces) {
+      this.addSpace(space)
+    }
+    this.publicSpaces = (await this.storage.findAll(ctx, core.class.Space, { private: false })).map((p) => p._id)
+  }
+
+  private removeMemberSpace (member: Ref<Account>, space: Ref<Space>): void {
+    const arr = this.allowedSpaces[member]
+    if (arr !== undefined) {
+      const index = arr.findIndex((p) => p === space)
+      if (index !== -1) {
+        arr.splice(index, 1)
+        this.allowedSpaces[member] = arr
+      }
+    }
+  }
+
+  private removeSpace (_id: Ref<Space>): void {
+    const space = this.privateSpaces[_id]
+    if (space !== undefined) {
+      for (const member of space.members) {
+        this.removeMemberSpace(member, space._id)
+      }
+      this.privateSpaces[_id] = undefined
+    }
+  }
+
+  private handleCreate (tx: TxCUD<Space>): void {
+    const createTx = tx as TxCreateDoc<Space>
+    if (!this.storage.hierarchy.isDerived(createTx.objectClass, core.class.Space)) return
+    if (createTx.attributes.private) {
+      const res = TxProcessor.buildDoc2Doc<Space>([createTx])
+      if (res !== undefined) {
+        this.addSpace(res)
+      }
+    } else {
+      this.publicSpaces.push(createTx.objectId)
+    }
+  }
+
+  private pushMembersHandle (addedMembers: Ref<Account> | Position<Ref<Account>>, space: Ref<Space>): void {
+    if (typeof addedMembers === 'object') {
+      for (const member of addedMembers.$each) {
+        this.addMemberSpace(member, space)
+      }
+    } else {
+      this.addMemberSpace(addedMembers, space)
+    }
+  }
+
+  private pullMembersHandle (removedMembers: Partial<Ref<Account>> | PullArray<Ref<Account>>, space: Ref<Space>): void {
+    if (typeof removedMembers === 'object') {
+      const { $in } = removedMembers as PullArray<Ref<Account>>
+      if ($in !== undefined) {
+        for (const member of $in) {
+          this.removeMemberSpace(member, space)
+        }
+      }
+    } else {
+      this.removeMemberSpace(removedMembers, space)
+    }
+  }
+
+  private syncMembers (members: Ref<Account>[], space: Ref<Space>): void {
+    const oldMembers = new Set(members)
+    const newMembers = new Set(members)
+    for (const old of oldMembers) {
+      if (!oldMembers.has(old)) {
+        this.removeMemberSpace(old, space)
+      }
+    }
+    for (const newMem of newMembers) {
+      if (!newMembers.has(newMem)) {
+        this.addMemberSpace(newMem, space)
+      }
+    }
+  }
+
+  private removePublicSpace (_id: Ref<Space>): void {
+    const publicIndex = this.publicSpaces.findIndex((p) => p === _id)
+    if (publicIndex !== -1) {
+      this.publicSpaces.splice(publicIndex, 1)
+    }
+  }
+
+  private async handleUpdate (ctx: SessionContext, tx: TxCUD<Space>): Promise<void> {
+    const updateDoc = tx as TxUpdateDoc<Space>
+    if (!this.storage.hierarchy.isDerived(updateDoc.objectClass, core.class.Space)) return
+
+    if (updateDoc.operations.private !== undefined) {
+      if (updateDoc.operations.private) {
+        const res = (await this.storage.findAll(ctx, core.class.Space, { _id: updateDoc.objectId }))[0]
+        if (res !== undefined) {
+          res.private = true
+          this.addSpace(res)
+          this.removePublicSpace(res._id)
+        }
+      } else if (!updateDoc.operations.private) {
+        this.removeSpace(updateDoc.objectId)
+        this.publicSpaces.push(updateDoc.objectId)
+      }
+    }
+
+    let space = this.privateSpaces[updateDoc.objectId]
+    if (space !== undefined) {
+      if (updateDoc.operations.members !== undefined) {
+        this.syncMembers(updateDoc.operations.members, space._id)
+      }
+      if (updateDoc.operations.$push?.members !== undefined) {
+        this.pushMembersHandle(updateDoc.operations.$push.members, space._id)
+      }
+
+      if (updateDoc.operations.$pull?.members !== undefined) {
+        this.pullMembersHandle(updateDoc.operations.$pull.members, space._id)
+      }
+      space = TxProcessor.updateDoc2Doc(space, updateDoc)
+    }
+  }
+
+  private handleRemove (tx: TxCUD<Space>): void {
+    const removeTx = tx as TxRemoveDoc<Space>
+    if (!this.storage.hierarchy.isDerived(removeTx.objectClass, core.class.Space)) return
+    if (removeTx._class !== core.class.TxCreateDoc) return
+    this.removeSpace(tx.objectId)
+    this.removePublicSpace(tx.objectId)
+  }
+
+  private async handleTx (ctx: SessionContext, tx: TxCUD<Space>): Promise<void> {
+    if (tx._class === core.class.TxCreateDoc) {
+      this.handleCreate(tx)
+    } else if (tx._class === core.class.TxUpdateDoc) {
+      await this.handleUpdate(ctx, tx)
+    } else if (tx._class === core.class.TxRemoveDoc) {
+      this.handleRemove(tx)
+    }
+  }
+
+  async getTargets (accounts: Ref<Account>[] | undefined): Promise<string[] | undefined> {
+    if (accounts === undefined) return
+    const users = await this.storage.modelDb.findAll(core.class.Account, { _id: { $in: accounts } })
+    return users.map((p) => p.email)
+  }
+
+  async tx (ctx: SessionContext, tx: Tx): Promise<TxMiddlewareResult> {
+    const h = this.storage.hierarchy
+    let targets: string[] | undefined
+
+    if (h.isDerived(tx._class, core.class.TxCUD)) {
+      const cudTx = tx as TxCUD<Doc>
+      const isSpace = h.isDerived(cudTx.objectClass, core.class.Space)
+      if (isSpace) {
+        await this.handleTx(ctx, cudTx as TxCUD<Space>)
+      }
+      const space = this.privateSpaces[tx.objectSpace]
+      if (space !== undefined) {
+        const account = await getUser(this.storage, ctx)
+        if (account !== core.account.System) {
+          const allowed = this.allowedSpaces[account]
+          if (allowed === undefined || !allowed.includes(isSpace ? (cudTx.objectId as Ref<Space>) : tx.objectSpace)) {
+            throw new PlatformError(new Status(Severity.ERROR, platform.status.Forbidden, {}))
+          }
+        }
+        targets = await this.getTargets(this.privateSpaces[tx.objectSpace]?.members)
+      }
+    }
+
+    const res = await this.provideTx(ctx, tx)
+    return [res[0], res[1], mergeTargets(targets, res[2])]
+  }
+
+  private async getAllAllowedSpaces (ctx: SessionContext): Promise<Ref<Space>[]> {
+    let userSpaces: Ref<Space>[] = []
+    try {
+      const account = await getUser(this.storage, ctx)
+      userSpaces = this.allowedSpaces[account] ?? []
+      return [...userSpaces, account as string as Ref<Space>, ...this.publicSpaces, ...this.systemSpaces]
+    } catch {
+      return [...this.publicSpaces, ...this.systemSpaces]
+    }
+  }
+
+  private async mergeQuery<T extends Doc>(
+    ctx: SessionContext,
+    query: ObjQueryType<T['space']>
+  ): Promise<ObjQueryType<T['space']>> {
+    const spaces = await this.getAllAllowedSpaces(ctx)
+    if (typeof query === 'string') {
+      if (!spaces.includes(query)) {
+        throw new PlatformError(new Status(Severity.ERROR, platform.status.Forbidden, {}))
+      }
+    } else if (query.$in !== undefined) {
+      query.$in = query.$in.filter((p) => spaces.includes(p))
+    } else {
+      query.$in = spaces
+    }
+    return query
+  }
+
+  override async findAll<T extends Doc>(
+    ctx: SessionContext,
+    _class: Ref<Class<T>>,
+    query: DocumentQuery<T>,
+    options?: FindOptions<T>
+  ): Promise<FindResult<T>> {
+    const newQuery = query
+    if (query.space !== undefined) {
+      newQuery.space = await this.mergeQuery(ctx, query.space)
+    } else {
+      const spaces = await this.getAllAllowedSpaces(ctx)
+      newQuery.space = { $in: spaces }
+    }
+    const findResult = await this.provideFindAll(ctx, _class, newQuery, options)
+    if (options?.lookup !== undefined) {
+      for (const object of findResult) {
+        if (object.$lookup !== undefined) {
+          await this.filterLookup(ctx, object.$lookup)
+        }
+      }
+    }
+    return findResult
+  }
+
+  async isUnavailable (ctx: SessionContext, space: Ref<Space>): Promise<boolean> {
+    if (this.privateSpaces[space] === undefined) return false
+    const account = await getUser(this.storage, ctx)
+    if (account === core.account.System) return false
+    return !this.allowedSpaces[account]?.includes(space)
+  }
+
+  async filterLookup<T extends Doc>(ctx: SessionContext, lookup: LookupData<T>): Promise<void> {
+    for (const key in lookup) {
+      const val = lookup[key]
+      if (Array.isArray(val)) {
+        const arr: AttachedDoc[] = []
+        for (const value of val) {
+          if (!(await this.isUnavailable(ctx, value.space))) {
+            arr.push(value)
+          }
+        }
+        lookup[key] = arr as any
+      } else if (val !== undefined) {
+        if (await this.isUnavailable(ctx, val.space)) {
+          lookup[key] = undefined
+        }
+      }
+    }
+  }
+}

server/middleware/src/utils.ts

@@ -0,0 +1,41 @@
+//
+// Copyright © 2023 Hardcore Engineering Inc.
+//
+// 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 core, { Account, Ref, ServerStorage } from '@hcengineering/core'
+import platform, { PlatformError, Severity, Status } from '@hcengineering/platform'
+import { SessionContext } from '@hcengineering/server-core'
+
+export function mergeTargets (current: string[] | undefined, prev: string[] | undefined): string[] | undefined {
+  if (current === undefined) return prev
+  if (prev === undefined) return current
+  const res: string[] = []
+  for (const value of current) {
+    if (prev.includes(value)) {
+      res.push(value)
+    }
+  }
+  return res
+}
+
+export async function getUser (storage: ServerStorage, ctx: SessionContext): Promise<Ref<Account>> {
+  if (ctx.userEmail === undefined) {
+    throw new PlatformError(new Status(Severity.ERROR, platform.status.Forbidden, {}))
+  }
+  const account = (await storage.modelDb.findAll(core.class.Account, { email: ctx.userEmail }))[0]
+  if (account === undefined) {
+    throw new PlatformError(new Status(Severity.ERROR, platform.status.Forbidden, {}))
+  }
+  return account._id
+}

server/ws/src/__tests__/server.test.ts

@@ -66,7 +66,7 @@ describe('server', () => {
         query: DocumentQuery<T>,
         options?: FindOptions<T>
       ): Promise<FindResult<T>> => toFindResult([]),
-      tx: async (ctx: SessionContext, tx: Tx): Promise<[TxResult, Tx[], string | undefined]> => [{}, [], undefined],
+      tx: async (ctx: SessionContext, tx: Tx): Promise<[TxResult, Tx[], string[] | undefined]> => [{}, [], undefined],
       close: async () => {},
       storage: {} as unknown as ServerStorage,
       domains: async () => [],

server/ws/src/server.ts

@@ -68,7 +68,7 @@ class SessionManager {
     workspace = this.workspaces.get(wsString)
 
     if (workspace === undefined) {
-      workspace = this.createWorkspace(pipelineFactory, token)
+      workspace = this.createWorkspace(ctx, pipelineFactory, token)
     }
 
     if (token.extra?.model === 'upgrade') {
@@ -85,7 +85,9 @@ class SessionManager {
       }
       if (LOGGING_ENABLED) console.log('no sessions for workspace', wsString)
       // Re-create pipeline.
-      workspace.pipeline = pipelineFactory(token.workspace, true, (tx) => this.broadcastAll(workspace as Workspace, tx))
+      workspace.pipeline = pipelineFactory(ctx, token.workspace, true, (tx) =>
+        this.broadcastAll(workspace as Workspace, tx)
+      )
 
       const pipeline = await workspace.pipeline
       const session = this.createSession(token, pipeline)
@@ -129,11 +131,11 @@ class SessionManager {
     }
   }
 
-  private createWorkspace (pipelineFactory: PipelineFactory, token: Token): Workspace {
+  private createWorkspace (ctx: MeasureContext, pipelineFactory: PipelineFactory, token: Token): Workspace {
     const upgrade = token.extra?.model === 'upgrade'
     const workspace: Workspace = {
       id: generateId(),
-      pipeline: pipelineFactory(token.workspace, upgrade, (tx) => this.broadcastAll(workspace, tx)),
+      pipeline: pipelineFactory(ctx, token.workspace, upgrade, (tx) => this.broadcastAll(workspace, tx)),
       sessions: [],
       upgrade
     }
@@ -272,7 +274,7 @@ class SessionManager {
     }
   }
 
-  broadcast (from: Session | null, workspaceId: WorkspaceId, resp: Response<any>, target?: string): void {
+  broadcast (from: Session | null, workspaceId: WorkspaceId, resp: Response<any>, target?: string[]): void {
     const workspace = this.workspaces.get(toWorkspaceString(workspaceId))
     if (workspace === undefined) {
       console.error(new Error('internal: cannot find sessions'))
@@ -284,7 +286,7 @@ class SessionManager {
       if (session[0] !== from) {
         if (target === undefined) {
           session[1].send(msg)
-        } else if (session[0].getUser() === target) {
+        } else if (target.includes(session[0].getUser())) {
           session[1].send(msg)
         }
       }

server/ws/src/types.ts

@@ -41,10 +41,15 @@ export type BroadcastCall = (
   from: Session | null,
   workspaceId: WorkspaceId,
   resp: Response<any>,
-  target?: string
+  target?: string[]
 ) => void
 
 /**
  * @public
  */
-export type PipelineFactory = (ws: WorkspaceId, upgrade: boolean, broadcast: (tx: Tx[]) => void) => Promise<Pipeline>
+export type PipelineFactory = (
+  ctx: MeasureContext,
+  ws: WorkspaceId,
+  upgrade: boolean,
+  broadcast: (tx: Tx[]) => void
+) => Promise<Pipeline>