[Refactor] Centralize integration lifecycle transitions

apps/web/src/app/api/mcp-oauth/callback/route.ts

@@ -1,13 +1,7 @@
 import { createHash } from 'node:crypto';
 import { type NextRequest, NextResponse } from 'next/server';
 
-import {
-  and,
-  db,
-  mcpConnections,
-  deploymentMcpEnablements,
-  eq,
-} from '@roomote/db/server';
+import { db, mcpConnections, eq } from '@roomote/db/server';
 import {
   getMcpIntegration,
   getMcpIntegrationDefaultDisabledTools,
@@ -30,7 +24,8 @@ import { authorize } from '@/lib/server';
 import { bootstrapWebRuntimeEnv } from '@/lib/server/bootstrap-runtime-env';
 import { getPublicAppUrl } from '@/lib/server/get-public-app-url';
 import { logger } from '@/lib/server/logger';
-import { captureIntegrationLifecycleEvent } from '@/lib/server/integration-telemetry';
+import { captureIntegrationConnectionTransitions } from '@/lib/server/integration-telemetry';
+import { enableDeploymentMcpIntegration } from '@/lib/server/deployment-mcp-connection';
 import {
   hydrateLinearMcpConnectionAfterOauth,
   LinearReplayIdentityMismatchError,
@@ -367,7 +362,6 @@ export async function GET(request: NextRequest) {
         connection,
         tokens,
         replayToken: oauthState.replayToken,
-        enabledByUserId: userId,
       });
     } else {
       failureStage = 'token_storage';
@@ -381,51 +375,19 @@ export async function GET(request: NextRequest) {
       failureStage = 'deployment_enablement';
       const defaultDisabledTools =
         getMcpIntegrationDefaultDisabledTools(integration);
-      const [reenabled] = await db
-        .update(deploymentMcpEnablements)
-        .set({
-          enabled: true,
-          enabledByUserId: userId,
-          updatedAt: new Date(),
-        })
-        .where(
-          and(
-            eq(deploymentMcpEnablements.mcpId, integration.id),
-            eq(deploymentMcpEnablements.enabled, false),
-          ),
-        )
-        .returning({ mcpId: deploymentMcpEnablements.mcpId });
-      const [inserted] = reenabled
-        ? []
-        : await db
-            .insert(deploymentMcpEnablements)
-            .values({
-              mcpId: integration.id,
-              enabled: true,
-              enabledByUserId: userId,
-              ...(defaultDisabledTools.length > 0
-                ? {
-                    disabledTools: [...defaultDisabledTools],
-                  }
-                : {}),
-            })
-            .onConflictDoNothing({ target: deploymentMcpEnablements.mcpId })
-            .returning({ mcpId: deploymentMcpEnablements.mcpId });
-      integrationBecameEnabled = Boolean(reenabled || inserted);
+      integrationBecameEnabled = await enableDeploymentMcpIntegration({
+        mcpId: integration.id,
+        userId,
+        defaultDisabledTools: [...defaultDisabledTools],
+      });
     }
 
-    captureIntegrationLifecycleEvent(
-      'integration_connected',
-      connection.mcpId,
+    captureIntegrationConnectionTransitions({
+      integrationId: connection.mcpId,
       userId,
-    );
-    if (integrationBecameEnabled && integration) {
-      captureIntegrationLifecycleEvent(
-        'integration_enabled',
-        integration.id,
-        userId,
-      );
-    }
+      connected: true,
+      enabled: integrationBecameEnabled,
+    });
 
     return redirectToResult({ status: 'connected' });
   } catch (error) {

apps/web/src/lib/server/deployment-mcp-connection.ts

@@ -0,0 +1,202 @@
+import {
+  and,
+  db,
+  deploymentMcpEnablements,
+  eq,
+  isNull,
+  mcpConnections,
+  type DatabaseOrTransaction,
+} from '@roomote/db/server';
+import type { McpConnectionAuthConfig } from '@roomote/types';
+
+import { captureIntegrationConnectionTransitions } from '@/lib/server/integration-telemetry';
+
+type SaveAuthenticatedDeploymentMcpConnectionInput<
+  TAuthConfig extends McpConnectionAuthConfig,
+> = {
+  mcpId: string;
+  userId: string;
+  buildAuthConfig: (
+    existingAuthConfig: McpConnectionAuthConfig | null,
+  ) => TAuthConfig | Promise<TAuthConfig>;
+  clearOauthTokens?: boolean;
+  resetDisabledTools?: boolean;
+};
+
+async function persistDeploymentMcpIntegrationEnabled(
+  database: DatabaseOrTransaction,
+  input: {
+    mcpId: string;
+    userId: string;
+    defaultDisabledTools?: string[];
+    resetDisabledTools?: boolean;
+  },
+): Promise<boolean> {
+  const updatedDisabledTools = input.resetDisabledTools ? null : undefined;
+  const insertedDisabledTools = input.resetDisabledTools
+    ? null
+    : input.defaultDisabledTools?.length
+      ? input.defaultDisabledTools
+      : undefined;
+  const [reenabled] = await database
+    .update(deploymentMcpEnablements)
+    .set({
+      enabled: true,
+      enabledByUserId: input.userId,
+      ...(updatedDisabledTools !== undefined
+        ? { disabledTools: updatedDisabledTools }
+        : {}),
+      updatedAt: new Date(),
+    })
+    .where(
+      and(
+        eq(deploymentMcpEnablements.mcpId, input.mcpId),
+        eq(deploymentMcpEnablements.enabled, false),
+      ),
+    )
+    .returning({ mcpId: deploymentMcpEnablements.mcpId });
+
+  if (reenabled) {
+    return true;
+  }
+
+  const [inserted] = await database
+    .insert(deploymentMcpEnablements)
+    .values({
+      mcpId: input.mcpId,
+      enabled: true,
+      enabledByUserId: input.userId,
+      ...(insertedDisabledTools !== undefined
+        ? { disabledTools: insertedDisabledTools }
+        : {}),
+    })
+    .onConflictDoNothing({ target: deploymentMcpEnablements.mcpId })
+    .returning({ mcpId: deploymentMcpEnablements.mcpId });
+
+  if (inserted) {
+    return true;
+  }
+
+  const [enabledAfterConflict] = await database
+    .update(deploymentMcpEnablements)
+    .set({
+      enabled: true,
+      enabledByUserId: input.userId,
+      ...(updatedDisabledTools !== undefined
+        ? { disabledTools: updatedDisabledTools }
+        : {}),
+      updatedAt: new Date(),
+    })
+    .where(
+      and(
+        eq(deploymentMcpEnablements.mcpId, input.mcpId),
+        eq(deploymentMcpEnablements.enabled, false),
+      ),
+    )
+    .returning({ mcpId: deploymentMcpEnablements.mcpId });
+
+  if (enabledAfterConflict) {
+    return true;
+  }
+
+  await database
+    .update(deploymentMcpEnablements)
+    .set({
+      enabledByUserId: input.userId,
+      ...(updatedDisabledTools !== undefined
+        ? { disabledTools: updatedDisabledTools }
+        : {}),
+      updatedAt: new Date(),
+    })
+    .where(eq(deploymentMcpEnablements.mcpId, input.mcpId));
+
+  return false;
+}
+
+export async function saveAuthenticatedDeploymentMcpConnection<
+  TAuthConfig extends McpConnectionAuthConfig,
+>(
+  input: SaveAuthenticatedDeploymentMcpConnectionInput<TAuthConfig>,
+): Promise<TAuthConfig> {
+  const existingConnection = await db.query.mcpConnections.findFirst({
+    where: and(
+      eq(mcpConnections.mcpId, input.mcpId),
+      isNull(mcpConnections.userId),
+    ),
+    columns: { authConfig: true },
+  });
+  const authConfig = await input.buildAuthConfig(
+    existingConnection?.authConfig ?? null,
+  );
+
+  const transitions = await db.transaction(async (tx) => {
+    const [insertedConnection] = await tx
+      .insert(mcpConnections)
+      .values({
+        userId: null,
+        mcpId: input.mcpId,
+        connectionRole: 'default',
+        authConfig,
+        enabled: true,
+        authStatus: 'authenticated',
+      })
+      .onConflictDoNothing({
+        target: [
+          mcpConnections.userId,
+          mcpConnections.mcpId,
+          mcpConnections.connectionRole,
+        ],
+      })
+      .returning({ id: mcpConnections.id });
+
+    if (!insertedConnection) {
+      await tx
+        .update(mcpConnections)
+        .set({
+          authConfig,
+          ...(input.clearOauthTokens
+            ? {
+                accessToken: null,
+                refreshToken: null,
+                tokenExpiresAt: null,
+                scopes: null,
+              }
+            : {}),
+          enabled: true,
+          authStatus: 'authenticated',
+          updatedAt: new Date(),
+        })
+        .where(
+          and(
+            eq(mcpConnections.mcpId, input.mcpId),
+            isNull(mcpConnections.userId),
+            eq(mcpConnections.connectionRole, 'default'),
+          ),
+        );
+    }
+
+    const enabled = await persistDeploymentMcpIntegrationEnabled(tx, {
+      mcpId: input.mcpId,
+      userId: input.userId,
+      resetDisabledTools: input.resetDisabledTools,
+    });
+
+    return { connected: Boolean(insertedConnection), enabled };
+  });
+
+  captureIntegrationConnectionTransitions({
+    integrationId: input.mcpId,
+    userId: input.userId,
+    ...transitions,
+  });
+
+  return authConfig;
+}
+
+export async function enableDeploymentMcpIntegration(input: {
+  mcpId: string;
+  userId: string;
+  defaultDisabledTools: string[];
+}): Promise<boolean> {
+  return persistDeploymentMcpIntegrationEnabled(db, input);
+}

apps/web/src/lib/server/integration-telemetry.ts

@@ -16,3 +16,25 @@ export function captureIntegrationLifecycleEvent(
     properties: { integration_id: integrationId },
   });
 }
+
+export function captureIntegrationConnectionTransitions(input: {
+  integrationId: string;
+  userId: string;
+  connected: boolean;
+  enabled: boolean;
+}): void {
+  if (input.connected) {
+    captureIntegrationLifecycleEvent(
+      'integration_connected',
+      input.integrationId,
+      input.userId,
+    );
+  }
+  if (input.enabled) {
+    captureIntegrationLifecycleEvent(
+      'integration_enabled',
+      input.integrationId,
+      input.userId,
+    );
+  }
+}