feat: add s3 localstack, upload api

web/src/server/createRun.ts

@@ -1,11 +1,10 @@
 "use server";
 
-import { ComfyAPI_Run } from "../app/api/create-run/route";
 import { db } from "@/db/db";
 import { workflowRunsTable } from "@/db/schema";
+import { ComfyAPI_Run } from "@/types/ComfyAPI_Run";
 import { eq } from "drizzle-orm";
 import { revalidatePath } from "next/cache";
-import { NextResponse } from "next/server";
 import "server-only";
 
 export async function createRun(
@@ -18,9 +17,10 @@ export async function createRun(
   });
 
   if (!machine) {
-    return new Response("Machine not found", {
-      status: 404,
-    });
+    throw new Error("Machine not found");
+    // return new Response("Machine not found", {
+    //   status: 404,
+    // });
   }
 
   const workflow_version_data =
@@ -38,9 +38,10 @@ export async function createRun(
   //   })
   // : null;
   if (!workflow_version_data) {
-    return new Response("Workflow version not found", {
-      status: 404,
-    });
+    throw new Error("Workflow version not found");
+    // return new Response("Workflow version not found", {
+    //   status: 404,
+    // });
   }
 
   const comfyui_endpoint = `${machine.endpoint}/comfy-deploy/run`;
@@ -54,22 +55,22 @@ export async function createRun(
     body: JSON.stringify({
       workflow_api: workflow_version_data.workflow_api,
       status_endpoint: `${origin}/api/update-run`,
+      file_upload_endpoint: `${origin}/api/file-upload`,
     }),
-  })
-    .then(async (res) => ComfyAPI_Run.parseAsync(await res.json()))
-    .catch((error) => {
-      console.error(error);
-      return new Response(error.details, {
-        status: 500,
-      });
-    });
+  }).then(async (res) => ComfyAPI_Run.parseAsync(await res.json()));
+  // .catch((error) => {
+  //   console.error(error);
+  //   return new Response(error.details, {
+  //     status: 500,
+  //   });
+  // });
 
-  console.log(result);
+  // console.log(result);
 
-  // return the error
-  if (result instanceof Response) {
-    return result;
-  }
+  // // return the error
+  // if (result instanceof Response) {
+  //   return result;
+  // }
 
   // Add to our db
   const workflow_run = await db
@@ -84,12 +85,14 @@ export async function createRun(
 
   revalidatePath(`/${workflow_version_data.workflow_id}`);
 
-  return NextResponse.json(
-    {
-      workflow_run_id: workflow_run[0].id,
-    },
-    {
-      status: 200,
-    }
-  );
+  return workflow_run[0].id;
+
+  // return NextResponse.json(
+  //   {
+  //     workflow_run_id: workflow_run[0].id,
+  //   },
+  //   {
+  //     status: 200,
+  //   }
+  // );
 }

web/src/server/curdMachine.ts

@@ -40,7 +40,17 @@ export async function addMachine(name: string, endpoint: string) {
   revalidatePath("/machines");
 }
 
