fix(integrations): enforce current task ownership

apps/api/src/gitea-integration/controllers/import-gitea-issues.ts

@@ -3,6 +3,7 @@ import { HTTPException } from "hono/http-exception";
 import db from "../../database";
 import {
   activityTable,
+  externalLinkTable,
   integrationTable,
   labelTable,
   projectTable,
@@ -14,6 +15,7 @@ import { isKaneoComment } from "../../plugins/gitea/utils/comment-origin";
 import {
   createGiteaClient,
   type GiteaIssue,
+  type GiteaComment,
   type GiteaLabel,
   type GiteaPullRequest,
 } from "../../plugins/gitea/utils/gitea-api";
@@ -29,6 +31,12 @@ import {
 import { formatTaskDescriptionFromIssue } from "../../plugins/github/utils/format";
 import { claimTaskNumber } from "../../task/controllers/claim-task-numbers";
 
+import {
+  type IntegrationDatabase,
+  linkedTaskScope,
+  withIntegrationTask,
+} from "../../plugins/github/services/integration-task-scope";
+
 type ImportResult = {
   imported: number;
   updated: number;
@@ -209,71 +217,104 @@ async function importSingleIssue(
   const priority = extractIssuePriority(adaptedLabels);
   const status = extractIssueStatus(adaptedLabels);
 
+  const comments = await fetchIssueComments(issue.number, config, client);
+
   if (existingLink) {
-    const updateData: Record<string, unknown> = {
-      title: issue.title,
-      description: formatTaskDescriptionFromIssue(issue.body),
-    };
+    const result = await withIntegrationTask(
+      existingLink.taskId,
+      { id: integrationId, projectId, project: { workspaceId } },
+      async (database) => {
+        const [linked] = await database
+          .select({ id: externalLinkTable.id })
+          .from(externalLinkTable)
+          .where(
+            and(
+              eq(externalLinkTable.id, existingLink.id),
+              eq(externalLinkTable.taskId, existingLink.taskId),
+              eq(externalLinkTable.integrationId, integrationId),
+            ),
+          )
+          .for("update");
+        if (!linked) return "skipped" as const;
 
-    if (priority) updateData.priority = priority;
-    if (status) updateData.status = status;
+        const updateData: Record<string, unknown> = {
+          title: issue.title,
+          description: formatTaskDescriptionFromIssue(issue.body),
+        };
 
-    await db
-      .update(taskTable)
-      .set(updateData)
-      .where(eq(taskTable.id, existingLink.taskId));
+        if (priority) updateData.priority = priority;
+        if (status) updateData.status = status;
 
-    await importLabelsForTask(labels, existingLink.taskId, workspaceId);
+        await database
+          .update(taskTable)
+          .set(updateData)
+          .where(linkedTaskScope(existingLink.taskId, projectId));
 
-    await importCommentsForTask(
-      issue.number,
-      existingLink.taskId,
-      config,
-      client,
-    );
+        await importLabelsForTask(
+          labels,
+          existingLink.taskId,
+          workspaceId,
+          database,
+        );
 
-    return "updated";
+        await importCommentsForTask(comments, existingLink.taskId, database);
+
+        return "updated" as const;
+      },
+    );
+    return result ?? "skipped";
   }
 
-  const createdTask = await db.transaction(async (tx) => {
-    const nextNumber = await claimTaskNumber(projectId, tx);
+  const createdTask = await withIntegrationTask(
+    null,
+    { id: integrationId, projectId, project: { workspaceId } },
+    async (tx) => {
+      const nextNumber = await claimTaskNumber(projectId, tx);
 
-    const taskValues: typeof taskTable.$inferInsert = {
-      projectId,
-      userId: null,
-      title: issue.title,
-      description: formatTaskDescriptionFromIssue(issue.body),
-      status: status || "to-do",
-      priority: priority ?? "low",
-      number: nextNumber,
-    };
+      const taskValues: typeof taskTable.$inferInsert = {
+        projectId,
+        userId: null,
+        title: issue.title,
+        description: formatTaskDescriptionFromIssue(issue.body),
+        status: status || "to-do",
+        priority: priority ?? "low",
+        number: nextNumber,
+      };
 
-    const [created] = await tx.insert(taskTable).values(taskValues).returning();
+      const [created] = await tx
+        .insert(taskTable)
+        .values(taskValues)
+        .returning();
 
-    if (!created) {
-      throw new Error("Failed to create task");
-    }
+      if (!created) {
+        throw new Error("Failed to create task");
+      }
 
-    return created;
-  });
+      await createExternalLink(
+        {
+          taskId: created.id,
+          integrationId,
+          resourceType: "issue",
+          externalId: issue.number.toString(),
+          url: issue.html_url,
+          title: issue.title,
+          metadata: {
+            state: issue.state,
+            createdFrom: "gitea-import",
+            author: issue.user?.login ?? issue.user?.username,
+          },
+        },
+        tx,
+      );
 
-  await createExternalLink({
-    taskId: createdTask.id,
-    integrationId,
-    resourceType: "issue",
-    externalId: issue.number.toString(),
-    url: issue.html_url,
-    title: issue.title,
-    metadata: {
-      state: issue.state,
-      createdFrom: "gitea-import",
-      author: issue.user?.login ?? issue.user?.username,
-    },
-  });
+      await importLabelsForTask(labels, created.id, workspaceId, tx);
 
-  await importLabelsForTask(labels, createdTask.id, workspaceId);
+      await importCommentsForTask(comments, created.id, tx);
 
-  await importCommentsForTask(issue.number, createdTask.id, config, client);
+      return created;
+    },
+  );
+  if (!createdTask) return "skipped";
 
   await publishEvent("task.created", {
     ...createdTask,
@@ -293,6 +334,7 @@ async function importLabelsForTask(
   issueLabels: GiteaIssue["labels"],
   taskId: string,
   workspaceId: string,
+  database: IntegrationDatabase = db,
 ): Promise<void> {
   const nonSystemLabels = (issueLabels ?? [])
     .map((label) => {
@@ -316,7 +358,7 @@ async function importLabelsForTask(
   const expectedNames = nonSystemLabels.map((label) => label.name);
 
   if (expectedNames.length > 0) {
-    await db
+    await database
       .delete(labelTable)
       .where(
         and(
@@ -325,10 +367,10 @@ async function importLabelsForTask(
         ),
       );
   } else {
-    await db.delete(labelTable).where(eq(labelTable.taskId, taskId));
+    await database.delete(labelTable).where(eq(labelTable.taskId, taskId));
   }
 
-  const existingLabelsOnTask = await db.query.labelTable.findMany({
+  const existingLabelsOnTask = await database.query.labelTable.findMany({
     where:
       expectedNames.length > 0
         ? and(
@@ -347,7 +389,7 @@ async function importLabelsForTask(
       continue;
     }
 
-    const existingWorkspaceLabel = await db.query.labelTable.findFirst({
+    const existingWorkspaceLabel = await database.query.labelTable.findFirst({
       where: and(
         eq(labelTable.workspaceId, workspaceId),
         eq(labelTable.name, labelData.name),
@@ -356,7 +398,7 @@ async function importLabelsForTask(
 
     const colorToUse = existingWorkspaceLabel?.color || labelData.color;
 
-    await db
+    await database
       .insert(labelTable)
       .values({
         name: labelData.name,
@@ -370,18 +412,12 @@ async function importLabelsForTask(
   }
 }
 
-async function importCommentsForTask(
+async function fetchIssueComments(
   issueNumber: number,
-  taskId: string,
   config: GiteaConfig,
   client: ReturnType<typeof createGiteaClient>,
-): Promise<void> {
-  const allComments: Array<{
-    id: number;
-    body: string;
-    html_url: string;
-    user?: { login?: string; username?: string; avatar_url?: string } | null;
-  }> = [];
+): Promise<GiteaComment[]> {
+  const allComments: GiteaComment[] = [];
   let page = 1;
 
   while (true) {
@@ -401,13 +437,21 @@ async function importCommentsForTask(
     page++;
   }
 
+  return allComments;
+}
+
+async function importCommentsForTask(
+  allComments: GiteaComment[],
+  taskId: string,
+  database: IntegrationDatabase,
+): Promise<void> {
   for (const comment of allComments) {
     const username = comment.user?.login ?? comment.user?.username ?? "";
     if (username.endsWith("[bot]") || isKaneoComment(comment.body)) {
       continue;
     }
 
-    await db
+    await database
       .insert(activityTable)
       .values({
         taskId,

apps/api/src/github-integration/controllers/import-issues.ts

@@ -2,6 +2,7 @@ import { createId } from "@paralleldrive/cuid2";
 import { and, asc, eq, isNull } from "drizzle-orm";
 import { HTTPException } from "hono/http-exception";
 import db from "../../database";
+import { publishEvent } from "../../events";
 import {
   activityTable,
   columnTable,
@@ -173,6 +174,17 @@ export async function importIssues(projectId: string, runId?: string) {
       }
       const currentRun: typeof githubImportTable.$inferSelect = run;
       run = await db.transaction(async (tx) => {
+        const [currentProject] = await tx
+          .select()
+          .from(projectTable)
+          .where(
+            and(
+              eq(projectTable.id, project.id),
+              eq(projectTable.workspaceId, project.workspaceId),
+            ),
+          )
+          .for("key share");
+        if (!currentProject) throw conflict();
         // Serialize against integration changes and webhook issue creation. No
         // provider request is made while this transaction holds row locks.
         const [currentIntegration] = await tx
@@ -201,6 +213,9 @@ export async function importIssues(projectId: string, runId?: string) {
         if (!saved) throw conflict();
         return saved;
       });
+      // Each bounded page is durable before other clients refresh, including
+      // continuation pages that change labels, comments or linked resources.
+      await publishEvent("project.updated", { projectId });
     }
     return importProgress(run.runId, run.state);
   });
@@ -305,13 +320,29 @@ async function applyPage(
   }
   const current = state.currentIssue;
   if (!current) throw new Error("Import continuation missing");
-  const task = await tx.query.taskTable.findFirst({
-    where: and(
-      eq(taskTable.id, current.taskId),
-      eq(taskTable.projectId, project.id),
-    ),
-  });
-  if (!task) {
+  const [task] = await tx
+    .select()
+    .from(taskTable)
+    .where(
+      and(
+        eq(taskTable.id, current.taskId),
+        eq(taskTable.projectId, project.id),
+      ),
+    )
+    .for("no key update");
+  const [linked] = await tx
+    .select({ id: externalLinkTable.id })
+    .from(externalLinkTable)
+    .where(
+      and(
+        eq(externalLinkTable.integrationId, integrationId),
+        eq(externalLinkTable.resourceType, "issue"),
+        eq(externalLinkTable.externalId, String(current.number)),
+        eq(externalLinkTable.taskId, current.taskId),
+      ),
+    )
+    .for("update");
+  if (!task || !linked) {
     state.skipped++;
     finishIssue(state);
     return;
@@ -408,13 +439,20 @@ async function importIssue(
   const priority = extractIssuePriority(issue.labels.nodes);
   const status = extractIssueStatus(issue.labels.nodes);
   if (link) {
-    const task = await tx.query.taskTable.findFirst({
-      where: and(
-        eq(taskTable.id, link.taskId),
-        eq(taskTable.projectId, projectId),
-      ),
-    });
+    const [task] = await tx
+      .select()
+      .from(taskTable)
+      .where(
+        and(eq(taskTable.id, link.taskId), eq(taskTable.projectId, projectId)),
+      )
+      .for("no key update");
     if (!task) return null;
+    const [linked] = await tx
+      .select({ id: externalLinkTable.id })
+      .from(externalLinkTable)
+      .where(eq(externalLinkTable.id, link.id))
+      .for("update");
+    if (!linked) return null;
     const [updated] = await tx
       .update(taskTable)
       .set({
@@ -540,6 +578,12 @@ async function linkPull(
     database: tx,
   });
   if (!task) return;
+  const [scopedTask] = await tx
+    .select({ id: taskTable.id })
+    .from(taskTable)
+    .where(and(eq(taskTable.id, task.id), eq(taskTable.projectId, project.id)))
+    .for("share");
+  if (!scopedTask) return;
   await tx.insert(externalLinkTable).values({
     taskId: task.id,
     integrationId,