token refresh, and filter inactive connections

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, {

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<void> {
+    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<void> {
     const { eventName, workspaceId, ...payload } = job.data;
 

apps/server/src/core/integration/repos/integration-connection.repo.ts

@@ -148,6 +148,19 @@ export class IntegrationConnectionRepo {
       .execute();
   }
 
+  async findExpiringTokens(
+    expiresBeforeMs: number,
+  ): Promise<IntegrationConnection[]> {
+    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,

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',
 }