chore: release v2.4.2

docs: add DB_EXPORT_PAGE_SIZE to .env.prod.example
fix(docker): remove the need for runtime variables at build time (#1067 )
2024-02-13 15:53:15 +01:00 · 2024-02-13 15:48:49 +01:00 · 2024-02-13 15:40:16 +01:00 · 2024-02-13 12:55:12 +00:00
23 changed files with 999 additions and 623 deletions
@@ -29,4 +29,8 @@ S3_ENDPOINT=
 S3_ACCESS_KEY_ID=
 S3_SECRET_ACCESS_KEY=
 S3_BUCKET_NAME=
-S3_REGION=
+S3_REGION=
+
+# Set during docker build of application
+# Used to disable environment verification at build time
+# DOCKER_BUILD=1
@@ -57,6 +57,9 @@ SALT="salt"
 # S3_BUCKET_NAME=
 # S3_REGION=

+# Exports are streamed to S3 in pages to avoid memory issues
+# The page size can be adjusted if needed to optimize performance
+# DB_EXPORT_PAGE_SIZE=1000



@@ -33,10 +33,6 @@ jobs:
  test-docker-build:
    runs-on: ubuntu-latest
    env:
-      DATABASE_URL: postgresql://postgres:postgres@localhost:5432/postgres
-      NEXTAUTH_SECRET: "secret"
-      SALT: "salt"
-      NEXTAUTH_URL: "http://localhost:3030"
      REGISTRY: ghcr.io
      IMAGE_NAME: ${{ github.repository }}

@@ -54,11 +50,6 @@ jobs:
        with:
          context: .
          push: false
-          build-args: |
-            DATABASE_URL=${{ env.DATABASE_URL }}
-            NEXTAUTH_SECRET=${{ env.NEXTAUTH_SECRET }}
-            NEXTAUTH_URL=${{ env.NEXTAUTH_URL }}
-            SALT=${{ env.SALT }}

  tests:
    runs-on: ubuntu-latest
@@ -152,10 +143,6 @@ jobs:
    environment: "protected branches"
    runs-on: ubuntu-latest
    env:
-      DATABASE_URL: postgresql://postgres:postgres@localhost:5432/postgres
-      NEXTAUTH_SECRET: "secret"
-      SALT: "salt"
-      NEXTAUTH_URL: "http://localhost:3030"
      REGISTRY: ghcr.io
      IMAGE_NAME: ${{ github.repository }}
    permissions:
@@ -198,8 +185,3 @@ jobs:
          push: true
          tags: ${{ steps.meta.outputs.tags }}
          labels: ${{ steps.meta.outputs.labels }}
-          build-args: |
-            DATABASE_URL=${{ env.DATABASE_URL }}
-            NEXTAUTH_SECRET=${{ env.NEXTAUTH_SECRET }}
-            NEXTAUTH_URL=${{ env.NEXTAUTH_URL }}
-            SALT=${{ env.SALT }}
@@ -1,19 +1,11 @@
 # Base image
 FROM node:20-alpine AS base
-ARG DATABASE_URL
-ARG NEXTAUTH_SECRET
-ARG NEXTAUTH_URL
-ARG SALT

 # It's important to update the index before installing packages to ensure you're getting the latest versions.
 # Check https://github.com/nodejs/docker-node/tree/b4117f9333da4138b03a546ec926ef50a31506c3#nodealpine to understand why libc6-compat might be needed.
 RUN apk update && apk upgrade --no-cache libcrypto3 libssl3 libc6-compat

 FROM base AS deps
-ARG DATABASE_URL
-ARG NEXTAUTH_SECRET
-ARG NEXTAUTH_URL
-ARG SALT

 WORKDIR /app

@@ -29,10 +21,6 @@ RUN \

 # Rebuild the source code only when needed
 FROM base AS builder
-ARG DATABASE_URL
-ARG NEXTAUTH_SECRET
-ARG NEXTAUTH_URL
-ARG SALT

 WORKDIR /app
 COPY --from=deps /app/node_modules ./node_modules
@@ -46,6 +34,9 @@ RUN rm -f ./src/middleware.ts
 # Uncomment the following line in case you want to disable telemetry during the build.
 ENV NEXT_TELEMETRY_DISABLED 1

+# Disable validation of environment variables during build
+ENV DOCKER_BUILD 1
+
 # Generate prisma client
 RUN npx prisma generate

@@ -54,10 +45,6 @@ RUN npm run build

 # Production image, copy all the files and run next
 FROM base AS runner
-ARG DATABASE_URL
-ARG NEXTAUTH_SECRET
-ARG NEXTAUTH_URL
-ARG SALT

 RUN apk add --no-cache dumb-init

@@ -66,6 +53,8 @@ WORKDIR /app
 ENV NODE_ENV production
 # Uncomment the following line in case you want to disable telemetry during runtime.
 ENV NEXT_TELEMETRY_DISABLED 1
+# Needed to re-enable validation of environment variables during runtime
+ENV DOCKER_BUILD 0

 RUN addgroup --system --gid 1001 nodejs
 RUN adduser --system --uid 1001 nextjs
@@ -4,17 +4,11 @@ services:
  langfuse-server:
    build:
      dockerfile: Dockerfile
-      args:
-        - DATABASE_URL=postgresql://postgres:postgres@db:5432/postgres
-        - NEXTAUTH_SECRET=mysecret
-        - SALT=mysalt
-        - NEXTAUTH_URL=http://localhost:3000
    depends_on:
      - db
    ports:
      - "3000:3000"
    environment:
-      - NODE_ENV=production
      - DATABASE_URL=postgresql://postgres:postgres@db:5432/postgres
      - NEXTAUTH_SECRET=mysecret
      - SALT=mysalt
@@ -8,7 +8,6 @@ services:
    ports:
      - "3000:3000"
    environment:
-      - NODE_ENV=production
      - DATABASE_URL=postgresql://postgres:postgres@db:5432/postgres
      - NEXTAUTH_SECRET=mysecret
      - SALT=mysalt
@@ -1,16 +1,17 @@
 {
  "name": "langfuse-core",
-  "version": "2.4.1",
+  "version": "2.4.2",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "langfuse-core",
-      "version": "2.4.1",
+      "version": "2.4.2",
      "hasInstallScript": true,
      "dependencies": {
        "@anthropic-ai/tokenizer": "^0.0.4",
        "@aws-sdk/client-s3": "^3.507.0",
+        "@aws-sdk/lib-storage": "^3.511.0",
        "@aws-sdk/s3-request-presigner": "^3.507.0",
        "@headlessui/react": "^1.7.18",
        "@heroicons/react": "^2.1.1",
@@ -664,6 +665,26 @@
        "node": ">=14.0.0"
      }
    },
+    "node_modules/@aws-sdk/lib-storage": {
+      "version": "3.511.0",
+      "resolved": "https://registry.npmjs.org/@aws-sdk/lib-storage/-/lib-storage-3.511.0.tgz",
+      "integrity": "sha512-inEbSyqzGxiQs8aEnkGdxw9ZDn370mRHOdE1TB/GvVe9buQVyZ2hQvOY5WBVOaIGDIxGpuUzVvr4o89XreU19w==",
+      "dependencies": {
+        "@smithy/abort-controller": "^2.1.1",
+        "@smithy/middleware-endpoint": "^2.4.1",
+        "@smithy/smithy-client": "^2.3.1",
+        "buffer": "5.6.0",
+        "events": "3.3.0",
+        "stream-browserify": "3.0.0",
+        "tslib": "^2.5.0"
+      },
+      "engines": {
+        "node": ">=14.0.0"
+      },
+      "peerDependencies": {
+        "@aws-sdk/client-s3": "^3.0.0"
+      }
+    },
    "node_modules/@aws-sdk/middleware-bucket-endpoint": {
      "version": "3.502.0",
      "resolved": "https://registry.npmjs.org/@aws-sdk/middleware-bucket-endpoint/-/middleware-bucket-endpoint-3.502.0.tgz",
@@ -7723,6 +7744,15 @@
        "node-int64": "^0.4.0"
      }
    },
