Use twentyORM in Timeline messaging

packages/twenty-server/src/engine/core-modules/messaging/dtos/timeline-thread-participant.dto.ts

@@ -1,14 +1,14 @@
-import { ObjectType, Field } from '@nestjs/graphql';
+import { Field, ObjectType } from '@nestjs/graphql';
 
 import { UUIDScalarType } from 'src/engine/api/graphql/workspace-schema-builder/graphql-types/scalars';
 
 @ObjectType('TimelineThreadParticipant')
 export class TimelineThreadParticipant {
   @Field(() => UUIDScalarType, { nullable: true })
-  personId: string;
+  personId: string | null;
 
   @Field(() => UUIDScalarType, { nullable: true })
-  workspaceMemberId: string;
+  workspaceMemberId: string | null;
 
   @Field()
   firstName: string;

packages/twenty-server/src/engine/core-modules/messaging/services/get-messages.service.ts

@@ -0,0 +1,101 @@
+import { Injectable } from '@nestjs/common';
+
+import { TIMELINE_THREADS_DEFAULT_PAGE_SIZE } from 'src/engine/core-modules/messaging/constants/messaging.constants';
+import { TimelineThreadsWithTotal } from 'src/engine/core-modules/messaging/dtos/timeline-threads-with-total.dto';
+import { TimelineMessagingService } from 'src/engine/core-modules/messaging/services/timeline-messaging.service';
+import { formatThreads } from 'src/engine/core-modules/messaging/utils/format-threads.util';
+import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
+import { PersonWorkspaceEntity } from 'src/modules/person/standard-objects/person.workspace-entity';
+
+@Injectable()
+export class GetMessagesService {
+  constructor(
+    private readonly twentyORMManager: TwentyORMManager,
+    private readonly timelineMessagingService: TimelineMessagingService,
+  ) {}
+
+  async getMessagesFromPersonIds(
+    workspaceMemberId: string,
+    personIds: string[],
+    page = 1,
+    pageSize: number = TIMELINE_THREADS_DEFAULT_PAGE_SIZE,
+  ): Promise<TimelineThreadsWithTotal> {
+    const offset = (page - 1) * pageSize;
+
+    const { messageThreads, totalNumberOfThreads } =
+      await this.timelineMessagingService.getAndCountMessageThreads(
+        personIds,
+        offset,
+        pageSize,
+      );
+
+    if (!messageThreads) {
+      return {
+        totalNumberOfThreads: 0,
+        timelineThreads: [],
+      };
+    }
+
+    const messageThreadIds = messageThreads.map(
+      (messageThread) => messageThread.id,
+    );
+
+    const threadParticipantsByThreadId =
+      await this.timelineMessagingService.getThreadParticipantsByThreadId(
+        messageThreadIds,
+      );
+
+    const threadVisibilityByThreadId =
+      await this.timelineMessagingService.getThreadVisibilityByThreadId(
+        messageThreadIds,
+        workspaceMemberId,
+      );
+
+    return {
+      totalNumberOfThreads,
+      timelineThreads: formatThreads(
+        messageThreads,
+        threadParticipantsByThreadId,
+        threadVisibilityByThreadId,
+      ),
+    };
+  }
+
+  async getMessagesFromCompanyId(
+    workspaceMemberId: string,
+    companyId: string,
+    page = 1,
+    pageSize: number = TIMELINE_THREADS_DEFAULT_PAGE_SIZE,
+  ): Promise<TimelineThreadsWithTotal> {
+    const personRepository =
+      await this.twentyORMManager.getRepository<PersonWorkspaceEntity>(
+        'person',
+      );
+    const personIds = (
+      await personRepository.find({
+        where: {
+          companyId,
+        },
+        select: {
+          id: true,
+        },
+      })
+    ).map((person) => person.id);
+
+    if (personIds.length === 0) {
+      return {
+        totalNumberOfThreads: 0,
+        timelineThreads: [],
+      };
+    }
+
+    const messageThreads = await this.getMessagesFromPersonIds(
+      workspaceMemberId,
+      personIds,
+      page,
+      pageSize,
+    );
+
+    return messageThreads;
+  }
+}

packages/twenty-server/src/engine/core-modules/messaging/services/timeline-messaging.service.ts

@@ -0,0 +1,259 @@
+import { Injectable } from '@nestjs/common';
+
+import { Any, Not } from 'typeorm';
+
+import { TimelineThread } from 'src/engine/core-modules/messaging/dtos/timeline-thread.dto';
+import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
+import { MessageChannelVisibility } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
+import { MessageParticipantWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-participant.workspace-entity';
+import { MessageThreadWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-thread.workspace-entity';
+
+@Injectable()
+export class TimelineMessagingService {
+  constructor(private readonly twentyORMManager: TwentyORMManager) {}
+
+  public async getAndCountMessageThreads(
+    personIds: string[],
+    offset: number,
+    pageSize: number,
+  ): Promise<{
+    messageThreads: Omit<
+      TimelineThread,
+      | 'firstParticipant'
+      | 'lastTwoParticipants'
+      | 'participantCount'
+      | 'read'
+      | 'visibility'
+    >[];
+    totalNumberOfThreads: number;
+  }> {
+    const messageThreadRepository =
+      await this.twentyORMManager.getRepository<MessageThreadWorkspaceEntity>(
+        'messageThread',
+      );
+
+    const [messageThreadIds, totalNumberOfThreads] =
+      await messageThreadRepository.findAndCount({
+        select: {
+          id: true,
+        },
+        where: {
+          messages: {
+            messageParticipants: {
+              personId: Any(personIds),
+            },
+          },
+        },
+        skip: offset,
+        take: pageSize,
+      });
+
+    const messageThreads = await messageThreadRepository.find({
+      select: {
+        id: true,
+      },
+      where: {
+        id: Any(messageThreadIds.map((thread) => thread.id)),
+      },
+      order: {
+        messages: {
+          receivedAt: 'DESC',
+        },
+      },
+      relations: ['messages'],
+    });
+
+    return {
+      messageThreads: messageThreads.map((messageThread) => {
+        const lastMessage = messageThread.messages[0];
+        const firstMessage =
+          messageThread.messages[messageThread.messages.length - 1];
+
+        return {
+          id: messageThread.id,
+          subject: firstMessage.subject,
+          lastMessageBody: lastMessage.text,
+          lastMessageReceivedAt: lastMessage.receivedAt ?? new Date(),
+          numberOfMessagesInThread: messageThread.messages.length,
+        };
+      }),
+      totalNumberOfThreads,
+    };
+  }
+
+  public async getThreadParticipantsByThreadId(
+    messageThreadIds: string[],
+  ): Promise<{
+    [key: string]: MessageParticipantWorkspaceEntity[];
+  }> {
+    const messageParticipantRepository =
+      await this.twentyORMManager.getRepository<MessageParticipantWorkspaceEntity>(
+        'messageParticipant',
+      );
+    const threadParticipants = await messageParticipantRepository
+      .createQueryBuilder()
+      .select('messageParticipant')
+      .addSelect('message.messageThreadId')
+      .addSelect('message.receivedAt')
+      .leftJoinAndSelect('messageParticipant.person', 'person')
+      .leftJoinAndSelect(
+        'messageParticipant.workspaceMember',
+        'workspaceMember',
+      )
+      .leftJoin('messageParticipant.message', 'message')
+      .where('message.messageThreadId = ANY(:messageThreadIds)', {
+        messageThreadIds,
+      })
+      .andWhere('messageParticipant.role = :role', { role: 'from' })
+      .orderBy('message.messageThreadId')
+      .distinctOn(['message.messageThreadId', 'messageParticipant.handle'])
+      .getMany();
+
+    // This is because subqueries are not handled by twentyORM
+    const orderedThreadParticipants = threadParticipants.sort(
+      (a, b) =>
+        (a.message.receivedAt ?? new Date()).getTime() -
+        (b.message.receivedAt ?? new Date()).getTime(),
+    );
+
+    // This is because composite fields are not handled correctly by the ORM
+    const threadParticipantsWithCompositeFields = orderedThreadParticipants.map(
+      (threadParticipant) => ({
+        ...threadParticipant,
+        person: {
+          id: threadParticipant.person?.id,
+          name: {
+            //eslint-disable-next-line
+            //@ts-ignore
+            firstName: threadParticipant.person?.nameFirstName,
+            //eslint-disable-next-line
+            //@ts-ignore
+            lastName: threadParticipant.person?.nameLastName,
+          },
+          avatarUrl: threadParticipant.person?.avatarUrl,
+        },
+        workspaceMember: {
+          id: threadParticipant.workspaceMember?.id,
+          name: {
+            //eslint-disable-next-line
+            //@ts-ignore
+            firstName: threadParticipant.workspaceMember?.nameFirstName,
+            //eslint-disable-next-line
+            //@ts-ignore
+            lastName: threadParticipant.workspaceMember?.nameLastName,
+          },
+          avatarUrl: threadParticipant.workspaceMember?.avatarUrl,
+        },
+      }),
+    );
+
+    return threadParticipantsWithCompositeFields.reduce(
+      (threadParticipantsAcc, threadParticipant) => {
+        if (!threadParticipant.message.messageThreadId)
+          return threadParticipantsAcc;
+
+        if (!threadParticipantsAcc[threadParticipant.message.messageThreadId])
+          threadParticipantsAcc[threadParticipant.message.messageThreadId] = [];
+
+        threadParticipantsAcc[threadParticipant.message.messageThreadId].push(
+          threadParticipant,
+        );
+
+        return threadParticipantsAcc;
+      },
+      {},
+    );
+  }
+
+  public async getThreadVisibilityByThreadId(
+    messageThreadIds: string[],
+    workspaceMemberId: string,
+  ): Promise<{
+    [key: string]: MessageChannelVisibility;
+  }> {
+    const messageThreadRepository =
+      await this.twentyORMManager.getRepository<MessageThreadWorkspaceEntity>(
+        'messageThread',
+      );
+
+    const threadsWithoutWorkspaceMember = await messageThreadRepository.find({
+      select: {
+        id: true,
+      },
+      where: {
+        id: Any(messageThreadIds),
+        messages: {
+          messageChannelMessageAssociations: {
+            messageChannel: {
+              connectedAccount: {
+                accountOwnerId: Not(workspaceMemberId),
+              },
+            },
+          },
+        },
+      },
+    });
+
+    const threadIdsWithoutWorkspaceMember = threadsWithoutWorkspaceMember.map(
+      (thread) => thread.id,
+    );
+
+    const threadVisibility = await messageThreadRepository
+      .createQueryBuilder()
+      .select('messageThread.id', 'id')
+      .addSelect('messageChannel.visibility', 'visibility')
+      .leftJoin('messageThread.messages', 'message')
+      .leftJoin(
+        'message.messageChannelMessageAssociations',
+        'messageChannelMessageAssociation',
+      )
+      .leftJoin(
+        'messageChannelMessageAssociation.messageChannel',
+        'messageChannel',
+      )
+      .where('messageThread.id = ANY(:messageThreadIds)', {
+        messageThreadIds: threadIdsWithoutWorkspaceMember,
+      })
+      .getRawMany();
+
+    const visibilityValues = Object.values(MessageChannelVisibility);
+
+    const threadVisibilityByThreadIdForWhichWorkspaceMemberIsNotInParticipants:
+      | {
+          [key: string]: MessageChannelVisibility;
+        }
+      | undefined = threadVisibility?.reduce(
+      (threadVisibilityAcc, threadVisibility) => {
+        threadVisibilityAcc[threadVisibility.id] =
+          visibilityValues[
+            Math.max(
+              visibilityValues.indexOf(threadVisibility.visibility),
+              visibilityValues.indexOf(
+                threadVisibilityAcc[threadVisibility.id] ??
+                  MessageChannelVisibility.METADATA,
+              ),
+            )
+          ];
+
+        return threadVisibilityAcc;
+      },
+      {},
+    );
+
+    const threadVisibilityByThreadId: {
+      [key: string]: MessageChannelVisibility;
+    } = messageThreadIds.reduce((threadVisibilityAcc, messageThreadId) => {
+      // If the workspace member is not in the participants of the thread, use the visibility value from the query
+      threadVisibilityAcc[messageThreadId] =
+        threadIdsWithoutWorkspaceMember.includes(messageThreadId)
+          ? (threadVisibilityByThreadIdForWhichWorkspaceMemberIsNotInParticipants?.[
+              messageThreadId
+            ] ?? MessageChannelVisibility.METADATA)
+          : MessageChannelVisibility.SHARE_EVERYTHING;
+
+      return threadVisibilityAcc;
+    }, {});
+
+    return threadVisibilityByThreadId;
+  }
+}