From 0c15fbfbd2ac8eccc8fdcff8eaa4eb4dcf592c81 Mon Sep 17 00:00:00 2001 From: Amirhossein Khalili Date: Sun, 21 Jun 2026 02:35:13 +0330 Subject: [PATCH] feat(events): add cursor infinite event feed --- backend/jobs/tests/test_api.py | 43 +++++++++++++ backend/jobs/views.py | 22 ++++--- frontend/src/api.ts | 19 ++++-- frontend/src/pages/DashboardPage.tsx | 2 +- frontend/src/pages/EventsPage.tsx | 95 ++++++++++++++++++++++------ frontend/src/styles.css | 19 ++++++ frontend/src/types.ts | 6 ++ 7 files changed, 170 insertions(+), 36 deletions(-) create mode 100644 backend/jobs/tests/test_api.py diff --git a/backend/jobs/tests/test_api.py b/backend/jobs/tests/test_api.py new file mode 100644 index 0000000..f99d05a --- /dev/null +++ b/backend/jobs/tests/test_api.py @@ -0,0 +1,43 @@ +from urllib.parse import urlparse + +import pytest +from rest_framework.test import APIClient + +from jobs.services import create_job + + +@pytest.mark.django_db +def test_global_events_use_cursor_pagination(): + create_job(job_type="demo.success") + create_job(job_type="demo.fail") + create_job(job_type="demo.slow") + + response = APIClient().get("/api/job-events/?limit=2") + + assert response.status_code == 200 + body = response.json() + assert body["next"] is not None + assert body["previous"] is None + assert len(body["results"]) == 2 + assert body["results"][0]["id"] > body["results"][1]["id"] + + +@pytest.mark.django_db +def test_global_events_next_cursor_returns_older_events(): + create_job(job_type="demo.success") + create_job(job_type="demo.fail") + create_job(job_type="demo.slow") + client = APIClient() + first_response = client.get("/api/job-events/?limit=2") + first_body = first_response.json() + parsed_next = urlparse(first_body["next"]) + next_path = f"{parsed_next.path}?{parsed_next.query}" + + next_response = client.get(next_path) + + assert next_response.status_code == 200 + body = next_response.json() + assert body["next"] is None + assert body["previous"] is not None + assert len(body["results"]) == 1 + assert body["results"][0]["id"] < first_body["results"][1]["id"] diff --git a/backend/jobs/views.py b/backend/jobs/views.py index 5acaff9..ff28863 100644 --- a/backend/jobs/views.py +++ b/backend/jobs/views.py @@ -1,6 +1,7 @@ from django.db import connection from rest_framework import generics, status from rest_framework.exceptions import NotFound, ValidationError +from rest_framework.pagination import CursorPagination from rest_framework.response import Response from rest_framework.views import APIView @@ -9,6 +10,13 @@ from jobs.serializers import JobCreateSerializer, JobEventSerializer, JobSeriali from jobs.services import create_job, job_stats, retry_failed_job +class JobEventCursorPagination(CursorPagination): + page_size = 50 + page_size_query_param = "limit" + max_page_size = 200 + ordering = "-id" + + class HealthAPIView(APIView): def get(self, request): try: @@ -63,21 +71,17 @@ class JobEventsAPIView(APIView): class GlobalJobEventsAPIView(APIView): def get(self, request): - queryset = JobEvent.objects.order_by("id") - after_id = request.query_params.get("after_id") + queryset = JobEvent.objects.order_by("-id") event_type = request.query_params.get("type") job_id = request.query_params.get("job_id") - if after_id: - queryset = queryset.filter(id__gt=after_id) if event_type: queryset = queryset.filter(type=event_type) if job_id: queryset = queryset.filter(job_id=job_id) - try: - limit = min(int(request.query_params.get("limit", "100")), 500) - except ValueError as exc: - raise ValidationError({"limit": "Must be an integer."}) from exc - return Response(JobEventSerializer(queryset[:limit], many=True).data) + + paginator = JobEventCursorPagination() + page = paginator.paginate_queryset(queryset, request, view=self) + return paginator.get_paginated_response(JobEventSerializer(page, many=True).data) class JobStatsAPIView(APIView): diff --git a/frontend/src/api.ts b/frontend/src/api.ts index 8d5b015..96a2c01 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -1,9 +1,10 @@ -import type { Health, Job, JobEvent, JobStats } from "./types"; +import type { CursorPage, Health, Job, JobEvent, JobStats } from "./types"; const API_BASE_URL = import.meta.env.VITE_API_BASE_URL ?? "http://localhost:8000/api"; async function request(path: string, init?: RequestInit): Promise { - const response = await fetch(`${API_BASE_URL}${path}`, { + const url = /^https?:\/\//i.test(path) ? path : `${API_BASE_URL}${path}`; + const response = await fetch(url, { headers: { "Content-Type": "application/json", ...(init?.headers ?? {}) }, ...init }); @@ -23,6 +24,13 @@ export type CreateJobBody = { idempotency_key?: string | null; }; +function normalizeCursorPage(value: CursorPage | T[]): CursorPage { + if (Array.isArray(value)) { + return { next: null, previous: null, results: value }; + } + return value; +} + export const api = { health: () => request("/health/"), listJobs: () => request("/jobs/"), @@ -31,12 +39,13 @@ export const api = { retryJob: (jobId: string) => request(`/jobs/${jobId}/retry/`, { method: "POST" }), getStats: () => request("/jobs/stats/"), listJobEvents: (jobId: string) => request(`/jobs/${jobId}/events/`), - listEvents: (filters: { after_id?: number; limit?: number; job_id?: string; type?: string } = {}) => { + listEvents: (filters: { limit?: number; job_id?: string; type?: string } = {}) => { const params = new URLSearchParams(); Object.entries(filters).forEach(([key, value]) => { if (value !== undefined && value !== null && value !== "") params.set(key, String(value)); }); const query = params.toString(); - return request(`/job-events/${query ? `?${query}` : ""}`); - } + return request | JobEvent[]>(`/job-events/${query ? `?${query}` : ""}`).then(normalizeCursorPage); + }, + listEventsPage: (url: string) => request | JobEvent[]>(url).then(normalizeCursorPage) }; diff --git a/frontend/src/pages/DashboardPage.tsx b/frontend/src/pages/DashboardPage.tsx index 2647b0f..9e5ff1d 100644 --- a/frontend/src/pages/DashboardPage.tsx +++ b/frontend/src/pages/DashboardPage.tsx @@ -66,7 +66,7 @@ export function DashboardPage() { ]); setJobs(jobsData); setStats({ ...emptyStats, ...statsData, by_status: { ...emptyStats.by_status, ...statsData.by_status } }); - setEvents(eventsData); + setEvents(eventsData.results); setHealth(healthData); }, []); diff --git a/frontend/src/pages/EventsPage.tsx b/frontend/src/pages/EventsPage.tsx index c4b8a70..17345a1 100644 --- a/frontend/src/pages/EventsPage.tsx +++ b/frontend/src/pages/EventsPage.tsx @@ -1,4 +1,4 @@ -import { useCallback, useEffect, useMemo, useState } from "react"; +import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { Link } from "react-router-dom"; import { toast } from "sonner"; @@ -8,36 +8,71 @@ import { DateTime } from "../components/DateTime"; import { EmptyState } from "../components/EmptyState"; import type { JobEvent } from "../types"; +const EVENT_PAGE_SIZE = 50; + +function mergeEvents(current: JobEvent[], incoming: JobEvent[]) { + const seen = new Set(current.map((event) => event.id)); + return [...current, ...incoming.filter((event) => !seen.has(event.id))].sort((a, b) => b.id - a.id); +} + export function EventsPage() { const [events, setEvents] = useState([]); + const [nextPageUrl, setNextPageUrl] = useState(null); + const [loading, setLoading] = useState(true); + const [loadingMore, setLoadingMore] = useState(false); + const [lastRefreshAt, setLastRefreshAt] = useState(null); + const sentinelRef = useRef(null); - const refreshInitial = useCallback(async () => { - setEvents(await api.listEvents({ limit: 100 })); + const refreshFirstPage = useCallback(async (replace = false) => { + if (replace) setLoading(true); + try { + const page = await api.listEvents({ limit: EVENT_PAGE_SIZE }); + setNextPageUrl((current) => (replace ? page.next : current)); + setEvents((current) => (replace ? page.results : mergeEvents(current, page.results))); + setLastRefreshAt(new Date().toISOString()); + } finally { + if (replace) setLoading(false); + } }, []); useEffect(() => { - void refreshInitial().catch((caught) => toast.error(caught instanceof Error ? caught.message : String(caught))); - }, [refreshInitial]); + void refreshFirstPage(true).catch((caught) => toast.error(caught instanceof Error ? caught.message : String(caught))); + }, [refreshFirstPage]); useEffect(() => { const id = window.setInterval(() => { - setEvents((current) => { - const afterId = current.at(-1)?.id; - void api - .listEvents({ after_id: afterId, limit: 100 }) - .then((incoming) => { - if (!incoming.length) return; - setEvents((latest) => { - const seen = new Set(latest.map((event) => event.id)); - return [...latest, ...incoming.filter((event) => !seen.has(event.id))].slice(-200); - }); - }) - .catch(() => undefined); - return current; - }); + void refreshFirstPage(false).catch(() => undefined); }, 1000); return () => window.clearInterval(id); - }, []); + }, [refreshFirstPage]); + + const loadMore = useCallback(async () => { + if (!nextPageUrl || loadingMore) return; + setLoadingMore(true); + try { + const page = await api.listEventsPage(nextPageUrl); + setNextPageUrl(page.next); + setEvents((current) => mergeEvents(current, page.results)); + } catch (caught) { + toast.error(caught instanceof Error ? caught.message : String(caught)); + } finally { + setLoadingMore(false); + } + }, [loadingMore, nextPageUrl]); + + useEffect(() => { + const node = sentinelRef.current; + if (!node || !nextPageUrl) return undefined; + + const observer = new IntersectionObserver( + (entries) => { + if (entries.some((entry) => entry.isIntersecting)) void loadMore(); + }, + { rootMargin: "280px 0px" } + ); + observer.observe(node); + return () => observer.disconnect(); + }, [loadMore, nextPageUrl]); const newestFirst = useMemo(() => [...events].sort((a, b) => b.id - a.id), [events]); @@ -48,9 +83,17 @@ export function EventsPage() { Audit log

Events

+
+ Live cursor feed + {newestFirst.length} loaded +
+
+

Global Event Feed

+ Last refresh {lastRefreshAt ? new Date(lastRefreshAt).toLocaleTimeString() : "-"} +
@@ -82,7 +125,17 @@ export function EventsPage() { ))}
- {!newestFirst.length && } + {loading && !newestFirst.length && } + {!loading && !newestFirst.length && } +
+
+ {loadingMore && Loading older events...} + {!loadingMore && nextPageUrl && ( + + )} + {!loadingMore && !nextPageUrl && newestFirst.length > 0 && End of event history}
diff --git a/frontend/src/styles.css b/frontend/src/styles.css index d746ed5..f8687d1 100644 --- a/frontend/src/styles.css +++ b/frontend/src/styles.css @@ -968,6 +968,25 @@ label { justify-content: flex-end; } +.event-feed-meta { + align-items: center; + display: flex; + flex-wrap: wrap; + gap: 8px; + justify-content: flex-end; +} + +.infinite-sentinel { + align-items: center; + color: var(--muted); + display: flex; + font-size: 13px; + font-weight: 900; + justify-content: center; + min-height: 64px; + padding-top: 12px; +} + .filter-grid { grid-template-columns: repeat(4, minmax(0, 1fr)); } diff --git a/frontend/src/types.ts b/frontend/src/types.ts index 62763d9..60b4b13 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -30,6 +30,12 @@ export type JobEvent = { created_at: string; }; +export type CursorPage = { + next: string | null; + previous: string | null; + results: T[]; +}; + export type JobStats = { total: number; by_status: Record;