admin/twenty
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix redis concurrency issue in health metrics + remove ongoing status count (#10717)

Browse Source 
### Context
For calendar and message sync job health monitoring, we used to
increment a counter in redis cache which could lead to concurrency
issue.

### Solution
- Update to a set structure in place of counter + use sAdd redis method
which is atomic
- Each minute another counter was incremented on a new cache key ->
Update to a 15s window
- Remove ONGOING status not needed. We only need status at job end (or
fail).


###  Potential improvements
- Check for cache key existence before fetching data to avoid useless
call to redis ?

closes https://github.com/twentyhq/twenty/issues/10070
This commit is contained in:
Etienne
2025-03-07 14:41:46 +01:00
committed by GitHub
parent 02a085df4f
commit 96035f0ccf
11 changed files with 203 additions and 251 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/twenty-front/src/modules/settings/admin-panel/health-status/components/SettingsAdminHealthAccountSyncCountersTable.tsx
View File

@ -34,8 +34,8 @@ export const SettingsAdminHealthAccountSyncCountersTable = ({
<TableCell align="right">{details.counters.NOT_SYNCED}</TableCell>
</TableRow>
<TableRow>
<TableCell>Sync Ongoing</TableCell>
<TableCell align="right">{details.counters.ONGOING}</TableCell>
<TableCell>Active Sync</TableCell>
<TableCell align="right">{details.counters.ACTIVE}</TableCell>
</TableRow>
<TableRow>
<TableCell>Total Jobs</TableCell>

37
packages/twenty-server/src/engine/core-modules/cache-storage/services/cache-storage.service.ts
View File

@ -25,25 +25,42 @@ export class CacheStorageService {
return this.cache.del(`${this.namespace}:${key}`);
}
async setAdd(key: string, value: string[]) {
async setAdd(key: string, value: string[], ttl?: number) {
if (value.length === 0) {
return;
}
if (this.isRedisCache()) {
return (this.cache as RedisCache).store.client.sAdd(
await (this.cache as RedisCache).store.client.sAdd(
`${this.namespace}:${key}`,
value,
);
if (ttl) {
await (this.cache as RedisCache).store.client.expire(
`${this.namespace}:${key}`,
ttl / 1000,
);
}
return;
}
this.get(key).then((res: string[]) => {
if (res) {
this.set(key, [...res, ...value]);
this.set(key, [...res, ...value], ttl);
} else {
this.set(key, value);
this.set(key, value, ttl);
}
});
}
async countAllSetMembers(cacheKeys: string[]) {
return (
await Promise.all(cacheKeys.map((key) => this.getSetLength(key) || 0))
).reduce((acc, setLength) => acc + setLength, 0);
}
async setPop(key: string, size = 1) {
if (this.isRedisCache()) {
return (this.cache as RedisCache).store.client.sPop(
@ -63,6 +80,18 @@ export class CacheStorageService {
});
}
async getSetLength(key: string) {
if (this.isRedisCache()) {
return await (this.cache as RedisCache).store.client.sCard(
`${this.namespace}:${key}`,
);
}
return this.get(key).then((res: string[]) => {
return res.length;
});
}
async flush() {
return this.cache.reset();
}

2
packages/twenty-server/src/engine/core-modules/captcha/captcha.guard.ts
View File

@ -26,7 +26,7 @@ export class CaptchaGuard implements CanActivate {
if (result.success) {
return true;
} else {
await this.healthCacheService.incrementInvalidCaptchaCounter();
await this.healthCacheService.updateInvalidCaptchaCache(token);
throw new BadRequestException(
'Invalid Captcha, please try another device',

2
packages/twenty-server/src/engine/core-modules/environment/environment-variables.ts
View File

@ -959,7 +959,7 @@ export class EnvironmentVariables {
@IsNumber()
@CastToPositiveNumber()
@IsOptional()
HEALTH_MONITORING_TIME_WINDOW_IN_MINUTES = 5;
HEALTH_METRICS_TIME_WINDOW_IN_MINUTES = 5;
@EnvironmentVariablesMetadata({
group: EnvironmentVariablesGroup.Other,

11
packages/twenty-server/src/engine/core-modules/health/controllers/metrics.controller.ts
View File

@ -1,14 +1,17 @@
import { Controller, Get } from '@nestjs/common';
import { HealthCacheService } from 'src/engine/core-modules/health/health-cache.service';
import { HealthCounterCacheKeys } from 'src/engine/core-modules/health/types/health-counter-cache-keys.type';
@Controller('metricsz')
@Controller('metrics')
export class MetricsController {
constructor(private readonly healthCacheService: HealthCacheService) {}
@Get('/message-channel-sync-job-by-status-counter')
getMessageChannelSyncJobByStatusCounter() {
return this.healthCacheService.getMessageChannelSyncJobByStatusCounter();
return this.healthCacheService.countChannelSyncJobByStatus(
HealthCounterCacheKeys.MessageChannelSyncJobByStatus,
);
}
@Get('/invalid-captcha-counter')
@ -18,6 +21,8 @@ export class MetricsController {
@Get('/calendar-channel-sync-job-by-status-counter')
getCalendarChannelSyncJobByStatusCounter() {
return this.healthCacheService.getCalendarChannelSyncJobByStatusCounter();
return this.healthCacheService.countChannelSyncJobByStatus(
HealthCounterCacheKeys.CalendarEventSyncJobByStatus,
);
}
}

223
packages/twenty-server/src/engine/core-modules/health/health-cache.service.ts
View File

@ -9,9 +9,11 @@ import { HealthCounterCacheKeys } from 'src/engine/core-modules/health/types/hea
import { CalendarChannelSyncStatus } from 'src/modules/calendar/common/standard-objects/calendar-channel.workspace-entity';
import { MessageChannelSyncStatus } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
const CACHE_BUCKET_DURATION_MS = 15000; // 15 seconds window for each cache bucket
@Injectable()
export class HealthCacheService {
private readonly healthMonitoringTimeWindowInMinutes: number;
private readonly healthMetricsTimeWindowInMinutes: number;
private readonly healthCacheTtl: number;
constructor(
@ -19,168 +21,119 @@ export class HealthCacheService {
private readonly cacheStorage: CacheStorageService,
private readonly environmentService: EnvironmentService,
) {
this.healthMonitoringTimeWindowInMinutes = this.environmentService.get(
'HEALTH_MONITORING_TIME_WINDOW_IN_MINUTES',
this.healthMetricsTimeWindowInMinutes = this.environmentService.get(
'HEALTH_METRICS_TIME_WINDOW_IN_MINUTES',
);
this.healthCacheTtl = this.healthMetricsTimeWindowInMinutes * 60000 * 2;
}
private getCacheBucketStartTimestamp(timestamp: number): number {
return (
Math.floor(timestamp / CACHE_BUCKET_DURATION_MS) *
CACHE_BUCKET_DURATION_MS
);
this.healthCacheTtl = this.healthMonitoringTimeWindowInMinutes * 60000 * 2;
}
private getCacheKeyWithTimestamp(key: string, timestamp?: number): string {
const minuteTimestamp = timestamp ?? Math.floor(Date.now() / 60000) * 60000;
const currentIntervalTimestamp =
timestamp ?? this.getCacheBucketStartTimestamp(Date.now());
return `${key}:${minuteTimestamp}`;
return `${key}:${currentIntervalTimestamp}`;
}
private getLastXMinutesTimestamps(minutes: number): number[] {
const currentMinuteTimestamp = Math.floor(Date.now() / 60000) * 60000;
private getLastCacheBucketStartTimestampsFromDate(
cacheBucketsCount: number,
date: number = Date.now(),
): number[] {
const currentIntervalTimestamp = this.getCacheBucketStartTimestamp(date);
return Array.from(
{ length: minutes },
(_, i) => currentMinuteTimestamp - i * 60000,
{ length: cacheBucketsCount },
(_, i) => currentIntervalTimestamp - i * CACHE_BUCKET_DURATION_MS,
);
}
async incrementMessageChannelSyncJobByStatusCounter(
status: MessageChannelSyncStatus,
increment: number,
async updateMessageOrCalendarChannelSyncJobByStatusCache(
key: HealthCounterCacheKeys,
status: MessageChannelSyncStatus | CalendarChannelSyncStatus,
messageChannelIds: string[],
) {
const cacheKey = this.getCacheKeyWithTimestamp(
HealthCounterCacheKeys.MessageChannelSyncJobByStatus,
);
const currentCounter =
await this.cacheStorage.get<AccountSyncJobByStatusCounter>(cacheKey);
const updatedCounter = {
...(currentCounter || {}),
[status]: (currentCounter?.[status] || 0) + increment,
};
return await this.cacheStorage.set(
cacheKey,
updatedCounter,
return await this.cacheStorage.setAdd(
this.getCacheKeyWithTimestamp(`${key}:${status}`),
messageChannelIds,
this.healthCacheTtl,
);
}
async getMessageChannelSyncJobByStatusCounter() {
const cacheKeys = this.getLastXMinutesTimestamps(
this.healthMonitoringTimeWindowInMinutes,
).map((timestamp) =>
this.getCacheKeyWithTimestamp(
HealthCounterCacheKeys.MessageChannelSyncJobByStatus,
timestamp,
),
);
const aggregatedCounter = Object.fromEntries(
Object.values(MessageChannelSyncStatus).map((status) => [status, 0]),
);
for (const key of cacheKeys) {
const counter =
await this.cacheStorage.get<AccountSyncJobByStatusCounter>(key);
if (!counter) continue;
for (const [status, count] of Object.entries(counter) as [
MessageChannelSyncStatus,
number,
][]) {
aggregatedCounter[status] += count;
}
async countChannelSyncJobByStatus(
key:
| HealthCounterCacheKeys.MessageChannelSyncJobByStatus
| HealthCounterCacheKeys.CalendarEventSyncJobByStatus,
timeWindowInSeconds: number = this.healthMetricsTimeWindowInMinutes * 60,
): Promise<AccountSyncJobByStatusCounter> {
if ((timeWindowInSeconds * 1000) % CACHE_BUCKET_DURATION_MS !== 0) {
throw new Error(
`Time window must be divisible by ${CACHE_BUCKET_DURATION_MS}`,
);
}
return aggregatedCounter;
const now = Date.now();
const countByStatus = {} as AccountSyncJobByStatusCounter;
const statuses =
key === HealthCounterCacheKeys.MessageChannelSyncJobByStatus
? Object.values(MessageChannelSyncStatus).filter(
(status) => status !== MessageChannelSyncStatus.ONGOING,
)
: Object.values(CalendarChannelSyncStatus).filter(
(status) => status !== CalendarChannelSyncStatus.ONGOING,
);
const cacheBuckets =
timeWindowInSeconds / (CACHE_BUCKET_DURATION_MS / 1000);
for (const status of statuses) {
const cacheKeys = this.computeTimeStampedCacheKeys(
`${key}:${status}`,
cacheBuckets,
now,
);
const channelIdsCount =
await this.cacheStorage.countAllSetMembers(cacheKeys);
countByStatus[status] = channelIdsCount;
}
return countByStatus;
}
async incrementInvalidCaptchaCounter() {
const cacheKey = this.getCacheKeyWithTimestamp(
HealthCounterCacheKeys.InvalidCaptcha,
);
computeTimeStampedCacheKeys(
key: string,
cacheBucketsCount: number,
date: number = Date.now(),
) {
return this.getLastCacheBucketStartTimestampsFromDate(
cacheBucketsCount,
date,
).map((timestamp) => this.getCacheKeyWithTimestamp(key, timestamp));
}
const currentCounter = await this.cacheStorage.get<number>(cacheKey);
const updatedCounter = (currentCounter || 0) + 1;
return await this.cacheStorage.set(
cacheKey,
updatedCounter,
async updateInvalidCaptchaCache(captchaToken: string) {
return await this.cacheStorage.setAdd(
this.getCacheKeyWithTimestamp(HealthCounterCacheKeys.InvalidCaptcha),
[captchaToken],
this.healthCacheTtl,
);
}
async getInvalidCaptchaCounter() {
const cacheKeys = this.getLastXMinutesTimestamps(
this.healthMonitoringTimeWindowInMinutes,
).map((timestamp) =>
this.getCacheKeyWithTimestamp(
async getInvalidCaptchaCounter(
timeWindowInSeconds: number = this.healthMetricsTimeWindowInMinutes * 60,
) {
return await this.cacheStorage.countAllSetMembers(
this.computeTimeStampedCacheKeys(
HealthCounterCacheKeys.InvalidCaptcha,
timestamp,
timeWindowInSeconds / (CACHE_BUCKET_DURATION_MS / 1000),
),
);
let aggregatedCounter = 0;
for (const key of cacheKeys) {
const counter = await this.cacheStorage.get<number>(key);
aggregatedCounter += counter || 0;
}
return aggregatedCounter;
}
async incrementCalendarChannelSyncJobByStatusCounter(
status: CalendarChannelSyncStatus,
increment: number,
) {
const cacheKey = this.getCacheKeyWithTimestamp(
HealthCounterCacheKeys.CalendarEventSyncJobByStatus,
);
const currentCounter =
await this.cacheStorage.get<AccountSyncJobByStatusCounter>(cacheKey);
const updatedCounter = {
...(currentCounter || {}),
[status]: (currentCounter?.[status] || 0) + increment,
};
return await this.cacheStorage.set(
cacheKey,
updatedCounter,
this.healthCacheTtl,
);
}
async getCalendarChannelSyncJobByStatusCounter() {
const cacheKeys = this.getLastXMinutesTimestamps(
this.healthMonitoringTimeWindowInMinutes,
).map((timestamp) =>
this.getCacheKeyWithTimestamp(
HealthCounterCacheKeys.CalendarEventSyncJobByStatus,
timestamp,
),
);
const aggregatedCounter = Object.fromEntries(
Object.values(CalendarChannelSyncStatus).map((status) => [status, 0]),
);
for (const key of cacheKeys) {
const counter =
await this.cacheStorage.get<AccountSyncJobByStatusCounter>(key);
if (!counter) continue;
for (const [status, count] of Object.entries(counter) as [
CalendarChannelSyncStatus,
number,
][]) {
aggregatedCounter[status] += count;
}
}
return aggregatedCounter;
}
}

121
packages/twenty-server/src/engine/core-modules/health/indicators/__tests__/connected-account.health.spec.ts
View File

@ -16,8 +16,7 @@ describe('ConnectedAccountHealth', () => {
beforeEach(async () => {
healthCacheService = {
getMessageChannelSyncJobByStatusCounter: jest.fn(),
getCalendarChannelSyncJobByStatusCounter: jest.fn(),
countChannelSyncJobByStatus: jest.fn(),
} as any;
healthIndicatorService = {
@ -65,22 +64,17 @@ describe('ConnectedAccountHealth', () => {
describe('message sync health', () => {
it('should return up status when no message sync jobs are present', async () => {
healthCacheService.getMessageChannelSyncJobByStatusCounter.mockResolvedValue(
{
healthCacheService.countChannelSyncJobByStatus
.mockResolvedValueOnce({
[MessageChannelSyncStatus.NOT_SYNCED]: 0,
[MessageChannelSyncStatus.ONGOING]: 0,
[MessageChannelSyncStatus.ACTIVE]: 0,
[MessageChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS]: 0,
[MessageChannelSyncStatus.FAILED_UNKNOWN]: 0,
},
);
healthCacheService.getCalendarChannelSyncJobByStatusCounter.mockResolvedValue(
{
})
.mockResolvedValueOnce({
[CalendarChannelSyncStatus.NOT_SYNCED]: 0,
[CalendarChannelSyncStatus.ACTIVE]: 0,
},
);
});
const result = await service.isHealthy();
@ -98,22 +92,17 @@ describe('ConnectedAccountHealth', () => {
});
it(`should return down status when message sync failure rate is above ${METRICS_FAILURE_RATE_THRESHOLD}%`, async () => {
healthCacheService.getMessageChannelSyncJobByStatusCounter.mockResolvedValue(
{
healthCacheService.countChannelSyncJobByStatus
.mockResolvedValueOnce({
[MessageChannelSyncStatus.NOT_SYNCED]: 0,
[MessageChannelSyncStatus.ONGOING]: 1,
[MessageChannelSyncStatus.ACTIVE]: 1,
[MessageChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS]: 2,
[MessageChannelSyncStatus.FAILED_UNKNOWN]: 2,
},
);
healthCacheService.getCalendarChannelSyncJobByStatusCounter.mockResolvedValue(
{
})
.mockResolvedValueOnce({
[CalendarChannelSyncStatus.NOT_SYNCED]: 0,
[CalendarChannelSyncStatus.ACTIVE]: 1,
},
);
});
const result = await service.isHealthy();
@ -127,28 +116,23 @@ describe('ConnectedAccountHealth', () => {
);
expect(
result.connectedAccount.details.messageSync.details.failureRate,
).toBe(33.33);
).toBe(40);
});
});
describe('calendar sync health', () => {
it('should return up status when no calendar sync jobs are present', async () => {
healthCacheService.getMessageChannelSyncJobByStatusCounter.mockResolvedValue(
{
healthCacheService.countChannelSyncJobByStatus
.mockResolvedValueOnce({
[MessageChannelSyncStatus.NOT_SYNCED]: 0,
[MessageChannelSyncStatus.ACTIVE]: 0,
},
);
healthCacheService.getCalendarChannelSyncJobByStatusCounter.mockResolvedValue(
{
})
.mockResolvedValueOnce({
[CalendarChannelSyncStatus.NOT_SYNCED]: 0,
[CalendarChannelSyncStatus.ONGOING]: 0,
[CalendarChannelSyncStatus.ACTIVE]: 0,
[CalendarChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS]: 0,
[CalendarChannelSyncStatus.FAILED_UNKNOWN]: 0,
},
);
});
const result = await service.isHealthy();
@ -166,22 +150,17 @@ describe('ConnectedAccountHealth', () => {
});
it(`should return down status when calendar sync failure rate is above ${METRICS_FAILURE_RATE_THRESHOLD}%`, async () => {
healthCacheService.getMessageChannelSyncJobByStatusCounter.mockResolvedValue(
{
healthCacheService.countChannelSyncJobByStatus
.mockResolvedValueOnce({
[MessageChannelSyncStatus.NOT_SYNCED]: 0,
[MessageChannelSyncStatus.ACTIVE]: 1,
},
);
healthCacheService.getCalendarChannelSyncJobByStatusCounter.mockResolvedValue(
{
})
.mockResolvedValueOnce({
[CalendarChannelSyncStatus.NOT_SYNCED]: 0,
[CalendarChannelSyncStatus.ONGOING]: 1,
[CalendarChannelSyncStatus.ACTIVE]: 1,
[CalendarChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS]: 2,
[CalendarChannelSyncStatus.FAILED_UNKNOWN]: 2,
},
);
});
const result = await service.isHealthy();
@ -195,25 +174,22 @@ describe('ConnectedAccountHealth', () => {
);
expect(
result.connectedAccount.details.calendarSync.details.failureRate,
).toBe(33.33);
).toBe(40);
});
});
describe('timeout handling', () => {
it('should handle message sync timeout', async () => {
healthCacheService.getMessageChannelSyncJobByStatusCounter.mockImplementationOnce(
() =>
healthCacheService.countChannelSyncJobByStatus
.mockResolvedValueOnce(
new Promise((resolve) =>
setTimeout(resolve, HEALTH_INDICATORS_TIMEOUT + 100),
),
);
healthCacheService.getCalendarChannelSyncJobByStatusCounter.mockResolvedValue(
{
)
.mockResolvedValueOnce({
[CalendarChannelSyncStatus.NOT_SYNCED]: 0,
[CalendarChannelSyncStatus.ACTIVE]: 1,
},
);
});
const healthCheckPromise = service.isHealthy();
@ -231,19 +207,16 @@ describe('ConnectedAccountHealth', () => {
});
it('should handle calendar sync timeout', async () => {
healthCacheService.getMessageChannelSyncJobByStatusCounter.mockResolvedValue(
{
healthCacheService.countChannelSyncJobByStatus
.mockResolvedValueOnce({
[MessageChannelSyncStatus.NOT_SYNCED]: 0,
[MessageChannelSyncStatus.ACTIVE]: 1,
},
);
healthCacheService.getCalendarChannelSyncJobByStatusCounter.mockImplementationOnce(
() =>
})
.mockResolvedValueOnce(
new Promise((resolve) =>
setTimeout(resolve, HEALTH_INDICATORS_TIMEOUT + 100),
),
);
);
const healthCheckPromise = service.isHealthy();
@ -263,21 +236,17 @@ describe('ConnectedAccountHealth', () => {
describe('combined health check', () => {
it('should return combined status with both checks healthy', async () => {
healthCacheService.getMessageChannelSyncJobByStatusCounter.mockResolvedValue(
{
healthCacheService.countChannelSyncJobByStatus
.mockResolvedValueOnce({
[MessageChannelSyncStatus.NOT_SYNCED]: 0,
[MessageChannelSyncStatus.ACTIVE]: 8,
[MessageChannelSyncStatus.FAILED_UNKNOWN]: 1,
},
);
healthCacheService.getCalendarChannelSyncJobByStatusCounter.mockResolvedValue(
{
})
.mockResolvedValueOnce({
[CalendarChannelSyncStatus.NOT_SYNCED]: 0,
[CalendarChannelSyncStatus.ACTIVE]: 8,
[CalendarChannelSyncStatus.FAILED_UNKNOWN]: 1,
},
);
});
const result = await service.isHealthy();
@ -287,21 +256,17 @@ describe('ConnectedAccountHealth', () => {
});
it('should return down status when both syncs fail', async () => {
healthCacheService.getMessageChannelSyncJobByStatusCounter.mockResolvedValue(
{
healthCacheService.countChannelSyncJobByStatus
.mockResolvedValueOnce({
[MessageChannelSyncStatus.NOT_SYNCED]: 0,
[MessageChannelSyncStatus.ACTIVE]: 1,
[MessageChannelSyncStatus.FAILED_UNKNOWN]: 2,
},
);
healthCacheService.getCalendarChannelSyncJobByStatusCounter.mockResolvedValue(
{
})
.mockResolvedValueOnce({
[CalendarChannelSyncStatus.NOT_SYNCED]: 0,
[CalendarChannelSyncStatus.ACTIVE]: 1,
[CalendarChannelSyncStatus.FAILED_UNKNOWN]: 2,
},
);
});
const result = await service.isHealthy();

9
packages/twenty-server/src/engine/core-modules/health/indicators/connected-account.health.ts
View File

@ -7,6 +7,7 @@ import {
import { HEALTH_ERROR_MESSAGES } from 'src/engine/core-modules/health/constants/health-error-messages.constants';
import { METRICS_FAILURE_RATE_THRESHOLD } from 'src/engine/core-modules/health/constants/metrics-failure-rate-threshold.const';
import { HealthCacheService } from 'src/engine/core-modules/health/health-cache.service';
import { HealthCounterCacheKeys } from 'src/engine/core-modules/health/types/health-counter-cache-keys.type';
import { withHealthCheckTimeout } from 'src/engine/core-modules/health/utils/health-check-timeout.util';
@Injectable()
@ -21,7 +22,9 @@ export class ConnectedAccountHealth {
try {
const counters = await withHealthCheckTimeout(
this.healthCacheService.getMessageChannelSyncJobByStatusCounter(),
this.healthCacheService.countChannelSyncJobByStatus(
HealthCounterCacheKeys.MessageChannelSyncJobByStatus,
),
HEALTH_ERROR_MESSAGES.MESSAGE_SYNC_TIMEOUT,
);
@ -70,7 +73,9 @@ export class ConnectedAccountHealth {
try {
const counters = await withHealthCheckTimeout(
this.healthCacheService.getCalendarChannelSyncJobByStatusCounter(),
this.healthCacheService.countChannelSyncJobByStatus(
HealthCounterCacheKeys.CalendarEventSyncJobByStatus,
),
HEALTH_ERROR_MESSAGES.CALENDAR_SYNC_TIMEOUT,
);

3
packages/twenty-server/src/engine/core-modules/health/types/account-sync-metrics.types.ts
View File

@ -5,9 +5,6 @@ export class AccountSyncJobByStatusCounter {
@Field(() => Number, { nullable: true })
NOT_SYNCED?: number;
@Field(() => Number, { nullable: true })
ONGOING?: number;
@Field(() => Number, { nullable: true })
ACTIVE?: number;

21
packages/twenty-server/src/modules/calendar/common/services/calendar-channel-sync-status.service.ts
View File

@ -6,6 +6,7 @@ import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decora
import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service';
import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum';
import { HealthCacheService } from 'src/engine/core-modules/health/health-cache.service';
import { HealthCounterCacheKeys } from 'src/engine/core-modules/health/types/health-counter-cache-keys.type';
import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
import {
CalendarChannelSyncStage,
@ -79,11 +80,6 @@ export class CalendarChannelSyncStatusService {
syncStatus: CalendarChannelSyncStatus.ONGOING,
syncStageStartedAt: new Date().toISOString(),
});
await this.healthCacheService.incrementCalendarChannelSyncJobByStatusCounter(
CalendarChannelSyncStatus.ONGOING,
calendarChannelIds.length,
);
}
public async resetAndScheduleFullCalendarEventListFetch(
@ -183,9 +179,10 @@ export class CalendarChannelSyncStatusService {
await this.schedulePartialCalendarEventListFetch(calendarChannelIds);
await this.healthCacheService.incrementCalendarChannelSyncJobByStatusCounter(
await this.healthCacheService.updateMessageOrCalendarChannelSyncJobByStatusCache(
HealthCounterCacheKeys.CalendarEventSyncJobByStatus,
CalendarChannelSyncStatus.ACTIVE,
calendarChannelIds.length,
calendarChannelIds,
);
}
@ -213,9 +210,10 @@ export class CalendarChannelSyncStatusService {
syncStage: CalendarChannelSyncStage.FAILED,
});
await this.healthCacheService.incrementCalendarChannelSyncJobByStatusCounter(
await this.healthCacheService.updateMessageOrCalendarChannelSyncJobByStatusCache(
HealthCounterCacheKeys.CalendarEventSyncJobByStatus,
CalendarChannelSyncStatus.FAILED_UNKNOWN,
calendarChannelIds.length,
calendarChannelIds,
);
}
@ -268,9 +266,10 @@ export class CalendarChannelSyncStatusService {
workspaceId,
);
await this.healthCacheService.incrementCalendarChannelSyncJobByStatusCounter(
await this.healthCacheService.updateMessageOrCalendarChannelSyncJobByStatusCache(
HealthCounterCacheKeys.CalendarEventSyncJobByStatus,
CalendarChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS,
calendarChannelIds.length,
calendarChannelIds,
);
}

21
packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts
View File

@ -6,6 +6,7 @@ import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decora
import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service';
import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum';
import { HealthCacheService } from 'src/engine/core-modules/health/health-cache.service';
import { HealthCounterCacheKeys } from 'src/engine/core-modules/health/types/health-counter-cache-keys.type';
import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
import { AccountsToReconnectService } from 'src/modules/connected-account/services/accounts-to-reconnect.service';
import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
@ -129,11 +130,6 @@ export class MessageChannelSyncStatusService {
syncStatus: MessageChannelSyncStatus.ONGOING,
syncStageStartedAt: new Date().toISOString(),
});
await this.healthCacheService.incrementMessageChannelSyncJobByStatusCounter(
MessageChannelSyncStatus.ONGOING,
messageChannelIds.length,
);
}
public async markAsCompletedAndSchedulePartialMessageListFetch(
@ -156,9 +152,10 @@ export class MessageChannelSyncStatusService {
syncedAt: new Date().toISOString(),
});
await this.healthCacheService.incrementMessageChannelSyncJobByStatusCounter(
await this.healthCacheService.updateMessageOrCalendarChannelSyncJobByStatusCache(
HealthCounterCacheKeys.MessageChannelSyncJobByStatus,
MessageChannelSyncStatus.ACTIVE,
messageChannelIds.length,
messageChannelIds,
);
}
@ -202,9 +199,10 @@ export class MessageChannelSyncStatusService {
syncStatus: MessageChannelSyncStatus.FAILED_UNKNOWN,
});
await this.healthCacheService.incrementMessageChannelSyncJobByStatusCounter(
await this.healthCacheService.updateMessageOrCalendarChannelSyncJobByStatusCache(
HealthCounterCacheKeys.MessageChannelSyncJobByStatus,
MessageChannelSyncStatus.FAILED_UNKNOWN,
messageChannelIds.length,
messageChannelIds,
);
}
@ -232,9 +230,10 @@ export class MessageChannelSyncStatusService {
syncStatus: MessageChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS,
});
await this.healthCacheService.incrementMessageChannelSyncJobByStatusCounter(
await this.healthCacheService.updateMessageOrCalendarChannelSyncJobByStatusCache(
HealthCounterCacheKeys.MessageChannelSyncJobByStatus,
MessageChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS,
messageChannelIds.length,
messageChannelIds,
);
const connectedAccountRepository =