Pipeline Editor

-- Pipeline storage schema
CREATE TABLE pipelines (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  name VARCHAR(255) NOT NULL,
  description TEXT,
  nodes JSONB NOT NULL,           -- React Flow nodes
  edges JSONB NOT NULL,           -- React Flow edges
  organization_id INTEGER REFERENCES organizations(id),
  created_by_id INTEGER REFERENCES users(id),
  access VARCHAR(50) DEFAULT 'private',  -- private, organization, public
  created_at TIMESTAMP DEFAULT NOW(),
  updated_at TIMESTAMP DEFAULT NOW()
);

-- Example node structure
{
  "id": "source-1",
  "type": "source",
  "position": { "x": 100, "y": 100 },
  "data": {
    "label": "Customer Data",
    "catalog": "main",
    "schema": "sales",
    "table": "customers"
  }
}

// Zustand store for pipeline state
import { create } from 'zustand';
import { addEdge, applyNodeChanges, applyEdgeChanges } from 'reactflow';

interface PipelineStore {
  nodes: Node[];
  edges: Edge[];
  selectedNode: Node | null;

  // Node operations
  addNode: (node: Node) => void;
  updateNodeData: (nodeId: string, data: Partial<NodeData>) => void;
  removeNode: (nodeId: string) => void;

  // Edge operations
  connectNodes: (connection: Connection) => void;
  removeEdge: (edgeId: string) => void;

  // Selection
  selectNode: (node: Node | null) => void;

  // Persistence
  loadPipeline: (pipeline: Pipeline) => void;
  savePipeline: () => Pipeline;
}

const usePipelineStore = create<PipelineStore>((set, get) => ({
  nodes: [],
  edges: [],
  selectedNode: null,

  addNode: (node) => set((state) => ({
    nodes: [...state.nodes, node]
  })),

  updateNodeData: (nodeId, data) => set((state) => ({
    nodes: state.nodes.map(n =>
      n.id === nodeId ? { ...n, data: { ...n.data, ...data } } : n
    )
  })),

  connectNodes: (connection) => set((state) => ({
    edges: addEdge(connection, state.edges)
  })),

  selectNode: (node) => set({ selectedNode: node }),

  loadPipeline: (pipeline) => set({
    nodes: pipeline.nodes,
    edges: pipeline.edges,
  }),

  savePipeline: () => ({
    nodes: get().nodes,
    edges: get().edges,
  }),
}));

// Convert visual pipeline to DLT code
const convertToDLT = (nodes: Node[], edges: Edge[]): string => {
  const dltCode: string[] = [];

  // Process nodes in topological order
  const sortedNodes = topologicalSort(nodes, edges);

  for (const node of sortedNodes) {
    switch (node.type) {
      case 'source':
        dltCode.push(`
@dlt.table(name="${node.data.outputName}")
def ${node.id.replace('-', '_')}():
    return spark.table("${node.data.catalog}.${node.data.schema}.${node.data.table}")
`);
        break;

      case 'filter':
        const inputNode = getInputNode(node, edges);
        dltCode.push(`
@dlt.table(name="${node.data.outputName}")
def ${node.id.replace('-', '_')}():
    return dlt.read("${inputNode.data.outputName}").filter("${node.data.condition}")
`);
        break;

      // ... more node types
    }
  }

  return dltCode.join('\n');
};