stop processing files when ever he/she wants

.dockerignore

@@ -5,3 +5,6 @@ npm-debug.log
 *.md
 .next
 .git
+cypress.config.*
+.cypress/*
+cypress/

DataSource/index.tsx

@@ -7,6 +7,7 @@ import { ConfigDropbox } from './Dropbox/ConfigDropbox';
 import { DropboxPickerProvider } from './Dropbox/DropboxPicker/dropbox-picker.context';
 import { ConfigGoogleDrive } from './GoogleDrive/ConfigGoogleDrive';
 import { ConfigAws } from './Aws/ConfigAws';
+import { StopConnection } from '@/components/StopConnection/StopConnection';
 
 export const DataSource = ({ connection, token, status }: { connection: ConnectionQuery, token: string | undefined | null, status: "PROCESSING" | "FINISHED" | undefined }) => {
   switch (connection.service) {
@@ -15,11 +16,13 @@ export const DataSource = ({ connection, token, status }: { connection: Connecti
         {connection.isConfigSet && <SyncConnection connection={connection} status={status} />}
         <ConfigGoogleDrive connection={connection} token={token} status={status} />
         <DeleteConnection connection={connection} status={status} />
+        <StopConnection connection={connection} status={status} />
       </>
     case "DIRECT_UPLOAD":
       return <>
         <UpdateConfigDirect connection={connection} status={status} />
         <DeleteConnection connection={connection} status={status} />
+        <StopConnection connection={connection} status={status} />
       </>
     case 'DROPBOX':
       return <>
@@ -28,17 +31,20 @@ export const DataSource = ({ connection, token, status }: { connection: Connecti
           <ConfigDropbox connection={connection} status={status} token={token} />
         </DropboxPickerProvider>
         <DeleteConnection connection={connection} status={status} />
+        <StopConnection connection={connection} status={status} />
       </>
-    case 'AWS': 
+    case 'AWS':
       return <>
-      {connection.isConfigSet && <SyncConnection connection={connection} status={status} />}
-      <ConfigAws connection={connection} status={status} token={token} />
-      <DeleteConnection connection={connection} status={status} />
-    </>
+        {connection.isConfigSet && <SyncConnection connection={connection} status={status} />}
+        <ConfigAws connection={connection} status={status} token={token} />
+        <DeleteConnection connection={connection} status={status} />
+        <StopConnection connection={connection} status={status} />
+      </>
     default:
       return <>
         {connection.isConfigSet && <SyncConnection connection={connection} status={status} />}
         <DeleteConnection connection={connection} status={status} />
+        <StopConnection connection={connection} status={status} />
       </>
   }
 }

actions/connctions/set/index.ts

@@ -1,5 +1,7 @@
 "use server"
 import { authOptions } from "@/auth";
+import { databaseDrizzle } from "@/db";
+import { connections } from "@/db/schemas/connections";
 import { setConnectionToProcess } from "@/fileProcessors/connectors";
 import { fromErrorToFormState, toFormState } from "@/lib/zodErrorHandle";
 import { addToProcessFilesQueue } from "@/workers/queues/jobs/processFiles.job";
@@ -17,7 +19,10 @@ export async function setConnectionConfig(_: FormState, formData: FormData) {
 
     formData.set("userId", session.user.id)
     const config = await setConnectionToProcess(formData)
-    await addToProcessFilesQueue(config)
+    const jobId = await addToProcessFilesQueue(config)
+    await databaseDrizzle.update(connections).set({
+      jobId: jobId
+    })
     revalidatePath("/connections");
     return toFormState("SUCCESS", "start processing");
   } catch (e) {

actions/connctions/stop/index.ts

@@ -0,0 +1,31 @@
+"use server"
+import { authOptions } from "@/auth";
+import { databaseDrizzle } from "@/db";
+import { fromErrorToFormState, toFormState } from "@/lib/zodErrorHandle";
+import { redisConnection } from "@/workers/redis";
+import { getServerSession } from "next-auth";
+
+type FormState = {
+  message: string;
+};
+
+export async function stopProcessing(_: FormState, formData: FormData) {
+  const session = await getServerSession(authOptions);
+  try {
+    if (!session?.user?.id) throw new Error("forbidden");
+    const connectionId = formData.get("connectionId");
+    if (!connectionId) throw new Error("Missing connection id")
+
+    const conn = await databaseDrizzle.query.connections.findFirst({
+      where: (c, u) => u.eq(c.id, connectionId.toString()),
+      columns: {
+        jobId: true,
+      }
+    })
+    if (!conn || !conn.jobId) throw new Error("Connection Not Found or connection not processing in this time")
+    await redisConnection.set(`cancel-job:${conn.jobId}`, '1'); 
+    return toFormState("SUCCESS", "stop processing");
+  } catch (e) {
+    return fromErrorToFormState(e);
+  }
+}

actions/connctions/sync/index.ts

@@ -57,10 +57,6 @@ export const syncConnectionConfig = async (_: FormState, formData: FormData) =>
     })
     if (!user) throw new Error("no such account")
 
-    await databaseDrizzle.update(connections).set({
-      isSyncing: true,
-    }).where(eq(connections.id, connectionId));
-
     const { currentConn, othersConn } = user.connections.reduce((acc, c) => {
       if (c.id === connectionId) {
         acc.currentConn = c;
@@ -71,8 +67,7 @@ export const syncConnectionConfig = async (_: FormState, formData: FormData) =>
     }, { currentConn: null as Conn | null, othersConn: [] as Conn[] });
 
     if (!currentConn) throw new Error("no such connection")
-
-    await addToProcessFilesQueue({
+    const jobId = await addToProcessFilesQueue({
       connectionId: connectionId,
       service: currentConn.service,
       metadata: currentConn.metadata || null,
@@ -81,6 +76,10 @@ export const syncConnectionConfig = async (_: FormState, formData: FormData) =>
       files: [],
       links: []
     })
+    await databaseDrizzle.update(connections).set({
+      isSyncing: true,
+      jobId: jobId
+    }).where(eq(connections.id, connectionId));
 
     revalidatePath("/connections");
     return toFormState("SUCCESS", "start processing");

app/api/connections/[id]/files/route.ts

@@ -85,6 +85,8 @@ export async function DELETE(request: NextRequest, { params }: Params) {
       where: (c, ops) => ops.and(ops.eq(c.userId, userId), ops.eq(c.id, id)),
       columns: {
         id: true,
+        isSyncing: true,
+        jobId: true,
       },
     })
 
@@ -94,6 +96,16 @@ export async function DELETE(request: NextRequest, { params }: Params) {
         message: "Connection not found or access denied",
       }, { status: 403 })
     }
+
+    if (connection.isSyncing || connection.jobId) {
+      return NextResponse.json(
+        {
+          code: 'processing_in_progress',
+          message: 'Cannot delete while files are being processed. Please cancel the active processing operation first.'
+        },
+        { status: 409 }
+      )
+    }
     const filesToDelete = validation.data.file
       ? [validation.data.file]
       : validation.data.files ?? [];

app/api/connections/[id]/route.ts

@@ -107,7 +107,10 @@ export async function DELETE(request: NextRequest, { params }: Params) {
     const { id } = await params
     const { data: conn, error: deleteError } = await tryAndCatch(databaseDrizzle.query.connections.findFirst({
       where: (conn, ops) => ops.and(ops.eq(conn.id, id), ops.eq(conn.userId, userId)),
-      columns: {},
+      columns: {
+        jobId: true,
+        isSyncing: true,
+      },
       with: {
         files: {
           columns: {
@@ -136,16 +139,25 @@ export async function DELETE(request: NextRequest, { params }: Params) {
       )
     }
 
+    if (conn.isSyncing || conn.jobId) {
+      return NextResponse.json(
+        {
+          code: 'processing_in_progress',
+          message: 'Cannot delete while files are being processed. Please cancel the active processing operation first.'
+        },
+        { status: 409 }
+      )
+    }
     for (const { chunksIds } of conn.files) {
-      await qdrantCLient.delete(qdrant_collection_name, {
+      await tryAndCatch(qdrantCLient.delete(qdrant_collection_name, {
         points: chunksIds,
         wait: wait === "true",
-      })
+      }))
     }
 
-    await databaseDrizzle
+    await tryAndCatch(databaseDrizzle
       .delete(connections)
-      .where(eq(connections.id, id))
+      .where(eq(connections.id, id)))
 
     return NextResponse.json({
       code: "ok",
@@ -246,10 +258,10 @@ export async function PUT(request: NextRequest, { params }: Params) {
 
       let error: Error | null = null;
       if (links.length > 0 || files.length > 0) {
-        const {error:err}  = await tryAndCatch(directProcessFiles(filesConfig))
+        const { error: err } = await tryAndCatch(directProcessFiles(filesConfig))
         error = err;
       } else {
-        const {error:err} = await tryAndCatch(connectionProcessFiles(filesConfig))
+        const { error: err } = await tryAndCatch(connectionProcessFiles(filesConfig))
         error = err
       }
 

app/api/connections/[id]/stop/route.ts

@@ -0,0 +1,78 @@
+import { databaseDrizzle } from "@/db"
+import { checkAuth } from "@/lib/api_key"
+import { tryAndCatch } from "@/lib/try-catch"
+import { redisConnection } from "@/workers/redis"
+import { NextRequest, NextResponse } from "next/server"
+import { APIError } from "openai"
+
+
+type Params = {
+  params: Promise<{
+    id: string
+  }>
+}
+
+export async function POST(request: NextRequest, { params }: Params) {
+  try {
+    const {
+      data: userId,
+      error: authError,
+    } = await tryAndCatch<string, APIError>(checkAuth(request))
+    if (authError) {
+      return NextResponse.json({
+        code: authError.code,
+        message: authError.message,
+      }, { status: authError.status })
+    }
+
+    const { id } = await params
+    const { data: conn, error: queryError } = await tryAndCatch(databaseDrizzle.query.connections.findFirst({
+      where: (conn, ops) => ops.and(ops.eq(conn.userId, userId!), ops.eq(conn.id, id)),
+      columns: {
+        jobId: true,
+      }
+    }))
+    if (queryError) {
+      return NextResponse.json(
+        {
+          code: "internal_server_error",
+          message: "Failed to load connection status",
+        },
+        { status: 500 },
+      )
+    }
+    if (!conn) {
+      return NextResponse.json(
+        { code: 'not_found', message: 'Connection not found' },
+        { status: 404 }
+      )
+    }
+    if (!conn.jobId) {
+      return NextResponse.json(
+        { code: 'not_processing', message: 'No active processing job' },
+        { status: 400 }
+      )
+    }
+    const { error: redisError } = await tryAndCatch(
+      redisConnection.set(`cancel-job:${conn.jobId}`, '1')
+    )
+
+    if (redisError) {
+      return NextResponse.json(
+        { code: 'internal_error', message: 'Failed to cancel processing' },
+        { status: 500 }
+      )
+    }
+
+    return NextResponse.json(
+      { code: 'ok', message: 'Processing cancellation requested' },
+      { status: 200 }
+    )
+
+  } catch (error: any) {
+    return NextResponse.json(
+      { code: "internal_server_error", message: error.message },
+      { status: 500 },
+    );
+  }
+}

components/Connections/Connections.tsx

@@ -19,9 +19,11 @@ import {
   TooltipProvider,
   TooltipTrigger,
 } from "@/components/ui/tooltip"
+import { useRouter } from "next/navigation";
 
 export default function Connections({ connections, tokens }: { connections: ConnectionQuery[], tokens: ConnectionToken }) {
   const [isMounted, setIsMounted] = useState(false)
+  const route = useRouter()
   const [connProgress, setConnProgress] = useState<ConnectionProgress | null>(null);
 
   useEffect(() => {
@@ -41,6 +43,12 @@ export default function Connections({ connections, tokens }: { connections: Conn
 
   }, [])
 
+  useEffect(() => {
+    if (connProgress?.status === 'FINISHED') {
+      route.refresh();
+    }
+  }, [connProgress]);
+
   return connections.map(connection => {
     const progress = connection.id === connProgress?.connectionId ? connProgress : null;
 

components/Connectors/Connectors.tsx

@@ -3,14 +3,18 @@ import { FaGoogleDrive, FaDropbox } from "react-icons/fa";
 import { ConnectionBtn } from './ConnectionBtn';
 import { SiAwslambda } from 'react-icons/si';
 
+const isCloud = process.env.DCUP_ENV === 'CLOUD';
+
 export const Connectors = async () => {
   const connectors = [
-    {
-      id: 'google-drive',
-      name: 'Google Drive',
-      icon: <FaGoogleDrive className="w-6 h-6" />,
-      description: 'Connect your Google Drive to access documents and files',
-    },
+    ...(!isCloud ? [
+      {
+        id: 'google-drive',
+        name: 'Google Drive',
+        icon: <FaGoogleDrive className="w-6 h-6" />,
+        description: 'Connect your Google Drive to access documents and files',
+      },
+    ] : []),
     {
       id: "dropbox",
       name: "Dropbox",

components/StopConnection/StopConnection.tsx

@@ -0,0 +1,46 @@
+"use client"
+import { Button } from "../ui/button"
+import { Pause } from "lucide-react"
+import { useTransition } from "react"
+import { toast } from "@/hooks/use-toast"
+import { EMPTY_FORM_STATE } from "@/lib/zodErrorHandle"
+import { ConnectionQuery } from "@/app/(protected)/connections/page"
+import { stopProcessing } from "@/actions/connctions/stop"
+
+
+export const StopConnection = ({ connection, status }: {
+  connection: ConnectionQuery,
+  status: "PROCESSING" | "FINISHED" | undefined
+}) => {
+  const [isPending, startTransition] = useTransition();
+
+  const handleStopConnection = () => {
+    startTransition(async () => {
+      try {
+        const formData = new FormData();
+        formData.set("connectionId", connection.id)
+        const res = await stopProcessing(EMPTY_FORM_STATE, formData)
+        if (res.status !== 'SUCCESS') {
+          throw new Error(res.message)
+        }
+        toast({
+          title: res.message,
+        });
+
+      } catch (error: any) {
+        toast({
+          variant: "destructive",
+          title: "Uh oh! Something went wrong.",
+          description: error.message || "An unexpected error occurred.",
+        });
+      }
+    })
+  }
+
+  return (
+    <Button data-test={`stop-connection`} onClick={handleStopConnection} size='sm' variant={'ghost'} disabled={isPending || status === 'FINISHED' || (!status && !connection.isSyncing)}>
+      <Pause/>
+      Stop
+    </Button>
+  )
+}