Merge pull request #83 from rosemary21/feature/export-streams-csv

feature/export-streams-csv

Author: greatest0fallt1me

src/api/v1/streams.test.ts

@@ -3,6 +3,17 @@ import app from "../../index";
 import { StreamRepository } from "../../repositories/streamRepository";
 import { Stream } from "../../db/schema";
 
+const TEST_SECRET = "test-jwt-secret-that-is-at-least-32-chars!!";
+
+// Inject JWT_SECRET so the requireAuth middleware has something to check against
+beforeAll(() => {
+  process.env.JWT_SECRET = TEST_SECRET;
+});
+
+afterAll(() => {
+  delete process.env.JWT_SECRET;
+});
+
 describe("Stream API Routes", () => {
   beforeAll(() => {
     process.env.API_KEYS = "test-1234";
@@ -265,3 +276,212 @@ describe("Stream API Routes", () => {
     });
   });
 });
+
+// ---------------------------------------------------------------------------
+// CSV Export endpoint
+// ---------------------------------------------------------------------------
+
+const makeExportStream = (overrides: Record<string, unknown> = {}) => ({
+  id: "aaaaaaaa-1111-1111-1111-000000000001",
+  payer: "0xPayerAddress",
+  recipient: "0xRecipientAddress",
+  status: "active",
+  ratePerSecond: "0.001",
+  startTime: new Date("2024-01-01T00:00:00Z"),
+  endTime: null,
+  totalAmount: "3600.0",
+  lastSettledAt: new Date("2024-01-01T00:00:00Z"),
+  createdAt: new Date("2024-06-01T12:00:00Z"),
+  updatedAt: new Date("2024-06-01T12:00:00Z"),
+  ...overrides,
+});
+
+describe("GET /api/v1/streams/export.csv", () => {
+  describe("authentication", () => {
+    it("should return 401 when no Authorization header is present", async () => {
+      const response = await request(app).get("/api/v1/streams/export.csv");
+      expect(response.status).toBe(401);
+      expect(response.body.error).toBe("Unauthorized");
+    });
+
+    it("should return 401 when the token is wrong", async () => {
+      const response = await request(app)
+        .get("/api/v1/streams/export.csv")
+        .set("Authorization", "Bearer wrong-token");
+      expect(response.status).toBe(401);
+      expect(response.body.error).toBe("Unauthorized");
+    });
+
+    it("should return 401 when Authorization header is malformed (no Bearer prefix)", async () => {
+      const response = await request(app)
+        .get("/api/v1/streams/export.csv")
+        .set("Authorization", TEST_SECRET);
+      expect(response.status).toBe(401);
+      expect(response.body.error).toBe("Unauthorized");
+    });
+  });
+
+  describe("response format", () => {
+    it("should return Content-Type text/csv for a valid request", async () => {
+      const spy = jest
+        .spyOn(StreamRepository.prototype, "findForExport")
+        .mockResolvedValue({ rows: [], nextCursor: null });
+
+      const response = await request(app)
+        .get("/api/v1/streams/export.csv")
+        .set("Authorization", `Bearer ${TEST_SECRET}`);
+
+      expect(response.status).toBe(200);
+      expect(response.headers["content-type"]).toMatch(/text\/csv/);
+      spy.mockRestore();
+    });
+
+    it("should set Content-Disposition to attachment with a .csv filename", async () => {
+      const spy = jest
+        .spyOn(StreamRepository.prototype, "findForExport")
+        .mockResolvedValue({ rows: [], nextCursor: null });
+
+      const response = await request(app)
+        .get("/api/v1/streams/export.csv")
+        .set("Authorization", `Bearer ${TEST_SECRET}`);
+
+      expect(response.headers["content-disposition"]).toMatch(/attachment/);
+      expect(response.headers["content-disposition"]).toMatch(/\.csv/);
+      spy.mockRestore();
+    });
+
+    it("should include a header row in the CSV", async () => {
+      const spy = jest
+        .spyOn(StreamRepository.prototype, "findForExport")
+        .mockResolvedValue({ rows: [], nextCursor: null });
+
+      const response = await request(app)
+        .get("/api/v1/streams/export.csv")
+        .set("Authorization", `Bearer ${TEST_SECRET}`);
+
+      const lines = response.text.split("\r\n").filter(Boolean);
+      expect(lines[0]).toBe(
+        "id,payer,recipient,status,ratePerSecond,startTime,endTime,totalAmount,lastSettledAt,createdAt,updatedAt"
+      );
+      spy.mockRestore();
+    });
+  });
+
+  describe("row count and content", () => {
+    it("should return only the header row when there are no streams", async () => {
+      const spy = jest
+        .spyOn(StreamRepository.prototype, "findForExport")
+        .mockResolvedValue({ rows: [], nextCursor: null });
+
+      const response = await request(app)
+        .get("/api/v1/streams/export.csv")
+        .set("Authorization", `Bearer ${TEST_SECRET}`);
+
+      const lines = response.text.split("\r\n").filter(Boolean);
+      expect(lines).toHaveLength(1); // header only
+      spy.mockRestore();
+    });
+
+    it("should contain one data row per stream in the fixture", async () => {
+      const fixtures = [makeExportStream(), makeExportStream({ id: "bbbbbbbb-2222-2222-2222-000000000002" })];
+      const spy = jest
+        .spyOn(StreamRepository.prototype, "findForExport")
+        .mockResolvedValue({ rows: fixtures as never[], nextCursor: null });
+
+      const response = await request(app)
+        .get("/api/v1/streams/export.csv")
+        .set("Authorization", `Bearer ${TEST_SECRET}`);
+
+      const lines = response.text.split("\r\n").filter(Boolean);
+      // 1 header + 2 data rows
+      expect(lines).toHaveLength(3);
+      expect(lines[1]).toContain(fixtures[0].id);
+      expect(lines[2]).toContain(fixtures[1].id);
+      spy.mockRestore();
+    });
+
+    it("should accumulate rows across multiple cursor pages", async () => {
+      const page1 = [makeExportStream({ id: "id-page1" })];
+      const page2 = [makeExportStream({ id: "id-page2" })];
+      const cursor = { createdAt: new Date(), id: "id-page1" };
+
+      const spy = jest
+        .spyOn(StreamRepository.prototype, "findForExport")
+        .mockResolvedValueOnce({ rows: page1 as never[], nextCursor: cursor })
+        .mockResolvedValueOnce({ rows: page2 as never[], nextCursor: null });
+
+      const response = await request(app)
+        .get("/api/v1/streams/export.csv")
+        .set("Authorization", `Bearer ${TEST_SECRET}`);
+
+      const lines = response.text.split("\r\n").filter(Boolean);
+      expect(lines).toHaveLength(3); // header + 2 rows across 2 pages
+      expect(response.text).toContain("id-page1");
+      expect(response.text).toContain("id-page2");
+      spy.mockRestore();
+    });
+  });
+
+  describe("CSV content correctness", () => {
+    it("should properly escape fields containing commas", async () => {
+      const streamWithComma = makeExportStream({ payer: "address,with,commas" });
+      const spy = jest
+        .spyOn(StreamRepository.prototype, "findForExport")
+        .mockResolvedValue({ rows: [streamWithComma as never], nextCursor: null });
+
+      const response = await request(app)
+        .get("/api/v1/streams/export.csv")
+        .set("Authorization", `Bearer ${TEST_SECRET}`);
+
+      expect(response.text).toContain('"address,with,commas"');
+      spy.mockRestore();
+    });
+
+    it("should properly escape fields containing double quotes", async () => {
+      const streamWithQuote = makeExportStream({ recipient: 'has"quote' });
+      const spy = jest
+        .spyOn(StreamRepository.prototype, "findForExport")
+        .mockResolvedValue({ rows: [streamWithQuote as never], nextCursor: null });
+
+      const response = await request(app)
+        .get("/api/v1/streams/export.csv")
+        .set("Authorization", `Bearer ${TEST_SECRET}`);
+
+      expect(response.text).toContain('"has""quote"');
+      spy.mockRestore();
+    });
+
+    it("should output empty string for null endTime", async () => {
+      const spy = jest
+        .spyOn(StreamRepository.prototype, "findForExport")
+        .mockResolvedValue({ rows: [makeExportStream({ endTime: null }) as never], nextCursor: null });
+
+      const response = await request(app)
+        .get("/api/v1/streams/export.csv")
+        .set("Authorization", `Bearer ${TEST_SECRET}`);
+
+      // The endTime column (6th, 0-indexed) should be empty between commas
+      const dataLine = response.text.split("\r\n")[1];
+      const cols = dataLine.split(",");
+      expect(cols[6]).toBe(""); // endTime column
+      spy.mockRestore();
+    });
+  });
+
+  describe("query filter pass-through", () => {
+    it("should forward payer, recipient, and status filters to the repository", async () => {
+      const spy = jest
+        .spyOn(StreamRepository.prototype, "findForExport")
+        .mockResolvedValue({ rows: [], nextCursor: null });
+
+      await request(app)
+        .get("/api/v1/streams/export.csv?payer=0xA&recipient=0xB&status=paused")
+        .set("Authorization", `Bearer ${TEST_SECRET}`);
+
+      expect(spy).toHaveBeenCalledWith(
+        expect.objectContaining({ payer: "0xA", recipient: "0xB", status: "paused" })
+      );
+      spy.mockRestore();
+    });
+  });
+});

