diff --git a/apps/client/src/features/integration/components/connection-row.tsx b/apps/client/src/features/integration/components/connection-row.tsx index af192498d..017616bd8 100644 --- a/apps/client/src/features/integration/components/connection-row.tsx +++ b/apps/client/src/features/integration/components/connection-row.tsx @@ -6,8 +6,7 @@ import { getIntegrationIcon } from "./integration-icons"; type ConnectionRowProps = { definition: IntegrationDefinition; connection?: UserConnection; - installed: boolean; - onConnect: (integrationId: string) => void; + onConnect: (type: string) => void; onDisconnect: (integrationId: string) => void; isDisconnecting?: boolean; }; @@ -15,14 +14,11 @@ type ConnectionRowProps = { export default function ConnectionRow({ definition, connection, - installed, onConnect, onDisconnect, isDisconnecting, }: ConnectionRowProps) { const { t } = useTranslation(); - const isConnected = !!connection; - const isAvailable = installed && (connection?.isEnabled ?? true); return ( @@ -47,7 +42,7 @@ export default function ConnectionRow({ - {isConnected ? ( + {connection ? ( <> {t("Connected")} @@ -63,7 +58,7 @@ export default function ConnectionRow({ {t("Disconnect")} - ) : isAvailable ? ( + ) : ( - ) : ( - - {t("Not available")} - )} diff --git a/apps/client/src/features/integration/pages/connections.tsx b/apps/client/src/features/integration/pages/connections.tsx index 708179197..d395f70fd 100644 --- a/apps/client/src/features/integration/pages/connections.tsx +++ b/apps/client/src/features/integration/pages/connections.tsx @@ -80,9 +80,12 @@ export default function Connections() { ) : ( {available - .filter((def) => def.capabilities.includes("oauth")) - .map((def) => { + .filter((def) => { + if (!def.capabilities.includes("oauth")) return false; const installation = installed?.find((i) => i.type === def.type); + return installation?.isEnabled; + }) + .map((def) => { const connection = myConnections?.find( (c) => c.type === def.type, ); @@ -92,7 +95,6 @@ export default function Connections() { key={def.type} definition={def} connection={connection} - installed={!!installation} onConnect={handleConnect} onDisconnect={handleDisconnect} isDisconnecting={disconnectMutation.isPending} diff --git a/apps/server/src/core/integration/integration.listener.ts b/apps/server/src/core/integration/integration.listener.ts index e2a7239ac..ca59d6540 100644 --- a/apps/server/src/core/integration/integration.listener.ts +++ b/apps/server/src/core/integration/integration.listener.ts @@ -1,17 +1,34 @@ -import { Injectable } from '@nestjs/common'; +import { Injectable, Logger, OnApplicationBootstrap } from '@nestjs/common'; import { OnEvent } from '@nestjs/event-emitter'; import { InjectQueue } from '@nestjs/bullmq'; import { Queue } from 'bullmq'; import { QueueJob, QueueName } from '../../integrations/queue/constants/queue.constants'; import { EventName } from '../../common/events/event.contants'; +const TOKEN_REFRESH_SCHEDULER_ID = 'integration-token-refresh-scheduler'; +const TOKEN_REFRESH_INTERVAL_MS = 10 * 60 * 1000; // 10 minutes + @Injectable() -export class IntegrationListener { +export class IntegrationListener implements OnApplicationBootstrap { + private readonly logger = new Logger(IntegrationListener.name); + constructor( @InjectQueue(QueueName.INTEGRATION_QUEUE) private readonly integrationQueue: Queue, ) {} + async onApplicationBootstrap() { + await this.integrationQueue.upsertJobScheduler( + TOKEN_REFRESH_SCHEDULER_ID, + { every: TOKEN_REFRESH_INTERVAL_MS }, + { + name: QueueJob.INTEGRATION_TOKEN_REFRESH, + data: {}, + }, + ); + this.logger.debug('Integration token refresh scheduler created'); + } + @OnEvent(EventName.PAGE_CREATED) async onPageCreated(payload: any) { await this.integrationQueue.add(QueueJob.INTEGRATION_EVENT, { diff --git a/apps/server/src/core/integration/integration.processor.ts b/apps/server/src/core/integration/integration.processor.ts index ad856cd35..a00e9616c 100644 --- a/apps/server/src/core/integration/integration.processor.ts +++ b/apps/server/src/core/integration/integration.processor.ts @@ -7,6 +7,8 @@ import { IntegrationRepo } from './repos/integration.repo'; import { IntegrationConnectionRepo } from './repos/integration-connection.repo'; import { OAuthService } from './oauth/oauth.service'; +const TOKEN_REFRESH_WINDOW_MS = 15 * 60 * 1000; // 15 minutes + @Processor(QueueName.INTEGRATION_QUEUE) export class IntegrationProcessor extends WorkerHost { private readonly logger = new Logger(IntegrationProcessor.name); @@ -25,11 +27,38 @@ export class IntegrationProcessor extends WorkerHost { case QueueJob.INTEGRATION_EVENT: await this.handleIntegrationEvent(job); break; + case QueueJob.INTEGRATION_TOKEN_REFRESH: + await this.handleTokenRefresh(); + break; default: this.logger.warn(`Unknown job: ${job.name}`); } } + private async handleTokenRefresh(): Promise { + const connections = await this.connectionRepo.findExpiringTokens( + TOKEN_REFRESH_WINDOW_MS, + ); + + if (connections.length === 0) { + return; + } + + this.logger.log( + `Refreshing tokens for ${connections.length} connection(s)`, + ); + + for (const connection of connections) { + try { + await this.oauthService.getValidAccessToken(connection); + } catch (err) { + this.logger.error( + `Token refresh failed for connection ${connection.id}: ${(err as Error).message}`, + ); + } + } + } + private async handleIntegrationEvent(job: Job): Promise { const { eventName, workspaceId, ...payload } = job.data; diff --git a/apps/server/src/core/integration/repos/integration-connection.repo.ts b/apps/server/src/core/integration/repos/integration-connection.repo.ts index 93e164dac..c555affb2 100644 --- a/apps/server/src/core/integration/repos/integration-connection.repo.ts +++ b/apps/server/src/core/integration/repos/integration-connection.repo.ts @@ -148,6 +148,19 @@ export class IntegrationConnectionRepo { .execute(); } + async findExpiringTokens( + expiresBeforeMs: number, + ): Promise { + const threshold = new Date(Date.now() + expiresBeforeMs); + return this.db + .selectFrom('integrationConnections') + .selectAll() + .where('refreshToken', 'is not', null) + .where('tokenExpiresAt', 'is not', null) + .where('tokenExpiresAt', '<', threshold) + .execute(); + } + async deleteByIntegration( integrationId: string, trx?: KyselyTransaction, diff --git a/apps/server/src/integrations/queue/constants/queue.constants.ts b/apps/server/src/integrations/queue/constants/queue.constants.ts index 3a2608eac..a000926cd 100644 --- a/apps/server/src/integrations/queue/constants/queue.constants.ts +++ b/apps/server/src/integrations/queue/constants/queue.constants.ts @@ -70,4 +70,5 @@ export enum QueueJob { PAGE_MENTION_NOTIFICATION = 'page-mention-notification', INTEGRATION_EVENT = 'integration-event', + INTEGRATION_TOKEN_REFRESH = 'integration-token-refresh', }