feat: implement direct CSV upload support for cloud storage

app/api/csv-temp/[id]/route.ts

@@ -1,6 +1,6 @@
 import { NextRequest, NextResponse } from "next/server";
 import { Readable } from "stream";
-import { getCsvStorageProvider } from "@/app/lib/csv-storage";
+import { getCsvStorageProvider, getResolvedCsvStorageMode } from "@/app/lib/csv-storage";
 import { openLocalCsvReadStream } from "@/app/lib/csv-storage-local";
 import { getClient } from "@/app/api/auth/[...nextauth]/options";
 import { getCorsHeaders, resolveReadOnly } from "@/app/api/utils";
@@ -29,6 +29,12 @@ export async function GET(
     }
     const key = normalizeCsvKey(id);
 
+    // This endpoint is only for local HTTP-serving mode. If CSV storage is S3
+    // or Blob, do not expose any local read path.
+    if (getResolvedCsvStorageMode() !== "local") {
+        return new NextResponse("Not found.", { status: 404 });
+    }
+
     const url = new URL(request.url);
     const owner = url.searchParams.get("o") ?? "";
     const token = url.searchParams.get("t") ?? "";

app/api/csv-temp/direct/route.ts

@@ -0,0 +1,133 @@
+import { NextRequest, NextResponse } from "next/server";
+import { getCorsHeaders, resolveReadOnly } from "../../utils";
+import { getClient } from "../../auth/[...nextauth]/options";
+import {
+    getCsvStorageProvider,
+    getResolvedCsvStorageMode,
+    type CsvDirectUploadRequest,
+    type CsvDirectUploadTarget,
+} from "@/app/lib/csv-storage";
+import { generateCsvKey, hashOwner } from "@/app/lib/csv-key";
+import { CSV_UPLOAD_ENABLED } from "@/lib/graphUpload";
+import { getCsvMaxUploadBytes } from "@/app/lib/csv-temp-config";
+
+interface DirectUploadBody {
+    filename?: string;
+    contentType?: string;
+    sizeBytes?: number;
+}
+
+function sanitizeDirectUploadRequest(body: unknown): CsvDirectUploadRequest | null {
+    if (!body || typeof body !== "object" || Array.isArray(body)) return null;
+
+    const input = body as Record<string, unknown>;
+    if (typeof input.filename !== "string" || !input.filename.trim()) return null;
+    if (typeof input.sizeBytes !== "number" || !Number.isFinite(input.sizeBytes) || input.sizeBytes <= 0) return null;
+    if (input.contentType !== undefined && typeof input.contentType !== "string") return null;
+
+    const maxBytes = getCsvMaxUploadBytes();
+    const filename = input.filename.trim();
+    const contentType = (typeof input.contentType === "string" ? input.contentType : "text/csv").trim() || "text/csv";
+
+    if (input.sizeBytes > maxBytes) {
+        return {
+            filename,
+            contentType,
+            sizeBytes: Number.POSITIVE_INFINITY,
+        };
+    }
+
+    return {
+        filename,
+        contentType,
+        sizeBytes: Math.floor(input.sizeBytes),
+    };
+}
+
+export async function OPTIONS(request: Request) {
+    return new NextResponse(null, { status: 204, headers: getCorsHeaders(request) });
+}
+
+export async function POST(request: NextRequest) {
+    if (!CSV_UPLOAD_ENABLED) {
+        return NextResponse.json(
+            { message: "CSV upload is temporarily disabled." },
+            { status: 403, headers: getCorsHeaders(request) }
+        );
+    }
+
+    const session = await getClient(request);
+    if (session instanceof NextResponse) return session;
+
+    if (resolveReadOnly(request, session.user.role)) {
+        return NextResponse.json(
+            { message: "You do not have permission to upload files." },
+            { status: 403, headers: getCorsHeaders(request) }
+        );
+    }
+
+    let body: unknown;
+    try {
+        body = await request.json();
+    } catch {
+        return NextResponse.json(
+            { message: "Invalid request body." },
+            { status: 400, headers: getCorsHeaders(request) }
+        );
+    }
+
+    const parsed = sanitizeDirectUploadRequest(body);
+    if (!parsed) {
+        return NextResponse.json(
+            { message: "filename and sizeBytes are required." },
+            { status: 400, headers: getCorsHeaders(request) }
+        );
+    }
+
+    if (!Number.isFinite(parsed.sizeBytes)) {
+        const maxMb = Math.floor(getCsvMaxUploadBytes() / (1024 * 1024));
+        return NextResponse.json(
+            { message: `File exceeds the ${maxMb} MB limit.` },
+            { status: 413, headers: getCorsHeaders(request) }
+        );
+    }
+
+    const mode = getResolvedCsvStorageMode();
+    if (mode === "local") {
+        return NextResponse.json(
+            { message: "Direct upload is not enabled for local CSV storage." },
+            { status: 409, headers: getCorsHeaders(request) }
+        );
+    }
+
+    const owner = hashOwner(session.user.id);
+    const key = generateCsvKey();
+    const provider = getCsvStorageProvider();
+
+    let target: CsvDirectUploadTarget | null;
+    try {
+        target = await provider.createDirectUploadTarget?.(owner, key, parsed) ?? null;
+    } catch (error) {
+        console.error("[csv-temp/direct] failed to prepare direct upload target:", error);
+        return NextResponse.json(
+            { message: "Failed to prepare direct upload." },
+            { status: 500, headers: getCorsHeaders(request) }
+        );
+    }
+
+    if (!target) {
+        return NextResponse.json(
+            { message: "Direct upload is not available for the configured CSV storage." },
+            { status: 409, headers: getCorsHeaders(request) }
+        );
+    }
+
+    return NextResponse.json(
+        {
+            key,
+            mode,
+            upload: target,
+        },
+        { status: 201, headers: getCorsHeaders(request) }
+    );
+}

