✨ Backend for block insert

next/prisma/schema.prisma

@@ -178,13 +178,15 @@ model WorkflowNode {
 }
 
 model NodeBlock {
-  id               String @id @default(cuid())
-  workflow_node_id String
+  id       String @id @default(cuid())
+  node_ref String @db.VarChar(12)
+  node_id  String
 
   type  String
   input Json
 
-  workflow_node WorkflowNode @relation(fields: [workflow_node_id], references: [id], onDelete: Cascade)
+  workflow_node WorkflowNode @relation(fields: [node_id], references: [id], onDelete: Cascade)
 
+  @@unique([node_id, node_ref])
   @@map("node_block")
 }

next/src/components/sidebar/links.tsx

@@ -24,6 +24,14 @@ export const PAGE_LINKS: LinkMetadata[] = [
     icon: FaHome,
     className: "group-hover:text-color-secondary",
   },
+  // TODO: Uncomment once enabled
+  // {
+  //   name: "Flows",
+  //   href: "/workflow",
+  //   icon: FaWater,
+  //   badge: "Alpha",
+  //   className: "transition-transform group-hover:scale-110",
+  // },
   {
     name: "Templates",
     href: "/templates",

next/src/pages/workflow/index.tsx

@@ -1,16 +1,16 @@
 import type { GetStaticProps, NextPage } from "next";
 import { useQuery } from "@tanstack/react-query";
-import { useSession } from "next-auth/react";
 import { useRouter } from "next/router";
 import WorkflowApi from "../../services/workflow/workflowApi";
 import { languages } from "../../utils/languages";
 import { serverSideTranslations } from "next-i18next/serverSideTranslations";
 import nextI18NextConfig from "../../../next-i18next.config";
 import DashboardLayout from "../../layout/dashboard";
 import EmptyWorkflowButton from "../../components/workflow/EmptyWorkflow";
+import { useAuth } from "../../hooks/useAuth";
 
 const WorkflowList: NextPage = () => {
-  const { data: session } = useSession();
+  const { session, signIn } = useAuth();
   const router = useRouter();
 
   const api = new WorkflowApi(session?.accessToken);
@@ -38,6 +38,11 @@ const WorkflowList: NextPage = () => {
         ))}
         <EmptyWorkflowButton
           onClick={() => {
+            if (!session) {
+              signIn().catch(console.error);
+              return;
+            }
+
             api
               .create({
                 name: "New Workflow",

platform/reworkd_platform/db/base.py

@@ -3,10 +3,9 @@
 from typing import Optional, Type, TypeVar
 
 from fastapi import HTTPException
-from pydantic import BaseModel
-from sqlalchemy import String, DateTime, func
+from sqlalchemy import DateTime, String, func
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 
 from reworkd_platform.db.meta import meta
 
@@ -64,10 +63,6 @@ def mark_deleted(self) -> "TrackedModel":
         self.delete_date = datetime.now()
         return self
 
-    def to_schema(self) -> BaseModel:
-        """Converts the model to a schema."""
-        raise NotImplementedError
-
 
 class UserMixin:
     user_id = mapped_column(String, name="user_id", nullable=False)

platform/reworkd_platform/db/crud/edge.py

@@ -0,0 +1,50 @@
+from typing import Dict, List, Set
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from reworkd_platform.db.models.workflow import WorkflowEdgeModel
+from reworkd_platform.web.api.workflow.schemas import EdgeUpsert
+
+
+class EdgeCRUD:
+    def __init__(self, session: AsyncSession) -> None:
+        self.session = session
+        self.source_target_map: Dict[str, Set[str]] = {}
+
+    def add_edge(self, source: str, target: str) -> bool:
+        if source not in self.source_target_map:
+            self.source_target_map[source] = set()
+
+        targets = self.source_target_map[source]
+        if target in targets:
+            return False
+
+        targets.add(target)
+        return True
+
+    async def get_edges(self, workflow_id: str) -> Dict[str, WorkflowEdgeModel]:
+        query = select(WorkflowEdgeModel).where(
+            WorkflowEdgeModel.workflow_id == workflow_id
+        )
+
+        return {
+            edge.id: edge
+            for edge in (await self.session.execute(query)).scalars().all()
+        }
+
+    async def create_edge(self, edge_upsert: EdgeUpsert, workflow_id: str) -> None:
+        await WorkflowEdgeModel(
+            workflow_id=workflow_id,
+            source=edge_upsert.source,
+            target=edge_upsert.target,
+        ).save(self.session)
+
+    async def delete_old_edges(
+        self, edges_to_keep: List[EdgeUpsert], all_edges: Dict[str, WorkflowEdgeModel]
+    ) -> None:
+        edges_to_keep_ids = {e.id for e in edges_to_keep}
+
+        for edge in all_edges.values():
+            if edge.id not in edges_to_keep_ids:
+                await edge.delete(self.session)

platform/reworkd_platform/db/crud/node.py

@@ -0,0 +1,84 @@
+from typing import Dict, List
+
+from sqlalchemy import and_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from reworkd_platform.db.models.workflow import NodeBlockModel, WorkflowNodeModel
+from reworkd_platform.web.api.workflow.schemas import NodeUpsert
+
+
+class NodeCRUD:
+    def __init__(self, session: AsyncSession):
+        self.session = session
+
+    async def get_nodes(self, workflow_id: str) -> Dict[str, WorkflowNodeModel]:
+        query = select(WorkflowNodeModel).where(
+            and_(
+                WorkflowNodeModel.workflow_id == workflow_id,
+                WorkflowNodeModel.delete_date.is_(None),
+            )
+        )
+
+        return {
+            node.id: node
+            for node in (await self.session.execute(query)).scalars().all()
+        }
+
+    async def get_node_blocks(self, node_refs: List[str]) -> Dict[str, NodeBlockModel]:
+        """
+        Returns an object mapping node_ref to NodeBlockModel
+        """
+        query = select(NodeBlockModel).where(
+            NodeBlockModel.node_ref.in_(node_refs),
+        )
+
+        return {
+            block.node_ref: block
+            for block in (await self.session.execute(query)).scalars().all()
+        }
+
+    async def create_node_with_block(
+        self, n: NodeUpsert, workflow_id: str
+    ) -> WorkflowNodeModel:
+        node = await WorkflowNodeModel(
+            workflow_id=workflow_id,
+            id=n.id,
+            ref=n.ref,
+            pos_x=n.pos_x,
+            pos_y=n.pos_y,
+        ).save(self.session)
+
+        await NodeBlockModel(
+            node_ref=node.ref,
+            node_id=node.id,
+            type=n.block.type,
+            input=n.block.input,
+        ).save(self.session)
+
+        return node
+
+    async def update_node_with_block(
+        self,
+        node: NodeUpsert,
+        existing_node: WorkflowNodeModel,
+        existing_block: NodeBlockModel,
+    ) -> None:
+        existing_node.pos_x = node.pos_x
+        existing_node.pos_y = node.pos_y
+        await existing_node.save(self.session)
+
+        existing_block.type = node.block.type
+        existing_block.input = node.block.input
+        await existing_block.save(self.session)
+
+    async def mark_old_nodes_deleted(
+        self,
+        nodes: List[NodeUpsert],
+        all_nodes: Dict[str, WorkflowNodeModel],
+    ) -> None:
+        node_ids = {n.id for n in nodes}
+        [
+            await node.mark_deleted().save(self.session)
+            for node in all_nodes.values()
+            if node.id not in node_ids
+        ]