diff --git a/packages/core/src/client.ts b/packages/core/src/client.ts index 8d56516f27..c5e490d3d8 100644 --- a/packages/core/src/client.ts +++ b/packages/core/src/client.ts @@ -21,12 +21,10 @@ import { Hierarchy } from './hierarchy' import { MeasureContext, MeasureMetricsContext } from './measurements' import { ModelDb } from './memdb' import type { DocumentQuery, FindOptions, FindResult, FulltextStorage, Storage, TxResult, WithLookup } from './storage' -import { SearchOptions, SearchQuery, SearchResult, SortingOrder } from './storage' -import { Tx, TxCUD } from './tx' +import { SearchOptions, SearchQuery, SearchResult } from './storage' +import { Tx, TxCUD, type TxWorkspaceEvent } from './tx' import { toFindResult } from './utils' -const transactionThreshold = 500 - /** * @public */ @@ -85,11 +83,13 @@ export interface ClientConnection extends Storage, FulltextStorage, BackupClient isConnected: () => boolean close: () => Promise - onConnect?: (event: ClientConnectEvent, data: any) => Promise + onConnect?: (event: ClientConnectEvent, lastTx: string | undefined, data: any) => Promise // If hash is passed, will return LoadModelResponse loadModel: (last: Timestamp, hash?: string) => Promise getAccount: () => Promise + + getLastHash?: (ctx: MeasureContext) => Promise } class ClientImpl implements AccountClient, BackupClient { @@ -236,7 +236,7 @@ export async function createClient ( let hierarchy = new Hierarchy() let model = new ModelDb(hierarchy) - let lastTx: number = 0 + let lastTx: string | undefined function txHandler (...tx: Tx[]): void { if (tx == null || tx.length === 0) { @@ -248,7 +248,11 @@ export async function createClient ( // eslint-disable-next-line @typescript-eslint/no-floating-promises client.updateFromRemote(...tx) } - lastTx = tx.reduce((cur, it) => (it.modifiedOn > cur ? it.modifiedOn : cur), 0) + for (const t of tx) { + if (t._class === core.class.TxWorkspaceEvent) { + lastTx = (t as TxWorkspaceEvent).params.lastTx + } + } } const conn = await ctx.with('connect', {}, () => connect(txHandler)) @@ -264,11 +268,14 @@ export async function createClient ( txHandler(...txBuffer) txBuffer = undefined - const oldOnConnect: ((event: ClientConnectEvent, data: any) => Promise) | undefined = conn.onConnect - conn.onConnect = async (event, data) => { + const oldOnConnect: + | ((event: ClientConnectEvent, lastTx: string | undefined, data: any) => Promise) + | undefined = conn.onConnect + conn.onConnect = async (event, _lastTx, data) => { console.log('Client: onConnect', event) if (event === ClientConnectEvent.Maintenance) { - await oldOnConnect?.(ClientConnectEvent.Maintenance, data) + lastTx = _lastTx + await oldOnConnect?.(ClientConnectEvent.Maintenance, _lastTx, data) return } // Find all new transactions and apply @@ -282,51 +289,27 @@ export async function createClient ( model = new ModelDb(hierarchy) await ctx.with('build-model', {}, (ctx) => buildModel(ctx, loadModelResponse, modelFilter, hierarchy, model)) - await oldOnConnect?.(ClientConnectEvent.Upgraded, data) - + await oldOnConnect?.(ClientConnectEvent.Upgraded, _lastTx, data) + lastTx = _lastTx // No need to fetch more stuff since upgrade was happened. return } - if (event === ClientConnectEvent.Connected) { + if (event === ClientConnectEvent.Connected && _lastTx !== lastTx && lastTx === undefined) { // No need to do anything here since we connected. - await oldOnConnect?.(event, data) + await oldOnConnect?.(event, _lastTx, data) + lastTx = _lastTx return } - // We need to look for last {transactionThreshold} transactions and if it is more since lastTx one we receive, we need to perform full refresh. - if (lastTx === 0) { - await oldOnConnect?.(ClientConnectEvent.Refresh, data) + if (_lastTx === lastTx) { + // Same lastTx, no need to refresh + await oldOnConnect?.(ClientConnectEvent.Reconnected, _lastTx, data) return } - const atxes = await ctx.with('find-atx', {}, () => - conn.findAll( - core.class.Tx, - { modifiedOn: { $gt: lastTx }, objectSpace: { $ne: core.space.Model } }, - { sort: { modifiedOn: SortingOrder.Ascending, _id: SortingOrder.Ascending }, limit: transactionThreshold } - ) - ) - - let needFullRefresh = false - // if we have attachment document create/delete we need to full refresh, since some derived data could be missing - for (const tx of atxes) { - if ( - (tx as TxCUD).attachedTo !== undefined && - (tx._class === core.class.TxCreateDoc || tx._class === core.class.TxRemoveDoc) - ) { - needFullRefresh = true - break - } - } - - if (atxes.length < transactionThreshold && !needFullRefresh) { - console.log('applying input transactions', atxes.length) - txHandler(...atxes) - await oldOnConnect?.(ClientConnectEvent.Reconnected, data) - } else { - // We need to trigger full refresh on queries, etc. - await oldOnConnect?.(ClientConnectEvent.Refresh, data) - } + lastTx = _lastTx + // We need to trigger full refresh on queries, etc. + await oldOnConnect?.(ClientConnectEvent.Refresh, lastTx, data) } return client @@ -344,6 +327,10 @@ async function tryLoadModel ( hash: '' } + if (conn.getLastHash !== undefined && (await conn.getLastHash(ctx)) === current.hash) { + // We have same model hash. + return current + } const lastTxTime = getLastTxTime(current.transactions) const result = await ctx.with('connection-load-model', { hash: current.hash !== '' }, (ctx) => conn.loadModel(lastTxTime, current.hash) @@ -357,21 +344,23 @@ async function tryLoadModel ( hash: '' } } - - // Save concatenated - void ctx - .with('persistence-store', {}, (ctx) => - persistence?.store({ - ...result, - transactions: !result.full ? current.transactions.concat(result.transactions) : result.transactions + const transactions = current.transactions.concat(result.transactions) + if (result.hash !== current.hash) { + // Save concatenated, if have some more of them. + void ctx + .with('persistence-store', {}, (ctx) => + persistence?.store({ + ...result, + transactions: !result.full ? transactions : result.transactions + }) + ) + .catch((err) => { + Analytics.handleError(err) }) - ) - .catch((err) => { - Analytics.handleError(err) - }) + } if (!result.full && !reload) { - result.transactions = current.transactions.concat(result.transactions) + result.transactions = transactions } return result diff --git a/packages/core/src/tx.ts b/packages/core/src/tx.ts index 13df6a1b16..956cf15972 100644 --- a/packages/core/src/tx.ts +++ b/packages/core/src/tx.ts @@ -51,7 +51,8 @@ export enum WorkspaceEvent { IndexingUpdate, SecurityChange, MaintenanceNotification, - BulkUpdate + BulkUpdate, + LastTx } /** diff --git a/packages/ui/src/components/DropdownLabels.svelte b/packages/ui/src/components/DropdownLabels.svelte index 2f58b2d8cd..5228af1abe 100644 --- a/packages/ui/src/components/DropdownLabels.svelte +++ b/packages/ui/src/components/DropdownLabels.svelte @@ -50,9 +50,11 @@ let container: HTMLElement let opened: boolean = false - $: selectedItem = multiselect ? items.filter((p) => selected?.includes(p.id)) : items.find((x) => x.id === selected) - $: if (autoSelect && selected === undefined && items[0] !== undefined) { - selected = multiselect ? [items[0].id] : items[0].id + $: selectedItem = multiselect + ? (items ?? []).filter((p) => selected?.includes(p.id)) + : (items ?? []).find((x) => x.id === selected) + $: if (autoSelect && selected === undefined && items?.[0] !== undefined) { + selected = multiselect ? [items?.[0]?.id] : items?.[0]?.id } const dispatch = createEventDispatcher() @@ -111,7 +113,7 @@ {:else if Array.isArray(selectedItem)} {#if selectedItem.length > 0} - {#each selectedItem as seleceted, i} + {#each selectedItem as seleceted} {seleceted.label} {/each} {:else} diff --git a/plugins/client-resources/src/connection.ts b/plugins/client-resources/src/connection.ts index 37241ce9d8..0535c8b126 100644 --- a/plugins/client-resources/src/connection.ts +++ b/plugins/client-resources/src/connection.ts @@ -108,10 +108,12 @@ class Connection implements ClientConnection { private helloRecieved: boolean = false - onConnect?: (event: ClientConnectEvent, data: any) => Promise + onConnect?: (event: ClientConnectEvent, lastTx: string | undefined, data: any) => Promise rpcHandler = new RPCHandler() + lastHash?: string + constructor ( private readonly ctx: MeasureContext, private readonly url: string, @@ -144,6 +146,11 @@ class Connection implements ClientConnection { this.scheduleOpen(this.ctx, false) } + async getLastHash (ctx: MeasureContext): Promise { + await this.waitOpenConnection(ctx) + return this.lastHash + } + private schedulePing (socketId: number): void { clearInterval(this.interval) this.pingResponse = Date.now() @@ -272,7 +279,7 @@ class Connection implements ClientConnection { if (resp.id === -1) { this.delay = 0 if (resp.result?.state === 'upgrading') { - void this.onConnect?.(ClientConnectEvent.Maintenance, resp.result.stats) + void this.onConnect?.(ClientConnectEvent.Maintenance, undefined, resp.result.stats) this.upgrading = true this.delay = 3 return @@ -286,6 +293,7 @@ class Connection implements ClientConnection { // We need to clear dial timer, since we recieve hello response. clearTimeout(this.dialTimer) this.dialTimer = null + this.lastHash = (resp as HelloResponse).lastHash const serverVersion = helloResp.serverVersion console.log('Connected to server:', serverVersion) @@ -315,6 +323,7 @@ class Connection implements ClientConnection { void this.onConnect?.( (resp as HelloResponse).reconnect === true ? ClientConnectEvent.Reconnected : ClientConnectEvent.Connected, + (resp as HelloResponse).lastTx, this.sessionId ) this.schedulePing(socketId) diff --git a/plugins/client-resources/src/index.ts b/plugins/client-resources/src/index.ts index d98ad98f12..089b3bbd4d 100644 --- a/plugins/client-resources/src/index.ts +++ b/plugins/client-resources/src/index.ts @@ -118,10 +118,10 @@ export default async () => { reject(new Error(`Connection timeout, and no connection established to ${endpoint}`)) } }, connectTimeout) - newOpt.onConnect = async (event, data) => { + newOpt.onConnect = async (event, lastTx, data) => { // Any event is fine, it means server is alive. clearTimeout(connectTO) - await opt?.onConnect?.(event, data) + await opt?.onConnect?.(event, lastTx, data) resolve() } }) diff --git a/plugins/client/src/index.ts b/plugins/client/src/index.ts index 3bd5ccccc2..04f9245097 100644 --- a/plugins/client/src/index.ts +++ b/plugins/client/src/index.ts @@ -62,7 +62,7 @@ export interface ClientFactoryOptions { onUpgrade?: () => void onUnauthorized?: () => void onArchived?: () => void - onConnect?: (event: ClientConnectEvent, data: any) => Promise + onConnect?: (event: ClientConnectEvent, lastTx: string | undefined, data: any) => Promise ctx?: MeasureContext onDialTimeout?: () => void | Promise } diff --git a/plugins/recruit-resources/src/components/VacancyPresenter.svelte b/plugins/recruit-resources/src/components/VacancyPresenter.svelte index ab8f5d998a..35ff43c74e 100644 --- a/plugins/recruit-resources/src/components/VacancyPresenter.svelte +++ b/plugins/recruit-resources/src/components/VacancyPresenter.svelte @@ -31,7 +31,7 @@ export let type: ObjectPresenterType = 'link' const dispatch = createEventDispatcher() - $: accentColor = getPlatformAvatarColorForTextDef(value.name, $themeStore.dark) + $: accentColor = getPlatformAvatarColorForTextDef(value?.name ?? '', $themeStore.dark) $: dispatch('accent-color', accentColor) onMount(() => { diff --git a/server/core/src/types.ts b/server/core/src/types.ts index 189caca7dc..6233c93adf 100644 --- a/server/core/src/types.ts +++ b/server/core/src/types.ts @@ -162,6 +162,11 @@ export interface DBAdapterManager { export interface PipelineContext { workspace: WorkspaceIdWithUrl + + lastTx?: string + + lastHash?: string + hierarchy: Hierarchy modelDb: ModelDb branding: Branding | null diff --git a/server/middleware/src/model.ts b/server/middleware/src/model.ts index 9da40d896f..499f48e01c 100644 --- a/server/middleware/src/model.ts +++ b/server/middleware/src/model.ts @@ -115,6 +115,7 @@ export class ModelMiddleware extends BaseMiddleware implements Middleware { private setLastHash (hash: string): void { this.lastHash = hash + this.context.lastHash = this.lastHash this.lastHashResponse = Promise.resolve({ full: false, hash, diff --git a/server/middleware/src/txPush.ts b/server/middleware/src/txPush.ts index 645ceec041..cf301ae86c 100644 --- a/server/middleware/src/txPush.ts +++ b/server/middleware/src/txPush.ts @@ -16,12 +16,16 @@ import core, { DOMAIN_TRANSIENT, DOMAIN_TX, + generateId, TxProcessor, + WorkspaceEvent, type Doc, type MeasureContext, + type SessionData, type Tx, type TxCUD, - type TxResult + type TxResult, + type TxWorkspaceEvent } from '@hcengineering/core' import { PlatformError, unknownError } from '@hcengineering/platform' import type { DBAdapterManager, Middleware, PipelineContext, TxMiddlewareResult } from '@hcengineering/server-core' @@ -68,6 +72,22 @@ export class TxMiddleware extends BaseMiddleware implements Middleware { txes: Array.from(new Set(txToStore.map((it) => it._class))) } ) + // We need to remember last Tx Id in context, so it will be used during reconnect to track a requirement for refresh. + this.context.lastTx = txToStore[txToStore.length - 1]._id + // We need to deliver information to all clients so far. + const evt: TxWorkspaceEvent = { + _class: core.class.TxWorkspaceEvent, + _id: generateId(), + event: WorkspaceEvent.LastTx, + modifiedBy: core.account.System, + modifiedOn: Date.now(), + objectSpace: core.space.DerivedTx, + space: core.space.DerivedTx, + params: { + lastTx: this.context.lastTx + } + } + ;(ctx.contextData as SessionData).broadcast.txes.push(evt) } if (txPromise !== undefined) { await txPromise diff --git a/server/rpc/src/rpc.ts b/server/rpc/src/rpc.ts index 5ae60ae81f..ac9022c389 100644 --- a/server/rpc/src/rpc.ts +++ b/server/rpc/src/rpc.ts @@ -46,6 +46,8 @@ export interface HelloResponse extends Response { binary: boolean reconnect?: boolean serverVersion: string + lastTx?: string + lastHash?: string // Last model hash } function replacer (key: string, value: any): any { diff --git a/server/server/src/sessionManager.ts b/server/server/src/sessionManager.ts index 193598e130..7a00401bd0 100644 --- a/server/server/src/sessionManager.ts +++ b/server/server/src/sessionManager.ts @@ -990,7 +990,7 @@ class TSessionManager implements SessionManager { return } if (request.id === -1 && request.method === 'hello') { - this.handleHello(request, service, ctx, workspace, ws, requestCtx) + await this.handleHello(request, service, ctx, workspace, ws, requestCtx) return } if (request.id === -2 && request.method === 'forceClose') { @@ -1053,14 +1053,14 @@ class TSessionManager implements SessionManager { }) } - private handleHello( + private async handleHello( request: Request, service: S, ctx: MeasureContext, workspace: string, ws: ConnectionSocket, requestCtx: MeasureContext - ): void { + ): Promise { const hello = request as HelloRequest service.binaryMode = hello.binary ?? false service.useCompression = hello.compression ?? false @@ -1080,12 +1080,16 @@ class TSessionManager implements SessionManager { if (reconnect) { this.reconnectIds.delete(service.sessionId) } + const pipeline = + service.workspace.pipeline instanceof Promise ? await service.workspace.pipeline : service.workspace.pipeline const helloResponse: HelloResponse = { id: -1, result: 'hello', binary: service.binaryMode, reconnect, - serverVersion: this.serverVersion + serverVersion: this.serverVersion, + lastTx: pipeline.context.lastTx, + lastHash: pipeline.context.lastHash } ws.send(requestCtx, helloResponse, false, false) }