feat(integrations): add Linear integration with bidirectional sync

.github/workflows/deploy-preview.yml

@@ -138,6 +138,12 @@ jobs:
           POSTHOG_PROJECT_ID: ${{ secrets.POSTHOG_PROJECT_ID }}
           KV_REST_API_URL: ${{ secrets.KV_REST_API_URL }}
           KV_REST_API_TOKEN: ${{ secrets.KV_REST_API_TOKEN }}
+          LINEAR_CLIENT_ID: ${{ secrets.LINEAR_CLIENT_ID }}
+          LINEAR_CLIENT_SECRET: ${{ secrets.LINEAR_CLIENT_SECRET }}
+          LINEAR_WEBHOOK_SECRET: ${{ secrets.LINEAR_WEBHOOK_SECRET }}
+          QSTASH_TOKEN: ${{ secrets.QSTASH_TOKEN }}
+          QSTASH_CURRENT_SIGNING_KEY: ${{ secrets.QSTASH_CURRENT_SIGNING_KEY }}
+          QSTASH_NEXT_SIGNING_KEY: ${{ secrets.QSTASH_NEXT_SIGNING_KEY }}
         run: |
           vercel pull --yes --environment=preview --token=$VERCEL_TOKEN
           vercel build --token=$VERCEL_TOKEN
@@ -160,7 +166,13 @@ jobs:
             --env POSTHOG_API_KEY=$POSTHOG_API_KEY \
             --env POSTHOG_PROJECT_ID=$POSTHOG_PROJECT_ID \
             --env KV_REST_API_URL=$KV_REST_API_URL \
-            --env KV_REST_API_TOKEN=$KV_REST_API_TOKEN)
+            --env KV_REST_API_TOKEN=$KV_REST_API_TOKEN \
+            --env LINEAR_CLIENT_ID=$LINEAR_CLIENT_ID \
+            --env LINEAR_CLIENT_SECRET=$LINEAR_CLIENT_SECRET \
+            --env LINEAR_WEBHOOK_SECRET=$LINEAR_WEBHOOK_SECRET \
+            --env QSTASH_TOKEN=$QSTASH_TOKEN \
+            --env QSTASH_CURRENT_SIGNING_KEY=$QSTASH_CURRENT_SIGNING_KEY \
+            --env QSTASH_NEXT_SIGNING_KEY=$QSTASH_NEXT_SIGNING_KEY)
           vercel alias $VERCEL_URL ${{ env.API_ALIAS }} --scope=$VERCEL_ORG_ID --token=$VERCEL_TOKEN
           echo "vercel_url=$VERCEL_URL" >> $GITHUB_OUTPUT
 

.github/workflows/deploy-production.yml

@@ -88,6 +88,12 @@ jobs:
           POSTHOG_PROJECT_ID: ${{ secrets.POSTHOG_PROJECT_ID }}
           KV_REST_API_URL: ${{ secrets.KV_REST_API_URL }}
           KV_REST_API_TOKEN: ${{ secrets.KV_REST_API_TOKEN }}
+          LINEAR_CLIENT_ID: ${{ secrets.LINEAR_CLIENT_ID }}
+          LINEAR_CLIENT_SECRET: ${{ secrets.LINEAR_CLIENT_SECRET }}
+          LINEAR_WEBHOOK_SECRET: ${{ secrets.LINEAR_WEBHOOK_SECRET }}
+          QSTASH_TOKEN: ${{ secrets.QSTASH_TOKEN }}
+          QSTASH_CURRENT_SIGNING_KEY: ${{ secrets.QSTASH_CURRENT_SIGNING_KEY }}
+          QSTASH_NEXT_SIGNING_KEY: ${{ secrets.QSTASH_NEXT_SIGNING_KEY }}
         run: |
           vercel pull --yes --environment=production --token=$VERCEL_TOKEN
           vercel build --prod --token=$VERCEL_TOKEN
