Merge pull request #96 from lmnr-ai/dev

Dev

Author: skull8888888

app-server/src/routes/datasets.rs

@@ -215,16 +215,6 @@ async fn update_datapoint_data(
     )
     .await?;
 
-    semantic_search
-        .delete_embeddings(
-            &project_id.to_string(),
-            vec![HashMap::from([(
-                "id".to_string(),
-                datapoint_id.to_string(),
-            )])],
-        )
-        .await?;
-
     let dataset = db::datasets::get_dataset(&db.pool, project_id, dataset_id).await?;
     if dataset.indexed_on.is_some() {
         dataset

app-server/src/traces/consumer.rs

@@ -29,6 +29,32 @@ use crate::{
 };
 
 pub async fn process_queue_spans(
+    pipeline_runner: Arc<PipelineRunner>,
+    db: Arc<DB>,
+    cache: Arc<Cache>,
+    semantic_search: Arc<SemanticSearch>,
+    language_model_runner: Arc<LanguageModelRunner>,
+    rabbitmq_connection: Arc<Connection>,
+    clickhouse: clickhouse::Client,
+    chunker_runner: Arc<chunk::runner::ChunkerRunner>,
+) {
+    loop {
+        inner_process_queue_spans(
+            pipeline_runner.clone(),
+            db.clone(),
+            cache.clone(),
+            semantic_search.clone(),
+            language_model_runner.clone(),
+            rabbitmq_connection.clone(),
+            clickhouse.clone(),
+            chunker_runner.clone(),
+        )
+        .await;
+        log::warn!("Span listener exited. Creating a new RabbitMQ channel...");
+    }
+}
+
+async fn inner_process_queue_spans(
     pipeline_runner: Arc<PipelineRunner>,
     db: Arc<DB>,
     cache: Arc<Cache>,
@@ -171,7 +197,12 @@ pub async fn process_queue_spans(
         }
 
         if let Err(e) = db::spans::record_span(&db.pool, &span).await {
-            log::error!("Failed to record span [{}]: {:?}", span.span_id, e);
+            log::error!(
+                "Failed to record span. span_id [{}], project_id [{}]: {:?}",
+                span.span_id,
+                rabbitmq_span_message.project_id,
+                e
+            );
         } else {
             // ack the message as soon as the span is recorded
             let _ = delivery
@@ -184,7 +215,12 @@ pub async fn process_queue_spans(
         // TODO: Queue batches and send them every 1-2 seconds
         let insert_span_res = ch::spans::insert_span(clickhouse.clone(), &ch_span).await;
         if let Err(e) = insert_span_res {
-            log::error!("Failed to insert span into Clickhouse: {:?}", e);
+            log::error!(
+                "Failed to insert span into Clickhouse. span_id [{}], project_id [{}]: {:?}",
+                span.span_id,
+                rabbitmq_span_message.project_id,
+                e
+            );
         }
 
         let registered_label_classes = match get_registered_label_classes_for_path(
@@ -196,7 +232,11 @@ pub async fn process_queue_spans(
         {
             Ok(classes) => classes,
             Err(e) => {
-                log::error!("Failed to get registered label classes: {:?}", e);
+                log::error!(
+                    "Failed to get registered label classes. project_id [{}]: {:?}",
+                    rabbitmq_span_message.project_id,
+                    e
+                );
                 Vec::new() // Return an empty vector if there's an error
             }
         };
@@ -217,5 +257,5 @@ pub async fn process_queue_spans(
         }
     }
 
-    log::info!("Shutting down span listener");
+    log::warn!("RabbitMQ closed connection. Shutting down span listener");
 }

frontend/app/api/projects/[projectId]/datasets/[datasetId]/datapoints/[datapointId]/route.ts

@@ -1,6 +1,8 @@
-import { authOptions } from '@/lib/auth';
-import { fetcher } from '@/lib/utils';
-import { getServerSession } from 'next-auth';
+import { db } from '@/lib/db/drizzle';
+import { datasetDatapoints } from '@/lib/db/schema';
+import { and, eq } from 'drizzle-orm';
+import { z } from 'zod';
+import { isCurrentUserMemberOfProject } from '@/lib/db/utils';
 
 export async function POST(
   req: Request,
@@ -9,22 +11,56 @@ export async function POST(
   }: { params: { projectId: string; datasetId: string; datapointId: string } }
 ): Promise<Response> {
   const projectId = params.projectId;
+
+  if (!await isCurrentUserMemberOfProject(projectId)) {
+    return new Response('Unauthorized', { status: 401 });
+  }
+
   const datasetId = params.datasetId;
   const datapointId = params.datapointId;
-  const session = await getServerSession(authOptions);
-  const user = session!.user;
 
   const body = await req.json();
 
-  return await fetcher(
-    `/projects/${projectId}/datasets/${datasetId}/datapoints/${datapointId}`,
-    {
-      method: 'POST',
-      headers: {
-        'Content-Type': 'application/json',
-        Authorization: `Bearer ${user.apiKey}`
-      },
-      body: JSON.stringify(body)
+  const schema = z.object({
+    data: z.record(z.unknown()),
+    target: z.record(z.unknown()).nullable(),
+    metadata: z.record(z.unknown()).nullable(),
+  });
+
+  const result = schema.safeParse(body);
+  if (!result.success) {
+    console.error('Invalid request body', result.error);
+    return new Response('Invalid request body', { status: 400 });
+  }
+
+  const { data, target, metadata } = result.data;
+
+  try {
+    const updatedDatapoint = await db
+      .update(datasetDatapoints)
+      .set({
+        data,
+        target,
+        metadata,
+      })
+      .where(
+        and(
+          eq(datasetDatapoints.id, datapointId),
+          eq(datasetDatapoints.datasetId, datasetId)
+        )
+      )
+      .returning();
+
+    if (updatedDatapoint.length === 0) {
+      return new Response('Datapoint not found', { status: 404 });
     }
-  );
+
+    return new Response(JSON.stringify(updatedDatapoint[0]), {
+      status: 200,
+      headers: { 'Content-Type': 'application/json' },
+    });
+  } catch (error) {
+    console.error('Error updating datapoint:', error);
+    return new Response('Internal Server Error', { status: 500 });
+  }
 }

frontend/app/api/projects/[projectId]/datasets/[datasetId]/datapoints/route.ts

@@ -2,6 +2,10 @@ import { getServerSession } from 'next-auth';
 import { authOptions } from '@/lib/auth';
 import { fetcher } from '@/lib/utils';
 import { NextRequest } from 'next/server';
+import { db } from '@/lib/db/drizzle';
+import { datasetDatapoints } from '@/lib/db/schema';
+import { and, inArray, eq } from 'drizzle-orm';
+import { isCurrentUserMemberOfProject } from '@/lib/db/utils';
 
 export async function GET(
   req: NextRequest,
@@ -53,20 +57,30 @@ export async function DELETE(
 ): Promise<Response> {
   const projectId = params.projectId;
   const datasetId = params.datasetId;
-  const session = await getServerSession(authOptions);
-  const user = session!.user;
 
-  const body = await req.json();
+  if (!(await isCurrentUserMemberOfProject(projectId))) {
+    return new Response(JSON.stringify({ error: "User is not a member of the project" }), { status: 403 });
+  }
 
-  return await fetcher(
-    `/projects/${projectId}/datasets/${datasetId}/datapoints`,
-    {
-      method: 'DELETE',
-      headers: {
-        'Content-Type': 'application/json',
-        Authorization: `Bearer ${user.apiKey}`
-      },
-      body: JSON.stringify(body)
-    }
-  );
+  const { searchParams } = new URL(req.url);
+  const datapointIds = searchParams.get('datapointIds')?.split(',');
+
+  if (!datapointIds) {
+    return new Response('At least one Datapoint ID is required', { status: 400 });
+  }
+
+  try {
+    await db.delete(datasetDatapoints)
+      .where(
+        and(
+          inArray(datasetDatapoints.id, datapointIds),
+          eq(datasetDatapoints.datasetId, datasetId)
+        )
+      );
+
+    return new Response('datasetDatapoints deleted successfully', { status: 200 });
+  } catch (error) {
+    console.error('Error deleting datasetDatapoints:', error);
+    return new Response('Error deleting datasetDatapoints', { status: 500 });
+  }
 }

frontend/app/api/projects/[projectId]/evaluations/route.ts

@@ -1,19 +1,52 @@
-import { getServerSession } from 'next-auth';
-import { authOptions } from '@/lib/auth';
-import { fetcher } from '@/lib/utils';
+import { db } from '@/lib/db/drizzle';
+import { evaluations } from '@/lib/db/schema';
+import { and, desc, eq, inArray } from 'drizzle-orm';
+import { isCurrentUserMemberOfProject } from '@/lib/db/utils';
 
 export async function GET(
   req: Request,
   { params }: { params: { projectId: string } }
 ): Promise<Response> {
   const projectId = params.projectId;
-  const session = await getServerSession(authOptions);
-  const user = session!.user;
-
-  return await fetcher(`/projects/${projectId}/evaluations`, {
-    method: 'GET',
-    headers: {
-      Authorization: `Bearer ${user.apiKey}`
-    }
-  });
+
+  if (!(await isCurrentUserMemberOfProject(projectId))) {
+    return new Response(JSON.stringify({ error: "User is not a member of the project" }), { status: 403 });
+  }
+
+  const result = await db.select().from(evaluations).where(eq(evaluations.projectId, projectId)).orderBy(desc(evaluations.createdAt));
+
+  return Response.json(result);
+}
+
+export async function DELETE(
+  req: Request,
+  { params }: { params: { projectId: string } }
+): Promise<Response> {
+  const projectId = params.projectId;
+
+  if (!(await isCurrentUserMemberOfProject(projectId))) {
+    return new Response(JSON.stringify({ error: "User is not a member of the project" }), { status: 403 });
+  }
+
+  const { searchParams } = new URL(req.url);
+  const evaluationIds = searchParams.get('evaluationIds')?.split(',');
+
+  if (!evaluationIds) {
+    return new Response('At least one Evaluation ID is required', { status: 400 });
+  }
+
+  try {
+    await db.delete(evaluations)
+      .where(
+        and(
+          inArray(evaluations.id, evaluationIds),
+          eq(evaluations.projectId, projectId)
+        )
+      );
+
+    return new Response('Evaluations deleted successfully', { status: 200 });
+  } catch (error) {
+    console.error('Error deleting evaluations:', error);
+    return new Response('Error deleting evaluations', { status: 500 });
+  }
 }

frontend/app/error.tsx

@@ -2,7 +2,7 @@
 
 import Link from 'next/link';
 import Image from 'next/image';
-import icon from '@/assets/logo/icon_light.svg';
+import icon from '@/assets/logo/icon.png';
 
 export default function Error({
   error,

frontend/app/globals.css

@@ -35,8 +35,8 @@
 
     --radius: 0.5rem; */
 
-    --background: 240 8% 8%;
-    --foreground: 240 8% 88%;
+    --background: 240 0% 6%;
+    --foreground: 240 8% 95%;
 
     --card: 240 8% 10%;
     --card-foreground: 240 8% 80%;
@@ -50,17 +50,17 @@
     --secondary: 232 9% 17%;
     --secondary-foreground: 240 8% 70%;
 
-    --muted: 232 9% 17%;
+    --muted: 240 6% 16%;
     --muted-foreground: 215 20.2% 65.1%;
 
     --accent: 232 9% 17%;
-    --accent-foreground: 210 40% 98%;
+    --accent-foreground: 210 100% 100%;
 
     --destructive: 0 62.8% 30.6%;
     --destructive-foreground: 210 40% 98%;
 
-    --border: 240 6% 20%;
-    --input: 240 6% 20%;
+    --border: 240 6% 18%;
+    --input: 240 6% 18%;
     --ring: 212 96% 78%;
 
     --radius: 0.5rem;

frontend/app/project/[projectId]/evaluations/page.tsx

@@ -1,10 +1,5 @@
-import { authOptions } from '@/lib/auth';
-import { getServerSession } from 'next-auth';
-import { redirect } from 'next/navigation';
 import { Metadata } from 'next';
 import Evaluations from '@/components/evaluations/evaluations';
-import { fetcherJSON } from '@/lib/utils';
-import { Evaluation } from '@/lib/evaluation/types';
 
 export const metadata: Metadata = {
   title: 'Evaluations'
@@ -15,22 +10,5 @@ export default async function EvaluationsPage({
 }: {
   params: { projectId: string };
 }) {
-  const session = await getServerSession(authOptions);
-  if (!session) {
-    redirect('/sign-in');
-  }
-
-  const user = session.user;
-
-  const evaluations = (await fetcherJSON(
-    `/projects/${params.projectId}/evaluations/`,
-    {
-      method: 'GET',
-      headers: {
-        Authorization: `Bearer ${user.apiKey}`
-      }
-    }
-  )) as Evaluation[];
-
-  return <Evaluations evaluations={evaluations} />;
+  return <Evaluations />;
 }