feat: real-time sync across devices via Server-Sent Events

Adds SSE-based push notifications so changes made on one device
appear instantly on all other connected devices for the same user.

- SSE notification hub (src/lib/sse.ts) with per-user client tracking
- Stream endpoint (/api/events/stream) with 30s heartbeat
- Task and list API routes notify connected clients on mutations
- Client auto-connects with 5s reconnect on connection loss
- Existing 2-minute polling sync remains as fallback

v1.53.0

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
mARTin 2026-03-20 07:44:00 +01:00
parent 3c17f6a89d
commit 876bd1047b
6 changed files with 165 additions and 1 deletions

View File

@ -1,6 +1,6 @@
{
"name": "my-weekly-todo-list",
"version": "1.52.1",
"version": "1.53.0",
"description": "A web-based weekly task management application that organizes to-dos and calendar events in a single, intuitive weekly view",
"main": "index.js",
"scripts": {

View File

@ -0,0 +1,50 @@
import { getServerSession } from 'next-auth';
import { authOptions } from '@/lib/auth';
import { addClient, removeClient } from '@/lib/sse';
export const dynamic = 'force-dynamic';
export const runtime = 'nodejs';
export async function GET() {
const session = await getServerSession(authOptions);
const userId = (session?.user as any)?.id;
if (!userId) {
return new Response('Unauthorized', { status: 401 });
}
const encoder = new TextEncoder();
let sseClient: ReturnType<typeof addClient> | null = null;
let heartbeatInterval: NodeJS.Timeout | null = null;
const stream = new ReadableStream({
start(controller) {
sseClient = addClient(userId, controller);
// Send initial connected event
controller.enqueue(encoder.encode(`event: connected\ndata: ${JSON.stringify({ clientId: sseClient.id })}\n\n`));
// Heartbeat every 30s to keep connection alive
heartbeatInterval = setInterval(() => {
try {
controller.enqueue(encoder.encode(`: heartbeat\n\n`));
} catch {
if (heartbeatInterval) clearInterval(heartbeatInterval);
if (sseClient) removeClient(sseClient);
}
}, 30000);
},
cancel() {
if (heartbeatInterval) clearInterval(heartbeatInterval);
if (sseClient) removeClient(sseClient);
},
});
return new Response(stream, {
headers: {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache, no-transform',
'Connection': 'keep-alive',
'X-Accel-Buffering': 'no',
},
});
}

View File

@ -2,6 +2,7 @@ import { NextRequest, NextResponse } from 'next/server';
import { getServerSession } from 'next-auth';
import { authOptions } from "@/lib/auth";
import { PrismaClient } from '@prisma/client';
import { notifyUser } from '@/lib/sse';
const prisma = new PrismaClient();
@ -76,6 +77,7 @@ export async function POST(request: NextRequest) {
include: { tasks: true } // Return with empty tasks array for frontend consistency
});
notifyUser(userId, "list-changed", { action: "created" });
return NextResponse.json({ list });
} catch (error) {
console.error('Error creating someday list:', error);
@ -143,6 +145,7 @@ export async function DELETE(request: NextRequest) {
where: { id }
});
notifyUser((session.user as any).id, "list-changed", { action: "deleted" });
return NextResponse.json({ success: true });
} catch (error) {
console.error('Error deleting someday list:', error);
@ -182,6 +185,7 @@ export async function PATCH(request: NextRequest) {
});
await Promise.all(updates);
notifyUser((session.user as any).id, "list-changed", { action: "reordered" });
return NextResponse.json({ success: true });
}
@ -216,6 +220,7 @@ export async function PATCH(request: NextRequest) {
data
});
notifyUser((session.user as any).id, "list-changed", { action: "updated" });
return NextResponse.json({ list });
} catch (error) {
console.error('Error updating someday list:', error);

View File

@ -2,6 +2,7 @@ import { NextRequest, NextResponse } from 'next/server';
import { getServerSession } from 'next-auth';
import { authOptions } from "@/lib/auth";
import { PrismaClient, Task } from '@prisma/client';
import { notifyUser } from '@/lib/sse';
const prisma = new PrismaClient();
@ -336,6 +337,7 @@ export async function POST(request: NextRequest) {
},
});
notifyUser(userId, "tasks-changed", { action: "created", taskId: task.id });
return NextResponse.json({ task });
} catch (error) {
console.error('Error creating task:', error);
@ -426,6 +428,7 @@ export async function PATCH(request: NextRequest) {
}
});
notifyUser(userId, "tasks-changed", { action: "created", taskId: newTask.id });
return NextResponse.json({ task: newTask });
}
@ -475,6 +478,7 @@ export async function PATCH(request: NextRequest) {
// which would exist ALONGSIDE the one we projected.
// By removing it, we rely purely on the projection system (or manual materialization via interaction).
notifyUser(userId, "tasks-changed", { action: "updated", taskId: task.id });
return NextResponse.json({ task });
} catch (error) {
console.error('Error updating task:', error);
@ -549,6 +553,7 @@ export async function DELETE(request: NextRequest) {
completed: true // Marked done so it doesn't appear pending
}
});
notifyUser(userId, "tasks-changed", { action: "deleted" });
return NextResponse.json({ message: 'Virtual task dismissed' });
}
@ -571,6 +576,7 @@ export async function DELETE(request: NextRequest) {
await prisma.task.delete({
where: { id },
});
notifyUser(userId, "tasks-changed", { action: "deleted", taskId: id });
return NextResponse.json({ message: 'Task permanently deleted' });
}
@ -580,6 +586,7 @@ export async function DELETE(request: NextRequest) {
data: { deletedAt: new Date() },
});
notifyUser(userId, "tasks-changed", { action: "deleted", taskId: id });
return NextResponse.json({ message: 'Task moved to trash' });
} catch (error) {
console.error('Error deleting task:', error);

View File

@ -2731,6 +2731,44 @@ export default function WeeklyView() {
return () => clearInterval(interval);
}, [session]);
// SSE real-time sync: listen for server-pushed task/list changes
useEffect(() => {
if (!session) return;
let eventSource: EventSource | null = null;
let reconnectTimeout: NodeJS.Timeout | null = null;
const connect = () => {
eventSource = new EventSource("/api/events/stream");
eventSource.addEventListener("connected", () => {
console.log("[SSE] Connected for real-time sync");
});
eventSource.addEventListener("tasks-changed", () => {
console.log("[SSE] Tasks changed remotely, refetching...");
fetchTasks();
});
eventSource.addEventListener("list-changed", () => {
console.log("[SSE] Lists changed remotely, refetching...");
fetchTasks();
});
eventSource.onerror = () => {
console.log("[SSE] Connection lost, reconnecting in 5s...");
eventSource?.close();
reconnectTimeout = setTimeout(connect, 5000);
};
};
connect();
return () => {
eventSource?.close();
if (reconnectTimeout) clearTimeout(reconnectTimeout);
};
}, [session]);
// Periodic background calendar cache refresh (every 2 minutes)
useEffect(() => {
if (!session) return;

64
src/lib/sse.ts Normal file
View File

@ -0,0 +1,64 @@
// Server-Sent Events notification hub
// In-memory map of userId → Set of writable stream controllers
type SSEClient = {
id: string;
controller: ReadableStreamDefaultController;
userId: string;
};
const clients = new Map<string, Set<SSEClient>>();
export function addClient(userId: string, controller: ReadableStreamDefaultController): SSEClient {
const client: SSEClient = {
id: Math.random().toString(36).slice(2),
controller,
userId,
};
if (!clients.has(userId)) {
clients.set(userId, new Set());
}
clients.get(userId)!.add(client);
return client;
}
export function removeClient(client: SSEClient) {
const userClients = clients.get(client.userId);
if (userClients) {
userClients.delete(client);
if (userClients.size === 0) {
clients.delete(client.userId);
}
}
}
export type SSEEventType = "task-created" | "task-updated" | "task-deleted" | "tasks-changed" | "list-changed" | "profile-changed";
export function notifyUser(userId: string, event: SSEEventType, data?: any) {
const userClients = clients.get(userId);
if (!userClients || userClients.size === 0) return;
const payload = `event: ${event}\ndata: ${JSON.stringify(data || {})}\n\n`;
const encoder = new TextEncoder();
const encoded = encoder.encode(payload);
for (const client of userClients) {
try {
client.controller.enqueue(encoded);
} catch {
// Client disconnected, clean up
removeClient(client);
}
}
}
export function getClientCount(userId?: string): number {
if (userId) {
return clients.get(userId)?.size || 0;
}
let total = 0;
for (const set of clients.values()) {
total += set.size;
}
return total;
}