-export async function deleteMachine(machine_id: string) {
-  await db.delete(machinesTable).where(eq(machinesTable.id, machine_id));
-  revalidatePath("/machines");
+export async function deleteMachine(
+  machine_id: string
+): Promise<{ message: string; error?: boolean }> {
+  try {
+    await db.delete(machinesTable).where(eq(machinesTable.id, machine_id));
+    revalidatePath("/machines");
+    return { message: "Machine Deleted" };
+  } catch (error: unknown) {
+    return {
+      message: `Error: ${error.detail}`,
+      error: true,
+    };
+  }
 }

web/src/server/findAllRuns.tsx

@@ -0,0 +1,23 @@
+import { db } from "@/db/db";
+import { workflowRunsTable } from "@/db/schema";
+import { desc, eq } from "drizzle-orm";
+
+export async function findAllRuns(workflow_id: string) {
+  return await db.query.workflowRunsTable.findMany({
+    where: eq(workflowRunsTable.workflow_id, workflow_id),
+    orderBy: desc(workflowRunsTable.created_at),
+    with: {
+      machine: {
+        columns: {
+          name: true,
+          endpoint: true,
+        },
+      },
+      version: {
+        columns: {
+          version: true,
+        },
+      },
+    },
+  });
+}

web/src/server/findFirstTableWithVersion.tsx

@@ -0,0 +1,10 @@
+import { db } from "@/db/db";
+import { workflowTable, workflowVersionTable } from "@/db/schema";
+import { desc, eq } from "drizzle-orm";
+
+export async function findFirstTableWithVersion(workflow_id: string) {
+  return await db.query.workflowTable.findFirst({
+    with: { versions: { orderBy: desc(workflowVersionTable.version) } },
+    where: eq(workflowTable.id, workflow_id),
+  });
+}

web/src/server/getRunsOutput.tsx

@@ -0,0 +1,12 @@
+"use server";
+
+import { db } from "@/db/db";
+import { workflowRunOutputs } from "@/db/schema";
+import { eq } from "drizzle-orm";
+
+export async function getRunsOutput(run_id: string) {
+  return await db
+    .select()
+    .from(workflowRunOutputs)
+    .where(eq(workflowRunOutputs.run_id, run_id));
+}

web/src/server/resource.ts

@@ -0,0 +1,106 @@
+import type { PutObjectCommandInput } from "@aws-sdk/client-s3";
+import {
+  DeleteObjectCommand,
+  GetObjectCommand,
+  PutObjectCommand,
+  S3,
+} from "@aws-sdk/client-s3";
+import { getSignedUrl } from "@aws-sdk/s3-request-presigner";
+
+const s3Client = new S3({
+  endpoint: process.env.SPACES_ENDPOINT, //"https://nyc3.digitaloceanspaces.com",
+  region: process.env.SPACES_REGION, //"nyc3",
+  credentials: {
+    accessKeyId: process.env.SPACES_KEY!,
+    secretAccessKey: process.env.SPACES_SECRET!,
+  },
+  forcePathStyle: true,
+});
+
+function replaceCDNUrl(url: string) {
+  url = url.replace(
+    process.env.SPACES_ENDPOINT!,
+    process.env.SPACES_ENDPOINT_CDN!
+  );
+  return url;
+}
+
+export type ResourceObject = {
+  resourceBucket: string;
+  resourceId: string;
+  resourceType: "image/png" | "application/zip" | string;
+  isPublic?: boolean;
+};
+
+export async function handleResourceUpload(
+  resource: Partial<ResourceObject>
+): Promise<string> {
+  const p: PutObjectCommandInput = {
+    Key: resource.resourceId,
+    Bucket: resource.resourceBucket,
+    ContentType: resource.resourceType,
+  };
+
+  // Only set ACL if resource is public
+  if (resource.isPublic) {
+    p.ACL = "public-read";
+  }
+
+  const url = await getSignedUrl(s3Client, new PutObjectCommand(p), {
+    expiresIn: 5 * 60,
+  });
+
+  return url;
+}
+
+export async function handResourceRemove(
+  resource: Partial<ResourceObject>
+): Promise<boolean> {
+  console.log("Removing resources", resource);
+  try {
+    const result = await s3Client.send(
+      new DeleteObjectCommand({
+        Key: `/public-download/sdk/${resource.resourceId}`,
+        Bucket: resource.resourceBucket,
+      })
+    );
+    console.log(result);
+  } catch (err) {
+    console.log("Error", err);
+    return false;
+  }
+  return true;
+}
+
+export async function handleResourceDownload(
+  resource: Partial<ResourceObject>
+): Promise<string> {
+  const url = await getSignedUrl(
+    s3Client,
+    new GetObjectCommand({
+      Key: resource.resourceId,
+      Bucket: resource.resourceBucket,
+
+      ResponseCacheControl: "no-cache, no-store",
+    }),
+    { expiresIn: 5 * 60 }
+  );
+  return replaceCDNUrl(url);
+}
+
+export async function handleResourceDelete(
+  resource: ResourceObject
+): Promise<string> {
+  try {
+    const result = await s3Client.send(
+      new DeleteObjectCommand({
+        Key: resource.resourceId,
+        Bucket: resource.resourceBucket,
+      })
+    );
+  } catch (e) {
+    //TODO handle error
+    return "error";
+  }
+  return "ok";
+}