feat(realtime): publish user-scoped update events
This commit is contained in:
parent
65dcbddeea
commit
485e26bbda
8 changed files with 93 additions and 4 deletions
|
|
@ -1,4 +1,6 @@
|
||||||
import { NextRequest } from "next/server";
|
import { NextRequest } from "next/server";
|
||||||
|
import config from "@payload-config";
|
||||||
|
import { getPayload } from "payload";
|
||||||
import { type SSEEventClient, subscribe, unsubscribe } from "@/lib/realtime/bus";
|
import { type SSEEventClient, subscribe, unsubscribe } from "@/lib/realtime/bus";
|
||||||
import { getPresence } from "@/lib/realtime/presence";
|
import { getPresence } from "@/lib/realtime/presence";
|
||||||
|
|
||||||
|
|
@ -6,6 +8,8 @@ export const dynamic = "force-dynamic";
|
||||||
export const runtime = "nodejs";
|
export const runtime = "nodejs";
|
||||||
|
|
||||||
export async function GET(req: NextRequest) {
|
export async function GET(req: NextRequest) {
|
||||||
|
const payload = await getPayload({ config: await config });
|
||||||
|
const { user } = await payload.auth({ headers: req.headers, canSetHeaders: false });
|
||||||
const encoder = new TextEncoder();
|
const encoder = new TextEncoder();
|
||||||
const channel = req.nextUrl.searchParams.get("channel");
|
const channel = req.nextUrl.searchParams.get("channel");
|
||||||
|
|
||||||
|
|
@ -23,6 +27,7 @@ export async function GET(req: NextRequest) {
|
||||||
start(controller) {
|
start(controller) {
|
||||||
client = {
|
client = {
|
||||||
send: (data) => controller.enqueue(encoder.encode(data)),
|
send: (data) => controller.enqueue(encoder.encode(data)),
|
||||||
|
userId: user ? Number(user.id) : undefined,
|
||||||
events: channel ? new Set(["presence"]) : undefined,
|
events: channel ? new Set(["presence"]) : undefined,
|
||||||
};
|
};
|
||||||
subscribe(client);
|
subscribe(client);
|
||||||
|
|
|
||||||
|
|
@ -1,5 +1,6 @@
|
||||||
import { CollectionConfig } from "payload";
|
import { CollectionConfig } from "payload";
|
||||||
import { Award } from "@/payload-types";
|
import { Award } from "@/payload-types";
|
||||||
|
import { broadcastToUser } from "@/lib/realtime/bus";
|
||||||
import { requirePermission, hasPermission } from "@/utils/access-control/hasPermission";
|
import { requirePermission, hasPermission } from "@/utils/access-control/hasPermission";
|
||||||
|
|
||||||
export const Profiles: CollectionConfig = {
|
export const Profiles: CollectionConfig = {
|
||||||
|
|
@ -23,6 +24,16 @@ export const Profiles: CollectionConfig = {
|
||||||
return await hasPermission(req.payload, req.user, "profiles:delete");
|
return await hasPermission(req.payload, req.user, "profiles:delete");
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
|
hooks: {
|
||||||
|
afterChange: [
|
||||||
|
async ({ doc }) => {
|
||||||
|
const userId = typeof doc.user === "object" ? doc.user?.id : doc.user;
|
||||||
|
if (typeof userId === "number") {
|
||||||
|
broadcastToUser("profile-update", userId, { profileId: doc.id });
|
||||||
|
}
|
||||||
|
},
|
||||||
|
],
|
||||||
|
},
|
||||||
fields: [
|
fields: [
|
||||||
{
|
{
|
||||||
name: "user",
|
name: "user",
|
||||||
|
|
|
||||||
|
|
@ -15,6 +15,7 @@ import {
|
||||||
} from "@/components/ui/dropdown-menu";
|
} from "@/components/ui/dropdown-menu";
|
||||||
import { cn } from "@/lib/utils";
|
import { cn } from "@/lib/utils";
|
||||||
import { notificationLabel } from "@/lib/notifications";
|
import { notificationLabel } from "@/lib/notifications";
|
||||||
|
import { NOTIFICATION_EVENT } from "@/hooks/useGameTick";
|
||||||
import {
|
import {
|
||||||
markAllNotificationsRead,
|
markAllNotificationsRead,
|
||||||
markNotificationsRead,
|
markNotificationsRead,
|
||||||
|
|
@ -41,7 +42,7 @@ function timeAgo(iso: string): string {
|
||||||
return `${days}d ago`;
|
return `${days}d ago`;
|
||||||
}
|
}
|
||||||
|
|
||||||
const POLL_MS = 30_000;
|
const POLL_MS = 60_000;
|
||||||
|
|
||||||
export function NotificationsBell() {
|
export function NotificationsBell() {
|
||||||
const router = useRouter();
|
const router = useRouter();
|
||||||
|
|
@ -74,8 +75,11 @@ export function NotificationsBell() {
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
void fetchNotifications();
|
void fetchNotifications();
|
||||||
intervalRef.current = setInterval(() => void fetchNotifications(), POLL_MS);
|
intervalRef.current = setInterval(() => void fetchNotifications(), POLL_MS);
|
||||||
|
const onNotification = () => void fetchNotifications();
|
||||||
|
window.addEventListener(NOTIFICATION_EVENT, onNotification);
|
||||||
return () => {
|
return () => {
|
||||||
if (intervalRef.current) clearInterval(intervalRef.current);
|
if (intervalRef.current) clearInterval(intervalRef.current);
|
||||||
|
window.removeEventListener(NOTIFICATION_EVENT, onNotification);
|
||||||
};
|
};
|
||||||
}, [fetchNotifications]);
|
}, [fetchNotifications]);
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -2,7 +2,14 @@
|
||||||
|
|
||||||
import { useEffect } from "react";
|
import { useEffect } from "react";
|
||||||
import { useRouter } from "next/navigation";
|
import { useRouter } from "next/navigation";
|
||||||
import { GAME_TICK_EVENT, type GameTickDetail } from "@/hooks/useGameTick";
|
import {
|
||||||
|
GAME_EVENT_EVENT,
|
||||||
|
GAME_TICK_EVENT,
|
||||||
|
NOTIFICATION_EVENT,
|
||||||
|
PROFILE_UPDATE_EVENT,
|
||||||
|
type GameTickDetail,
|
||||||
|
type ProfileUpdateDetail,
|
||||||
|
} from "@/hooks/useGameTick";
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Opens a Server-Sent Events connection to the game tick stream.
|
* Opens a Server-Sent Events connection to the game tick stream.
|
||||||
|
|
@ -15,6 +22,14 @@ export function GameTickRealtime() {
|
||||||
|
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
const source = new EventSource("/api/realtime");
|
const source = new EventSource("/api/realtime");
|
||||||
|
let refreshTimer: ReturnType<typeof setTimeout> | null = null;
|
||||||
|
const refresh = (delay = 250) => {
|
||||||
|
if (refreshTimer) return;
|
||||||
|
refreshTimer = setTimeout(() => {
|
||||||
|
refreshTimer = null;
|
||||||
|
router.refresh();
|
||||||
|
}, delay);
|
||||||
|
};
|
||||||
source.addEventListener("game-tick", (event) => {
|
source.addEventListener("game-tick", (event) => {
|
||||||
let detail: GameTickDetail = {};
|
let detail: GameTickDetail = {};
|
||||||
try {
|
try {
|
||||||
|
|
@ -22,12 +37,32 @@ export function GameTickRealtime() {
|
||||||
} catch {
|
} catch {
|
||||||
// ignore malformed payload
|
// ignore malformed payload
|
||||||
}
|
}
|
||||||
router.refresh();
|
refresh();
|
||||||
window.dispatchEvent(
|
window.dispatchEvent(
|
||||||
new CustomEvent(GAME_TICK_EVENT, { detail }),
|
new CustomEvent(GAME_TICK_EVENT, { detail }),
|
||||||
);
|
);
|
||||||
});
|
});
|
||||||
return () => source.close();
|
source.addEventListener("profile-update", (event) => {
|
||||||
|
let detail: ProfileUpdateDetail = {};
|
||||||
|
try {
|
||||||
|
detail = JSON.parse(event.data ?? "{}");
|
||||||
|
} catch {
|
||||||
|
// ignore malformed payload
|
||||||
|
}
|
||||||
|
refresh(detail.xpGained ? 1700 : 250);
|
||||||
|
window.dispatchEvent(new CustomEvent(PROFILE_UPDATE_EVENT, { detail }));
|
||||||
|
});
|
||||||
|
source.addEventListener("game-event", () => {
|
||||||
|
refresh();
|
||||||
|
window.dispatchEvent(new CustomEvent(GAME_EVENT_EVENT));
|
||||||
|
});
|
||||||
|
source.addEventListener("notification", (event) => {
|
||||||
|
window.dispatchEvent(new CustomEvent(NOTIFICATION_EVENT, { detail: event.data }));
|
||||||
|
});
|
||||||
|
return () => {
|
||||||
|
if (refreshTimer) clearTimeout(refreshTimer);
|
||||||
|
source.close();
|
||||||
|
};
|
||||||
}, [router]);
|
}, [router]);
|
||||||
|
|
||||||
return null;
|
return null;
|
||||||
|
|
|
||||||
|
|
@ -3,11 +3,19 @@
|
||||||
import { useEffect, useRef, useState } from "react";
|
import { useEffect, useRef, useState } from "react";
|
||||||
|
|
||||||
export const GAME_TICK_EVENT = "ptf:game-tick";
|
export const GAME_TICK_EVENT = "ptf:game-tick";
|
||||||
|
export const PROFILE_UPDATE_EVENT = "ptf:profile-update";
|
||||||
|
export const GAME_EVENT_EVENT = "ptf:game-event";
|
||||||
|
export const NOTIFICATION_EVENT = "ptf:notification";
|
||||||
|
|
||||||
export interface GameTickDetail {
|
export interface GameTickDetail {
|
||||||
processedAt?: string;
|
processedAt?: string;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export interface ProfileUpdateDetail {
|
||||||
|
reason?: string;
|
||||||
|
xpGained?: number;
|
||||||
|
}
|
||||||
|
|
||||||
export function useGameTick(onTick?: () => void): number {
|
export function useGameTick(onTick?: () => void): number {
|
||||||
const [count, setCount] = useState(0);
|
const [count, setCount] = useState(0);
|
||||||
const callbackRef = useRef(onTick);
|
const callbackRef = useRef(onTick);
|
||||||
|
|
|
||||||
|
|
@ -1,3 +1,5 @@
|
||||||
|
import { broadcastToUser } from "@/lib/realtime/bus";
|
||||||
|
|
||||||
type PayloadType = Awaited<ReturnType<typeof import("payload").getPayload>>;
|
type PayloadType = Awaited<ReturnType<typeof import("payload").getPayload>>;
|
||||||
|
|
||||||
export interface NotifyInput {
|
export interface NotifyInput {
|
||||||
|
|
@ -40,6 +42,7 @@ export async function notifyUser(payload: PayloadType, input: NotifyInput): Prom
|
||||||
overrideAccess: true,
|
overrideAccess: true,
|
||||||
depth: 0,
|
depth: 0,
|
||||||
});
|
});
|
||||||
|
broadcastToUser("notification", input.userId, { type: input.type ?? null });
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
console.error("[Notifications] Failed to create notification:", err);
|
console.error("[Notifications] Failed to create notification:", err);
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -1,6 +1,7 @@
|
||||||
export interface SSEEventClient {
|
export interface SSEEventClient {
|
||||||
send: (data: string) => void;
|
send: (data: string) => void;
|
||||||
events?: Set<string>;
|
events?: Set<string>;
|
||||||
|
userId?: number;
|
||||||
}
|
}
|
||||||
|
|
||||||
const clients = new Set<SSEEventClient>();
|
const clients = new Set<SSEEventClient>();
|
||||||
|
|
@ -30,3 +31,18 @@ export function broadcast(event: string, data: unknown): number {
|
||||||
}
|
}
|
||||||
return clients.size;
|
return clients.size;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export function broadcastToUser(event: string, userId: number, data: unknown): number {
|
||||||
|
const payload = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
|
||||||
|
const targetUserId = Number(userId);
|
||||||
|
for (const client of clients) {
|
||||||
|
if (client.userId !== targetUserId) continue;
|
||||||
|
if (client.events && !client.events.has(event)) continue;
|
||||||
|
try {
|
||||||
|
client.send(payload);
|
||||||
|
} catch {
|
||||||
|
clients.delete(client);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return clients.size;
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,5 @@
|
||||||
import type { EventType } from "./eventTypes";
|
import type { EventType } from "./eventTypes";
|
||||||
|
import { broadcast } from "@/lib/realtime/bus";
|
||||||
|
|
||||||
interface GameEventInput {
|
interface GameEventInput {
|
||||||
type: EventType;
|
type: EventType;
|
||||||
|
|
@ -52,6 +53,12 @@ export async function emitGameEvent(
|
||||||
},
|
},
|
||||||
depth: 0,
|
depth: 0,
|
||||||
});
|
});
|
||||||
|
broadcast("game-event", {
|
||||||
|
type: event.type,
|
||||||
|
actor: event.actor ?? null,
|
||||||
|
targetCollection: event.targetCollection ?? null,
|
||||||
|
targetId: event.targetId ?? null,
|
||||||
|
});
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
console.error("[EventLog] Failed to emit event:", err);
|
console.error("[EventLog] Failed to emit event:", err);
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue