diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index bb85ac9..462690d 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -41,8 +41,8 @@ jobs: - name: Migrate database run: | sleep 2 - npx prisma generate --sql --schema=libs/prisma/prisma/schema.prisma npx nx run-many -t prisma-deploy --parallel=4 + npx prisma generate --sql --schema=libs/prisma/prisma/schema.prisma env: DATABASE_URL: ${{ secrets.DATABASE_URL }} FLY_API_TOKEN: ${{ secrets.FLY_API_TOKEN_POSTGRES2 }} diff --git a/apps/frontend/app/api/clan/[clanName]/activity/route.ts b/apps/frontend/app/api/clan/[clanName]/activity/route.ts new file mode 100644 index 0000000..01e7f09 --- /dev/null +++ b/apps/frontend/app/api/clan/[clanName]/activity/route.ts @@ -0,0 +1,33 @@ +import { NextRequest, NextResponse } from 'next/server'; +import { z } from 'zod'; +import { decodeString } from '../../../../../utils/encoding'; +import prisma from '../../../../../utils/prisma'; +import { getClanActivity, rangeParamsFromSearch } from '../../../../../utils/activity'; + +const paramsSchema = z.object({ + clanName: z.string().transform(decodeString), +}); + +export async function GET( + request: NextRequest, + { params }: { params: { clanName: string } } +) { + const parsedParams = paramsSchema.safeParse(params); + + if (!parsedParams.success) { + return NextResponse.json({ error: 'Bad request' }, { status: 400 }); + } + + const clan = await prisma.clan.findUnique({ + where: { name: parsedParams.data.clanName }, + select: { id: true }, + }); + + if (clan === null) { + return NextResponse.json({ error: 'Not found' }, { status: 404 }); + } + + return NextResponse.json( + await getClanActivity(clan.id, rangeParamsFromSearch(request.nextUrl.searchParams)) + ); +} diff --git a/apps/frontend/app/api/daily-players/route.ts b/apps/frontend/app/api/daily-players/route.ts new file mode 100644 index 0000000..0235378 --- /dev/null +++ b/apps/frontend/app/api/daily-players/route.ts @@ -0,0 +1,8 @@ +import { NextRequest, NextResponse } from 'next/server'; +import { getDailyPlayers } from '../../../utils/dailyPlayers'; + +export async function GET(request: NextRequest) { + return NextResponse.json( + await getDailyPlayers(request.nextUrl.searchParams.get('range') ?? '90d') + ); +} diff --git a/apps/frontend/app/api/gametype/[gameTypeName]/activity/route.ts b/apps/frontend/app/api/gametype/[gameTypeName]/activity/route.ts new file mode 100644 index 0000000..4632867 --- /dev/null +++ b/apps/frontend/app/api/gametype/[gameTypeName]/activity/route.ts @@ -0,0 +1,33 @@ +import { NextRequest, NextResponse } from 'next/server'; +import { z } from 'zod'; +import { decodeString } from '../../../../../utils/encoding'; +import prisma from '../../../../../utils/prisma'; +import { getGameTypeActivity, rangeParamsFromSearch } from '../../../../../utils/activity'; + +const paramsSchema = z.object({ + gameTypeName: z.string().transform(decodeString), +}); + +export async function GET( + request: NextRequest, + { params }: { params: { gameTypeName: string } } +) { + const parsedParams = paramsSchema.safeParse(params); + + if (!parsedParams.success) { + return NextResponse.json({ error: 'Bad request' }, { status: 400 }); + } + + const gameType = await prisma.gameType.findUnique({ + where: { name: parsedParams.data.gameTypeName }, + select: { id: true }, + }); + + if (gameType === null) { + return NextResponse.json({ error: 'Not found' }, { status: 404 }); + } + + return NextResponse.json( + await getGameTypeActivity(gameType.id, rangeParamsFromSearch(request.nextUrl.searchParams)) + ); +} diff --git a/apps/frontend/app/api/gametype/[gameTypeName]/map/[mapName]/activity/route.ts b/apps/frontend/app/api/gametype/[gameTypeName]/map/[mapName]/activity/route.ts new file mode 100644 index 0000000..ac4c747 --- /dev/null +++ b/apps/frontend/app/api/gametype/[gameTypeName]/map/[mapName]/activity/route.ts @@ -0,0 +1,39 @@ +import { NextRequest, NextResponse } from 'next/server'; +import { z } from 'zod'; +import { decodeString } from '../../../../../../../utils/encoding'; +import prisma from '../../../../../../../utils/prisma'; +import { getMapActivity, rangeParamsFromSearch } from '../../../../../../../utils/activity'; + +const paramsSchema = z.object({ + gameTypeName: z.string().transform(decodeString), + mapName: z.string().transform(decodeString), +}); + +export async function GET( + request: NextRequest, + { params }: { params: { gameTypeName: string; mapName: string } } +) { + const parsedParams = paramsSchema.safeParse(params); + + if (!parsedParams.success) { + return NextResponse.json({ error: 'Bad request' }, { status: 400 }); + } + + const map = await prisma.map.findUnique({ + where: { + name_gameTypeName: { + name: parsedParams.data.mapName, + gameTypeName: parsedParams.data.gameTypeName, + }, + }, + select: { id: true }, + }); + + if (map === null) { + return NextResponse.json({ error: 'Not found' }, { status: 404 }); + } + + return NextResponse.json( + await getMapActivity(map.id, rangeParamsFromSearch(request.nextUrl.searchParams)) + ); +} diff --git a/apps/frontend/app/api/player/[playerName]/activity/route.ts b/apps/frontend/app/api/player/[playerName]/activity/route.ts new file mode 100644 index 0000000..9d95270 --- /dev/null +++ b/apps/frontend/app/api/player/[playerName]/activity/route.ts @@ -0,0 +1,33 @@ +import { NextRequest, NextResponse } from 'next/server'; +import { z } from 'zod'; +import { decodeString } from '../../../../../utils/encoding'; +import prisma from '../../../../../utils/prisma'; +import { getPlayerActivity, rangeParamsFromSearch } from '../../../../../utils/activity'; + +const paramsSchema = z.object({ + playerName: z.string().transform(decodeString), +}); + +export async function GET( + request: NextRequest, + { params }: { params: { playerName: string } } +) { + const parsedParams = paramsSchema.safeParse(params); + + if (!parsedParams.success) { + return NextResponse.json({ error: 'Bad request' }, { status: 400 }); + } + + const player = await prisma.player.findUnique({ + where: { name: parsedParams.data.playerName }, + select: { id: true }, + }); + + if (player === null) { + return NextResponse.json({ error: 'Not found' }, { status: 404 }); + } + + return NextResponse.json( + await getPlayerActivity(player.id, rangeParamsFromSearch(request.nextUrl.searchParams)) + ); +} diff --git a/apps/frontend/app/api/server/[ip]/[port]/activity/route.ts b/apps/frontend/app/api/server/[ip]/[port]/activity/route.ts new file mode 100644 index 0000000..0809343 --- /dev/null +++ b/apps/frontend/app/api/server/[ip]/[port]/activity/route.ts @@ -0,0 +1,36 @@ +import { NextRequest, NextResponse } from 'next/server'; +import { z } from 'zod'; +import { decodeIp } from '../../../../../../utils/encoding'; +import prisma from '../../../../../../utils/prisma'; +import { getServerActivity, rangeParamsFromSearch } from '../../../../../../utils/activity'; + +const paramsSchema = z.object({ + ip: z.string().transform(decodeIp), + port: z.coerce.number().int().positive().max(65535), +}); + +export async function GET( + request: NextRequest, + { params }: { params: { ip: string; port: string } } +) { + const parsedParams = paramsSchema.safeParse(params); + + if (!parsedParams.success) { + return NextResponse.json({ error: 'Bad request' }, { status: 400 }); + } + + const { ip, port } = parsedParams.data; + + const gameServer = await prisma.gameServer.findUnique({ + where: { ip_port: { ip, port } }, + select: { id: true }, + }); + + if (gameServer === null) { + return NextResponse.json({ error: 'Not found' }, { status: 404 }); + } + + return NextResponse.json( + await getServerActivity(gameServer.id, rangeParamsFromSearch(request.nextUrl.searchParams)) + ); +} diff --git a/apps/frontend/app/clan/[clanName]/layout.tsx b/apps/frontend/app/clan/[clanName]/layout.tsx index 5528b03..579aa2b 100644 --- a/apps/frontend/app/clan/[clanName]/layout.tsx +++ b/apps/frontend/app/clan/[clanName]/layout.tsx @@ -4,7 +4,10 @@ import { LayoutTabs } from './LayoutTabs'; import { paramsSchema } from './schema'; import { z } from 'zod'; import { formatPlayTime } from '../../../utils/format'; +import { encodeString } from '../../../utils/encoding'; import { ClanPlayerCount } from './ClanPlayerCount'; +import { ActivityCalendarSection } from '../../../components/ActivityCalendarSection'; +import { getClanActivity } from '../../../utils/activity'; export default async function Index({ params, @@ -17,6 +20,7 @@ export default async function Index({ const clan = await prisma.clan.findUnique({ select: { + id: true, name: true, playTime: true, activePlayerCount: true, @@ -29,7 +33,11 @@ export default async function Index({ }, }); - const [gameTypeCount, mapCount] = await Promise.all([ + if (clan === null) { + return notFound(); + } + + const [gameTypeCount, mapCount, activity] = await Promise.all([ prisma.clanInfoGameType.count({ where: { clan: { @@ -44,27 +52,32 @@ export default async function Index({ }, } }), + getClanActivity(clan.id, { range: '1y' }), ]); - if (clan === null) { - return notFound(); - } - return (
-
-
-

{clan.name}

- +
+ -
- +
+

{clan.name}

+
+ + + + + Playtime: {formatPlayTime(clan.playTime)} + +
+
+
+
+
+ +
+

{gameTypeName}

+
+
+
+ +
+
+ +
+

{mapName}

+ + + {gameTypeName} + + +
+
+
+
+ +

0.7 servers and map counts

diff --git a/apps/frontend/app/player/[playerName]/layout.tsx b/apps/frontend/app/player/[playerName]/layout.tsx index d1d6396..331373d 100644 --- a/apps/frontend/app/player/[playerName]/layout.tsx +++ b/apps/frontend/app/player/[playerName]/layout.tsx @@ -3,11 +3,13 @@ import { z } from 'zod'; import { notFound } from 'next/navigation'; import Link from 'next/link'; import Image from 'next/image'; -import { formatPlayTime } from '../../../utils/format'; import prisma from '../../../utils/prisma'; import { LayoutTabs } from './LayoutTabs'; import { LastSeen } from '../../../components/LastSeen'; +import { formatPlayTime } from '../../../utils/format'; import { encodeString } from '../../../utils/encoding'; +import { ActivityCalendarSection } from '../../../components/ActivityCalendarSection'; +import { getPlayerActivity } from '../../../utils/activity'; export default async function Index({ params, @@ -20,16 +22,16 @@ export default async function Index({ const player = await prisma.player.findUnique({ select: { + id: true, name: true, - playTime: true, clanName: true, + playTime: true, lastSeenAt: true, gameServerStateClients: { select: { gameServerState: { select: { - createdAt: true, gameServer: { select: { ip: true, @@ -40,28 +42,17 @@ export default async function Index({ }, }, }, - - playerInfoMaps: { - select: { - map: { - select: { - name: true, - gameTypeName: true, - }, - }, - playTime: true, - }, - orderBy: { - playTime: 'desc', - }, - }, }, where: { name: playerName, }, }); - const [mapCount, gameTypeCount, clanCount] = await Promise.all([ + if (player === null) { + return notFound(); + } + + const [mapCount, gameTypeCount, clanCount, activity] = await Promise.all([ prisma.playerInfoMap.count({ where: { playerName, @@ -77,42 +68,49 @@ export default async function Index({ playerName, }, }), + getPlayerActivity(player.id, { range: '1y' }), ]); - if (player === null) { - return notFound(); - } - return (
-
- Player -
-

{player.name}

- - {player.clanName !== null && ( - - {player.clanName} - - )} - -
- +
+ Player +
+

{player.name}

+
+ {player.clanName !== null && ( + + + {player.clanName} + + + )} + + Playtime: {formatPlayTime(player.playTime)} + + + client.gameServerState.gameServer) + .filter((gameServer) => gameServer !== null)} + lastSeenAt={player.lastSeenAt} + /> + +
+
+
+
; } + const activity = await getServerActivity(gameServer.id, { range: '1y' }); + return (
-
-
-

- {gameServer.gameServerState.name} -

-
- - - {gameServer.gameServerState.map.gameTypeName} - - - - - {gameServer.gameServerState.map.name} - - - {`${gameServer.gameServerState.numClients} / ${gameServer.gameServerState.maxClients} clients`} - - Playtime: {formatPlayTime(gameServer.playTime)} - +
+
+ +
+

+ {gameServer.gameServerState.name} +

+
+ + + {gameServer.gameServerState.map.gameTypeName} + + + + + {gameServer.gameServerState.map.name} + + + {`${gameServer.gameServerState.numClients} / ${gameServer.gameServerState.maxClients} clients`} + + Playtime: {formatPlayTime(gameServer.playTime)} + +
-
-
-

- Server address -

- - {ipAndPort(gameServer.ip, gameServer.port)} - -
+

Teerank

@@ -213,6 +253,58 @@ export default async function Index() { +

Rollup

+
+
+ Latest day +
+ + {latestRollupDay === null ? 'None' : formatUtcDay(latestRollupDay)} + + {rollupOnTime ? ( + OK + ) : ( + Late + )} +
+
+ +
+ Oldest day +
+ + {oldestRollupDay === null ? 'None' : formatUtcDay(oldestRollupDay)} + +
+
+ +
+ Backfill gap +
+ {missingRollupDays > 0 && ( + + {missingRollupDays} days missing + + )} + {missingRollupDays === 0 ? ( + Complete + ) : ( + Filling + )} +
+
+ +
+ Failed jobs +
+ {rollupFailedCount} + {rollupFailedCount > 0 && ( + Failing + )} +
+
+
+

Teeworlds

{masterServers.map((masterServer) => ( diff --git a/apps/frontend/components/ActivityCalendar.tsx b/apps/frontend/components/ActivityCalendar.tsx new file mode 100644 index 0000000..bfa00ed --- /dev/null +++ b/apps/frontend/components/ActivityCalendar.tsx @@ -0,0 +1,192 @@ +'use client'; + +import { useEffect, useRef, useState } from 'react'; +import { format } from 'date-fns'; +import { DAY_MS, addUtcDays, formatUtcDay, parseUtcDay } from '@teerank/teerank/date'; + +export type ActivityDay = { + day: string; // YYYY-MM-DD + value: number; + tooltip: string; +}; + +export type ActivityPayload = { + range: string; + from: string; + to: string; + days: ActivityDay[]; +}; + +const EMPTY_COLOR = '#00000010'; +const LEVEL_COLORS = ['#e6dcae', '#d2ba6a', '#b3922e', '#997700']; + +const TOOLTIP_DELAY_MS = 300; + +function quantile(sorted: number[], q: number) { + return sorted[Math.min(sorted.length - 1, Math.floor(sorted.length * q))]; +} + +function cellColor(value: number | undefined, thresholds: number[]) { + if (value === undefined || value <= 0) { + return EMPTY_COLOR; + } + + for (const [level, threshold] of thresholds.entries()) { + if (value <= threshold) { + return LEVEL_COLORS[level]; + } + } + + return LEVEL_COLORS[LEVEL_COLORS.length - 1]; +} + +type Tooltip = { + x: number; + y: number; + text: string; +}; + +export function ActivityCalendar({ + payload, + showLabels, +}: { + payload: ActivityPayload; + showLabels: boolean; +}) { + const containerRef = useRef(null); + const [containerWidth, setContainerWidth] = useState(0); + const [tooltip, setTooltip] = useState(null); + const timerRef = useRef>(); + + useEffect(() => { + const container = containerRef.current; + + if (container === null) { + return; + } + + const observer = new ResizeObserver(() => setContainerWidth(container.clientWidth)); + observer.observe(container); + setContainerWidth(container.clientWidth); + + return () => observer.disconnect(); + }, []); + + useEffect(() => () => clearTimeout(timerRef.current), []); + + const showTooltip = (event: React.MouseEvent, text: string) => { + const bounds = event.currentTarget.getBoundingClientRect(); + clearTimeout(timerRef.current); + timerRef.current = setTimeout(() => { + setTooltip({ x: bounds.left + bounds.width / 2, y: bounds.top, text }); + }, TOOLTIP_DELAY_MS); + }; + + const hideTooltip = () => { + clearTimeout(timerRef.current); + setTooltip(null); + }; + + const from = parseUtcDay(payload.from); + const to = parseUtcDay(payload.to); + const days = new Map(payload.days.map((day) => [day.day, day])); + + const positives = payload.days + .map(({ value }) => value) + .filter((value) => value > 0) + .sort((a, b) => a - b); + const thresholds = [quantile(positives, 0.25), quantile(positives, 0.5), quantile(positives, 0.75)]; + + // Columns start on the Sunday on or before the first day. + const startSunday = addUtcDays(from, -from.getUTCDay()); + const weeks = Math.floor((to.getTime() - startSunday.getTime()) / DAY_MS / 7) + 1; + + const pitch = containerWidth === 0 ? 14 : Math.min(22, Math.max(12, containerWidth / weeks)); + const cell = pitch * 0.78; + const width = weeks * pitch - (pitch - cell); + const height = 7 * pitch - (pitch - cell); + + const cells: React.ReactNode[] = []; + const monthLabels: { week: number; label: string }[] = []; + let previousMonth = -1; + let lastLabeledColumn = -3; + + for (let week = 0; week < weeks; week++) { + let columnMonth: number | null = null; + + for (let weekday = 0; weekday < 7; weekday++) { + const date = addUtcDays(startSunday, week * 7 + weekday); + + if (date < from || date > to) { + continue; + } + + columnMonth ??= date.getUTCMonth(); + + const iso = formatUtcDay(date); + const day = days.get(iso); + const text = day?.tooltip ?? `No activity on ${format(date, 'MMM d, yyyy')}`; + + cells.push( + showTooltip(event, text)} + onMouseLeave={hideTooltip} + /> + ); + } + + // Label columns where a new month starts, keeping labels apart. + if (columnMonth !== null && columnMonth !== previousMonth) { + if (week - lastLabeledColumn >= 3 && week <= weeks - 3) { + lastLabeledColumn = week; + monthLabels.push({ + week, + label: format(new Date(Date.UTC(2000, columnMonth, 1)), 'MMM'), + }); + } + previousMonth = columnMonth; + } + } + + return ( +
+ + {cells} + + +
+ {monthLabels.map(({ week, label }) => ( + + {label} + + ))} +
+ + {tooltip !== null && ( +
+ {tooltip.text} +
+ )} +
+ ); +} diff --git a/apps/frontend/components/ActivityCalendarSection.tsx b/apps/frontend/components/ActivityCalendarSection.tsx new file mode 100644 index 0000000..f2cbece --- /dev/null +++ b/apps/frontend/components/ActivityCalendarSection.tsx @@ -0,0 +1,84 @@ +'use client'; + +import { useState } from 'react'; +import { DAY_MS, formatUtcDay, parseUtcDay, utcYesterday } from '@teerank/teerank/date'; +import { ActivityCalendar, ActivityPayload } from './ActivityCalendar'; + +const ARROW_CLASSES = + 'absolute inset-y-0 z-20 flex w-8 items-center justify-center text-lg ' + + 'leading-none text-[#970] transition-opacity xl:w-20'; + +export function ActivityCalendarSection({ + apiPath, + initial, +}: { + apiPath: string; + initial: ActivityPayload; +}) { + const [payload, setPayload] = useState(initial); + const [loading, setLoading] = useState(false); + const [active, setActive] = useState(false); + + const atLatest = payload.to >= formatUtcDay(utcYesterday()); + const chromeClasses = active ? 'visible opacity-100' : 'invisible opacity-0'; + + const shiftWindow = async (direction: 1 | -1) => { + const from = parseUtcDay(payload.from); + const to = parseUtcDay(payload.to); + const spanMs = to.getTime() - from.getTime() + DAY_MS; + + let newFrom = new Date(from.getTime() + direction * spanMs); + let newTo = new Date(to.getTime() + direction * spanMs); + + const yesterday = utcYesterday(); + if (newTo > yesterday) { + newTo = yesterday; + newFrom = new Date(newTo.getTime() - spanMs + DAY_MS); + } + + setLoading(true); + try { + const response = await fetch( + `${apiPath}?from=${formatUtcDay(newFrom)}&to=${formatUtcDay(newTo)}` + ); + if (response.ok) { + setPayload(await response.json()); + } + } finally { + setLoading(false); + } + }; + + return ( +
setActive(false)} + > + + {!atLatest && ( + + )} + +
setActive(true)} + > +
+ +
+
+
+ ); +} diff --git a/apps/frontend/components/Chart.tsx b/apps/frontend/components/Chart.tsx new file mode 100644 index 0000000..e1f2dee --- /dev/null +++ b/apps/frontend/components/Chart.tsx @@ -0,0 +1,200 @@ +'use client'; + +import { useState } from 'react'; +import { format } from 'date-fns'; +import { formatInteger } from '../utils/format'; + +export type ChartPoint = { + date: Date; + value: number | null; +}; + +const WIDTH = 1000; +const HEIGHT = 260; +const MARGIN = { top: 12, right: 12, bottom: 32, left: 90 }; +const INNER_WIDTH = WIDTH - MARGIN.left - MARGIN.right; +const INNER_HEIGHT = HEIGHT - MARGIN.top - MARGIN.bottom; + +function niceCeiling(value: number) { + if (value <= 0) { + return 1; + } + + const magnitude = Math.pow(10, Math.floor(Math.log10(value))); + + for (const factor of [1, 2, 2.5, 5, 10]) { + if (factor * magnitude >= value) { + return factor * magnitude; + } + } + + return 10 * magnitude; +} + +function slotX(index: number, count: number) { + return MARGIN.left + ((index + 0.5) / count) * INNER_WIDTH; +} + +function xTickIndices(count: number) { + const tickCount = Math.min(5, count); + const indices = new Set(); + + for (let tick = 0; tick < tickCount; tick++) { + indices.add(Math.round((tick / Math.max(1, tickCount - 1)) * (count - 1))); + } + + return [...indices]; +} + +function XAxis({ + dates, + formatDate, + fontSize, +}: { + dates: Date[]; + formatDate: (date: Date) => string; + fontSize: number; +}) { + return ( + <> + {xTickIndices(dates.length).map((index) => ( + + {formatDate(dates[index])} + + ))} + + ); +} + +function YAxis({ + ticks, + scaleY, + formatValue, + fontSize, +}: { + ticks: number[]; + scaleY: (value: number) => number; + formatValue: (value: number) => string; + fontSize: number; +}) { + return ( + <> + {ticks.map((tick) => ( + + + + {formatValue(tick)} + + + ))} + + ); +} + +function EmptyChart({ label }: { label: string }) { + return ( +
+ {label} +
+ ); +} + +export function BarChart({ + points, + formatDate = (date) => format(date, 'MMM d'), + formatTooltipDate = formatDate, + formatValue = formatInteger, + emptyLabel = 'No data yet', + fontSize = 14, +}: { + points: ChartPoint[]; + formatDate?: (date: Date) => string; + formatTooltipDate?: (date: Date) => string; + formatValue?: (value: number) => string; + emptyLabel?: string; + // Bump for charts rendered at half width, where the viewBox scales text down. + fontSize?: number; +}) { + const [tooltip, setTooltip] = useState<{ x: number; y: number; text: string } | null>(null); + + const showTooltip = (event: React.MouseEvent, text: string) => { + const bounds = event.currentTarget.getBoundingClientRect(); + setTooltip({ x: bounds.left + bounds.width / 2, y: bounds.top, text }); + }; + + const hideTooltip = () => { + setTooltip(null); + }; + + const values = points.flatMap((point) => (point.value === null ? [] : point.value)); + + if (values.length === 0) { + return ; + } + + const max = niceCeiling(Math.max(...values)); + const scaleY = (value: number) => MARGIN.top + INNER_HEIGHT * (1 - value / max); + const barWidth = (INNER_WIDTH / points.length) * 0.7; + + return ( +
+ + + point.date)} formatDate={formatDate} fontSize={fontSize} /> + + {points.map((point, index) => + point.value === null ? null : ( + + showTooltip(event, `${formatValue(point.value as number)} on ${formatTooltipDate(point.date)}`) + } + onMouseLeave={hideTooltip} + /> + ) + )} + + + {tooltip !== null && ( +
+ {tooltip.text} +
+ )} +
+ ); +} diff --git a/apps/frontend/components/DailyPlayersSection.tsx b/apps/frontend/components/DailyPlayersSection.tsx new file mode 100644 index 0000000..43f3822 --- /dev/null +++ b/apps/frontend/components/DailyPlayersSection.tsx @@ -0,0 +1,80 @@ +'use client'; + +import { useState } from 'react'; +import { BarChart } from './Chart'; +import { fillSeries } from '../utils/series'; +import { eachUtcDay, parseUtcDay } from '@teerank/teerank/date'; +import { DailyPlayersPayload } from '../utils/dailyPlayers'; + +const PRESETS: { key: string; label: string }[] = [ + { key: '30d', label: 'Last 30 days' }, + { key: '90d', label: 'Last 90 days' }, + { key: '1y', label: 'Last year' }, + { key: 'all', label: 'All time' }, +]; + +export function DailyPlayersSection({ initial }: { initial: DailyPlayersPayload }) { + const [payload, setPayload] = useState(initial); + const [loading, setLoading] = useState(false); + const [open, setOpen] = useState(false); + + const select = async (range: string) => { + setOpen(false); + setLoading(true); + try { + const response = await fetch(`/api/daily-players?range=${range}`); + if (response.ok) { + setPayload(await response.json()); + } + } finally { + setLoading(false); + } + }; + + const domain = eachUtcDay(parseUtcDay(payload.from), parseUtcDay(payload.to)); + + return ( +
+
+

Daily players

+ + + + {open && ( + <> + setOpen(false)} /> + + {PRESETS.map((preset) => ( + + ))} + + + )} + +
+ +
+ ({ at: parseUtcDay(day), value: players })) + )} + emptyLabel="No history yet — the first rollup lands tomorrow" + /> +
+
+ ); +} diff --git a/apps/frontend/utils/activity.ts b/apps/frontend/utils/activity.ts new file mode 100644 index 0000000..be12d5d --- /dev/null +++ b/apps/frontend/utils/activity.ts @@ -0,0 +1,182 @@ +import { format } from 'date-fns'; +import prisma from './prisma'; +import { formatInteger, formatPlayTime } from './format'; +import { DAY_MS, addUtcDays, formatUtcDay, parseUtcDay, utcYesterday } from '@teerank/teerank/date'; +import type { ActivityDay, ActivityPayload } from '../components/ActivityCalendar'; + +export type RangeParams = { + range?: string; + from?: string; + to?: string; +}; + +const MAX_SPAN_DAYS = 5 * 366; + +const PRESET_DAYS: Record = { + '30d': 30, + '90d': 90, + '1y': 365, +}; + +export function rangeParamsFromSearch(searchParams: URLSearchParams): RangeParams { + return { + range: searchParams.get('range') ?? undefined, + from: searchParams.get('from') ?? undefined, + to: searchParams.get('to') ?? undefined, + }; +} + +function parseDayParam(day: string | undefined) { + if (day === undefined || !/^\d{4}-\d{2}-\d{2}$/.test(day)) { + return null; + } + + const date = parseUtcDay(day); + return Number.isNaN(date.getTime()) ? null : date; +} + +async function resolveDomain( + params: RangeParams, + getMinDay: () => Promise +): Promise<{ range: string; from: Date; to: Date }> { + const yesterday = utcYesterday(); + const customFrom = parseDayParam(params.from); + const customTo = parseDayParam(params.to); + + if (customFrom !== null && customTo !== null && customFrom <= customTo) { + const to = customTo > yesterday ? yesterday : customTo; + const minFrom = addUtcDays(to, -(MAX_SPAN_DAYS - 1)); + const from = customFrom < minFrom ? minFrom : customFrom; + return { range: 'custom', from, to }; + } + + if (params.range === 'all') { + const minDay = await getMinDay(); + const minFrom = addUtcDays(yesterday, -(MAX_SPAN_DAYS - 1)); + const from = + minDay === null || minDay > yesterday + ? yesterday + : new Date(Math.floor(minDay.getTime() / DAY_MS) * DAY_MS); + return { range: 'all', from: from < minFrom ? minFrom : from, to: yesterday }; + } + + const days = PRESET_DAYS[params.range ?? '90d'] ?? 90; + return { + range: PRESET_DAYS[params.range ?? ''] !== undefined ? (params.range as string) : '90d', + from: addUtcDays(yesterday, -(days - 1)), + to: yesterday, + }; +} + +function payload(range: string, from: Date, to: Date, days: ActivityDay[]): ActivityPayload { + return { range, from: formatUtcDay(from), to: formatUtcDay(to), days }; +} + +function tooltipDate(day: Date) { + return format(day, 'MMM d, yyyy'); +} + +export async function getPlayerActivity(playerId: number, params: RangeParams) { + const { range, from, to } = await resolveDomain(params, async () => { + const result = await prisma.playerDay.aggregate({ + _min: { day: true }, + where: { playerId }, + }); + return result._min.day; + }); + + const rows = await prisma.playerDay.findMany({ + where: { playerId, day: { gte: from, lte: to } }, + select: { day: true, playTime: true }, + }); + + return payload(range, from, to, rows.map((row) => ({ + day: formatUtcDay(row.day), + value: row.playTime, + tooltip: `${formatPlayTime(BigInt(row.playTime))} on ${tooltipDate(row.day)}`, + }))); +} + +export async function getClanActivity(clanId: number, params: RangeParams) { + const { range, from, to } = await resolveDomain(params, async () => { + const result = await prisma.clanDay.aggregate({ + _min: { day: true }, + where: { clanId }, + }); + return result._min.day; + }); + + const rows = await prisma.clanDay.findMany({ + where: { clanId, day: { gte: from, lte: to } }, + select: { day: true, playTime: true, playerCount: true }, + }); + + return payload(range, from, to, rows.map((row) => ({ + day: formatUtcDay(row.day), + value: row.playTime, + tooltip: `${formatPlayTime(BigInt(row.playTime))} · ${formatInteger(row.playerCount)} players on ${tooltipDate(row.day)}`, + }))); +} + +export async function getGameTypeActivity(gameTypeId: number, params: RangeParams) { + const { range, from, to } = await resolveDomain(params, async () => { + const result = await prisma.gameTypeDay.aggregate({ + _min: { day: true }, + where: { gameTypeId }, + }); + return result._min.day; + }); + + const rows = await prisma.gameTypeDay.findMany({ + where: { gameTypeId, day: { gte: from, lte: to } }, + select: { day: true, playTime: true, playerCount: true }, + }); + + return payload(range, from, to, rows.map((row) => ({ + day: formatUtcDay(row.day), + value: row.playTime, + tooltip: `${formatPlayTime(BigInt(row.playTime))} · ${formatInteger(row.playerCount)} players on ${tooltipDate(row.day)}`, + }))); +} + +export async function getMapActivity(mapId: number, params: RangeParams) { + const { range, from, to } = await resolveDomain(params, async () => { + const result = await prisma.mapDay.aggregate({ + _min: { day: true }, + where: { mapId }, + }); + return result._min.day; + }); + + const rows = await prisma.mapDay.findMany({ + where: { mapId, day: { gte: from, lte: to } }, + select: { day: true, playTime: true, playerCount: true }, + }); + + return payload(range, from, to, rows.map((row) => ({ + day: formatUtcDay(row.day), + value: row.playTime, + tooltip: `${formatPlayTime(BigInt(row.playTime))} · ${formatInteger(row.playerCount)} players on ${tooltipDate(row.day)}`, + }))); +} + +export async function getServerActivity(gameServerId: number, params: RangeParams) { + const { range, from, to } = await resolveDomain(params, async () => { + const result = await prisma.serverDay.aggregate({ + _min: { day: true }, + where: { gameServerId }, + }); + return result._min.day; + }); + + const rows = await prisma.serverDay.findMany({ + where: { gameServerId, day: { gte: from, lte: to } }, + select: { day: true, avgClients: true, maxClients: true }, + }); + + return payload(range, from, to, rows.map((row) => ({ + day: formatUtcDay(row.day), + value: row.avgClients, + tooltip: `${row.avgClients} avg clients · peak ${row.maxClients} on ${tooltipDate(row.day)}`, + }))); +} diff --git a/apps/frontend/utils/dailyPlayers.ts b/apps/frontend/utils/dailyPlayers.ts new file mode 100644 index 0000000..6acf5b3 --- /dev/null +++ b/apps/frontend/utils/dailyPlayers.ts @@ -0,0 +1,51 @@ +import { unstable_cache } from 'next/cache'; +import prisma from './prisma'; +import { addUtcDays, formatUtcDay, utcYesterday } from '@teerank/teerank/date'; + +export type DailyPlayersPayload = { + range: string; + from: string; + to: string; + days: { day: string; players: number }[]; +}; + +const MAX_SPAN_DAYS = 5 * 366; + +const PRESET_DAYS: Record = { + '30d': 30, + '90d': 90, + '1y': 365, +}; + +export const getDailyPlayers = unstable_cache( + async (range: string): Promise => { + const to = utcYesterday(); + let from: Date; + + if (range === 'all') { + const result = await prisma.playerDay.aggregate({ _min: { day: true } }); + const minFrom = addUtcDays(to, -(MAX_SPAN_DAYS - 1)); + const minDay = result._min.day; + from = minDay === null || minDay > to ? to : minDay < minFrom ? minFrom : minDay; + } else { + range = PRESET_DAYS[range] !== undefined ? range : '90d'; + from = addUtcDays(to, -(PRESET_DAYS[range] - 1)); + } + + const rows = await prisma.playerDay.groupBy({ + by: ['day'], + where: { day: { gte: from } }, + _count: { _all: true }, + orderBy: { day: 'asc' }, + }); + + return { + range, + from: formatUtcDay(from), + to: formatUtcDay(to), + days: rows.map(({ day, _count }) => ({ day: formatUtcDay(day), players: _count._all })), + }; + }, + ['home-daily-players'], + { revalidate: 3600 } +); diff --git a/apps/frontend/utils/series.ts b/apps/frontend/utils/series.ts new file mode 100644 index 0000000..aed50c2 --- /dev/null +++ b/apps/frontend/utils/series.ts @@ -0,0 +1,17 @@ +import { ChartPoint } from '../components/Chart'; +import { addUtcDays, utcYesterday } from '@teerank/teerank/date'; + +export function lastUtcDays(count: number) { + const yesterday = utcYesterday(); + + return Array.from({ length: count }, (_, index) => addUtcDays(yesterday, index - (count - 1))); +} + +export function fillSeries(domain: Date[], rows: { at: Date; value: number }[]): ChartPoint[] { + const values = new Map(rows.map((row) => [row.at.getTime(), row.value])); + + return domain.map((date) => ({ + date, + value: values.get(date.getTime()) ?? null, + })); +} diff --git a/apps/scheduler/src/main.ts b/apps/scheduler/src/main.ts index fe15433..3e737eb 100644 --- a/apps/scheduler/src/main.ts +++ b/apps/scheduler/src/main.ts @@ -8,6 +8,8 @@ import { fillClanActivePlayerCountScheduler } from './schedulers/fillClanActiveP import { cleanAllQueues } from '@teerank/teerank'; import { updateGlobalCountsScheduler } from './schedulers/updateGlobalCountsScheduler'; import { archiveSnapshotsScheduler } from './schedulers/archiveSnapshotsScheduler'; +import { rollupDayScheduler } from './schedulers/rollupDayScheduler'; +import { rollupBackfillScheduler } from './schedulers/rollupBackfillScheduler'; async function main() { if (process.env.NODE_ENV === 'development') { @@ -25,6 +27,8 @@ async function main() { fillClanActivePlayerCountScheduler(); updateGlobalCountsScheduler(); archiveSnapshotsScheduler(); + rollupDayScheduler(); + rollupBackfillScheduler(); } main(); diff --git a/apps/scheduler/src/schedulers/rollupBackfillScheduler.ts b/apps/scheduler/src/schedulers/rollupBackfillScheduler.ts new file mode 100644 index 0000000..b1c279e --- /dev/null +++ b/apps/scheduler/src/schedulers/rollupBackfillScheduler.ts @@ -0,0 +1,66 @@ +import { ListObjectsV2Command } from "@aws-sdk/client-s3"; +import { hoursToMilliseconds, minutesToMilliseconds } from "date-fns"; +import { + S3_BUCKET, + SNAPSHOT_RETENTION_HOURS, + addUtcDays, + formatUtcDay, + getEnvInt, + getS3Client, + parseUtcDay, + scheduleRollupBackfill, +} from "@teerank/teerank"; +import { schedule } from "../utils"; +import { prisma } from "../prisma"; + +const ROLLUP_BACKFILL_DAYS_PER_TICK = getEnvInt('ROLLUP_BACKFILL_DAYS_PER_TICK', 4); + +async function listArchivedDays() { + const s3 = getS3Client(); + const days: string[] = []; + let continuationToken: string | undefined; + + do { + const result = await s3.send(new ListObjectsV2Command({ + Bucket: S3_BUCKET, + Prefix: 'snapshots/', + Delimiter: '/', + ContinuationToken: continuationToken, + })); + + for (const prefix of result.CommonPrefixes ?? []) { + const match = prefix.Prefix?.match(/dt=(\d{4}-\d{2}-\d{2})\/$/); + if (match !== null && match !== undefined) { + days.push(match[1]); + } + } + + continuationToken = result.NextContinuationToken; + } while (continuationToken !== undefined); + + return days.sort(); +} + +export function rollupBackfillScheduler() { + schedule(minutesToMilliseconds(15), async () => { + const days = (await listArchivedDays()).slice(0, -1).filter((day) => { + const dayEndMs = addUtcDays(parseUtcDay(day), 1).getTime(); + return dayEndMs + hoursToMilliseconds(SNAPSHOT_RETENTION_HOURS) <= Date.now(); + }); + + if (days.length === 0) { + return; + } + + const rolledUpDays = await prisma.playerDay.groupBy({ + by: ['day'], + }); + const rolledUp = new Set(rolledUpDays.map(({ day }) => formatUtcDay(day))); + + const missing = days.filter((day) => !rolledUp.has(day)).slice(0, ROLLUP_BACKFILL_DAYS_PER_TICK); + + for (const day of missing) { + await scheduleRollupBackfill({ day }); + } + }); +} diff --git a/apps/scheduler/src/schedulers/rollupDayScheduler.ts b/apps/scheduler/src/schedulers/rollupDayScheduler.ts new file mode 100644 index 0000000..d29da1c --- /dev/null +++ b/apps/scheduler/src/schedulers/rollupDayScheduler.ts @@ -0,0 +1,9 @@ +import { hoursToMilliseconds } from "date-fns"; +import { schedule } from "../utils"; +import { addUtcDays, formatUtcDay, scheduleRollupDay } from "@teerank/teerank"; + +export function rollupDayScheduler() { + schedule(hoursToMilliseconds(1), async () => { + await scheduleRollupDay({ day: formatUtcDay(addUtcDays(new Date(), -1)) }); + }); +} diff --git a/apps/worker/src/main.ts b/apps/worker/src/main.ts index 8d13d76..af03d89 100644 --- a/apps/worker/src/main.ts +++ b/apps/worker/src/main.ts @@ -7,6 +7,8 @@ import { startUpdateGameTypesCountsWorker } from "./workers/updateGameTypesCount import { startFillClanActivePlayerCountWorker } from "./workers/fillClanActivePlayerCount"; import { startUpdateGlobalCountsWorker } from "./workers/updateGlobalCounts"; import { startArchiveSnapshotsWorker } from "./workers/archiveSnapshots"; +import { startRollupDayWorker } from "./workers/rollupDay"; +import { startRollupBackfillWorker } from "./workers/rollupBackfill"; async function main() { const workers = await Promise.all([ @@ -19,6 +21,8 @@ async function main() { startFillClanActivePlayerCountWorker(), startUpdateGlobalCountsWorker(), startArchiveSnapshotsWorker(), + startRollupDayWorker(), + startRollupBackfillWorker(), ]); async function gracefulShutdown(signal: string) { diff --git a/apps/worker/src/parquet.ts b/apps/worker/src/parquet.ts index f5262d0..b38ab7c 100644 --- a/apps/worker/src/parquet.ts +++ b/apps/worker/src/parquet.ts @@ -5,6 +5,7 @@ import { Table, TimestampMillisecond, Utf8, + tableFromIPC, tableToIPC, vectorFromArray, } from "apache-arrow"; @@ -12,6 +13,7 @@ import { Compression, Table as WasmTable, WriterPropertiesBuilder, + readParquet, writeParquet, } from "parquet-wasm"; @@ -67,3 +69,34 @@ export function encodeSnapshotRowsToParquet(rows: SnapshotArchiveRow[]): Uint8Ar return writeParquet(wasmTable, writerProperties); } + +export function decodeSnapshotRowsFromParquet(parquet: Uint8Array): SnapshotArchiveRow[] { + const table = tableFromIPC(readParquet(parquet).intoIPCStream()); + const rows: SnapshotArchiveRow[] = []; + + for (let index = 0; index < table.numRows; index++) { + const row = table.get(index)!.toJSON(); + + rows.push({ + snapshotId: Number(row.snapshotId), + createdAt: new Date(Number(row.createdAt)), + gameServerId: row.gameServerId, + serverName: row.serverName, + version: row.version, + mapId: row.mapId, + mapName: row.mapName, + gameTypeName: row.gameTypeName, + numPlayers: row.numPlayers, + maxPlayers: row.maxPlayers, + numClients: row.numClients, + maxClients: row.maxClients, + playerName: row.playerName, + clanName: row.clanName, + score: row.score, + country: row.country, + inGame: row.inGame, + }); + } + + return rows; +} diff --git a/apps/worker/src/rollup/aggregateDay.ts b/apps/worker/src/rollup/aggregateDay.ts new file mode 100644 index 0000000..f13e1d6 --- /dev/null +++ b/apps/worker/src/rollup/aggregateDay.ts @@ -0,0 +1,153 @@ +import { millisecondsInHour } from "date-fns"; +import { removeDuplicatedClients } from "../utils"; + +export const OBSERVATION_SECONDS = 5 * 60; + +export type RollupSnapshot = { + createdAt: Date; + gameServerId: number; + mapId: number; + gameTypeName: string; + numClients: number; + clients: { + playerName: string; + clanName: string | null; + inGame: boolean; + }[]; +}; + +export type DayRollup = { + players: { playerName: string; playTime: number }[]; + serverDays: { gameServerId: number; avgClients: number; maxClients: number }[]; + maps: { mapId: number; playTime: number; playerCount: number }[]; + gameTypes: { gameTypeName: string; playTime: number; playerCount: number }[]; + clans: { clanName: string; playTime: number; playerCount: number }[]; +}; + +type ServerHourAggregate = { + gameServerId: number; + snapshotCount: number; + clientSum: number; + maxClients: number; +}; + +type PresenceAggregate = { + playTime: number; + players: Set; +}; + +function getOrCreate(map: Map, key: K, create: () => V): V { + let value = map.get(key); + + if (value === undefined) { + value = create(); + map.set(key, value); + } + + return value; +} + +function newPresence(): PresenceAggregate { + return { playTime: 0, players: new Set() }; +} + +export class DayAggregator { + private players = new Map(); + private serverHours = new Map(); + private maps = new Map(); + private gameTypes = new Map(); + private clans = new Map(); + + addSnapshot(snapshot: RollupSnapshot) { + const clients = removeDuplicatedClients(snapshot.clients); + const inGameCount = clients.filter((client) => client.inGame).length; + + const hour = Math.floor(snapshot.createdAt.getTime() / millisecondsInHour); + const serverHour = getOrCreate(this.serverHours, `${snapshot.gameServerId}\0${hour}`, () => ({ + gameServerId: snapshot.gameServerId, + snapshotCount: 0, + clientSum: 0, + maxClients: 0, + })); + + serverHour.snapshotCount += 1; + serverHour.clientSum += snapshot.numClients; + serverHour.maxClients = Math.max(serverHour.maxClients, snapshot.numClients); + + if (clients.length === 0) { + return; + } + + const map = getOrCreate(this.maps, snapshot.mapId, newPresence); + const gameType = getOrCreate(this.gameTypes, snapshot.gameTypeName, newPresence); + + map.playTime += inGameCount * OBSERVATION_SECONDS; + gameType.playTime += inGameCount * OBSERVATION_SECONDS; + + for (const client of clients) { + const playTime = client.inGame ? OBSERVATION_SECONDS : 0; + + this.players.set(client.playerName, (this.players.get(client.playerName) ?? 0) + playTime); + map.players.add(client.playerName); + gameType.players.add(client.playerName); + + if (client.clanName !== null) { + const clan = getOrCreate(this.clans, client.clanName, newPresence); + clan.playTime += playTime; + clan.players.add(client.playerName); + } + } + } + + finalize(): DayRollup { + const serverDays = new Map(); + + for (const serverHour of this.serverHours.values()) { + if (serverHour.clientSum === 0) { + continue; + } + + const serverDay = getOrCreate(serverDays, serverHour.gameServerId, () => ({ + hourAverages: [], + maxClients: 0, + })); + + serverDay.hourAverages.push(serverHour.clientSum / serverHour.snapshotCount); + serverDay.maxClients = Math.max(serverDay.maxClients, serverHour.maxClients); + } + + return { + players: [...this.players.entries()].map(([playerName, playTime]) => ({ + playerName, + playTime, + })), + + serverDays: [...serverDays.entries()].map(([gameServerId, aggregate]) => ({ + gameServerId, + avgClients: Math.round( + aggregate.hourAverages.reduce((sum, average) => sum + average, 0) / + aggregate.hourAverages.length + ), + maxClients: aggregate.maxClients, + })), + + maps: [...this.maps.entries()].map(([mapId, aggregate]) => ({ + mapId, + playTime: aggregate.playTime, + playerCount: aggregate.players.size, + })), + + gameTypes: [...this.gameTypes.entries()].map(([gameTypeName, aggregate]) => ({ + gameTypeName, + playTime: aggregate.playTime, + playerCount: aggregate.players.size, + })), + + clans: [...this.clans.entries()].map(([clanName, aggregate]) => ({ + clanName, + playTime: aggregate.playTime, + playerCount: aggregate.players.size, + })), + }; + } +} diff --git a/apps/worker/src/rollup/partitions.ts b/apps/worker/src/rollup/partitions.ts new file mode 100644 index 0000000..047890b --- /dev/null +++ b/apps/worker/src/rollup/partitions.ts @@ -0,0 +1,19 @@ +import { createRollupPartition } from "@prisma/client/sql"; +import { prisma } from "../prisma"; + +const PARTITIONED_TABLES = ["PlayerDay", "ServerDay", "MapDay", "GameTypeDay", "ClanDay"]; + +function monthStart(date: Date, offsetMonths = 0) { + return new Date(Date.UTC(date.getUTCFullYear(), date.getUTCMonth() + offsetMonths, 1)); +} + +export async function ensureRollupPartitions(day: Date) { + for (const offset of [0, 1]) { + const from = monthStart(day, offset); + const to = monthStart(day, offset + 1); + + for (const table of PARTITIONED_TABLES) { + await prisma.$queryRawTyped(createRollupPartition(table, from, to)); + } + } +} diff --git a/apps/worker/src/rollup/writeDayRollup.ts b/apps/worker/src/rollup/writeDayRollup.ts new file mode 100644 index 0000000..38f2b37 --- /dev/null +++ b/apps/worker/src/rollup/writeDayRollup.ts @@ -0,0 +1,163 @@ +import { chunk } from "lodash"; +import { minutesToMilliseconds } from "date-fns"; +import { formatUtcDay } from "@teerank/teerank"; +import { prisma } from "../prisma"; +import { DayRollup } from "./aggregateDay"; +import { ensureRollupPartitions } from "./partitions"; + +const LOOKUP_CHUNK_SIZE = 5_000; +const INSERT_CHUNK_SIZE = 2_000; + +async function lookupIds( + names: string[], + findMany: (names: string[]) => Promise<{ name: string; id: number }[]> +) { + const ids = new Map(); + + for (const names_ of chunk(names, LOOKUP_CHUNK_SIZE)) { + for (const { name, id } of await findMany(names_)) { + ids.set(name, id); + } + } + + return ids; +} + +export async function writeDayRollup(day: Date, rollup: DayRollup) { + await ensureRollupPartitions(day); + + const [playerIds, clanIds, gameTypeIds] = await Promise.all([ + lookupIds( + rollup.players.map((row) => row.playerName), + (names) => + prisma.player.findMany({ + where: { name: { in: names } }, + select: { name: true, id: true }, + }) + ), + lookupIds( + rollup.clans.map((row) => row.clanName), + (names) => + prisma.clan.findMany({ + where: { name: { in: names } }, + select: { name: true, id: true }, + }) + ), + lookupIds( + rollup.gameTypes.map((row) => row.gameTypeName), + (names) => + prisma.gameType.findMany({ + where: { name: { in: names } }, + select: { name: true, id: true }, + }) + ), + ]); + + // Names that no longer resolve (deleted players/clans) are dropped: the + // rollup has no string columns to keep them under. + let droppedRows = 0; + + const playerRows = rollup.players.flatMap((row) => { + const playerId = playerIds.get(row.playerName); + + if (playerId === undefined) { + droppedRows += 1; + return []; + } + + return { + day, + playerId, + playTime: row.playTime, + }; + }); + + const serverRows = rollup.serverDays.map((row) => ({ + day, + gameServerId: row.gameServerId, + avgClients: row.avgClients, + maxClients: row.maxClients, + })); + + const mapRows = rollup.maps.map((row) => ({ + day, + mapId: row.mapId, + playTime: row.playTime, + playerCount: row.playerCount, + })); + + const gameTypeRows = rollup.gameTypes.flatMap((row) => { + const gameTypeId = gameTypeIds.get(row.gameTypeName); + + if (gameTypeId === undefined) { + droppedRows += 1; + return []; + } + + return { + day, + gameTypeId, + playTime: row.playTime, + playerCount: row.playerCount, + }; + }); + + const clanRows = rollup.clans.flatMap((row) => { + const clanId = clanIds.get(row.clanName); + + if (clanId === undefined) { + droppedRows += 1; + return []; + } + + return { + day, + clanId, + playTime: row.playTime, + playerCount: row.playerCount, + }; + }); + + // Inserting a day in key order keeps the btrees ~90% full instead of the + // ~70% random inserts leave; the reverse indexes lead with the same id. + playerRows.sort((a, b) => a.playerId - b.playerId); + serverRows.sort((a, b) => a.gameServerId - b.gameServerId); + mapRows.sort((a, b) => a.mapId - b.mapId); + gameTypeRows.sort((a, b) => a.gameTypeId - b.gameTypeId); + clanRows.sort((a, b) => a.clanId - b.clanId); + + await prisma.$transaction( + async (tx) => { + await tx.playerDay.deleteMany({ where: { day } }); + await tx.serverDay.deleteMany({ where: { day } }); + await tx.mapDay.deleteMany({ where: { day } }); + await tx.gameTypeDay.deleteMany({ where: { day } }); + await tx.clanDay.deleteMany({ where: { day } }); + + for (const rows of chunk(playerRows, INSERT_CHUNK_SIZE)) { + await tx.playerDay.createMany({ data: rows }); + } + for (const rows of chunk(serverRows, INSERT_CHUNK_SIZE)) { + await tx.serverDay.createMany({ data: rows }); + } + for (const rows of chunk(mapRows, INSERT_CHUNK_SIZE)) { + await tx.mapDay.createMany({ data: rows }); + } + for (const rows of chunk(gameTypeRows, INSERT_CHUNK_SIZE)) { + await tx.gameTypeDay.createMany({ data: rows }); + } + for (const rows of chunk(clanRows, INSERT_CHUNK_SIZE)) { + await tx.clanDay.createMany({ data: rows }); + } + }, + { timeout: minutesToMilliseconds(5), maxWait: minutesToMilliseconds(1) } + ); + + const dayLabel = formatUtcDay(day); + console.log( + `Rolled up ${dayLabel}: ${playerRows.length} PlayerDay, ` + + `${serverRows.length} ServerDay, ${mapRows.length} MapDay, ` + + `${gameTypeRows.length} GameTypeDay, ${clanRows.length} ClanDay` + + (droppedRows > 0 ? ` (${droppedRows} rows dropped: unresolved names)` : '') + ); +} diff --git a/apps/worker/src/snapshots.ts b/apps/worker/src/snapshots.ts new file mode 100644 index 0000000..d3d1f5c --- /dev/null +++ b/apps/worker/src/snapshots.ts @@ -0,0 +1,69 @@ +import { Prisma } from "@prisma/client"; +import { prisma } from "./prisma"; + +const snapshotSelect = { + id: true, + createdAt: true, + gameServerId: true, + mapId: true, + numPlayers: true, + maxPlayers: true, + numClients: true, + maxClients: true, + map: { + select: { + name: true, + gameTypeName: true, + }, + }, + clients: { + select: { + playerName: true, + clanName: true, + score: true, + country: true, + inGame: true, + }, + }, +} satisfies Prisma.GameServerSnapshotSelect; + +export type IteratedSnapshot = Prisma.GameServerSnapshotGetPayload<{ + select: typeof snapshotSelect; +}>; + +export async function* iterateSnapshots({ + from, + to, + batchSize, +}: { + from: Date; + to: Date; + batchSize: number; +}): AsyncGenerator { + let cursor = 0; + + for (;;) { + const snapshots = await prisma.gameServerSnapshot.findMany({ + where: { + createdAt: { gte: from, lt: to }, + id: { gt: cursor }, + }, + orderBy: { + id: 'asc', + }, + take: batchSize, + select: snapshotSelect, + }); + + if (snapshots.length === 0) { + return; + } + + yield* snapshots; + cursor = snapshots[snapshots.length - 1].id; + + if (snapshots.length < batchSize) { + return; + } + } +} diff --git a/apps/worker/src/workers/rollupBackfill.ts b/apps/worker/src/workers/rollupBackfill.ts new file mode 100644 index 0000000..3789632 --- /dev/null +++ b/apps/worker/src/workers/rollupBackfill.ts @@ -0,0 +1,129 @@ +import { GetObjectCommand, ListObjectsV2Command } from "@aws-sdk/client-s3"; +import { hoursToMilliseconds } from "date-fns"; +import { + RollupBackfillJobData, + S3_BUCKET, + SNAPSHOT_RETENTION_HOURS, + addUtcDays, + getEnvInt, + getS3Client, + parseUtcDay, + processRollupBackfillJobs, +} from "@teerank/teerank"; +import { SnapshotArchiveRow, decodeSnapshotRowsFromParquet } from "../parquet"; +import { DayAggregator, RollupSnapshot } from "../rollup/aggregateDay"; +import { writeDayRollup } from "../rollup/writeDayRollup"; +import { isDayRolledUp } from "./rollupDay"; + +const ROLLUP_TIME_BUDGET_MS = getEnvInt('ROLLUP_TIME_BUDGET_MS', 10 * 60 * 1000); + +async function listDayObjectKeys(day: string) { + const s3 = getS3Client(); + const keys: string[] = []; + let continuationToken: string | undefined; + + do { + const result = await s3.send(new ListObjectsV2Command({ + Bucket: S3_BUCKET, + Prefix: `snapshots/dt=${day}/`, + ContinuationToken: continuationToken, + })); + + for (const object of result.Contents ?? []) { + if (object.Key !== undefined) { + keys.push(object.Key); + } + } + + continuationToken = result.NextContinuationToken; + } while (continuationToken !== undefined); + + return keys.sort(); +} + +// Archive rows are flat client observations; regroup them per snapshot so the +// aggregation sees the same shape as the live rollup. +function addArchiveRows(aggregator: DayAggregator, rows: SnapshotArchiveRow[], day: Date, dayEnd: Date) { + const snapshots = new Map(); + + for (const row of rows) { + if (row.createdAt < day || row.createdAt >= dayEnd) { + continue; + } + + let snapshot = snapshots.get(row.snapshotId); + + if (snapshot === undefined) { + snapshot = { + createdAt: row.createdAt, + gameServerId: row.gameServerId, + mapId: row.mapId, + gameTypeName: row.gameTypeName, + numClients: row.numClients, + clients: [], + }; + snapshots.set(row.snapshotId, snapshot); + } + + if (row.playerName !== null) { + snapshot.clients.push({ + playerName: row.playerName, + clanName: row.clanName, + inGame: row.inGame ?? false, + }); + } + } + + for (const snapshot of snapshots.values()) { + aggregator.addSnapshot(snapshot); + } +} + +export async function rollupBackfill(data: RollupBackfillJobData) { + const startedAt = Date.now(); + const day = parseUtcDay(data.day); + const dayEnd = addUtcDays(day, 1); + + // The archive only holds all of a day's snapshots once the retention window + // has moved past the day's end. + if (dayEnd.getTime() + hoursToMilliseconds(SNAPSHOT_RETENTION_HOURS) > Date.now()) { + console.log(`Backfill for ${data.day} skipped: day may not be fully archived`); + return; + } + + if (await isDayRolledUp(day)) { + console.log(`Backfill for ${data.day} skipped: already rolled up`); + return; + } + + const keys = await listDayObjectKeys(data.day); + + if (keys.length === 0) { + console.log(`Backfill for ${data.day} skipped: no archive objects`); + return; + } + + const s3 = getS3Client(); + const aggregator = new DayAggregator(); + + for (const key of keys) { + if (Date.now() - startedAt > ROLLUP_TIME_BUDGET_MS) { + throw new Error(`Backfill for ${data.day} exceeded time budget, nothing written`); + } + + const object = await s3.send(new GetObjectCommand({ Bucket: S3_BUCKET, Key: key })); + const body = await object.Body?.transformToByteArray(); + + if (body === undefined) { + throw new Error(`Archive object ${key} has no body`); + } + + addArchiveRows(aggregator, decodeSnapshotRowsFromParquet(body), day, dayEnd); + } + + await writeDayRollup(day, aggregator.finalize()); +} + +export async function startRollupBackfillWorker() { + return processRollupBackfillJobs(rollupBackfill); +} diff --git a/apps/worker/src/workers/rollupDay.test.ts b/apps/worker/src/workers/rollupDay.test.ts new file mode 100644 index 0000000..5cc0fe6 --- /dev/null +++ b/apps/worker/src/workers/rollupDay.test.ts @@ -0,0 +1,210 @@ +import { addMinutes } from "date-fns"; +import { prismaMock } from "../../test/mockPrisma"; +import { DayAggregator, OBSERVATION_SECONDS, RollupSnapshot } from "../rollup/aggregateDay"; +import { rollupDay } from "./rollupDay"; + +const day = new Date('2026-08-19T00:00:00.000Z'); + +const newClient = (playerName: string, clanName: string | null = null, inGame = true) => ({ + playerName, + clanName, + inGame, +}); + +const newSnapshot = ( + minutes: number, + clients: RollupSnapshot['clients'], + overrides: Partial = {} +): RollupSnapshot => ({ + createdAt: addMinutes(day, minutes), + gameServerId: 1, + mapId: 1, + gameTypeName: 'CTF', + numClients: clients.length, + clients, + ...overrides, +}); + +describe('DayAggregator', () => { + test('one in-game player with a clan', () => { + const aggregator = new DayAggregator(); + aggregator.addSnapshot(newSnapshot(0, [newClient('player0', 'clan0')])); + + const rollup = aggregator.finalize(); + + expect(rollup.players).toEqual([{ playerName: 'player0', playTime: OBSERVATION_SECONDS }]); + expect(rollup.serverDays).toEqual([{ gameServerId: 1, avgClients: 1, maxClients: 1 }]); + expect(rollup.maps).toEqual([{ mapId: 1, playTime: OBSERVATION_SECONDS, playerCount: 1 }]); + expect(rollup.gameTypes).toEqual([ + { gameTypeName: 'CTF', playTime: OBSERVATION_SECONDS, playerCount: 1 }, + ]); + expect(rollup.clans).toEqual([ + { clanName: 'clan0', playTime: OBSERVATION_SECONDS, playerCount: 1 }, + ]); + }); + + test('spectators count for presence but not playtime', () => { + const aggregator = new DayAggregator(); + aggregator.addSnapshot(newSnapshot(0, [newClient('player0', 'clan0', false)])); + + const rollup = aggregator.finalize(); + + expect(rollup.players).toEqual([{ playerName: 'player0', playTime: 0 }]); + expect(rollup.maps[0]).toEqual({ mapId: 1, playTime: 0, playerCount: 1 }); + expect(rollup.clans[0]).toEqual({ clanName: 'clan0', playTime: 0, playerCount: 1 }); + // A spectator is still a connected client. + expect(rollup.serverDays[0].avgClients).toBe(1); + }); + + test('duplicated client names are counted once', () => { + const aggregator = new DayAggregator(); + aggregator.addSnapshot(newSnapshot(0, [newClient('player0'), newClient('player0')])); + + const rollup = aggregator.finalize(); + + expect(rollup.players).toEqual([{ playerName: 'player0', playTime: OBSERVATION_SECONDS }]); + }); + + test('servers with only empty snapshots are skipped', () => { + const aggregator = new DayAggregator(); + aggregator.addSnapshot(newSnapshot(0, [])); + aggregator.addSnapshot(newSnapshot(0, [newClient('player0')], { gameServerId: 2 })); + + const rollup = aggregator.finalize(); + + expect(rollup.serverDays).toEqual([{ gameServerId: 2, avgClients: 1, maxClients: 1 }]); + expect(rollup.maps).toEqual([{ mapId: 1, playTime: OBSERVATION_SECONDS, playerCount: 1 }]); + }); + + test('empty hours do not dilute the daily average', () => { + const aggregator = new DayAggregator(); + aggregator.addSnapshot(newSnapshot(0, [newClient('player0'), newClient('player1')])); + aggregator.addSnapshot(newSnapshot(5, [newClient('player0'), newClient('player1')])); + aggregator.addSnapshot(newSnapshot(60, [])); + aggregator.addSnapshot(newSnapshot(120, [newClient('player0'), newClient('player1'), newClient('player2'), newClient('player3')])); + + const rollup = aggregator.finalize(); + + // Hour 0 averages 2, hour 1 is empty and ignored, hour 2 averages 4. + expect(rollup.serverDays).toEqual([{ gameServerId: 1, avgClients: 3, maxClients: 4 }]); + }); + + test('distinct players across snapshots and gametypes', () => { + const aggregator = new DayAggregator(); + aggregator.addSnapshot(newSnapshot(0, [newClient('player0', 'clan0')])); + aggregator.addSnapshot( + newSnapshot(5, [newClient('player0', 'clan0'), newClient('player1')], { + gameServerId: 2, + mapId: 2, + gameTypeName: 'DM', + }) + ); + + const rollup = aggregator.finalize(); + + expect(rollup.players).toEqual([ + { playerName: 'player0', playTime: 2 * OBSERVATION_SECONDS }, + { playerName: 'player1', playTime: OBSERVATION_SECONDS }, + ]); + expect(rollup.gameTypes).toEqual([ + { gameTypeName: 'CTF', playTime: OBSERVATION_SECONDS, playerCount: 1 }, + { gameTypeName: 'DM', playTime: 2 * OBSERVATION_SECONDS, playerCount: 2 }, + ]); + expect(rollup.clans).toEqual([ + { clanName: 'clan0', playTime: 2 * OBSERVATION_SECONDS, playerCount: 1 }, + ]); + }); +}); + +describe('rollupDay', () => { + const mockLookups = () => { + prismaMock.player.findMany.mockResolvedValue([{ name: 'player0', id: 11 }] as never); + prismaMock.clan.findMany.mockResolvedValue([{ name: 'clan0', id: 21 }] as never); + prismaMock.gameType.findMany.mockResolvedValue([{ name: 'CTF', id: 31 }] as never); + prismaMock.$queryRawTyped.mockResolvedValue([] as never); + prismaMock.$transaction.mockImplementation(((callback: (tx: unknown) => unknown) => + callback(prismaMock)) as never); + }; + + test('skips a day that is already rolled up', async () => { + prismaMock.playerDay.findFirst.mockResolvedValue({ playerId: 1 } as never); + + await rollupDay({ day: '2026-08-19' }); + + expect(prismaMock.gameServerSnapshot.findMany).not.toHaveBeenCalled(); + expect(prismaMock.$transaction).not.toHaveBeenCalled(); + }); + + test('skips a day that is not over', async () => { + const today = new Date().toISOString().slice(0, 10); + + await rollupDay({ day: today }); + + expect(prismaMock.playerDay.findFirst).not.toHaveBeenCalled(); + expect(prismaMock.gameServerSnapshot.findMany).not.toHaveBeenCalled(); + }); + + test('aggregates a day and writes all five tables', async () => { + prismaMock.playerDay.findFirst.mockResolvedValue(null); + prismaMock.gameServerSnapshot.findMany.mockResolvedValue([ + { + id: 1, + createdAt: addMinutes(day, 30), + gameServerId: 1, + mapId: 1, + numClients: 1, + map: { gameTypeName: 'CTF' }, + clients: [{ playerName: 'player0', clanName: 'clan0', inGame: true }], + }, + ] as never); + mockLookups(); + + await rollupDay({ day: '2026-08-19' }); + + expect(prismaMock.playerDay.deleteMany).toHaveBeenCalledWith({ where: { day } }); + expect(prismaMock.serverDay.deleteMany).toHaveBeenCalledWith({ where: { day } }); + + expect(prismaMock.playerDay.createMany).toHaveBeenCalledWith({ + data: [{ day, playerId: 11, playTime: OBSERVATION_SECONDS }], + }); + expect(prismaMock.serverDay.createMany).toHaveBeenCalledWith({ + data: [{ day, gameServerId: 1, avgClients: 1, maxClients: 1 }], + }); + expect(prismaMock.mapDay.createMany).toHaveBeenCalledWith({ + data: [{ day, mapId: 1, playTime: OBSERVATION_SECONDS, playerCount: 1 }], + }); + expect(prismaMock.gameTypeDay.createMany).toHaveBeenCalledWith({ + data: [{ day, gameTypeId: 31, playTime: OBSERVATION_SECONDS, playerCount: 1 }], + }); + expect(prismaMock.clanDay.createMany).toHaveBeenCalledWith({ + data: [{ day, clanId: 21, playTime: OBSERVATION_SECONDS, playerCount: 1 }], + }); + }); + + test('players that no longer exist are dropped', async () => { + prismaMock.playerDay.findFirst.mockResolvedValue(null); + prismaMock.gameServerSnapshot.findMany.mockResolvedValue([ + { + id: 1, + createdAt: addMinutes(day, 30), + gameServerId: 1, + mapId: 1, + numClients: 2, + map: { gameTypeName: 'CTF' }, + clients: [ + { playerName: 'player0', clanName: 'clan0', inGame: true }, + { playerName: 'deleted', clanName: null, inGame: true }, + ], + }, + ] as never); + mockLookups(); + + await rollupDay({ day: '2026-08-19' }); + + const call = prismaMock.playerDay.createMany.mock.calls[0][0] as { + data: { playerId: number }[]; + }; + expect(call.data).toHaveLength(1); + expect(call.data[0].playerId).toBe(11); + }); +}); diff --git a/apps/worker/src/workers/rollupDay.ts b/apps/worker/src/workers/rollupDay.ts new file mode 100644 index 0000000..b67fafe --- /dev/null +++ b/apps/worker/src/workers/rollupDay.ts @@ -0,0 +1,66 @@ +import { + RollupDayJobData, + addUtcDays, + getEnvInt, + parseUtcDay, + processRollupDayJobs, +} from "@teerank/teerank"; +import { prisma } from "../prisma"; +import { iterateSnapshots } from "../snapshots"; +import { DayAggregator } from "../rollup/aggregateDay"; +import { writeDayRollup } from "../rollup/writeDayRollup"; + +const ROLLUP_BATCH_SIZE = getEnvInt('ROLLUP_BATCH_SIZE', 2000); +const ROLLUP_TIME_BUDGET_MS = getEnvInt('ROLLUP_TIME_BUDGET_MS', 10 * 60 * 1000); + +export async function isDayRolledUp(day: Date) { + const existing = await prisma.playerDay.findFirst({ + where: { day }, + select: { playerId: true }, + }); + + return existing !== null; +} + +export async function rollupDay(data: RollupDayJobData) { + const startedAt = Date.now(); + const day = parseUtcDay(data.day); + const dayEnd = addUtcDays(day, 1); + + if (dayEnd.getTime() > Date.now()) { + console.log(`Rollup for ${data.day} skipped: day is not over`); + return; + } + + if (await isDayRolledUp(day)) { + console.log(`Rollup for ${data.day} skipped: already rolled up`); + return; + } + + const aggregator = new DayAggregator(); + + for await (const snapshot of iterateSnapshots({ + from: day, + to: dayEnd, + batchSize: ROLLUP_BATCH_SIZE, + })) { + if (Date.now() - startedAt > ROLLUP_TIME_BUDGET_MS) { + throw new Error(`Rollup for ${data.day} exceeded time budget, nothing written`); + } + + aggregator.addSnapshot({ + createdAt: snapshot.createdAt, + gameServerId: snapshot.gameServerId, + mapId: snapshot.mapId, + gameTypeName: snapshot.map.gameTypeName, + numClients: snapshot.numClients, + clients: snapshot.clients, + }); + } + + await writeDayRollup(day, aggregator.finalize()); +} + +export async function startRollupDayWorker() { + return processRollupDayJobs(rollupDay); +} diff --git a/libs/prisma/prisma/migrations/20260820000000_rollup_tables/migration.sql b/libs/prisma/prisma/migrations/20260820000000_rollup_tables/migration.sql new file mode 100644 index 0000000..362f29a --- /dev/null +++ b/libs/prisma/prisma/migrations/20260820000000_rollup_tables/migration.sql @@ -0,0 +1,85 @@ +-- Rollup tables are partitioned by range on day so retention changes are +-- DETACH+DROP instead of delete-and-vacuum. Postgres requires the partition +-- key in the primary key, which every key below satisfies. The rollup workers +-- create future partitions ahead of time (see apps/worker/src/rollup); the +-- current and next month are created here so the first job has somewhere to +-- write. + +-- CreateTable +CREATE TABLE "PlayerDay" ( + "day" DATE NOT NULL, + "playerId" INTEGER NOT NULL, + "playTime" INTEGER NOT NULL, + + CONSTRAINT "PlayerDay_pkey" PRIMARY KEY ("day","playerId") +) PARTITION BY RANGE ("day"); + +-- CreateIndex +CREATE INDEX "PlayerDay_playerId_day_idx" ON "PlayerDay"("playerId", "day"); + +-- CreateTable +CREATE TABLE "ServerDay" ( + "day" DATE NOT NULL, + "gameServerId" INTEGER NOT NULL, + "avgClients" SMALLINT NOT NULL, + "maxClients" SMALLINT NOT NULL, + + CONSTRAINT "ServerDay_pkey" PRIMARY KEY ("gameServerId","day") +) PARTITION BY RANGE ("day"); + +-- CreateTable +CREATE TABLE "MapDay" ( + "day" DATE NOT NULL, + "mapId" INTEGER NOT NULL, + "playTime" INTEGER NOT NULL, + "playerCount" INTEGER NOT NULL, + + CONSTRAINT "MapDay_pkey" PRIMARY KEY ("mapId","day") +) PARTITION BY RANGE ("day"); + +-- CreateTable +CREATE TABLE "GameTypeDay" ( + "day" DATE NOT NULL, + "gameTypeId" INTEGER NOT NULL, + "playTime" INTEGER NOT NULL, + "playerCount" INTEGER NOT NULL, + + CONSTRAINT "GameTypeDay_pkey" PRIMARY KEY ("gameTypeId","day") +) PARTITION BY RANGE ("day"); + +-- CreateTable +CREATE TABLE "ClanDay" ( + "day" DATE NOT NULL, + "clanId" INTEGER NOT NULL, + "playTime" INTEGER NOT NULL, + "playerCount" SMALLINT NOT NULL, + + CONSTRAINT "ClanDay_pkey" PRIMARY KEY ("clanId","day") +) PARTITION BY RANGE ("day"); + +-- CreateFunction +-- DDL cannot be a prepared statement, so partition creation lives in a +-- function the rollup workers call through typedSql. +CREATE FUNCTION create_rollup_partition(table_name text, from_day date, to_day date) RETURNS text AS $$ +DECLARE + partition_name text := table_name || '_' || to_char(from_day, 'YYYY_MM'); +BEGIN + EXECUTE format( + 'CREATE TABLE IF NOT EXISTS %I PARTITION OF %I FOR VALUES FROM (%L) TO (%L)', + partition_name, table_name, from_day, to_day + ); + RETURN partition_name; +END; +$$ LANGUAGE plpgsql; + +-- CreatePartitions +CREATE TABLE "PlayerDay_2026_08" PARTITION OF "PlayerDay" FOR VALUES FROM ('2026-08-01') TO ('2026-09-01'); +CREATE TABLE "PlayerDay_2026_09" PARTITION OF "PlayerDay" FOR VALUES FROM ('2026-09-01') TO ('2026-10-01'); +CREATE TABLE "ServerDay_2026_08" PARTITION OF "ServerDay" FOR VALUES FROM ('2026-08-01') TO ('2026-09-01'); +CREATE TABLE "ServerDay_2026_09" PARTITION OF "ServerDay" FOR VALUES FROM ('2026-09-01') TO ('2026-10-01'); +CREATE TABLE "MapDay_2026_08" PARTITION OF "MapDay" FOR VALUES FROM ('2026-08-01') TO ('2026-09-01'); +CREATE TABLE "MapDay_2026_09" PARTITION OF "MapDay" FOR VALUES FROM ('2026-09-01') TO ('2026-10-01'); +CREATE TABLE "GameTypeDay_2026_08" PARTITION OF "GameTypeDay" FOR VALUES FROM ('2026-08-01') TO ('2026-09-01'); +CREATE TABLE "GameTypeDay_2026_09" PARTITION OF "GameTypeDay" FOR VALUES FROM ('2026-09-01') TO ('2026-10-01'); +CREATE TABLE "ClanDay_2026_08" PARTITION OF "ClanDay" FOR VALUES FROM ('2026-08-01') TO ('2026-09-01'); +CREATE TABLE "ClanDay_2026_09" PARTITION OF "ClanDay" FOR VALUES FROM ('2026-09-01') TO ('2026-10-01'); diff --git a/libs/prisma/prisma/schema.prisma b/libs/prisma/prisma/schema.prisma index 5e6f168..2d0c901 100644 --- a/libs/prisma/prisma/schema.prisma +++ b/libs/prisma/prisma/schema.prisma @@ -319,3 +319,53 @@ model GlobalCounts { clanCount Int @default(0) gameServerCount Int @default(0) } + +// Daily/hourly rollups of the snapshot history. No foreign keys: ids are +// resolved at write time, and rows must survive source-row deletion. The +// tables are partitioned by range on day/hour (see the migration); every +// primary key leads with the partition key. + +model PlayerDay { + day DateTime @db.Date + playerId Int + playTime Int // seconds + + @@id([day, playerId]) + @@index([playerId, day]) +} + +model ServerDay { + day DateTime @db.Date + gameServerId Int + avgClients Int @db.SmallInt + maxClients Int @db.SmallInt + + @@id([gameServerId, day]) +} + +model MapDay { + day DateTime @db.Date + mapId Int + playTime Int + playerCount Int + + @@id([mapId, day]) +} + +model GameTypeDay { + day DateTime @db.Date + gameTypeId Int + playTime Int + playerCount Int + + @@id([gameTypeId, day]) +} + +model ClanDay { + day DateTime @db.Date + clanId Int + playTime Int + playerCount Int @db.SmallInt + + @@id([clanId, day]) +} diff --git a/libs/prisma/prisma/sql/createRollupPartition.sql b/libs/prisma/prisma/sql/createRollupPartition.sql new file mode 100644 index 0000000..cef9eaa --- /dev/null +++ b/libs/prisma/prisma/sql/createRollupPartition.sql @@ -0,0 +1 @@ +SELECT create_rollup_partition($1, $2::date, $3::date) AS "partition"; diff --git a/libs/teerank/src/index.ts b/libs/teerank/src/index.ts index 3d2908b..1233235 100644 --- a/libs/teerank/src/index.ts +++ b/libs/teerank/src/index.ts @@ -1,4 +1,5 @@ export * from './lib/utils'; +export * from './lib/date'; export * from './lib/schemas'; export * from './lib/bullmq'; export * from './lib/redisConfig'; diff --git a/libs/teerank/src/lib/bullmq/cleanAllQueues.ts b/libs/teerank/src/lib/bullmq/cleanAllQueues.ts index 5300375..7602c59 100644 --- a/libs/teerank/src/lib/bullmq/cleanAllQueues.ts +++ b/libs/teerank/src/lib/bullmq/cleanAllQueues.ts @@ -7,6 +7,8 @@ import { cleanUpdateGlobalCountsQueue } from "./queueUpdateGlobalCounts"; import { cleanUpdatePlayTimeQueue } from "./queueUpdatePlayTime"; import { cleanRankPlayerQueue } from "./queueRankPlayer"; import { cleanArchiveSnapshotsQueue } from "./queueArchiveSnapshots"; +import { cleanRollupDayQueue } from "./queueRollupDay"; +import { cleanRollupBackfillQueue } from "./queueRollupBackfill"; export async function cleanAllQueues() { await Promise.all([ @@ -19,5 +21,7 @@ export async function cleanAllQueues() { cleanUpdatePlayTimeQueue(), cleanRankPlayerQueue(), cleanArchiveSnapshotsQueue(), + cleanRollupDayQueue(), + cleanRollupBackfillQueue(), ]); } diff --git a/libs/teerank/src/lib/bullmq/index.ts b/libs/teerank/src/lib/bullmq/index.ts index 6108830..523fbbd 100644 --- a/libs/teerank/src/lib/bullmq/index.ts +++ b/libs/teerank/src/lib/bullmq/index.ts @@ -9,3 +9,5 @@ export * from './queueMapCount'; export * from './queueFillClanActivePlayerCount'; export * from './queueUpdateGlobalCounts'; export * from './queueArchiveSnapshots'; +export * from './queueRollupDay'; +export * from './queueRollupBackfill'; diff --git a/libs/teerank/src/lib/bullmq/queueRollupBackfill.ts b/libs/teerank/src/lib/bullmq/queueRollupBackfill.ts new file mode 100644 index 0000000..b6277a2 --- /dev/null +++ b/libs/teerank/src/lib/bullmq/queueRollupBackfill.ts @@ -0,0 +1,57 @@ +import { Job, Queue, Worker } from "bullmq"; +import { bullmqConnection } from "./config"; +import { z } from "zod"; +import { hoursToSeconds } from "date-fns"; +import { utcDaySchema } from "../schemas"; + +let rollupBackfillQueue: Queue | null = null; + +const QUEUE_NAME_ROLLUP_BACKFILL = 'rollup-backfill'; + +function getQueueRollupBackfill() { + rollupBackfillQueue ??= new Queue(QUEUE_NAME_ROLLUP_BACKFILL, { connection: bullmqConnection }); + return rollupBackfillQueue; +} + +const schema = z.object({ + day: utcDaySchema, +}); + +export type RollupBackfillJobData = z.infer; + +export async function scheduleRollupBackfill(data: RollupBackfillJobData) { + const queue = getQueueRollupBackfill(); + await queue.add(`rollup-backfill-${data.day}`, data, { + deduplication: { + id: `rollup-backfill-${data.day}`, + } + }); +} + +export async function processRollupBackfillJobs(processor: (data: RollupBackfillJobData) => Promise) { + const jobProcessor = async (job: Job) => { + const data = schema.parse(job.data); + await processor(data); + } + + return new Worker(QUEUE_NAME_ROLLUP_BACKFILL, jobProcessor, { + connection: bullmqConnection, + concurrency: 1, + removeOnComplete: { + age: hoursToSeconds(6), + }, + removeOnFail: { + count: 1000, + } + }); +} + +export async function cleanRollupBackfillQueue() { + await getQueueRollupBackfill().obliterate({ + force: true, + }); +} + +export async function getRollupBackfillFailedCount() { + return getQueueRollupBackfill().getFailedCount(); +} diff --git a/libs/teerank/src/lib/bullmq/queueRollupDay.ts b/libs/teerank/src/lib/bullmq/queueRollupDay.ts new file mode 100644 index 0000000..c549a97 --- /dev/null +++ b/libs/teerank/src/lib/bullmq/queueRollupDay.ts @@ -0,0 +1,57 @@ +import { Job, Queue, Worker } from "bullmq"; +import { bullmqConnection } from "./config"; +import { z } from "zod"; +import { hoursToSeconds } from "date-fns"; +import { utcDaySchema } from "../schemas"; + +let rollupDayQueue: Queue | null = null; + +const QUEUE_NAME_ROLLUP_DAY = 'rollup-day'; + +function getQueueRollupDay() { + rollupDayQueue ??= new Queue(QUEUE_NAME_ROLLUP_DAY, { connection: bullmqConnection }); + return rollupDayQueue; +} + +const schema = z.object({ + day: utcDaySchema, +}); + +export type RollupDayJobData = z.infer; + +export async function scheduleRollupDay(data: RollupDayJobData) { + const queue = getQueueRollupDay(); + await queue.add(`rollup-day-${data.day}`, data, { + deduplication: { + id: `rollup-day-${data.day}`, + } + }); +} + +export async function processRollupDayJobs(processor: (data: RollupDayJobData) => Promise) { + const jobProcessor = async (job: Job) => { + const data = schema.parse(job.data); + await processor(data); + } + + return new Worker(QUEUE_NAME_ROLLUP_DAY, jobProcessor, { + connection: bullmqConnection, + concurrency: 1, + removeOnComplete: { + age: hoursToSeconds(6), + }, + removeOnFail: { + count: 1000, + } + }); +} + +export async function cleanRollupDayQueue() { + await getQueueRollupDay().obliterate({ + force: true, + }); +} + +export async function getRollupDayFailedCount() { + return getQueueRollupDay().getFailedCount(); +} diff --git a/libs/teerank/src/lib/date.ts b/libs/teerank/src/lib/date.ts new file mode 100644 index 0000000..3a93260 --- /dev/null +++ b/libs/teerank/src/lib/date.ts @@ -0,0 +1,29 @@ +import { hoursToMilliseconds } from 'date-fns'; + +export const DAY_MS = hoursToMilliseconds(24); + +export function parseUtcDay(day: string) { + return new Date(`${day}T00:00:00.000Z`); +} + +export function formatUtcDay(date: Date) { + return date.toISOString().slice(0, 10); +} + +export function addUtcDays(date: Date, days: number) { + return new Date(date.getTime() + days * DAY_MS); +} + +export function startOfUtcDay(date: Date) { + return new Date(Math.floor(date.getTime() / DAY_MS) * DAY_MS); +} + +export function utcYesterday() { + return addUtcDays(startOfUtcDay(new Date()), -1); +} + +export function eachUtcDay(from: Date, to: Date) { + const count = Math.round((to.getTime() - from.getTime()) / DAY_MS) + 1; + + return Array.from({ length: count }, (_, index) => addUtcDays(from, index)); +} diff --git a/libs/teerank/src/lib/schemas.ts b/libs/teerank/src/lib/schemas.ts index a177a9c..d8d8bc8 100644 --- a/libs/teerank/src/lib/schemas.ts +++ b/libs/teerank/src/lib/schemas.ts @@ -1,5 +1,7 @@ import { z } from 'zod'; +export const utcDaySchema = z.string().regex(/^\d{4}-\d{2}-\d{2}$/); + export const indexPlayerSchema = z.object({ name: z.string(), clanName: z.string().nullable(), diff --git a/tsconfig.base.json b/tsconfig.base.json index 11b62a5..1c1c187 100644 --- a/tsconfig.base.json +++ b/tsconfig.base.json @@ -16,7 +16,8 @@ "baseUrl": ".", "paths": { "@teerank/prisma": ["libs/prisma/src/index.ts"], - "@teerank/teerank": ["libs/teerank/src/index.ts"] + "@teerank/teerank": ["libs/teerank/src/index.ts"], + "@teerank/teerank/date": ["libs/teerank/src/lib/date.ts"] } }, "exclude": ["node_modules", "tmp"]