diff --git a/.vscode/launch.json b/.vscode/launch.json index 0bd68d2913..cb4d54a67d 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -124,7 +124,7 @@ // "DB_URL": "postgresql://postgres:example@localhost:5432", "DB_URL": "postgresql://root@huly.local:26258/defaultdb?sslmode=disable", // "GREEN_URL": "http://huly.local:6767?token=secret", - "SERVER_PORT": "3334", + "SERVER_PORT": "3335", "METRICS_CONSOLE": "false", "DEBUG_PRINT_SQL": "true", "METRICS_FILE": "${workspaceRoot}/metrics.txt", // Show metrics in console evert 30 seconds., @@ -135,7 +135,7 @@ "FRONT_URL": "http://localhost:8083", "ACCOUNTS_URL": "http://localhost:3003", "MODEL_JSON": "${workspaceRoot}/models/all/bundle/model.json", - "MODEL_VERSION": "0.7.75", + "MODEL_VERSION": "0.7.110", "STATS_URL": "http://huly.local:4901", "QUEUE_CONFIG": "localhost:19093" }, @@ -248,7 +248,7 @@ "request": "launch", "args": ["src/__start.ts"], "env": { - "PORT": "4900", + "PORT": "4901", "SERVER_SECRET": "secret" }, "runtimeVersion": "20", @@ -585,13 +585,12 @@ "PLATFORM_OPERATION_LOGGING": "true", "FRONT_URL": "http://localhost:8080", "PORT": "3500", - "STATS_URL": "http://huly.local:4900" + "STATS_URL": "http://huly.local:4901" }, "runtimeArgs": ["--nolazy", "-r", "ts-node/register"], "sourceMaps": true, "cwd": "${workspaceRoot}/services/github/pod-github", "protocol": "inspector", - "attachSimplePort": 0, "outputCapture": "std" }, { diff --git a/dev/tool/src/benchmark.ts b/dev/tool/src/benchmark.ts index ca3bc6f3ef..8a11b574eb 100644 --- a/dev/tool/src/benchmark.ts +++ b/dev/tool/src/benchmark.ts @@ -254,14 +254,7 @@ export async function benchmark ( // operations = 0 requestTime = 0 // transfer = 0 - const r = extract( - json.metrics as Metrics, - '🧲 session', - 'client', - 'handleRequest', - 'process', - 'find-all' - ) + const r = extract(json.metrics as Metrics, '🧲 session', 'client', 'process', 'find-all') operations = (r?.operations ?? 0) - oldOperations oldOperations = r?.operations ?? 0 diff --git a/dev/tool/src/workspace.ts b/dev/tool/src/workspace.ts index 4cfda47138..7d22191308 100644 --- a/dev/tool/src/workspace.ts +++ b/dev/tool/src/workspace.ts @@ -157,7 +157,7 @@ export async function backupRestore ( dataId: workspace.dataId, url: workspace.url } - const result: boolean = await ctx.with('restore', { workspace: workspace.url }, (ctx) => + const result: boolean = await ctx.with('restore', {}, (ctx) => restore(ctx, '', wsUrl, storage, { date: -1, skip: new Set(skipDomains), diff --git a/packages/core/src/measurements/metrics.ts b/packages/core/src/measurements/metrics.ts index fc6bfe7852..a5ec906cb4 100644 --- a/packages/core/src/measurements/metrics.ts +++ b/packages/core/src/measurements/metrics.ts @@ -84,7 +84,9 @@ export function updateMeasure ( const fParams = typeof fullParams === 'function' ? fullParams() : fullParams // Update params if required - for (const [k, v] of Object.entries(params)) { + const pparams = Object.entries(params) + if (pparams.length > 0) { + const [k, v] = pparams[0] let params = metrics.params[k] if (params === undefined) { params = {} @@ -106,7 +108,22 @@ export function updateMeasure ( param.operations++ } // Do not update top results for params. - // param.topResult = getUpdatedTopResult(param.topResult, ed - st, fParams) + if (pparams.length > 1) { + // We need to update all other params as counters. + if (param.topResult === undefined) { + param.topResult = [] + } + for (const [, v] of pparams.slice(1)) { + const r = (param.topResult ?? []).find((it) => it.params[`${v}`] === true) + if (r !== undefined) { + r.value += 1 // Counter of operations + r.time = (r.time ?? 0) + (value ?? ed - st) + } else { + param.topResult.push({ params: { [`${v}`]: true }, value: 1, time: value ?? ed - st }) + } + } + param.topResult.sort((a, b) => b.value - a.value) + } } // Update leaf data if (override === true) { diff --git a/packages/core/src/measurements/types.ts b/packages/core/src/measurements/types.ts index 52925ce27a..84b9fa483a 100644 --- a/packages/core/src/measurements/types.ts +++ b/packages/core/src/measurements/types.ts @@ -21,6 +21,7 @@ export interface MetricsData { value: number topResult?: { value: number + time?: number params: FullParamsType }[] } diff --git a/packages/core/src/server.ts b/packages/core/src/server.ts index 96fd04ef13..e277606948 100644 --- a/packages/core/src/server.ts +++ b/packages/core/src/server.ts @@ -46,6 +46,7 @@ export interface SessionData { contextCache: Map removedMap: Map, Doc> account: Account + service: string sessionId: string admin?: boolean isTriggerCtx?: boolean diff --git a/plugins/workbench-resources/src/components/statistics/MetricsInfo.svelte b/plugins/workbench-resources/src/components/statistics/MetricsInfo.svelte index 82680fefa7..f4eb29c222 100644 --- a/plugins/workbench-resources/src/components/statistics/MetricsInfo.svelte +++ b/plugins/workbench-resources/src/components/statistics/MetricsInfo.svelte @@ -151,16 +151,25 @@ {#if childExpandable}
{#each vv.topResult ?? [] as r} - - -
- Time:{toTime(r.value)} -
- - 
-                    {JSON.stringify(r, null, 2)}
-
- + +
+ +
+ {Object.entries(r.params)[0][0]} +
+ + +
+ {r.value} +
+ + +
+ {toTime(r.time ?? 0)} +
+ +
+ {/each}
{/if} diff --git a/pods/fulltext/src/__tests__/utils.ts b/pods/fulltext/src/__tests__/utils.ts index b40f9aec8c..308a381f9d 100644 --- a/pods/fulltext/src/__tests__/utils.ts +++ b/pods/fulltext/src/__tests__/utils.ts @@ -76,7 +76,7 @@ export async function preparePipeline ( const middlewares: MiddlewareCreator[] = [ TxMiddleware.create, // Store tx into transaction domain - FullTextMiddleware.create('', generateToken(systemAccountUuid, wsIds.uuid)), + FullTextMiddleware.create('', generateToken(systemAccountUuid, wsIds.uuid, { service: 'fulltext' })), LowLevelMiddleware.create, QueryJoinMiddleware.create, DomainFindMiddleware.create, diff --git a/pods/fulltext/src/manager.ts b/pods/fulltext/src/manager.ts index 3acfe152c1..3b0393cbf0 100644 --- a/pods/fulltext/src/manager.ts +++ b/pods/fulltext/src/manager.ts @@ -116,7 +116,14 @@ export class WorkspaceManager { for (const m of msg) { const ws = m.id as WorkspaceUuid - const indexer = await this.getIndexer(this.ctx, ws, generateToken(systemAccountUuid, ws), true) + const indexer = await this.getIndexer( + this.ctx, + ws, + generateToken(systemAccountUuid, ws, { + service: 'fulltext' + }), + true + ) await indexer?.fulltext.processDocuments(this.ctx, m.value, control) } } @@ -135,7 +142,12 @@ export class WorkspaceManager { mm.type === QueueWorkspaceEvent.Restored || mm.type === QueueWorkspaceEvent.FullReindex ) { - const indexer = await this.getIndexer(this.ctx, ws, generateToken(systemAccountUuid, ws), true) + const indexer = await this.getIndexer( + this.ctx, + ws, + generateToken(systemAccountUuid, ws, { service: 'fulltext' }), + true + ) if (indexer !== undefined) { await indexer.dropWorkspace() // TODO: Add heartbeat const classes = await indexer.getIndexClassess() @@ -149,7 +161,7 @@ export class WorkspaceManager { mm.type === QueueWorkspaceEvent.Archived || mm.type === QueueWorkspaceEvent.ClearIndex ) { - const token = generateToken(systemAccountUuid, ws) + const token = generateToken(systemAccountUuid, ws, { service: 'fulltext' }) const workspaceInfo = await this.getWorkspaceInfo(token) if (workspaceInfo !== undefined) { if (workspaceInfo.dataId != null) { @@ -158,7 +170,12 @@ export class WorkspaceManager { await this.fulltextAdapter.clean(this.ctx, workspaceInfo.uuid) } } else if (mm.type === QueueWorkspaceEvent.Reindex) { - const indexer = await this.getIndexer(this.ctx, ws, generateToken(systemAccountUuid, ws), true) + const indexer = await this.getIndexer( + this.ctx, + ws, + generateToken(systemAccountUuid, ws, { service: 'fulltext' }), + true + ) const mmd = mm as QueueWorkspaceReindexMessage await indexer?.reindex(this.ctx, mmd.domain, mmd.classes, control) } diff --git a/pods/fulltext/src/workspace.ts b/pods/fulltext/src/workspace.ts index a762cb5ee1..dbf67292e2 100644 --- a/pods/fulltext/src/workspace.ts +++ b/pods/fulltext/src/workspace.ts @@ -95,7 +95,7 @@ export class WorkspaceIndexer { throw new PlatformError(unknownError('Default adapter should be set')) } - const token = generateToken(systemAccountUuid, workspace.uuid) + const token = generateToken(systemAccountUuid, workspace.uuid, { service: 'fulltext' }) const transactorEndpoint = await endpointProvider(token) result.fulltext = new FullTextIndexPipeline( diff --git a/pods/server/src/rpc.ts b/pods/server/src/rpc.ts index 3d0d6d2c06..39c51465c7 100644 --- a/pods/server/src/rpc.ts +++ b/pods/server/src/rpc.ts @@ -40,6 +40,7 @@ interface RPCClientInfo { client: ConnectionSocket session: Session workspaceId: string + context: MeasureContext } const gzipAsync = promisify(gzip) @@ -173,12 +174,12 @@ export function registerRPC (app: Express, sessions: SessionManager, ctx: Measur }) return } - transactorRpc = { session: s.session, client: cs, workspaceId: s.workspaceId } + transactorRpc = { session: s.session, client: cs, workspaceId: s.workspaceId, context: s.context } rpcSessions.set(token, transactorRpc) } const rpc = transactorRpc - const rateLimit = await sessions.handleRPC(ctx, rpc.session, rpc.client, async (ctx, rateLimit) => { + const rateLimit = await sessions.handleRPC(rpc.context, rpc.session, rpc.client, async (ctx, rateLimit) => { await operation(ctx, rpc.session, rateLimit, token) }) if (rateLimit !== undefined) { diff --git a/pods/server/src/server_http.ts b/pods/server/src/server_http.ts index 8eeb993d6d..4a092bf830 100644 --- a/pods/server/src/server_http.ts +++ b/pods/server/src/server_http.ts @@ -320,7 +320,7 @@ export function startHttpServer ( } await ctx.with( 'storage upload', - { workspace: wsIds.uuid }, + {}, async (ctx) => { await externalStorage.put( ctx, @@ -336,7 +336,7 @@ export function startHttpServer ( }) res.end(JSON.stringify({ success: true })) }, - { file: name, contentType } + { file: name, contentType, workspace: wsIds.uuid } ) } catch (err: any) { Analytics.handleError(err) diff --git a/pods/stats/src/stats.ts b/pods/stats/src/stats.ts index 1400e1c8a4..438863bab7 100644 --- a/pods/stats/src/stats.ts +++ b/pods/stats/src/stats.ts @@ -15,6 +15,7 @@ import { } from '@hcengineering/server-core' import serverToken, { decodeToken } from '@hcengineering/server-token' import cors from '@koa/cors' +import type { IncomingHttpHeaders } from 'http' import Koa from 'koa' import bodyParser from 'koa-bodyparser' import Router from 'koa-router' @@ -36,6 +37,14 @@ interface OverviewStatistics { workspaces: WorkspaceStatistics[] } +const extractAuthorizationToken = (headers: IncomingHttpHeaders): string | undefined => { + try { + return headers.authorization?.slice(7) ?? undefined + } catch { + return undefined + } +} + /** * @public */ @@ -70,7 +79,7 @@ export function serveStats (ctx: MeasureContext, onClose?: () => void): void { router.get('/api/v1/overview', (req, res) => { try { - const token = req.query.token as string + const token = (req.query.token as string) ?? extractAuthorizationToken(req.headers) const payload = decodeToken(token) const admin = payload.extra?.admin === 'true' if (!admin) { @@ -98,7 +107,7 @@ export function serveStats (ctx: MeasureContext, onClose?: () => void): void { const json: Record> = {} for (const [k, v] of statistics.entries()) { if (Date.now() - v.lastUpdate > serviceTimeout) { - timeouts.set(v.serviceName, (timeouts.get(v.serviceName) ?? 0) + 1) + timeouts.set(k, (timeouts.get(k) ?? 0) + 1) toClean.push(k) continue } @@ -117,6 +126,7 @@ export function serveStats (ctx: MeasureContext, onClose?: () => void): void { } } for (const k of toClean) { + timeouts.delete(k) statistics.delete(k) } @@ -139,7 +149,7 @@ export function serveStats (ctx: MeasureContext, onClose?: () => void): void { router.get('/api/v1/statistics', (req, res) => { try { - const token = req.query.token as string + const token = (req.query.token as string) ?? extractAuthorizationToken(req.headers) const payload = decodeToken(token) const admin = payload.extra?.admin === 'true' ctx.info('get stats', { admin, service: req.query.name }) @@ -167,9 +177,9 @@ export function serveStats (ctx: MeasureContext, onClose?: () => void): void { }) router.put('/api/v1/statistics', (req, res) => { try { - const token = req.query.token as string + const token = (req.query.token as string) ?? extractAuthorizationToken(req.headers) const payload = decodeToken(token) - const service = payload.extra?.service === 'true' + const service = payload.extra?.service != null const serviceName = (req.query.name as string) ?? '' if (service) { ctx.info('put stats', { service: req.query.name, len: req.request.length }) diff --git a/server-plugins/calendar-resources/src/index.ts b/server-plugins/calendar-resources/src/index.ts index 70cf5b19cc..1024775879 100644 --- a/server-plugins/calendar-resources/src/index.ts +++ b/server-plugins/calendar-resources/src/index.ts @@ -319,7 +319,7 @@ async function sendEventToService ( method: 'POST', keepalive: true, headers: { - Authorization: 'Bearer ' + generateToken(systemAccountUuid, workspace), + Authorization: 'Bearer ' + generateToken(systemAccountUuid, workspace, { service: 'calendar' }), 'Content-Type': 'application/json' }, body: JSON.stringify({ diff --git a/server/account-service/src/index.ts b/server/account-service/src/index.ts index cb6d01c1c6..1839dba90d 100644 --- a/server/account-service/src/index.ts +++ b/server/account-service/src/index.ts @@ -295,7 +295,7 @@ export function serveAccount (measureCtx: MeasureContext, brandings: BrandingMap router.put('/api/v1/manage', async (req, res) => { try { - const token = req.query.token as string + const token = (req.query.token as string) ?? extractToken(req.headers) const payload = decodeToken(token) if (payload.extra?.admin !== 'true') { req.res.writeHead(404, {}) @@ -358,7 +358,15 @@ export function serveAccount (measureCtx: MeasureContext, brandings: BrandingMap host = new URL(origin).host } const branding = host !== undefined ? brandings[host] : null - const result = await measureCtx.with(request.method, {}, (mctx) => { + + let source = '' + try { + source = (token != null ? decodeToken(token).extra?.service : undefined) ?? '🤦‍♂️user' + } catch (err) { + // Ignore + } + + const result = await measureCtx.with(request.method, { source }, (mctx) => { if (method === undefined || typeof method !== 'function') { const response = { id: request.id, diff --git a/server/collaborator/src/extensions/authentication.ts b/server/collaborator/src/extensions/authentication.ts index cea8d66336..f80a6568a5 100644 --- a/server/collaborator/src/extensions/authentication.ts +++ b/server/collaborator/src/extensions/authentication.ts @@ -36,30 +36,35 @@ export class AuthenticationExtension implements Extension { const ctx = this.configuration.ctx const { workspaceId } = decodeDocumentId(data.documentName) - return await ctx.with('authenticate', { workspaceId }, async () => { - const token = decodeToken(data.token) - const readonly = isGuest(token) + return await ctx.with( + 'authenticate', + {}, + async () => { + const token = decodeToken(data.token) + const readonly = isGuest(token) - ctx.info('authenticate', { - workspaceId, - account: token.account, - mode: token.extra?.mode ?? '', - readonly - }) + ctx.info('authenticate', { + workspaceId, + account: token.account, + mode: token.extra?.mode ?? '', + readonly + }) - if (readonly) { - data.connection.readOnly = true - } + if (readonly) { + data.connection.readOnly = true + } - // verify workspace can be accessed with the token - const ids = await getWorkspaceIds(data.token) + // verify workspace can be accessed with the token + const ids = await getWorkspaceIds(data.token) - // verify workspace uuid in the document matches the token - if (ids.uuid !== workspaceId) { - throw new Error('documentName must include workspace id') - } + // verify workspace uuid in the document matches the token + if (ids.uuid !== workspaceId) { + throw new Error('documentName must include workspace id') + } - return buildContext(data, ids) - }) + return buildContext(data, ids) + }, + { workspaceId } + ) } } diff --git a/server/collaborator/src/server.ts b/server/collaborator/src/server.ts index 30eff08426..341992c274 100644 --- a/server/collaborator/src/server.ts +++ b/server/collaborator/src/server.ts @@ -186,17 +186,24 @@ export async function start (ctx: MeasureContext, config: Config, storageAdapter const context = await getContext(rawToken, token) rpcCtx.info('rpc', { method: request.method, connectionId: context.connectionId, mode: token.extra?.mode ?? '' }) - await rpcCtx.with('/rpc', { method: request.method }, async (ctx) => { - try { - const response: RpcResponse = await rpcCtx.with(request.method, {}, (ctx) => { - return method(ctx, context, documentId, request.payload, { hocuspocus, storageAdapter, transformer }) - }) - res.status(200).send(response) - } catch (err: any) { - Analytics.handleError(err) - res.status(500).send({ error: err.message }) + await rpcCtx.with( + '/rpc', + { + source: token.extra?.service ?? '🤦‍♂️user', + method: request.method + }, + async (ctx) => { + try { + const response: RpcResponse = await rpcCtx.with(request.method, {}, (ctx) => { + return method(ctx, context, documentId, request.payload, { hocuspocus, storageAdapter, transformer }) + }) + res.status(200).send(response) + } catch (err: any) { + Analytics.handleError(err) + res.status(500).send({ error: err.message }) + } } - }) + ) }) const wss = new WebSocketServer({ diff --git a/server/core/src/pipeline.ts b/server/core/src/pipeline.ts index 4c37c9f9f6..b9664b0e50 100644 --- a/server/core/src/pipeline.ts +++ b/server/core/src/pipeline.ts @@ -45,7 +45,7 @@ export async function createPipeline ( constructors: MiddlewareCreator[], context: PipelineContext ): Promise { - return await PipelineImpl.create(ctx.newChild('pipeline-operations', {}), constructors, context) + return await PipelineImpl.create(ctx, constructors, context) } class PipelineImpl implements Pipeline { diff --git a/server/core/src/stats.ts b/server/core/src/stats.ts index cdcfc072dd..d978d0d827 100644 --- a/server/core/src/stats.ts +++ b/server/core/src/stats.ts @@ -100,13 +100,14 @@ export function initStatisticsContext ( let errorToSend = 0 if (metricsFile !== undefined || ops?.logConsole === true || statsUrl !== undefined) { + metricsContext.info('using stats url', { statsUrl, service: serviceName ?? '' }) if (metricsFile !== undefined) { console.info('storing measurements into local file', metricsFile) } let oldMetricsValue = '' const serviceId = encodeURIComponent(os.hostname() + '-' + serviceName) - let prev: Promise | undefined + let prev: Promise | Promise | undefined const handleError = (err: any): void => { errorToSend++ if (errorToSend % 2 === 0) { @@ -138,7 +139,7 @@ export function initStatisticsContext ( return } if (statsUrl !== undefined) { - const token = generateToken(systemAccountUuid, undefined, { service: 'true' }) + const token = generateToken(systemAccountUuid, undefined, { service: serviceName }) const data: ServiceStatistics = { serviceName: ops?.serviceName?.() ?? serviceName, cpu: getCPUInfo(), @@ -149,20 +150,18 @@ export function initStatisticsContext ( const statData = JSON.stringify(data) - prev = fetch( - concatLink(statsUrl, '/api/v1/statistics') + `/?token=${encodeURIComponent(token)}&name=${serviceId}`, - { - method: 'PUT', - headers: { - 'Content-Type': 'application/json' - }, - body: statData - } - ) - .catch(handleError) - .then(() => { + prev = fetch(concatLink(statsUrl, '/api/v1/statistics') + `/?name=${serviceId}`, { + method: 'PUT', + headers: { + 'Content-Type': 'application/json', + authorization: `Bearer ${token}` + }, + body: statData + }) + .finally(() => { prev = undefined }) + .catch(handleError) } } catch (err: any) { handleError(err) diff --git a/server/core/src/types.ts b/server/core/src/types.ts index 3460aafa4e..6a18512ce4 100644 --- a/server/core/src/types.ts +++ b/server/core/src/types.ts @@ -563,6 +563,8 @@ export interface ClientSessionCtx { */ export interface Session { workspace: WorkspaceIds + + token: Token createTime: number // Session restore information diff --git a/server/core/src/utils.ts b/server/core/src/utils.ts index 696f973e23..84a213dd48 100644 --- a/server/core/src/utils.ts +++ b/server/core/src/utils.ts @@ -169,7 +169,8 @@ export class SessionDataImpl implements SessionData { _removedMap: Map, Doc> | undefined, _contextCache: Map | undefined, readonly modelDb: ModelDb, - readonly socialStringsToUsers: Map + readonly socialStringsToUsers: Map, + readonly service: string ) { this._removedMap = _removedMap this._contextCache = _contextCache @@ -246,7 +247,8 @@ export function wrapPipeline ( undefined, undefined, pipeline.context.modelDb, - new Map() + new Map(), + 'transactor' ) ctx.contextData = contextData if (pipeline.context.lowLevelStorage === undefined) { diff --git a/server/front/src/index.ts b/server/front/src/index.ts index 31e46a6ea3..51c606f9fb 100644 --- a/server/front/src/index.ts +++ b/server/front/src/index.ts @@ -57,9 +57,9 @@ async function storageUpload ( const data = file.tempFilePath !== undefined ? fs.createReadStream(file.tempFilePath) : file.data const resp = await ctx.with( 'storage upload', - { workspace: wsIds.uuid }, + {}, (ctx) => storageAdapter.put(ctx, wsIds, uuid, data, file.mimetype, file.size), - { file: file.name, contentType: file.mimetype } + { file: file.name, contentType: file.mimetype, workspace: wsIds.uuid } ) ctx.info('storage upload', resp) @@ -522,15 +522,18 @@ export function start ( const range = req.headers.range if (range !== undefined) { - await ctx.with('file-range', { workspace: wsIds.uuid }, (ctx) => - getFileRange(ctx, blobInfo as PlatformBlob, range, config.storageAdapter, wsIds, res) + await ctx.with( + 'file-range', + {}, + (ctx) => getFileRange(ctx, blobInfo as PlatformBlob, range, config.storageAdapter, wsIds, res), + { workspace: wsIds.uuid } ) } else { await ctx.with( 'file', - { workspace: wsIds.uuid }, + {}, (ctx) => getFile(ctx, blobInfo as PlatformBlob, config.storageAdapter, wsIds, req, res), - { uuid } + { uuid, workspace: wsIds.uuid } ) } } catch (error: any) { diff --git a/server/indexer/src/indexer/indexer.ts b/server/indexer/src/indexer/indexer.ts index a8bf946eae..36714ec17a 100644 --- a/server/indexer/src/indexer/indexer.ts +++ b/server/indexer/src/indexer/indexer.ts @@ -563,7 +563,8 @@ export class FullTextIndexPipeline implements FullTextPipeline { undefined, undefined, this.model, - new Map() + new Map(), + 'fulltext' ) } diff --git a/server/middleware/src/contextName.ts b/server/middleware/src/contextName.ts index 4e712462df..0de34a3e0c 100644 --- a/server/middleware/src/contextName.ts +++ b/server/middleware/src/contextName.ts @@ -37,7 +37,7 @@ export class ContextNameMiddleware extends BaseMiddleware implements Middleware return new ContextNameMiddleware(context, next) } - async tx (ctx: MeasureContext, txes: Tx[]): Promise { + async tx (ctx: MeasureContext, txes: Tx[]): Promise { let measureName: string | undefined const tx = txes.find((it) => it._class === core.class.TxApplyIf) @@ -53,10 +53,10 @@ export class ContextNameMiddleware extends BaseMiddleware implements Middleware const result = await ctx.with( measureName !== undefined ? `📶 ${measureName}` : 'client-tx', - { _class: tx?._class }, + { source: ctx.contextData.service }, (ctx) => { ;({ opLogMetrics, op } = registerOperationLog(ctx)) - return this.provideTx(ctx as MeasureContext, txes) + return this.provideTx(ctx, txes) } ) updateOperationLog(opLogMetrics, op) diff --git a/server/middleware/src/domainFind.ts b/server/middleware/src/domainFind.ts index 6678bd2408..1883034cc1 100644 --- a/server/middleware/src/domainFind.ts +++ b/server/middleware/src/domainFind.ts @@ -22,6 +22,7 @@ import { type FindResult, type MeasureContext, type Ref, + type SessionData, DOMAIN_MODEL } from '@hcengineering/core' import { PlatformError, unknownError } from '@hcengineering/platform' @@ -51,7 +52,7 @@ export class DomainFindMiddleware extends BaseMiddleware implements Middleware { } findAll( - ctx: MeasureContext, + ctx: MeasureContext, _class: Ref>, query: DocumentQuery, options?: ServerFindOptions @@ -67,7 +68,7 @@ export class DomainFindMiddleware extends BaseMiddleware implements Middleware { } return ctx.with( p + '-find-all', - { _class }, + { source: ctx.contextData?.service ?? 'system', _class }, (ctx) => { return this.adapterManager.getAdapter(domain, false).findAll(ctx, _class, query, options) }, diff --git a/server/middleware/src/triggers.ts b/server/middleware/src/triggers.ts index f36ab705b2..b2044c3524 100644 --- a/server/middleware/src/triggers.ts +++ b/server/middleware/src/triggers.ts @@ -226,7 +226,8 @@ export class TriggersMiddleware extends BaseMiddleware implements Middleware { sctx.removedMap, sctx.contextCache, this.context.modelDb, - sctx.socialStringsToUsers + sctx.socialStringsToUsers, + sctx.service ) ctx.contextData = asyncContextData const aresult = await this.triggers.apply( diff --git a/server/server-pipeline/src/pipeline.ts b/server/server-pipeline/src/pipeline.ts index 6709d095cf..cdd30289c7 100644 --- a/server/server-pipeline/src/pipeline.ts +++ b/server/server-pipeline/src/pipeline.ts @@ -133,7 +133,12 @@ export function createServerPipeline ( TxMiddleware.create, // Store tx into transaction domain ...(opt.disableTriggers === true ? [] : [TriggersMiddleware.create]), ...(opt.fulltextUrl !== undefined - ? [FullTextMiddleware.create(opt.fulltextUrl, generateToken(systemAccountUuid, workspace.uuid))] + ? [ + FullTextMiddleware.create( + opt.fulltextUrl, + generateToken(systemAccountUuid, workspace.uuid, { service: 'transactor' }) + ) + ] : []), LowLevelMiddleware.create, QueryJoinMiddleware.create, diff --git a/server/server/src/client.ts b/server/server/src/client.ts index 430df99ef6..8101f29aa8 100644 --- a/server/server/src/client.ts +++ b/server/server/src/client.ts @@ -95,7 +95,7 @@ export class ClientSession implements Session { isAdmin: boolean constructor ( - protected readonly token: Token, + readonly token: Token, readonly workspace: WorkspaceIds, readonly account: Account, readonly info: LoginInfoWithWorkspaces, @@ -164,7 +164,8 @@ export class ClientSession implements Session { undefined, undefined, ctx.pipeline.context.modelDb, - ctx.socialStringsToUsers + ctx.socialStringsToUsers, + this.token.extra?.service ?? '🤦‍♂️user' ) ctx.ctx.contextData = contextData } diff --git a/server/server/src/sessionManager.ts b/server/server/src/sessionManager.ts index 3c047d1891..3bf8f51d1f 100644 --- a/server/server/src/sessionManager.ts +++ b/server/server/src/sessionManager.ts @@ -121,6 +121,7 @@ export class TSessionManager implements SessionManager { now: number = Date.now() + ticksContext: MeasureContext constructor ( readonly ctx: MeasureContext, readonly timeouts: Timeouts, @@ -145,6 +146,8 @@ export class TSessionManager implements SessionManager { } this.workspaceProducer = this.queue.getProducer(ctx.newChild('queue', {}), QueueTopic.Workspace) this.usersProducer = this.queue.getProducer(ctx.newChild('queue', {}), QueueTopic.Users) + + this.ticksContext = ctx.newChild('ticks', {}) } scheduleMaintenance (timeMinutes: number): void { @@ -218,7 +221,7 @@ export class TSessionManager implements SessionManager { break } } - void this.getWorkspaceInfo(workspace.token, connected).catch(() => { + void this.getWorkspaceInfo(this.ticksContext, workspace.token, connected).catch(() => { // Ignore }) } catch (err: any) { @@ -280,7 +283,7 @@ export class TSessionManager implements SessionManager { this.ctx.warn('session hang, closing...', { wsId, user: s.session.getUser() }) // Force close workspace if only one client and it hang. - void this.close(this.ctx, s.socket, wsId).catch((err) => { + void this.close(this.ticksContext, s.socket, wsId).catch((err) => { this.ctx.error('failed to close', err) }) continue @@ -293,7 +296,7 @@ export class TSessionManager implements SessionManager { // And ping other wize s.session.lastPing = now if (s.socket.checkState()) { - void s.socket.send(this.ctx, { result: pingConst }, s.session.binaryMode, s.session.useCompression) + void s.socket.send(this.ticksContext, { result: pingConst }, s.session.binaryMode, s.session.useCompression) } } for (const r of s.session.requests.values()) { @@ -343,7 +346,12 @@ export class TSessionManager implements SessionManager { ) } - async getWorkspaceInfo (token: string, updateLastVisit = true): Promise { + @withContext('🧭 get-workspace-info') + async getWorkspaceInfo ( + ctx: MeasureContext, + token: string, + updateLastVisit = true + ): Promise { try { return await getAccountClient(this.accountsUrl, token).getWorkspaceInfo(updateLastVisit) } catch (err: any) { @@ -354,7 +362,8 @@ export class TSessionManager implements SessionManager { } } - async getLoginWithWorkspaceInfo (token: string): Promise { + @withContext('🧭 get-login-with-workspace-info') + async getLoginWithWorkspaceInfo (ctx: MeasureContext, token: string): Promise { try { const accountClient = getAccountClient(this.accountsUrl, token) return await accountClient.getLoginWithWorkspaceInfo() @@ -374,6 +383,7 @@ export class TSessionManager implements SessionManager { tickCounter = 0 + @withContext('🧭 get-workspace') async getWorkspace ( ctx: MeasureContext, workspaceUuid: WorkspaceUuid, @@ -480,7 +490,6 @@ export class TSessionManager implements SessionManager { return { workspace } } - @withContext('📲 add-session') async addSession ( ctx: MeasureContext, ws: ConnectionSocket, @@ -488,92 +497,94 @@ export class TSessionManager implements SessionManager { rawToken: string, sessionId: string | undefined ): Promise { - let account: LoginInfoWithWorkspaces | undefined + return await ctx.with('📲 add-session', { source: token.extra?.service ?? '🤦‍♂️user' }, async (ctx) => { + let account: LoginInfoWithWorkspaces | undefined - try { - account = await this.getLoginWithWorkspaceInfo(rawToken) - } catch (err: any) { - return { error: err } - } + try { + account = await this.getLoginWithWorkspaceInfo(ctx, rawToken) + } catch (err: any) { + return { error: err } + } - if (account === undefined) { - return { error: new Error('Account not found or not available'), terminate: true } - } + if (account === undefined) { + return { error: new Error('Account not found or not available'), terminate: true } + } - let wsInfo = account.workspaces[token.workspace] + let wsInfo = account.workspaces[token.workspace] - if (wsInfo === undefined) { - // In case of guest or system account - // We need to get workspace info for system account. - const workspaceInfo = await this.getWorkspaceInfo(rawToken, false) - if (workspaceInfo === undefined) { + if (wsInfo === undefined) { + // In case of guest or system account + // We need to get workspace info for system account. + const workspaceInfo = await this.getWorkspaceInfo(ctx, rawToken, false) + if (workspaceInfo === undefined) { + return { error: new Error('Workspace not found or not available'), terminate: true } + } + wsInfo = { + url: workspaceInfo.url, + mode: workspaceInfo.mode, + dataId: workspaceInfo.dataId, + version: { + versionMajor: workspaceInfo.versionMajor, + versionMinor: workspaceInfo.versionMinor, + versionPatch: workspaceInfo.versionPatch + }, + role: AccountRole.Owner, + endpoint: { externalUrl: '', internalUrl: '', region: workspaceInfo.region ?? '' }, + progress: workspaceInfo.processingProgress + } + } + const { workspace, resp } = await this.getWorkspace(ctx.parent ?? ctx, token.workspace, wsInfo, token, ws) + if (resp !== undefined) { + return resp + } + + if (workspace === undefined || account === undefined) { + // Should not happen return { error: new Error('Workspace not found or not available'), terminate: true } } - wsInfo = { - url: workspaceInfo.url, - mode: workspaceInfo.mode, - dataId: workspaceInfo.dataId, - version: { - versionMajor: workspaceInfo.versionMajor, - versionMinor: workspaceInfo.versionMinor, - versionPatch: workspaceInfo.versionPatch - }, - role: AccountRole.Owner, - endpoint: { externalUrl: '', internalUrl: '', region: workspaceInfo.region ?? '' }, - progress: workspaceInfo.processingProgress + + const oldSession = sessionId !== undefined ? workspace.sessions?.get(sessionId) : undefined + if (oldSession !== undefined) { + // Just close old socket for old session id. + await this.close(ctx, oldSession.socket, workspace.wsId.uuid) } - } - const { workspace, resp } = await this.getWorkspace(ctx, token.workspace, wsInfo, token, ws) - if (resp !== undefined) { - return resp - } - if (workspace === undefined || account === undefined) { - // Should not happen - return { error: new Error('Workspace not found or not available'), terminate: true } - } + const session = this.createSession(token, workspace.wsId, account) - const oldSession = sessionId !== undefined ? workspace.sessions?.get(sessionId) : undefined - if (oldSession !== undefined) { - // Just close old socket for old session id. - await this.close(ctx, oldSession.socket, workspace.wsId.uuid) - } + session.sessionId = sessionId !== undefined && (sessionId ?? '').trim().length > 0 ? sessionId : generateId() + session.sessionInstanceId = generateId() + const tickHash = this.tickCounter % ticksPerSecond - const session = this.createSession(token, workspace.wsId, account) + this.sessions.set(ws.id, { session, socket: ws, tickHash }) + // We need to delete previous session with Id if found. + this.tickCounter++ + workspace.sessions.set(session.sessionId, { session, socket: ws, tickHash }) - session.sessionId = sessionId !== undefined && (sessionId ?? '').trim().length > 0 ? sessionId : generateId() - session.sessionInstanceId = generateId() - const tickHash = this.tickCounter % ticksPerSecond + const accountUuid = account.account + if (accountUuid !== systemAccountUuid && accountUuid !== guestAccount) { + await this.usersProducer.send(workspace.wsId.uuid, [ + userEvents.login({ + user: accountUuid, + sessions: this.countUserSessions(workspace, accountUuid), + socialIds: account.socialIds.map((it) => it._id) + }) + ]) + } - this.sessions.set(ws.id, { session, socket: ws, tickHash }) - // We need to delete previous session with Id if found. - this.tickCounter++ - workspace.sessions.set(session.sessionId, { session, socket: ws, tickHash }) + // Mark workspace as init completed and we had at least one client. + if (!workspace.workspaceInitCompleted) { + workspace.workspaceInitCompleted = true + } - const accountUuid = account.account - if (accountUuid !== systemAccountUuid && accountUuid !== guestAccount) { - await this.usersProducer.send(workspace.wsId.uuid, [ - userEvents.login({ - user: accountUuid, - sessions: this.countUserSessions(workspace, accountUuid), - socialIds: account.socialIds.map((it) => it._id) - }) - ]) - } - - // Mark workspace as init completed and we had at least one client. - if (!workspace.workspaceInitCompleted) { - workspace.workspaceInitCompleted = true - } - - if (this.timeMinutes > 0) { - void ws - .send(ctx, { result: this.createMaintenanceWarning() }, session.binaryMode, session.useCompression) - .catch((err) => { - ctx.error('failed to send maintenance warning', err) - }) - } - return { session, context: workspace.context, workspaceId: workspace.wsId.uuid } + if (this.timeMinutes > 0) { + void ws + .send(ctx, { result: this.createMaintenanceWarning() }, session.binaryMode, session.useCompression) + .catch((err) => { + ctx.error('failed to send maintenance warning', err) + }) + } + return { session, context: workspace.context, workspaceId: workspace.wsId.uuid } + }) } private async switchToUpgradeSession ( @@ -744,7 +755,7 @@ export class TSessionManager implements SessionManager { } const workspace: Workspace = new Workspace( context, - generateToken(systemAccountUuid, token.workspace), + generateToken(systemAccountUuid, token.workspace, { service: 'transactor' }), factory, this.tickCounter % ticksPerSecond, workspaceSoftShutdownTicks, @@ -1077,14 +1088,17 @@ export class TSessionManager implements SessionManager { () => Date.now() ) - handleRequest( + async handleRequest( requestCtx: MeasureContext, service: S, ws: ConnectionSocket, request: Request, workspaceId: WorkspaceUuid ): Promise { - const userCtx = requestCtx.newChild('📞 client', {}) + const userCtx = requestCtx.newChild('📞 client', { + source: service.token.extra?.service ?? '🤦‍♂️user', + mode: '🧭 handleRequest' + }) const rateLimit = this.limitter.checkRateLimit(service.getUser()) // If remaining is 0, rate limit is exceeded if (rateLimit?.remaining === 0) { @@ -1098,104 +1112,102 @@ export class TSessionManager implements SessionManager { service.binaryMode, service.useCompression ) - return Promise.resolve() + return } // Calculate total number of clients const reqId = generateId() const st = Date.now() - return userCtx - .with('🧭 handleRequest', {}, async (ctx) => { - if (request.time != null) { - const delta = Date.now() - request.time - requestCtx.measure('msg-receive-delta', delta) - } - const workspace = this.workspaces.get(workspaceId) - if (workspace === undefined || workspace.closing !== undefined) { - await ws.send( - ctx, - { - id: request.id, - error: unknownError('Workspace is closing') - }, - service.binaryMode, - service.useCompression - ) - return - } - if (request.id === -1 && request.method === 'hello') { - await this.handleHello(request, service, ctx, workspace, ws, requestCtx) - return - } - if (request.id === -2 && request.method === 'forceClose') { - // TODO: we chould allow this only for admin or system accounts - let done = false - const wsRef = this.workspaces.get(workspaceId) - if (wsRef?.upgrade ?? false) { - done = true - this.ctx.warn('FORCE CLOSE', { workspace: workspaceId }) - // In case of upgrade, we need to force close workspace not in interval handler - await this.forceClose(workspaceId, ws) - } - const forceCloseResponse: Response = { + try { + if (request.time != null) { + const delta = Date.now() - request.time + requestCtx.measure('msg-receive-delta', delta) + } + const workspace = this.workspaces.get(workspaceId) + if (workspace === undefined || workspace.closing !== undefined) { + await ws.send( + userCtx, + { id: request.id, - result: done - } - await ws.send(ctx, forceCloseResponse, service.binaryMode, service.useCompression) - return + error: unknownError('Workspace is closing') + }, + service.binaryMode, + service.useCompression + ) + return + } + if (request.id === -1 && request.method === 'hello') { + await this.handleHello(request, service, userCtx, workspace, ws, requestCtx) + return + } + if (request.id === -2 && request.method === 'forceClose') { + // TODO: we chould allow this only for admin or system accounts + let done = false + const wsRef = this.workspaces.get(workspaceId) + if (wsRef?.upgrade ?? false) { + done = true + this.ctx.warn('FORCE CLOSE', { workspace: workspaceId }) + // In case of upgrade, we need to force close workspace not in interval handler + await this.forceClose(workspaceId, ws) + } + const forceCloseResponse: Response = { + id: request.id, + result: done + } + await ws.send(userCtx, forceCloseResponse, service.binaryMode, service.useCompression) + return + } + + service.requests.set(reqId, { + id: reqId, + params: request, + start: st + }) + if (request.id === -1 && request.method === '#upgrade') { + ws.close() + return + } + + const f = (service as any)[request.method] + try { + const params = [...request.params] + + if (ws.isBackpressure()) { + await ws.backpressure(userCtx) } - service.requests.set(reqId, { - id: reqId, - params: request, - start: st - }) - if (request.id === -1 && request.method === '#upgrade') { - ws.close() - return - } - - const f = (service as any)[request.method] - try { - const params = [...request.params] - - if (ws.isBackpressure()) { - await ws.backpressure(ctx) - } - - await workspace.with(async (pipeline, communicationApi) => { - await ctx.with('🧨 process', {}, (callTx) => - f.apply(service, [ - this.createOpContext(callTx, userCtx, pipeline, communicationApi, request.id, service, ws, rateLimit), - ...params - ]) - ) - }) - } catch (err: any) { - Analytics.handleError(err) - if (LOGGING_ENABLED) { - this.ctx.error('error handle request', { error: err, request }) - } - await ws.send( - userCtx, - { - id: request.id, - error: unknownError(err), - result: JSON.parse(JSON.stringify(err?.stack)) - }, - service.binaryMode, - service.useCompression + await workspace.with(async (pipeline, communicationApi) => { + await userCtx.with('🧨 process', {}, (callTx) => + f.apply(service, [ + this.createOpContext(callTx, userCtx, pipeline, communicationApi, request.id, service, ws, rateLimit), + ...params + ]) ) + }) + } catch (err: any) { + Analytics.handleError(err) + if (LOGGING_ENABLED) { + this.ctx.error('error handle request', { error: err, request }) } - }) - .finally(() => { - userCtx.end() - service.requests.delete(reqId) - }) + await ws.send( + userCtx, + { + id: request.id, + error: unknownError(err), + result: JSON.parse(JSON.stringify(err?.stack)) + }, + service.binaryMode, + service.useCompression + ) + } + } finally { + userCtx.end() + service.requests.delete(reqId) + } } - handleRPC( + async handleRPC( requestCtx: MeasureContext, service: S, ws: ConnectionSocket, @@ -1204,65 +1216,66 @@ export class TSessionManager implements SessionManager { const rateLimitStatus = this.limitter.checkRateLimit(service.getUser()) // If remaining is 0, rate limit is exceeded if (rateLimitStatus?.remaining === 0) { - return Promise.resolve(rateLimitStatus) + return await Promise.resolve(rateLimitStatus) } - const userCtx = requestCtx.newChild('📞 client', {}) + const userCtx = requestCtx.newChild('📞 client', { + source: service.token.extra?.service ?? '🤦‍♂️user', + mode: '🧭 handleRPC' + }) // Calculate total number of clients const reqId = generateId() const st = Date.now() - return userCtx - .with('🧭 handleRPC', {}, async (ctx) => { - const workspace = this.workspaces.get(service.workspace.uuid) - if (workspace === undefined || workspace.closing !== undefined) { - throw new Error('Workspace is closing') - } + try { + const workspace = this.workspaces.get(service.workspace.uuid) + if (workspace === undefined || workspace.closing !== undefined) { + throw new Error('Workspace is closing') + } - service.requests.set(reqId, { - id: reqId, - params: {}, - start: st - }) + service.requests.set(reqId, { + id: reqId, + params: {}, + start: st + }) - try { - await workspace.with(async (pipeline, communicationApi) => { - const uctx = this.createOpContext( - ctx, - userCtx, - pipeline, - communicationApi, - reqId, - service, - ws, - rateLimitStatus - ) - await operation(uctx) - }) - } catch (err: any) { - Analytics.handleError(err) - if (LOGGING_ENABLED) { - this.ctx.error('error handle request', { error: err }) - } - await ws.send( + try { + await workspace.with(async (pipeline, communicationApi) => { + const uctx = this.createOpContext( userCtx, - { - id: reqId, - error: unknownError(err), - result: JSON.parse(JSON.stringify(err?.stack)) - }, - service.binaryMode, - service.useCompression + userCtx, + pipeline, + communicationApi, + reqId, + service, + ws, + rateLimitStatus ) - throw err + await operation(uctx) + }) + } catch (err: any) { + Analytics.handleError(err) + if (LOGGING_ENABLED) { + this.ctx.error('error handle request', { error: err }) } - return undefined - }) - .finally(() => { - userCtx.end() - service.requests.delete(reqId) - }) + await ws.send( + userCtx, + { + id: reqId, + error: unknownError(err), + result: JSON.parse(JSON.stringify(err?.stack)) + }, + service.binaryMode, + service.useCompression + ) + throw err + } + return undefined + } finally { + userCtx.end() + service.requests.delete(reqId) + } } entryToUserStats = (session: Session, socket: ConnectionSocket): UserStatistics => { diff --git a/server/workspace-service/src/ws-operations.ts b/server/workspace-service/src/ws-operations.ts index e560bee2a0..2194254d38 100644 --- a/server/workspace-service/src/ws-operations.ts +++ b/server/workspace-service/src/ws-operations.ts @@ -322,7 +322,8 @@ export async function upgradeWorkspaceWith ( undefined, undefined, pipeline.context.modelDb, - new Map() + new Map(), + 'workspace' ) ctx.contextData = contextData await handleWsEvent?.('upgrade-started', version, 0) diff --git a/services/ai-bot/pod-ai-bot/src/start.ts b/services/ai-bot/pod-ai-bot/src/start.ts index 64dddd1629..71ed6c24ea 100644 --- a/services/ai-bot/pod-ai-bot/src/start.ts +++ b/services/ai-bot/pod-ai-bot/src/start.ts @@ -63,7 +63,10 @@ export const start = async (): Promise => { ctx.info('AI person uuid', { personUuid }) const storage = await getDbStorage() - const socialIds: SocialId[] = await getAccountClient(config.AccountsURL, generateToken(personUuid)).getSocialIds() + const socialIds: SocialId[] = await getAccountClient( + config.AccountsURL, + generateToken(personUuid, undefined, { service: 'aibot' }) + ).getSocialIds() const aiControl = new AIControl(personUuid, socialIds, storage, ctx) diff --git a/services/analytics-collector/pod-analytics-collector/src/collector.ts b/services/analytics-collector/pod-analytics-collector/src/collector.ts index 7701dd43d5..8f58f42da5 100644 --- a/services/analytics-collector/pod-analytics-collector/src/collector.ts +++ b/services/analytics-collector/pod-analytics-collector/src/collector.ts @@ -133,7 +133,7 @@ export class Collector { return true } - const rawToken = generateToken(token.account, token.workspace, token.extra) + const rawToken = generateToken(token.account, token.workspace, { ...token.extra, service: 'analytics-collector' }) const wsInfo = await getAccountClient(config.AccountsUrl, rawToken).getWorkspaceInfo() this.ctx.info('workspace info', wsInfo) diff --git a/services/calendar/pod-calendar-mailer/src/utils.ts b/services/calendar/pod-calendar-mailer/src/utils.ts index 7981084beb..bf9065ea4e 100644 --- a/services/calendar/pod-calendar-mailer/src/utils.ts +++ b/services/calendar/pod-calendar-mailer/src/utils.ts @@ -24,7 +24,7 @@ export async function getClient ( workspaceUuid: WorkspaceUuid, socialId?: PersonId ): Promise<{ client: TxOperations, accountClient: AccountClient }> { - const token = generateToken(systemAccountUuid, workspaceUuid) + const token = generateToken(systemAccountUuid, workspaceUuid, { service: 'calendar-mailer' }) let accountClient = getAccountClient(config.accountsUrl, token) if (socialId !== undefined && socialId !== core.account.System) { @@ -32,7 +32,7 @@ export async function getClient ( if (personUuid === undefined) { throw new Error('Global person not found') } - const token = generateToken(personUuid, workspaceUuid) + const token = generateToken(personUuid, workspaceUuid, { service: 'calendar-mailer' }) accountClient = getAccountClient(config.accountsUrl, token) } diff --git a/services/calendar/pod-calendar/src/auth.ts b/services/calendar/pod-calendar/src/auth.ts index 7157cb54c0..3e99cdf332 100644 --- a/services/calendar/pod-calendar/src/auth.ts +++ b/services/calendar/pod-calendar/src/auth.ts @@ -55,17 +55,22 @@ export class AuthController { state: State, code: string ): Promise { - await ctx.with('Create auth controller', { workspace: state.workspace, user: state.userId }, async () => { - const mutex = await lock(`${state.workspace}:${state.userId}`) - try { - const client = await getClient(getWorkspaceToken(state.workspace)) - const txOp = new TxOperations(client, core.account.System) - const controller = new AuthController(ctx, accountClient, txOp, state) - await controller.process(code) - } finally { - mutex() - } - }) + await ctx.with( + 'Create auth controller', + {}, + async () => { + const mutex = await lock(`${state.workspace}:${state.userId}`) + try { + const client = await getClient(getWorkspaceToken(state.workspace)) + const txOp = new TxOperations(client, core.account.System) + const controller = new AuthController(ctx, accountClient, txOp, state) + await controller.process(code) + } finally { + mutex() + } + }, + { workspace: state.workspace, user: state.userId } + ) } static async signout ( @@ -75,22 +80,27 @@ export class AuthController { workspace: WorkspaceUuid, value: GoogleEmail ): Promise { - await ctx.with('Signout auth controller', { workspace, userId }, async () => { - const mutex = await lock(`${workspace}:${userId}`) - try { - const client = await getClient(getWorkspaceToken(workspace)) - const txOp = new TxOperations(client, core.account.System) - const controller = new AuthController(ctx, accountClient, txOp, { - userId, - workspace - }) - await controller.signout(value) - } catch (err) { - ctx.error('signout', { workspace, userId, err }) - } finally { - mutex() - } - }) + await ctx.with( + 'Signout auth controller', + {}, + async () => { + const mutex = await lock(`${workspace}:${userId}`) + try { + const client = await getClient(getWorkspaceToken(workspace)) + const txOp = new TxOperations(client, core.account.System) + const controller = new AuthController(ctx, accountClient, txOp, { + userId, + workspace + }) + await controller.signout(value) + } catch (err) { + ctx.error('signout', { workspace, userId, err }) + } finally { + mutex() + } + }, + { workspace, userId } + ) } private async signout (value: GoogleEmail): Promise { @@ -167,7 +177,7 @@ export class AuthController { private async setWorkspaceIntegration (res: AuthResult): Promise { await this.ctx.with( 'Set workspace integration', - { user: this.user.userId, workspace: this.user.workspace, email: res.email }, + {}, async () => { const integrations = await this.client.findAll(setting.class.Integration, { createdBy: this.user.userId, @@ -207,26 +217,36 @@ export class AuthController { }) } } + }, + { + user: this.user.userId, + workspace: this.user.workspace, + email: res.email } ) } private async createAccIntegrationIfNotExists (): Promise { - await this.ctx.with('Create account integration if not exists', { user: this.user.userId }, async () => { - const integration = await this.accountClient.getIntegration({ - socialId: this.user.userId, - kind: CALENDAR_INTEGRATION, - workspaceUuid: this.user.workspace - }) - if (integration != null) { - return - } - await this.accountClient.createIntegration({ - socialId: this.user.userId, - kind: CALENDAR_INTEGRATION, - workspaceUuid: this.user.workspace - }) - }) + await this.ctx.with( + 'Create account integration if not exists', + {}, + async () => { + const integration = await this.accountClient.getIntegration({ + socialId: this.user.userId, + kind: CALENDAR_INTEGRATION, + workspaceUuid: this.user.workspace + }) + if (integration != null) { + return + } + await this.accountClient.createIntegration({ + socialId: this.user.userId, + kind: CALENDAR_INTEGRATION, + workspaceUuid: this.user.workspace + }) + }, + { user: this.user.userId } + ) } private async updateToken (token: Credentials, email: GoogleEmail): Promise { diff --git a/services/calendar/pod-calendar/src/pushHandler.ts b/services/calendar/pod-calendar/src/pushHandler.ts index bca596c1cd..2b28c4bd99 100644 --- a/services/calendar/pod-calendar/src/pushHandler.ts +++ b/services/calendar/pod-calendar/src/pushHandler.ts @@ -28,14 +28,19 @@ export class PushHandler { ) {} async sync (token: Token, calendarId: string | null): Promise { - await this.ctx.with('Push handler', { workspace: token.workspace, user: token.userId }, async () => { - const client = await getClient(getWorkspaceToken(token.workspace)) - const txOp = new TxOperations(client, core.account.System) - const res = getGoogleClient() - res.auth.setCredentials(token) - await IncomingSyncManager.push(this.ctx, this.accountClient, txOp, token, res.google, calendarId) - await txOp.close() - }) + await this.ctx.with( + 'Push handler', + {}, + async () => { + const client = await getClient(getWorkspaceToken(token.workspace)) + const txOp = new TxOperations(client, core.account.System) + const res = getGoogleClient() + res.auth.setCredentials(token) + await IncomingSyncManager.push(this.ctx, this.accountClient, txOp, token, res.google, calendarId) + await txOp.close() + }, + { workspace: token.workspace, user: token.userId } + ) } async push (email: GoogleEmail, mode: 'events' | 'calendar', calendarId?: string): Promise { diff --git a/services/github/pod-github/src/platform.ts b/services/github/pod-github/src/platform.ts index 5aa3b3ccbd..c0168ee770 100644 --- a/services/github/pod-github/src/platform.ts +++ b/services/github/pod-github/src/platform.ts @@ -282,16 +282,21 @@ export class PlatformWorker { } ctx.info('add integration', { workspace, installationId, accountId }) - await ctx.with('add integration', { workspace, installationId, accountId }, async (ctx) => { - await accountsClient.createIntegration({ - kind: 'github', - workspaceUuid: record.workspace, - socialId: record.accountId, - data: { installationId: record.installationId } - }) + await ctx.with( + 'add integration', + {}, + async (ctx) => { + await accountsClient.createIntegration({ + kind: 'github', + workspaceUuid: record.workspace, + socialId: record.accountId, + data: { installationId: record.installationId } + }) - this.integrations.push(record) - }) + this.integrations.push(record) + }, + { workspace, installationId, accountId } + ) // We need to query installations to be sure we have it, in case event is delayed or not received. await this.updateInstallation(installationId) @@ -491,14 +496,16 @@ export class PlatformWorker { if (!revoke) { const personSpace = await client.findOne(contact.class.PersonSpace, { person: person._id }) if (personSpace !== undefined && person.personUuid !== undefined) { - await createNotification(client, person, { - user: person.personUuid, - space: personSpace._id, - message: github.string.AuthenticatedWithGithub, - props: { - login: update.login - } - }) + if (update.login != null) { + await createNotification(client, person, { + user: person.personUuid, + space: personSpace._id, + message: github.string.AuthenticatedWithGithub, + props: { + login: update.login + } + }) + } } if (dta?._id !== undefined) { diff --git a/services/github/pod-github/src/sync/issues.ts b/services/github/pod-github/src/sync/issues.ts index d38c20f643..acda90aaa4 100644 --- a/services/github/pod-github/src/sync/issues.ts +++ b/services/github/pod-github/src/sync/issues.ts @@ -887,7 +887,7 @@ export class IssueSyncManager extends IssueSyncManagerBase implements DocSyncMan try { const response: any = await this.ctx.with( 'graphql.listIssue', - { prj: prj.name, repo: repo.name }, + {}, () => integration.octokit.graphql( `query listIssues { diff --git a/services/github/pod-github/src/worker.ts b/services/github/pod-github/src/worker.ts index adbe44e492..23699c68c6 100644 --- a/services/github/pod-github/src/worker.ts +++ b/services/github/pod-github/src/worker.ts @@ -1557,7 +1557,7 @@ export class GithubWorker implements IntegrationManager { } await this.ctx.withLog( 'external sync', - { installation: integration.installationName, workspace: this.workspace.uuid }, + {}, async () => { const enabled = integration.enabled && integration.octokit !== undefined @@ -1652,13 +1652,15 @@ export class GithubWorker implements IntegrationManager { } await this.ctx.withLog( 'external sync', - { _class: _class.join(', '), workspace: this.workspace.uuid }, + { _class: _class.join(', ') }, async () => { await mapper.externalFullSync(integration, derivedClient, _projects, _repositories) - } + }, + { installation: integration.installationName, workspace: this.workspace.uuid } ) } - } + }, + { installation: integration.installationName, workspace: this.workspace.uuid } ) } } diff --git a/services/gmail/pod-gmail/src/accounts.ts b/services/gmail/pod-gmail/src/accounts.ts index 9f1c5f9650..2323691cc9 100644 --- a/services/gmail/pod-gmail/src/accounts.ts +++ b/services/gmail/pod-gmail/src/accounts.ts @@ -21,7 +21,7 @@ import { serviceToken } from './utils' export async function getAccountPerson (account: AccountUuid): Promise { try { - const accountClient = getAccountClient(generateToken(account)) + const accountClient = getAccountClient(generateToken(account, undefined, { service: 'gmail' })) return await accountClient.getPerson() } catch (e) { console.error(e) @@ -31,7 +31,7 @@ export async function getAccountPerson (account: AccountUuid): Promise { try { - const accountClient = getAccountClient(generateToken(account)) + const accountClient = getAccountClient(generateToken(account, undefined, { service: 'gmail' })) return await accountClient.getSocialIds() } catch (e) { console.error(e) diff --git a/services/love/src/billing.ts b/services/love/src/billing.ts index 2c308f452f..bc2fe3bbe5 100644 --- a/services/love/src/billing.ts +++ b/services/love/src/billing.ts @@ -77,7 +77,7 @@ export async function saveLiveKitSessionBilling (ctx: MeasureContext, sessionId: const workspace = session.roomName.split('_')[0] as WorkspaceUuid const endpoint = concatLink(config.BillingUrl, `/api/v1/billing/${workspace}/livekit/session`) - const token = generateToken(systemAccountUuid, workspace) + const token = generateToken(systemAccountUuid, workspace, { service: 'love' }) try { const res = await fetch(endpoint, { @@ -116,7 +116,7 @@ export async function saveLiveKitEgressBilling (ctx: MeasureContext, egress: Egr const workspace = egress.roomName.split('_')[0] as WorkspaceUuid const endpoint = concatLink(config.BillingUrl, `/api/v1/billing/${workspace}/livekit/egress`) - const token = generateToken(systemAccountUuid, workspace) + const token = generateToken(systemAccountUuid, workspace, { service: 'love' }) try { const res = await fetch(endpoint, { diff --git a/services/mail/pod-inbound-mail/src/client.ts b/services/mail/pod-inbound-mail/src/client.ts index 775e5406bf..7745f6d6ef 100644 --- a/services/mail/pod-inbound-mail/src/client.ts +++ b/services/mail/pod-inbound-mail/src/client.ts @@ -21,7 +21,7 @@ import { getClient } from '@hcengineering/kvs-client' import config from './config' // TODO: Find account UUID from mailboxes and use personal workspace -export const mailServiceToken = generateToken(systemAccountUuid, undefined, { service: 'mail' }, config.secret) +export const mailServiceToken = generateToken(systemAccountUuid, undefined, { service: 'inbound-mail' }, config.secret) export const baseConfig: BaseConfig = { AccountsURL: config.accountsUrl, KvsUrl: config.kvsUrl, diff --git a/services/msg2file/src/worker.ts b/services/msg2file/src/worker.ts index b62658a078..5488052402 100644 --- a/services/msg2file/src/worker.ts +++ b/services/msg2file/src/worker.ts @@ -61,13 +61,14 @@ export async function job (ctx: MeasureContext, storage: StorageAdapter, db: Pos await rateLimiter.add(async () => { await ctx.with( 'process', + {}, + async () => { + await processRecord(ctx, record, db, storage) + }, { workspace: record.workspace, card: record.card, attempt: record.attempt - }, - async () => { - await processRecord(ctx, record, db, storage) } ) }) diff --git a/services/telegram-bot/pod-telegram-bot/src/account.ts b/services/telegram-bot/pod-telegram-bot/src/account.ts index e7aabb6313..5c1f4298dd 100644 --- a/services/telegram-bot/pod-telegram-bot/src/account.ts +++ b/services/telegram-bot/pod-telegram-bot/src/account.ts @@ -30,7 +30,7 @@ import { IntegrationInfo } from './types' export async function getAccountPerson (account: AccountUuid): Promise { try { - const accountClient = getAccountClient(generateToken(account)) + const accountClient = getAccountClient(generateToken(account, undefined, { service: 'telegram-bot' })) return await accountClient.getPerson() } catch (e) { console.error(e) @@ -40,7 +40,7 @@ export async function getAccountPerson (account: AccountUuid): Promise { try { - const accountClient = getAccountClient(generateToken(account)) + const accountClient = getAccountClient(generateToken(account, undefined, { service: 'telegram-bot' })) return await accountClient.getSocialIds() } catch (e) { console.error(e) @@ -49,7 +49,7 @@ export async function getAccountSocialIds (account: AccountUuid): Promise { - const client = getAccountClient(generateToken(account)) + const client = getAccountClient(generateToken(account, undefined, { serviece: 'telegram-bot' })) const integrations = await client.listIntegrations({ kind: 'telegram-bot' }) if (integrations.length === 0) return [] const socialIds = await getAccountSocialIds(account) @@ -118,7 +118,7 @@ export async function getAnyIntegrationByAccount ( account: AccountUuid, workspace?: WorkspaceUuid ): Promise { - const client = getAccountClient(generateToken(account)) + const client = getAccountClient(generateToken(account, undefined, { service: 'telegram-bot' })) const integrations = await client.listIntegrations({ kind: 'telegram-bot', workspaceUuid: workspace }) if (integrations.length === 0) return undefined diff --git a/services/telegram-bot/pod-telegram-bot/src/worker.ts b/services/telegram-bot/pod-telegram-bot/src/worker.ts index d9b869ebd9..c56da7b511 100644 --- a/services/telegram-bot/pod-telegram-bot/src/worker.ts +++ b/services/telegram-bot/pod-telegram-bot/src/worker.ts @@ -292,7 +292,7 @@ export class PlatformWorker { } try { - const accountClient = getAccountClient(generateToken(account, workspaceId)) + const accountClient = getAccountClient(generateToken(account, workspaceId, { service: 'telegram-bot' })) const result = await accountClient.getWorkspaceInfo(false) if (result === undefined) { diff --git a/services/telegram-bot/pod-telegram-bot/src/workspace.ts b/services/telegram-bot/pod-telegram-bot/src/workspace.ts index 8e3682edbd..d7893f7454 100644 --- a/services/telegram-bot/pod-telegram-bot/src/workspace.ts +++ b/services/telegram-bot/pod-telegram-bot/src/workspace.ts @@ -55,7 +55,7 @@ export class WorkspaceClient { ctx: MeasureContext, storage: StorageAdapter ): Promise { - const token = generateToken(account, workspace) + const token = generateToken(account, workspace, { service: 'telegram-bot' }) const endpoint = await getTransactorEndpoint(token) const client = createRestClient(endpoint, workspace, token) const model = await client.getModel() diff --git a/ws-tests/docker-compose.yaml b/ws-tests/docker-compose.yaml index 936c22ccc8..ac4098aeff 100644 --- a/ws-tests/docker-compose.yaml +++ b/ws-tests/docker-compose.yaml @@ -400,4 +400,4 @@ services: - PASSWORD=password - AVATAR_PATH=./avatar.png - AVATAR_CONTENT_TYPE=.png - - STATS_URL=http://huly.local:4900 + - STATS_URL=http://huly.local:4901