QFIX: Check last visit in gmail service (#9300)

* QFIX: Check last visit in gmail service Signed-off-by: Andrey Sobolev <haiodo@gmail.com> * QFix: Fix workspace loop and tests Signed-off-by: Artem Savchenko <armisav@gmail.com> * QFix: check pending workspaces in background Signed-off-by: Artem Savchenko <armisav@gmail.com> * QFix: Handle errors in checkPendingWorkspaces Signed-off-by: Artem Savchenko <armisav@gmail.com> --------- Signed-off-by: Andrey Sobolev <haiodo@gmail.com> Signed-off-by: Artem Savchenko <armisav@gmail.com> Co-authored-by: Artem Savchenko <armisav@gmail.com>
2026-08-26 14:22:23 +02:00 · 2025-06-20 11:57:40 +07:00
parent c74ce51be8
commit 3fc81cdbdb
6 changed files with 189 additions and 64 deletions
@@ -6,7 +6,8 @@ const config = {
  Credentials: 'test-credentials',
  WATCH_TOPIC_NAME: 'test-watch-topic',
  FooterMessage: '<br><br><p>Sent via <a href="https://huly.io">Huly</a></p>',
-  InitLimit: 50
+  InitLimit: 50,
+  WorkspaceInactivityInterval: 3
 }

 export default config
@@ -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<WorkspaceUuid, Token[]>()
+      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
@@ -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<keyof Config>)
@@ -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<string, WorkspaceClient> = new Map<string, WorkspaceClient>()
@@ -78,10 +82,10 @@ export class GmailController {
  async startAll (): Promise<void> {
    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<WorkspaceUuid>(
        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<void>((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<WorkspaceUuid>, sysClient: AccountClient): Promise<void> {
+    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<void>((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<WorkspaceUuid>, sysClient: AccountClient): Promise<Set<WorkspaceUuid>> {
+    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<void>((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<void> {
    const workspaceClient = await this.getWorkspaceClient(workspace)
    const clients: GmailClient[] = []
@@ -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<Integration | null> {
  const client = getAccountClient(serviceToken())
  return await client.getIntegration({ kind: GMAIL_INTEGRATION, socialId, workspaceUuid: workspaceUuid ?? null })
 }

-export async function getIntegrations (token: string): Promise<Integration[]> {
-  const client = getAccountClient(token)
+export async function getIntegrations (client: AccountClient, token: string): Promise<Integration[]> {
  return (await client.listIntegrations({ kind: GMAIL_INTEGRATION })) ?? []
 }

@@ -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<Token[]> {
-  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<Map<WorkspaceUuid, Token[]>> {
+  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<WorkspaceUuid, Token[]>()
+  for (const entry of byWorkspaces.entries()) {
+    result.set(
+      entry[0],
+      entry[1].map((secret: { secret: string }) => JSON.parse(secret.secret))
+    )
+  }
+  return result
 }