wip

apps/server/src/common/features.ts

@@ -23,6 +23,7 @@ export const Feature = {
   PERSONAL_SPACES: 'spaces:personal',
   DOCX_EXPORT: 'export:docx',
   BASES: 'bases',
+  INTEGRATIONS: 'integrations',
 } as const;
 
 export type FeatureKey = (typeof Feature)[keyof typeof Feature];

apps/server/src/core/integration/dto/integration.dto.ts

@@ -1,5 +1,4 @@
 import {
-  IsBoolean,
   IsNotEmpty,
   IsObject,
   IsOptional,
@@ -28,10 +27,6 @@ export class UpdateIntegrationDto {
   @IsOptional()
   @IsObject()
   settings?: Record<string, any>;
-
-  @IsOptional()
-  @IsBoolean()
-  isEnabled?: boolean;
 }
 
 export class IntegrationIdDto {

apps/server/src/core/integration/integration-connection.service.ts

@@ -28,7 +28,7 @@ export class IntegrationConnectionService {
     );
 
     return {
-      connected: !!connection,
+      connected: !!connection && !connection.invalidatedAt,
       providerUserId: connection?.providerUserId ?? undefined,
     };
   }
@@ -61,9 +61,9 @@ export class IntegrationConnectionService {
     return rows.map((row) => ({
       integrationId: row.integrationId,
       type: row.type,
-      isEnabled: row.isEnabled,
       providerUserId: row.providerUserId ?? null,
       connectedAt: row.createdAt,
+      invalidatedAt: row.invalidatedAt ?? null,
     }));
   }
 

apps/server/src/core/integration/integration.controller.ts

@@ -19,11 +19,14 @@ import {
   UpdateIntegrationDto,
   IntegrationIdDto,
 } from './dto/integration.dto';
+import { IntegrationRegistry } from './registry/integration-registry';
 import WorkspaceAbilityFactory from '../casl/abilities/workspace-ability.factory';
 import {
   WorkspaceCaslAction,
   WorkspaceCaslSubject,
 } from '../casl/interfaces/workspace-ability.type';
+import { LicenseCheckService } from '../../integrations/environment/license-check.service';
+import { Feature } from '../../common/features';
 
 @Controller('integrations')
 export class IntegrationController {
@@ -31,8 +34,22 @@ export class IntegrationController {
     private readonly integrationService: IntegrationService,
     private readonly connectionService: IntegrationConnectionService,
     private readonly workspaceAbility: WorkspaceAbilityFactory,
+    private readonly licenseCheckService: LicenseCheckService,
+    private readonly registry: IntegrationRegistry,
   ) {}
 
+  private assertIntegrationsLicensed(workspace: Workspace) {
+    if (
+      !this.licenseCheckService.hasFeature(
+        workspace.licenseKey,
+        Feature.INTEGRATIONS,
+        workspace.plan,
+      )
+    ) {
+      throw new ForbiddenException('This feature requires a valid license');
+    }
+  }
+
   @UseGuards(JwtAuthGuard)
   @HttpCode(HttpStatus.OK)
   @Post('available')
@@ -67,6 +84,9 @@ export class IntegrationController {
       throw new ForbiddenException();
     }
 
+    if (this.registry.getProvider(dto.type)?.definition.requiresLicense) {
+      this.assertIntegrationsLicensed(workspace);
+    }
     return this.integrationService.install(dto.type, workspace.id, user.id);
   }
 
@@ -112,7 +132,6 @@ export class IntegrationController {
 
     return this.integrationService.update(dto.integrationId, workspace.id, {
       settings: dto.settings,
-      isEnabled: dto.isEnabled,
     });
   }
 

apps/server/src/core/integration/integration.processor.ts

@@ -1,5 +1,7 @@
-import { Processor, WorkerHost } from '@nestjs/bullmq';
-import { Logger } from '@nestjs/common';
+import { OnWorkerEvent, Processor, WorkerHost } from '@nestjs/bullmq';
+import { Logger, NotFoundException } from '@nestjs/common';
+import { IntegrationConnection } from '@docmost/db/types/entity.types';
+import { TokenInvalidError } from './registry/integration-provider.interface';
 import { Job } from 'bullmq';
 import { QueueJob, QueueName } from '../../integrations/queue/constants/queue.constants';
 import { IntegrationRegistry } from './registry/integration-registry';
@@ -35,6 +37,13 @@ export class IntegrationProcessor extends WorkerHost {
     }
   }
 
