diff --git a/communication b/communication index 2449f29168..2e04e5c091 160000 --- a/communication +++ b/communication @@ -1 +1 @@ -Subproject commit 2449f29168059e0a3f55cf79815057219332acf7 +Subproject commit 2e04e5c0914ec3ce99e896d8f0eb91622570e873 diff --git a/dev/tool/src/github.ts b/dev/tool/src/github.ts index 477b5343dd..6496af53c3 100644 --- a/dev/tool/src/github.ts +++ b/dev/tool/src/github.ts @@ -16,7 +16,7 @@ import core, { import { getAccountsFromTxes, getSocialKeyByOldEmail } from '@hcengineering/model-core' import { getAccountClient } from '@hcengineering/server-client' import { createDummyStorageAdapter, wrapPipeline, type PipelineFactory } from '@hcengineering/server-core' -import { createBackupPipeline } from '@hcengineering/server-pipeline' +import { createBackupPipeline, createEmptyBroadcastOps } from '@hcengineering/server-pipeline' import { generateToken } from '@hcengineering/server-token' import type { Db } from 'mongodb' @@ -106,7 +106,7 @@ export async function performGithubAccountMigrations ( it.workspace = ws.uuid replaces.set(it.workspace, ws.uuid) - const pipeline = await factory(metricsContext, ws, (): void => {}, null, null) + const pipeline = await factory(metricsContext, ws, createEmptyBroadcastOps(), null) const client = wrapPipeline(metricsContext, pipeline, ws, false) const systemAccounts = [core.account.System, core.account.ConfigUser] diff --git a/dev/tool/src/gmail.ts b/dev/tool/src/gmail.ts index 22793607fa..0046854ec3 100644 --- a/dev/tool/src/gmail.ts +++ b/dev/tool/src/gmail.ts @@ -17,14 +17,14 @@ import core, { isDeletingMode, systemAccountUuid } from '@hcengineering/core' -import { getAccountClient } from '@hcengineering/server-client' import { getClient as getKvsClient } from '@hcengineering/kvs-client' -import { generateToken } from '@hcengineering/server-token' import { getAccountsFromTxes, getSocialKeyByOldEmail } from '@hcengineering/model-core' +import { getAccountClient } from '@hcengineering/server-client' +import { generateToken } from '@hcengineering/server-token' -import type { Db } from 'mongodb' import { type PipelineFactory, createDummyStorageAdapter, wrapPipeline } from '@hcengineering/server-core' -import { createBackupPipeline } from '@hcengineering/server-pipeline' +import { createBackupPipeline, createEmptyBroadcastOps } from '@hcengineering/server-pipeline' +import type { Db } from 'mongodb' // Old token and history types interface Credentials { @@ -305,7 +305,7 @@ export async function loadAccounts ( factory: PipelineFactory, metricsContext: MeasureMetricsContext ): Promise<(Doc & { email?: string })[]> { - const pipeline = await factory(metricsContext, ws, (): void => {}, null, null) + const pipeline = await factory(metricsContext, ws, createEmptyBroadcastOps(), null) const client = wrapPipeline(metricsContext, pipeline, ws, false) const accountsTxes: TxCUD[] = [] diff --git a/dev/tool/src/markup.ts b/dev/tool/src/markup.ts index ec806797b5..0a4a89119c 100644 --- a/dev/tool/src/markup.ts +++ b/dev/tool/src/markup.ts @@ -20,42 +20,46 @@ import { yDocCopyXmlField, yDocFromBuffer } from '@hcengineering/collaboration' -import { withRetry } from '@hcengineering/retry' +import documents from '@hcengineering/controlled-documents' import core, { + type AnyAttribute, type Blob, + type Class, + DOMAIN_TX, type Doc, + type Domain, type Hierarchy, + type LowLevelStorage, type MeasureContext, + MeasureMetricsContext, + RateLimiter, type Ref, + SortingOrder, type Tx, type TxCreateDoc, type TxUpdateDoc, - DOMAIN_TX, - SortingOrder, type WorkspaceIds, + type WorkspaceUuid, + groupByArray, + isArchivingMode, + isDeletingMode, makeCollabId, makeCollabYdocId, makeDocCollabId, - MeasureMetricsContext, - systemAccountUuid, - isArchivingMode, - isDeletingMode, - type Domain, - type AnyAttribute, - type LowLevelStorage, - type Class, - RateLimiter, - type WorkspaceUuid, - groupByArray + systemAccountUuid } from '@hcengineering/core' import document, { type Document } from '@hcengineering/document' -import documents from '@hcengineering/controlled-documents' -import { DOMAIN_DOCUMENT } from '@hcengineering/model-document' import { DOMAIN_DOCUMENTS } from '@hcengineering/model-controlled-documents' +import { DOMAIN_DOCUMENT } from '@hcengineering/model-document' import { getDBClient } from '@hcengineering/postgres' -import { type PipelineFactory, type StorageAdapter, createDummyStorageAdapter } from '@hcengineering/server-core' +import { withRetry } from '@hcengineering/retry' import { getAccountClient } from '@hcengineering/server-client' -import { createBackupPipeline, sharedPipelineContextVars } from '@hcengineering/server-pipeline' +import { type PipelineFactory, type StorageAdapter, createDummyStorageAdapter } from '@hcengineering/server-core' +import { + createBackupPipeline, + createEmptyBroadcastOps, + sharedPipelineContextVars +} from '@hcengineering/server-pipeline' import { generateToken } from '@hcengineering/server-token' import { isEmptyMarkup } from '@hcengineering/text-core' @@ -424,7 +428,7 @@ export async function restoreMarkupRefs ( ctx.info('processing workspace', { uuid, name, url, region, classes }) try { - const pipeline = await factory(ctx, workspace, (): void => {}, null, null) + const pipeline = await factory(ctx, workspace, createEmptyBroadcastOps(), null) try { const { hierarchy, lowLevelStorage } = pipeline.context diff --git a/dev/tool/src/workspace.ts b/dev/tool/src/workspace.ts index 7d22191308..ef955129a3 100644 --- a/dev/tool/src/workspace.ts +++ b/dev/tool/src/workspace.ts @@ -39,6 +39,7 @@ import { import { createStorageFromConfig, storageConfigFromEnv } from '@hcengineering/server-storage' import { connect } from '@hcengineering/server-tool' import { generateModelDiff, printDiff } from './mdiff' +import { createEmptyBroadcastOps } from '@hcengineering/server-pipeline' export async function diffWorkspace (mongoUrl: string, dbName: string, rawTxes: Tx[]): Promise { const client = getMongoClient(mongoUrl) @@ -164,7 +165,7 @@ export async function backupRestore ( recheck: false, storageAdapter: workspaceStorage, getConnection: async () => { - return wrapPipeline(ctx, await pipelineFactory(ctx, wsUrl, () => {}, null, null), wsUrl) + return wrapPipeline(ctx, await pipelineFactory(ctx, wsUrl, createEmptyBroadcastOps(), null), wsUrl) } }) ) diff --git a/models/core/src/migration.ts b/models/core/src/migration.ts index c642a468ea..a6ff27c61a 100644 --- a/models/core/src/migration.ts +++ b/models/core/src/migration.ts @@ -927,7 +927,7 @@ export const coreOperation: MigrateOperation = { attachedTo: objectId, attachedToClass: objectClass, ...(tx as any).tx, - objectSpace: (tx as any).tx.objectSpace ?? tx.objectClass + objectSpace: (tx as any).tx.objectSpace ?? tx.objectSpace } }) ) diff --git a/packages/api-client/src/rest/tx.ts b/packages/api-client/src/rest/tx.ts index 46f365c059..e84e0039fd 100644 --- a/packages/api-client/src/rest/tx.ts +++ b/packages/api-client/src/rest/tx.ts @@ -19,10 +19,14 @@ import { type Client, type Doc, type DocumentQuery, + type DomainParams, + type DomainRequestOptions, + type DomainResult, type FindOptions, type FindResult, Hierarchy, ModelDb, + type OperationDomain, type Ref, type SearchOptions, type SearchQuery, @@ -72,6 +76,14 @@ class RestTxClient implements Client { return toFindResult(result, data.total) } + async domainRequest( + domain: OperationDomain, + params: DomainParams, + options?: DomainRequestOptions + ): Promise> { + throw new Error('Domain request operation not supported') + } + async findOne( _class: Ref>, query: DocumentQuery, diff --git a/packages/core/src/__tests__/client.test.ts b/packages/core/src/__tests__/client.test.ts index 1695232430..e4770b566b 100644 --- a/packages/core/src/__tests__/client.test.ts +++ b/packages/core/src/__tests__/client.test.ts @@ -22,7 +22,15 @@ import core from '../component' import { Hierarchy } from '../hierarchy' import { ModelDb, TxDb } from '../memdb' import { TxOperations } from '../operations' -import type { DocumentQuery, FindResult, SearchOptions, SearchQuery, SearchResult, TxResult } from '../storage' +import type { + DocumentQuery, + DomainResult, + FindResult, + SearchOptions, + SearchQuery, + SearchResult, + TxResult +} from '../storage' import { type Tx, TxFactory, TxProcessor } from '../tx' import { fillConfiguration, generateId, pluginFilterTx } from '../utils' import { connect } from './connection' @@ -123,6 +131,10 @@ describe('client', () => { findAll = findAll pushHandler = (): void => {} + domainRequest (): Promise { + return Promise.resolve({ domain: 'test' as Domain, value: null }) + } + searchFulltext = async (query: SearchQuery, options: SearchOptions): Promise => { return { docs: [] } } diff --git a/packages/core/src/__tests__/connection.ts b/packages/core/src/__tests__/connection.ts index 69d704a233..93f2775f59 100644 --- a/packages/core/src/__tests__/connection.ts +++ b/packages/core/src/__tests__/connection.ts @@ -19,7 +19,15 @@ import { type ClientConnection } from '../client' import core from '../component' import { Hierarchy } from '../hierarchy' import { ModelDb, TxDb } from '../memdb' -import type { DocumentQuery, FindResult, SearchOptions, SearchQuery, SearchResult, TxResult } from '../storage' +import type { + DocumentQuery, + DomainResult, + FindResult, + SearchOptions, + SearchQuery, + SearchResult, + TxResult +} from '../storage' import type { Tx } from '../tx' import { DOMAIN_TX } from '../tx' import { genMinModel } from './minmodel' @@ -74,6 +82,10 @@ export async function connect (handler: (tx: Tx) => void): Promise { + return await Promise.resolve({ domain: 'test' as Domain, value: null }) + } + async close (): Promise {} async loadChunk (domain: Domain, idx?: number): Promise { diff --git a/packages/core/src/__tests__/memdb.test.ts b/packages/core/src/__tests__/memdb.test.ts index 33333d2b13..9d4f3b1717 100644 --- a/packages/core/src/__tests__/memdb.test.ts +++ b/packages/core/src/__tests__/memdb.test.ts @@ -13,8 +13,8 @@ // limitations under the License. // -import { type Client } from '..' -import type { Class, Doc, Obj, Ref } from '../classes' +import { type Client, type DomainParams, type DomainRequestOptions, type DomainResult } from '..' +import type { Class, Doc, Obj, OperationDomain, Ref } from '../classes' import core from '../component' import { Hierarchy } from '../hierarchy' import { ModelDb, TxDb } from '../memdb' @@ -22,11 +22,11 @@ import { TxOperations } from '../operations' import { type DocumentQuery, type FindOptions, - SortingOrder, - type WithLookup, - type SearchQuery, type SearchOptions, - type SearchResult + type SearchQuery, + type SearchResult, + SortingOrder, + type WithLookup } from '../storage' import { type Tx } from '../tx' import { genMinModel, test, type TestMixin } from './minmodel' @@ -56,6 +56,14 @@ class ClientModel extends ModelDb implements Client { return { docs: [] } } + async domainRequest( + domain: OperationDomain, + params: DomainParams, + options?: DomainRequestOptions + ): Promise> { + return { domain, value: null as any } + } + async close (): Promise {} } diff --git a/packages/core/src/classes.ts b/packages/core/src/classes.ts index d963720b0c..80dd6e26a1 100644 --- a/packages/core/src/classes.ts +++ b/packages/core/src/classes.ts @@ -238,6 +238,11 @@ export interface Classifier extends Doc, UXObject { */ export type Domain = string & { __domain: true } +/** + * @public + */ +export type OperationDomain = string & { __domain: true } + /** * @public */ diff --git a/packages/core/src/client.ts b/packages/core/src/client.ts index 5a1e823615..99a2ccdf59 100644 --- a/packages/core/src/client.ts +++ b/packages/core/src/client.ts @@ -15,13 +15,23 @@ import { Analytics } from '@hcengineering/analytics' import { type BackupClient, type DocChunk } from './backup' -import { type Class, DOMAIN_MODEL, type Doc, type Domain, type Ref, type Timestamp } from './classes' +import { + type Class, + DOMAIN_MODEL, + type Doc, + type Domain, + type OperationDomain, + type Ref, + type Timestamp +} from './classes' import core from './component' import { Hierarchy } from './hierarchy' import { type MeasureContext, MeasureMetricsContext } from './measurements' import { ModelDb } from './memdb' import type { DocumentQuery, + DomainParams, + DomainResult, FindOptions, FindResult, FulltextStorage, @@ -40,6 +50,9 @@ import { platformNow, platformNowDiff, toFindResult } from './utils' */ export type TxHandler = (...tx: Tx[]) => void +export interface DomainRequestOptions { + retry?: boolean +} /** * @public */ @@ -53,7 +66,12 @@ export interface Client extends Storage, FulltextStorage { options?: FindOptions ) => Promise | undefined> close: () => Promise - getConnection?: () => ClientConnection + + domainRequest: ( + domain: OperationDomain, + params: DomainParams, + options?: DomainRequestOptions + ) => Promise> } /** @@ -81,8 +99,6 @@ export enum ClientConnectEvent { Maintenance // In case workspace are in maintenance mode } -export type Handler = (...result: any[]) => void - /** * @public */ @@ -94,9 +110,9 @@ export interface ClientConnection extends Storage, FulltextStorage, BackupClient // If hash is passed, will return LoadModelResponse loadModel: (last: Timestamp, hash?: string) => Promise - getLastHash?: (ctx: MeasureContext) => Promise - pushHandler: (handler: Handler) => void + pushHandler: (handler: TxHandler) => void + domainRequest: (ctx: OperationDomain, params: DomainParams, options?: DomainRequestOptions) => Promise } class ClientImpl implements Client, BackupClient { @@ -147,6 +163,14 @@ class ClientImpl implements Client, BackupClient { return await this.conn.searchFulltext(query, options) } + async domainRequest ( + ctx: OperationDomain, + params: DomainParams, + options?: DomainRequestOptions + ): Promise { + return await this.conn.domainRequest(ctx, params, options) + } + async findOne( _class: Ref>, query: DocumentQuery, diff --git a/packages/core/src/component.ts b/packages/core/src/component.ts index 3cf9fa75ca..58b01ceeb5 100644 --- a/packages/core/src/component.ts +++ b/packages/core/src/component.ts @@ -70,6 +70,7 @@ import type { TxApplyIf, TxCUD, TxCreateDoc, + TxDomainEvent, TxMixin, TxModelUpgrade, TxRemoveDoc, @@ -111,6 +112,7 @@ export default plugin(coreId, { Tx: '' as Ref>, TxModelUpgrade: '' as Ref>, TxWorkspaceEvent: '' as Ref>, + TxDomainEvent: '' as Ref>, TxApplyIf: '' as Ref>, TxCUD: '' as Ref>>, TxCreateDoc: '' as Ref>>, @@ -195,7 +197,8 @@ export default plugin(coreId, { Model: '' as Ref, Space: '' as Ref, Configuration: '' as Ref, - Workspace: '' as Ref + Workspace: '' as Ref, + Domain: '' as Ref }, account: { System: '' as PersonId, diff --git a/packages/core/src/operations.ts b/packages/core/src/operations.ts index 467b004900..8a182e5514 100644 --- a/packages/core/src/operations.ts +++ b/packages/core/src/operations.ts @@ -11,7 +11,6 @@ import { toFindResult } from '.' import type { - PersonId, AnyAttribute, AttachedData, AttachedDoc, @@ -19,6 +18,8 @@ import type { Data, Doc, Mixin, + OperationDomain, + PersonId, Ref, Space, Timestamp @@ -27,6 +28,8 @@ import { type Client } from './client' import core from './component' import type { DocumentQuery, + DomainParams, + DomainResult, FindOptions, FindResult, SearchOptions, @@ -44,7 +47,7 @@ import { type DocumentClassQuery, type Tx, type TxApplyResult, type TxCUD, TxFac * * `notify` is not supported by TxOperations. */ -export class TxOperations implements Omit { +export class TxOperations implements Omit { readonly txFactory: TxFactory constructor ( @@ -83,6 +86,10 @@ export class TxOperations implements Omit { return this.client.findOne(_class, query, options) } + domainRequest(domain: OperationDomain, params: DomainParams): Promise> { + return this.client.domainRequest(domain, params) + } + searchFulltext (query: SearchQuery, options: SearchOptions): Promise { return this.client.searchFulltext(query, options) } @@ -468,6 +475,7 @@ export class ApplyOperations extends TxOperations { findOne: (_class, query, options?) => ops.client.findOne(_class, query, options), findAll: (_class, query, options?) => ops.client.findAll(_class, query, options), searchFulltext: (query, options) => ops.client.searchFulltext(query, options), + domainRequest: (domain, params) => ops.client.domainRequest(domain, params), tx: async (tx): Promise => { if (TxProcessor.isExtendsCUD(tx._class)) { this.txes.push(tx as TxCUD) @@ -559,6 +567,7 @@ export class TxBuilder extends TxOperations { findOne: async (_class, query, options?) => undefined, findAll: async (_class, query, options?) => toFindResult([]), searchFulltext: async (query, options) => ({ docs: [] }), + domainRequest: async (domain, params) => ({ domain, value: null as any }), tx: async (tx): Promise => { if (TxProcessor.isExtendsCUD(tx._class)) { this.txes.push(tx as TxCUD) diff --git a/packages/core/src/server.ts b/packages/core/src/server.ts index f763fcff96..fbe68c57fb 100644 --- a/packages/core/src/server.ts +++ b/packages/core/src/server.ts @@ -38,12 +38,14 @@ export interface StorageIterator { close: (ctx: MeasureContext) => Promise } -export type BroadcastTargets = Record string[] | undefined> +export type BroadcastTargets = Record AccountUuid[] | undefined> export interface SessionData { broadcast: { txes: Tx[] targets: BroadcastTargets // A set of broadcast filters if required + queue: Tx[] // Queue only broadcast + sessions: Record // Session based broadcast } contextCache: Map removedMap: Map, Doc> diff --git a/packages/core/src/storage.ts b/packages/core/src/storage.ts index a1959df782..ae7ec7643c 100644 --- a/packages/core/src/storage.ts +++ b/packages/core/src/storage.ts @@ -16,7 +16,7 @@ import type { Asset, Resource } from '@hcengineering/platform' import type { KeysByType } from 'simplytyped' -import type { Association, AttachedDoc, Class, Doc, Ref, Space } from './classes' +import type { Association, AttachedDoc, Class, Doc, Domain, Ref, Space } from './classes' import type { Tx } from './tx' /** @@ -205,6 +205,13 @@ export type FindResult = WithLookup[] & { lookupMap?: Record } +export type DomainParams = Record + +export interface DomainResult { + domain: Domain + value: T +} + /** * @public */ diff --git a/packages/core/src/tx.ts b/packages/core/src/tx.ts index 3be9c812ba..cc40b62b0a 100644 --- a/packages/core/src/tx.ts +++ b/packages/core/src/tx.ts @@ -22,6 +22,7 @@ import type { Doc, Domain, Mixin, + OperationDomain, PersonId, PropertyType, Ref, @@ -63,6 +64,10 @@ export interface TxWorkspaceEvent extends Tx { event: WorkspaceEvent params: T } +export interface TxDomainEvent extends Tx { + domain: OperationDomain + event: T +} /** * @public diff --git a/packages/presentation/src/communication.ts b/packages/presentation/src/communication.ts index 2de394d714..721227a9ba 100644 --- a/packages/presentation/src/communication.ts +++ b/packages/presentation/src/communication.ts @@ -12,6 +12,33 @@ // See the License for the specific language governing permissions and // limitations under the License. // +import { initLiveQueries } from '@hcengineering/communication-client-query' +import { + type AddCollaboratorsEvent, + type AttachBlobsOperation, + type AttachLinkPreviewsOperation, + type BlobPatchEvent, + type CreateMessageEvent, + type CreateMessageResult, + type DetachBlobsOperation, + type DetachLinkPreviewsOperation, + type Event, + type EventResult, + type LinkPreviewPatchEvent, + MessageEventType, + NotificationEventType, + type ReactionPatchEvent, + type RemoveCollaboratorsEvent, + type RemoveNotificationContextEvent, + type RemovePatchEvent, + type SetBlobsOperation, + type SetLinkPreviewsOperation, + type ThreadPatchEvent, + type UpdateNotificationContextEvent, + type UpdateNotificationEvent, + type UpdateNotificationQuery, + type UpdatePatchEvent +} from '@hcengineering/communication-sdk-types' import { type AccountID, type BlobData, @@ -38,41 +65,22 @@ import { type NotificationContext, type SocialID } from '@hcengineering/communication-types' -import { - type AddCollaboratorsEvent, - type CreateMessageEvent, - type CreateMessageResult, - type EventResult, - type RemoveCollaboratorsEvent, - type RemoveNotificationContextEvent, - type UpdateNotificationContextEvent, - type UpdateNotificationEvent, - type UpdateNotificationQuery, - type Event, - type ThreadPatchEvent, - MessageEventType, - type UpdatePatchEvent, - type RemovePatchEvent, - type ReactionPatchEvent, - NotificationEventType, - type BlobPatchEvent, - type AttachBlobsOperation, - type DetachBlobsOperation, - type SetBlobsOperation, - type LinkPreviewPatchEvent, - type AttachLinkPreviewsOperation, - type DetachLinkPreviewsOperation, - type SetLinkPreviewsOperation -} from '@hcengineering/communication-sdk-types' -import { - type Client as PlatformClient, - type ClientConnection as PlatformConnection, +import core, { generateId, getCurrentAccount, - SocialIdType + type OperationDomain, + type Client as PlatformClient, + SocialIdType, + type Tx, + type TxDomainEvent } from '@hcengineering/core' import { onDestroy } from 'svelte' -import { + +import { generateLinkPreviewId } from '@hcengineering/communication-shared' +import { getCurrentWorkspaceUuid, getFilesUrl } from './file' +import { addTxListener, removeTxListener, type TxListener } from './utils' + +export { createCollaboratorsQuery, createLabelsQuery, createMessagesQuery, @@ -81,29 +89,6 @@ import { initLiveQueries, type MessageQueryParams } from '@hcengineering/communication-client-query' -import { generateLinkPreviewId } from '@hcengineering/communication-shared' - -import { getCurrentWorkspaceUuid, getFilesUrl } from './file' - -export { - createMessagesQuery, - createNotificationsQuery, - createNotificationContextsQuery, - createLabelsQuery, - createCollaboratorsQuery -} -export type { MessageQueryParams } - -interface Connection extends PlatformConnection { - findMessages: (params: FindMessagesParams, queryId?: number) => Promise - findMessagesGroups: (params: FindMessagesGroupsParams) => Promise - findNotificationContexts: (params: FindNotificationContextParams, queryId?: number) => Promise - findNotifications: (params: FindNotificationsParams, queryId?: number) => Promise - findLabels: (params: FindLabelsParams) => Promise - findCollaborators: (params: FindCollaboratorsParams, queryId?: number) => Promise - sendEvent: (event: Event) => Promise - unsubscribeQuery: (id: number) => Promise -} let client: CommunicationClient @@ -114,11 +99,10 @@ export function getCommunicationClient (): CommunicationClient { } export async function setCommunicationClient (platformClient: PlatformClient): Promise { - const connection = platformClient.getConnection?.() - if (connection === undefined) { - return + if (client !== undefined) { + client.close() } - const _client = new Client(connection as unknown as Connection) + const _client = new Client(platformClient) initLiveQueries(_client, getCurrentWorkspaceUuid(), getFilesUrl(), onDestroy) client = _client onClientListeners.forEach((fn) => { @@ -126,15 +110,22 @@ export async function setCommunicationClient (platformClient: PlatformClient): P }) } +const COMMUNICATION = 'communication' as OperationDomain + class Client { - constructor (private readonly connection: Connection) { - connection.pushHandler((...events: any[]) => { - for (const event of events) { - if (event != null && 'type' in event) { - this.onEvent(event as Event) - } + txHandler: TxListener + constructor (private readonly connection: PlatformClient) { + this.txHandler = this.doHandleEvents.bind(this) + addTxListener(this.txHandler) + } + + doHandleEvents (events: Tx[]): void { + for (const event of events) { + if (event._class === core.class.TxDomainEvent && (event as TxDomainEvent).domain === COMMUNICATION) { + const evt = event as TxDomainEvent + this.onEvent(evt.event) } - }) + } } onEvent: (event: Event) => void = () => {} @@ -375,43 +366,74 @@ class Client { } async findMessages (params: FindMessagesParams, queryId?: number): Promise { - return await this.connection.findMessages(params, queryId) + return ( + await this.connection.domainRequest(COMMUNICATION, { + findMessages: { params, queryId } + }) + ).value } async findMessagesGroups (params: FindMessagesGroupsParams): Promise { - return await this.connection.findMessagesGroups(params) + return ( + await this.connection.domainRequest(COMMUNICATION, { + findMessagesGroups: { params } + }) + ).value } async findNotificationContexts ( params: FindNotificationContextParams, queryId?: number ): Promise { - return await this.connection.findNotificationContexts(params, queryId) + return ( + await this.connection.domainRequest(COMMUNICATION, { + findNotificationContexts: { params, queryId } + }) + ).value } async findNotifications (params: FindNotificationsParams, queryId?: number): Promise { - return await this.connection.findNotifications(params, queryId) + return ( + await this.connection.domainRequest(COMMUNICATION, { + findNotifications: { params, queryId } + }) + ).value } async findLabels (params: FindLabelsParams): Promise { - return await this.connection.findLabels(params) + return ( + await this.connection.domainRequest(COMMUNICATION, { + findLabels: { params } + }) + ).value } - async findCollaborators (params: FindCollaboratorsParams, queryId?: number): Promise { - return await this.connection.findCollaborators(params, queryId) + async findCollaborators (params: FindCollaboratorsParams): Promise { + return ( + await this.connection.domainRequest(COMMUNICATION, { + findCollaborators: { params } + }) + ).value } async unsubscribeQuery (id: number): Promise { - await this.connection.unsubscribeQuery(id) + await this.connection.domainRequest(COMMUNICATION, { + unsubscribeQuery: id + }) } close (): void { - // do nothing + removeTxListener(this.txHandler) } private async sendEvent (event: Event): Promise { const ev: Event = { ...event, _id: generateId() } - const eventPromise = this.connection.sendEvent(ev) + + const eventPromise: Promise = this.connection + .domainRequest(COMMUNICATION, { + event: ev + }) + .then((result) => result.value) this.onRequest(ev, eventPromise) return await eventPromise } diff --git a/packages/presentation/src/pipeline.ts b/packages/presentation/src/pipeline.ts index b1c17c9069..6a6df593ab 100644 --- a/packages/presentation/src/pipeline.ts +++ b/packages/presentation/src/pipeline.ts @@ -4,10 +4,14 @@ import { type Client, type Doc, type DocumentQuery, + type DomainParams, + type DomainRequestOptions, + type DomainResult, type FindOptions, type FindResult, type Hierarchy, type ModelDb, + type OperationDomain, type QuerySelector, type Ref, type SearchOptions, @@ -35,6 +39,12 @@ export interface PresentationMiddleware { options?: FindOptions ) => Promise> + domainRequest: ( + domain: OperationDomain, + params: DomainParams, + options?: DomainRequestOptions + ) => Promise> + findOne: ( _class: Ref>, query: DocumentQuery, @@ -102,6 +112,27 @@ export class PresentationPipelineImpl implements PresentationPipeline { return current } + async domainRequest( + domain: OperationDomain, + params: DomainParams, + options?: DomainRequestOptions + ): Promise> { + try { + return this.head !== undefined + ? await this.head.domainRequest(domain, params, options) + : await this.client.domainRequest(domain, params, options) + } catch (err: any) { + if (err instanceof PlatformError) { + if (err.status.code === platform.status.ConnectionClosed) { + return { domain, value: null as any } + } + } + const status = unknownError(err) + await setPlatformStatus(status) + return { domain, value: null as any } + } + } + async findAll( _class: Ref>, query: DocumentQuery, @@ -269,6 +300,25 @@ export abstract class BasePresentationMiddleware { return await this.client.findOne(_class, query, options) } + async domainRequest( + domain: OperationDomain, + params: DomainParams, + options?: DomainRequestOptions + ): Promise> { + return await this.provideDomainRequest(domain, params, options) + } + + protected async provideDomainRequest( + domain: OperationDomain, + params: DomainParams, + options?: DomainRequestOptions + ): Promise> { + if (this.next !== undefined) { + return await this.next.domainRequest(domain, params, options) + } + return await this.client.domainRequest(domain, params, options) + } + protected async provideSubscribe( _class: Ref>, query: DocumentQuery, @@ -317,6 +367,14 @@ export class OptimizeQueryMiddleware extends BasePresentationMiddleware implemen return await this.provideTx(tx) } + async domainRequest( + domain: OperationDomain, + params: DomainParams, + options?: DomainRequestOptions + ): Promise> { + return await this.provideDomainRequest(domain, params, options) + } + async subscribe( _class: Ref>, query: DocumentQuery, diff --git a/packages/presentation/src/plugin.ts b/packages/presentation/src/plugin.ts index a3f191c19b..5921067d7c 100644 --- a/packages/presentation/src/plugin.ts +++ b/packages/presentation/src/plugin.ts @@ -15,26 +15,31 @@ // import { - type WorkspaceDataId, - type WorkspaceUuid, type Class, type Client, type Doc, type DocumentQuery, + type DomainParams, + type DomainRequestOptions, + type DomainResult, type FindOptions, type FindResult, type Mixin, + type OperationDomain, type Ref, type SearchOptions, type SearchQuery, type SearchResult, type Tx, type TxResult, - type WithLookup + type WithLookup, + type WorkspaceDataId, + type WorkspaceUuid } from '@hcengineering/core' import type { Asset, IntlString, Metadata, Plugin, StatusCode } from '@hcengineering/platform' import { plugin } from '@hcengineering/platform' import { type ComponentExtensionId } from '@hcengineering/ui/src/types' +import { type UploadConfig } from './file' import { type PresentationMiddlewareFactory } from './pipeline' import type { PreviewConfig } from './preview' import { @@ -45,7 +50,6 @@ import { type InstantTransactions, type ObjectSearchCategory } from './types' -import { type UploadConfig } from './file' /** * @public @@ -70,6 +74,13 @@ export interface ClientHook { options?: FindOptions ) => Promise | undefined> + domainRequest: ( + client: Client, + domain: OperationDomain, + params: DomainParams, + options?: DomainRequestOptions + ) => Promise> + tx: (client: Client, tx: Tx) => Promise searchFulltext: (client: Client, query: SearchQuery, options: SearchOptions) => Promise diff --git a/packages/presentation/src/utils.ts b/packages/presentation/src/utils.ts index 58ed908355..c8a71d14a8 100644 --- a/packages/presentation/src/utils.ts +++ b/packages/presentation/src/utils.ts @@ -22,10 +22,12 @@ import core, { type AttachedDoc, type Class, type Client, - type ClientConnection, type Collection, type Doc, type DocumentQuery, + type DomainParams, + type DomainRequestOptions, + type DomainResult, type FindOptions, type FindResult, getCurrentAccount, @@ -34,6 +36,7 @@ import core, { type Mixin, type ModelDb, type Obj, + type OperationDomain, reduceCalls, type Ref, type RefTo, @@ -61,7 +64,7 @@ import { onDestroy } from 'svelte' import { get, writable } from 'svelte/store' import { type KeyedAttribute } from '..' -import { OptimizeQueryMiddleware, PresentationPipelineImpl, type PresentationPipeline } from './pipeline' +import { OptimizeQueryMiddleware, type PresentationPipeline, PresentationPipelineImpl } from './pipeline' import plugin, { type ClientHook } from './plugin' export { reduceCalls } from '@hcengineering/core' @@ -304,10 +307,6 @@ class ClientHookImpl implements Client { return await this.client.findOne(_class, query, options) } - get getConnection (): (() => ClientConnection) | undefined { - return this.client.getConnection - } - async close (): Promise { await this.client.close() } @@ -323,6 +322,17 @@ class ClientHookImpl implements Client { return await this.client.findAll(_class, query, options) } + async domainRequest( + domain: OperationDomain, + params: DomainParams, + options?: DomainRequestOptions + ): Promise> { + if (this.hook !== undefined) { + return await this.hook.domainRequest(this.client, domain, params, options) + } + return await this.client.domainRequest(domain, params, options) + } + async tx (tx: Tx): Promise { if (this.hook !== undefined) { return await this.hook.tx(this.client, tx) diff --git a/packages/query/src/__tests__/connection.ts b/packages/query/src/__tests__/connection.ts index b09e3894d9..6c3acc7b65 100644 --- a/packages/query/src/__tests__/connection.ts +++ b/packages/query/src/__tests__/connection.ts @@ -28,7 +28,6 @@ import core, { FindResult, FulltextStorage, generateId, - Handler, Hierarchy, LoadModelResponse, ModelDb, @@ -39,7 +38,12 @@ import core, { Timestamp, Tx, TxDb, - TxResult + TxResult, + type DomainParams, + type DomainRequestOptions, + type DomainResult, + type OperationDomain, + type TxHandler } from '@hcengineering/core' import { genMinModel } from './minmodel' @@ -49,7 +53,7 @@ BackupClient & FulltextStorage & { isConnected: () => boolean loadModel: (last: Timestamp, hash?: string) => Promise - pushHandler: (handler: Handler) => void + pushHandler: (handler: TxHandler) => void } > { const txes = genMinModel() @@ -99,6 +103,14 @@ FulltextStorage & { return (await this.findAll(_class, query, { ...options, limit: 1 })).shift() } + async domainRequest ( + domain: OperationDomain, + params: DomainParams, + options?: DomainRequestOptions + ): Promise { + return { domain, value: null } + } + getHierarchy (): Hierarchy { return this.hierarchy } diff --git a/packages/query/src/index.ts b/packages/query/src/index.ts index 0836e996c1..d119cdfa5a 100644 --- a/packages/query/src/index.ts +++ b/packages/query/src/index.ts @@ -58,7 +58,11 @@ import core, { platformNow, reduceCalls, shouldShowArchived, - toFindResult + toFindResult, + type DomainParams, + type DomainRequestOptions, + type DomainResult, + type OperationDomain } from '@hcengineering/core' import { PlatformError } from '@hcengineering/platform' import { deepEqual } from 'fast-equals' @@ -230,6 +234,14 @@ export class LiveQuery implements WithTx, Client { return toFindResult(q.result.getClone(), q.total) } + async domainRequest( + domain: OperationDomain, + params: DomainParams, + options?: DomainRequestOptions + ): Promise> { + return await this.client.domainRequest(domain, params, options) + } + searchFulltext (query: SearchQuery, options: SearchOptions): Promise { return this.client.searchFulltext(query, options) } diff --git a/plugins/client-resources/src/connection.ts b/plugins/client-resources/src/connection.ts index e77aab8d3b..ba72e49f24 100644 --- a/plugins/client-resources/src/connection.ts +++ b/plugins/client-resources/src/connection.ts @@ -28,17 +28,20 @@ import core, { ClientConnectEvent, ClientConnection, clone, - Handler, Doc, DocChunk, DocumentQuery, Domain, + type DomainParams, + type DomainRequestOptions, + type DomainResult, FindOptions, FindResult, generateId, LoadModelResponse, type MeasureContext, MeasureMetricsContext, + type OperationDomain, type PersonUuid, Ref, SearchOptions, @@ -60,22 +63,8 @@ import platform, { Status, UNAUTHORIZED } from '@hcengineering/platform' +import { HelloRequest, HelloResponse, type RateLimitInfo, ReqId, type Response, RPCHandler } from '@hcengineering/rpc' import { uncompress } from 'snappyjs' -import { HelloRequest, HelloResponse, ReqId, type Response, RPCHandler, type RateLimitInfo } from '@hcengineering/rpc' -import { EventResult } from '@hcengineering/communication-sdk-types' -import { - FindLabelsParams, - FindMessagesGroupsParams, - FindMessagesParams, - FindNotificationContextParams, - FindNotificationsParams, - FindCollaboratorsParams, - Label, - Message, - MessagesGroup, - NotificationContext, - Collaborator -} from '@hcengineering/communication-types' const SECOND = 1000 const pingTimeout = 10 * SECOND @@ -142,7 +131,7 @@ class Connection implements ClientConnection { lastHash?: string - handlers: Handler[] = [] + handlers: TxHandler[] = [] constructor ( private readonly ctx: MeasureContext, @@ -177,7 +166,7 @@ class Connection implements ClientConnection { this.scheduleOpen(this.ctx, false) } - pushHandler (handler: Handler): void { + pushHandler (handler: TxHandler): void { this.handlers.push(handler) } @@ -908,44 +897,19 @@ class Connection implements ClientConnection { return this.sendRequest({ method: 'searchFulltext', params: [query, options] }) } + domainRequest (domain: OperationDomain, params: DomainParams, options?: DomainRequestOptions): Promise { + return this.sendRequest({ + method: 'domainRequest', + params: [domain, params], + retry: async () => { + return options?.retry ?? false + } + }) + } + sendForceClose (): Promise { return this.sendRequest({ method: 'forceClose', params: [], allowReconnect: false, overrideId: -2, once: true }) } - - async sendEvent (event: Event): Promise { - return await this.sendRequest({ method: 'event', params: [event] }) - } - - async findMessages (params: FindMessagesParams, queryId?: number): Promise { - return await this.sendRequest({ method: 'findMessages', params: [params, queryId] }) - } - - async findLabels (params: FindLabelsParams): Promise { - return await this.sendRequest({ method: 'findLabels', params: [params] }) - } - - async findMessagesGroups (params: FindMessagesGroupsParams): Promise { - return await this.sendRequest({ method: 'findMessagesGroups', params: [params] }) - } - - async findNotificationContexts ( - params: FindNotificationContextParams, - queryId?: number - ): Promise { - return await this.sendRequest({ method: 'findNotificationContexts', params: [params, queryId] }) - } - - async findNotifications (params: FindNotificationsParams, queryId?: number): Promise { - return await this.sendRequest({ method: 'findNotifications', params: [params, queryId] }) - } - - async findCollaborators (params: FindCollaboratorsParams): Promise { - return await this.sendRequest({ method: 'findCollaborators', params: [params] }) - } - - async unsubscribeQuery (id: number): Promise { - await this.sendRequest({ method: 'unsubscribeQuery', params: [id] }) - } } /** diff --git a/plugins/devmodel-resources/src/index.ts b/plugins/devmodel-resources/src/index.ts index 7caed91a26..90ff300fb8 100644 --- a/plugins/devmodel-resources/src/index.ts +++ b/plugins/devmodel-resources/src/index.ts @@ -22,8 +22,12 @@ import core, { type Client, type Doc, type DocumentQuery, + type DomainParams, + type DomainRequestOptions, + type DomainResult, type FindOptions, type FindResult, + type OperationDomain, type Ref, type SearchOptions, type SearchQuery, @@ -137,6 +141,26 @@ export class PresentationClientHook implements ClientHook { return result } + async domainRequest ( + client: Client, + domain: OperationDomain, + params: DomainParams, + options?: DomainRequestOptions + ): Promise { + const result = await client.domainRequest(domain, params) + if (this.notifyEnabled) { + console.debug( + 'devmodel# domainRequest=>', + domain, + testing ? JSON.stringify(cutObjectArray(params)).slice(0, 160) : params, + options, + 'result => ', + result + ) + } + return result + } + async searchFulltext (client: Client, query: SearchQuery, options: SearchOptions): Promise { const result = await client.searchFulltext(query, options) if (this.notifyEnabled) { diff --git a/pods/fulltext/src/__tests__/utils.ts b/pods/fulltext/src/__tests__/utils.ts index 308a381f9d..18cb24e0f0 100644 --- a/pods/fulltext/src/__tests__/utils.ts +++ b/pods/fulltext/src/__tests__/utils.ts @@ -29,6 +29,7 @@ import { type PlatformQueue } from '@hcengineering/server-core' import { + createEmptyBroadcastOps, getConfig, registerAdapterFactory, registerDestroyFactory, @@ -85,7 +86,7 @@ export async function preparePipeline ( DBAdapterInitMiddleware.create, ModelMiddleware.create(model), DBAdapterMiddleware.create(conf), // Configure DB adapters - ...(useBroadcast ? [BroadcastMiddleware.create((ctx, tx) => {})] : []) + ...(useBroadcast ? [BroadcastMiddleware.create(createEmptyBroadcastOps())] : []) ] const hierarchy = new Hierarchy() @@ -96,8 +97,7 @@ export async function preparePipeline ( modelDb, hierarchy, storageAdapter: storage, - contextVars: {}, - communicationApi: null + contextVars: {} } const pipeline = await createPipeline(toolCtx, middlewares, context) return { pipeline, wsIds } diff --git a/pods/fulltext/src/workspace.ts b/pods/fulltext/src/workspace.ts index 7f00ad62fd..54d38e7201 100644 --- a/pods/fulltext/src/workspace.ts +++ b/pods/fulltext/src/workspace.ts @@ -87,9 +87,7 @@ export class WorkspaceIndexer { modelDb, hierarchy, storageAdapter: externalStorage, - contextVars: {}, - // TODO: Communication API ?? - communicationApi: null + contextVars: {} } result.pipeline = await createPipeline(ctx, middlewares, context) diff --git a/pods/server/src/__tests__/server.test.ts b/pods/server/src/__tests__/server.test.ts index 1e8f751a7d..a508002f91 100644 --- a/pods/server/src/__tests__/server.test.ts +++ b/pods/server/src/__tests__/server.test.ts @@ -81,6 +81,8 @@ describe('server', () => { [], undefined ], + domainRequest: async (ctx, domain, params) => ({ domain, value: null as any }), + closeSession: async (ctx, sessionId) => {}, close: async () => {}, domains: async () => [], groupBy: async () => new Map(), @@ -97,9 +99,6 @@ describe('server', () => { loadModel: async (ctx, lastModelTx, hash) => [] } }, - communicationApiFactory: async () => { - return {} as any - }, brandingMap: {}, accountsUrl: '', queue: createDummyQueue() @@ -199,6 +198,8 @@ describe('server', () => { [], undefined ], + domainRequest: async (ctx, domain, params) => ({ domain, value: null as any }), + closeSession: async (ctx, sessionId) => {}, groupBy: async () => new Map(), close: async () => {}, domains: async () => [], @@ -215,9 +216,6 @@ describe('server', () => { loadModel: async (ctx, lastModelTx, hash) => [] } }, - communicationApiFactory: async () => { - return {} as any - }, brandingMap: {}, accountsUrl: '', queue: createDummyQueue() diff --git a/pods/server/src/rpc.ts b/pods/server/src/rpc.ts index f6907fd912..53599b1a78 100644 --- a/pods/server/src/rpc.ts +++ b/pods/server/src/rpc.ts @@ -1,31 +1,33 @@ -import core, { - buildSocialIdString, - generateId, - systemAccountUuid, - pickPrimarySocialId, - TxFactory, - TxProcessor, - type AttachedData, - type Data, - type Class, - type Doc, - type MeasureContext, - type Ref, - type SearchOptions, - type SearchQuery, - type TxCUD -} from '@hcengineering/core' -import type { ClientSessionCtx, ConnectionSocket, Session, SessionManager } from '@hcengineering/server-core' -import { decodeToken } from '@hcengineering/server-token' -import { rpcJSONReplacer, type RateLimitInfo } from '@hcengineering/rpc' +import { getClient as getAccountClientRaw, type AccountClient } from '@hcengineering/account-client' import contact, { AvatarType, combineName, - type SocialIdentity, type Person, + type SocialIdentity, type SocialIdentityRef } from '@hcengineering/contact' -import { type AccountClient, getClient as getAccountClientRaw } from '@hcengineering/account-client' +import core, { + buildSocialIdString, + generateId, + pickPrimarySocialId, + systemAccountUuid, + TxFactory, + TxProcessor, + type AttachedData, + type Class, + type Data, + type Doc, + type MeasureContext, + type OperationDomain, + type Ref, + type SearchOptions, + type SearchQuery, + type TxCUD, + type TxDomainEvent +} from '@hcengineering/core' +import { rpcJSONReplacer, type RateLimitInfo } from '@hcengineering/rpc' +import type { ClientSessionCtx, ConnectionSocket, Session, SessionManager } from '@hcengineering/server-core' +import { decodeToken } from '@hcengineering/server-token' import { createHash } from 'crypto' import { type Express, type Response as ExpressResponse, type Request } from 'express' @@ -36,6 +38,8 @@ import { gzip } from 'zlib' import { retrieveJson } from './utils' import { unknownError } from '@hcengineering/platform' + +export const COMMUNICATION_DOMAIN = 'communication' as OperationDomain interface RPCClientInfo { client: ConnectionSocket session: Session @@ -247,8 +251,16 @@ export function registerRPC (app: Express, sessions: SessionManager, ctx: Measur void withSession(req, res, async (ctx, session, rateLimit) => { const tx: any = (await retrieveJson(req)) ?? {} - const result = await session.txRaw(ctx, tx) - await sendJson(req, res, result.result, rateLimitToHeaders(rateLimit)) + if (tx._class === core.class.TxDomainEvent) { + const domainTx = tx as TxDomainEvent + const result = await session.domainRequestRaw(ctx, domainTx.domain, { + event: domainTx.event + }) + await sendJson(req, res, result.value, rateLimitToHeaders(rateLimit)) + } else { + const result = await session.txRaw(ctx, tx) + await sendJson(req, res, result.result, rateLimitToHeaders(rateLimit)) + } }) }) app.get('/api/v1/account/:workspaceId', (req, res) => { @@ -301,28 +313,17 @@ export function registerRPC (app: Express, sessions: SessionManager, ctx: Measur }) }) - app.get('/api/v1/find-messages/:workspaceId', (req, res) => { + app.get('/api/v1/request/:domain/:operation/:workspaceId', (req, res) => { void withSession(req, res, async (ctx, session) => { + const domain = req.params.domain as OperationDomain + const operation = req.params.operation + const params = req.query.params !== undefined ? JSON.parse(req.query.params as string) : {} - const result = await session.findMessagesRaw(ctx, params) - await sendJson(req, res, result) - }) - }) - app.get('/api/v1/find-messages-groups/:workspaceId', (req, res) => { - void withSession(req, res, async (ctx, session) => { - const params = req.query.params !== undefined ? JSON.parse(req.query.params as string) : {} - - const result = await session.findMessagesGroupsRaw(ctx, params) - await sendJson(req, res, result) - }) - }) - app.post('/api/v1/event/:workspaceId', (req, res) => { - void withSession(req, res, async (ctx, session) => { - const event: any = (await retrieveJson(req)) ?? {} - - const result = await session.eventRaw(ctx, event) - await sendJson(req, res, result) + const result = await session.domainRequestRaw(ctx, domain, { + [operation]: { params } + }) + await sendJson(req, res, result.value) }) }) diff --git a/pods/server/src/server.ts b/pods/server/src/server.ts index 3fb68c94ff..8b752cd324 100644 --- a/pods/server/src/server.ts +++ b/pods/server/src/server.ts @@ -111,13 +111,6 @@ export function start ( const externalStorage = buildStorageFromConfig(opt.storageConfig) - const pipelineFactory = createServerPipeline( - metrics, - dbUrl, - model, - { ...opt, externalStorage, adapterSecurity: isAdapterSecurity(dbUrl), queue: opt.queue }, - {} - ) const communicationApiFactory: CommunicationApiFactory = async (ctx, workspace, broadcastSessions) => { if (dbUrl.startsWith('mongodb') || !opt.communicationApiEnabled) { return { @@ -140,13 +133,19 @@ export function start ( ctx.newChild('💬 communication api', {}), workspace.uuid, dbUrl, - broadcastSessions as any // FIXME when communication will be inside the repo + broadcastSessions ) } + const pipelineFactory = createServerPipeline( + metrics, + dbUrl, + model, + { ...opt, externalStorage, adapterSecurity: isAdapterSecurity(dbUrl), queue: opt.queue, communicationApiFactory }, + {} + ) const sessionManager = startSessionManager(metrics, { pipelineFactory, - communicationApiFactory, brandingMap: opt.brandingMap, enableCompression: opt.enableCompression, accountsUrl: opt.accountsUrl, diff --git a/server-plugins/activity-resources/src/newActivity.ts b/server-plugins/activity-resources/src/newActivity.ts index e0a55b48e3..af6295e005 100644 --- a/server-plugins/activity-resources/src/newActivity.ts +++ b/server-plugins/activity-resources/src/newActivity.ts @@ -2,8 +2,8 @@ import core, { type Class, type Doc, type Hierarchy, + type OperationDomain, type Ref, - systemAccount, type TxCreateDoc, type TxCUD, TxProcessor @@ -12,7 +12,7 @@ import { type Card } from '@hcengineering/card' import { type TriggerControl } from '@hcengineering/server-core' import activity from '@hcengineering/activity' import { type ActivityControl } from '@hcengineering/server-activity' -import { type ServerApi as CommunicationApi, MessageEventType } from '@hcengineering/communication-sdk-types' +import { MessageEventType, type CreateMessageEvent } from '@hcengineering/communication-sdk-types' import { type ActivityAttributeUpdate, type ActivityMessageExtra, @@ -29,9 +29,8 @@ export async function generateActivity ( control: TriggerControl, cache: Map, Card> ): Promise { - const { hierarchy, communicationApi } = control + const { hierarchy } = control - if (communicationApi == null) return if (tx.space === core.space.DerivedTx) return if ( @@ -44,7 +43,7 @@ export async function generateActivity ( switch (tx._class) { case core.class.TxCreateDoc: { const card = TxProcessor.createDoc2Doc(tx as TxCreateDoc) - await createMessages(tx, control, card, communicationApi) + await createMessages(tx, control, card) break } case core.class.TxMixin: @@ -54,18 +53,13 @@ export async function generateActivity ( (await control.findAll(control.ctx, tx.objectClass, { _id: tx.objectId }, { limit: 1 }))[0] if (card !== undefined) { cache.set(tx.objectId, card) - await createMessages(tx, control, card, communicationApi) + await createMessages(tx, control, card) } } } } -async function createMessages ( - tx: TxCUD, - control: TriggerControl, - card: Card | undefined, - api: CommunicationApi -): Promise { +async function createMessages (tx: TxCUD, control: TriggerControl, card: Card | undefined): Promise { if (card === undefined) return const action = getActivityAction(control, tx) @@ -82,21 +76,18 @@ async function createMessages ( } for (const data of result) { - void api.event( - { - account: systemAccount - }, - { - type: MessageEventType.CreateMessage, - messageType: MessageType.Activity, - cardId: card._id, - cardType: card._class, - content: await getActivityContent(control, data, card), - socialId: tx.modifiedBy, - extra: data, - date: new Date(tx.modifiedOn) - } - ) + const event: CreateMessageEvent = { + type: MessageEventType.CreateMessage, + messageType: MessageType.Activity, + cardId: card._id, + cardType: card._class, + content: await getActivityContent(control, data, card), + socialId: tx.modifiedBy, + extra: data, + date: new Date(tx.modifiedOn) + } + + void control.domainRequest(control.ctx, 'communication' as OperationDomain, { event }) } } diff --git a/server-plugins/card-resources/src/index.ts b/server-plugins/card-resources/src/index.ts index f5b4bce89f..93d4489a60 100644 --- a/server-plugins/card-resources/src/index.ts +++ b/server-plugins/card-resources/src/index.ts @@ -22,9 +22,9 @@ import core, { fillDefaults, getDiffUpdate, Mixin, + OperationDomain, Ref, splitMixinUpdate, - systemAccount, Tx, TxCreateDoc, TxMixin, @@ -35,7 +35,13 @@ import core, { import { TriggerControl } from '@hcengineering/server-core' import setting from '@hcengineering/setting' import view from '@hcengineering/view' -import { CardEventType, NotificationEventType } from '@hcengineering/communication-sdk-types' +import { + AddCollaboratorsEvent, + CardEventType, + NotificationEventType, + RemoveCardEvent, + UpdateCardTypeEvent +} from '@hcengineering/communication-sdk-types' import { getEmployee, getPersonSpaces } from '@hcengineering/server-contact' import contact from '@hcengineering/contact' @@ -277,15 +283,16 @@ async function OnCardRemove (ctx: TxRemoveDoc[], control: TriggerControl): ) } - void control.communicationApi?.event( - { account: systemAccount }, - { - type: CardEventType.RemoveCard, - cardId: removedCard._id, - date: new Date(removeTx.createdOn ?? removeTx.modifiedOn), - socialId: removedCard.modifiedBy - } - ) + const event: RemoveCardEvent = { + type: CardEventType.RemoveCard, + cardId: removedCard._id, + date: new Date(removeTx.createdOn ?? removeTx.modifiedOn), + socialId: removedCard.modifiedBy + } + + void control.domainRequest(control.ctx, 'communication' as OperationDomain, { + event + }) return res } @@ -348,16 +355,16 @@ async function OnCardUpdate (ctx: TxUpdateDoc[], control: TriggerControl): res.push(...(await updateParentInfoName(control, doc._id, updateTx.operations.title, doc._id))) } if ((updateTx.operations as any)._class !== undefined) { - void control.communicationApi?.event( - { account: systemAccount as any }, - { - type: CardEventType.UpdateCardType, - cardId: doc._id, - cardType: (updateTx.operations as any)._class, - socialId: updateTx.createdBy ?? updateTx.modifiedBy, - date: new Date(updateTx.createdOn ?? updateTx.modifiedOn) - } - ) + const event: UpdateCardTypeEvent = { + type: CardEventType.UpdateCardType, + cardId: doc._id, + cardType: (updateTx.operations as any)._class, + socialId: updateTx.createdBy ?? updateTx.modifiedBy, + date: new Date(updateTx.createdOn ?? updateTx.modifiedOn) + } + void control.domainRequest(control.ctx, 'communication' as OperationDomain, { + event + }) } return res @@ -427,9 +434,6 @@ async function OnCardCreate (ctx: TxCreateDoc[], control: TriggerControl): } async function updateCollaborators (control: TriggerControl, ctx: TxCreateDoc[]): Promise { - const { communicationApi } = control - if (communicationApi == null) return - for (const tx of ctx) { const modifier = await getEmployee(control, tx.modifiedBy) const collaborators: AccountUuid[] = [] @@ -448,17 +452,17 @@ async function updateCollaborators (control: TriggerControl, ctx: TxCreateDoc { if (pipeline === undefined) { - pipeline = await this.pipelineFactory(ctx, wsIds, () => {}, null, null) + pipeline = await this.pipelineFactory( + ctx, + wsIds, + { + broadcast: () => {}, + broadcastSessions: () => {} + }, + null + ) } return wrapPipeline(ctx, pipeline, wsIds) }, @@ -500,7 +508,15 @@ export async function doRestoreWorkspace ( cleanIndexState, getConnection: async () => { if (pipeline === undefined) { - pipeline = await pipelineFactory(ctx, wsIds, () => {}, null, null) + pipeline = await pipelineFactory( + ctx, + wsIds, + { + broadcast: () => {}, + broadcastSessions: () => {} + }, + null + ) } return wrapPipeline(ctx, pipeline, wsIds) }, diff --git a/server/core/src/base.ts b/server/core/src/base.ts index c23e7950e7..85133ca02e 100644 --- a/server/core/src/base.ts +++ b/server/core/src/base.ts @@ -18,10 +18,12 @@ import { type Doc, type DocumentQuery, type Domain, + type DomainParams, type FindOptions, type FindResult, type LoadModelResponse, type MeasureContext, + type OperationDomain, type Ref, type SearchOptions, type SearchQuery, @@ -31,6 +33,7 @@ import { toFindResult, type Tx } from '@hcengineering/core' +import type { DomainResult } from '@hcengineering/core/src/storage' import type { Middleware, PipelineContext, TxMiddlewareResult } from './types' export const emptyFindResult = Promise.resolve(toFindResult([])) @@ -141,4 +144,30 @@ export abstract class BaseMiddleware implements Middleware { } return emptySearchResult } + + domainRequest (ctx: MeasureContext, domain: OperationDomain, params: DomainParams): Promise { + return this.provideDomainRequest(ctx, domain, params) + } + + protected async provideDomainRequest ( + ctx: MeasureContext, + domain: OperationDomain, + params: DomainParams + ): Promise { + if (this.next !== undefined) { + return await this.next.domainRequest(ctx, domain, params) + } + return { domain, value: null } + } + + provideCloseSession (ctx: MeasureContext, sessionId: string): Promise { + if (this.next !== undefined) { + return this.next.closeSession(ctx, sessionId) + } + return Promise.resolve() + } + + closeSession (ctx: MeasureContext, sessionId: string): Promise { + return this.provideCloseSession(ctx, sessionId) + } } diff --git a/server/core/src/pipeline.ts b/server/core/src/pipeline.ts index b9664b0e50..f71cbc101c 100644 --- a/server/core/src/pipeline.ts +++ b/server/core/src/pipeline.ts @@ -21,10 +21,13 @@ import { type Doc, type DocumentQuery, type Domain, + type DomainParams, + type DomainResult, type FindOptions, type FindResult, type LoadModelResponse, type MeasureContext, + type OperationDomain, type Ref, type SearchOptions, type SearchQuery, @@ -100,11 +103,11 @@ class PipelineImpl implements Pipeline { query: DocumentQuery, options?: FindOptions ): Promise> { - return this.head !== undefined ? this.head.findAll(ctx, _class, query, options) : Promise.resolve(toFindResult([])) + return this.head?.findAll(ctx, _class, query, options) ?? Promise.resolve(toFindResult([])) } loadModel (ctx: MeasureContext, lastModelTx: Timestamp, hash?: string): Promise { - return this.head !== undefined ? this.head.loadModel(ctx, lastModelTx, hash) : Promise.resolve([]) + return this.head?.loadModel(ctx, lastModelTx, hash) ?? Promise.resolve([]) } groupBy( @@ -113,24 +116,33 @@ class PipelineImpl implements Pipeline { field: string, query?: DocumentQuery

): Promise> { - return this.head !== undefined ? this.head.groupBy(ctx, domain, field, query) : Promise.resolve(new Map()) + return this.head?.groupBy(ctx, domain, field, query) ?? Promise.resolve(new Map()) } searchFulltext (ctx: MeasureContext, query: SearchQuery, options: SearchOptions): Promise { - return this.head !== undefined ? this.head.searchFulltext(ctx, query, options) : Promise.resolve({ docs: [] }) + return this.head?.searchFulltext(ctx, query, options) ?? Promise.resolve({ docs: [] }) } tx (ctx: MeasureContext, tx: Tx[]): Promise { - if (this.head !== undefined) { - return this.head.tx(ctx, tx) - } - return Promise.resolve({}) + return this.head?.tx(ctx, tx) ?? Promise.resolve({}) } handleBroadcast (ctx: MeasureContext): Promise { return this.head?.handleBroadcast(ctx) ?? emptyBroadcastResult } + domainRequest( + ctx: MeasureContext, + domain: OperationDomain, + params: DomainParams + ): Promise> { + return this.head?.domainRequest(ctx, domain, params) ?? Promise.resolve({ domain, value: undefined }) + } + + closeSession (ctx: MeasureContext, sessionId: string): Promise { + return this.head?.closeSession(ctx, sessionId) ?? Promise.resolve() + } + async close (): Promise { for (const mw of this.middlewares) { try { diff --git a/server/core/src/types.ts b/server/core/src/types.ts index 9e43061e71..96a110adcf 100644 --- a/server/core/src/types.ts +++ b/server/core/src/types.ts @@ -13,17 +13,7 @@ // limitations under the License. // -import { - type ServerApi as CommunicationApi, - type Event as CommunicationEvent, - type EventResult -} from '@hcengineering/communication-sdk-types' -import { - type FindMessagesGroupsParams, - type FindMessagesParams, - type Message, - type MessagesGroup -} from '@hcengineering/communication-types' +import { type ServerApi as CommunicationApi } from '@hcengineering/communication-sdk-types' import { type Account, type AccountUuid, @@ -32,6 +22,8 @@ import { type Doc, type DocumentQuery, type Domain, + type DomainParams, + type DomainResult, type FindOptions, type FindResult, type Hierarchy, @@ -40,6 +32,7 @@ import { type MeasureContext, type ModelDb, type Obj, + type OperationDomain, type PersonId, type Ref, type SearchOptions, @@ -63,9 +56,9 @@ import type { Token } from '@hcengineering/server-token' import { type Readable } from 'stream' import type { DbAdapter, DomainHelper } from './adapter' +import { type PlatformQueue, type PlatformQueueProducer, type QueueTopic } from './queue' import type { StatisticsElement, WorkspaceStatistics } from './stats' import { type StorageAdapter } from './storage' -import { type PlatformQueueProducer, type QueueTopic, type PlatformQueue } from './queue' export interface ServerFindOptions extends FindOptions { prefix?: string @@ -119,20 +112,29 @@ export interface Middleware { hash?: string ) => Promise + // Operation to some specific domain + domainRequest: ( + ctx: MeasureContext, + domain: OperationDomain, + params: DomainParams + ) => Promise + closeSession: (ctx: MeasureContext, sessionId: string) => Promise close: () => Promise } /** * @public */ -export type BroadcastFunc = ( - ctx: MeasureContext, - tx: Tx[], - targets?: string | string[], - exclude?: string[] -) => void +export interface BroadcastOps { + broadcast: (ctx: MeasureContext, tx: Tx[], targets?: AccountUuid | AccountUuid[], exclude?: AccountUuid[]) => void -export type BroadcastSessionsFunc = (ctx: MeasureContext, sessionIds: string[], result: any) => void + broadcastSessions: (measure: MeasureContext, sessionIds: Record) => void +} + +export interface BroadcastSessionsFunc { + broadcast: (ctx: MeasureContext, sessionIds: string[], result: any) => void + enqueue: (ctx: MeasureContext, result: any) => void +} /** * @public @@ -199,7 +201,6 @@ export interface PipelineContext { contextVars: Record broadcastEvent?: (ctx: MeasureContext, tx: Tx[]) => Promise - communicationApi: CommunicationApi | null userStatusMap?: Map, { online: boolean, user: AccountUuid }> } /** @@ -228,6 +229,14 @@ export interface Pipeline { ) => Promise handleBroadcast: (ctx: MeasureContext) => Promise + + domainRequest: ( + ctx: MeasureContext, + domain: OperationDomain, + params: DomainParams + ) => Promise> + + closeSession: (ctx: MeasureContext, sessionId: string) => Promise } /** @@ -236,9 +245,8 @@ export interface Pipeline { export type PipelineFactory = ( ctx: MeasureContext, ws: WorkspaceIds, - broadcast: BroadcastFunc, - branding: Branding | null, - communicationApi: CommunicationApi | null + broadcast: BroadcastOps, + branding: Branding | null ) => Promise export type CommunicationApiFactory = ( @@ -266,11 +274,10 @@ export interface TriggerControl { modelDb: ModelDb removedMap: Map, Doc> userStatusMap: Map, { online: boolean, user: AccountUuid }> + domainRequest: (ctx: MeasureContext, domain: OperationDomain, params: DomainParams) => Promise queue?: PlatformQueue - communicationApi: CommunicationApi | null - // Cache per workspace cache: Map // Cache per root tx @@ -543,8 +550,6 @@ export interface ClientSessionCtx { ctx: MeasureContext pipeline: Pipeline - communicationApi: CommunicationApi - socialStringsToUsers: Map requestId: ReqId | undefined sendResponse: (id: ReqId | undefined, msg: any) => Promise @@ -629,11 +634,9 @@ export interface Session { includeSessionContext: (ctx: ClientSessionCtx) => void - eventRaw: (ctx: ClientSessionCtx, event: CommunicationEvent) => Promise - findMessagesRaw: (ctx: ClientSessionCtx, params: FindMessagesParams) => Promise - findMessagesGroupsRaw: (ctx: ClientSessionCtx, params: FindMessagesGroupsParams) => Promise - updateLast: () => void + + domainRequestRaw: (ctx: ClientSessionCtx, domain: OperationDomain, params: DomainParams) => Promise } /** @@ -694,7 +697,12 @@ export interface SessionManager { sessionId: string | undefined ) => Promise - broadcastAll: (workspace: WorkspaceUuid, tx: Tx[], targets?: string[]) => void + broadcastAll: ( + workspace: WorkspaceUuid, + tx: Tx[], + targets?: AccountUuid | AccountUuid[], + exclude?: AccountUuid[] + ) => void close: (ctx: MeasureContext, ws: ConnectionSocket, workspaceId: WorkspaceUuid) => Promise @@ -728,7 +736,6 @@ export interface SessionManager { ctx: MeasureContext, sendCtx: MeasureContext, pipeline: Pipeline, - communicationApi: CommunicationApi, requestId: Request['id'], service: Session, ws: ConnectionSocket, diff --git a/server/core/src/utils.ts b/server/core/src/utils.ts index 84a213dd48..9abfe9466f 100644 --- a/server/core/src/utils.ts +++ b/server/core/src/utils.ts @@ -16,10 +16,13 @@ import core, { type DocChunk, type DocumentQuery, type Domain, + type DomainParams, + type DomainResult, type FindOptions, type FindResult, type MeasureContext, type ModelDb, + type OperationDomain, type PersonId, type Ref, type SearchResult, @@ -181,7 +184,9 @@ export class SessionDataImpl implements SessionData { if (this._broadcast === undefined) { this._broadcast = { targets: {}, - txes: [] + txes: [], + queue: [], + sessions: {} } } return this._broadcast @@ -241,7 +246,7 @@ export function wrapPipeline ( systemAccount, 'pipeline', true, - { targets: {}, txes: [] }, + { targets: {}, txes: [], queue: [], sessions: {} }, wsIds, true, undefined, @@ -260,6 +265,9 @@ export function wrapPipeline ( findAll: (_class, query, options) => pipeline.findAll(ctx, _class, query, options), findOne: async (_class, query, options) => (await pipeline.findAll(ctx, _class, query, { ...options, limit: 1 })).shift(), + domainRequest: async (domain, params) => { + return await pipeline.domainRequest(ctx, domain, params) + }, clean: (domain, docs) => backupOps.clean(ctx, domain, docs), close: () => pipeline.close(), closeChunk: (idx) => backupOps.closeChunk(ctx, idx), @@ -309,6 +317,10 @@ export function wrapAdapterToClient (ctx: MeasureContext, storageAdapter: DbAdap return (await storageAdapter.findAll(ctx, _class, query, options)) as any } + async domainRequest(domain: OperationDomain, params: DomainParams): Promise> { + return { domain, value: null as any } + } + async tx (tx: Tx): Promise { return await storageAdapter.tx(ctx, tx) } diff --git a/server/middleware/src/broadcast.ts b/server/middleware/src/broadcast.ts index 9e6f5f1c32..75b55c6920 100644 --- a/server/middleware/src/broadcast.ts +++ b/server/middleware/src/broadcast.ts @@ -15,6 +15,7 @@ import { TxProcessor, + type AccountUuid, type BroadcastTargets, type Class, type Doc, @@ -25,7 +26,7 @@ import { type TxCUD } from '@hcengineering/core' import type { - BroadcastFunc, + BroadcastOps, Middleware, MiddlewareCreator, PipelineContext, @@ -40,13 +41,13 @@ export class BroadcastMiddleware extends BaseMiddleware implements Middleware { constructor ( context: PipelineContext, protected readonly next: Middleware | undefined, - readonly broadcast: BroadcastFunc + readonly broadcast: BroadcastOps ) { super(context, next) context.broadcastEvent = (ctx, tx) => this.doBroadcast(ctx, tx) } - static create (broadcast: BroadcastFunc): MiddlewareCreator { + static create (broadcast: BroadcastOps): MiddlewareCreator { return async (ctx, pipelineContext, next) => new BroadcastMiddleware(pipelineContext, next, broadcast) } @@ -54,6 +55,10 @@ export class BroadcastMiddleware extends BaseMiddleware implements Middleware { await this.next?.handleBroadcast(ctx) await this.doBroadcast(ctx, ctx.contextData.broadcast.txes, ctx.contextData.broadcast.targets) + + if (Object.keys(ctx.contextData.broadcast.sessions).length > 0) { + this.broadcast.broadcastSessions(ctx, ctx.contextData.broadcast.sessions) + } } tx (ctx: MeasureContext, tx: Tx[]): Promise { @@ -70,9 +75,9 @@ export class BroadcastMiddleware extends BaseMiddleware implements Middleware { // Combine targets by sender - const toSendTarget = new Map() + const toSendTarget = new Map() - const getTxes = (key: string): Tx[] => { + const getTxes = (key: AccountUuid | ''): Tx[] => { let txes = toSendTarget.get(key) if (txes === undefined) { txes = [...(toSendTarget.get('') ?? [])] // We also need to add all from to all @@ -83,7 +88,7 @@ export class BroadcastMiddleware extends BaseMiddleware implements Middleware { // Put current user as send target for (const txd of tx) { - let target: string[] | undefined + let target: AccountUuid[] | undefined for (const tt of Object.values(targets ?? {})) { target = tt(txd) if (target !== undefined) { @@ -107,8 +112,8 @@ export class BroadcastMiddleware extends BaseMiddleware implements Middleware { const handleSend = async ( ctx: MeasureContext, derived: Tx[], - target?: string, - exclude?: string[] + target?: AccountUuid, + exclude?: AccountUuid[] ): Promise => { if (derived.length === 0) { return @@ -118,7 +123,7 @@ export class BroadcastMiddleware extends BaseMiddleware implements Middleware { await this.sendWithPart(derived, ctx, target, exclude) } else { // Let's send after our response will go out - this.broadcast(ctx, derived, target, exclude) + this.broadcast.broadcast(ctx, derived, target, exclude) } } @@ -127,17 +132,17 @@ export class BroadcastMiddleware extends BaseMiddleware implements Middleware { // Then send targeted and all other for (const [k, v] of toSendTarget.entries()) { - void handleSend(ctx, v, k) + void handleSend(ctx, v, k as AccountUuid) } // Send all other except us. - await handleSend(ctx, toSendAll, undefined, Array.from(toSendTarget.keys())) + await handleSend(ctx, toSendAll, undefined, Array.from(toSendTarget.keys()) as AccountUuid[]) } private async sendWithPart ( derived: Tx[], ctx: MeasureContext, - target: string | undefined, - exclude: string[] | undefined + target: AccountUuid | undefined, + exclude: AccountUuid[] | undefined ): Promise { const classes = new Set>>() for (const dtx of derived) { @@ -150,6 +155,6 @@ export class BroadcastMiddleware extends BaseMiddleware implements Middleware { } } const bevent = createBroadcastEvent(Array.from(classes)) - this.broadcast(ctx, [bevent], target, exclude) + this.broadcast.broadcast(ctx, [bevent], target, exclude) } } diff --git a/server/middleware/src/liveQuery.ts b/server/middleware/src/liveQuery.ts index 8d570a3944..f76d8dfda2 100644 --- a/server/middleware/src/liveQuery.ts +++ b/server/middleware/src/liveQuery.ts @@ -73,6 +73,9 @@ export class LiveQueryMiddleware extends BaseMiddleware implements Middleware { results.total )[0] }, + domainRequest: async (domain, params) => { + return await this.provideDomainRequest(metrics, domain, params) + }, tx: (tx) => { return Promise.resolve({}) }, diff --git a/server/middleware/src/private.ts b/server/middleware/src/private.ts index f8e4b82e25..685020bce7 100644 --- a/server/middleware/src/private.ts +++ b/server/middleware/src/private.ts @@ -29,7 +29,8 @@ import core, { type TxCUD, TxProcessor, systemAccountUuid, - type SessionData + type SessionData, + type AccountUuid } from '@hcengineering/core' import platform, { PlatformError, Severity, Status } from '@hcengineering/platform' import { @@ -69,7 +70,7 @@ export class PrivateMiddleware extends BaseMiddleware implements Middleware { tx (ctx: MeasureContext, txes: Tx[]): Promise { for (const tx of txes) { - let target: PersonUuid[] | undefined + let target: AccountUuid[] | undefined if (this.isTargetDomain(tx)) { const account = ctx.contextData.account if (!account.socialIds.includes(tx.modifiedBy) && account.uuid !== systemAccountUuid) { diff --git a/server/middleware/src/queue.ts b/server/middleware/src/queue.ts index 588bb9a943..5386ac2809 100644 --- a/server/middleware/src/queue.ts +++ b/server/middleware/src/queue.ts @@ -50,7 +50,10 @@ export class QueueMiddleware extends BaseMiddleware { } await Promise.all([ this.provideBroadcast(ctx), - this.txProducer.send(this.context.workspace.uuid, ctx.contextData.broadcast.txes) + this.txProducer.send( + this.context.workspace.uuid, + ctx.contextData.broadcast.txes.concat(ctx.contextData.broadcast.queue) + ) ]) } } diff --git a/server/middleware/src/spaceSecurity.ts b/server/middleware/src/spaceSecurity.ts index 2351ea6c01..f5598ffef2 100644 --- a/server/middleware/src/spaceSecurity.ts +++ b/server/middleware/src/spaceSecurity.ts @@ -343,7 +343,7 @@ export class SpaceSecurityMiddleware extends BaseMiddleware implements Middlewar } } - getTargets (accounts: AccountUuid[]): string[] { + getTargets (accounts: AccountUuid[]): AccountUuid[] { const res = Array.from(new Set(accounts)) // We need to add system account for targets for integrations to work properly res.push(systemAccountUuid) diff --git a/server/middleware/src/triggers.ts b/server/middleware/src/triggers.ts index fb2cfcf97e..3c7de2c1a5 100644 --- a/server/middleware/src/triggers.ts +++ b/server/middleware/src/triggers.ts @@ -141,8 +141,10 @@ export class TriggersMiddleware extends BaseMiddleware implements Middleware { modelDb: this.context.modelDb, hierarchy: this.context.hierarchy, cache: this.cache, - communicationApi: this.context.communicationApi, userStatusMap: this.context.userStatusMap ?? new Map(), + domainRequest: async (ctx, domain, params) => { + return (await this.context.head?.domainRequest(ctx, domain, params)) ?? { domain, value: undefined } + }, apply: async (ctx, tx, needResult) => { if (needResult === true) { return (await this.context.derived?.tx(ctx, tx)) ?? {} @@ -223,7 +225,7 @@ export class TriggersMiddleware extends BaseMiddleware implements Middleware { sctx.account, sctx.sessionId, sctx.admin, - { txes: [], targets: {} }, + { txes: [], targets: {}, queue: [], sessions: {} }, this.context.workspace, true, sctx.removedMap, diff --git a/server/server-pipeline/package.json b/server/server-pipeline/package.json index 03309efa19..7f6ed41c06 100644 --- a/server/server-pipeline/package.json +++ b/server/server-pipeline/package.json @@ -178,6 +178,8 @@ "@hcengineering/chat": "^0.6.0", "@hcengineering/card": "^0.6.0", "@hcengineering/mail": "^0.6.0", - "@hcengineering/kafka": "^0.6.0" + "@hcengineering/kafka": "^0.6.0", + "@hcengineering/communication-types": "^0.1.0", + "@hcengineering/communication-sdk-types": "^0.1.0" } } diff --git a/server/server-pipeline/src/communication.ts b/server/server-pipeline/src/communication.ts new file mode 100644 index 0000000000..c09a88bfa8 --- /dev/null +++ b/server/server-pipeline/src/communication.ts @@ -0,0 +1,140 @@ +// +// Copyright © 2022 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 type { SessionData as CommunicationSession, Event, ServerApi } from '@hcengineering/communication-sdk-types' +import core, { + generateId, + type DomainParams, + type DomainResult, + type MeasureContext, + type OperationDomain, + type SessionData, + type TxDomainEvent +} from '@hcengineering/core' +import type { + CommunicationApiFactory, + Middleware, + MiddlewareCreator, + PipelineContext +} from '@hcengineering/server-core' +import { BaseMiddleware } from '@hcengineering/server-core' + +export const COMMUNICATION_DOMAIN = 'communication' as OperationDomain + +/** + * @public + */ +export class CommunicationMiddleware extends BaseMiddleware implements Middleware { + constructor ( + readonly ctx: MeasureContext, + context: PipelineContext, + readonly next: Middleware | undefined, + readonly communicationApi: ServerApi + ) { + super(context, next) + } + + static create (communicationApiFactory: CommunicationApiFactory): MiddlewareCreator { + return async (ctx, context, next): Promise => { + const communicationApi = await communicationApiFactory(ctx, context.workspace, { + broadcast: (ctx, sessions, result: Event) => { + const { contextData, evt } = CommunicationMiddleware.wrapEvent(ctx, result) + for (const s of sessions) { + contextData.broadcast.sessions[s] = (contextData.broadcast.sessions[s] ?? []).concat(evt) + } + }, + enqueue: (ctx, result: Event) => { + const { contextData, evt } = CommunicationMiddleware.wrapEvent(ctx, result) + contextData.broadcast.queue.push(evt) + } + }) + return new CommunicationMiddleware(ctx, context, next, communicationApi) + } + } + + private static wrapEvent (ctx: MeasureContext, result: Event): { contextData: SessionData, evt: TxDomainEvent } { + const contextData = ctx.contextData as SessionData + const evt: TxDomainEvent = { + _id: generateId(), + space: core.space.Tx, + objectSpace: core.space.Domain, + _class: core.class.TxDomainEvent, + domain: COMMUNICATION_DOMAIN, + event: result, + modifiedBy: contextData.account.primarySocialId, + modifiedOn: Date.now() + } + return { contextData, evt } + } + + async domainRequest (ctx: MeasureContext, domain: OperationDomain, params: DomainParams): Promise { + if (domain === COMMUNICATION_DOMAIN) { + return { + domain, + value: await this.handleCommand(ctx, params) + } + } else { + return await this.provideDomainRequest(ctx, domain, params) + } + } + + async handleCommand (_ctx: MeasureContext, args: DomainParams): Promise { + const ctx = this.getCommunicationCtx(_ctx) + if (args.findMessages !== undefined) { + const { params, queryId } = args.findMessages + return await this.communicationApi.findMessages(ctx, params, queryId) + } + if (args.findMessagesGroups !== undefined) { + const { params } = args.findMessagesGroups + return await this.communicationApi.findMessagesGroups(ctx, params) + } + if (args.findNotificationContexts !== undefined) { + const { params, queryId } = args.findNotificationContexts + return await this.communicationApi.findNotificationContexts(ctx, params, queryId) + } + if (args.findNotifications !== undefined) { + const { params, queryId } = args.findNotifications + return await this.communicationApi.findNotifications(ctx, params, queryId) + } + if (args.findLabels !== undefined) { + const { params } = args.findLabels + return await this.communicationApi.findLabels(ctx, params) + } + if (args.findCollaborators !== undefined) { + const { params } = args.findCollaborators + return await this.communicationApi.findCollaborators(ctx, params) + } + if (args.unsubscribeQuery !== undefined) { + const { id } = args.unsubscribeQuery + await this.communicationApi.unsubscribeQuery(ctx, id) + return + } + if (args.event !== undefined) { + const event = args.event + return await this.communicationApi.event(ctx, event) + } + return {} + } + + private getCommunicationCtx (ctx: MeasureContext): CommunicationSession { + return { + ...ctx, + sessionId: ctx.contextData.sessionId, + derived: ctx.contextData.isTriggerCtx === true, + // TODO: We should decide what to do with communications package and remove this workaround + account: ctx.contextData.account + } + } +} diff --git a/server/server-pipeline/src/pipeline.ts b/server/server-pipeline/src/pipeline.ts index 9ce5493946..0865abb2cf 100644 --- a/server/server-pipeline/src/pipeline.ts +++ b/server/server-pipeline/src/pipeline.ts @@ -46,6 +46,8 @@ import { createInMemoryAdapter, createNullAdapter, createPipeline, + type BroadcastOps, + type CommunicationApiFactory, type DbAdapterFactory, type DbConfiguration, type Middleware, @@ -59,6 +61,7 @@ import { } from '@hcengineering/server-core' import { generateToken } from '@hcengineering/server-token' import { createStorageDataAdapter } from './blobStorage' +import { CommunicationMiddleware } from './communication' /** * @public @@ -109,10 +112,11 @@ export function createServerPipeline ( extraLogging?: boolean // If passed, will log every request/etc. pipelineContextVars?: Record + communicationApiFactory?: CommunicationApiFactory }, extensions?: Partial ): PipelineFactory { - return (ctx, workspace, broadcast, branding, communicationApi) => { + return (ctx, workspace, broadcast, branding) => { const metricsCtx = opt.usePassedCtx === true ? ctx : metrics const wsMetrics = metricsCtx.newChild('🧲 session', {}) const conf = getConfig(metrics, dbUrl, wsMetrics, opt, extensions) @@ -124,6 +128,9 @@ export function createServerPipeline ( FindSecurityMiddleware.create, PluginConfigurationMiddleware.create, PrivateMiddleware.create, + ...(opt.communicationApiFactory !== undefined + ? [CommunicationMiddleware.create(opt.communicationApiFactory)] + : []), (ctx: MeasureContext, context: PipelineContext, next?: Middleware) => SpaceSecurityMiddleware.create(opt.adapterSecurity ?? false, ctx, context, next), SpacePermissionsMiddleware.create, @@ -163,8 +170,7 @@ export function createServerPipeline ( hierarchy, queue: opt.queue, storageAdapter: opt.externalStorage, - contextVars: opt.pipelineContextVars ?? sharedPipelineContextVars, - communicationApi + contextVars: opt.pipelineContextVars ?? sharedPipelineContextVars } return createPipeline(ctx, middlewares, context) } @@ -185,7 +191,7 @@ export function createBackupPipeline ( externalStorage: StorageAdapter } ): PipelineFactory { - return (ctx, workspace, broadcast, branding, communicationApi) => { + return (ctx, workspace, broadcast, branding) => { const metricsCtx = opt.usePassedCtx === true ? ctx : metrics const wsMetrics = metricsCtx.newChild('🧲 backup', {}) const conf = getConfig(metrics, dbUrl, wsMetrics, { @@ -211,13 +217,19 @@ export function createBackupPipeline ( modelDb, hierarchy, storageAdapter: opt.externalStorage, - contextVars: {}, - communicationApi + contextVars: {} } return createPipeline(ctx, middlewares, context) } } +export function createEmptyBroadcastOps (): BroadcastOps { + return { + broadcast: (): void => {}, + broadcastSessions: (): void => {} + } +} + export async function getServerPipeline ( ctx: MeasureContext, model: Tx[], @@ -227,6 +239,7 @@ export async function getServerPipeline ( opt?: { queue?: PlatformQueue disableTriggers?: boolean + communicationApiFactory?: CommunicationApiFactory } ): Promise { const pipelineFactory = createServerPipeline(ctx, dbUrl, model, { @@ -234,11 +247,11 @@ export async function getServerPipeline ( usePassedCtx: true, disableTriggers: opt?.disableTriggers ?? false, adapterSecurity: isAdapterSecurity(dbUrl), - queue: opt?.queue + queue: opt?.queue, + communicationApiFactory: opt?.communicationApiFactory }) - // TODO: Communication API ?? - return await pipelineFactory(ctx, wsUrl, () => {}, null, null) + return await pipelineFactory(ctx, wsUrl, createEmptyBroadcastOps(), null) } const txAdapterFactories: Record = {} diff --git a/server/server/package.json b/server/server/package.json index c5c45e1d08..201f6aabf3 100644 --- a/server/server/package.json +++ b/server/server/package.json @@ -37,8 +37,6 @@ "dependencies": { "@hcengineering/account-client": "^0.6.0", "@hcengineering/analytics": "^0.6.0", - "@hcengineering/communication-types": "^0.1.0", - "@hcengineering/communication-sdk-types": "^0.1.0", "@hcengineering/core": "^0.6.32", "@hcengineering/platform": "^0.6.11", "@hcengineering/rpc": "^0.6.5", diff --git a/server/server/src/client.ts b/server/server/src/client.ts index 9ddb0f36de..1c92fd2f42 100644 --- a/server/server/src/client.ts +++ b/server/server/src/client.ts @@ -15,33 +15,21 @@ import type { LoginInfoWithWorkspaces } from '@hcengineering/account-client' import { - type Event as CommunicationEvent, - type SessionData as CommunicationSession, - type EventResult -} from '@hcengineering/communication-sdk-types' -import { - type FindCollaboratorsParams, - type FindLabelsParams, - type FindMessagesGroupsParams, - type FindMessagesParams, - type FindNotificationContextParams, - type FindNotificationsParams, - type Message, - type MessagesGroup -} from '@hcengineering/communication-types' -import { - type AccountUuid, generateId, TxProcessor, type Account, + type AccountUuid, type Class, type Doc, type DocumentQuery, type Domain, + type DomainParams, + type DomainResult, type FindOptions, type FindResult, type LoadModelResponse, type MeasureContext, + type OperationDomain, type PersonId, type Ref, type SearchOptions, @@ -392,70 +380,49 @@ export class ClientSession implements Session { await ctx.sendResponse(ctx.requestId, {}) } - async eventRaw (ctx: ClientSessionCtx, event: CommunicationEvent): Promise { - this.lastRequest = Date.now() - return await ctx.communicationApi.event(this.getCommunicationCtx(), event) - } + async domainRequest (ctx: ClientSessionCtx, domain: OperationDomain, params: DomainParams): Promise { + try { + const result: DomainResult = await this.domainRequestRaw(ctx, domain, params) + await ctx.sendResponse(ctx.requestId, result) - async event (ctx: ClientSessionCtx, event: CommunicationEvent): Promise { - const result = await this.eventRaw(ctx, event) - await ctx.sendResponse(ctx.requestId, result) - } + // We need to broadcast all collected transactions + const broadcastPromise = ctx.pipeline.handleBroadcast(ctx.ctx) - async findMessagesRaw (ctx: ClientSessionCtx, params: FindMessagesParams, queryId?: number): Promise { - this.lastRequest = Date.now() - return await ctx.communicationApi.findMessages(this.getCommunicationCtx(), params, queryId) - } - - async findMessages (ctx: ClientSessionCtx, params: FindMessagesParams, queryId?: number): Promise { - const result = await this.findMessagesRaw(ctx, params, queryId) - await ctx.sendResponse(ctx.requestId, result) - } - - async findMessagesGroupsRaw (ctx: ClientSessionCtx, params: FindMessagesGroupsParams): Promise { - this.lastRequest = Date.now() - return await ctx.communicationApi.findMessagesGroups(this.getCommunicationCtx(), params) - } - - async findMessagesGroups (ctx: ClientSessionCtx, params: FindMessagesGroupsParams): Promise { - const result = await this.findMessagesGroupsRaw(ctx, params) - await ctx.sendResponse(ctx.requestId, result) - } - - async findNotifications (ctx: ClientSessionCtx, params: FindNotificationsParams): Promise { - const result = await ctx.communicationApi.findNotifications(this.getCommunicationCtx(), params) - await ctx.sendResponse(ctx.requestId, result) - } - - async findNotificationContexts ( - ctx: ClientSessionCtx, - params: FindNotificationContextParams, - queryId?: number - ): Promise { - const result = await ctx.communicationApi.findNotificationContexts(this.getCommunicationCtx(), params, queryId) - await ctx.sendResponse(ctx.requestId, result) - } - - async findLabels (ctx: ClientSessionCtx, params: FindLabelsParams): Promise { - const result = await ctx.communicationApi.findLabels(this.getCommunicationCtx(), params) - await ctx.sendResponse(ctx.requestId, result) - } - - async findCollaborators (ctx: ClientSessionCtx, params: FindCollaboratorsParams): Promise { - const result = await ctx.communicationApi.findCollaborators(this.getCommunicationCtx(), params) - await ctx.sendResponse(ctx.requestId, result) - } - - async unsubscribeQuery (ctx: ClientSessionCtx, id: number): Promise { - this.lastRequest = Date.now() - await ctx.communicationApi.unsubscribeQuery(this.getCommunicationCtx(), id) - await ctx.sendResponse(ctx.requestId, {}) - } - - private getCommunicationCtx (): CommunicationSession { - return { - sessionId: this.sessionId, - account: this.account + // ok we could perform async requests if any + const asyncs = (ctx.ctx.contextData as SessionData).asyncRequests ?? [] + let asyncsPromise: Promise | undefined + if (asyncs.length > 0) { + const handleAyncs = async (): Promise => { + try { + for (const r of asyncs) { + await r(ctx.ctx) + } + } catch (err: any) { + ctx.ctx.error('failed to handleAsyncs', { err }) + } + } + asyncsPromise = handleAyncs() + } + await broadcastPromise + if (asyncsPromise !== undefined) { + await asyncsPromise + } + } catch (err) { + await ctx.sendError(ctx.requestId, 'Failed to findAll', unknownError(err)) + ctx.ctx.error('failed to findAll', { err }) } } + + async domainRequestRaw (ctx: ClientSessionCtx, domain: OperationDomain, params: DomainParams): Promise { + this.lastRequest = Date.now() + this.total.find++ + this.current.find++ + this.includeSessionContext(ctx) + const result: DomainResult = await ctx.ctx.with('domainRequest', { domain }, (_ctx) => + _ctx.with(`${domain}-${Object.keys(params)[0]}`, {}, (_ctx) => ctx.pipeline.domainRequest(_ctx, domain, params), { + params + }) + ) + return result + } } diff --git a/server/server/src/sessionManager.ts b/server/server/src/sessionManager.ts index 6ecb460e31..8eccae8f51 100644 --- a/server/server/src/sessionManager.ts +++ b/server/server/src/sessionManager.ts @@ -19,7 +19,6 @@ import { type LoginInfoWorkspace } from '@hcengineering/account-client' import { Analytics } from '@hcengineering/analytics' -import { type ServerApi as CommunicationApi } from '@hcengineering/communication-sdk-types' import core, { AccountRole, type AccountUuid, @@ -63,7 +62,6 @@ import { import { type AddSessionResponse, type ClientSessionCtx, - type CommunicationApiFactory, type ConnectionSocket, type GetWorkspaceResponse, LOGGING_ENABLED, @@ -85,7 +83,7 @@ import { import { generateToken, type Token } from '@hcengineering/server-token' import { ClientSession } from './client' import { sendResponse } from './utils' -import { type PipelinePair, Workspace } from './workspace' +import { Workspace } from './workspace' const ticksPerSecond = 20 const workspaceSoftShutdownTicks = 15 * ticksPerSecond @@ -135,8 +133,7 @@ export class TSessionManager implements SessionManager { readonly enableCompression: boolean, readonly doHandleTick: boolean = true, readonly queue: PlatformQueue, - readonly pipelineFactory: PipelineFactory, - readonly communicationApiFactory: CommunicationApiFactory + readonly pipelineFactory: PipelineFactory ) { if (this.doHandleTick) { this.checkInterval = setInterval(() => { @@ -630,7 +627,12 @@ export class TSessionManager implements SessionManager { await this.doCloseAll(workspace, 0, 'upgrade', ws) } - broadcastAll (workspace: WorkspaceUuid, tx: Tx[], target?: string | string[], exclude?: string[]): void { + broadcastAll ( + workspace: WorkspaceUuid, + tx: Tx[], + target?: AccountUuid | AccountUuid[], + exclude?: AccountUuid[] + ): void { const ws = this.workspaces.get(workspace) if (ws === undefined) { return @@ -638,7 +640,7 @@ export class TSessionManager implements SessionManager { this.doBroadcast(ws, tx, target, exclude) } - doBroadcast (ws: Workspace, tx: Tx[], target?: string | string[], exclude?: string[]): void { + doBroadcast (ws: Workspace, tx: Tx[], target?: AccountUuid | AccountUuid[], exclude?: AccountUuid[]): void { if (ws.upgrade) { return } @@ -674,22 +676,21 @@ export class TSessionManager implements SessionManager { } } - broadcastSessions (measure: MeasureContext, workspace: Workspace, sessionIds: string[], result: any): void { - if (workspace.upgrade) { - return - } + broadcastSessions (measure: MeasureContext, sessionIds: Record): void { const ctx = measure.newChild('📬 broadcast sessions', {}) - const sessions = [...workspace.sessions.values()].filter((it) => { - if (it === undefined || it.session.sessionId === '') { - return false - } - return sessionIds.includes(it.session.sessionId) - }) + const allSessions = Array.from(this.sessions.values()) + const sessions = Object.entries(sessionIds).map(([sessionId, txes]) => ({ + session: allSessions.find((it) => it.session.sessionId === sessionId), + txes + })) function send (): void { for (const session of sessions) { + if (session.session === undefined) { + continue + } try { - void sendResponse(ctx, session.session, session.socket, { result }) + void sendResponse(ctx, session.session.session, session.session.socket, { result: session.txes }) } catch (err: any) { Analytics.handleError(err) ctx.error('error during send', { error: err }) @@ -709,8 +710,8 @@ export class TSessionManager implements SessionManager { from: Session | null, workspaceId: WorkspaceUuid, resp: Tx[], - target: string | undefined, - exclude?: string[] + target: AccountUuid | undefined, + exclude?: AccountUuid[] ): void { const workspace = this.workspaces.get(workspaceId) if (workspace === undefined) { @@ -760,24 +761,21 @@ export class TSessionManager implements SessionManager { url: workspaceUrl } - const factory = async (): Promise => { - const communicationApi = await this.communicationApiFactory( - pipelineCtx, - workspaceIds, - (ctx, sessionIds, result) => { - this.broadcastSessions(ctx, workspace, sessionIds, result) - } - ) + const factory = async (): Promise => { const pipeline = await this.pipelineFactory( pipelineCtx, workspaceIds, - (ctx, tx, targets, exclude) => { - this.broadcastAll(workspaceIds.uuid, tx, targets, exclude) + { + broadcast: (ctx, tx, targets, exclude) => { + this.broadcastAll(workspaceIds.uuid, tx, targets, exclude) + }, + broadcastSessions: (ctx, sessions) => { + this.broadcastSessions(ctx, sessions) + } }, - branding, - communicationApi + branding ) - return { pipeline, communicationApi } + return pipeline } const workspace: Workspace = new Workspace( context, @@ -797,7 +795,6 @@ export class TSessionManager implements SessionManager { private async trySetStatus ( ctx: MeasureContext, pipeline: Pipeline, - communicationApi: CommunicationApi, session: Session, online: boolean, workspaceId: WorkspaceUuid @@ -806,7 +803,7 @@ export class TSessionManager implements SessionManager { if (current !== undefined) { await current } - const promise = this.setStatus(ctx, pipeline, communicationApi, session, online, workspaceId) + const promise = this.setStatus(ctx, pipeline, session, online, workspaceId) this.statusPromises.set(session.getUser(), promise) await promise this.statusPromises.delete(session.getUser()) @@ -815,7 +812,6 @@ export class TSessionManager implements SessionManager { private async setStatus ( ctx: MeasureContext, pipeline: Pipeline, - communicationApi: CommunicationApi, session: Session, online: boolean, workspaceId: WorkspaceUuid @@ -828,7 +824,6 @@ export class TSessionManager implements SessionManager { const clientCtx: ClientSessionCtx = { requestId: undefined, pipeline, - communicationApi, sendResponse: async () => { // No response }, @@ -898,13 +893,13 @@ export class TSessionManager implements SessionManager { if (workspace !== undefined) { const another = Array.from(workspace.sessions.values()).findIndex((p) => p.session.getUser() === user) if (another === -1 && !workspace.upgrade) { - void workspace.with(async (pipeline, communicationApi) => { - await communicationApi.closeSession(sessionRef.session.sessionId) + void workspace.with(async (pipeline) => { + await pipeline.closeSession(ctx, sessionRef.session.sessionId) + // await communicationApi.closeSession(sessionRef.session.sessionId) if (user !== guestAccount && user !== systemAccountUuid) { await this.trySetStatus( workspace.context, pipeline, - communicationApi, sessionRef.session, false, workspaceUuid @@ -1051,7 +1046,6 @@ export class TSessionManager implements SessionManager { ctx: MeasureContext, sendCtx: MeasureContext, pipeline: Pipeline, - communicationApi: CommunicationApi, requestId: Request['id'], service: Session, ws: ConnectionSocket, @@ -1061,7 +1055,6 @@ export class TSessionManager implements SessionManager { return { ctx, pipeline, - communicationApi, requestId, sendResponse: (reqId, msg) => sendResponse(sendCtx, service, ws, { @@ -1221,10 +1214,10 @@ export class TSessionManager implements SessionManager { await ws.backpressure(userCtx) } - await workspace.with(async (pipeline, communicationApi) => { + await workspace.with(async (pipeline) => { await userCtx.with('🧨 process', {}, (callTx) => f.apply(service, [ - this.createOpContext(callTx, userCtx, pipeline, communicationApi, request.id, service, ws, rateLimit), + this.createOpContext(callTx, userCtx, pipeline, request.id, service, ws, rateLimit), ...params ]) ) @@ -1285,17 +1278,8 @@ export class TSessionManager implements SessionManager { }) try { - await workspace.with(async (pipeline, communicationApi) => { - const uctx = this.createOpContext( - userCtx, - userCtx, - pipeline, - communicationApi, - reqId, - service, - ws, - rateLimitStatus - ) + await workspace.with(async (pipeline) => { + const uctx = this.createOpContext(userCtx, userCtx, pipeline, reqId, service, ws, rateLimitStatus) await operation(uctx, rateLimitStatus) }) } catch (err: any) { @@ -1379,7 +1363,7 @@ export class TSessionManager implements SessionManager { } const account = service.getRawAccount() - await workspace.with(async (pipeline, communicationApi) => { + await workspace.with(async (pipeline) => { const helloResponse: HelloResponse = { id: -1, result: 'hello', @@ -1394,12 +1378,10 @@ export class TSessionManager implements SessionManager { await ws.send(requestCtx, helloResponse, false, false) }) if (account.uuid !== guestAccount && account.uuid !== systemAccountUuid) { - void workspace.with(async (pipeline, communicationApi) => { + void workspace.with(async (pipeline) => { // We do not need to wait for set-status, just return session to client await ctx - .with('set-status', {}, (ctx) => - this.trySetStatus(ctx, pipeline, communicationApi, service, true, service.workspace.uuid) - ) + .with('set-status', {}, (ctx) => this.trySetStatus(ctx, pipeline, service, true, service.workspace.uuid)) .catch(() => {}) }) } @@ -1423,8 +1405,7 @@ export function createSessionManager ( enableCompression: boolean, doHandleTick: boolean = true, queue: PlatformQueue, - pipelineFactory: PipelineFactory, - communicationApiFactory: CommunicationApiFactory + pipelineFactory: PipelineFactory ): SessionManager { return new TSessionManager( ctx, @@ -1435,14 +1416,12 @@ export function createSessionManager ( enableCompression, doHandleTick, queue, - pipelineFactory, - communicationApiFactory + pipelineFactory ) } export interface SessionManagerOptions extends Partial { pipelineFactory: PipelineFactory - communicationApiFactory: CommunicationApiFactory brandingMap: BrandingMap enableCompression?: boolean accountsUrl: string @@ -1469,8 +1448,7 @@ export function startSessionManager (ctx: MeasureContext, opt: SessionManagerOpt opt.enableCompression ?? false, true, opt.queue, - opt.pipelineFactory, - opt.communicationApiFactory + opt.pipelineFactory ) return sessions } diff --git a/server/server/src/workspace.ts b/server/server/src/workspace.ts index 575dac0909..458bd71fef 100644 --- a/server/server/src/workspace.ts +++ b/server/server/src/workspace.ts @@ -14,7 +14,6 @@ // import { Analytics } from '@hcengineering/analytics' -import { type ServerApi as CommunicationApi } from '@hcengineering/communication-sdk-types' import { type Branding, type MeasureContext, type WorkspaceIds } from '@hcengineering/core' import type { ConnectionSocket, Pipeline, Session } from '@hcengineering/server-core' @@ -23,17 +22,13 @@ interface TickHandler { operation: () => void } -export interface PipelinePair { - pipeline: Pipeline - communicationApi: CommunicationApi -} -export type WorkspacePipelineFactory = () => Promise +export type WorkspacePipelineFactory = () => Promise /** * @public */ export class Workspace { - pipeline?: PipelinePair | Promise + pipeline?: Pipeline | Promise upgrade: boolean = false closing?: Promise @@ -61,22 +56,22 @@ export class Workspace { this.softShutdown = softShutdown } - private getPipelinePair (): PipelinePair | Promise { + private getPipeline (): Pipeline | Promise { if (this.pipeline === undefined) { this.pipeline = this.factory() } return this.pipeline } - async with(op: (pipeline: Pipeline, communicationApi: CommunicationApi) => Promise): Promise { + async with(op: (pipeline: Pipeline) => Promise): Promise { this.operations++ - let pair = this.getPipelinePair() - if (pair instanceof Promise) { - pair = await pair - this.pipeline = pair + let pipeline = this.getPipeline() + if (pipeline instanceof Promise) { + pipeline = await pipeline + this.pipeline = pipeline } try { - return await op(pair.pipeline, pair.communicationApi) + return await op(pipeline) } finally { this.operations-- } @@ -86,7 +81,7 @@ export class Workspace { if (this.pipeline === undefined) { return } - const { pipeline, communicationApi } = await this.pipeline + const pipeline = await this.pipeline const closePipeline = async (): Promise => { try { await ctx.with('close-pipeline', {}, async () => { @@ -98,20 +93,10 @@ export class Workspace { } } - const closeCommunicationApi = async (): Promise => { - try { - await ctx.with('close-communication-api', {}, async () => { - await communicationApi.close() - }) - } catch (err: any) { - Analytics.handleError(err) - ctx.error('close-pipeline-error', { error: err }) - } - } await ctx.with('closing', {}, async () => { const to = timeoutPromise(120000) - const closePromises = [closePipeline(), closeCommunicationApi()] - await Promise.race([Promise.all(closePromises), to.promise]) + const closePromises = closePipeline() + await Promise.race([closePromises, to.promise]) to.cancelHandle() }) } diff --git a/services/msg2file/src/worker.ts b/services/msg2file/src/worker.ts index e08911da1b..07f82d1c2a 100644 --- a/services/msg2file/src/worker.ts +++ b/services/msg2file/src/worker.ts @@ -387,18 +387,21 @@ async function createGroup ( ): Promise { await retry( async () => - await client.event({ - type: MessageEventType.CreateMessagesGroup, - group: { - cardId, - blobId, - fromDate, - toDate, - count + await client.event( + { + type: MessageEventType.CreateMessagesGroup, + group: { + cardId, + blobId, + fromDate, + toDate, + count + }, + socialId: core.account.System, + date: new Date() }, - socialId: core.account.System, - date: new Date() - }), + core.account.System + ), { retries: 3 } ) } @@ -406,13 +409,16 @@ async function createGroup ( async function removeGroup (client: CommunicationRestClient, cardId: CardID, blobId: Ref): Promise { await retry( async () => - await client.event({ - type: MessageEventType.RemoveMessagesGroup, - cardId, - blobId, - socialId: core.account.System, - date: new Date() - }), + await client.event( + { + type: MessageEventType.RemoveMessagesGroup, + cardId, + blobId, + socialId: core.account.System, + date: new Date() + }, + core.account.System + ), { retries: 3 } ) }