fix(convex): bound skill health reads

convex/leaderboards.test.ts

@@ -1,12 +1,20 @@
 /* @vitest-environment node */
 import { describe, expect, it, vi } from "vitest";
-import { rebuildTrendingLeaderboardInternal } from "./leaderboards";
+import {
+  rebuildTrendingLeaderboardAction,
+  rebuildTrendingLeaderboardInternal,
+} from "./leaderboards";
 
-const handler = (
+const mutationHandler = (
   rebuildTrendingLeaderboardInternal as unknown as {
     _handler: (ctx: unknown, args: { limit?: number }) => Promise<unknown>;
   }
 )._handler;
+const actionHandler = (
+  rebuildTrendingLeaderboardAction as unknown as {
+    _handler: (ctx: unknown, args: { limit?: number }) => Promise<unknown>;
+  }
+)._handler;
 
 describe("leaderboards.rebuildTrendingLeaderboardInternal", () => {
   it("schedules the action-based rebuild instead of reading daily stats inline", async () => {
@@ -30,11 +38,44 @@ describe("leaderboards.rebuildTrendingLeaderboardInternal", () => {
       },
     } as never;
 
-    const result = await handler(ctx, { limit: 500 });
+    const result = await mutationHandler(ctx, { limit: 500 });
 
     expect(runAfter).toHaveBeenCalledTimes(1);
     expect(runAfter.mock.calls[0]?.[0]).toBe(0);
     expect(runAfter.mock.calls[0]?.[2]).toEqual({ limit: 200 });
     expect(result).toEqual({ ok: true, count: 0, scheduled: true });
   });
+
+  it("rebuild action pages daily stats instead of collecting a whole day", async () => {
+    const runQuery = vi.fn(async (_ref: unknown, args: Record<string, unknown>) => {
+      if (Array.isArray(args.entries)) return args.entries;
+      return {
+        rows: [
+          {
+            skillId: "skills:one",
+            installs: 1,
+            downloads: 2,
+          },
+        ],
+        isDone: true,
+        continueCursor: "",
+      };
+    });
+    const runMutation = vi.fn(async () => ({ ok: true }));
+
+    const result = await actionHandler(
+      {
+        runQuery,
+        runMutation,
+      },
+      { limit: 5 },
+    );
+
+    expect(result).toEqual({ ok: true, count: 1 });
+    expect(runQuery).toHaveBeenCalledWith(
+      expect.anything(),
+      expect.objectContaining({ cursor: null, limit: 5000 }),
+    );
+    expect(runMutation).toHaveBeenCalledTimes(2);
+  });
 });

convex/leaderboards.ts

@@ -1,10 +1,10 @@
 import { v } from "convex/values";
 import { internal } from "./_generated/api";
+import type { Id } from "./_generated/dataModel";
 import { internalAction, internalMutation, internalQuery } from "./functions";
 import {
-  buildTrendingEntriesFromDailyRows,
+  compareTrendingEntries,
   getTrendingRange,
-  queryDailyStats,
   takeTopNonSuspiciousTrendingEntries,
   takeTopTrendingEntries,
   TRENDING_LEADERBOARD_KIND,
@@ -13,17 +13,37 @@ import {
 
 const MAX_TRENDING_LIMIT = 200;
 const KEEP_LEADERBOARD_ENTRIES = 3;
+const DAILY_STATS_PAGE_SIZE = 5_000;
 
 // ---------------------------------------------------------------------------
 // Action → Query → Mutation pattern (avoids 32K document-read limit)
 // ---------------------------------------------------------------------------
 
-/** Reads a single day's skillDailyStats in its own query transaction. */
-export const getDailyStats = internalQuery({
-  args: { day: v.number() },
-  handler: async (ctx, { day }) => {
-    const rows = await queryDailyStats(ctx, day);
-    return rows.map((r) => ({ skillId: r.skillId, installs: r.installs, downloads: r.downloads }));
+/** Reads one page of a single day's skillDailyStats in its own query transaction. */
+export const getDailyStatsPage = internalQuery({
+  args: {
+    day: v.number(),
+    cursor: v.union(v.string(), v.null()),
+    limit: v.optional(v.number()),
+  },
+  handler: async (ctx, { day, cursor, limit }) => {
+    const page = await ctx.db
+      .query("skillDailyStats")
+      .withIndex("by_day", (q) => q.eq("day", day))
+      .paginate({
+        cursor,
+        numItems: Math.min(limit ?? DAILY_STATS_PAGE_SIZE, DAILY_STATS_PAGE_SIZE),
+      });
+
+    return {
+      rows: page.page.map((r) => ({
+        skillId: r.skillId,
+        installs: r.installs,
+        downloads: r.downloads,
+      })),
+      isDone: page.isDone,
+      continueCursor: page.continueCursor,
+    };
   },
 });
 
@@ -92,10 +112,38 @@ export const rebuildTrendingLeaderboardAction = internalAction({
     const now = Date.now();
     const { startDay, endDay } = getTrendingRange(now);
     const dayKeys = Array.from({ length: endDay - startDay + 1 }, (_, i) => startDay + i);
-    const perDayRows = await Promise.all(
-      dayKeys.map((day) => ctx.runQuery(internal.leaderboards.getDailyStats, { day })),
-    );
-    const entries = buildTrendingEntriesFromDailyRows(perDayRows);
+    const totals = new Map<Id<"skills">, { installs: number; downloads: number }>();
+
+    for (const day of dayKeys) {
+      let cursor: string | null = null;
+      let isDone = false;
+      while (!isDone) {
+        const page: {
+          rows: Array<{ skillId: Id<"skills">; installs: number; downloads: number }>;
+          isDone: boolean;
+          continueCursor: string;
+        } = await ctx.runQuery(internal.leaderboards.getDailyStatsPage, {
+          day,
+          cursor,
+          limit: DAILY_STATS_PAGE_SIZE,
+        });
+        for (const row of page.rows) {
+          const current = totals.get(row.skillId) ?? { installs: 0, downloads: 0 };
+          current.installs += row.installs;
+          current.downloads += row.downloads;
+          totals.set(row.skillId, current);
+        }
+        cursor = page.continueCursor;
+        isDone = page.isDone;
+      }
+    }
+
+    const entries = Array.from(totals, ([skillId, entry]) => ({
+      skillId,
+      installs: entry.installs,
+      downloads: entry.downloads,
+      score: entry.installs,
+    })).sort((a, b) => compareTrendingEntries(b, a));
     const items = takeTopTrendingEntries(entries, limit);
     const nonSuspicious = await ctx.runQuery(
       internal.leaderboards.filterTopNonSuspiciousTrendingEntries,

convex/skillStatEvents.ts

@@ -22,6 +22,7 @@ import { internal } from "./_generated/api";
 import type { Doc, Id } from "./_generated/dataModel";
 import type { MutationCtx } from "./_generated/server";
 import { internalAction, internalMutation, internalQuery } from "./functions";
+import { toDayKey } from "./lib/leaderboards";
 import { applySkillStatDeltas, bumpDailySkillStats } from "./lib/skillStats";
 import { adjustUserSkillStatsForSkillChange } from "./lib/userSkillStats";
 
@@ -203,7 +204,7 @@ function aggregateEvents(events: Doc<"skillStatEvents">[]): AggregatedDeltas {
 export const processSkillStatEventsInternal = internalMutation({
   args: { batchSize: v.optional(v.number()) },
   handler: async (ctx, args) => {
-    const batchSize = args.batchSize ?? 500;
+    const batchSize = Math.max(1, Math.min(args.batchSize ?? 100, 100));
     const now = Date.now();
 
     // Level 1: Fetch a batch of unprocessed events
@@ -359,17 +360,45 @@ export const applyAggregatedStatsAndUpdateCursor = internalMutation({
   },
   handler: async (ctx, args) => {
     const now = Date.now();
+    const dailyStats = new Map<
+      string,
+      { skillId: Id<"skills">; occurredAt: number; downloads: number; installs: number }
+    >();
 
-    // Update daily stats for trending/leaderboards
     for (const delta of args.skillDeltas) {
       for (const occurredAt of delta.downloadEvents) {
-        await bumpDailySkillStats(ctx, { skillId: delta.skillId, now: occurredAt, downloads: 1 });
+        const key = `${delta.skillId}:${toDayKey(occurredAt)}`;
+        const current = dailyStats.get(key) ?? {
+          skillId: delta.skillId,
+          occurredAt,
+          downloads: 0,
+          installs: 0,
+        };
+        current.downloads += 1;
+        dailyStats.set(key, current);
       }
       for (const occurredAt of delta.installNewEvents) {
-        await bumpDailySkillStats(ctx, { skillId: delta.skillId, now: occurredAt, installs: 1 });
+        const key = `${delta.skillId}:${toDayKey(occurredAt)}`;
+        const current = dailyStats.get(key) ?? {
+          skillId: delta.skillId,
+          occurredAt,
+          downloads: 0,
+          installs: 0,
+        };
+        current.installs += 1;
+        dailyStats.set(key, current);
       }
     }
 
+    for (const stat of dailyStats.values()) {
+      await bumpDailySkillStats(ctx, {
+        skillId: stat.skillId,
+        now: stat.occurredAt,
+        downloads: stat.downloads,
+        installs: stat.installs,
+      });
+    }
+
     // Update cursor position (upsert)
     const existingCursor = await ctx.db
       .query("skillStatUpdateCursors")

convex/skills.ts

@@ -124,8 +124,8 @@ const MAX_LIST_LIMIT = 50;
 const MAX_PUBLIC_LIST_LIMIT = 200;
 const MAX_LIST_BULK_LIMIT = 200;
 const MAX_LIST_TAKE = 1000;
-const MAX_SKILL_CATALOG_SCAN_DOCUMENTS = 30_000;
-const MAX_SKILL_CATALOG_SCAN_PAGES = 200;
+const MAX_SKILL_CATALOG_SCAN_DOCUMENTS = 2_000;
+const MAX_SKILL_CATALOG_SCAN_PAGES = 20;
 const MAX_SKILL_CATALOG_SEARCH_PAGE_SIZE = 200;
 const HARD_DELETE_BATCH_SIZE = 100;
 const HARD_DELETE_VERSION_BATCH_SIZE = 10;
@@ -3269,6 +3269,7 @@ export const listPackageCatalogPage = query({
       loops += 1;
       const effectivePageSize = Math.min(
         remainingScanBudget,
+        250,
         offset > 0 && pageSize
           ? Math.max(pageSize, offset + 1)
           : Math.max(targetCount * 3, targetCount),