+  // Route worker-level errors (e.g. lock renewal after laptop sleep) through
+  // the logger instead of bullmq's raw console.error fallback.
+  @OnWorkerEvent('error')
+  onError(err: Error): void {
+    this.logger.error(`Worker error: ${err.message}`);
+  }
+
   private async handleTokenRefresh(): Promise<void> {
     const connections = await this.connectionRepo.findExpiringTokens(
       TOKEN_REFRESH_WINDOW_MS,
@@ -55,6 +64,15 @@ export class IntegrationProcessor extends WorkerHost {
         this.logger.error(
           `Token refresh failed for connection ${connection.id}: ${(err as Error).message}`,
         );
+        // Dead credential or orphaned row: retire it so findExpiringTokens stops selecting it.
+        if (
+          err instanceof NotFoundException ||
+          err instanceof TokenInvalidError
+        ) {
+          await this.connectionRepo
+            .invalidate(connection.id)
+            .catch(() => undefined);
+        }
       }
     }
   }
@@ -67,7 +85,7 @@ export class IntegrationProcessor extends WorkerHost {
     }
 
     const integrations =
-      await this.integrationRepo.findEnabledByWorkspace(workspaceId);
+      await this.integrationRepo.findAllByWorkspace(workspaceId);
 
     for (const integration of integrations) {
       const provider = this.registry.getProvider(integration.type);
@@ -75,12 +93,13 @@ export class IntegrationProcessor extends WorkerHost {
         continue;
       }
 
+      let connection: IntegrationConnection | undefined;
       try {
         const connections = await this.connectionRepo.findByIntegration(
           integration.id,
         );
 
-        const connection = connections[0];
+        connection = connections[0];
         let accessToken: string | undefined;
 
         if (connection) {
@@ -103,6 +122,11 @@ export class IntegrationProcessor extends WorkerHost {
         this.logger.error(
           `Integration event handler failed for ${integration.type}: ${(err as Error).message}`,
         );
+        if (err instanceof TokenInvalidError && connection) {
+          await this.connectionRepo
+            .invalidate(connection.id)
+            .catch(() => undefined);
+        }
       }
     }
   }

apps/server/src/core/integration/integration.service.ts

@@ -3,7 +3,12 @@ import {
   Injectable,
   NotFoundException,
 } from '@nestjs/common';
+import { InjectKysely } from 'nestjs-kysely';
+import { KyselyDB } from '@docmost/db/types/kysely.types';
+import { executeTx } from '@docmost/db/utils';
 import { IntegrationRepo } from './repos/integration.repo';
+import { IntegrationConnectionRepo } from './repos/integration-connection.repo';
+import { IntegrationWebhookRepo } from './repos/integration-webhook.repo';
 import { IntegrationRegistry } from './registry/integration-registry';
 import { Integration } from '@docmost/db/types/entity.types';
 import { validateIntegrationSettings } from './dto/integration-settings.schema';
@@ -11,7 +16,10 @@ import { validateIntegrationSettings } from './dto/integration-settings.schema';
 @Injectable()
 export class IntegrationService {
   constructor(
+    @InjectKysely() private readonly db: KyselyDB,
     private readonly integrationRepo: IntegrationRepo,
+    private readonly connectionRepo: IntegrationConnectionRepo,
+    private readonly webhookRepo: IntegrationWebhookRepo,
     private readonly registry: IntegrationRegistry,
   ) {}
 
@@ -33,7 +41,7 @@ export class IntegrationService {
     userId: string,
   ): Promise<Integration> {
     const provider = this.registry.getProvider(type);
-    if (!provider) {
+    if (!provider || provider.definition.hidden) {
       throw new BadRequestException(`Unknown integration type: ${type}`);
     }
 
@@ -59,13 +67,18 @@ export class IntegrationService {
     if (!integration || integration.workspaceId !== workspaceId) {
       throw new NotFoundException('Integration not found');
     }
-    await this.integrationRepo.softDelete(integrationId);
+    // Delete child rows first so no orphan connections keep feeding the token refresh scheduler.
+    await executeTx(this.db, async (trx) => {
+      await this.connectionRepo.deleteByIntegration(integrationId, trx);
+      await this.webhookRepo.deleteByIntegration(integrationId, trx);
+      await this.integrationRepo.softDelete(integrationId, trx);
+    });
   }
 
   async update(
     integrationId: string,
     workspaceId: string,
-    data: { settings?: Record<string, any>; isEnabled?: boolean },
+    data: { settings?: Record<string, any> },
   ): Promise<Integration> {
     const integration = await this.integrationRepo.findById(integrationId);
     if (!integration || integration.workspaceId !== workspaceId) {
@@ -85,7 +98,6 @@ export class IntegrationService {
 
     return this.integrationRepo.update(integrationId, {
       ...(data.settings !== undefined && { settings: data.settings }),
-      ...(data.isEnabled !== undefined && { isEnabled: data.isEnabled }),
     });
   }
 }

apps/server/src/core/integration/oauth/oauth.controller.ts

@@ -24,6 +24,10 @@ import {
   OAuthInstallDto,
 } from '../dto/integration.dto';
 import { IntegrationConnectionService } from '../integration-connection.service';
+import { IntegrationRegistry } from '../registry/integration-registry';
+import { LicenseCheckService } from '../../../integrations/environment/license-check.service';
+import { Feature } from '../../../common/features';
+import { ForbiddenException } from '@nestjs/common';
 
 @Controller('integrations/oauth')
 export class OAuthController {
@@ -32,6 +36,8 @@ export class OAuthController {
   constructor(
     private readonly oauthService: OAuthService,
     private readonly connectionService: IntegrationConnectionService,
+    private readonly licenseCheckService: LicenseCheckService,
+    private readonly registry: IntegrationRegistry,
   ) {}
 
   @UseGuards(JwtAuthGuard)
@@ -66,6 +72,19 @@ export class OAuthController {
     @AuthUser() user: User,
     @AuthWorkspace() workspace: Workspace,
   ) {
+    // This flow creates the integration row on callback success; gate it
+    // like a plain install.
+    if (
+      this.registry.getProvider(dto.type)?.definition.requiresLicense &&
+      !this.licenseCheckService.hasFeature(
+        workspace.licenseKey,
+        Feature.INTEGRATIONS,
+        workspace.plan,
+      )
+    ) {
+      throw new ForbiddenException('This feature requires a valid license');
+    }
+
     const { authorizationUrl } = await this.oauthService.getInstallAuthorizationUrl(
       dto.type,
       workspace.id,

apps/server/src/core/integration/providers/github/github.service.ts

@@ -1,6 +1,7 @@
 import { Injectable, Logger } from '@nestjs/common';
 import { UnfurlResult } from '../../registry/integration-provider.interface';
 import { relativeTime } from '../../utils/relative-time';
+import { providerApiFetch } from '../../utils/provider-fetch';
 
 @Injectable()
 export class GitHubService {
@@ -253,7 +254,7 @@ export class GitHubService {
     apiBaseUrl: string,
     path: string,
   ): Promise<any> {
-    const response = await fetch(`${apiBaseUrl}${path}`, {
+    const response = await providerApiFetch('GitHub', `${apiBaseUrl}${path}`, {
       headers: {
         Authorization: `Bearer ${accessToken}`,
         Accept: 'application/vnd.github.v3+json',
@@ -261,12 +262,6 @@ export class GitHubService {
       },
     });
 
-    if (!response.ok) {
-      throw new Error(
-        `GitHub API error: ${response.status} ${response.statusText}`,
-      );
-    }
-
     return response.json();
   }
 }

apps/server/src/core/integration/providers/gitlab/gitlab.service.ts

@@ -1,6 +1,7 @@
 import { Injectable, Logger } from '@nestjs/common';
 import { UnfurlResult } from '../../registry/integration-provider.interface';
 import { relativeTime } from '../../utils/relative-time';
+import { providerApiFetch } from '../../utils/provider-fetch';
 
 @Injectable()
 export class GitLabService {
@@ -240,19 +241,13 @@ export class GitLabService {
     apiBaseUrl: string,
     path: string,
   ): Promise<any> {
-    const response = await fetch(`${apiBaseUrl}${path}`, {
+    const response = await providerApiFetch('GitLab', `${apiBaseUrl}${path}`, {
       headers: {
         Authorization: `Bearer ${accessToken}`,
         Accept: 'application/json',
       },
     });
 
-    if (!response.ok) {
-      throw new Error(
-        `GitLab API error: ${response.status} ${response.statusText}`,
-      );
-    }
-
     return response.json();
   }
 }

apps/server/src/core/integration/registry/integration-provider.interface.ts

@@ -35,6 +35,11 @@ export type IntegrationDefinition = {
   capabilities: IntegrationCapability[];
   oauth?: OAuthConfig;
   unfurlPatterns?: UnfurlPattern[];
+  // Kept out of the available list and refused for install; existing
+  // installations keep unfurling.
+  hidden?: boolean;
+  // Install requires the INTEGRATIONS license feature; unset = free.
+  requiresLicense?: boolean;
 };
 
 export type ConnectedEvent = {
@@ -84,6 +89,26 @@ export class UnfurlForbiddenError extends Error {
   }
 }
 
+// Thrown when the provider definitively rejects the stored credential (API 401,
+// or invalid_grant at the token endpoint). Callers retire the connection.
+export class TokenInvalidError extends Error {
+  constructor(message = 'Integration credential is no longer valid') {
+    super(message);
+    this.name = 'TokenInvalidError';
+  }
+}
+
+export class ProviderApiError extends Error {
+  constructor(
+    readonly provider: string,
+    readonly status: number,
+    statusText = '',
+  ) {
+    super(`${provider} API error: ${status} ${statusText}`.trimEnd());
+    this.name = 'ProviderApiError';
+  }
+}
+
 export type LinkDescription = {
   title: string;
   description?: string;

apps/server/src/core/integration/registry/integration-registry.ts

@@ -21,7 +21,9 @@ export class IntegrationRegistry {
   }
 
   getAvailableIntegrations(): IntegrationDefinition[] {
-    return this.getAllProviders().map((p) => p.definition);
+    return this.getAllProviders()
+      .map((p) => p.definition)
+      .filter((definition) => !definition.hidden);
   }
 
   findUnfurlProvider(

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

@@ -91,6 +91,7 @@ export class IntegrationConnectionRepo {
             accessToken: connection.accessToken,
             refreshToken: connection.refreshToken,
             tokenExpiresAt: connection.tokenExpiresAt,
+            invalidatedAt: null,
             scopes: connection.scopes,
             providerUserId: connection.providerUserId,
             metadata: connection.metadata,
@@ -123,6 +124,7 @@ export class IntegrationConnectionRepo {
           accessToken: input.accessToken,
           refreshToken: input.refreshToken ?? null,
           tokenExpiresAt: input.tokenExpiresAt ?? null,
+          invalidatedAt: null,
           scopes: input.scopes ?? null,
           userId: input.userId,
         },
@@ -200,9 +202,9 @@ export class IntegrationConnectionRepo {
       .select([
         'integrationConnections.integrationId',
         'integrations.type',
-        'integrations.isEnabled',
         'integrationConnections.providerUserId',
         'integrationConnections.createdAt',
+        'integrationConnections.invalidatedAt',
       ])
       .where('integrationConnections.userId', '=', userId)
       .where('integrations.workspaceId', '=', workspaceId)
@@ -216,10 +218,31 @@ export class IntegrationConnectionRepo {
     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)
+      .innerJoin(
+        'integrations',
+        'integrations.id',
+        'integrationConnections.integrationId',
+      )
+      .selectAll('integrationConnections')
+      .where('integrations.deletedAt', 'is', null)
+      .where('integrationConnections.invalidatedAt', 'is', null)
+      .where('integrationConnections.refreshToken', 'is not', null)
+      .where('integrationConnections.tokenExpiresAt', 'is not', null)
+      .where('integrationConnections.tokenExpiresAt', '<', threshold)
+      .execute();
+  }
+
+  // Retire a rejected credential: flag for reconnect UX, drop the dead refresh token; no-op if the row is gone.
+  async invalidate(connectionId: string): Promise<void> {
+    await this.db
+      .updateTable('integrationConnections')
+      .set({
+        invalidatedAt: new Date(),
+        refreshToken: null,
+        tokenExpiresAt: null,
+        updatedAt: new Date(),
+      })
+      .where('id', '=', connectionId)
       .execute();
   }
 

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

@@ -41,20 +41,6 @@ export class IntegrationRepo {
       .executeTakeFirst();
   }
 
-  async findEnabledByWorkspace(
-    workspaceId: string,
-    trx?: KyselyTransaction,
-  ): Promise<Integration[]> {
-    const db = dbOrTx(this.db, trx);
-    return db
-      .selectFrom('integrations')
-      .selectAll()
-      .where('workspaceId', '=', workspaceId)
-      .where('isEnabled', '=', true)
-      .where('deletedAt', 'is', null)
-      .execute();
-  }
-
   async findAllByWorkspace(
     workspaceId: string,
     trx?: KyselyTransaction,
@@ -91,7 +77,6 @@ export class IntegrationRepo {
       .onConflict((oc) =>
         oc.columns(['type', 'workspaceId']).doUpdateSet({
           deletedAt: null,
-          isEnabled: true,
           installedById: integration.installedById,
           updatedAt: new Date(),
         }),
@@ -135,7 +120,6 @@ export class IntegrationRepo {
       .selectFrom('integrations')
       .selectAll()
       .where('type', '=', type)
-      .where('isEnabled', '=', true)
       .where('deletedAt', 'is', null)
       .where(sql<string>`settings->>${sql.lit(key)}`, '=', value)
       .executeTakeFirst();

apps/server/src/core/integration/unfurl/unfurl.service.ts

@@ -7,6 +7,8 @@ import {
   UnfurlResult,
   UnfurlNeedsConnection,
   UnfurlForbiddenError,
+  TokenInvalidError,
+  ProviderApiError,
   IntegrationProvider,
 } from '../registry/integration-provider.interface';
 import { RedisService } from '@nestjs-labs/nestjs-ioredis';
@@ -14,6 +16,9 @@ import type { Redis } from 'ioredis';
 import * as crypto from 'crypto';
 
 const UNFURL_CACHE_TTL = 300; // 5 minutes
+// Transient failures get a short negative cache so a broken provider is not
+// re-fetched on every view; 404s cache at the normal TTL (the target is gone).
+const UNFURL_ERROR_CACHE_TTL = 60;
 const UNFURL_CACHE_PREFIX = 'unfurl:';
 
 @Injectable()
@@ -66,11 +71,12 @@ export class UnfurlService {
             userId,
           );
 
-    if (!connection) {
+    if (!connection || connection.invalidatedAt) {
+      // Dead workspace connections need an admin re-install; members get no card.
       if (connectionScope === 'workspace') {
         return null;
       }
-      // Not cached: the card should load as soon as the user connects.
+      // Not cached: the card should load as soon as the user (re)connects.
       return this.buildNeedsConnection(
         provider,
         integration.id,
@@ -105,14 +111,45 @@ export class UnfurlService {
     } catch (err) {
       // Not-authorized is an expected outcome (no card), not an error.
       if (err instanceof UnfurlForbiddenError) {
-        this.logger.debug(`Unfurl not authorized for ${url}`);
+        this.logger.debug(
+          `Unfurl not authorized for ${url}: ${(err as Error).message}`,
+        );
+        await this.cacheNull(cacheKey, UNFURL_ERROR_CACHE_TTL);
         return null;
       }
+      if (err instanceof TokenInvalidError) {
+        this.logger.warn(
+          `Retiring connection ${connection.id}: ${(err as Error).message}`,
+        );
+        await this.connectionRepo
+          .invalidate(connection.id)
+          .catch(() => undefined);
+        if (connectionScope === 'workspace') {
+          return null;
+        }
+        // Not cached so the card heals the moment the user reconnects.
+        return this.buildNeedsConnection(
+          provider,
+          integration.id,
+          patternType,
+          match,
+          url,
+        );
+      }
       this.logger.error(`Unfurl failed for ${url}: ${(err as Error).message}`);
+      const ttl =
+        err instanceof ProviderApiError && err.status === 404
+          ? UNFURL_CACHE_TTL
+          : UNFURL_ERROR_CACHE_TTL;
+      await this.cacheNull(cacheKey, ttl);
       return null;
     }
   }
 
+  private async cacheNull(cacheKey: string, ttl: number): Promise<void> {
+    await this.redis.set(cacheKey, 'null', 'EX', ttl);
+  }
+
   async purgeUserCache(workspaceId: string, userId: string): Promise<void> {
     const pattern = `${UNFURL_CACHE_PREFIX}${workspaceId}:${userId}:*`;
     try {
@@ -167,7 +204,6 @@ export class UnfurlService {
     patternType: string;
     integration: {
       id: string;
-      isEnabled: boolean;
       type: string;
       settings: unknown;
     };
@@ -178,13 +214,13 @@ export class UnfurlService {
         workspaceId,
         staticResult.provider.definition.type,
       );
-      if (integration && integration.isEnabled) {
+      if (integration) {
         return { ...staticResult, integration };
       }
     }
 
     const integrations =
-      await this.integrationRepo.findEnabledByWorkspace(workspaceId);
+      await this.integrationRepo.findAllByWorkspace(workspaceId);
 
     for (const integration of integrations) {
       const provider = this.registry.getProvider(integration.type);

apps/server/src/database/migrations/20260807T1264122-integrations.ts

@@ -11,7 +11,6 @@ export async function up(db: Kysely<any>): Promise<void> {
       col.references('workspaces.id').onDelete('cascade').notNull(),
     )
     .addColumn('type', 'text', (col) => col.notNull())
-    .addColumn('is_enabled', 'boolean', (col) => col.notNull().defaultTo(true))
     .addColumn('settings', 'jsonb')
     .addColumn('installed_by_id', 'uuid', (col) =>
       col.references('users.id').onDelete('set null'),
@@ -50,6 +49,8 @@ export async function up(db: Kysely<any>): Promise<void> {
     .addColumn('access_token', 'text')
     .addColumn('refresh_token', 'text')
     .addColumn('token_expires_at', 'timestamptz')
+    // Set when the provider definitively rejects the credential; a reconnect clears it.
+    .addColumn('invalidated_at', 'timestamptz')
     .addColumn('scopes', 'text')
     .addColumn('metadata', 'jsonb')
     // 'workspace' = one shared bot/app connection per integration (Slack);

apps/server/src/database/types/db.d.ts

@@ -510,7 +510,6 @@ export interface Integrations {
   id: Generated<string>;
   workspaceId: string;
   type: string;
-  isEnabled: Generated<boolean>;
   settings: Json | null;
   installedById: string | null;
   createdAt: Generated<Timestamp>;
@@ -527,6 +526,7 @@ export interface IntegrationConnections {
   accessToken: string | null;
   refreshToken: string | null;
   tokenExpiresAt: Timestamp | null;
+  invalidatedAt: Timestamp | null;
   scopes: string | null;
   kind: string;
   metadata: Json | null;