wieerwill/platform
1
0
Fork 0
mirror of https://github.com/hcengineering/platform.git synced 2025-05-14 03:10:27 +00:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

UBERF-10375: Fix full email messages sync (#8758)

Browse Source 
Signed-off-by: Artem Savchenko <armisav@gmail.com>
This commit is contained in:
Artyom Savchenko 2025-04-29 14:03:04 +07:00 committed by GitHub
parent 89d2d6872a
commit 5311096e05
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 58 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

91
services/gmail/pod-gmail/src/message/sync.ts
View File

@ -88,6 +88,16 @@ export class SyncManager {
await this.keyValueClient.setValue(historyKey, history) await this.keyValueClient.setValue(historyKey, history)
} }
private async getPageToken (userId: PersonId): Promise<string | null> {
const pageTokenKey = this.getPageTokenKey(userId)
return await this.keyValueClient.getValue<string>(pageTokenKey)
}
private async setPageToken (userId: PersonId, pageToken: string): Promise<void> {
const pageTokenKey = this.getPageTokenKey(userId)
await this.keyValueClient.setValue(pageTokenKey, pageToken)
}
private async partSync (userId: PersonId, userEmail: string | undefined, historyId: string): Promise<void> { private async partSync (userId: PersonId, userEmail: string | undefined, historyId: string): Promise<void> {
if (userEmail === undefined) { if (userEmail === undefined) {
throw new Error('Cannot sync without user email') throw new Error('Cannot sync without user email')
@ -151,7 +161,10 @@ export class SyncManager {
if (userEmail === undefined) { if (userEmail === undefined) {
throw new Error('Cannot sync without user email') throw new Error('Cannot sync without user email')
} }
const pageToken: string | undefined = undefined
// Get saved page token if exists to resume sync
let pageToken: string | undefined = (await this.getPageToken(userId)) ?? undefined
const query: gmail_v1.Params$Resource$Users$Messages$List = { const query: gmail_v1.Params$Resource$Users$Messages$List = {
userId: 'me', userId: 'me',
pageToken pageToken
@ -160,48 +173,56 @@ export class SyncManager {
query.q = q query.q = q
} }
let currentHistoryId: string | undefined let currentHistoryId: string | undefined
let totalProcessedMessages = 0
try { try {
const messagesIds: string[] = [] // Process one page at a time
while (true) { while (true) {
await this.rateLimiter.take(5) await this.rateLimiter.take(5)
const messages = await this.gmail.messages.list(query) const messages = await this.gmail.messages.list(query)
if (query.pageToken == null) {
this.ctx.info('Total messages', { const ids = messages.data.messages?.map((p) => p.id).filter((id) => id != null) ?? []
workspace: this.workspace, this.ctx.info('Processing page', {
userId, workspace: this.workspace,
resultSizeEstimate: messages.data.resultSizeEstimate userId,
}) messagesInPage: ids.length,
} totalProcessed: totalProcessedMessages,
const ids = messages.data.messages?.map((p) => p.id) ?? [] currentHistoryId,
for (let index = 0; index < ids.length; index++) { pageToken: query.pageToken
const id = ids[index] })
for (const id of ids) {
if (id == null) continue if (id == null) continue
messagesIds.push(id) try {
const message = await this.getMessage(id)
const historyId = message.data.historyId
await this.messageManager.saveMessage(message, userEmail)
if (historyId != null && q === undefined) {
if (currentHistoryId == null || Number(currentHistoryId) < Number(historyId)) {
currentHistoryId = historyId
}
}
} catch (err: any) {
this.ctx.error('Full sync message error', { workspace: this.workspace, userId, messageId: id, err })
}
} }
totalProcessedMessages += ids.length
if (messages.data.nextPageToken == null) { if (messages.data.nextPageToken == null) {
this.ctx.info('Break', { totalNewMessages: messagesIds.length }) this.ctx.info('Completed sync', { workspace: this.workspace, userId, totalMessages: totalProcessedMessages })
break break
} }
query.pageToken = messages.data.nextPageToken
// Update page token for the next iteration
pageToken = messages.data.nextPageToken
query.pageToken = pageToken
await this.setPageToken(userId, pageToken)
} }
for (let index = messagesIds.length - 1; index >= 0; index--) {
const id = messagesIds[index] if (currentHistoryId != null) {
try { await this.setHistoryId(userId, currentHistoryId)
const message = await this.getMessage(id)
const historyId = message.data.historyId
await this.messageManager.saveMessage(message, userEmail)
if (historyId != null && q === undefined) {
if (currentHistoryId == null || Number(currentHistoryId) < Number(historyId)) {
await this.setHistoryId(userId, historyId)
currentHistoryId = historyId
}
}
if (index % 500 === 0) {
this.ctx.info('Remaining messages to sync', { workspace: this.workspace, userId, count: index })
}
} catch (err: any) {
this.ctx.error('Full sync message error', { workspace: this.workspace, userId, err })
}
} }
this.ctx.info('Full sync finished', { workspaceUuid: this.workspace, userId, userEmail }) this.ctx.info('Full sync finished', { workspaceUuid: this.workspace, userId, userEmail })
} catch (err) { } catch (err) {
@ -243,4 +264,8 @@ export class SyncManager {
private getHistoryKey (userId: PersonId): string { private getHistoryKey (userId: PersonId): string {
return `history:${this.workspace}:${userId}` return `history:${this.workspace}:${userId}`
} }
private getPageTokenKey (userId: PersonId): string {
return `page-token:${this.workspace}:${userId}`
}
} }