fix(messaging): refactor SentMessagePersistenceService (#20077)

refactored `SentMessagePersistenceService` to be thin wrapper over `saveMessagesAndEnqueueContactCreation` this fixes a bug where the old logic did not call match participants causing messages to not show up
2026-04-27 17:26:41 +05:30
parent 15c52d3a39
commit 6545ca274c
8 changed files with 253 additions and 171 deletions
@@ -117,6 +117,7 @@ import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/mess
    MessagingOngoingStaleCronCommand,
    MessagingRelaunchFailedMessageChannelsCronCommand,
    MessagingProcessGroupEmailActionsService,
+    MessagingSaveMessagesAndEnqueueContactCreationService,
  ],
 })
 export class MessagingImportManagerModule {}
@@ -6,6 +6,7 @@ import { MessageFolderEntity } from 'src/engine/metadata-modules/message-folder/
 import { OAuth2ClientManagerModule } from 'src/modules/connected-account/oauth2-client-manager/oauth2-client-manager.module';
 import { MessagingIMAPDriverModule } from 'src/modules/messaging/message-import-manager/drivers/imap/messaging-imap-driver.module';
 import { MessagingSmtpDriverModule } from 'src/modules/messaging/message-import-manager/drivers/smtp/messaging-smtp-driver.module';
+import { MessagingImportManagerModule } from 'src/modules/messaging/message-import-manager/messaging-import-manager.module';
 import { GmailMessageOutboundService } from 'src/modules/messaging/message-outbound-manager/drivers/gmail/services/gmail-message-outbound.service';
 import { ImapSmtpMessageOutboundService } from 'src/modules/messaging/message-outbound-manager/drivers/imap/services/imap-smtp-message-outbound.service';
 import { MicrosoftMessageOutboundService } from 'src/modules/messaging/message-outbound-manager/drivers/microsoft/services/microsoft-message-outbound.service';
@@ -18,6 +19,7 @@ import { SentMessagePersistenceService } from 'src/modules/messaging/message-out
    OAuth2ClientManagerModule,
    MessagingIMAPDriverModule,
    MessagingSmtpDriverModule,
