feat: add real-time stats page with live visitor tracking and page view analytics

convex/_generated/api.d.ts

@@ -8,10 +8,12 @@
  * @module
  */
 
+import type * as crons from "../crons.js";
 import type * as http from "../http.js";
 import type * as pages from "../pages.js";
 import type * as posts from "../posts.js";
 import type * as rss from "../rss.js";
+import type * as stats from "../stats.js";
 
 import type {
   ApiFromModules,
@@ -20,10 +22,12 @@ import type {
 } from "convex/server";
 
 declare const fullApi: ApiFromModules<{
+  crons: typeof crons;
   http: typeof http;
   pages: typeof pages;
   posts: typeof posts;
   rss: typeof rss;
+  stats: typeof stats;
 }>;
 
 /**

convex/crons.ts

@@ -0,0 +1,15 @@
+import { cronJobs } from "convex/server";
+import { internal } from "./_generated/api";
+
+const crons = cronJobs();
+
+// Clean up stale sessions every 5 minutes
+crons.interval(
+  "cleanup stale sessions",
+  { minutes: 5 },
+  internal.stats.cleanupStaleSessions,
+  {}
+);
+
+export default crons;
+

convex/schema.ts

@@ -50,4 +50,24 @@ export default defineSchema({
     key: v.string(),
     value: v.any(),
   }).index("by_key", ["key"]),
+
+  // Page view events for analytics (event records pattern)
+  pageViews: defineTable({
+    path: v.string(),
+    pageType: v.string(), // "blog" | "page" | "home" | "stats"
+    sessionId: v.string(),
+    timestamp: v.number(),
+  })
+    .index("by_path", ["path"])
+    .index("by_timestamp", ["timestamp"])
+    .index("by_session_path", ["sessionId", "path"]),
+
+  // Active sessions for real-time visitor tracking
+  activeSessions: defineTable({
+    sessionId: v.string(),
+    currentPath: v.string(),
+    lastSeen: v.number(),
+  })
+    .index("by_sessionId", ["sessionId"])
+    .index("by_lastSeen", ["lastSeen"]),
 });

convex/stats.ts

@@ -0,0 +1,227 @@
+import { query, mutation, internalMutation } from "./_generated/server";
+import { v } from "convex/values";
+
+// Deduplication window: 30 minutes in milliseconds
+const DEDUP_WINDOW_MS = 30 * 60 * 1000;
+
+// Session timeout: 2 minutes in milliseconds
+const SESSION_TIMEOUT_MS = 2 * 60 * 1000;
+
+/**
+ * Record a page view event.
+ * Idempotent: same session viewing same path within 30min = 1 view.
+ */
+export const recordPageView = mutation({
+  args: {
+    path: v.string(),
+    pageType: v.string(),
+    sessionId: v.string(),
+  },
+  returns: v.null(),
+  handler: async (ctx, args) => {
+    const now = Date.now();
+    const dedupCutoff = now - DEDUP_WINDOW_MS;
+
+    // Check for recent view from same session on same path
+    const recentView = await ctx.db
+      .query("pageViews")
+      .withIndex("by_session_path", (q) =>
+        q.eq("sessionId", args.sessionId).eq("path", args.path)
+      )
+      .order("desc")
+      .first();
+
+    // Early return if already viewed within dedup window
+    if (recentView && recentView.timestamp > dedupCutoff) {
+      return null;
+    }
+
+    // Insert new view event
+    await ctx.db.insert("pageViews", {
+      path: args.path,
+      pageType: args.pageType,
+      sessionId: args.sessionId,
+      timestamp: now,
+    });
+
+    return null;
+  },
+});
+
+/**
+ * Update active session heartbeat.
+ * Creates or updates session with current path and timestamp.
+ */
+export const heartbeat = mutation({
+  args: {
+    sessionId: v.string(),
+    currentPath: v.string(),
+  },
+  returns: v.null(),
+  handler: async (ctx, args) => {
+    const now = Date.now();
+
+    // Find existing session by sessionId
+    const existingSession = await ctx.db
+      .query("activeSessions")
+      .withIndex("by_sessionId", (q) => q.eq("sessionId", args.sessionId))
+      .first();
+
+    if (existingSession) {
+      // Update existing session
+      await ctx.db.patch(existingSession._id, {
+        currentPath: args.currentPath,
+        lastSeen: now,
+      });
+    } else {
+      // Create new session
+      await ctx.db.insert("activeSessions", {
+        sessionId: args.sessionId,
+        currentPath: args.currentPath,
+        lastSeen: now,
+      });
+    }
+
+    return null;
+  },
+});
+
+/**
+ * Get all stats for the stats page.
+ * Real-time subscription via useQuery.
+ */
+export const getStats = query({
+  args: {},
+  returns: v.object({
+    activeVisitors: v.number(),
+    activeByPath: v.array(
+      v.object({
+        path: v.string(),
+        count: v.number(),
+      })
+    ),
+    totalPageViews: v.number(),
+    uniqueVisitors: v.number(),
+    publishedPosts: v.number(),
+    publishedPages: v.number(),
+    pageStats: v.array(
+      v.object({
+        path: v.string(),
+        title: v.string(),
+        pageType: v.string(),
+        views: v.number(),
+      })
+    ),
+  }),
+  handler: async (ctx) => {
+    const now = Date.now();
+    const sessionCutoff = now - SESSION_TIMEOUT_MS;
+
+    // Get active sessions (heartbeat within last 2 minutes)
+    const activeSessions = await ctx.db
+      .query("activeSessions")
+      .withIndex("by_lastSeen", (q) => q.gt("lastSeen", sessionCutoff))
+      .collect();
+
+    // Count active visitors by path
+    const activeByPathMap: Record<string, number> = {};
+    for (const session of activeSessions) {
+      activeByPathMap[session.currentPath] =
+        (activeByPathMap[session.currentPath] || 0) + 1;
+    }
+    const activeByPath = Object.entries(activeByPathMap)
+      .map(([path, count]) => ({ path, count }))
+      .sort((a, b) => b.count - a.count);
+
+    // Get all page views
+    const allViews = await ctx.db.query("pageViews").collect();
+
+    // Aggregate views by path and count unique sessions
+    const viewsByPath: Record<string, number> = {};
+    const uniqueSessions = new Set<string>();
+
+    for (const view of allViews) {
+      viewsByPath[view.path] = (viewsByPath[view.path] || 0) + 1;
+      uniqueSessions.add(view.sessionId);
+    }
+
+    // Get published posts and pages for titles
+    const posts = await ctx.db
+      .query("posts")
+      .withIndex("by_published", (q) => q.eq("published", true))
+      .collect();
+
+    const pages = await ctx.db
+      .query("pages")
+      .withIndex("by_published", (q) => q.eq("published", true))
+      .collect();
+
+    // Build page stats array with titles
+    const pageStats = Object.entries(viewsByPath)
+      .map(([path, views]) => {
+        // Match path to post or page
+        const slug = path.startsWith("/") ? path.slice(1) : path;
+        const post = posts.find((p) => p.slug === slug);
+        const page = pages.find((p) => p.slug === slug);
+
+        let title = path;
+        let pageType = "other";
+
+        if (path === "/" || path === "") {
+          title = "Home";
+          pageType = "home";
+        } else if (path === "/stats") {
+          title = "Stats";
+          pageType = "stats";
+        } else if (post) {
+          title = post.title;
+          pageType = "blog";
+        } else if (page) {
+          title = page.title;
+          pageType = "page";
+        }
+
+        return {
+          path,
+          title,
+          pageType,
+          views,
+        };
+      })
+      .sort((a, b) => b.views - a.views);
+
+    return {
+      activeVisitors: activeSessions.length,
+      activeByPath,
+      totalPageViews: allViews.length,
+      uniqueVisitors: uniqueSessions.size,
+      publishedPosts: posts.length,
+      publishedPages: pages.length,
+      pageStats,
+    };
+  },
+});
+
+/**
+ * Internal mutation to clean up stale sessions.
+ * Called by cron job every 5 minutes.
+ */
+export const cleanupStaleSessions = internalMutation({
+  args: {},
+  returns: v.number(),
+  handler: async (ctx) => {
+    const cutoff = Date.now() - SESSION_TIMEOUT_MS;
+
+    // Get all stale sessions
+    const staleSessions = await ctx.db
+      .query("activeSessions")
+      .withIndex("by_lastSeen", (q) => q.lt("lastSeen", cutoff))
+      .collect();
+
+    // Delete in parallel
+    await Promise.all(staleSessions.map((session) => ctx.db.delete(session._id)));
+
+    return staleSessions.length;
+  },
+});
+