diff --git a/services/gmail/pod-gmail/src/__mocks__/config.ts b/services/gmail/pod-gmail/src/__mocks__/config.ts index fbc6ce1c62..75bdcf4a5f 100644 --- a/services/gmail/pod-gmail/src/__mocks__/config.ts +++ b/services/gmail/pod-gmail/src/__mocks__/config.ts @@ -6,7 +6,8 @@ const config = { Credentials: 'test-credentials', WATCH_TOPIC_NAME: 'test-watch-topic', FooterMessage: '

Sent via Huly

', - InitLimit: 50 + InitLimit: 50, + WorkspaceInactivityInterval: 3 } export default config diff --git a/services/gmail/pod-gmail/src/__tests__/gmailController.test.ts b/services/gmail/pod-gmail/src/__tests__/gmailController.test.ts index 49355dad89..bfe9c892e7 100644 --- a/services/gmail/pod-gmail/src/__tests__/gmailController.test.ts +++ b/services/gmail/pod-gmail/src/__tests__/gmailController.test.ts @@ -43,10 +43,12 @@ describe('GmailController', () => { const workspaceAId: WorkspaceUuid = 'workspace-a' as any const workspaceBId: WorkspaceUuid = 'workspace-b' as any + const inactiveWorkspaceId: WorkspaceUuid = 'inactive-workspace' as any + const outdatedWorkspaceId: WorkspaceUuid = 'outdated-workspace' as any const workspaceATokens: Token[] = [ - { userId: 'user1', workspace: workspaceAId, token: 'token1' } as any, - { userId: 'user2', workspace: workspaceAId, token: 'token2' } as any + { uuid: workspaceAId, userId: 'user1', workspace: workspaceAId, token: 'token1' } as any, + { uuid: workspaceBId, userId: 'user2', workspace: workspaceAId, token: 'token2' } as any ] const workspaceBTokens: Token[] = [ @@ -55,6 +57,18 @@ describe('GmailController', () => { { userId: 'user5', workspace: workspaceBId, token: 'token5' } as any ] + const inactiveWorkspaceTokens: Token[] = [ + { userId: 'user3', workspace: workspaceBId, token: 'token6' } as any, + { userId: 'user4', workspace: workspaceBId, token: 'token7' } as any, + { userId: 'user5', workspace: workspaceBId, token: 'token8' } as any + ] + + const outdatedWorkspaceTokens: Token[] = [ + { userId: 'user3', workspace: workspaceBId, token: 'token6' } as any, + { userId: 'user4', workspace: workspaceBId, token: 'token7' } as any, + { userId: 'user5', workspace: workspaceBId, token: 'token8' } as any + ] + beforeEach(() => { jest.clearAllMocks() @@ -82,7 +96,12 @@ describe('GmailController', () => { // Create mock clients with unique properties mockGmailClients = new Map() - const allUsers = [...workspaceATokens, ...workspaceBTokens].map((token) => token.userId) + const allUsers = [ + ...workspaceATokens, + ...workspaceBTokens, + ...inactiveWorkspaceTokens, + ...outdatedWorkspaceTokens + ].map((token) => token.userId) allUsers.forEach((userId) => { mockGmailClients.set(userId, { startSync: jest.fn().mockResolvedValue(undefined), @@ -108,14 +127,40 @@ describe('GmailController', () => { // Mock getWorkspaceTokens jest.spyOn(tokens, 'getWorkspaceTokens').mockImplementation(async (_, workspaceId) => { - if (workspaceId === workspaceAId) return workspaceATokens - if (workspaceId === workspaceBId) return workspaceBTokens - return [] + const result = new Map() + if (workspaceId === workspaceAId || workspaceId === undefined) result.set(workspaceAId, workspaceATokens) + if (workspaceId === workspaceBId || workspaceId === undefined) result.set(workspaceBId, workspaceBTokens) + if (workspaceId === inactiveWorkspaceId || workspaceId === undefined) { + result.set(inactiveWorkspaceId, inactiveWorkspaceTokens) + } + if (workspaceId === outdatedWorkspaceId || workspaceId === undefined) { + result.set(outdatedWorkspaceId, outdatedWorkspaceTokens) + } + + return result }) // Mock getAccountClient jest.spyOn(serverClient, 'getAccountClient').mockReturnValue({ - getWorkspaceInfo: jest.fn().mockResolvedValue({ mode: 'active' }) + getWorkspaceInfo: jest.fn().mockResolvedValue({ mode: 'active' }), + getWorkspacesInfo: jest.fn().mockResolvedValue([ + { uuid: workspaceAId, workspaceUuid: workspaceAId, name: 'Workspace A', mode: 'active', lastVisit: Date.now() }, + { uuid: workspaceBId, workspaceUuid: workspaceBId, name: 'Workspace B', mode: 'active', lastVisit: Date.now() }, + { + uuid: inactiveWorkspaceId, + workspaceUuid: inactiveWorkspaceId, + name: 'Inactive Workspace', + mode: 'archived', + lastVisit: Date.now() + }, + { + uuid: outdatedWorkspaceId, + workspaceUuid: outdatedWorkspaceId, + name: 'Outdated Workspace', + mode: 'active', + lastVisit: Date.now() - 7 * 3600 * 24 * 1000 + } + ]) } as any) // Mock serviceToken diff --git a/services/gmail/pod-gmail/src/config.ts b/services/gmail/pod-gmail/src/config.ts index 9f4f717424..5af4c24074 100644 --- a/services/gmail/pod-gmail/src/config.ts +++ b/services/gmail/pod-gmail/src/config.ts @@ -31,6 +31,8 @@ interface Config extends BaseConfig { QueueConfig: string QueueRegion: string CommunicationTopic: string + + WorkspaceInactivityInterval: number // Interval in days to stop workspace synchronization if not visited } const envMap: { [key in keyof Config]: string } = { @@ -47,7 +49,8 @@ const envMap: { [key in keyof Config]: string } = { Version: 'VERSION', QueueConfig: 'QUEUE_CONFIG', QueueRegion: 'QUEUE_REGION', - CommunicationTopic: 'COMMUNICATION_TOPIC' + CommunicationTopic: 'COMMUNICATION_TOPIC', + WorkspaceInactivityInterval: 'WORKSPACE_INACTIVITY_INTERVAL' } const parseNumber = (str: string | undefined): number | undefined => (str !== undefined ? Number(str) : undefined) @@ -74,7 +77,8 @@ const config: Config = (() => { Version: version, QueueConfig: process.env[envMap.QueueConfig] ?? '', QueueRegion: process.env[envMap.QueueRegion] ?? '', - CommunicationTopic: process.env[envMap.CommunicationTopic] ?? 'hulygun' + CommunicationTopic: process.env[envMap.CommunicationTopic] ?? 'hulygun', + WorkspaceInactivityInterval: parseNumber(process.env[envMap.WorkspaceInactivityInterval] ?? '3') // In days } const missingEnv = (Object.keys(params) as Array) diff --git a/services/gmail/pod-gmail/src/gmailController.ts b/services/gmail/pod-gmail/src/gmailController.ts index 09404e31b0..de896bab97 100644 --- a/services/gmail/pod-gmail/src/gmailController.ts +++ b/services/gmail/pod-gmail/src/gmailController.ts @@ -16,25 +16,29 @@ import { AccountUuid, isActiveMode, + isArchivingMode, + isDeletingMode, MeasureContext, RateLimiter, + WorkspaceInfoWithStatus, WorkspaceUuid, type PersonId } from '@hcengineering/core' -import type { StorageAdapter } from '@hcengineering/server-core' import { normalizeEmail } from '@hcengineering/mail-common' +import type { StorageAdapter } from '@hcengineering/server-core' +import { getAccountClient } from '@hcengineering/server-client' import { decode64 } from './base64' import config from './config' import { type GmailClient } from './gmail' -import { type ProjectCredentials, type Token, type User } from './types' -import { WorkspaceClient } from './workspaceClient' -import { getAccountClient } from '@hcengineering/server-client' import { getIntegrations } from './integrations' -import { serviceToken } from './utils' import { getWorkspaceTokens } from './tokens' +import { type ProjectCredentials, type Token, type User } from './types' +import { serviceToken } from './utils' +import { WorkspaceClient } from './workspaceClient' import { AuthProvider } from './gmail/auth' +import { AccountClient } from '@hcengineering/account-client' export class GmailController { private readonly workspaces: Map = new Map() @@ -78,10 +82,10 @@ export class GmailController { async startAll (): Promise { try { const token = serviceToken() - const integrations = await getIntegrations(token) + const sysClient = getAccountClient(token) + const integrations = await getIntegrations(sysClient, token) this.ctx.info('Start integrations', { count: integrations.length }) - const limiter = new RateLimiter(config.InitLimit) const workspaceIds = new Set( integrations .map((integration) => { @@ -93,45 +97,98 @@ export class GmailController { }) .filter((id): id is WorkspaceUuid => id != null) ) - this.ctx.info('Workspaces with integrations', { count: workspaceIds.size }) - - for (const workspace of workspaceIds) { - try { - const wsToken = serviceToken(workspace) - const accountClient = getAccountClient(wsToken) - - const tokens = await getWorkspaceTokens(accountClient, workspace) - await limiter.add(async () => { - const info = await accountClient.getWorkspaceInfo() - - if (info === undefined) { - this.ctx.info('workspace not found', { workspaceUuid: workspace }) - return - } - if (!isActiveMode(info.mode)) { - this.ctx.info('workspace is not active', { workspaceUuid: workspace }) - return - } - this.ctx.info('Use stored tokens', { count: tokens.length }) - const startPromise = this.startWorkspace(workspace, tokens) - const timeoutPromise = new Promise((resolve) => { - setTimeout(() => { - resolve() - }, 60000) - }) - await Promise.race([startPromise, timeoutPromise]) - }) - } catch (err: any) { - this.ctx.error('Failed to create workspace client', { workspaceUuid: workspace, error: err.message }) - } - } - - await limiter.waitProcessing() + const pendingWorkspaces = await this.startWorkspaces(workspaceIds, sysClient) + this.ctx.info('Pending workspaces', { count: pendingWorkspaces.size }) + // Start async check for pending workspaces + void this.checkPendingWorkspaces(pendingWorkspaces, sysClient) } catch (err: any) { this.ctx.error('Failed to start existing integrations', { error: err.message }) } } + async checkPendingWorkspaces (workspaceIds: Set, sysClient: AccountClient): Promise { + try { + let unprocessedWorkspaces = new Set(workspaceIds) + while (unprocessedWorkspaces.size > 0) { + unprocessedWorkspaces = await this.startWorkspaces(unprocessedWorkspaces, sysClient) + if (unprocessedWorkspaces.size > 0) { + this.ctx.info('Waiting for pending workspaces', { count: unprocessedWorkspaces.size }) + await new Promise((resolve) => { + setTimeout( + () => { + resolve() + }, + 5 * 60 * 1000 + ) // Wait 5 minutes + }) + } + } + } catch (err: any) { + this.ctx.error('Failed to check pending workspaces', { error: err.message, workspaceIds }) + } + } + + async startWorkspaces (workspaceIds: Set, sysClient: AccountClient): Promise> { + const unprocessedWorkspaces = new Set(workspaceIds) + const limiter = new RateLimiter(config.InitLimit) + this.ctx.info('Workspaces with integrations', { count: unprocessedWorkspaces.size }) + + const workspaceWithInfo = await sysClient.getWorkspacesInfo(Array.from(unprocessedWorkspaces)) + + const allTokens = await getWorkspaceTokens(sysClient) + + for (const info of workspaceWithInfo) { + const workspace = info.uuid + try { + const { needSync, needRecheck } = this.checkWorkspace(info) + if (!needSync) { + if (!needRecheck) unprocessedWorkspaces.delete(workspace) + continue + } + + // So we will not start it one more time. + unprocessedWorkspaces.delete(workspace) + + const tokens = allTokens.get(workspace) ?? [] + await limiter.add(async () => { + this.ctx.info('Use stored tokens', { count: tokens.length }) + const startPromise = this.startWorkspace(workspace, tokens) + const timeoutPromise = new Promise((resolve) => { + setTimeout(() => { + // Not connected, recheck again. + unprocessedWorkspaces.add(workspace) + resolve() + }, 60000) + }) + await Promise.race([startPromise, timeoutPromise]) + }) + } catch (err: any) { + this.ctx.error('Failed to create workspace client', { workspaceUuid: workspace, error: err.message }) + } + } + + await limiter.waitProcessing() + return unprocessedWorkspaces + } + + checkWorkspace (info: WorkspaceInfoWithStatus): { needSync: boolean, needRecheck: boolean } { + if (isArchivingMode(info.mode) || isDeletingMode(info.mode)) { + this.ctx.info('workspace is in archiving or deleting mode, skipping', { workspaceUuid: info.uuid }) + return { needSync: false, needRecheck: false } + } + if (!isActiveMode(info.mode)) { + this.ctx.info('workspace is not active, skipping for now.', { workspaceUuid: info.uuid }) + return { needSync: false, needRecheck: true } + } + const lastVisit = (Date.now() - (info.lastVisit ?? 0)) / (3600 * 24 * 1000) // In days + + if (lastVisit > config.WorkspaceInactivityInterval) { + this.ctx.warn('workspace is inactive for too long, skipping for now.', { workspaceUuid: info.uuid }) + return { needSync: false, needRecheck: true } + } + return { needSync: true, needRecheck: false } + } + async startWorkspace (workspace: WorkspaceUuid, tokens: Token[]): Promise { const workspaceClient = await this.getWorkspaceClient(workspace) const clients: GmailClient[] = [] diff --git a/services/gmail/pod-gmail/src/integrations.ts b/services/gmail/pod-gmail/src/integrations.ts index 8dc0b92902..2578a079b0 100644 --- a/services/gmail/pod-gmail/src/integrations.ts +++ b/services/gmail/pod-gmail/src/integrations.ts @@ -13,23 +13,22 @@ // limitations under the License. // +import { Integration, type AccountClient } from '@hcengineering/account-client' import { AccountUuid, MeasureContext, PersonId, TxOperations, WorkspaceUuid } from '@hcengineering/core' -import { getAccountClient } from '@hcengineering/server-client' -import { Integration } from '@hcengineering/account-client' import gmail from '@hcengineering/gmail' +import { getAccountClient } from '@hcengineering/server-client' import setting from '@hcengineering/setting' -import { serviceToken } from './utils' -import { GMAIL_INTEGRATION } from './types' import { getAccountSocialIds } from './accounts' +import { GMAIL_INTEGRATION } from './types' +import { serviceToken } from './utils' export async function getIntegration (socialId: PersonId, workspaceUuid?: WorkspaceUuid): Promise { const client = getAccountClient(serviceToken()) return await client.getIntegration({ kind: GMAIL_INTEGRATION, socialId, workspaceUuid: workspaceUuid ?? null }) } -export async function getIntegrations (token: string): Promise { - const client = getAccountClient(token) +export async function getIntegrations (client: AccountClient, token: string): Promise { return (await client.listIntegrations({ kind: GMAIL_INTEGRATION })) ?? [] } diff --git a/services/gmail/pod-gmail/src/tokens.ts b/services/gmail/pod-gmail/src/tokens.ts index 0120c91ae5..7b5faae98d 100644 --- a/services/gmail/pod-gmail/src/tokens.ts +++ b/services/gmail/pod-gmail/src/tokens.ts @@ -13,8 +13,8 @@ // limitations under the License. // -import { MeasureContext, PersonId, WorkspaceUuid } from '@hcengineering/core' import type { AccountClient } from '@hcengineering/account-client' +import { groupByArray, MeasureContext, PersonId, WorkspaceUuid } from '@hcengineering/core' import { getAccountClient } from '@hcengineering/server-client' import { GMAIL_INTEGRATION, SecretType, Token } from './types' @@ -77,10 +77,29 @@ export class TokenStorage { } } -export async function getWorkspaceTokens (accountClient: AccountClient, workspace: WorkspaceUuid): Promise { - const secrets = await accountClient.listIntegrationsSecrets({ - kind: GMAIL_INTEGRATION, - workspaceUuid: workspace - }) - return secrets.map((secret: { secret: string }) => JSON.parse(secret.secret)) +export async function getWorkspaceTokens ( + accountClient: AccountClient, + workspace?: WorkspaceUuid +): Promise> { + const secrets = ( + await accountClient.listIntegrationsSecrets( + workspace !== undefined + ? { + kind: GMAIL_INTEGRATION, + workspaceUuid: workspace + } + : { kind: GMAIL_INTEGRATION } + ) + ).filter((it) => it.workspaceUuid != null) + + const byWorkspaces = groupByArray(secrets, (it) => it.workspaceUuid as WorkspaceUuid) + + const result = new Map() + for (const entry of byWorkspaces.entries()) { + result.set( + entry[0], + entry[1].map((secret: { secret: string }) => JSON.parse(secret.secret)) + ) + } + return result }