app/api/csv-temp/route.ts

@@ -1,7 +1,7 @@
 import { NextRequest, NextResponse } from "next/server";
 import { getCorsHeaders, resolveReadOnly } from "../utils";
 import { getClient } from "../auth/[...nextauth]/options";
-import { getCsvStorageProvider } from "@/app/lib/csv-storage";
+import { getCsvStorageProvider, getResolvedCsvStorageMode } from "@/app/lib/csv-storage";
 import { generateCsvKey, hashOwner } from "@/app/lib/csv-key";
 import { CSV_UPLOAD_ENABLED } from "@/lib/graphUpload";
 import { streamCsvUpload } from "./stream-csv";
@@ -44,6 +44,17 @@ export async function POST(request: NextRequest) {
 
     const owner = hashOwner(session.user.id);
     const key = generateCsvKey();
+
+    // In cloud modes, CSV bytes must go directly from browser -> storage via
+    // /api/csv-temp/direct. Reject proxied multipart uploads here to prevent
+    // accidental backend buffering/streaming fallback.
+    if (getResolvedCsvStorageMode() !== "local") {
+        return NextResponse.json(
+            { message: "Direct upload required: use /api/csv-temp/direct for cloud CSV storage." },
+            { status: 409, headers: getCorsHeaders(request) }
+        );
+    }
+
     const provider = getCsvStorageProvider();
 
     const result = await streamCsvUpload(request, (body) => provider.store(owner, key, body));

app/api/graph/[graph]/load-csv/route.ts

@@ -5,13 +5,41 @@ import { getCsvStorageProvider } from "@/app/lib/csv-storage";
 import { hashOwner, isValidCsvKey, normalizeCsvKey } from "@/app/lib/csv-key";
 import { assertFalkorFetchableCsvUrl, CsvUrlConfigError } from "@/app/lib/csv-load-url";
 import { CSV_UPLOAD_ENABLED, containsLoadCsv } from "@/lib/graphUpload";
+import { CSV_HEAD_INSPECT_BYTES, headLooksBinary, stripLeadingBom } from "@/app/api/csv-temp/csv-head";
 
 /**
  * Interval for keep-alive heartbeat bytes streamed while a (potentially long,
  * atomic) LOAD CSV runs, so idle proxies/load balancers don't drop the request.
  */
 const HEARTBEAT_INTERVAL_MS = 15_000;
 
+async function validateCsvHeadFromUrl(csvUrl: string): Promise<string | null> {
+    try {
+        const response = await fetch(csvUrl, {
+            method: "GET",
+            headers: { Range: `bytes=0-${CSV_HEAD_INSPECT_BYTES - 1}` },
+            cache: "no-store",
+        });
+
+        // Let LOAD CSV surface connectivity/auth errors with provider-specific
+        // diagnostics; this pre-check is only for content sanity validation.
+        if (!response.ok) return null;
+
+        const raw = Buffer.from(await response.arrayBuffer());
+        const head = stripLeadingBom(raw);
+
+        if (head.length === 0) {
+            return "The uploaded CSV is empty.";
+        }
+        if (headLooksBinary(head)) {
+            return "The uploaded file does not appear to be a valid CSV (contains binary data).";
+        }
+        return null;
+    } catch {
+        return null;
+    }
+}
+
 interface LoadCsvBody {
     key?: string;
     withHeaders?: boolean;
@@ -131,6 +159,14 @@ export async function POST(
             throw configError;
         }
 
+        const csvHeadError = await validateCsvHeadFromUrl(csvUrl);
+        if (csvHeadError) {
+            return NextResponse.json(
+                { message: csvHeadError },
+                { status: 422, headers: getCorsHeaders(request) }
+            );
+        }
+
         const graph = session.client.selectGraph(graphId);
         const withHeadersClause = withHeaders ? "WITH HEADERS " : "";
         const loadCsvQuery = `LOAD CSV ${withHeadersClause}FROM $csvUrl AS row\n${body.trim()}`;
@@ -169,8 +205,20 @@ export async function POST(
                     const raw = (queryError as Error).message || "Failed to execute the LOAD CSV query.";
                     const isHeaderError = /failed reading csv header row/i.test(raw);
                     const isFetchError = /(unsupported uri|error opening csv uri)/i.test(raw);
+                    const csvUrlHost = (() => {
+                        try {
+                            return new URL(csvUrl).host;
+                        } catch {
+                            return "unknown";
+                        }
+                    })();
+                    const isLikelyLocalEndpoint = /localhost:|127\.0\.0\.1:/i.test(csvUrl);
                     const message = isHeaderError
-                        ? "Failed reading CSV header row. Verify the file has a valid header row, or turn off 'Use CSV headers' and access columns as row[0], row[1], ..."
+                        ? (
+                            (isFetchError && isLikelyLocalEndpoint)
+                                ? `Failed reading CSV header row. FalkorDB also reported it could not reach the storage URL from inside its container (host: ${csvUrlHost}). Set S3_READ_URL_HOST to a host reachable by FalkorDB (for example 172.17.0.1) and retry, or run FalkorDB on the host network.`
+                                : `Failed reading CSV header row. Verify the file has a valid header row, or turn off 'Use CSV headers' and access columns as row[0], row[1], ... (resolved storage host: ${csvUrlHost}).`
+                        )
                         : isFetchError
                             ? "FalkorDB could not fetch the uploaded CSV. Check the CSV storage configuration (CSV_STORAGE / CSV_SERVE_BASE_URL / IMPORT_FOLDER) so the database can reach the file."
                             : raw;

app/api/graph/[graph]/upload/route.ts

@@ -85,7 +85,9 @@ export async function POST(
     const batchText = await fs.promises.readFile(storedUpload.filePath, "utf-8");
 
     try {
-      const count = await executeCypherBatch(graph, batchText);
+      const count = await executeCypherBatch(graph, batchText, {
+        sourceExtension: extension as ".txt" | ".cql" | ".cypher",
+      });
 
       return NextResponse.json(
         { message: `Executed ${count} Cypher statement(s).` },

app/api/graph/[graph]/upload/upload-utils.test.ts

@@ -66,3 +66,56 @@ test("executeCypherBatch rejects a batch containing LOAD CSV without executing a
   );
   assert.equal(querySpy.mock.callCount(), 0);
 });
+
+test("executeCypherBatch (.txt) ignores markdown fences, directives, and non-Cypher full-line comments", async () => {
+  const { graph, querySpy } = makeGraph();
+
+  const count = await executeCypherBatch(
+    graph,
+    `
+\uFEFF
+\`\`\`cypher
+// comment line
+-- another comment
+# heading/comment
+:begin
+CREATE (:A);
+MATCH (n) RETURN n;
+\`\`\`
+`,
+    { sourceExtension: ".txt" }
+  );
+
+  assert.equal(count, 2);
+  assert.equal(querySpy.mock.callCount(), 2);
+  assert.equal(querySpy.mock.calls[0].arguments[0], "CREATE (:A)");
+  assert.equal(querySpy.mock.calls[1].arguments[0], "MATCH (n) RETURN n");
+});
+
+test("executeCypherBatch (.cypher) does not apply txt normalization", async () => {
+  const { graph } = makeGraph(async (query) => {
+    if (String(query).includes("# comment")) {
+      throw new Error("syntax error at #");
+    }
+    return { data: [] };
+  });
+
+  await assert.rejects(
+    () => executeCypherBatch(graph, "# comment\nCREATE (:A);", { sourceExtension: ".cypher" }),
+    /Failed to execute Cypher statement 1: syntax error at #/
+  );
+});
+
+test("executeCypherBatch (.txt) keeps unknown colon directives so they fail loudly", async () => {
+  const { graph } = makeGraph(async (query) => {
+    if (String(query).startsWith(":param")) {
+      throw new Error("syntax error at :");
+    }
+    return { data: [] };
+  });
+
+  await assert.rejects(
+    () => executeCypherBatch(graph, ":param x => 1\nCREATE (:A);", { sourceExtension: ".txt" }),
+    /Failed to execute Cypher statement 1: syntax error at :/
+  );
+});

app/api/graph/[graph]/upload/upload-utils.ts

@@ -4,13 +4,49 @@ import {
   splitCypherStatements,
 } from "../../../../../lib/graphUpload.ts";
 
+type CypherUploadExtension = ".txt" | ".cql" | ".cypher";
+
+interface ExecuteCypherBatchOptions {
+  sourceExtension?: CypherUploadExtension;
+}
+
 /**
- * Split a Cypher batch file into statements and execute them sequentially.
- * Returns the number of statements executed. Failures are annotated with the
- * statement number so partial-batch errors are actionable.
+ * .txt uploads are often copied from docs/issues and may include markdown
+ * fences, wrappers, or full-line comments. Strip only obvious non-Cypher
+ * lines to make uploads more forgiving without changing trusted .cypher/.cql.
+ *
+ * Note: FalkorDB executes Cypher only. This does not add SQL support; it only
+ * discards non-Cypher noise in loose text files before Cypher splitting.
  */
-export async function executeCypherBatch(graph: Graph, batchText: string): Promise<number> {
-  const statements = splitCypherStatements(batchText);
+function normalizeTxtBatch(batchText: string): string {
+  const withoutBom = batchText.replace(/^\uFEFF/, "");
+  const lines = withoutBom.split(/\r?\n/);
+
+  const kept = lines.filter((line) => {
+    const trimmed = line.trim();
+    if (!trimmed) return true;
+
+    if (trimmed.startsWith("```")) return false;
+    if (trimmed.startsWith("//")) return false;
+    if (trimmed.startsWith("--")) return false;
+    if (trimmed.startsWith("#")) return false;
+    // Keep arbitrary ":..." lines (for fail-loud behavior) and only drop
+    // known transaction wrappers often copied from cypher-shell snippets.
+    if (/^:(begin|commit)\b/i.test(trimmed)) return false;
+
+    return true;
+  });
+
+  return kept.join("\n");
+}
+
+export async function executeCypherBatch(
+  graph: Graph,
+  batchText: string,
+  options: ExecuteCypherBatchOptions = {}
+): Promise<number> {
+  const normalizedBatch = options.sourceExtension === ".txt" ? normalizeTxtBatch(batchText) : batchText;
+  const statements = splitCypherStatements(normalizedBatch);
 
   // Reject the whole batch before executing anything if any statement smuggles
   // its own `LOAD CSV` clause, so an uploaded batch can never trigger an