+    "node_modules/buffer": {
+      "version": "5.6.0",
+      "resolved": "https://registry.npmjs.org/buffer/-/buffer-5.6.0.tgz",
+      "integrity": "sha512-/gDYp/UtU0eA1ys8bOs9J6a+E/KWIY+DZ+Q2WESNUA0jFRsJOc0SNUO6xJ5SGA1xueg3NL65W6s+NY5l9cunuw==",
+      "dependencies": {
+        "base64-js": "^1.0.2",
+        "ieee754": "^1.1.4"
+      }
+    },
    "node_modules/buffer-crc32": {
      "version": "0.2.13",
      "resolved": "https://registry.npmjs.org/buffer-crc32/-/buffer-crc32-0.2.13.tgz",
@@ -10126,6 +10156,14 @@
      "resolved": "https://registry.npmjs.org/eventemitter3/-/eventemitter3-4.0.7.tgz",
      "integrity": "sha512-8guHBZCwKnFhYdHr2ysuRWErTwhoN2X8XELRlrRwpmfeY2jjuUN4taQMsULKUVo1K4DvZl+0pgfyoysHxvmvEw=="
    },
+    "node_modules/events": {
+      "version": "3.3.0",
+      "resolved": "https://registry.npmjs.org/events/-/events-3.3.0.tgz",
+      "integrity": "sha512-mQw+2fkQbALzQ7V0MY0IqdnXNOeTtP4r0lN9z7AAawCXgqea7bDii20AYrIBrFd/Hx0M2Ocz6S111CaFkUcb0Q==",
+      "engines": {
+        "node": ">=0.8.x"
+      }
+    },
    "node_modules/execa": {
      "version": "5.1.1",
      "resolved": "https://registry.npmjs.org/execa/-/execa-5.1.1.tgz",
@@ -11125,7 +11163,6 @@
      "version": "1.2.1",
      "resolved": "https://registry.npmjs.org/ieee754/-/ieee754-1.2.1.tgz",
      "integrity": "sha512-dcyqhDvX1C46lXZcVqCpK+FtMRQVdIMN6/Df5js2zouUsqG7I6sFxitIC+7KYK29KdXOLHdu9zL4sFnoVQnqaA==",
-      "dev": true,
      "funding": [
        {
          "type": "github",
@@ -15795,7 +15832,6 @@
      "version": "3.6.2",
      "resolved": "https://registry.npmjs.org/readable-stream/-/readable-stream-3.6.2.tgz",
      "integrity": "sha512-9u/sniCrY3D5WdsERHzHE4G2YCXqoG5FTHUiCC4SIbr6XcLZBY05ya9EKjYek9O5xOAwjGq+1JdGBAS7Q9ScoA==",
-      "dev": true,
      "dependencies": {
        "inherits": "^2.0.3",
        "string_decoder": "^1.1.1",
@@ -16485,7 +16521,6 @@
      "version": "5.2.1",
      "resolved": "https://registry.npmjs.org/safe-buffer/-/safe-buffer-5.2.1.tgz",
      "integrity": "sha512-rp3So07KcdmmKbGvgaNxQSJr7bGVSVk5S9Eq1F+ppbRo70+YeaDxkw5Dd8NPN+GD6bjnYm2VuPuCXmpuYvmCXQ==",
-      "dev": true,
      "funding": [
        {
          "type": "github",
@@ -16860,6 +16895,15 @@
        "node": ">= 0.4"
      }
    },
+    "node_modules/stream-browserify": {
+      "version": "3.0.0",
+      "resolved": "https://registry.npmjs.org/stream-browserify/-/stream-browserify-3.0.0.tgz",
+      "integrity": "sha512-H73RAHsVBapbim0tU2JwwOiXUj+fikfiaoYAKHF3VJfA0pe2BCzkhAHBlLG6REzE+2WNZcxOXjK7lkso+9euLA==",
+      "dependencies": {
+        "inherits": "~2.0.4",
+        "readable-stream": "^3.5.0"
+      }
+    },
    "node_modules/streamsearch": {
      "version": "1.1.0",
      "resolved": "https://registry.npmjs.org/streamsearch/-/streamsearch-1.1.0.tgz",
@@ -16872,7 +16916,6 @@
      "version": "1.3.0",
      "resolved": "https://registry.npmjs.org/string_decoder/-/string_decoder-1.3.0.tgz",
      "integrity": "sha512-hkRX8U1WjJFd8LsDJ2yQ/wWWxaopEsABU1XfkM8A+j0+85JAGppt16cr1Whg6KIbb4okU6Mql6BOj+uup/wKeA==",
-      "dev": true,
      "dependencies": {
        "safe-buffer": "~5.2.0"
      }
@@ -1,6 +1,6 @@
 {
  "name": "langfuse-core",
-  "version": "2.4.1",
+  "version": "2.4.2",
  "private": true,
  "scripts": {
    "prebuild": "cp generated/openapi-client/openapi.yml public/openapi-client.yml && cp generated/openapi-server/openapi.yml public/openapi-server.yml",
@@ -31,6 +31,7 @@
  "dependencies": {
    "@anthropic-ai/tokenizer": "^0.0.4",
    "@aws-sdk/client-s3": "^3.507.0",
+    "@aws-sdk/lib-storage": "^3.511.0",
    "@aws-sdk/s3-request-presigner": "^3.507.0",
    "@headlessui/react": "^1.7.18",
    "@heroicons/react": "^2.1.1",
@@ -0,0 +1,194 @@
+/** @jest-environment node */
+/* eslint-disable @typescript-eslint/no-unsafe-member-access */
+
+import { pruneDatabase } from "@/src/__tests__/test-utils";
+import { ModelUsageUnit } from "@/src/constants";
+import { appRouter } from "@/src/server/api/root";
+import { createInnerTRPCContext } from "@/src/server/api/trpc";
+import { prisma } from "@/src/server/db";
+import type { Session } from "next-auth";
+
+describe("observations.export RPC", () => {
+  const numberOfGenerations = 5;
+  const projectId = "7a88fb47-b4e2-43b8-a06c-a5ce950dc53a";
+
+  beforeAll(async () => {
+    // Disable S3 upload
+    process.env.S3_ENDPOINT = "";
+
+    await pruneDatabase();
+    const traceId = "trace-1";
+
+    await prisma.trace.create({
+      data: {
+        id: traceId,
+        name: "trace-name",
+        userId: "user-1",
+        projectId,
+        metadata: { key: "value" },
+        release: "1.0.0",
+        version: "2.0.0",
+      },
+    });
+
+    for (let i = 1; i <= numberOfGenerations; i++) {
+      await prisma.observation.create({
+        data: {
+          type: "GENERATION",
+          id: `generation-${i}`,
+          name: `generation-${i}`,
+          model: "gpt-3.5-turbo",
+          totalCost: 1,
+          startTime: new Date("2021-01-01T00:00:00.000Z"),
+          endTime: new Date("2021-01-01T00:00:05.000Z"),
+          project: { connect: { id: projectId } },
+          traceId,
+          input: [
+            {
+              role: "system",
+              content: "Be a helpful assistant",
+            },
+            {
+              role: "user",
+              content: "How can i create a React component?",
+            },
+          ],
+          output: {
+            completion: `Creating a React component can be done in two ways.`,
+          },
+          metadata: {
+            user: `user-@langfuse.com`,
+          },
+          unit: ModelUsageUnit.Tokens,
+        },
+      });
+    }
+  });
+
+  afterAll(async () => await pruneDatabase());
+
+  const session: Session = {
+    expires: "1",
+    user: {
+      id: "clgb17vnp000008jjere5g15i",
+      name: "John Doe",
+      projects: [
+        {
+          id: projectId,
+          role: "ADMIN",
+          name: "test",
+        },
+      ],
+      featureFlags: {
+        templateFlag: true,
+      },
+      admin: true,
+    },
+  };
+
+  const ctx = createInnerTRPCContext({ session });
+  const caller = appRouter.createCaller({ ...ctx, prisma });
+
+  it("should return a CSV file", async () => {
+    const result = await caller.generations.export({
+      fileFormat: "CSV",
+      orderBy: { column: "id", order: "ASC" },
+      filter: [
+        {
+          column: "start_time",
+          type: "datetime",
+          operator: ">",
+          value: new Date("1990-01-01"),
+        },
+      ],
+      projectId,
+      searchQuery: null,
+    });
+
+    if (result.type !== "data")
+      throw new Error("No data returned. Is S3 accidentally enabled?");
+    const { data, fileName } = result;
+
+    const fileExtension = fileName.split(".").pop();
+    expect(fileName).toContain(`lf-export-${projectId}`);
+    expect(fileExtension).toBe("csv");
+    expect(data.split("\n").filter(Boolean).length).toBe(
+      numberOfGenerations + 1,
+    );
+  });
+
+  it("should return a JSON file", async () => {
+    const result = await caller.generations.export({
+      fileFormat: "JSON",
+      orderBy: { column: "id", order: "ASC" },
+      filter: [
+        {
+          column: "start_time",
+          type: "datetime",
+          operator: ">",
+          value: new Date("1990-01-01"),
+        },
+      ],
+      projectId,
+      searchQuery: null,
+    });
+
+    if (result.type !== "data")
+      throw new Error("No data returned. Is S3 accidentally enabled?");
+    const { data, fileName } = result;
+
+    const fileExtension = fileName.split(".").pop();
+    expect(fileName).toContain(`lf-export-${projectId}`);
+    expect(fileExtension).toBe("json");
+
+    expect(JSON.parse(data).length).toBe(numberOfGenerations);
+  });
+
+  it("should return a OPENAI-JSONL file", async () => {
+    const result = await caller.generations.export({
+      fileFormat: "OPENAI-JSONL",
+      orderBy: { column: "id", order: "ASC" },
+      filter: [
+        {
+          column: "start_time",
+          type: "datetime",
+          operator: ">",
+          value: new Date("1990-01-01"),
+        },
+      ],
+      projectId,
+      searchQuery: null,
+    });
+
+    if (result.type !== "data")
+      throw new Error("No data returned. Is S3 accidentally enabled?");
+    const { data, fileName } = result;
+
+    const fileExtension = fileName.split(".").pop();
+    expect(fileName).toContain(`lf-export-${projectId}`);
+    expect(fileExtension).toBe("jsonl");
+
+    expect(data.split("\n").filter(Boolean).length).toBe(numberOfGenerations);
+  });
+
+  it("should throw on unsupported file formats", async () => {
+    const unsupportedFileFormat = "XLSX";
+
+    const call = caller.generations.export({
+      fileFormat: unsupportedFileFormat as unknown as "JSON",
+      orderBy: { column: "id", order: "ASC" },
+      filter: [
+        {
+          column: "start_time",
+          type: "datetime",
+          operator: ">",
+          value: new Date("1990-01-01"),
+        },
+      ],
+      projectId,
+      searchQuery: null,
+    });
+
+    await expect(call).rejects.toThrow();
+  });
+});
@@ -1 +1 @@
-export const VERSION = "v2.4.1";
+export const VERSION = "v2.4.2";
@@ -58,6 +58,8 @@ export const env = createEnv({
    S3_SECRET_ACCESS_KEY: z.string().optional(),
    S3_BUCKET_NAME: z.string().optional(),
    S3_REGION: z.string().optional(),
+    // Database exports
+    DB_EXPORT_PAGE_SIZE: z.number().optional(),
  },

  /**
@@ -120,5 +122,10 @@ export const env = createEnv({
    S3_SECRET_ACCESS_KEY: process.env.S3_SECRET_ACCESS_KEY,
    S3_BUCKET_NAME: process.env.S3_BUCKET_NAME,
    S3_REGION: process.env.S3_REGION,
+    // Database exports
+    DB_EXPORT_PAGE_SIZE: process.env.DB_EXPORT_PAGE_SIZE,
  },
+  // Skip validation in Docker builds
+  // DOCKER_BUILD is set in Dockerfile
+  skipValidation: process.env.DOCKER_BUILD === "1",
 });
@@ -1,573 +0,0 @@
-import { z } from "zod";
-
-import {
-  createTRPCRouter,
-  protectedProjectProcedure,
-} from "@/src/server/api/trpc";
-
-import { Prisma, type ObservationView } from "@prisma/client";
-import { jsonSchema, paginationZod } from "@/src/utils/zod";
-import { singleFilter } from "@/src/server/api/interfaces/filters";
-import {
-  datetimeFilterToPrismaSql,
-  filterToPrismaSql,
-} from "@/src/features/filters/server/filterToPrisma";
-import {
-  type ObservationOptions,
-  observationsTableCols,
-} from "@/src/server/api/definitions/observationsTable";
-import { usdFormatter } from "@/src/utils/numbers";
-import { env } from "@/src/env.mjs";
-import {
-  S3Client,
-  PutObjectCommand,
-  GetObjectCommand,
-} from "@aws-sdk/client-s3";
-import { getSignedUrl } from "@aws-sdk/s3-request-presigner";
-import {
-  exportFileFormats,
-  exportOptions,
-} from "@/src/server/api/interfaces/exportTypes";
-import { orderBy } from "@/src/server/api/interfaces/orderBy";
-import { orderByToPrismaSql } from "@/src/features/orderBy/server/orderByToPrisma";
-
-const GenerationTableOptions = z.object({
-  projectId: z.string(), // Required for protectedProjectProcedure
-  filter: z.array(singleFilter),
-  searchQuery: z.string().nullable(),
-  orderBy: orderBy,
-});
-
-const ListInputs = GenerationTableOptions.extend({
-  ...paginationZod,
-});
-
-// extend generationfilteroptions with export options
-const ExportInputs = GenerationTableOptions.extend({
-  fileFormat: z.enum(exportFileFormats),
-});
-
-export const generationsRouter = createTRPCRouter({
-  all: protectedProjectProcedure
-    .input(ListInputs)
-    .query(async ({ input, ctx }) => {
-      // ATTENTION: When making changes to this query, make sure to also update the export query
-      const searchCondition = input.searchQuery
-        ? Prisma.sql`AND (
-        o."id" ILIKE ${`%${input.searchQuery}%`} OR
-        o."name" ILIKE ${`%${input.searchQuery}%`} OR
-        o."model" ILIKE ${`%${input.searchQuery}%`} OR
-        t."name" ILIKE ${`%${input.searchQuery}%`}
-      )`
-        : Prisma.empty;
-
-      const filterCondition = filterToPrismaSql(
-        input.filter,
-        observationsTableCols,
-      );
-
-      const orderByCondition = orderByToPrismaSql(
-        input.orderBy,
-        observationsTableCols,
-      );
-
-      // to improve query performance, add timeseries filter to observation queries as well
-      const startTimeFilter = input.filter.find(
-        (f) => f.column === "start_time" && f.type === "datetime",
-      );
-      const datetimeFilter =
-        startTimeFilter && startTimeFilter.type === "datetime"
-          ? datetimeFilterToPrismaSql(
-              "start_time",
-              startTimeFilter.operator,
-              startTimeFilter.value,
-            )
-          : Prisma.empty;
-
-      const generations = await ctx.prisma.$queryRaw<
-        Array<
-          ObservationView & {
-            traceId: string;
-            traceName: string;
-            latency: number | null;
-          }
-        >
-      >(
-        Prisma.sql`
-          WITH observations_with_latency AS (
-            SELECT
-              o.*,
-              CASE WHEN o.end_time IS NULL THEN NULL ELSE (EXTRACT(EPOCH FROM o."end_time") - EXTRACT(EPOCH FROM o."start_time"))::double precision END AS "latency"
-            FROM observations_view o
-            WHERE o.type = 'GENERATION'
-            AND o.project_id = ${input.projectId}
-            ${datetimeFilter}
-          ),
-          -- used for filtering
-          scores_avg AS (
-            SELECT
-              trace_id,
-              observation_id,
-              jsonb_object_agg(name::text, avg_value::double precision) AS scores_avg
-            FROM (
-              SELECT
-                trace_id,
-                observation_id,
-                name,
-                avg(value) avg_value
-              FROM
-                scores
-              GROUP BY
-                1,
-                2,
-                3
-              ORDER BY
-                1) tmp
-            GROUP BY
-              1, 2
-          )
-          SELECT
-            o.id,
-            o.name,
-            o.model,
-            o.start_time as "startTime",
-            o.end_time as "endTime",
-            o.latency,
-            o.input,
-            o.output,
-            o.metadata,
-            o.trace_id as "traceId",
-            t.name as "traceName",
-            o.completion_start_time as "completionStartTime",
-            o.prompt_tokens as "promptTokens",
-            o.completion_tokens as "completionTokens",
-            o.total_tokens as "totalTokens",
-            o.level,
-            o.status_message as "statusMessage",
-            o.version,
-            o.model_id as "modelId",
-            o.input_price as "inputPrice",
-            o.output_price as "outputPrice",
-            o.total_price as "totalPrice",
-            o.calculated_input_cost as "calculatedInputCost",
-            o.calculated_output_cost as "calculatedOutputCost",
-            o.calculated_total_cost as "calculatedTotalCost"
-          FROM observations_with_latency o
-          JOIN traces t ON t.id = o.trace_id
-          LEFT JOIN scores_avg AS s_avg ON s_avg.trace_id = t.id and s_avg.observation_id = o.id
-          WHERE
-            t.project_id = ${input.projectId}
-            ${searchCondition}
-            ${filterCondition}
-            ${orderByCondition}
-          LIMIT ${input.limit}
-          OFFSET ${input.page * input.limit}
-        `,
-      );
-
-      const totalGenerations = await ctx.prisma.$queryRaw<
-        Array<{ count: bigint }>
-      >(
-        Prisma.sql`
-          WITH observations_with_latency AS (
-            SELECT
-              o.*,
-              CASE WHEN o.end_time IS NULL THEN NULL ELSE (EXTRACT(EPOCH FROM o."end_time") - EXTRACT(EPOCH FROM o."start_time"))::double precision END AS "latency"
-            FROM observations_view o
-            WHERE o.type = 'GENERATION'
-            AND o.project_id = ${input.projectId}
-            ${datetimeFilter}
-          ),
-          -- used for filtering
-          scores_avg AS (
-            SELECT
-              trace_id,
-              observation_id,
-              jsonb_object_agg(name::text, avg_value::double precision) AS scores_avg
-            FROM (
-              SELECT
-                trace_id,
-                observation_id,
-                name,
-                avg(value) avg_value
-              FROM
-                scores
-              GROUP BY
-                1,
-                2,
-                3
-              ORDER BY
-                1) tmp
-            GROUP BY
-              1, 2
-          )
-          SELECT
-            count(*)
-          FROM observations_with_latency o
-          JOIN traces t ON t.id = o.trace_id
-          LEFT JOIN scores_avg AS s_avg ON s_avg.trace_id = t.id and s_avg.observation_id = o.id
-          WHERE
-            t.project_id = ${input.projectId}
-            ${searchCondition}
-            ${filterCondition}
-        `,
-      );
-
-      const scores = await ctx.prisma.score.findMany({
-        where: {
-          trace: {
-            projectId: input.projectId,
-          },
-          observationId: {
-            in: generations.map((gen) => gen.id),
-          },
-        },
-      });
-      const count = totalGenerations[0]?.count;
-      return {
-        totalCount: count ? Number(count) : undefined,
-        generations: generations.map((generation) => {
-          const filteredScores = scores.filter(
-            (s) => s.observationId === generation.id,
-          );
-          return {
-            ...generation,
-            scores: filteredScores,
-          };
-        }),
-      };
-    }),
-
-  export: protectedProjectProcedure
-    .input(ExportInputs)
-    .query(async ({ input, ctx }) => {
-      // ATTENTION: When making changes to this query, make sure to also update the all query
-      const searchCondition = input.searchQuery
-        ? Prisma.sql`AND (
-        o."id" ILIKE ${`%${input.searchQuery}%`} OR
-        o."name" ILIKE ${`%${input.searchQuery}%`} OR
-        o."model" ILIKE ${`%${input.searchQuery}%`} OR
-        t."name" ILIKE ${`%${input.searchQuery}%`}
-      )`
-        : Prisma.empty;
-
-      const filterCondition = filterToPrismaSql(
-        input.filter,
-        observationsTableCols,
-      );
-
-      const orderByCondition = orderByToPrismaSql(
-        input.orderBy,
-        observationsTableCols,
-      );
-
-      // to improve query performance, add timeseries filter to observation queries as well
-      const startTimeFilter = input.filter.find(
-        (f) => f.column === "start_time" && f.type === "datetime",
-      );
-      const datetimeFilter =
-        startTimeFilter && startTimeFilter.type === "datetime"
-          ? datetimeFilterToPrismaSql(
-              "start_time",
-              startTimeFilter.operator,
-              startTimeFilter.value,
-            )
-          : Prisma.empty;
-
-      const generations = await ctx.prisma.$queryRaw<
-        Array<
-          ObservationView & {
-            traceId: string;
-            traceName: string;
-            latency: number | null;
-          }
-        >
-      >(
-        Prisma.sql`
-          WITH observations_with_latency AS (
-            SELECT
-              o.*,
-              CASE WHEN o.end_time IS NULL THEN NULL ELSE (EXTRACT(EPOCH FROM o."end_time") - EXTRACT(EPOCH FROM o."start_time"))::double precision END AS "latency"
-            FROM observations_view o
-            WHERE o.type = 'GENERATION'
-            AND o.project_id = ${input.projectId}
-            ${datetimeFilter}
-          ),
-          -- used for filtering
-          scores_avg AS (
-            SELECT
-              trace_id,
-              observation_id,
-              jsonb_object_agg(name::text, avg_value::double precision) AS scores_avg
-            FROM (
-              SELECT
-                trace_id,
-                observation_id,
-                name,
-                avg(value) avg_value
-              FROM
-                scores
-              GROUP BY
-                1,
-                2,
-                3
-              ORDER BY
-                1) tmp
-            GROUP BY
-              1, 2
-          )
-          SELECT
-            o.id,
-            o.name,
-            o.model,
-            o.start_time as "startTime",
-            o.end_time as "endTime",
-            o.latency,
-            o.input,
-            o.output,
-            o.metadata,
-            o.trace_id as "traceId",
-            t.name as "traceName",
-            o.completion_start_time as "completionStartTime",
-            o.prompt_tokens as "promptTokens",
-            o.completion_tokens as "completionTokens",
-            o.total_tokens as "totalTokens",
-            o.level,
-            o.status_message as "statusMessage",
-            o.version,
-            o.model_id as "modelId",
-            o.input_price as "inputPrice",
-            o.output_price as "outputPrice",
-            o.total_price as "totalPrice",
-            o.calculated_input_cost as "calculatedInputCost",
-            o.calculated_output_cost as "calculatedOutputCost",
-            o.calculated_total_cost as "calculatedTotalCost"
-          FROM observations_with_latency o
-          JOIN traces t ON t.id = o.trace_id
-          LEFT JOIN scores_avg AS s_avg ON s_avg.trace_id = t.id and s_avg.observation_id = o.id
-          WHERE
-            t.project_id = ${input.projectId}
-            ${searchCondition}
-            ${filterCondition}
-            ${orderByCondition}
-        `,
-      );
-
-      let output: string = "";
-
-      // create file
-      switch (input.fileFormat) {
-        case "CSV":
-          output = [
-            [
-              "traceId",
-              "name",
-              "model",
-              "startTime",
-              "endTime",
-              "cost",
-              "prompt",
-              "completion",
-              "metadata",
-            ],
-          ]
-            .concat(
-              generations.map((generation) =>
-                [
-                  generation.traceId,
-                  generation.name ?? "",
-                  generation.model ?? "",
-                  generation.startTime.toISOString(),
-                  generation.endTime?.toISOString() ?? "",
-                  generation.calculatedTotalCost
-                    ? usdFormatter(
-                        generation.calculatedTotalCost.toNumber(),
-                        2,
-                        8,
-                      )
-                    : "",
-                  JSON.stringify(generation.input),
-                  JSON.stringify(generation.output),
-                  JSON.stringify(generation.metadata),
-                ].map((field) => {
-                  const str = typeof field === "string" ? field : String(field);
-                  return `"${str.replace(/"/g, '""')}"`;
-                }),
-              ),
-            )
-            .map((row) => row.join(","))
-            .join("\n");
-          break;
-        case "JSON":
-          output = JSON.stringify(generations);
-          break;
-        case "OPENAI-JSONL":
-          const inputSchemaOpenAI = z.array(
-            z.object({
-              role: z.enum(["system", "user", "assistant"]),
-              content: z.string(),
-            }),
-          );
-          const outputSchema = z
-            .object({
-              completion: jsonSchema,
-            })
-            .or(jsonSchema);
-          output = generations
-            .map((generation) => ({
-              parsedInput: inputSchemaOpenAI.safeParse(generation.input),
-              parsedOutput: outputSchema.safeParse(generation.output),
-            }))
-            .filter((generation) => generation.parsedInput.success)
-            .map((generation) =>
-              generation.parsedInput.success // check for typescript validation, is always true due to previous filter
-                ? generation.parsedInput.data.concat(
-                    generation.parsedOutput.success
-                      ? [
-                          {
-                            role: "assistant",
-                            content:
-                              typeof generation.parsedOutput.data ===
-                                "object" &&
-                              "completion" in generation.parsedOutput.data
-                                ? JSON.stringify(
-                                    generation.parsedOutput.data.completion,
-                                  )
-                                : JSON.stringify(generation.parsedOutput.data),
-                          },
-                        ]
-                      : [],
-                  )
-                : [],
-            )
-            // to jsonl
-            .map((row) => JSON.stringify(row))
-            .join("\n");
-
-          break;
-        default:
-          throw new Error("Invalid export file format");
-      }
-
-      const fileName = `lf-export-${
-        input.projectId
-      }-${new Date().toISOString()}.${
-        exportOptions[input.fileFormat].extension
-      }`;
-
-      if (
-        env.S3_BUCKET_NAME &&
-        env.S3_ACCESS_KEY_ID &&
-        env.S3_SECRET_ACCESS_KEY &&
-        env.S3_ENDPOINT &&
-        env.S3_REGION
-      ) {
-        const client = new S3Client({
-          credentials: {
-            accessKeyId: env.S3_ACCESS_KEY_ID,
-            secretAccessKey: env.S3_SECRET_ACCESS_KEY,
-          },
-          endpoint: env.S3_ENDPOINT,
-          region: env.S3_REGION,
-        });
-        await client.send(
-          new PutObjectCommand({
-            Bucket: env.S3_BUCKET_NAME,
-            Key: fileName,
-            Body: output,
-            ContentType: exportOptions[input.fileFormat].fileType,
-          }),
-        );
-        const signedUrl = await getSignedUrl(
-          client,
-          new GetObjectCommand({
-            Bucket: env.S3_BUCKET_NAME,
-            Key: fileName,
-            ResponseContentDisposition: `attachment; filename="${fileName}"`,
-          }),
-          {
-            expiresIn: 60 * 60, // in 1 hour, signed url will expire
-          },
-        );
-        return {
-          type: "s3",
-          url: signedUrl,
-          fileName,
-        } as const;
-      } else {
-        return {
-          type: "data",
-          data: output,
-          fileName,
-        } as const;
-      }
-    }),
-  filterOptions: protectedProjectProcedure
-    .input(z.object({ projectId: z.string() }))
-    .query(async ({ input, ctx }) => {
-      const queryFilter = {
-        projectId: input.projectId,
-        type: "GENERATION",
-      } as const;
-
-      const scores = await ctx.prisma.score.groupBy({
-        where: {
-          observation: {
-            projectId: input.projectId,
-          },
-        },
-        by: ["name"],
-      });
-
-      const model = await ctx.prisma.observation.groupBy({
-        by: ["model"],
-        where: queryFilter,
-        _count: { _all: true },
-      });
-      const name = await ctx.prisma.observation.groupBy({
-        by: ["name"],
-        where: queryFilter,
-        _count: { _all: true },
-      });
-      const traceName = await ctx.prisma.$queryRaw<
-        Array<{
-          traceName: string | null;
-          count: number;
-        }>
-      >(Prisma.sql`
-        SELECT
-          t.name "traceName",
-          count(*)::int AS count
-        FROM traces t
-        JOIN observations o ON o.trace_id = t.id
-        WHERE o.type = 'GENERATION'
-          AND o.project_id = ${input.projectId}
-          AND t.project_id = ${input.projectId}
-        GROUP BY 1
-      `);
-
-      // typecheck filter options, needs to include all columns with options
-      const res: ObservationOptions = {
-        model: model
-          .filter((i) => i.model !== null)
-          .map((i) => ({
-            value: i.model as string,
-            count: i._count._all,
-          })),
-        name: name
-          .filter((i) => i.name !== null)
-          .map((i) => ({
-            value: i.name as string,
-            count: i._count._all,
-          })),
-        traceName: traceName
-          .filter((i) => i.traceName !== null)
-          .map((i) => ({
-            value: i.traceName as string,
-            count: i.count,
-          })),
-        scores_avg: scores.map((score) => score.name),
-      };
-      return res;
-    }),
-});
@@ -0,0 +1,61 @@
+import { Readable } from "stream";
+
+import { Prisma, type PrismaClient } from "@prisma/client";
+
+/**
+ * DatabaseReadStream fetches and streams database records in paginated batches,
+ * simulating a streaming behavior. This class is designed for efficient, memory-optimized chunking of
+ * database queries, ideal for processing large datasets with minimal memory overhead. It operates in
+ * object mode, directly streaming database entity objects.
+ *
+ * Note: Due to Prisma's lack of direct streaming support, this class implements a chunk-based approach
+ * rather than true database streaming. It fetches data in paginated batches determined by the pageSize.
+ * GitHub issue: https://github.com/prisma/prisma/issues/5055
+ *
+ * @param prisma - The PrismaClient instance for database queries.
+ * @param rawSqlQuery - A Prisma.Sql object representing the base SQL query, excluding OFFSET and LIMIT.
+ * @param pageSize - Number of records per batch, defining the chunk size.
+ *
+ * The class extends Node.js's Readable stream, using async iteration and Prisma's pagination for scalable
+ * data processing. It's suitable for applications requiring large dataset processing with a low memory footprint.
+ */
+export class DatabaseReadStream<EntityType> extends Readable {
+  private hasNextPage: boolean;
+  private offset: number;
+  private isReading: boolean;
+
+  constructor(
+    private prisma: PrismaClient,
+    private rawSqlQuery: Prisma.Sql,
+    private pageSize: number,
+  ) {
+    super({ objectMode: true }); // Set object mode to true to allow pushing objects to the stream rather than strings or buffers
+
+    this.isReading = false; // Prevent concurrent read executions
+    this.hasNextPage = true;
+    this.offset = 0;
+  }
+
+  async _read() {
+    if (!this.hasNextPage || this.isReading) return; // Avoid calling the database if there's no more data or if a read operation is already in progress
+
+    this.isReading = true;
+
+    try {
+      const query = Prisma.sql`${this.rawSqlQuery} OFFSET ${this.offset} LIMIT ${this.pageSize}`;
+      const rows = await this.prisma.$queryRaw<EntityType[]>(query);
+
+      if (rows.length > 0) {
+        rows.forEach((row) => this.push(row));
+        this.offset += rows.length;
+      } else {
+        this.hasNextPage = false;
+        this.push(null); // Signal end of stream
+      }
+    } catch (error) {
+      this.emit("error", error);
+    } finally {
+      this.isReading = false;
+    }
+  }
+}
@@ -0,0 +1,139 @@
+import {
+  datetimeFilterToPrismaSql,
+  filterToPrismaSql,
+} from "@/src/features/filters/server/filterToPrisma";
+import { orderByToPrismaSql } from "@/src/features/orderBy/server/orderByToPrisma";
+import { observationsTableCols } from "@/src/server/api/definitions/observationsTable";
+import { Prisma } from "@prisma/client";
+
+import { type GenerationsExportInput } from "../exportQuery";
+import { type GetAllGenerationsInput } from "../getAllQuery";
+
+type GetSqlFromInputParams =
+  | {
+      input: GenerationsExportInput;
+      type: "export";
+    }
+  | { input: GetAllGenerationsInput; type: "paginate" };
+
+export function getAllGenerationsSqlQuery({
+  input,
+  type,
+}: GetSqlFromInputParams) {
+  const searchCondition = input.searchQuery
+    ? Prisma.sql`AND (
+        o."id" ILIKE ${`%${input.searchQuery}%`} OR
+        o."name" ILIKE ${`%${input.searchQuery}%`} OR
+        o."model" ILIKE ${`%${input.searchQuery}%`} OR
+        t."name" ILIKE ${`%${input.searchQuery}%`}
+      )`
+    : Prisma.empty;
+
+  const filterCondition = filterToPrismaSql(
+    input.filter,
+    observationsTableCols,
+  );
+
+  const orderByCondition = orderByToPrismaSql(
+    input.orderBy,
+    observationsTableCols,
+  );
+
+  // to improve query performance, add timeseries filter to observation queries as well
+  const startTimeFilter = input.filter.find(
+    (f) => f.column === "start_time" && f.type === "datetime",
+  );
+  const datetimeFilter =
+    startTimeFilter && startTimeFilter.type === "datetime"
+      ? datetimeFilterToPrismaSql(
+          "start_time",
+          startTimeFilter.operator,
+          startTimeFilter.value,
+        )
+      : Prisma.empty;
+
+  // For exports: use a date cutoff filter to ignore ingested rows
+  const dateCutoffFilter =
+    type === "export"
+      ? datetimeFilterToPrismaSql("start_time", "<", new Date())
+      : Prisma.empty;
+
+  // For UI pagination: set LIMIT and OFFSET
+  const pagination =
+    type === "paginate"
+      ? Prisma.sql`LIMIT ${input.limit} OFFSET ${input.page * input.limit}`
+      : Prisma.empty;
+
+  const rawSqlQuery = Prisma.sql`
+      WITH observations_with_latency AS (
+        SELECT
+          o.*,
+          CASE WHEN o.end_time IS NULL THEN NULL ELSE (EXTRACT(EPOCH FROM o."end_time") - EXTRACT(EPOCH FROM o."start_time"))::double precision END AS "latency"
+        FROM observations_view o
+        WHERE o.type = 'GENERATION'
+        AND o.project_id = ${input.projectId}
+        ${datetimeFilter}
+        ${dateCutoffFilter}
+      ),
+      -- used for filtering
+      scores_avg AS (
+        SELECT
+          trace_id,
+          observation_id,
+          jsonb_object_agg(name::text, avg_value::double precision) AS scores_avg
+        FROM (
+          SELECT
+            trace_id,
+            observation_id,
+            name,
+            avg(value) avg_value
+          FROM
+            scores
+          GROUP BY
+            1,
+            2,
+            3
+          ORDER BY
+            1) tmp
+        GROUP BY
+          1, 2
+      )
+      SELECT
+        o.id,
+        o.name,
+        o.model,
+        o.start_time as "startTime",
+        o.end_time as "endTime",
+        o.latency,
+        o.input,
+        o.output,
+        o.metadata,
+        o.trace_id as "traceId",
+        t.name as "traceName",
+        o.completion_start_time as "completionStartTime",
+        o.prompt_tokens as "promptTokens",
+        o.completion_tokens as "completionTokens",
+        o.total_tokens as "totalTokens",
+        o.level,
+        o.status_message as "statusMessage",
+        o.version,
+        o.model_id as "modelId",
+        o.input_price as "inputPrice",
+        o.output_price as "outputPrice",
+        o.total_price as "totalPrice",
+        o.calculated_input_cost as "calculatedInputCost",
+        o.calculated_output_cost as "calculatedOutputCost",
+        o.calculated_total_cost as "calculatedTotalCost"
+      FROM observations_with_latency o
+      JOIN traces t ON t.id = o.trace_id
+      LEFT JOIN scores_avg AS s_avg ON s_avg.trace_id = t.id and s_avg.observation_id = o.id
+      WHERE
+        t.project_id = ${input.projectId}
+        ${searchCondition}
+        ${filterCondition}
+        ${orderByCondition}
+        ${pagination}
+    `;
+
+  return { rawSqlQuery, datetimeFilter, searchCondition, filterCondition };
+}
@@ -0,0 +1,91 @@
+import { type Transform } from "stream";
+import { z } from "zod";
+
+import { env } from "@/src/env.mjs";
+import {
+  exportFileFormats,
+  exportOptions,
+} from "@/src/server/api/interfaces/exportTypes";
+import { S3StorageService } from "@/src/server/api/services/S3StorageService";
+import { protectedProjectProcedure } from "@/src/server/api/trpc";
+import { type ObservationView } from "@prisma/client";
+
+import { DatabaseReadStream } from "../db/DatabaseReadStream";
+import { getAllGenerationsSqlQuery } from "../db/getAllGenerationsSqlQuery";
+import { GenerationTableOptions } from "../utils/GenerationTableOptions";
+import { transformStreamToCsv } from "./transforms/transformStreamToCsv";
+import { transformStreamToJson } from "./transforms/transformStreamToJson";
+import { transformStreamToJsonLines } from "./transforms/transformStreamToJsonLines";
+
+const generationsExportInput = GenerationTableOptions.extend({
+  fileFormat: z.enum(exportFileFormats),
+});
+export type GenerationsExportInput = z.infer<typeof generationsExportInput>;
+export type GenerationsExportResult =
+  | {
+      type: "s3";
+      fileName: string;
+      url: string;
+    }
+  | {
+      type: "data";
+      fileName: string;
+      data: string;
+    };
+
+export const generationsExportQuery = protectedProjectProcedure
+  .input(generationsExportInput)
+  .query<GenerationsExportResult>(async ({ input, ctx }) => {
+    const { rawSqlQuery } = getAllGenerationsSqlQuery({
+      input,
+      type: "export",
+    });
+    const queryPageSize = env.DB_EXPORT_PAGE_SIZE ?? 1000;
+    const dbReadStream = new DatabaseReadStream<ObservationView>(
+      ctx.prisma,
+      rawSqlQuery,
+      queryPageSize,
+    );
+
+    const streamTransformations: Record<
+      typeof input.fileFormat,
+      () => Transform
+    > = {
+      CSV: transformStreamToCsv,
+      JSON: transformStreamToJson,
+      "OPENAI-JSONL": transformStreamToJsonLines,
+    };
+    const transformation = streamTransformations[input.fileFormat];
+
+    const fileStream = dbReadStream.pipe(transformation());
+    const fileDate = new Date().toISOString();
+    const fileExtension = exportOptions[input.fileFormat].extension;
+    const fileName = `lf-export-${input.projectId}-${fileDate}.${fileExtension}`;
+
+    if (S3StorageService.getIsS3StorageConfigured(env)) {
+      const { signedUrl } = await new S3StorageService().uploadFile({
+        fileName,
+        fileType: exportOptions[input.fileFormat].fileType,
+        data: fileStream,
+      });
+
+      return {
+        type: "s3",
+        url: signedUrl,
+        fileName,
+      };
+    }
+
+    // Fall back to returning the data directly. This might fail for large exports due to memory constraints.
+    // Self-hosted instances should always run with sufficient memory or have S3 configured to avoid this.
+    let fileOutputString = "";
+    for await (const chunk of fileStream) {
+      fileOutputString += chunk;
+    }
+
+    return {
+      type: "data",
+      data: fileOutputString,
+      fileName,
+    };
+  });
@@ -0,0 +1,59 @@
+import { Transform, type TransformCallback } from "stream";
+
+import { usdFormatter } from "@/src/utils/numbers";
+
+import type { ObservationView } from "@prisma/client";
+
+export function transformStreamToCsv(): Transform {
+  let isFirstChunk = true;
+
+  return new Transform({
+    objectMode: true,
+    transform(
+      row: ObservationView,
+      encoding: BufferEncoding,
+      callback: TransformCallback,
+    ): void {
+      if (isFirstChunk) {
+        // Output the header if it's the first chunk
+        const csvHeader = [
+          "traceId",
+          "name",
+          "model",
+          "startTime",
+          "endTime",
+          "cost",
+          "prompt",
+          "completion",
+          "metadata",
+        ];
+
+        this.push(csvHeader.join(",") + "\n");
+
+        isFirstChunk = false;
+      }
+
+      // Convert the generation object to a CSV line and push it
+      const csvRow = [
+        row.traceId,
+        row.name ?? "",
+        row.model ?? "",
+        row.startTime.toISOString(),
+        row.endTime?.toISOString() ?? "",
+        row.calculatedTotalCost
+          ? usdFormatter(row.calculatedTotalCost.toNumber(), 2, 8)
+          : "",
+        JSON.stringify(row.input),
+        JSON.stringify(row.output),
+        JSON.stringify(row.metadata),
+      ].map((field) => {
+        const str = typeof field === "string" ? field : String(field);
+        return `"${str.replace(/"/g, '""')}"`;
+      });
+
+      this.push(csvRow.join(",") + "\n");
+
+      callback();
+    },
+  });
+}
@@ -0,0 +1,40 @@
+import { Transform, type TransformCallback } from "stream";
+
+import type { ObservationView } from "@prisma/client";
+
+export function transformStreamToJson(): Transform {
+  let isFirstElement = true;
+
+  return new Transform({
+    objectMode: true,
+
+    transform(
+      row: ObservationView,
+      encoding: BufferEncoding,
+      callback: TransformCallback,
+    ): void {
+      if (isFirstElement) {
+        this.push("["); // Push the opening bracket for the first element
+        isFirstElement = false; // Reset the flag after the first element
+      } else {
+        this.push(","); // For subsequent elements, prepend a comma
+      }
+
+      this.push(JSON.stringify(row)); // Push the current row as a JSON string
+
+      callback();
+    },
+
+    // 'final' is called when there is no more data to be consumed, but before the stream is finished.
+    final(callback: TransformCallback): void {
+      if (isFirstElement) {
+        // If no rows were processed, the opening bracket has not been pushed yet.
+        this.push("[]"); // Push an empty array to ensure valid JSON.
+      } else {
+        this.push("]"); // Close JSON array
+      }
+
+      callback();
+    },
+  });
+}
@@ -0,0 +1,49 @@
+import { Transform, type TransformCallback } from "stream";
+import { z } from "zod";
+
+import { jsonSchema } from "@/src/utils/zod";
+import { type ObservationView } from "@prisma/client";
+
+export function transformStreamToJsonLines(): Transform {
+  return new Transform({
+    objectMode: true,
+    transform(
+      row: ObservationView,
+      encoding: BufferEncoding,
+      callback: TransformCallback,
+    ): void {
+      const inputSchemaOpenAI = z.array(
+        z.object({
+          role: z.enum(["system", "user", "assistant"]),
+          content: z.string(),
+        }),
+      );
+
+      const outputSchema = z
+        .object({
+          completion: jsonSchema,
+        })
+        .or(jsonSchema);
+
+      const parsedInput = inputSchemaOpenAI.safeParse(row.input);
+      const parsedOutput = outputSchema.safeParse(row.output);
+
+      if (parsedInput.success && parsedOutput.success) {
+        const output = JSON.stringify([
+          ...parsedInput.data,
+          {
+            role: "assistant",
+            content:
+              typeof parsedOutput.data === "object" &&
+              "completion" in parsedOutput.data
+                ? JSON.stringify(parsedOutput.data.completion)
+                : JSON.stringify(parsedOutput.data),
+          },
+        ]);
+        this.push(output + "\n");
+      }
+
+      callback();
+    },
+  });
+}
@@ -0,0 +1,75 @@
+import { z } from "zod";
+
+import { type ObservationOptions } from "@/src/server/api/definitions/observationsTable";
+import { protectedProjectProcedure } from "@/src/server/api/trpc";
+import { Prisma } from "@prisma/client";
+
+export const filterOptionsQuery = protectedProjectProcedure
+  .input(z.object({ projectId: z.string() }))
+  .query(async ({ input, ctx }) => {
+    const queryFilter = {
+      projectId: input.projectId,
+      type: "GENERATION",
+    } as const;
+
+    const scores = await ctx.prisma.score.groupBy({
+      where: {
+        observation: {
+          projectId: input.projectId,
+        },
+      },
+      by: ["name"],
+    });
+
+    const model = await ctx.prisma.observation.groupBy({
+      by: ["model"],
+      where: queryFilter,
+      _count: { _all: true },
+    });
+    const name = await ctx.prisma.observation.groupBy({
+      by: ["name"],
+      where: queryFilter,
+      _count: { _all: true },
+    });
+    const traceName = await ctx.prisma.$queryRaw<
+      Array<{
+        traceName: string | null;
+        count: number;
+      }>
+    >(Prisma.sql`
+        SELECT
+          t.name "traceName",
+          count(*)::int AS count
+        FROM traces t
+        JOIN observations o ON o.trace_id = t.id
+        WHERE o.type = 'GENERATION'
+          AND o.project_id = ${input.projectId}
+          AND t.project_id = ${input.projectId}
+        GROUP BY 1
+      `);
+
+    // typecheck filter options, needs to include all columns with options
+    const res: ObservationOptions = {
+      model: model
+        .filter((i) => i.model !== null)
+        .map((i) => ({
+          value: i.model as string,
+          count: i._count._all,
+        })),
+      name: name
+        .filter((i) => i.name !== null)
+        .map((i) => ({
+          value: i.name as string,
+          count: i._count._all,
+        })),
+      traceName: traceName
+        .filter((i) => i.traceName !== null)
+        .map((i) => ({
+          value: i.traceName as string,
+          count: i.count,
+        })),
+      scores_avg: scores.map((score) => score.name),
+    };
+
+    return res;
+  });
@@ -0,0 +1,100 @@
+import { type z } from "zod";
+
+import { protectedProjectProcedure } from "@/src/server/api/trpc";
+import { paginationZod } from "@/src/utils/zod";
+import { type ObservationView, Prisma } from "@prisma/client";
+
+import { GenerationTableOptions } from "./utils/GenerationTableOptions";
+import { getAllGenerationsSqlQuery } from "@/src/server/api/routers/generations/db/getAllGenerationsSqlQuery";
+
+const getAllGenerationsInput = GenerationTableOptions.extend({
+  ...paginationZod,
+});
+export type GetAllGenerationsInput = z.infer<typeof getAllGenerationsInput>;
+
+export const getAllQuery = protectedProjectProcedure
+  .input(getAllGenerationsInput)
+  .query(async ({ input, ctx }) => {
+    const { rawSqlQuery, datetimeFilter, filterCondition, searchCondition } =
+      getAllGenerationsSqlQuery({ input, type: "paginate" });
+
+    const generations = await ctx.prisma.$queryRaw<
+      (ObservationView & {
+        traceId: string;
+        traceName: string;
+        latency: number | null;
+      })[]
+    >(rawSqlQuery);
+
+    const totalGenerations = await ctx.prisma.$queryRaw<
+      Array<{ count: bigint }>
+    >(
+      Prisma.sql`
+      WITH observations_with_latency AS (
+        SELECT
+          o.*,
+          CASE WHEN o.end_time IS NULL THEN NULL ELSE (EXTRACT(EPOCH FROM o."end_time") - EXTRACT(EPOCH FROM o."start_time"))::double precision END AS "latency"
+        FROM observations_view o
+        WHERE o.type = 'GENERATION'
+        AND o.project_id = ${input.projectId}
+        ${datetimeFilter}
+      ),
+      -- used for filtering
+      scores_avg AS (
+        SELECT
+          trace_id,
+          observation_id,
+          jsonb_object_agg(name::text, avg_value::double precision) AS scores_avg
+        FROM (
+          SELECT
+            trace_id,
+            observation_id,
+            name,
+            avg(value) avg_value
+          FROM
+            scores
+          GROUP BY
+            1,
+            2,
+            3
+          ORDER BY
+            1) tmp
+        GROUP BY
+          1, 2
+      )
+      SELECT
+        count(*)
+      FROM observations_with_latency o
+      JOIN traces t ON t.id = o.trace_id
+      LEFT JOIN scores_avg AS s_avg ON s_avg.trace_id = t.id and s_avg.observation_id = o.id
+      WHERE
+        t.project_id = ${input.projectId}
+        ${searchCondition}
+        ${filterCondition}
+    `,
+    );
+
+    const scores = await ctx.prisma.score.findMany({
+      where: {
+        trace: {
+          projectId: input.projectId,
+        },
+        observationId: {
+          in: generations.map((gen) => gen.id),
+        },
+      },
+    });
+    const count = totalGenerations[0]?.count;
+    return {
+      totalCount: count ? Number(count) : undefined,
+      generations: generations.map((generation) => {
+        const filteredScores = scores.filter(
+          (s) => s.observationId === generation.id,
+        );
+        return {
+          ...generation,
+          scores: filteredScores,
+        };
+      }),
+    };
+  });
@@ -0,0 +1,11 @@
+import { createTRPCRouter } from "@/src/server/api/trpc";
+
+import { generationsExportQuery } from "./exportQuery";
+import { filterOptionsQuery } from "./filterOptionsQuery";
+import { getAllQuery } from "./getAllQuery";
+
+export const generationsRouter = createTRPCRouter({
+  all: getAllQuery,
+  export: generationsExportQuery,
+  filterOptions: filterOptionsQuery,
+});
@@ -0,0 +1,10 @@
+import { z } from "zod";
+import { singleFilter } from "@/src/server/api/interfaces/filters";
+import { orderBy } from "@/src/server/api/interfaces/orderBy";
+
+export const GenerationTableOptions = z.object({
+  projectId: z.string(), // Required for protectedProjectProcedure
+  filter: z.array(singleFilter),
+  searchQuery: z.string().nullable(),
+  orderBy: orderBy,
+});
@@ -0,0 +1,98 @@
+import type { Readable } from "stream";
+import { env } from "@/src/env.mjs";
+import { GetObjectCommand, S3Client } from "@aws-sdk/client-s3";
+import { Upload } from "@aws-sdk/lib-storage";
+import { getSignedUrl } from "@aws-sdk/s3-request-presigner";
+
+type UploadFile = {
+  fileName: string;
+  fileType: string;
+  data: Readable | string;
+};
+
+class S3StorageService {
+  private client: S3Client;
+
+  constructor() {
+    if (!S3StorageService.getIsS3StorageConfigured(env)) {
+      throw new Error("S3 bucket is not configured");
+    }
+
+    this.client = new S3Client({
+      credentials: {
+        accessKeyId: env.S3_ACCESS_KEY_ID,
+        secretAccessKey: env.S3_SECRET_ACCESS_KEY,
+      },
+      endpoint: env.S3_ENDPOINT,
+      region: env.S3_REGION,
+    });
+  }
+
+  public async uploadFile({
+    fileName,
+    fileType,
+    data,
+  }: UploadFile): Promise<{ signedUrl: string }> {
+    try {
+      await new Upload({
+        client: this.client,
+        params: {
+          Bucket: env.S3_BUCKET_NAME,
+          Key: fileName,
+          Body: data,
+          ContentType: fileType,
+        },
+      }).done();
+
+      const expiresInOneHour = 60 * 60;
+      const signedUrl = await this.getSignedUrl(fileName, expiresInOneHour);
+
+      return { signedUrl };
+    } catch (err) {
+      console.error(err);
+
+      throw new Error("Failed to upload to S3 or generate signed URL");
+    }
+  }
+
+  private async getSignedUrl(
+    fileName: string,
+    ttlSeconds: number,
+  ): Promise<string> {
+    try {
+      return await getSignedUrl(
+        this.client,
+        new GetObjectCommand({
+          Bucket: env.S3_BUCKET_NAME,
+          Key: fileName,
+          ResponseContentDisposition: `attachment; filename="${fileName}"`,
+        }),
+        { expiresIn: ttlSeconds },
+      );
+    } catch (err) {
+      throw Error("Failed to generate signed URL");
+    }
+  }
+
+  static getIsS3StorageConfigured(
+    currentEnv: Env,
+  ): currentEnv is S3ConfiguredEnv {
+    return Boolean(
+      currentEnv.S3_BUCKET_NAME &&
+        currentEnv.S3_ACCESS_KEY_ID &&
+        currentEnv.S3_SECRET_ACCESS_KEY &&
+        currentEnv.S3_ENDPOINT &&
+        currentEnv.S3_REGION,
+    );
+  }
+}
+
+export { S3StorageService };
+
+type Env = typeof env;
+type S3ConfiguredEnv = Env & {
+  S3_ACCESS_KEY_ID: string;
+  S3_SECRET_ACCESS_KEY: string;
+  S3_ENDPOINT: string;
+  S3_REGION: string;
+};
Author	SHA1	Message	Date
Marc Klingen	2608878d20	chore: release v2.4.2 CI/CD / lint (push) Waiting to run Details CI/CD / test-docker-build (push) Waiting to run Details CI/CD / tests (18) (push) Waiting to run Details CI/CD / tests (20) (push) Waiting to run Details CI/CD / e2e-tests (push) Waiting to run Details CI/CD / all-ci-passed (push) Blocked by required conditions Details CI/CD / push-docker-image (push) Blocked by required conditions Details release.yml / release (push) Waiting to run Details	2024-02-13 15:53:15 +01:00
Marc Klingen	62a2857619	docs: add DB_EXPORT_PAGE_SIZE to .env.prod.example	2024-02-13 15:48:49 +01:00
Yash Khare GitHub Marc Klingen	0c1942c2b3	fix(docker): remove the need for runtime variables at build time (#1067 ) --------- Co-authored-by: Marc Klingen <git@marcklingen.com>	2024-02-13 15:40:16 +01:00
Hassieb Pakzad GitHub Max Deichmann	b95c93cdb1	fix: memory optimizes generation.export trpc to allow for larger exports (#1124 ) * fix: memory optimizes generation.export trpc * test: disable S3 uploads in export test * refactor: modularize generations.export * feat: adds streaming to S3 bucket * feat: adds DB_EXPORT_PAGE_SIZE environment variable * refactor: clean up generationsRouter --------- Co-authored-by: Max Deichmann <m.deichmann@tum.de>	2024-02-13 12:55:12 +00:00