+    MessagingImportManagerModule,
    TypeOrmModule.forFeature([MessageChannelEntity, MessageFolderEntity]),
  ],
  providers: [
@@ -1,185 +1,37 @@
-import { Injectable, Logger } from '@nestjs/common';
+import { Injectable } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';

-import { MessageParticipantRole } from 'twenty-shared/types';
-import { v4 } from 'uuid';
+import { Repository } from 'typeorm';

-import { type ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity';
-import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
-import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util';
-import { MessageDirection } from 'src/modules/messaging/common/enums/message-direction.enum';
-import { type MessageChannelMessageAssociationWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel-message-association.workspace-entity';
-import { type MessageParticipantWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-participant.workspace-entity';
-import { type MessageThreadWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-thread.workspace-entity';
-import { type MessageWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message.workspace-entity';
-import { type SendMessageResult } from 'src/modules/messaging/message-outbound-manager/types/send-message-result.type';
-
-type PersistSentMessageInput = {
-  sendResult: SendMessageResult;
-  subject: string;
-  body: string;
-  recipients: { to: string[]; cc: string[]; bcc: string[] };
-  connectedAccount: ConnectedAccountEntity;
-  messageChannelId: string;
-  inReplyTo?: string;
-  workspaceId: string;
-};
+import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity';
+import { MessagingSaveMessagesAndEnqueueContactCreationService } from 'src/modules/messaging/message-import-manager/services/messaging-save-messages-and-enqueue-contact-creation.service';
+import { type PersistSentMessageInput } from 'src/modules/messaging/message-outbound-manager/types/persist-sent-message-input.type';
+import { formatSentMessage } from 'src/modules/messaging/message-outbound-manager/utils/format-sent-message.util';

@Injectable()
 export class SentMessagePersistenceService {
-  private readonly logger = new Logger(SentMessagePersistenceService.name);
-
  constructor(
-    private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    @InjectRepository(MessageChannelEntity)
+    private readonly messageChannelRepository: Repository<MessageChannelEntity>,
+    private readonly saveMessagesAndEnqueueContactCreationService: MessagingSaveMessagesAndEnqueueContactCreationService,
  ) {}

  async persistSentMessage(input: PersistSentMessageInput): Promise<void> {
-    const authContext = buildSystemAuthContext(input.workspaceId);
-
-    await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageWorkspaceEntity>(
-          input.workspaceId,
-          'message',
-        );
-
-      const messageThreadRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageThreadWorkspaceEntity>(
-          input.workspaceId,
-          'messageThread',
-        );
-
-      const messageParticipantRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageParticipantWorkspaceEntity>(
-          input.workspaceId,
-          'messageParticipant',
-        );
-
-      const associationRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelMessageAssociationWorkspaceEntity>(
-          input.workspaceId,
-          'messageChannelMessageAssociation',
-        );
-
-      const messageThreadId = await this.findOrCreateThread({
-        messageRepository,
-        messageThreadRepository,
-        inReplyTo: input.inReplyTo,
-        subject: input.subject,
-      });
-
-      const messageId = v4();
-
-      await messageRepository.insert({
-        id: messageId,
-        headerMessageId: input.sendResult.headerMessageId,
-        subject: input.subject,
-        text: input.body,
-        receivedAt: new Date(),
-        messageThreadId,
-      });
-
-      const participants = this.buildParticipants(
-        messageId,
-        input.connectedAccount.handle ?? '',
-        input.recipients,
-      );
-
-      if (participants.length > 0) {
-        await messageParticipantRepository.insert(participants);
-      }
-
-      await associationRepository.insert({
-        messageChannelId: input.messageChannelId,
-        messageId,
-        messageExternalId: input.sendResult.messageExternalId ?? null,
-        messageThreadExternalId: input.sendResult.threadExternalId ?? null,
-        direction: MessageDirection.OUTGOING,
-      });
-    }, authContext);
-  }
-
-  private async findOrCreateThread({
-    messageRepository,
-    messageThreadRepository,
-    inReplyTo,
-    subject,
-  }: {
-    messageRepository: Awaited<
-      ReturnType<GlobalWorkspaceOrmManager['getRepository']>
-    >;
-    messageThreadRepository: Awaited<
-      ReturnType<GlobalWorkspaceOrmManager['getRepository']>
-    >;
-    inReplyTo?: string;
-    subject: string;
-  }): Promise<string> {
-    if (inReplyTo) {
-      const parentMessage = await messageRepository.findOne({
-        where: { headerMessageId: inReplyTo },
-      });
-
-      if (parentMessage?.messageThreadId) {
-        return parentMessage.messageThreadId;
-      }
-    }
-
-    const threadId = v4();
-
-    await messageThreadRepository.insert({
-      id: threadId,
-      subject,
+    const messageChannel = await this.messageChannelRepository.findOneOrFail({
+      where: {
+        id: input.messageChannelId,
+        workspaceId: input.workspaceId,
+      },
+      relations: { connectedAccount: true },
    });

-    return threadId;
-  }
+    const messageToSave = formatSentMessage(input);

-  private buildParticipants(
-    messageId: string,
-    senderHandle: string,
-    recipients: { to: string[]; cc: string[]; bcc: string[] },
-  ): Pick<
-    MessageParticipantWorkspaceEntity,
-    'messageId' | 'handle' | 'displayName' | 'role'
-  >[] {
-    const participants: Pick<
-      MessageParticipantWorkspaceEntity,
-      'messageId' | 'handle' | 'displayName' | 'role'
-    >[] = [];
-
-    participants.push({
-      messageId,
-      handle: senderHandle,
-      displayName: senderHandle,
-      role: MessageParticipantRole.FROM,
-    });
-
-    for (const email of recipients.to) {
-      participants.push({
-        messageId,
-        handle: email,
-        displayName: email,
-        role: MessageParticipantRole.TO,
-      });
-    }
-
-    for (const email of recipients.cc) {
-      participants.push({
-        messageId,
-        handle: email,
-        displayName: email,
-        role: MessageParticipantRole.CC,
-      });
-    }
-
-    for (const email of recipients.bcc) {
-      participants.push({
-        messageId,
-        handle: email,
-        displayName: email,
-        role: MessageParticipantRole.BCC,
-      });
-    }
-
-    return participants;
+    await this.saveMessagesAndEnqueueContactCreationService.saveMessagesAndEnqueueContactCreation(
+      [messageToSave],
+      messageChannel,
+      messageChannel.connectedAccount,
+      input.workspaceId,
+    );
  }
 }
@@ -0,0 +1,13 @@
+import { type ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity';
+import { type SendMessageResult } from 'src/modules/messaging/message-outbound-manager/types/send-message-result.type';
+
+export type PersistSentMessageInput = {
+  sendResult: SendMessageResult;
+  subject: string;
+  body: string;
+  recipients: { to: string[]; cc: string[]; bcc: string[] };
+  connectedAccount: Pick<ConnectedAccountEntity, 'id' | 'handle'>;
+  messageChannelId: string;
+  inReplyTo?: string;
+  workspaceId: string;
+};
@@ -0,0 +1,90 @@
+import { MessageParticipantRole } from 'twenty-shared/types';
+
+import { type ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity';
+import { MessageDirection } from 'src/modules/messaging/common/enums/message-direction.enum';
+import { type PersistSentMessageInput } from 'src/modules/messaging/message-outbound-manager/types/persist-sent-message-input.type';
+import { formatSentMessage } from 'src/modules/messaging/message-outbound-manager/utils/format-sent-message.util';
+
+const buildInput = (
+  overrides: Partial<PersistSentMessageInput> = {},
+): PersistSentMessageInput => ({
+  sendResult: {
+    headerMessageId: '<msg-1@mail.example>',
+    messageExternalId: 'gmail-external-1',
+    threadExternalId: 'thread-external-1',
+  },
+  subject: 'Quarterly review',
+  body: 'See attached.',
+  recipients: { to: [], cc: [], bcc: [] },
+  connectedAccount: {
+    handle: 'sender@example.com',
+  } as ConnectedAccountEntity,
+  messageChannelId: 'channel-1',
+  workspaceId: 'workspace-1',
+  ...overrides,
+});
+
+describe('formatSentMessage', () => {
+  it('should mark the message as OUTGOING with the sender as FROM participant', () => {
+    const message = formatSentMessage(buildInput());
+
+    expect(message.direction).toBe(MessageDirection.OUTGOING);
+    expect(message.participants).toContainEqual({
+      role: MessageParticipantRole.FROM,
+      handle: 'sender@example.com',
+      displayName: 'sender@example.com',
+    });
+  });
+
+  it('should emit one participant per to/cc/bcc recipient with correct roles', () => {
+    const message = formatSentMessage(
+      buildInput({
+        recipients: {
+          to: ['alice@example.com'],
+          cc: ['bob@example.com', 'carol@example.com'],
+          bcc: ['dave@example.com'],
+        },
+      }),
+    );
+
+    const rolesByHandle = Object.fromEntries(
+      message.participants.map((participant) => [
+        participant.handle,
+        participant.role,
+      ]),
+    );
+
+    expect(rolesByHandle).toEqual({
+      'sender@example.com': MessageParticipantRole.FROM,
+      'alice@example.com': MessageParticipantRole.TO,
+      'bob@example.com': MessageParticipantRole.CC,
+      'carol@example.com': MessageParticipantRole.CC,
+      'dave@example.com': MessageParticipantRole.BCC,
+    });
+  });
+
+  it('should fall back to the headerMessageId when the provider omits external ids so unrelated sends do not collide on a shared empty thread key', () => {
+    const message = formatSentMessage(
+      buildInput({
+        sendResult: {
+          headerMessageId: '<msg-2@mail.example>',
+          messageExternalId: undefined,
+          threadExternalId: undefined,
+        },
+      }),
+    );
+
+    expect(message.externalId).toBe('<msg-2@mail.example>');
+    expect(message.messageThreadExternalId).toBe('<msg-2@mail.example>');
+    expect(message.headerMessageId).toBe('<msg-2@mail.example>');
+  });
+
+  it('should copy subject and body verbatim and start with no folder associations', () => {
+    const message = formatSentMessage(buildInput());
+
+    expect(message.subject).toBe('Quarterly review');
+    expect(message.text).toBe('See attached.');
+    expect(message.attachments).toEqual([]);
+    expect(message.messageFolderIds).toBeUndefined();
+  });
+});
@@ -0,0 +1,41 @@
+import { resolveOutboundThreadExternalId } from 'src/modules/messaging/message-outbound-manager/utils/resolve-outbound-thread-external-id.util';
+
+describe('resolveOutboundThreadExternalId', () => {
+  it("should prefer the provider's thread id when present", () => {
+    const result = resolveOutboundThreadExternalId({
+      sendResult: {
+        headerMessageId: '<reply@mail.example>',
+        messageExternalId: 'gmail-id',
+        threadExternalId: 'gmail-thread-id',
+      },
+      inReplyTo: '<parent@mail.example>',
+    });
+
+    expect(result).toBe('gmail-thread-id');
+  });
+
+  it('should fall back to inReplyTo so IMAP/SMTP replies attach to the parent thread', () => {
+    const result = resolveOutboundThreadExternalId({
+      sendResult: {
+        headerMessageId: '<reply@mail.example>',
+        messageExternalId: undefined,
+        threadExternalId: undefined,
+      },
+      inReplyTo: '<parent@mail.example>',
+    });
+
+    expect(result).toBe('<parent@mail.example>');
+  });
+
+  it('should fall back to the headerMessageId for new IMAP/SMTP sends so unrelated threads do not collide', () => {
+    const result = resolveOutboundThreadExternalId({
+      sendResult: {
+        headerMessageId: '<msg@mail.example>',
+        messageExternalId: undefined,
+        threadExternalId: undefined,
+      },
+    });
+
+    expect(result).toBe('<msg@mail.example>');
+  });
+});
@@ -0,0 +1,58 @@
+import { isNonEmptyString } from '@sniptt/guards';
+import { MessageParticipantRole } from 'twenty-shared/types';
+
+import { MessageDirection } from 'src/modules/messaging/common/enums/message-direction.enum';
+import {
+  type MessageParticipant,
+  type MessageWithParticipants,
+} from 'src/modules/messaging/message-import-manager/types/message';
+import { type PersistSentMessageInput } from 'src/modules/messaging/message-outbound-manager/types/persist-sent-message-input.type';
+import { resolveOutboundThreadExternalId } from 'src/modules/messaging/message-outbound-manager/utils/resolve-outbound-thread-external-id.util';
+
+export const formatSentMessage = (
+  input: PersistSentMessageInput,
+): MessageWithParticipants => {
+  const senderHandle = input.connectedAccount.handle ?? '';
+
+  const participants: MessageParticipant[] = [
+    {
+      role: MessageParticipantRole.FROM,
+      handle: senderHandle,
+      displayName: senderHandle,
+    },
+    ...input.recipients.to.map((handle) => ({
+      role: MessageParticipantRole.TO,
+      handle,
+      displayName: handle,
+    })),
+    ...input.recipients.cc.map((handle) => ({
+      role: MessageParticipantRole.CC,
+      handle,
+      displayName: handle,
+    })),
+    ...input.recipients.bcc.map((handle) => ({
+      role: MessageParticipantRole.BCC,
+      handle,
+      displayName: handle,
+    })),
+  ];
+
+  const headerMessageId = input.sendResult.headerMessageId;
+
+  return {
+    externalId: isNonEmptyString(input.sendResult.messageExternalId)
+      ? input.sendResult.messageExternalId
+      : headerMessageId,
+    headerMessageId,
+    messageThreadExternalId: resolveOutboundThreadExternalId({
+      sendResult: input.sendResult,
+      inReplyTo: input.inReplyTo,
+    }),
+    subject: input.subject,
+    text: input.body,
+    receivedAt: new Date(),
+    direction: MessageDirection.OUTGOING,
+    attachments: [],
+    participants,
+  };
+};
@@ -0,0 +1,25 @@
+import { isNonEmptyString } from '@sniptt/guards';
+
+import { type SendMessageResult } from 'src/modules/messaging/message-outbound-manager/types/send-message-result.type';
+
+export const resolveOutboundThreadExternalId = ({
+  sendResult,
+  inReplyTo,
+}: {
+  sendResult: SendMessageResult;
+  inReplyTo?: string;
+}): string => {
+  if (isNonEmptyString(sendResult.threadExternalId)) {
+    return sendResult.threadExternalId;
+  }
+
+  // IMAP/SMTP have no server-side thread id. Reuse the parent's Message-ID so
+  // the reply attaches to the same thread the parent stored under.
+  if (isNonEmptyString(inReplyTo)) {
+    return inReplyTo;
+  }
+
+  // New IMAP/SMTP send: own Message-ID is unique per RFC822, so unrelated
+  // sends never collide on a shared empty thread key.
+  return sendResult.headerMessageId;
+};