feat(backend): implement bulk export of tickets to CSV format

backend/main.py

@@ -548,6 +548,38 @@ async def get_tickets(company_id: str | None = None):
     res = query.execute()
     return res.data
 
+
+@app.get("/tickets/export/csv")
+async def export_tickets_csv(company_id: str | None = None):
+    """
+    Stream all tickets as a downloadable CSV for admin audits.
+
+    Data is fetched from Supabase in bounded batches and serialized
+    incrementally, so large exports stream to the client without loading the
+    full dataset into memory. Cell values are sanitized against CSV
+    formula-injection payloads.
+    """
+    from backend.services.csv_exporter import stream_tickets_csv
+
+    if not supabase:
+        raise HTTPException(status_code=500, detail="Database connection not initialized")
+
+    def fetch_batch(offset: int, batch_size: int):
+        query = supabase.table("tickets").select("*").order("created_at", desc=True)
+        if company_id:
+            query = query.eq("company_id", company_id)
+        res = query.range(offset, offset + batch_size - 1).execute()
+        return res.data or []
+
+    return StreamingResponse(
+        stream_tickets_csv(fetch_batch),
+        media_type="text/csv",
+        headers={
+            "Content-Disposition": 'attachment; filename="tickets_export.csv"',
+            "Cache-Control": "no-store",
+        },
+    )
+
 @app.post("/tickets/save")
 async def save_ticket(request_body: TicketSaveRequest):
     """

backend/services/csv_exporter.py

@@ -0,0 +1,80 @@
+"""
+Streaming CSV exporter for ticket audits (issue #3901).
+
+Tickets are serialized one batch at a time and yielded as CSV text, so admin
+audits can download very large exports without buffering the whole dataset in
+memory. Cell values are sanitized against CSV/formula-injection payloads.
+
+Run with:  python -m unittest backend.tests.test_csv_exporter -v
+"""
+
+import csv
+import io
+from typing import Callable, Iterable, Iterator
+
+CSV_COLUMNS = [
+    "ticket_id",
+    "company_id",
+    "company",
+    "category",
+    "subcategory",
+    "priority",
+    "status",
+    "owner_id",
+    "created_at",
+    "sla_breach_at",
+    "resolved_at",
+]
+
+DEFAULT_BATCH_SIZE = 500
+
+# Cells starting with these characters can trigger formula execution when a
+# spreadsheet application opens the exported file (CSV formula injection).
+_FORMULA_PREFIXES = ("=", "+", "-", "@")
+
+
+def _safe_cell(value: object) -> str:
+    """Normalize a cell value and neutralize spreadsheet formula injection."""
+    if value is None:
+        return ""
+    text = str(value)
+    text = text.replace("\r", " ").replace("\n", " ").replace("\x00", "")
+    if text.startswith(_FORMULA_PREFIXES):
+        text = "'" + text
+    return text
+
+
+def _header_row() -> str:
+    buf = io.StringIO()
+    csv.writer(buf).writerow(CSV_COLUMNS)
+    return buf.getvalue()
+
+
+def _ticket_row(ticket: dict) -> str:
+    buf = io.StringIO()
+    csv.writer(buf).writerow([_safe_cell(ticket.get(col)) for col in CSV_COLUMNS])
+    return buf.getvalue()
+
+
+def stream_tickets_csv(
+    fetch_batch: Callable[[int, int], Iterable[dict]],
+    batch_size: int = DEFAULT_BATCH_SIZE,
+) -> Iterator[str]:
+    """
+    Yield CSV text for every ticket returned by ``fetch_batch(offset, batch_size)``.
+
+    ``fetch_batch`` is expected to return an iterable of ticket dicts (or an
+    empty iterable once the dataset is exhausted). Rows are streamed incrementally
+    rather than buffered as a single string.
+    """
+    yield _header_row()
+    offset = 0
+    while True:
+        batch = list(fetch_batch(offset, batch_size))
+        if not batch:
+            break
+        for ticket in batch:
+            yield _ticket_row(ticket)
+        if len(batch) < batch_size:
+            break
+        offset += batch_size

backend/tests/test_csv_exporter.py

@@ -0,0 +1,76 @@
+"""
+Unit tests for the streaming CSV exporter (issue #3901).
+
+Run with:  python -m unittest backend.tests.test_csv_exporter -v
+"""
+
+import csv
+import io
+import unittest
+
+from backend.services.csv_exporter import CSV_COLUMNS, _safe_cell, stream_tickets_csv
+
+
+def _parse(csv_text: str):
+    return list(csv.reader(io.StringIO(csv_text)))
+
+
+class CsvExporterTests(unittest.TestCase):
+    def test_header_row_emitted_first(self):
+        rows = list(stream_tickets_csv(lambda offset, size: []))
+        parsed = _parse("".join(rows))
+        self.assertEqual(parsed, [CSV_COLUMNS])
+
+    def test_single_batch(self):
+        tickets = [
+            {
+                "ticket_id": "T-1",
+                "company_id": "acme",
+                "company": "Acme Inc",
+                "category": "Hardware",
+                "subcategory": "Laptop",
+                "priority": "high",
+                "status": "open",
+                "owner_id": "u-1",
+                "created_at": "2026-01-01T00:00:00Z",
+                "sla_breach_at": "2026-01-02T00:00:00Z",
+                "resolved_at": "",
+            }
+        ]
+        rows = list(stream_tickets_csv(lambda offset, size: tickets if offset == 0 else []))
+        parsed = _parse("".join(rows))
+        self.assertEqual(len(parsed), 2)
+        self.assertEqual(parsed[1][0], "T-1")
+
+    def test_paginates_until_exhausted(self):
+        calls = []
+
+        def fetch_batch(offset, size):
+            calls.append(offset)
+            if offset == 0:
+                return [{"ticket_id": f"T-{i}", "company_id": "acme"} for i in range(3)]
+            if offset == 3:
+                return [{"ticket_id": "T-x", "company_id": "acme"}]
+            return []
+
+        rows = list(stream_tickets_csv(fetch_batch, batch_size=3))
+        parsed = _parse("".join(rows))
+        self.assertEqual(calls, [0, 3, 6])
+        self.assertEqual(len(parsed), 5)
+
+    def test_missing_fields_default_to_empty(self):
+        rows = list(stream_tickets_csv(lambda offset, size: [{"ticket_id": "T-9"}]))
+        parsed = _parse("".join(rows))
+        self.assertEqual(parsed[1][1], "")
+        self.assertEqual(len(parsed[1]), len(CSV_COLUMNS))
+
+    def test_formula_injection_neutralized(self):
+        for prefix in ("=", "+", "-", "@"):
+            self.assertTrue(_safe_cell(f"{prefix}SUM(A1:A9)").startswith("'"))
+
+    def test_newlines_stripped_from_cells(self):
+        self.assertEqual(_safe_cell("line1\nline2\rline3"), "line1 line2 line3")
+
+
+if __name__ == "__main__":
+    unittest.main()