UBERF-10925: Save gmail messages only for integration owner (#9061)

Signed-off-by: Artem Savchenko <armisav@gmail.com>

services/gmail/pod-gmail/src/message/adapter.ts

@@ -35,7 +35,7 @@ export function createMessageManager (
   socialId: SocialId
 ): IMessageManager {
   if (config.Version === 'v2') {
-    return new MessageManagerV2(ctx, attachmentHandler, client, keyValueClient, accountClient, token)
+    return new MessageManagerV2(ctx, attachmentHandler, client, keyValueClient, accountClient, token, socialId._id)
   } else {
     return new MessageManagerV1(ctx, client, attachmentHandler, socialId._id, workspace)
   }

services/gmail/pod-gmail/src/message/v2/message.ts

@@ -16,7 +16,7 @@ import { type GaxiosResponse } from 'gaxios'
 import { gmail_v1 } from 'googleapis'
 import sanitizeHtml from 'sanitize-html'
 
-import { TxOperations, type MeasureContext } from '@hcengineering/core'
+import { type MeasureContext, PersonId, TxOperations } from '@hcengineering/core'
 import {
   createMessages,
   parseEmailHeader,
@@ -40,7 +40,8 @@ export class MessageManagerV2 implements IMessageManager {
     private readonly txClient: TxOperations,
     private readonly keyValueClient: KeyValueClient,
     private readonly accountClient: AccountClient,
-    private readonly token: string
+    private readonly token: string,
+    private readonly personId: PersonId
   ) {}
 
   async saveMessage (message: GaxiosResponse<gmail_v1.Schema$Message>, me: string): Promise<void> {
@@ -65,7 +66,8 @@ export class MessageManagerV2 implements IMessageManager {
       this.token,
       this.wsInfo,
       res,
-      attachments
+      attachments,
+      [this.personId]
     )
   }
 }

services/mail/mail-common/src/__tests__/message.test.ts

@@ -0,0 +1,401 @@
+/* eslint-disable @typescript-eslint/unbound-method */
+//
+// Copyright © 2025 Hardcore Engineering Inc.
+//
+// Licensed under the Eclipse Public License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License. You may
+// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+import { createMessages } from '../message'
+import { WorkspaceLoginInfo } from '@hcengineering/account-client'
+import { PersonSpace } from '@hcengineering/contact'
+import { MeasureContext, PersonId, PersonUuid, Ref, TxOperations } from '@hcengineering/core'
+import { KeyValueClient } from '@hcengineering/kvs-client'
+import { Producer } from 'kafkajs'
+import { Attachment, BaseConfig, EmailContact, EmailMessage } from '../types'
+import { PersonCacheFactory } from '../person'
+import { PersonSpacesCacheFactory } from '../personSpaces'
+import { ChannelCacheFactory } from '../channel'
+import { ThreadLookupService } from '../thread'
+
+// Mock dependencies
+jest.mock('../person')
+jest.mock('../personSpaces')
+jest.mock('../channel')
+jest.mock('../thread')
+jest.mock('kafkajs')
+
+describe('createMessages', () => {
+  // Setup test vars
+  let mockConfig: BaseConfig
+  let mockCtx: MeasureContext
+  let mockTxClient: jest.Mocked<TxOperations>
+  let mockKvsClient: jest.Mocked<KeyValueClient>
+  let mockProducer: jest.Mocked<Producer>
+  let mockToken: string
+  let mockWsInfo: WorkspaceLoginInfo
+  let mockMessage: EmailMessage
+  let mockAttachments: Attachment[]
+
+  // Mock person objects
+  const mockFromPerson = {
+    socialId: 'from-person-id' as PersonId,
+    uuid: 'from-uuid' as PersonUuid,
+    localPerson: 'from-local'
+  }
+
+  const mockToPerson1 = {
+    socialId: 'to-person-id-1' as PersonId,
+    uuid: 'to-uuid-1' as PersonUuid,
+    localPerson: 'to-local-1'
+  }
+
+  const mockToPerson2 = {
+    socialId: 'to-person-id-2' as PersonId,
+    uuid: 'to-uuid-2' as PersonUuid,
+    localPerson: 'to-local-2'
+  }
+
+  // Mock email contacts
+  const mockFromContact: EmailContact = {
+    email: 'from@example.com',
+    firstName: 'From',
+    lastName: 'User'
+  }
+
+  const mockToContact1: EmailContact = {
+    email: 'to1@example.com',
+    firstName: 'To1',
+    lastName: 'User'
+  }
+
+  const mockToContact2: EmailContact = {
+    email: 'to2@example.com',
+    firstName: 'To2',
+    lastName: 'User'
+  }
+
+  // Mock PersonCache class
+  const mockPersonCache = {
+    ensurePerson: jest.fn()
+  }
+
+  // Mock PersonSpaces class
+  const mockPersonSpacesCache = {
+    getPersonSpaces: jest.fn()
+  }
+
+  // Mock ChannelCache class
+  const mockChannelCache = {
+    getOrCreateChannel: jest.fn()
+  }
+
+  // Mock ThreadLookupService class
+  const mockThreadLookup = {
+    getThreadId: jest.fn(),
+    getParentThreadId: jest.fn(),
+    setThreadId: jest.fn()
+  }
+
+  // Mock space
+  const mockSpace: PersonSpace = {
+    _id: 'space-id' as Ref<PersonSpace>,
+    _class: 'class.contact.PersonSpace' as any,
+    person: 'person-id' as any,
+    name: 'Test Space',
+    private: true
+  } as any
+
+  beforeEach(() => {
+    jest.resetAllMocks()
+
+    // Setup mocks
+    mockCtx = {
+      info: jest.fn(),
+      error: jest.fn(),
+      warn: jest.fn()
+    } as unknown as MeasureContext
+
+    mockTxClient = {
+      createDoc: jest.fn().mockResolvedValue('thread-id'),
+      createMixin: jest.fn().mockResolvedValue(undefined),
+      findOne: jest.fn()
+    } as unknown as jest.Mocked<TxOperations>
+
+    mockKvsClient = {} as unknown as jest.Mocked<KeyValueClient>
+
+    mockProducer = {
+      send: jest.fn().mockResolvedValue(undefined)
+    } as unknown as jest.Mocked<Producer>
+
+    mockToken = 'test-token'
+
+    mockWsInfo = {
+      workspace: 'workspace-id',
+      endpoint: 'wss://example.com',
+      token: 'ws-token',
+      workspaceUrl: 'https://example.com',
+      workspaceDataId: 'data-id'
+    } as any
+
+    mockMessage = {
+      mailId: 'test-mail-id',
+      from: mockFromContact,
+      to: [mockToContact1, mockToContact2],
+      copy: [],
+      subject: 'Test Subject',
+      textContent: 'Test Content',
+      htmlContent: '<p>Test Content</p>',
+      sendOn: Date.now()
+    } as any
+
+    mockAttachments = []
+
+    // Setup config
+    mockConfig = {
+      CommunicationTopic: 'communication-topic'
+    } as any
+
+    // Setup mock factories
+    ;(PersonCacheFactory.getInstance as jest.Mock).mockReturnValue(mockPersonCache)
+    ;(PersonSpacesCacheFactory.getInstance as jest.Mock).mockReturnValue(mockPersonSpacesCache)
+    ;(ChannelCacheFactory.getInstance as jest.Mock).mockReturnValue(mockChannelCache)
+    ;(ThreadLookupService.getInstance as jest.Mock).mockReturnValue(mockThreadLookup)
+
+    // Setup mock return values
+    mockPersonCache.ensurePerson
+      .mockResolvedValueOnce(mockFromPerson) // For from person
+      .mockResolvedValueOnce(mockToPerson1) // For first to person
+      .mockResolvedValueOnce(mockToPerson2) // For second to person
+
+    mockPersonSpacesCache.getPersonSpaces.mockResolvedValue([mockSpace])
+
+    mockChannelCache.getOrCreateChannel.mockResolvedValue('channel-id')
+
+    mockThreadLookup.getThreadId.mockResolvedValue(undefined)
+    mockThreadLookup.getParentThreadId.mockResolvedValue(undefined)
+  })
+
+  it('should use default participants when targetPersons is not provided', async () => {
+    // Act
+    await createMessages(
+      mockConfig,
+      mockCtx,
+      mockTxClient,
+      mockKvsClient,
+      mockProducer,
+      mockToken,
+      mockWsInfo,
+      mockMessage,
+      mockAttachments
+    )
+
+    // Assert
+    // This checks that createDoc was called with the expected participants list
+    // Default participants should include the sender and all recipients
+    expect(mockTxClient.createDoc).toHaveBeenCalledWith(
+      expect.any(String),
+      expect.any(String),
+      expect.objectContaining({
+        members: [mockFromPerson.socialId, mockToPerson1.socialId, mockToPerson2.socialId]
+      }),
+      expect.any(String),
+      expect.any(Number),
+      expect.any(String)
+    )
+
+    // Also check that the channel was created with the same participants
+    expect(mockChannelCache.getOrCreateChannel).toHaveBeenCalledWith(
+      mockSpace._id,
+      [mockFromPerson.socialId, mockToPerson1.socialId, mockToPerson2.socialId],
+      mockToContact1.email,
+      expect.any(String)
+    )
+  })
+
+  it('should use provided targetPersons when specified', async () => {
+    // Arrange
+    const customTargetPersons = ['custom-person-1', 'custom-person-2'] as PersonId[]
+
+    // Act
+    await createMessages(
+      mockConfig,
+      mockCtx,
+      mockTxClient,
+      mockKvsClient,
+      mockProducer,
+      mockToken,
+      mockWsInfo,
+      mockMessage,
+      mockAttachments,
+      customTargetPersons
+    )
+
+    // Assert
+    // Check that createDoc was called with the custom participants list
+    expect(mockTxClient.createDoc).toHaveBeenCalledWith(
+      expect.any(String),
+      expect.any(String),
+      expect.objectContaining({
+        members: customTargetPersons
+      }),
+      expect.any(String),
+      expect.any(Number),
+      expect.any(String)
+    )
+
+    // Also check that the channel was created with the custom participants
+    expect(mockChannelCache.getOrCreateChannel).toHaveBeenCalledWith(
+      mockSpace._id,
+      customTargetPersons,
+      mockToContact1.email,
+      expect.any(String)
+    )
+  })
+
+  it('should handle empty targetPersons array', async () => {
+    // Arrange
+    const emptyTargetPersons: PersonId[] = []
+
+    // Act
+    await createMessages(
+      mockConfig,
+      mockCtx,
+      mockTxClient,
+      mockKvsClient,
+      mockProducer,
+      mockToken,
+      mockWsInfo,
+      mockMessage,
+      mockAttachments,
+      emptyTargetPersons
+    )
+
+    // Assert
+    // Check that createDoc was called with the empty participants list
+    expect(mockTxClient.createDoc).toHaveBeenCalledWith(
+      expect.any(String),
+      expect.any(String),
+      expect.objectContaining({
+        members: [] // Empty array
+      }),
+      expect.any(String),
+      expect.any(Number),
+      expect.any(String)
+    )
+  })
+
+  it('should handle missing recipient persons gracefully', async () => {
+    // Arrange - setup person cache to return undefined for the second recipient
+    mockPersonCache.ensurePerson
+      .mockReset()
+      .mockResolvedValueOnce(mockFromPerson) // For from person
+      .mockResolvedValueOnce(mockToPerson1) // For first to person
+      .mockResolvedValueOnce(undefined) // For second to person (missing)
+
+    // Act
+    await createMessages(
+      mockConfig,
+      mockCtx,
+      mockTxClient,
+      mockKvsClient,
+      mockProducer,
+      mockToken,
+      mockWsInfo,
+      mockMessage,
+      mockAttachments
+    )
+
+    // Assert
+    // Only the available persons should be included in participants
+    expect(mockTxClient.createDoc).toHaveBeenCalledWith(
+      expect.any(String),
+      expect.any(String),
+      expect.objectContaining({
+        members: [
+          mockFromPerson.socialId,
+          mockToPerson1.socialId
+          // mockToPerson2.socialId should not be here
+        ]
+      }),
+      expect.any(String),
+      expect.any(Number),
+      expect.any(String)
+    )
+  })
+
+  it('should handle case where all recipients are missing', async () => {
+    // Arrange - setup person cache to return undefined for all recipients
+    mockPersonCache.ensurePerson
+      .mockReset()
+      .mockResolvedValueOnce(mockFromPerson) // For from person
+      .mockResolvedValueOnce(undefined) // For first to person (missing)
+      .mockResolvedValueOnce(undefined) // For second to person (missing)
+
+    // Act
+    await createMessages(
+      mockConfig,
+      mockCtx,
+      mockTxClient,
+      mockKvsClient,
+      mockProducer,
+      mockToken,
+      mockWsInfo,
+      mockMessage,
+      mockAttachments
+    )
+
+    // Assert
+    // Should log an error and not proceed with message creation
+    expect(mockCtx.error).toHaveBeenCalledWith('Unable to create message without a proper TO', expect.any(Object))
+    expect(mockTxClient.createDoc).not.toHaveBeenCalled()
+  })
+
+  it('should handle empty to list with copy recipients', async () => {
+    // Arrange - setup message with empty 'to' but populated 'copy'
+    const messageWithCopy: EmailMessage = {
+      ...mockMessage,
+      to: [],
+      copy: [mockToContact1]
+    }
+
+    mockPersonCache.ensurePerson
+      .mockReset()
+      .mockResolvedValueOnce(mockFromPerson) // For from person
+      .mockResolvedValueOnce(mockToPerson1) // For copy person
+
+    // Act
+    await createMessages(
+      mockConfig,
+      mockCtx,
+      mockTxClient,
+      mockKvsClient,
+      mockProducer,
+      mockToken,
+      mockWsInfo,
+      messageWithCopy,
+      mockAttachments
+    )
+
+    // Assert
+    // Should include the copy recipients in participants
+    expect(mockTxClient.createDoc).toHaveBeenCalledWith(
+      expect.any(String),
+      expect.any(String),
+      expect.objectContaining({
+        members: [mockFromPerson.socialId, mockToPerson1.socialId]
+      }),
+      expect.any(String),
+      expect.any(Number),
+      expect.any(String)
+    )
+  })
+})

services/mail/mail-common/src/message.ts

@@ -46,6 +46,32 @@ import { PersonSpacesCacheFactory } from './personSpaces'
 import { ChannelCache, ChannelCacheFactory } from './channel'
 import { ThreadLookupService } from './thread'
 
+/**
+ * Creates mail messages in the platform
+ *
+ * This function processes an email message and creates corresponding chat messages. It handles:
+ * - Ensuring persons exist for email addresses
+ * - Finding or creating channels for participants
+ * - Creating threads for messages
+ * - Uploading attachments to storage
+ * - Sending message events to Kafka
+ *
+ * @param {BaseConfig} config - Configuration options including storage and Kafka settings
+ * @param {MeasureContext} ctx - Context for logging and performance measurement
+ * @param {TxOperations} txClient - Client for database transactions
+ * @param {KeyValueClient} keyValueClient - Client for key-value storage operations
+ * @param {Producer} producer - Kafka producer for sending message events
+ * @param {string} token - Authentication token for API calls
+ * @param {WorkspaceLoginInfo} wsInfo - Workspace information including ID and URLs
+ * @param {EmailMessage} message - The email message to process
+ * @param {Attachment[]} attachments - Array of attachments for the message
+ * @param {PersonId[]} [targetPersons] - Optional list of specific persons who should receive the message.
+ *                                       If not provided, all existing accounts from email addresses will be used.
+ * @returns {Promise<void>} A promise that resolves when all messages have been created
+ * @throws Will log errors but not throw exceptions for partial failures
+ *
+ * @public
+ */
 export async function createMessages (
   config: BaseConfig,
   ctx: MeasureContext,
@@ -55,7 +81,8 @@ export async function createMessages (
   token: string,
   wsInfo: WorkspaceLoginInfo,
   message: EmailMessage,
-  attachments: Attachment[]
+  attachments: Attachment[],
+  targetPersons?: PersonId[]
 ): Promise<void> {
   const { mailId, from, subject, replyTo } = message
   const tos = [...(message.to ?? []), ...(message.copy ?? [])]
@@ -82,7 +109,7 @@ export async function createMessages (
   }
 
   const modifiedBy = fromPerson.socialId
-  const participants = [fromPerson.socialId, ...toPersons.map((p) => p.socialId)]
+  const participants = targetPersons ?? [fromPerson.socialId, ...toPersons.map((p) => p.socialId)]
   const content = getMdContent(ctx, message)
 
   const attachedBlobs: Attachment[] = []