[messaging] remove v2 feature flag (#4845)
## Context We are now removing Messaging V2 feature flag to use it everywhere. ## Implementation - renaming FetchWorkspaceMessagesCommandsModule to MessagingCommandModule to make it more generic since it it hosts all commands related to the messaging module - creating a crons folder inside commands and jobs crons should be named with xxx.cron.command.ts instead of xxx.command.ts. Same for jobs, jobs should be named with xxx.cron.job.ts. In a future PR we should make sure those CronJobs implement a CronJob interface since it's a bit different (a CronJob does not contain a payload compared to a Job) - Cron commands have been renamed to "cron:$module:command" so `fetch-all-workspaces-messages-from-cache:cron:start` has been renamed to `cron:messaging:gmail-fetch-messages-from-cache`. Also having to create a command to stop the cron is a bit painful to maintain so I removed them for now, this can be easily done manually with pg-boss or bull-mq - Removing full-sync and partial-sync commands as they were there for testing only, we might put them back at some point but we will have to adapt the code anyway. - Feature flag has been removed from the MessageChannel standard object to make sure those new columns are created during the next sync-metadata
This commit is contained in:
Weiko
@ -0,0 +1,67 @@
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
|
||||
import { Repository, In } from 'typeorm';
|
||||
|
||||
import { MessageQueueJob } from 'src/engine/integrations/message-queue/interfaces/message-queue-job.interface';
|
||||
|
||||
import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
|
||||
import { DataSourceEntity } from 'src/engine/metadata-modules/data-source/data-source.entity';
|
||||
import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
|
||||
import { MessageChannelRepository } from 'src/modules/messaging/repositories/message-channel.repository';
|
||||
import { MessageChannelObjectMetadata } from 'src/modules/messaging/standard-objects/message-channel.object-metadata';
|
||||
import { GmailFetchMessageContentFromCacheService } from 'src/modules/messaging/services/gmail-fetch-message-content-from-cache/gmail-fetch-message-content-from-cache.service';
|
||||
|
||||
@Injectable()
|
||||
export class FetchAllMessagesFromCacheCronJob
|
||||
implements MessageQueueJob<undefined>
|
||||
{
|
||||
private readonly logger = new Logger(FetchAllMessagesFromCacheCronJob.name);
|
||||
|
||||
constructor(
|
||||
@InjectRepository(Workspace, 'core')
|
||||
private readonly workspaceRepository: Repository<Workspace>,
|
||||
@InjectRepository(DataSourceEntity, 'metadata')
|
||||
private readonly dataSourceRepository: Repository<DataSourceEntity>,
|
||||
@InjectObjectMetadataRepository(MessageChannelObjectMetadata)
|
||||
private readonly messageChannelRepository: MessageChannelRepository,
|
||||
private readonly gmailFetchMessageContentFromCacheService: GmailFetchMessageContentFromCacheService,
|
||||
) {}
|
||||
|
||||
async handle(): Promise<void> {
|
||||
const workspaceIds = (
|
||||
await this.workspaceRepository.find({
|
||||
where: {
|
||||
subscriptionStatus: 'active',
|
||||
},
|
||||
select: ['id'],
|
||||
})
|
||||
).map((workspace) => workspace.id);
|
||||
|
||||
const dataSources = await this.dataSourceRepository.find({
|
||||
where: {
|
||||
workspaceId: In(workspaceIds),
|
||||
},
|
||||
});
|
||||
|
||||
const workspaceIdsWithDataSources = new Set(
|
||||
dataSources.map((dataSource) => dataSource.workspaceId),
|
||||
);
|
||||
|
||||
for (const workspaceId of workspaceIdsWithDataSources) {
|
||||
await this.fetchWorkspaceMessages(workspaceId);
|
||||
}
|
||||
}
|
||||
|
||||
private async fetchWorkspaceMessages(workspaceId: string): Promise<void> {
|
||||
const messageChannels =
|
||||
await this.messageChannelRepository.getAll(workspaceId);
|
||||
|
||||
for (const messageChannel of messageChannels) {
|
||||
await this.gmailFetchMessageContentFromCacheService.fetchMessageContentFromCache(
|
||||
workspaceId,
|
||||
messageChannel.connectedAccountId,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,83 @@
|
||||
import { Inject, Injectable, Logger } from '@nestjs/common';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
|
||||
import { Repository, In } from 'typeorm';
|
||||
|
||||
import { MessageQueueJob } from 'src/engine/integrations/message-queue/interfaces/message-queue-job.interface';
|
||||
|
||||
import { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
|
||||
import { MessageQueueService } from 'src/engine/integrations/message-queue/services/message-queue.service';
|
||||
import { ConnectedAccountRepository } from 'src/modules/connected-account/repositories/connected-account.repository';
|
||||
import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
|
||||
import { DataSourceEntity } from 'src/engine/metadata-modules/data-source/data-source.entity';
|
||||
import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
|
||||
import { ConnectedAccountObjectMetadata } from 'src/modules/connected-account/standard-objects/connected-account.object-metadata';
|
||||
import {
|
||||
GmailPartialSyncV2Job as GmailPartialSyncV2Job,
|
||||
GmailPartialSyncV2JobData as GmailPartialSyncV2JobData,
|
||||
} from 'src/modules/messaging/jobs/gmail-partial-sync-v2.job';
|
||||
|
||||
@Injectable()
|
||||
export class GmailPartialSyncCronJob implements MessageQueueJob<undefined> {
|
||||
private readonly logger = new Logger(GmailPartialSyncCronJob.name);
|
||||
|
||||
constructor(
|
||||
@InjectRepository(Workspace, 'core')
|
||||
private readonly workspaceRepository: Repository<Workspace>,
|
||||
@InjectRepository(DataSourceEntity, 'metadata')
|
||||
private readonly dataSourceRepository: Repository<DataSourceEntity>,
|
||||
@Inject(MessageQueue.messagingQueue)
|
||||
private readonly messageQueueService: MessageQueueService,
|
||||
@InjectObjectMetadataRepository(ConnectedAccountObjectMetadata)
|
||||
private readonly connectedAccountRepository: ConnectedAccountRepository,
|
||||
) {}
|
||||
|
||||
async handle(): Promise<void> {
|
||||
const workspaceIds = (
|
||||
await this.workspaceRepository.find({
|
||||
where: {
|
||||
subscriptionStatus: 'active',
|
||||
},
|
||||
select: ['id'],
|
||||
})
|
||||
).map((workspace) => workspace.id);
|
||||
|
||||
const dataSources = await this.dataSourceRepository.find({
|
||||
where: {
|
||||
workspaceId: In(workspaceIds),
|
||||
},
|
||||
});
|
||||
|
||||
const workspaceIdsWithDataSources = new Set(
|
||||
dataSources.map((dataSource) => dataSource.workspaceId),
|
||||
);
|
||||
|
||||
for (const workspaceId of workspaceIdsWithDataSources) {
|
||||
await this.enqueuePartialSyncs(workspaceId);
|
||||
}
|
||||
}
|
||||
|
||||
private async enqueuePartialSyncs(workspaceId: string): Promise<void> {
|
||||
try {
|
||||
const connectedAccounts =
|
||||
await this.connectedAccountRepository.getAll(workspaceId);
|
||||
|
||||
for (const connectedAccount of connectedAccounts) {
|
||||
await this.messageQueueService.add<GmailPartialSyncV2JobData>(
|
||||
GmailPartialSyncV2Job.name,
|
||||
{
|
||||
workspaceId,
|
||||
connectedAccountId: connectedAccount.id,
|
||||
},
|
||||
);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error(
|
||||
`Error while fetching workspace messages for workspace ${workspaceId}`,
|
||||
);
|
||||
this.logger.error(error);
|
||||
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -1,40 +0,0 @@
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
|
||||
import { MessageQueueJob } from 'src/engine/integrations/message-queue/interfaces/message-queue-job.interface';
|
||||
|
||||
import { CalendarEventCleanerService } from 'src/modules/calendar/services/calendar-event-cleaner/calendar-event-cleaner.service';
|
||||
|
||||
export type DeleteConnectedAccountAssociatedCalendarDataJobData = {
|
||||
workspaceId: string;
|
||||
connectedAccountId: string;
|
||||
};
|
||||
|
||||
@Injectable()
|
||||
export class DeleteConnectedAccountAssociatedCalendarDataJob
|
||||
implements
|
||||
MessageQueueJob<DeleteConnectedAccountAssociatedCalendarDataJobData>
|
||||
{
|
||||
private readonly logger = new Logger(
|
||||
DeleteConnectedAccountAssociatedCalendarDataJob.name,
|
||||
);
|
||||
|
||||
constructor(
|
||||
private readonly calendarEventCleanerService: CalendarEventCleanerService,
|
||||
) {}
|
||||
|
||||
async handle(
|
||||
data: DeleteConnectedAccountAssociatedCalendarDataJobData,
|
||||
): Promise<void> {
|
||||
this.logger.log(
|
||||
`Deleting connected account ${data.connectedAccountId} associated calendar data in workspace ${data.workspaceId}`,
|
||||
);
|
||||
|
||||
await this.calendarEventCleanerService.cleanWorkspaceCalendarEvents(
|
||||
data.workspaceId,
|
||||
);
|
||||
|
||||
this.logger.log(
|
||||
`Deleted connected account ${data.connectedAccountId} associated calendar data in workspace ${data.workspaceId}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
@ -1,49 +0,0 @@
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
|
||||
import { MessageQueueJob } from 'src/engine/integrations/message-queue/interfaces/message-queue-job.interface';
|
||||
|
||||
import { GoogleAPIRefreshAccessTokenService } from 'src/modules/connected-account/services/google-api-refresh-access-token/google-api-refresh-access-token.service';
|
||||
import { GmailFullSyncService } from 'src/modules/messaging/services/gmail-full-sync/gmail-full-sync.service';
|
||||
|
||||
export type GmailFullSyncJobData = {
|
||||
workspaceId: string;
|
||||
connectedAccountId: string;
|
||||
nextPageToken?: string;
|
||||
};
|
||||
|
||||
@Injectable()
|
||||
export class GmailFullSyncJob implements MessageQueueJob<GmailFullSyncJobData> {
|
||||
private readonly logger = new Logger(GmailFullSyncJob.name);
|
||||
|
||||
constructor(
|
||||
private readonly googleAPIsRefreshAccessTokenService: GoogleAPIRefreshAccessTokenService,
|
||||
private readonly gmailFullSyncService: GmailFullSyncService,
|
||||
) {}
|
||||
|
||||
async handle(data: GmailFullSyncJobData): Promise<void> {
|
||||
this.logger.log(
|
||||
`gmail full-sync for workspace ${data.workspaceId} and account ${
|
||||
data.connectedAccountId
|
||||
} ${data.nextPageToken ? `and ${data.nextPageToken} pageToken` : ''}`,
|
||||
);
|
||||
|
||||
try {
|
||||
await this.googleAPIsRefreshAccessTokenService.refreshAndSaveAccessToken(
|
||||
data.workspaceId,
|
||||
data.connectedAccountId,
|
||||
);
|
||||
} catch (e) {
|
||||
this.logger.error(
|
||||
`Error refreshing access token for connected account ${data.connectedAccountId} in workspace ${data.workspaceId}`,
|
||||
e,
|
||||
);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
await this.gmailFullSyncService.fetchConnectedAccountThreads(
|
||||
data.workspaceId,
|
||||
data.connectedAccountId,
|
||||
);
|
||||
}
|
||||
}
|
||||
@ -1,48 +0,0 @@
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
|
||||
import { MessageQueueJob } from 'src/engine/integrations/message-queue/interfaces/message-queue-job.interface';
|
||||
|
||||
import { GoogleAPIRefreshAccessTokenService } from 'src/modules/connected-account/services/google-api-refresh-access-token/google-api-refresh-access-token.service';
|
||||
import { GmailPartialSyncService } from 'src/modules/messaging/services/gmail-partial-sync/gmail-partial-sync.service';
|
||||
|
||||
export type GmailPartialSyncJobData = {
|
||||
workspaceId: string;
|
||||
connectedAccountId: string;
|
||||
};
|
||||
|
||||
@Injectable()
|
||||
export class GmailPartialSyncJob
|
||||
implements MessageQueueJob<GmailPartialSyncJobData>
|
||||
{
|
||||
private readonly logger = new Logger(GmailPartialSyncJob.name);
|
||||
|
||||
constructor(
|
||||
private readonly googleAPIsRefreshAccessTokenService: GoogleAPIRefreshAccessTokenService,
|
||||
private readonly gmailPartialSyncService: GmailPartialSyncService,
|
||||
) {}
|
||||
|
||||
async handle(data: GmailPartialSyncJobData): Promise<void> {
|
||||
this.logger.log(
|
||||
`gmail partial-sync for workspace ${data.workspaceId} and account ${data.connectedAccountId}`,
|
||||
);
|
||||
|
||||
try {
|
||||
await this.googleAPIsRefreshAccessTokenService.refreshAndSaveAccessToken(
|
||||
data.workspaceId,
|
||||
data.connectedAccountId,
|
||||
);
|
||||
} catch (e) {
|
||||
this.logger.error(
|
||||
`Error refreshing access token for connected account ${data.connectedAccountId} in workspace ${data.workspaceId}`,
|
||||
e,
|
||||
);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
await this.gmailPartialSyncService.fetchConnectedAccountThreads(
|
||||
data.workspaceId,
|
||||
data.connectedAccountId,
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user