src/repositories/streamRepository.test.ts

@@ -146,4 +146,77 @@ describe("StreamRepository", () => {
       expect(countQuery.where).toHaveBeenCalled();
     });
   });
+
+  describe("findForExport", () => {
+    const makeStream = (overrides: Partial<{ id: string; createdAt: Date }> = {}) => ({
+      id: overrides.id ?? "aaaaaaaa-0000-0000-0000-000000000001",
+      payer: "0xPayer",
+      recipient: "0xRecipient",
+      status: "active" as const,
+      ratePerSecond: "1.0",
+      startTime: new Date("2024-01-01T00:00:00Z"),
+      endTime: null,
+      totalAmount: "3600.0",
+      lastSettledAt: new Date("2024-01-01T00:00:00Z"),
+      createdAt: overrides.createdAt ?? new Date("2024-06-01T00:00:00Z"),
+      updatedAt: new Date("2024-06-01T00:00:00Z"),
+    });
+
+    it("should return all rows with null nextCursor when count <= batchSize", async () => {
+      const mockRows = [makeStream({ id: "a" }), makeStream({ id: "b" })];
+      (db.select as jest.Mock).mockReturnValue(createMockQuery(mockRows));
+
+      const result = await repository.findForExport({ batchSize: 500 });
+
+      expect(result.rows).toHaveLength(2);
+      expect(result.nextCursor).toBeNull();
+    });
+
+    it("should return a page and nextCursor when more rows than batchSize exist", async () => {
+      // Simulate batchSize=2 but DB returns 3 rows (batchSize+1)
+      const t = new Date("2024-06-01T00:00:00Z");
+      const r1 = makeStream({ id: "id-1", createdAt: t });
+      const r2 = makeStream({ id: "id-2", createdAt: t });
+      const r3 = makeStream({ id: "id-3", createdAt: t }); // sentinel
+      (db.select as jest.Mock).mockReturnValue(createMockQuery([r1, r2, r3]));
+
+      const result = await repository.findForExport({ batchSize: 2 });
+
+      expect(result.rows).toHaveLength(2);
+      expect(result.rows[0].id).toBe("id-1");
+      expect(result.rows[1].id).toBe("id-2");
+      expect(result.nextCursor).toEqual({ createdAt: t, id: "id-2" });
+    });
+
+    it("should apply payer/recipient/status filters", async () => {
+      (db.select as jest.Mock).mockReturnValue(createMockQuery([]));
+
+      await repository.findForExport({ payer: "0xA", recipient: "0xB", status: "paused" });
+
+      // select was called — no assertion on internals, just that it resolves without error
+      expect(db.select).toHaveBeenCalled();
+    });
+
+    it("should return empty rows with null nextCursor for empty result set", async () => {
+      (db.select as jest.Mock).mockReturnValue(createMockQuery([]));
+
+      const result = await repository.findForExport({});
+
+      expect(result.rows).toHaveLength(0);
+      expect(result.nextCursor).toBeNull();
+    });
+
+    it("should pass cursor arguments when provided", async () => {
+      const cursorDate = new Date("2024-05-01T00:00:00Z");
+      (db.select as jest.Mock).mockReturnValue(createMockQuery([]));
+
+      const result = await repository.findForExport({
+        cursorCreatedAt: cursorDate,
+        cursorId: "cursor-id",
+      });
+
+      expect(result.rows).toHaveLength(0);
+      expect(result.nextCursor).toBeNull();
+    });
+  });
 });

src/repositories/streamRepository.ts

@@ -1,4 +1,4 @@
-import { eq, and, desc, sql } from "drizzle-orm";
+import { eq, and, or, desc, lt, sql } from "drizzle-orm";
 import { db } from "../db/index";
 import { streams, Stream, NewStream } from "../db/schema";
 
@@ -18,6 +18,23 @@ export interface UpdateStreamParams {
   updatedAt?: Date;
 }
 
+export interface ExportParams {
+  payer?: string;
+  recipient?: string;
+  status?: "active" | "paused" | "cancelled" | "completed";
+  /** Exclusive cursor: resume after this (createdAt, id) pair (both must be set together). */
+  cursorCreatedAt?: Date;
+  cursorId?: string;
+  /** Number of rows per DB fetch (default 500). */
+  batchSize?: number;
+}
+
+export interface ExportBatch {
+  rows: Stream[];
+  /** null when this is the last page. */
+  nextCursor: { createdAt: Date; id: string } | null;
+}
+
 export class StreamRepository {
   async findById(id: string, includeDeleted = false): Promise<(Stream & { accruedEstimate: string }) | null> {
     const conditions = [eq(streams.id, id)];