@@ -110,7 +116,13 @@ jobs:
             --env POSTHOG_API_KEY=$POSTHOG_API_KEY \
             --env POSTHOG_PROJECT_ID=$POSTHOG_PROJECT_ID \
             --env KV_REST_API_URL=$KV_REST_API_URL \
-            --env KV_REST_API_TOKEN=$KV_REST_API_TOKEN
+            --env KV_REST_API_TOKEN=$KV_REST_API_TOKEN \
+            --env LINEAR_CLIENT_ID=$LINEAR_CLIENT_ID \
+            --env LINEAR_CLIENT_SECRET=$LINEAR_CLIENT_SECRET \
+            --env LINEAR_WEBHOOK_SECRET=$LINEAR_WEBHOOK_SECRET \
+            --env QSTASH_TOKEN=$QSTASH_TOKEN \
+            --env QSTASH_CURRENT_SIGNING_KEY=$QSTASH_CURRENT_SIGNING_KEY \
+            --env QSTASH_NEXT_SIGNING_KEY=$QSTASH_NEXT_SIGNING_KEY
 
   deploy-web:
     name: Deploy Web to Vercel

apps/api/package.json

@@ -13,16 +13,19 @@
 	"dependencies": {
 		"@clerk/backend": "^2.27.0",
 		"@clerk/nextjs": "^6.36.2",
+		"@linear/sdk": "^68.1.0",
 		"@sentry/nextjs": "^10.32.1",
 		"@superset/db": "workspace:*",
 		"@superset/shared": "workspace:*",
 		"@superset/trpc": "workspace:*",
 		"@t3-oss/env-nextjs": "^0.13.8",
 		"@trpc/server": "^11.7.1",
+		"@upstash/qstash": "^2.8.4",
 		"@vercel/blob": "^2.0.0",
 		"drizzle-orm": "0.45.1",
 		"import-in-the-middle": "2.0.1",
 		"jose": "^6.1.3",
+		"lodash.chunk": "^4.2.0",
 		"next": "^16.0.10",
 		"react": "^19.2.3",
 		"react-dom": "^19.2.3",
@@ -31,6 +34,7 @@
 	},
 	"devDependencies": {
 		"@superset/typescript": "workspace:*",
+		"@types/lodash.chunk": "^4.2.9",
 		"@types/node": "^24.9.1",
 		"@types/react": "^19.2.7",
 		"@types/react-dom": "^19.2.3",

apps/api/src/app/api/integrations/linear/callback/route.ts

@@ -0,0 +1,109 @@
+import { LinearClient } from "@linear/sdk";
+import { db } from "@superset/db/client";
+import { integrationConnections } from "@superset/db/schema";
+import { Client } from "@upstash/qstash";
+import { z } from "zod";
+import { env } from "@/env";
+
+const qstash = new Client({ token: env.QSTASH_TOKEN });
+
+const stateSchema = z.object({
+	organizationId: z.string().min(1),
+	userId: z.string().min(1),
+});
+
+export async function GET(request: Request) {
+	const url = new URL(request.url);
+	const code = url.searchParams.get("code");
+	const state = url.searchParams.get("state");
+	const error = url.searchParams.get("error");
+
+	if (error) {
+		return Response.redirect(
+			`${env.NEXT_PUBLIC_WEB_URL}/integrations/linear?error=oauth_denied`,
+		);
+	}
+
+	if (!code || !state) {
+		return Response.redirect(
+			`${env.NEXT_PUBLIC_WEB_URL}/integrations/linear?error=missing_params`,
+		);
+	}
+
+	const parsed = stateSchema.safeParse(
+		JSON.parse(Buffer.from(state, "base64url").toString("utf-8")),
+	);
+
+	if (!parsed.success) {
+		return Response.redirect(
+			`${env.NEXT_PUBLIC_WEB_URL}/integrations/linear?error=invalid_state`,
+		);
+	}
+
+	const { organizationId, userId } = parsed.data;
+
+	const tokenResponse = await fetch("https://api.linear.app/oauth/token", {
+		method: "POST",
+		headers: { "Content-Type": "application/x-www-form-urlencoded" },
+		body: new URLSearchParams({
+			grant_type: "authorization_code",
+			client_id: env.LINEAR_CLIENT_ID,
+			client_secret: env.LINEAR_CLIENT_SECRET,
+			redirect_uri: `${env.NEXT_PUBLIC_API_URL}/api/integrations/linear/callback`,
+			code,
+		}),
+	});
+
+	if (!tokenResponse.ok) {
+		return Response.redirect(
+			`${env.NEXT_PUBLIC_WEB_URL}/integrations/linear?error=token_exchange_failed`,
+		);
+	}
+
+	const tokenData: { access_token: string; expires_in?: number } =
+		await tokenResponse.json();
+
+	const linearClient = new LinearClient({
+		accessToken: tokenData.access_token,
+	});
+	const viewer = await linearClient.viewer;
+	const linearOrg = await viewer.organization;
+
+	const tokenExpiresAt = tokenData.expires_in
+		? new Date(Date.now() + tokenData.expires_in * 1000)
+		: null;
+
+	await db
+		.insert(integrationConnections)
+		.values({
+			organizationId,
+			connectedByUserId: userId,
+			provider: "linear",
+			accessToken: tokenData.access_token,
+			tokenExpiresAt,
+			externalOrgId: linearOrg.id,
+			externalOrgName: linearOrg.name,
+		})
+		.onConflictDoUpdate({
+			target: [
+				integrationConnections.organizationId,
+				integrationConnections.provider,
+			],
+			set: {
+				accessToken: tokenData.access_token,
+				tokenExpiresAt,
+				externalOrgId: linearOrg.id,
+				externalOrgName: linearOrg.name,
+				connectedByUserId: userId,
+				updatedAt: new Date(),
+			},
+		});
+
+	await qstash.publishJSON({
+		url: `${env.NEXT_PUBLIC_API_URL}/api/integrations/linear/jobs/initial-sync`,
+		body: { organizationId, creatorUserId: userId },
+		retries: 3,
+	});
+
+	return Response.redirect(`${env.NEXT_PUBLIC_WEB_URL}/integrations/linear`);
+}

apps/api/src/app/api/integrations/linear/connect/route.ts

@@ -0,0 +1,61 @@
+import { auth } from "@clerk/nextjs/server";
+import { db } from "@superset/db/client";
+import { organizationMembers, users } from "@superset/db/schema";
+import { and, eq } from "drizzle-orm";
+import { env } from "@/env";
+
+export async function GET(request: Request) {
+	const { userId: clerkUserId } = await auth();
+
+	if (!clerkUserId) {
+		return Response.json({ error: "Unauthorized" }, { status: 401 });
+	}
+
+	const url = new URL(request.url);
+	const organizationId = url.searchParams.get("organizationId");
+
+	if (!organizationId) {
+		return Response.json(
+			{ error: "Missing organizationId parameter" },
+			{ status: 400 },
+		);
+	}
+
+	const user = await db.query.users.findFirst({
+		where: eq(users.clerkId, clerkUserId),
+	});
+
+	if (!user) {
+		return Response.json({ error: "User not found" }, { status: 404 });
+	}
+
+	const membership = await db.query.organizationMembers.findFirst({
+		where: and(
+			eq(organizationMembers.organizationId, organizationId),
+			eq(organizationMembers.userId, user.id),
+		),
+	});
+
+	if (!membership) {
+		return Response.json(
+			{ error: "User is not a member of this organization" },
+			{ status: 403 },
+		);
+	}
+
+	const state = Buffer.from(
+		JSON.stringify({ organizationId, userId: user.id }),
+	).toString("base64url");
+
+	const linearAuthUrl = new URL("https://linear.app/oauth/authorize");
+	linearAuthUrl.searchParams.set("client_id", env.LINEAR_CLIENT_ID);
+	linearAuthUrl.searchParams.set(
+		"redirect_uri",
+		`${env.NEXT_PUBLIC_API_URL}/api/integrations/linear/callback`,
+	);
+	linearAuthUrl.searchParams.set("response_type", "code");
+	linearAuthUrl.searchParams.set("scope", "read,write,issues:create");
+	linearAuthUrl.searchParams.set("state", state);
+
+	return Response.redirect(linearAuthUrl.toString());
+}

apps/api/src/app/api/integrations/linear/jobs/initial-sync/route.ts

@@ -0,0 +1,126 @@
+import { LinearClient } from "@linear/sdk";
+import { buildConflictUpdateColumns, db } from "@superset/db";
+import { integrationConnections, tasks, users } from "@superset/db/schema";
+import { Receiver } from "@upstash/qstash";
+import { and, eq, inArray } from "drizzle-orm";
+import chunk from "lodash.chunk";
+import { z } from "zod";
+import { env } from "@/env";
+import { fetchAllIssues, mapIssueToTask } from "./utils";
+
+const BATCH_SIZE = 100;
+
+const receiver = new Receiver({
+	currentSigningKey: env.QSTASH_CURRENT_SIGNING_KEY,
+	nextSigningKey: env.QSTASH_NEXT_SIGNING_KEY,
+});
+
+const payloadSchema = z.object({
+	organizationId: z.string().min(1),
+	creatorUserId: z.string().min(1),
+});
+
+export async function POST(request: Request) {
+	const body = await request.text();
+	const signature = request.headers.get("upstash-signature");
+
+	if (!signature) {
+		return Response.json({ error: "Missing signature" }, { status: 401 });
+	}
+
+	const isValid = await receiver.verify({
+		body,
+		signature,
+		url: `${env.NEXT_PUBLIC_API_URL}/api/integrations/linear/jobs/initial-sync`,
+	});
+
+	if (!isValid) {
+		return Response.json({ error: "Invalid signature" }, { status: 401 });
+	}
+
+	const parsed = payloadSchema.safeParse(JSON.parse(body));
+	if (!parsed.success) {
+		return Response.json({ error: "Invalid payload" }, { status: 400 });
+	}
+
+	const { organizationId, creatorUserId } = parsed.data;
+
+	const connection = await db.query.integrationConnections.findFirst({
+		where: and(
+			eq(integrationConnections.organizationId, organizationId),
+			eq(integrationConnections.provider, "linear"),
+		),
+	});
+
+	if (!connection) {
+		return Response.json({ error: "No connection found", skipped: true });
+	}
+
+	const client = new LinearClient({ accessToken: connection.accessToken });
+	await performInitialSync(client, organizationId, creatorUserId);
+
+	return Response.json({ success: true });
+}
+
+async function performInitialSync(
+	client: LinearClient,
+	organizationId: string,
+	creatorUserId: string,
+) {
+	const issues = await fetchAllIssues(client);
+
+	if (issues.length === 0) {
+		return;
+	}
+
+	const assigneeEmails = [
+		...new Set(
+			issues.map((i) => i.assignee?.email).filter((e): e is string => !!e),
+		),
+	];
+
+	const matchedUsers =
+		assigneeEmails.length > 0
+			? await db.query.users.findMany({
+					where: inArray(users.email, assigneeEmails),
+				})
+			: [];
+
+	const userByEmail = new Map(matchedUsers.map((u) => [u.email, u.id]));
+
+	const taskValues = issues.map((issue) =>
+		mapIssueToTask(issue, organizationId, creatorUserId, userByEmail),
+	);
+
+	const batches = chunk(taskValues, BATCH_SIZE);
+
+	for (const batch of batches) {
+		await db
+			.insert(tasks)
+			.values(batch)
+			.onConflictDoUpdate({
+				target: [tasks.externalProvider, tasks.externalId],
+				set: {
+					...buildConflictUpdateColumns(tasks, [
+						"slug",
+						"title",
+						"description",
+						"status",
+						"statusColor",
+						"statusType",
+						"priority",
+						"assigneeId",
+						"estimate",
+						"dueDate",
+						"labels",
+						"startedAt",
+						"completedAt",
+						"externalKey",
+						"externalUrl",
+						"lastSyncedAt",
+					]),
+					syncError: null,
+				},
+			});
+	}
+}