65 lines
5.0 KiB
TypeScript
65 lines
5.0 KiB
TypeScript
import { randomUUID } from 'node:crypto';
|
|
import { MongoServerError, type Db } from 'mongodb';
|
|
import { ACTIVE_WINDOW_MS, RETENTION_MS, VISIT_SESSION_MS } from '../../constants/analytics.js';
|
|
import type { AnalyticsEvent, Presence, VisitMetadata, VisitSession } from '../shared/analytics-types.js';
|
|
import type { PlayerDocument } from './stats-store.js';
|
|
import { kstDay } from './analytics-report.js';
|
|
export interface AnalyticsStore {
|
|
db: Db;
|
|
touch(playerId: string, now: Date, visible: boolean, metadata: VisitMetadata, metadataFinal?: boolean): Promise<VisitSession>;
|
|
insert(event: AnalyticsEvent): Promise<void>;
|
|
sample(now: Date, presence: Presence, environment: string, quality?: { restartedAt: Date; failures: number; dropped: number; rateLimited: number }): Promise<void>;
|
|
cleanup(now: Date): Promise<void>;
|
|
}
|
|
export async function createAnalyticsStore(db: Db): Promise<AnalyticsStore> {
|
|
const players = db.collection<PlayerDocument & { visit?: VisitSession }>('players');
|
|
const sessions = db.collection<VisitSession & { _id: string }>('analytics_sessions');
|
|
const events = db.collection<AnalyticsEvent>('analytics_events');
|
|
const samples = db.collection('analytics_presence_samples');
|
|
for (const collection of [sessions, events, samples]) await collection.createIndex({ expiresAt: 1 }, { expireAfterSeconds: 0 });
|
|
await sessions.createIndex({ playerId: 1, lastAt: -1 });
|
|
await sessions.createIndex({ sessionId: 1 }, { unique: true });
|
|
for (const field of ['eventType', 'playerId', 'sessionId']) await events.createIndex({ [field]: 1, occurredAt: 1 });
|
|
await samples.createIndex({ occurredAt: 1 });
|
|
await players.createIndex({ 'visit.expiresAt': 1 });
|
|
const cleanup = async (now: Date) => { await players.updateMany({ 'visit.expiresAt': { $lte: now } }, { $unset: { visit: '' } }); };
|
|
await cleanup(new Date());
|
|
return {
|
|
db,
|
|
cleanup,
|
|
async touch(playerId, now, visible, metadata, metadataFinal = false) {
|
|
// The single player document atomically decides the session ID and activity delta.
|
|
// Snapshot writes to analytics_sessions never decide or increment counters independently.
|
|
const expired = { $or: [{ $eq: [{ $type: '$visit' }, 'missing'] }, { $lte: ['$visit.lastAt', new Date(+now - VISIT_SESSION_MS)] }] };
|
|
const fresh: VisitSession = { sessionId: randomUUID(), playerId, startedAt: now, lastAt: now, lastSignalAt: now, visible,
|
|
activeMs: 0, activeDays: visible ? [kstDay(now)] : [], metadata, metadataReady: metadataFinal, expiresAt: new Date(+now + RETENTION_MS) };
|
|
const delta = { $cond: [{ $and: [visible, '$visit.visible', { $lte: [{ $subtract: [now, '$visit.lastSignalAt'] }, ACTIVE_WINDOW_MS] }, { $gte: [now, '$visit.lastSignalAt'] }] }, { $subtract: [now, '$visit.lastSignalAt'] }, 0] };
|
|
const doc = await players.findOneAndUpdate({ _id: playerId }, [{ $set: {
|
|
visit: { $cond: [expired, { $literal: fresh }, { $mergeObjects: ['$visit', {
|
|
visible, lastSignalAt: now, lastAt: visible ? now : '$visit.lastAt',
|
|
activeMs: { $add: ['$visit.activeMs', delta] }, expiresAt: new Date(+now + RETENTION_MS),
|
|
activeDays: { $setUnion: [{ $filter: { input: { $ifNull: ['$visit.activeDays', []] }, as: 'day', cond: { $gte: ['$$day', kstDay(new Date(+now - RETENTION_MS))] } } }, visible ? [kstDay(now)] : []] },
|
|
metadata: { $cond: [{ $and: [metadataFinal, { $ne: ['$visit.metadataReady', true] }] }, { $literal: metadata }, '$visit.metadata'] },
|
|
metadataReady: { $or: [metadataFinal, { $eq: ['$visit.metadataReady', true] }] },
|
|
}] }] },
|
|
} }], { returnDocument: 'after' });
|
|
if (!doc?.visit) throw new Error('UNKNOWN_PLAYER');
|
|
const visit = doc.visit;
|
|
await sessions.updateOne({ _id: visit.sessionId }, {
|
|
$setOnInsert: { sessionId: visit.sessionId, playerId, startedAt: visit.startedAt, ...(!visit.metadataReady ? { metadata: visit.metadata } : {}) },
|
|
$max: { lastAt: visit.lastAt, lastSignalAt: visit.lastSignalAt, activeMs: visit.activeMs, expiresAt: visit.expiresAt },
|
|
$set: { visible: visit.visible, ...(visit.metadataReady ? { metadata: visit.metadata, metadataReady: true } : {}) },
|
|
$addToSet: { activeDays: { $each: visit.activeDays } },
|
|
}, { upsert: true });
|
|
return visit;
|
|
},
|
|
async insert(event) {
|
|
try { await events.insertOne({ ...event, expiresAt: new Date(+event.occurredAt + RETENTION_MS) }); }
|
|
catch (error) { if (!(error instanceof MongoServerError) || error.code !== 11000) throw error; }
|
|
await db.collection<{ _id: string; firstCollectedAt: Date }>('analytics_metadata').updateOne({ _id: event.environment }, { $min: { firstCollectedAt: event.occurredAt } }, { upsert: true });
|
|
if (event.eventType === 'page_view' && event.playerId) await players.updateOne({ _id: event.playerId }, { $min: { firstVisitAt: event.occurredAt } });
|
|
},
|
|
async sample(now, presence, environment, quality) { await samples.insertOne({ occurredAt: now, active: presence.active, source: presence.source, environment, ...(quality ?? {}), expiresAt: new Date(+now + RETENTION_MS) }); },
|
|
};
|
|
}
|