Jayden/AFFiNE-Mirror
0
0
Fork 0
mirror of https://github.com/toeverything/AFFiNE.git synced 2026-09-03 07:10:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(server): improve calendar sync queue (#14783)

Browse Source 
#### PR Dependency Tree


* **PR #14783** 👈

This tree was auto-generated by
[Charcoal](https://github.com/danerwilliams/charcoal)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
  * Configurable request timeout for calendar integrations.
* Calendar polling now enqueues per-subscription sync jobs (larger
batch) for improved throughput.

* **Bug Fixes / Improvements**
* Persisted next-sync timestamps and retry counts for more reliable
scheduling and retry behavior.
* Exponential backoff and webhook renewal now update scheduling
consistently.

* **Refactor**
* Calendar sync flow moved to a job-queue-driven design for better
concurrency and observability.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
DarkSky
2026-04-05 10:52:18 +08:00
committed by GitHub
parent bfcf7fc2ba
commit d975bf46fb
17 changed files with 352 additions and 307 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/backend/server/src/plugins/calendar/service.ts
+104 -135
View File
@@ -4,16 +4,15 @@ import { Injectable, Logger } from '@nestjs/common';
import { Transactional } from '@nestjs-cls/transactional';
import type { CalendarAccount, Prisma } from '@prisma/client';
import { addDays, subDays } from 'date-fns';
import { chunk } from 'lodash-es';
import {
CalendarProviderRequestError,
Config,
exponentialBackoffDelay,
GraphqlBadRequest,
Mutex,
JobQueue,
URLHelper,
} from '../../base';
import { SessionRedis } from '../../base/redis';
import { Models } from '../../models';
import type { CalendarCalDAVProviderPreset } from './config';
import {
@@ -29,11 +28,10 @@ import type { LinkCalDAVAccountInput } from './types';
const TOKEN_REFRESH_SKEW_MS = 60 * 1000;
const DEFAULT_PAST_DAYS = 90;
const DEFAULT_FUTURE_DAYS = 180;
const SYNC_FAILURE_BACKOFF_KEY_PREFIX = 'calendar:sync:backoff:';
const SYNC_FAILURE_BACKOFF_BASE_MS = 5 * 60 * 1000;
const SYNC_FAILURE_BACKOFF_MAX_MS = 6 * 60 * 60 * 1000;
const SYNC_FAILURE_BACKOFF_TTL_SECONDS = 24 * 60 * 60;
const ACCOUNT_SYNC_BATCH_SIZE = 10;
const DEFAULT_REFRESH_INTERVAL_MINUTES = 30;
const CHANNEL_RENEW_RETRY_MS = 15 * 60 * 1000;
@Injectable()
export class CalendarService {
@@ -43,8 +41,7 @@ export class CalendarService {
constructor(
private readonly models: Models,
private readonly providerFactory: CalendarProviderFactory<CalendarProvider>,
private readonly mutex: Mutex,
private readonly redis: SessionRedis,
private readonly queue: JobQueue,
private readonly config: Config,
private readonly url: URLHelper
) {}
@@ -87,10 +84,24 @@ export class CalendarService {
return null;
}
return await this.models.calendarAccount.updateRefreshInterval(
accountId,
refreshIntervalMinutes
const updatedAccount =
await this.models.calendarAccount.updateRefreshInterval(
accountId,
refreshIntervalMinutes
);
const subscriptions =
await this.models.calendarSubscription.listByAccountForSync(accountId);
await Promise.all(
subscriptions.map(subscription =>
this.models.calendarSubscription.updateSync(subscription.id, {
nextSyncAt: this.calculateNextSyncAt(
subscription.lastSyncAt ?? this.now(),
refreshIntervalMinutes
),
})
)
);
return updatedAccount;
}
async unlinkAccount(userId: string, accountId: string) {
@@ -315,25 +326,6 @@ export class CalendarService {
return;
}
const now = Date.now();
const backoff = await this.getSyncFailureBackoff(subscription.id);
if (backoff && now < backoff.nextRetryAt.getTime()) {
return;
}
await using lock = await this.mutex.acquire(
`calendar:subscription:${subscriptionId}`
);
if (!lock) {
return;
}
const lockedNow = Date.now();
const lockedBackoff = await this.getSyncFailureBackoff(subscription.id);
if (lockedBackoff && lockedNow < lockedBackoff.nextRetryAt.getTime()) {
return;
}
const provider = this.providerFactory.get(
account.provider as CalendarProviderName
);
@@ -417,30 +409,27 @@ export class CalendarService {
}
if (synced) {
await this.clearSyncFailureBackoff(subscription.id);
await this.ensureWebhookChannel(subscription, provider, accessToken);
}
await this.models.calendarSubscription.updateLastSyncAt(
subscription.id,
new Date()
);
}
async syncAccount(accountId: string) {
const account = await this.models.calendarAccount.get(accountId);
if (!account || account.status !== 'active') {
return;
}
const subscriptions =
await this.models.calendarSubscription.listByAccountForSync(accountId);
for (const batch of chunk(subscriptions, ACCOUNT_SYNC_BATCH_SIZE)) {
await Promise.allSettled(
batch.map(subscription =>
this.syncSubscription(subscription.id, { reason: 'polling' })
)
const syncedAt = this.now();
let nextSyncAt = this.calculateNextSyncAt(
syncedAt,
account.refreshIntervalMinutes
);
try {
await this.ensureWebhookChannel(subscription, provider, accessToken);
} catch (error) {
nextSyncAt = this.calculateChannelRetryAt(nextSyncAt);
this.logger.warn(
`Failed to ensure webhook channel for subscription ${subscription.id}`,
this.toError(error)
);
}
await this.models.calendarSubscription.updateSync(subscription.id, {
lastSyncAt: syncedAt,
nextSyncAt,
syncRetryCount: 0,
});
}
}
@@ -459,9 +448,18 @@ export class CalendarService {
params.to
);
const subscriptions =
await this.models.calendarSubscription.listWithAccounts(subscriptionIds);
const staleSubscriptions = subscriptions.filter(
subscription =>
subscription.enabled &&
subscription.account.status === 'active' &&
subscription.nextSyncAt.getTime() <= this.nowMs()
);
Promise.allSettled(
subscriptionIds.map(subscriptionId =>
this.syncSubscription(subscriptionId, { reason: 'on-demand' })
staleSubscriptions.map(subscription =>
this.enqueueSyncSubscription(subscription.id, 'on-demand')
)
).catch(error => {
this.logger.warn('Calendar on-demand sync failed', error as Error);
@@ -517,7 +515,7 @@ export class CalendarService {
return;
}
await this.syncSubscription(subscription.id, { reason: 'webhook' });
await this.enqueueSyncSubscription(subscription.id, 'webhook');
}
getWebhookToken() {
@@ -751,7 +749,7 @@ export class CalendarService {
}
private getSyncWindow() {
const now = new Date();
const now = this.now();
return {
timeMin: subDays(now, DEFAULT_PAST_DAYS).toISOString(),
timeMax: addDays(now, DEFAULT_FUTURE_DAYS).toISOString(),
@@ -771,7 +769,7 @@ export class CalendarService {
if (
accessToken &&
account.expiresAt &&
account.expiresAt.getTime() > Date.now() + TOKEN_REFRESH_SKEW_MS
account.expiresAt.getTime() > this.nowMs() + TOKEN_REFRESH_SKEW_MS
) {
return { accessToken };
}
@@ -835,7 +833,7 @@ export class CalendarService {
return;
}
const renewThreshold = Date.now() + 24 * 60 * 60 * 1000;
const renewThreshold = this.nowMs() + 24 * 60 * 60 * 1000;
if (
subscription.channelExpiration &&
subscription.channelExpiration.getTime() > renewThreshold
@@ -877,6 +875,7 @@ export class CalendarService {
subscription: {
id: string;
externalCalendarId: string;
syncRetryCount: number;
customChannelId: string | null;
customResourceId: string | null;
};
@@ -899,7 +898,6 @@ export class CalendarService {
}
if (this.isTokenInvalidError(params.error)) {
await this.clearSyncFailureBackoff(params.subscription.id);
await this.models.calendarAccount.invalidateAndPurge(
params.account.id,
this.formatSyncError(params.error)
@@ -907,18 +905,14 @@ export class CalendarService {
return;
}
const backoff = await this.bumpSyncFailureBackoff(params.subscription.id);
const interval = params.account.refreshIntervalMinutes ?? 60;
const lastSyncAt = this.calculateLastSyncAtForRetry(
backoff.nextRetryAt,
interval
);
await this.models.calendarSubscription.updateLastSyncAt(
params.subscription.id,
lastSyncAt
);
const attempt = params.subscription.syncRetryCount + 1;
const nextRetryAt = this.calculateFailureRetryAt(attempt);
await this.models.calendarSubscription.updateSync(params.subscription.id, {
nextSyncAt: nextRetryAt,
syncRetryCount: attempt,
});
this.logger.warn(
`Calendar sync failed for subscription ${params.subscription.id}, attempt ${backoff.attempt}, next retry at ${backoff.nextRetryAt.toISOString()}`,
`Calendar sync failed for subscription ${params.subscription.id}, attempt ${attempt}, next retry at ${nextRetryAt.toISOString()}`,
this.toError(params.error)
);
}
@@ -931,15 +925,6 @@ export class CalendarService {
return status === 404;
}
private calculateLastSyncAtForRetry(
nextRetryAt: Date,
refreshIntervalMinutes: number
) {
// Cron schedules by `now - lastSyncAt >= refreshInterval`, so back-calculate
// a synthetic lastSyncAt to defer the next attempt to `nextRetryAt`.
return new Date(nextRetryAt.getTime() - refreshIntervalMinutes * 60 * 1000);
}
private async disableSubscription(params: {
subscriptionId: string;
provider: CalendarProvider;
@@ -970,68 +955,52 @@ export class CalendarService {
await this.models.calendarSubscription.disableAndPurge(
params.subscriptionId
);
await this.clearSyncFailureBackoff(params.subscriptionId);
}
private getSyncFailureBackoffKey(subscriptionId: string) {
return `${SYNC_FAILURE_BACKOFF_KEY_PREFIX}${subscriptionId}`;
}
private async getSyncFailureBackoff(subscriptionId: string) {
const key = this.getSyncFailureBackoffKey(subscriptionId);
const value = await this.redis.get(key);
if (!value) {
return null;
}
try {
const parsed = JSON.parse(value) as {
attempt?: number;
nextRetryAt?: string;
};
if (!parsed.attempt || !parsed.nextRetryAt) {
return null;
async enqueueSyncSubscription(
subscriptionId: string,
reason: 'polling' | 'webhook' | 'on-demand'
) {
await this.queue.add(
'calendar.syncSubscription',
{
subscriptionId,
reason,
},
{
jobId: subscriptionId,
}
const nextRetryAt = new Date(parsed.nextRetryAt);
if (Number.isNaN(nextRetryAt.getTime())) {
return null;
}
return {
attempt: parsed.attempt,
nextRetryAt,
};
} catch {
return null;
}
);
}
private async bumpSyncFailureBackoff(subscriptionId: string) {
const state = await this.getSyncFailureBackoff(subscriptionId);
const attempt = (state?.attempt ?? 0) + 1;
const delay = Math.min(
SYNC_FAILURE_BACKOFF_BASE_MS * 2 ** (attempt - 1),
SYNC_FAILURE_BACKOFF_MAX_MS
);
const nextRetryAt = new Date(Date.now() + delay);
const key = this.getSyncFailureBackoffKey(subscriptionId);
await this.redis.set(
key,
JSON.stringify({
attempt,
nextRetryAt: nextRetryAt.toISOString(),
}),
'EX',
SYNC_FAILURE_BACKOFF_TTL_SECONDS
);
return {
attempt,
nextRetryAt,
};
private calculateNextSyncAt(base: Date, refreshIntervalMinutes?: number) {
const intervalMinutes =
refreshIntervalMinutes ?? DEFAULT_REFRESH_INTERVAL_MINUTES;
return new Date(base.getTime() + intervalMinutes * 60 * 1000);
}
private async clearSyncFailureBackoff(subscriptionId: string) {
const key = this.getSyncFailureBackoffKey(subscriptionId);
await this.redis.del(key);
private calculateChannelRetryAt(nextSyncAt: Date) {
return new Date(
Math.min(nextSyncAt.getTime(), this.nowMs() + CHANNEL_RENEW_RETRY_MS)
);
}
private calculateFailureRetryAt(attempt: number) {
return new Date(
this.nowMs() +
exponentialBackoffDelay(attempt - 1, {
baseDelayMs: SYNC_FAILURE_BACKOFF_BASE_MS,
maxDelayMs: SYNC_FAILURE_BACKOFF_MAX_MS,
})
);
}
private now() {
return new Date(this.nowMs());
}
private nowMs() {
return Date.now();
}
private formatSyncError(error: unknown) {