Fix calendar sync (#9392)

Signed-off-by: Denis Bykhov <bykhov.denis@gmail.com>

.vscode/launch.json

@@ -723,6 +723,25 @@
       "outputCapture": "std",
       "cwd": "${workspaceRoot}/services/analytics-collector/pod-analytics-collector"
     },
+    {
+      "name": "Debug calendar",
+      "type": "node",
+      "request": "launch",
+      "args": ["src/index.ts"],
+      "env": {
+        "ACCOUNTS_URL" : "http://localhost:3000",
+        "Credentials" : "",
+        "KVS_URL" : "http://localhost:8094",
+        "SECRET" : "secret",
+        "WATCH_URL" : "https://calendar.hc.engineering/push"
+      },
+      "runtimeArgs": ["--nolazy", "-r", "ts-node/register"],
+      "runtimeVersion": "20",
+      "sourceMaps": true,
+      "showAsyncStacks": true,
+      "outputCapture": "std",
+      "cwd": "${workspaceRoot}/services/calendar/pod-calendar"
+    },
     {
       "name": "Debug AI bot",
       "type": "node",

services/calendar/pod-calendar/src/auth.ts

@@ -140,7 +140,7 @@ export class AuthController {
     const authRes = await this.authorize(code)
     await this.setWorkspaceIntegration(authRes)
     if (authRes.success) {
-      void IncomingSyncManager.sync(
+      void IncomingSyncManager.initSync(
         this.ctx,
         this.accountClient,
         this.client,

services/calendar/pod-calendar/src/calendar.ts

@@ -22,7 +22,6 @@ import { OAuth2Client } from 'google-auth-library'
 import { calendar_v3 } from 'googleapis'
 import { removeUserByEmail } from './kvsUtils'
 import { getRateLimitter, RateLimiter } from './rateLimiter'
-import { IncomingSyncManager } from './sync'
 import { CALENDAR_INTEGRATION, type Token } from './types'
 import {
   convertDate,
@@ -77,21 +76,6 @@ export class CalendarClient {
     return calendarClient
   }
 
-  async startSync (): Promise<void> {
-    try {
-      await IncomingSyncManager.sync(
-        this.ctx,
-        this.accountClient,
-        this.client,
-        this.user,
-        this.user.email,
-        this.calendar
-      )
-    } catch (err) {
-      this.ctx.error('Start sync error', { workspace: this.user.workspace, user: this.user.userId, err })
-    }
-  }
-
   private areDatesEqual (first: calendar_v3.Schema$EventDateTime, second: calendar_v3.Schema$EventDateTime): boolean {
     if (first.date != null && second.date != null) {
       return new Date(first.date).getTime() === new Date(second.date).getTime()

services/calendar/pod-calendar/src/calendarController.ts

@@ -68,8 +68,12 @@ export class CalendarController {
         const integrations = groups.get(info.uuid) ?? []
         if (await this.checkWorkspace(info, integrations)) {
           await limiter.add(async () => {
-            this.ctx.info('start workspace', { workspace: info.uuid })
-            await WorkspaceClient.run(this.ctx, this.accountClient, info.uuid)
+            try {
+              this.ctx.info('start workspace', { workspace: info.uuid })
+              await WorkspaceClient.run(this.ctx, this.accountClient, info.uuid)
+            } catch (err) {
+              this.ctx.error('Failed to start workspace', { workspace: info.uuid, error: err })
+            }
           })
         }
       }

services/calendar/pod-calendar/src/client.ts

@@ -19,7 +19,7 @@ import { setMetadata } from '@hcengineering/platform'
 import { createClient, getTransactorEndpoint } from '@hcengineering/server-client'
 
 export async function getClient (token: string): Promise<Client> {
-  const endpoint = await getTransactorEndpoint(token)
+  const endpoint = await getTransactorEndpoint(token, 'external')
   setMetadata(client.metadata.FilterModel, 'client')
   return await createClient(endpoint, token)
 }

services/calendar/pod-calendar/src/index.ts

@@ -14,7 +14,7 @@
 //
 
 // eslint-disable-next-line
-require('dotenv').config()
+import 'dotenv/config'
 // eslint-disable-next-line
-const { main } = require('./main')
+import { main } from './main'
 void main()

services/calendar/pod-calendar/src/kvsUtils.ts

@@ -15,14 +15,16 @@ export function getKvsClient (): KeyValueClient {
 export async function getSyncHistory (workspace: WorkspaceUuid): Promise<number | undefined> {
   const client = getKvsClient()
   const key = `${CALENDAR_INTEGRATION}:calendarSync:${workspace}`
-  const res = await client.getValue<number>(key)
-  return res ?? undefined
+  try {
+    const res = await client.getValue<number>(key)
+    return res ?? undefined
+  } catch {}
 }
 
-export async function setSyncHistory (workspace: WorkspaceUuid): Promise<void> {
+export async function setSyncHistory (workspace: WorkspaceUuid, value: number): Promise<void> {
   const client = getKvsClient()
   const key = `${CALENDAR_INTEGRATION}:calendarSync:${workspace}`
-  await client.setValue(key, Date.now())
+  await client.setValue(key, value)
 }
 
 function calendarsHistoryKey (user: User, email: GoogleEmail): string {

services/calendar/pod-calendar/src/outcomingClient.ts

@@ -65,7 +65,7 @@ export class OutcomingClient {
     } else if (type === 'create') {
       await this.create(event, calendar)
     }
-    await setSyncHistory(this.workspace)
+    await setSyncHistory(this.workspace, Date.now())
   }
 
   private async convertBody (event: Event): Promise<calendar_v3.Schema$Event> {

services/calendar/pod-calendar/src/sync.ts

@@ -41,10 +41,23 @@ import setting from '@hcengineering/setting'
 import { htmlToMarkup } from '@hcengineering/text'
 import { deepEqual } from 'fast-equals'
 import { calendar_v3 } from 'googleapis'
-import { getCalendarsSyncHistory, getEventHistory, setCalendarsSyncHistory, setEventHistory } from './kvsUtils'
+import { getClient } from './client'
+import {
+  getCalendarsSyncHistory,
+  getEventHistory,
+  removeUserByEmail,
+  setCalendarsSyncHistory,
+  setEventHistory
+} from './kvsUtils'
 import { getRateLimitter, RateLimiter } from './rateLimiter'
-import { GoogleEmail, Token, User } from './types'
-import { parseRecurrenceStrings } from './utils'
+import { CALENDAR_INTEGRATION, GoogleEmail, Token, User } from './types'
+import {
+  getGoogleClient,
+  getWorkspaceToken,
+  parseRecurrenceStrings,
+  removeIntegrationSecret,
+  setCredentials
+} from './utils'
 import { WatchController } from './watch'
 
 const locks = new Map<string, Promise<void>>()
@@ -91,7 +104,7 @@ export class IncomingSyncManager {
     this.systemClient = new TxOperations(client.client, core.account.System)
   }
 
-  static async sync (
+  static async initSync (
     ctx: MeasureContext,
     accountClient: AccountClient,
     client: TxOperations,
@@ -108,6 +121,31 @@ export class IncomingSyncManager {
     }
   }
 
+  static async sync (ctx: MeasureContext, accountClient: AccountClient, user: Token, email: GoogleEmail): Promise<void> {
+    const client = await getClient(getWorkspaceToken(user.workspace))
+    const txOp = new TxOperations(client, user.userId)
+    const google = getGoogleClient()
+    const mutex = await lock(`${user.workspace}:${user.userId}:${email}`)
+    try {
+      const authSucces = await setCredentials(google.auth, user)
+      if (!authSucces) {
+        await removeUserByEmail(user, user.email)
+        await removeIntegrationSecret(ctx, accountClient, {
+          socialId: user.userId,
+          kind: CALENDAR_INTEGRATION,
+          workspaceUuid: user.workspace,
+          key: user.email
+        })
+        return
+      }
+      const syncManager = new IncomingSyncManager(ctx, accountClient, txOp, user, email, google.google)
+      await syncManager.startSync()
+    } finally {
+      mutex()
+      await txOp.close()
+    }
+  }
+
   private async fillParticipants (): Promise<void> {
     const personsBySocialId = await getPersonRefsBySocialIds(this.client)
     const emailSocialIds = await this.client.findAll(contact.class.SocialIdentity, { type: SocialIdType.EMAIL })

services/calendar/pod-calendar/src/workspaceClient.ts

@@ -28,6 +28,7 @@ import { CalendarClient } from './calendar'
 import { getClient } from './client'
 import config from './config'
 import { addUserByEmail, getSyncHistory, setSyncHistory } from './kvsUtils'
+import { IncomingSyncManager } from './sync'
 import { getWorkspaceTokens } from './tokens'
 import { GoogleEmail, Token } from './types'
 import { getWorkspaceToken } from './utils'
@@ -37,6 +38,7 @@ export class WorkspaceClient {
   private readonly calendarsByExternal = new Map<string, ExternalCalendar>()
   readonly calendarsById = new Map<Ref<ExternalCalendar>, ExternalCalendar>()
   readonly participants = new Map<Ref<Person>, string>()
+  private lastSync: number = 0
 
   private constructor (
     private readonly ctx: MeasureContext,
@@ -86,14 +88,16 @@ export class WorkspaceClient {
     const tokens = await getWorkspaceTokens(this.accountClient, this.workspace)
     for (const token of tokens) {
       if (token.workspaceUuid === null) continue
-      await addUserByEmail(JSON.parse(token.secret), token.key as GoogleEmail)
-      await this.createCalendarClient(JSON.parse(token.secret))
+      const parsedToken = JSON.parse(token.secret)
+      await addUserByEmail(parsedToken, token.key as GoogleEmail)
+      await this.createCalendarClient(parsedToken)
     }
     await this.getNewEvents()
     const limiter = new RateLimiter(config.InitLimit)
-    for (const client of this.clients.values()) {
+    for (const token of tokens) {
       await limiter.add(async () => {
-        await client.startSync()
+        const parsedToken = JSON.parse(token.secret)
+        await IncomingSyncManager.sync(this.ctx, this.accountClient, parsedToken, parsedToken.email)
       })
     }
   }
@@ -118,27 +122,43 @@ export class WorkspaceClient {
 
   // #region Events
 
-  private async getSyncTime (): Promise<number | undefined> {
-    return (await getSyncHistory(this.workspace)) ?? undefined
-  }
-
-  private async updateSyncTime (): Promise<void> {
-    await setSyncHistory(this.workspace)
+  private async updateSyncHistory (): Promise<void> {
+    try {
+      await setSyncHistory(this.workspace, this.lastSync)
+    } catch (err) {
+      this.ctx.error('Failed to update sync history', {
+        workspace: this.workspace,
+        error: err instanceof Error ? err.message : String(err)
+      })
+    }
   }
 
   private async getNewEvents (): Promise<void> {
-    const lastSync = await this.getSyncTime()
+    const lastSync = await getSyncHistory(this.workspace)
+    this.lastSync = lastSync ?? 0
     const query = lastSync !== undefined ? { modifiedOn: { $gt: lastSync } } : {}
-    const newEvents = await this.client.findAll(calendar.class.Event, query)
+    const newEvents = await this.client.findAll(calendar.class.Event, query, { sort: { modifiedOn: 1 } })
+    const interval = setInterval(() => {
+      void this.updateSyncHistory()
+    }, 5000)
     for (const newEvent of newEvents) {
-      const client = this.getCalendarClientByCalendar(newEvent.calendar as Ref<ExternalCalendar>)
-      if (client === undefined) {
-        this.ctx.warn('Client not found', { calendar: newEvent.calendar, workspace: this.workspace })
-        continue
+      try {
+        const client = this.getCalendarClientByCalendar(newEvent.calendar as Ref<ExternalCalendar>)
+        if (client === undefined) {
+          this.ctx.warn('Client not found', { calendar: newEvent.calendar, workspace: this.workspace })
+          continue
+        }
+        await client.syncMyEvent(newEvent)
+        this.lastSync = newEvent.modifiedOn
+      } catch (err) {
+        this.ctx.error('Failed to sync event', {
+          event: newEvent._id,
+          workspace: this.workspace,
+          error: err instanceof Error ? err.message : String(err)
+        })
       }
-      await client.syncMyEvent(newEvent)
-      await this.updateSyncTime()
     }
+    clearInterval(interval)
     this.ctx.info('all outcoming messages synced', { workspace: this.workspace })
   }