From a58294361b8fb12bed323b7502d7df4e6cbd01c7 Mon Sep 17 00:00:00 2001 From: Patrick Erichsen Date: Wed, 29 Jul 2026 23:45:14 -0700 Subject: [PATCH] fix: show honest canonical trending states (#3294) * fix: show honest canonical trending states * ci: guard CLAW-602 permanent Test deploy * fix: fail closed when trending discovery is unavailable * fix: preserve trending rows on pagination errors * feat: build native rolling trending feed * fix: decouple native trending from skills.sh * test: cover native trending rollout independence --- convex/_generated/api.d.ts | 4 + convex/canonicalTrending.test.ts | 206 ++++++++-- convex/canonicalTrending.ts | 226 +++++------ convex/canonicalTrendingTestFixtures.test.ts | 20 +- convex/canonicalTrendingTestFixtures.ts | 39 +- convex/crons.test.ts | 16 + convex/crons.ts | 7 + convex/http.test.ts | 21 +- convex/httpApiV1/trendingV1.test.ts | 13 +- convex/httpApiV1/trendingV1.ts | 5 - convex/lib/canonicalTrending.test.ts | 13 + convex/lib/canonicalTrending.ts | 12 +- convex/lib/retentionPolicy.ts | 12 + convex/lib/skillHourlyStats.test.ts | 380 ++++++++++++++++++ convex/lib/skillHourlyStats.ts | 188 +++++++++ convex/migrations.ts | 115 ++++++ convex/rolloutCapabilities.test.ts | 2 + convex/rolloutCapabilities.ts | 4 + convex/schema.ts | 30 ++ convex/skillHourlyStats.ts | 107 +++++ convex/skillStatEvents.test.ts | 7 + convex/skillStatEvents.ts | 40 ++ convex/stars.test.ts | 52 ++- convex/stars.ts | 44 +- .../home-listing-section.claw591.test.tsx | 38 +- src/__tests__/skills-index.claw591.test.tsx | 30 +- src/components/HomeListingSection.tsx | 68 +++- src/lib/catalogDiscoveryCapabilities.test.ts | 6 +- src/lib/catalogDiscoveryCapabilities.ts | 7 +- src/lib/homeListingData.claw591.test.ts | 46 ++- src/lib/homeListingData.ts | 66 ++- src/lib/trendingApi.ts | 2 + src/routes/skills/-SkillsResults.tsx | 31 +- src/routes/skills/-useSkillsBrowseModel.ts | 39 +- src/routes/skills/index.tsx | 1 + 35 files changed, 1583 insertions(+), 314 deletions(-) create mode 100644 convex/lib/skillHourlyStats.test.ts create mode 100644 convex/lib/skillHourlyStats.ts create mode 100644 convex/skillHourlyStats.ts diff --git a/convex/_generated/api.d.ts b/convex/_generated/api.d.ts index 1a3756e2..dffc2644 100644 --- a/convex/_generated/api.d.ts +++ b/convex/_generated/api.d.ts @@ -130,6 +130,7 @@ import type * as lib_skillBackfill from "../lib/skillBackfill.js"; import type * as lib_skillCards from "../lib/skillCards.js"; import type * as lib_skillDownloadBackfill from "../lib/skillDownloadBackfill.js"; import type * as lib_skillFileAccess from "../lib/skillFileAccess.js"; +import type * as lib_skillHourlyStats from "../lib/skillHourlyStats.js"; import type * as lib_skillInstallBackfill from "../lib/skillInstallBackfill.js"; import type * as lib_skillPresentation from "../lib/skillPresentation.js"; import type * as lib_skillPresentationBackfill from "../lib/skillPresentationBackfill.js"; @@ -182,6 +183,7 @@ import type * as securityDatasetNode from "../securityDatasetNode.js"; import type * as securityScan from "../securityScan.js"; import type * as securityScanDispatch from "../securityScanDispatch.js"; import type * as skillCards from "../skillCards.js"; +import type * as skillHourlyStats from "../skillHourlyStats.js"; import type * as skillPresentationAssets from "../skillPresentationAssets.js"; import type * as skillPresentationAssetsHttp from "../skillPresentationAssetsHttp.js"; import type * as skillPresentationBackfill from "../skillPresentationBackfill.js"; @@ -331,6 +333,7 @@ declare const fullApi: ApiFromModules<{ "lib/skillCards": typeof lib_skillCards; "lib/skillDownloadBackfill": typeof lib_skillDownloadBackfill; "lib/skillFileAccess": typeof lib_skillFileAccess; + "lib/skillHourlyStats": typeof lib_skillHourlyStats; "lib/skillInstallBackfill": typeof lib_skillInstallBackfill; "lib/skillPresentation": typeof lib_skillPresentation; "lib/skillPresentationBackfill": typeof lib_skillPresentationBackfill; @@ -383,6 +386,7 @@ declare const fullApi: ApiFromModules<{ securityScan: typeof securityScan; securityScanDispatch: typeof securityScanDispatch; skillCards: typeof skillCards; + skillHourlyStats: typeof skillHourlyStats; skillPresentationAssets: typeof skillPresentationAssets; skillPresentationAssetsHttp: typeof skillPresentationAssetsHttp; skillPresentationBackfill: typeof skillPresentationBackfill; diff --git a/convex/canonicalTrending.test.ts b/convex/canonicalTrending.test.ts index 2836639f..894ef8ca 100644 --- a/convex/canonicalTrending.test.ts +++ b/convex/canonicalTrending.test.ts @@ -3,6 +3,7 @@ import { convexTest } from "convex-test"; import { afterEach, describe, expect, it, vi } from "vitest"; import { internal } from "./_generated/api"; +import { getCompletedRolling24HourWindow } from "./lib/skillHourlyStats"; import schema from "./schema"; const modules = import.meta.glob("./**/*.ts"); @@ -160,11 +161,12 @@ describe("canonical Trending snapshot storage", () => { it("keeps pagination pinned to the snapshot encoded by the cursor", async () => { const t = convexTest(schema, modules); const source = await insertEligibleNativeSource(t, "pagination-source"); + const now = Date.now(); await t.mutation(internal.canonicalTrending.startSnapshotInternal, { snapshotId: "skills-1000", - generatedAt: 1_000, - expiresAt: Date.now() + 100_000, + generatedAt: now - 1_000, + expiresAt: now + 100_000, windowStartDay: 40, windowEndDay: 40, }); @@ -193,7 +195,7 @@ describe("canonical Trending snapshot storage", () => { }); await t.mutation(internal.canonicalTrending.finalizeSnapshotInternal, { snapshotId: "skills-1000", - completedAt: 1_050, + completedAt: now - 950, totalItems: 3, sourceCounts: { clawhubTrending: 1, clawhubRising: 1, skillsShTrending: 1 }, operations: { documentsRead: 12, documentsWritten: 4, functionCalls: 3 }, @@ -209,9 +211,9 @@ describe("canonical Trending snapshot storage", () => { expect(firstPage).toMatchObject({ kind: "skills", snapshotId: "skills-1000", - generatedAt: "1970-01-01T00:00:01.000Z", + generatedAt: new Date(now - 1_000).toISOString(), windowHours: 24, - rankingVersion: "skills-trending-v1", + rankingVersion: "skills-trending-v2", items: [ { id: "clawhub:one", rank: 1, lane: "clawhub-trending" }, { id: "clawhub:two", rank: 2, lane: "skills-sh-trending" }, @@ -221,8 +223,8 @@ describe("canonical Trending snapshot storage", () => { await t.mutation(internal.canonicalTrending.startSnapshotInternal, { snapshotId: "skills-2000", - generatedAt: 2_000, - expiresAt: Date.now() + 100_000, + generatedAt: now, + expiresAt: now + 100_000, windowStartDay: 41, windowEndDay: 41, }); @@ -239,7 +241,7 @@ describe("canonical Trending snapshot storage", () => { }); await t.mutation(internal.canonicalTrending.finalizeSnapshotInternal, { snapshotId: "skills-2000", - completedAt: 2_050, + completedAt: now + 50, totalItems: 1, sourceCounts: { clawhubTrending: 1, clawhubRising: 0, skillsShTrending: 0 }, operations: { documentsRead: 5, documentsWritten: 2, functionCalls: 3 }, @@ -309,7 +311,7 @@ describe("canonical Trending snapshot storage", () => { ).toEqual({ status: "invalid-cursor" }); }); - it("does not read or write source state while the rollout is dark", async () => { + it("reports unavailable without writing when native hourly stats are not ready", async () => { vi.stubEnv("CLAWHUB_ENV", "production"); vi.stubEnv("CLAWHUB_SKILLS_SH_ROLLOUT_MODE", "off"); const t = convexTest(schema, modules); @@ -319,10 +321,151 @@ describe("canonical Trending snapshot storage", () => { ctx.db.query("canonicalTrendingSnapshots").collect(), ); - expect(result).toEqual({ status: "disabled" }); + expect(result).toEqual({ status: "unavailable", reason: "hourly-stats-not-ready" }); expect(snapshots).toEqual([]); }); + it("anchors the rolling window to a fresh completed aggregation across cron boundaries", async () => { + const t = convexTest(schema, modules); + const now = 100 * 60 * 60 * 1_000 + 7 * 60 * 1_000; + const lastAggregationCompletedAt = now - 15 * 60 * 1_000; + await t.run(async (ctx) => { + await ctx.db.insert("skillHourlyStatStates", { + key: "canonical_trending", + liveStartedAt: now - 60 * 60 * 1_000, + eventBackfillThroughCreationTime: 100, + activeGeneration: 1, + backfillCompletedAt: now - 30 * 60 * 1_000, + lastAggregationCompletedAt, + updatedAt: lastAggregationCompletedAt, + }); + }); + + await expect( + t.mutation(internal.skillHourlyStats.sealForSnapshotInternal, { now }), + ).resolves.toMatchObject({ + startHour: 75, + endHour: 98, + startAt: 75 * 60 * 60 * 1_000, + endAt: 99 * 60 * 60 * 1_000, + lastAggregationCompletedAt, + sealedGeneration: 1, + }); + }); + + it("rejects hourly aggregation state once it is two hours old", async () => { + const t = convexTest(schema, modules); + const now = 100 * 60 * 60 * 1_000; + await t.run(async (ctx) => { + await ctx.db.insert("skillHourlyStatStates", { + key: "canonical_trending", + liveStartedAt: now - 3 * 60 * 60 * 1_000, + eventBackfillThroughCreationTime: 100, + activeGeneration: 1, + backfillCompletedAt: now - 3 * 60 * 60 * 1_000, + lastAggregationCompletedAt: now - 2 * 60 * 60 * 1_000, + updatedAt: now - 2 * 60 * 60 * 1_000, + }); + }); + + await expect( + t.mutation(internal.skillHourlyStats.sealForSnapshotInternal, { now }), + ).resolves.toBeNull(); + }); + + it("materializes native rolling activity while skills.sh is disabled", async () => { + vi.stubEnv("CLAWHUB_ENV", "production"); + vi.stubEnv("CLAWHUB_SKILLS_SH_ROLLOUT_MODE", "off"); + const t = convexTest(schema, modules); + const source = await insertEligibleNativeSource(t, "native-only"); + const now = Date.now(); + const window = getCompletedRolling24HourWindow(now); + await t.run(async (ctx) => { + await ctx.db.insert("skillHourlyStatStates", { + key: "canonical_trending", + liveStartedAt: now - 3_600_000, + eventBackfillThroughCreationTime: 100, + activeGeneration: 1, + backfillCompletedAt: now - 1_000, + lastAggregationCompletedAt: window.endAt + 1, + lastProcessedEventCreationTime: 100, + updatedAt: now, + }); + await ctx.db.insert("skillHourlyStats", { + skillId: source.skillId, + hour: window.endHour, + generation: 0, + downloads: 6, + installs: 8, + bookmarks: 3, + updatedAt: now, + expiresAt: now + 72 * 3_600_000, + }); + }); + + const result = await t.action(internal.canonicalTrending.materializeInternal, {}); + + expect(result).toMatchObject({ + status: "ready", + totalItems: 1, + sourceCounts: { clawhubTrending: 1, clawhubRising: 1, skillsShTrending: 0 }, + sample: [ + expect.objectContaining({ + id: expect.stringMatching(/^clawhub:/), + trending24hInstalls: 8, + }), + ], + }); + }); + + it("stops serving the last good snapshot after two hours", async () => { + const t = convexTest(schema, modules); + const now = Date.now(); + await t.mutation(internal.canonicalTrending.startSnapshotInternal, { + snapshotId: "skills-stale-serving", + generatedAt: now - 2 * 60 * 60 * 1_000 - 1, + expiresAt: now + 24 * 60 * 60 * 1_000, + windowStartDay: 40, + windowEndDay: 40, + }); + await t.mutation(internal.canonicalTrending.finalizeSnapshotInternal, { + snapshotId: "skills-stale-serving", + completedAt: now - 2 * 60 * 60 * 1_000, + totalItems: 0, + sourceCounts: { clawhubTrending: 0, clawhubRising: 0, skillsShTrending: 0 }, + operations: { documentsRead: 1, documentsWritten: 2, functionCalls: 2 }, + }); + + expect( + await t.query(internal.canonicalTrending.getPageInternal, { cursor: null, limit: 20 }), + ).toEqual({ status: "unavailable" }); + }); + + it("never serves a snapshot produced by the legacy ranking algorithm", async () => { + const t = convexTest(schema, modules); + const now = Date.now(); + await t.run(async (ctx) => { + await ctx.db.insert("canonicalTrendingSnapshots", { + snapshotId: "skills-legacy-ranking", + kind: "skills", + status: "ready", + rankingVersion: "skills-trending-v1", + generatedAt: now, + completedAt: now, + expiresAt: now + 24 * 60 * 60 * 1_000, + windowHours: 24, + windowStartDay: 40, + windowEndDay: 40, + writtenItems: 0, + totalItems: 0, + }); + }); + + expect( + await t.query(internal.canonicalTrending.getPageInternal, { cursor: null, limit: 20 }), + ).toEqual({ status: "unavailable" }); + }); + it("prunes expired snapshots independently while materialization is dark", async () => { vi.stubEnv("CLAWHUB_ENV", "production"); vi.stubEnv("CLAWHUB_SKILLS_SH_ROLLOUT_MODE", "off"); @@ -362,11 +505,12 @@ describe("canonical Trending snapshot storage", () => { expect(rows).toEqual({ snapshots: [], items: [] }); }); - it("materializes imported 24-hour metrics into a ready snapshot", async () => { + it("materializes hourly native metrics with a fresh enabled skills.sh contribution", async () => { vi.stubEnv("CLAWHUB_ENV", "test"); vi.stubEnv("CLAWHUB_SKILLS_SH_ROLLOUT_MODE", "test"); const t = convexTest(schema, modules); const now = Date.now(); + const window = getCompletedRolling24HourWindow(now); await t.run(async (ctx) => { const userId = await ctx.db.insert("users", { @@ -414,39 +558,25 @@ describe("canonical Trending snapshot storage", () => { createdAt: now - 1_000, updatedAt: now, }); - await ctx.db.insert("rankingMetricImports", { - datasetVersion: "ranking-test-v1", - checksum: "a".repeat(64), - generatedAt: new Date(now).toISOString(), - importedAt: now, - startDay: 1, - endDay: 60, - targetCount: 1, - skillTargetCount: 1, - packageTargetCount: 0, - dailyRowCount: 1, - importedSkillRows: 1, - importedPackageRows: 0, - unresolvedTargets: 0, - skippedOverlayRows: 0, + await ctx.db.insert("skillHourlyStatStates", { + key: "canonical_trending", + liveStartedAt: now - 3_600_000, + eventBackfillThroughCreationTime: 100, + activeGeneration: 1, + backfillCompletedAt: now - 1_000, + lastAggregationCompletedAt: window.endAt + 1, + lastProcessedEventCreationTime: 100, + updatedAt: now, }); - await ctx.db.insert("skillDailyStats", { + await ctx.db.insert("skillHourlyStats", { skillId, - day: 60, + hour: window.endHour, + generation: 0, downloads: 18, installs: 12, bookmarks: 4, - rankingDatasetVersion: "ranking-test-v1", - rankingImportedAt: now, updatedAt: now, - }); - await ctx.db.insert("skillDailyStats", { - skillId, - day: 61, - downloads: 999, - installs: 999, - bookmarks: 999, - updatedAt: now + 1, + expiresAt: now + 72 * 3_600_000, }); const trendingRunId = await ctx.db.insert("skillsShMirrorRuns", { snapshotId: "skills-sh-trending-runtime", diff --git a/convex/canonicalTrending.ts b/convex/canonicalTrending.ts index d6cda9d5..d62e2727 100644 --- a/convex/canonicalTrending.ts +++ b/convex/canonicalTrending.ts @@ -13,16 +13,20 @@ import { canonicalTrendingSourceRefValidator, decodeCanonicalTrendingCursor, encodeCanonicalTrendingCursor, + isFreshExternalTrendingRun, type CanonicalTrendingMaterializationCandidate, } from "./lib/canonicalTrending"; import { shouldExcludeSkillFromPublicBrowse } from "./lib/publicBrowse"; import { getRuntimeRolloutCapabilities } from "./lib/rolloutCapabilities"; +import { getCompletedRolling24HourWindow, sumRollingHourlyStats } from "./lib/skillHourlyStats"; import { isPublicSkillsShMirrorDigest } from "./lib/skillsShMirrorPublic"; import { assertTestSeedAllowed } from "./lib/testSeed"; const SOURCE_PAGE_SIZE = 250; const WRITE_BATCH_SIZE = 100; const SNAPSHOT_RETENTION_MS = 48 * 60 * 60 * 1_000; +const SNAPSHOT_MAX_SERVING_AGE_MS = 2 * 60 * 60 * 1_000; +const EXTERNAL_SOURCE_MAX_AGE_MS = 2 * 60 * 60 * 1_000; const RISING_MAX_AGE_MS = 30 * 24 * 60 * 60 * 1_000; const PRUNE_BATCH_SIZE = 500; const PRUNE_MAX_BATCHES = 20; @@ -32,8 +36,7 @@ const internalRefs = internal as unknown as { failSnapshotInternal: unknown; finalizeSnapshotInternal: unknown; getExternalSourcePageInternal: unknown; - getMetricSourcePageInternal: unknown; - getMetricWindowInternal: unknown; + getHourlySourcePageInternal: unknown; getNativeSourcePageInternal: unknown; getLatestCompletedTrendingRunInternal: unknown; pruneExpiredInternal: unknown; @@ -41,6 +44,9 @@ const internalRefs = internal as unknown as { startSnapshotInternal: unknown; writeItemsInternal: unknown; }; + skillHourlyStats: { + sealForSnapshotInternal: unknown; + }; }; async function pruneExpiredRows( @@ -118,26 +124,6 @@ async function collectSourcePages( return { rows, documentsRead, functionCalls }; } -export const getMetricWindowInternal = internalQuery({ - args: {}, - handler: async (ctx) => { - const latestImport = await ctx.db - .query("rankingMetricImports") - .withIndex("by_imported_at") - .order("desc") - .first(); - return latestImport - ? { - datasetVersion: latestImport.datasetVersion, - importedAt: latestImport.importedAt, - startDay: latestImport.endDay, - endDay: latestImport.endDay, - documentsRead: 1, - } - : null; - }, -}); - export const getNativeSourcePageInternal = internalQuery({ args: { paginationOpts: paginationOptsValidator }, handler: async (ctx, args) => { @@ -157,25 +143,21 @@ export const getNativeSourcePageInternal = internalQuery({ }, }); -export const getMetricSourcePageInternal = internalQuery({ +export const getHourlySourcePageInternal = internalQuery({ args: { - day: v.number(), - datasetVersion: v.string(), - importedAt: v.number(), + startHour: v.number(), + endHour: v.number(), + maxGeneration: v.number(), paginationOpts: paginationOptsValidator, }, handler: async (ctx, args) => { const result = await ctx.db - .query("skillDailyStats") - .withIndex("by_day", (q) => q.eq("day", args.day)) + .query("skillHourlyStats") + .withIndex("by_hour", (q) => q.gte("hour", args.startHour).lte("hour", args.endHour)) .paginate(args.paginationOpts); return { ...result, - page: result.page.filter( - (row) => - row.rankingDatasetVersion === args.datasetVersion && - row.rankingImportedAt === args.importedAt, - ), + page: result.page.filter((row) => row.generation <= args.maxGeneration), documentsRead: result.page.length, }; }, @@ -215,7 +197,11 @@ export const getLatestCompletedTrendingRunInternal = internalQuery({ q.and(q.eq(q.field("sourceView"), "trending"), q.eq(q.field("status"), "completed")), ) .first(); - return { runId: run?._id ?? null, documentsRead: Number(Boolean(run)) }; + return { + runId: run?._id ?? null, + completedAt: run?.completedAt ?? null, + documentsRead: Number(Boolean(run)), + }; }, }); @@ -226,6 +212,8 @@ export const startSnapshotInternal = internalMutation({ expiresAt: v.number(), windowStartDay: v.number(), windowEndDay: v.number(), + windowStartHour: v.optional(v.number()), + windowEndHour: v.optional(v.number()), }, handler: async (ctx, args) => { const existing = await ctx.db @@ -243,6 +231,8 @@ export const startSnapshotInternal = internalMutation({ windowHours: CANONICAL_TRENDING_WINDOW_HOURS, windowStartDay: args.windowStartDay, windowEndDay: args.windowEndDay, + windowStartHour: args.windowStartHour, + windowEndHour: args.windowEndHour, writtenItems: 0, }); }, @@ -388,10 +378,6 @@ export const materializeInternal = internalAction({ throw new Error("Invalid CLAW-590 proof snapshot ID"); } } - if (!getRuntimeRolloutCapabilities().skillsSh.runtimeEnabled) { - return { status: "disabled" as const }; - } - const startedAt = Date.now(); const snapshotId = args.proofSnapshotId ?? `skills-${startedAt}`; let snapshotStarted = false; @@ -400,94 +386,96 @@ export const materializeInternal = internalAction({ let documentsWritten = 0; try { - const metricWindow = (await ctx.runQuery( - internalRefs.canonicalTrending.getMetricWindowInternal as never, - {}, - )) as { - datasetVersion: string; - importedAt: number; - startDay: number; - endDay: number; - documentsRead: number; - } | null; - functionCalls += 1; - documentsRead += metricWindow?.documentsRead ?? 0; - if (!metricWindow) throw new Error("No 24-hour ranking metric window is available"); + type HourlyWindow = { + startHour: number; + endHour: number; + startAt: number; + endAt: number; + lastAggregationCompletedAt: number; + sealedGeneration: number; + }; + const proofWindow = + args.proofSnapshotId !== undefined + ? { + ...getCompletedRolling24HourWindow(startedAt), + lastAggregationCompletedAt: startedAt, + sealedGeneration: 0, + } + : null; + const hourlyWindow = proofWindow + ? proofWindow + : ((await ctx.runMutation( + internalRefs.skillHourlyStats.sealForSnapshotInternal as never, + { now: startedAt } as never, + )) as HourlyWindow | null); + if (!proofWindow) functionCalls += 1; + if (!hourlyWindow) { + return { status: "unavailable" as const, reason: "hourly-stats-not-ready" as const }; + } const nativeSource = (await collectSourcePages( ctx, internalRefs.canonicalTrending.getNativeSourcePageInternal, )) as CollectedSource>; - const metricSource = (await collectSourcePages( + const hourlySource = (await collectSourcePages( ctx, - internalRefs.canonicalTrending.getMetricSourcePageInternal, + internalRefs.canonicalTrending.getHourlySourcePageInternal, { - day: metricWindow.endDay, - datasetVersion: metricWindow.datasetVersion, - importedAt: metricWindow.importedAt, + startHour: hourlyWindow.startHour, + endHour: hourlyWindow.endHour, + maxGeneration: hourlyWindow.sealedGeneration, }, - )) as CollectedSource>; - const latestTrendingRun = (await ctx.runQuery( - internalRefs.canonicalTrending.getLatestCompletedTrendingRunInternal as never, - {}, - )) as { runId: Doc<"skillsShMirrorRuns">["_id"] | null; documentsRead: number }; - documentsRead += latestTrendingRun.documentsRead; - functionCalls += 1; - const externalSource = (await collectSourcePages( - ctx, - internalRefs.canonicalTrending.getExternalSourcePageInternal, - )) as CollectedSource>; + )) as CollectedSource>; + type TrendingRun = { + runId: Doc<"skillsShMirrorRuns">["_id"] | null; + completedAt: number | null; + documentsRead: number; + }; + let latestTrendingRun: TrendingRun | null = null; + let externalSource: CollectedSource> = { + rows: [], + documentsRead: 0, + functionCalls: 0, + }; + if (getRuntimeRolloutCapabilities().skillsSh.runtimeEnabled) { + const candidateRun = (await ctx.runQuery( + internalRefs.canonicalTrending.getLatestCompletedTrendingRunInternal as never, + {}, + )) as TrendingRun; + documentsRead += candidateRun.documentsRead; + functionCalls += 1; + if (isFreshExternalTrendingRun(candidateRun, startedAt, EXTERNAL_SOURCE_MAX_AGE_MS)) { + latestTrendingRun = candidateRun; + externalSource = (await collectSourcePages( + ctx, + internalRefs.canonicalTrending.getExternalSourcePageInternal, + )) as CollectedSource>; + } + } documentsRead += - nativeSource.documentsRead + metricSource.documentsRead + externalSource.documentsRead; + nativeSource.documentsRead + hourlySource.documentsRead + externalSource.documentsRead; functionCalls += - nativeSource.functionCalls + metricSource.functionCalls + externalSource.functionCalls; + nativeSource.functionCalls + hourlySource.functionCalls + externalSource.functionCalls; - const confirmedMetricWindow = (await ctx.runQuery( - internalRefs.canonicalTrending.getMetricWindowInternal as never, - {}, - )) as typeof metricWindow; - functionCalls += 1; - documentsRead += confirmedMetricWindow?.documentsRead ?? 0; - if ( - !confirmedMetricWindow || - confirmedMetricWindow.datasetVersion !== metricWindow.datasetVersion || - confirmedMetricWindow.importedAt !== metricWindow.importedAt - ) { - throw new Error("24-hour ranking metric import changed during materialization"); + if (latestTrendingRun) { + const confirmedTrendingRun = (await ctx.runQuery( + internalRefs.canonicalTrending.getLatestCompletedTrendingRunInternal as never, + {}, + )) as TrendingRun; + documentsRead += confirmedTrendingRun.documentsRead; + functionCalls += 1; + if (confirmedTrendingRun.runId !== latestTrendingRun.runId) { + throw new Error("skills.sh Trending run changed during materialization"); + } } - const confirmedTrendingRun = (await ctx.runQuery( - internalRefs.canonicalTrending.getLatestCompletedTrendingRunInternal as never, - {}, - )) as { runId: Doc<"skillsShMirrorRuns">["_id"] | null; documentsRead: number }; - documentsRead += confirmedTrendingRun.documentsRead; - functionCalls += 1; - if (confirmedTrendingRun.runId !== latestTrendingRun.runId) { - throw new Error("skills.sh Trending run changed during materialization"); - } - const latestTrendingRunId = latestTrendingRun.runId; - - const usageBySkill = new Map( - metricSource.rows.map((row) => [ - String(row.skillId), - { - installs: row.installs, - bookmarks: row.bookmarks ?? 0, - updatedAt: row.rankingImportedAt ?? row.updatedAt, - }, - ]), - ); + const usageBySkill = sumRollingHourlyStats(hourlySource.rows); const nativeCandidates = nativeSource.rows - .map((digest) => - buildNativeCanonicalTrendingCandidate( - digest, - usageBySkill.get(String(digest.skillId)) ?? { - installs: 0, - bookmarks: 0, - updatedAt: metricWindow.importedAt, - }, - ), - ) + .map((digest) => { + const usage = usageBySkill.get(String(digest.skillId)); + if (!usage || usage.downloads + usage.installs + usage.bookmarks <= 0) return null; + return buildNativeCanonicalTrendingCandidate(digest, usage); + }) .filter( (candidate): candidate is CanonicalTrendingMaterializationCandidate => candidate !== null, ); @@ -496,7 +484,7 @@ export const materializeInternal = internalAction({ .filter((candidate) => candidate.createdAt >= risingCutoff) .map((candidate) => ({ ...candidate, lane: "clawhub-rising" as const })); const externalCandidates = externalSource.rows - .filter((digest) => digest.trendingObservedRunId === latestTrendingRunId) + .filter((digest) => digest.trendingObservedRunId === latestTrendingRun?.runId) .map(buildExternalCanonicalTrendingCandidate) .filter( (candidate): candidate is CanonicalTrendingMaterializationCandidate => candidate !== null, @@ -514,8 +502,10 @@ export const materializeInternal = internalAction({ snapshotId, generatedAt: startedAt, expiresAt, - windowStartDay: metricWindow.startDay, - windowEndDay: metricWindow.endDay, + windowStartDay: Math.floor(hourlyWindow.startHour / 24), + windowEndDay: Math.floor(hourlyWindow.endHour / 24), + windowStartHour: hourlyWindow.startHour, + windowEndHour: hourlyWindow.endHour, } as never, ); snapshotStarted = true; @@ -634,6 +624,12 @@ export const getPageInternal = internalQuery({ .order("desc") .first(); if (!snapshot && !decoded) return { status: "unavailable" as const }; + if (snapshot && snapshot.generatedAt + SNAPSHOT_MAX_SERVING_AGE_MS <= now) { + return { status: decoded ? ("expired" as const) : ("unavailable" as const) }; + } + if (snapshot && snapshot.rankingVersion !== CANONICAL_TRENDING_RANKING_VERSION) { + return { status: decoded ? ("expired" as const) : ("unavailable" as const) }; + } if ( !snapshot || snapshot.status !== "ready" || diff --git a/convex/canonicalTrendingTestFixtures.test.ts b/convex/canonicalTrendingTestFixtures.test.ts index 385f84c4..398c65fc 100644 --- a/convex/canonicalTrendingTestFixtures.test.ts +++ b/convex/canonicalTrendingTestFixtures.test.ts @@ -23,24 +23,6 @@ afterEach(() => { describe("CLAW-590 permanent Test snapshot ownership", () => { it("seeds and removes an exact owned 20-row source corpus", async () => { const t = convexTest(schema, modules); - await t.run(async (ctx) => { - await ctx.db.insert("rankingMetricImports", { - datasetVersion: "claw-590-test-dataset", - checksum: "claw-590-test-checksum", - generatedAt: new Date(1_000).toISOString(), - importedAt: 2_000, - startDay: 1, - endDay: 1, - targetCount: 0, - skillTargetCount: 0, - packageTargetCount: 0, - dailyRowCount: 0, - importedSkillRows: 0, - importedPackageRows: 0, - unresolvedTargets: 0, - skippedOverlayRows: 0, - }); - }); await expect( t.mutation(internal.canonicalTrendingTestFixtures.seedCanonicalTrendingSourceFixture, { @@ -140,7 +122,7 @@ describe("CLAW-590 permanent Test snapshot ownership", () => { snapshotId: SNAPSHOT_ID, kind: "skills", status: "failed", - rankingVersion: "skills-trending-v1", + rankingVersion: "skills-trending-v2", generatedAt: 1_000, completedAt: 2_000, expiresAt: Date.now() + 100_000, diff --git a/convex/canonicalTrendingTestFixtures.ts b/convex/canonicalTrendingTestFixtures.ts index 3f161e6c..53b039e8 100644 --- a/convex/canonicalTrendingTestFixtures.ts +++ b/convex/canonicalTrendingTestFixtures.ts @@ -7,6 +7,7 @@ import { internalQuery, type QueryCtx, } from "./_generated/server"; +import { getCompletedRolling24HourWindow } from "./lib/skillHourlyStats"; import { assertTestSeedAllowed } from "./lib/testSeed"; const CONFIRM = "manage-claw-590-canonical-trending-test-proof"; @@ -156,13 +157,17 @@ function assertOwnedNativeDigest( } } -function assertOwnedDailyStat(stat: Doc<"skillDailyStats">, index: number, skillId: Id<"skills">) { +function assertOwnedHourlyStat( + stat: Doc<"skillHourlyStats">, + index: number, + skillId: Id<"skills">, +) { if ( stat.skillId !== skillId || + stat.generation !== 0 || + stat.downloads !== 100_000 - index || stat.installs !== 100_000 - index || - stat.bookmarks !== 10_000 - index || - !stat.rankingDatasetVersion || - stat.rankingImportedAt === undefined + stat.bookmarks !== 10_000 - index ) { throw new Error("CLAW-590 source fixture metric ownership mismatch"); } @@ -275,8 +280,8 @@ async function readOwnedSourceFixture(ctx: Pick) { .withIndex("by_skill", (q) => q.eq("skillId", skill._id)) .unique(), ctx.db - .query("skillDailyStats") - .withIndex("by_skill_day", (q) => q.eq("skillId", skill._id)) + .query("skillHourlyStats") + .withIndex("by_skill_and_hour_and_generation", (q) => q.eq("skillId", skill._id)) .collect(), ]); if (!version || !digest || stats.length !== 1) { @@ -284,7 +289,7 @@ async function readOwnedSourceFixture(ctx: Pick) { } assertOwnedVersion(version, skill._id, owner._id); assertOwnedNativeDigest(digest, index, skill._id, owner._id, version._id); - assertOwnedDailyStat(stats[0]!, index, skill._id); + assertOwnedHourlyStat(stats[0]!, index, skill._id); native.push({ owner, skill, version, digest, stat: stats[0]! }); } @@ -306,7 +311,7 @@ function assertOwnedSnapshot(snapshot: Doc<"canonicalTrendingSnapshots">, snapsh if ( snapshot.snapshotId !== snapshotId || snapshot.kind !== "skills" || - snapshot.rankingVersion !== "skills-trending-v1" || + snapshot.rankingVersion !== "skills-trending-v2" || snapshot.windowHours !== 24 ) { throw new Error("CLAW-590 proof snapshot ownership mismatch"); @@ -327,14 +332,8 @@ export const seedCanonicalTrendingSourceFixture = internalMutation({ }; } - const metricWindow = await ctx.db - .query("rankingMetricImports") - .withIndex("by_imported_at") - .order("desc") - .first(); - if (!metricWindow) throw new Error("CLAW-590 source fixture requires a ranking metric import"); - const now = Date.now(); + const window = getCompletedRolling24HourWindow(now); const users: Id<"users">[] = []; for (let index = 0; index < NATIVE_PUBLISHER_COUNT; index += 1) { users.push( @@ -420,15 +419,15 @@ export const seedCanonicalTrendingSourceFixture = internalMutation({ createdAt: now - index, updatedAt: now - index, }); - await ctx.db.insert("skillDailyStats", { + await ctx.db.insert("skillHourlyStats", { skillId, - day: metricWindow.endDay, + hour: window.endHour, + generation: 0, downloads: 100_000 - index, installs: 100_000 - index, bookmarks: 10_000 - index, - rankingDatasetVersion: metricWindow.datasetVersion, - rankingImportedAt: metricWindow.importedAt, - updatedAt: metricWindow.importedAt, + updatedAt: now, + expiresAt: now + 72 * 60 * 60 * 1_000, }); } diff --git a/convex/crons.test.ts b/convex/crons.test.ts index 9572e230..4b18c66f 100644 --- a/convex/crons.test.ts +++ b/convex/crons.test.ts @@ -12,6 +12,7 @@ const mocks = vi.hoisted(() => { const publisherTemporalAbuseScanPruneRef = Symbol("publisher-temporal-abuse-scan-prune"); const httpRateLimitKeysPruneRef = Symbol("http-rate-limit-keys-prune"); const skillStatEventPruneRef = Symbol("skill-stat-event-prune"); + const skillHourlyStatsPruneRef = Symbol("skill-hourly-stats-prune"); const packageStatEventPruneRef = Symbol("package-stat-event-prune"); const authSessionsPruneRef = Symbol("auth-sessions-prune"); const authRefreshTokensPruneRef = Symbol("auth-refresh-tokens-prune"); @@ -33,6 +34,7 @@ const mocks = vi.hoisted(() => { publisherTemporalAbuseScanPruneRef, httpRateLimitKeysPruneRef, skillStatEventPruneRef, + skillHourlyStatsPruneRef, packageStatEventPruneRef, authSessionsPruneRef, authRefreshTokensPruneRef, @@ -73,6 +75,9 @@ vi.mock("./_generated/api", () => ({ processSkillStatEventsInternal: Symbol("skill-doc-stat-sync"), pruneProcessedSkillStatEventsInternal: mocks.skillStatEventPruneRef, }, + skillHourlyStats: { + pruneExpiredInternal: mocks.skillHourlyStatsPruneRef, + }, packages: { processPackageStatEventsInternal: Symbol("package-stat-events"), pruneProcessedPackageStatEventsInternal: mocks.packageStatEventPruneRef, @@ -194,6 +199,17 @@ describe("crons", () => { ); }); + it("prunes expired hourly skill stats independently each hour", async () => { + await import("./crons"); + + expect(mocks.interval).toHaveBeenCalledWith( + "skill-hourly-stats-prune", + { hours: 1 }, + mocks.skillHourlyStatsPruneRef, + { batchSize: 500 }, + ); + }); + it("prunes expired skill scan requests in bounded continuation batches", async () => { await import("./crons"); diff --git a/convex/crons.ts b/convex/crons.ts index 78251da7..198aa78d 100644 --- a/convex/crons.ts +++ b/convex/crons.ts @@ -40,6 +40,13 @@ if (process.env.CLAWHUB_DISABLE_CRONS !== "1" && process.env.CLAWHUB_PREVIEW !== {}, ); + crons.interval( + "skill-hourly-stats-prune", + { hours: 1 }, + internal.skillHourlyStats.pruneExpiredInternal, + { batchSize: RETENTION_STANDARD_BATCH_SIZE }, + ); + crons.interval( "package-trending-leaderboard", { minutes: 60 }, diff --git a/convex/http.test.ts b/convex/http.test.ts index 982519f4..7f6103ae 100644 --- a/convex/http.test.ts +++ b/convex/http.test.ts @@ -97,22 +97,15 @@ describe("HTTP route rate limit defaults", () => { vi.unstubAllEnvs(); }); - it("keeps the dark canonical Trending route ahead of rate limiting", async () => { + it("keeps native Trending rate limited while the skills.sh lane is off", async () => { vi.stubEnv("CLAWHUB_ENV", "production"); vi.stubEnv("CLAWHUB_SKILLS_SH_ROLLOUT_MODE", "off"); - const route = http.lookup(ApiRoutes.trending, "GET"); - if (!route) throw new Error("Expected canonical Trending route"); - const [action] = route; - const { ctx, runMutation } = makeDeniedRateLimitCtx(); - - const response = await (action as unknown as WrappedHttpAction)._handler( - ctx, - new Request(`https://example.com${ApiRoutes.trending}`), - ); - - expect(response.status).toBe(404); - expect(response.headers.get("cache-control")).toBe("no-store"); - expect(runMutation).not.toHaveBeenCalled(); + await expectRouteUsesIpBucket({ + path: ApiRoutes.trending, + method: "GET", + bucket: "readIp", + rate: RATE_LIMITS.read.ip, + }); }); it("registers package version downloads behind the download limit", async () => { diff --git a/convex/httpApiV1/trendingV1.test.ts b/convex/httpApiV1/trendingV1.test.ts index 6c8f8218..0b9b2d6f 100644 --- a/convex/httpApiV1/trendingV1.test.ts +++ b/convex/httpApiV1/trendingV1.test.ts @@ -20,19 +20,20 @@ afterEach(() => { }); describe("canonical Trending HTTP API", () => { - it("stays dark before rate limiting while the skills.sh rollout is disabled", async () => { + it("serves native Trending independently while the skills.sh rollout is disabled", async () => { vi.stubEnv("CLAWHUB_SKILLS_SH_ROLLOUT_MODE", "off"); - const runQuery = vi.fn(); + const page = { kind: "skills", items: [] }; + const runQuery = vi.fn(async () => ({ status: "ok", page })); const response = await trendingV1Handler( { runQuery } as never, new Request("https://clawhub.ai/api/v1/trending?kind=skills"), ); - expect(response.status).toBe(404); - expect(response.headers.get("cache-control")).toBe("no-store"); - expect(applyRateLimit).not.toHaveBeenCalled(); - expect(runQuery).not.toHaveBeenCalled(); + expect(response.status).toBe(200); + expect(await response.json()).toEqual(page); + expect(applyRateLimit).toHaveBeenCalledOnce(); + expect(runQuery).toHaveBeenCalledOnce(); }); it("returns the materialized snapshot envelope without reordering cards", async () => { diff --git a/convex/httpApiV1/trendingV1.ts b/convex/httpApiV1/trendingV1.ts index 7ad74cee..6f95778d 100644 --- a/convex/httpApiV1/trendingV1.ts +++ b/convex/httpApiV1/trendingV1.ts @@ -1,7 +1,6 @@ import { internal } from "../_generated/api"; import type { ActionCtx } from "../_generated/server"; import { applyRateLimit } from "../lib/httpRateLimit"; -import { getRuntimeRolloutCapabilities } from "../lib/rolloutCapabilities"; import { json, text } from "./shared"; const internalRefs = internal as unknown as { @@ -27,10 +26,6 @@ function parseLimit(value: string | null) { } export async function trendingV1Handler(ctx: ActionCtx, request: Request) { - if (!getRuntimeRolloutCapabilities().skillsSh.runtimeEnabled) { - return text("Not found", 404, { "cache-control": "no-store" }); - } - const rate = await applyRateLimit(ctx, request, "read"); if (!rate.ok) return rate.response; diff --git a/convex/lib/canonicalTrending.test.ts b/convex/lib/canonicalTrending.test.ts index fb8908ee..4952c0bc 100644 --- a/convex/lib/canonicalTrending.test.ts +++ b/convex/lib/canonicalTrending.test.ts @@ -5,6 +5,7 @@ import { buildNativeCanonicalTrendingCandidate, decodeCanonicalTrendingCursor, encodeCanonicalTrendingCursor, + isFreshExternalTrendingRun, sortCanonicalTrendingPools, type CanonicalTrendingCandidate, } from "./canonicalTrending"; @@ -28,6 +29,18 @@ function candidate( } describe("canonical Trending ordering", () => { + it("admits skills.sh only while its latest completed run is fresh", () => { + expect(isFreshExternalTrendingRun({ runId: "run-1", completedAt: 8_001 }, 10_000, 2_000)).toBe( + true, + ); + expect(isFreshExternalTrendingRun({ runId: "run-1", completedAt: 8_000 }, 10_000, 2_000)).toBe( + false, + ); + expect(isFreshExternalTrendingRun({ runId: null, completedAt: 9_999 }, 10_000, 2_000)).toBe( + false, + ); + }); + it("interleaves complete pools as a continuous 40/20/40 feed", () => { const result = blendCanonicalTrendingPools({ clawhubTrending: Array.from({ length: 4 }, (_, index) => diff --git a/convex/lib/canonicalTrending.ts b/convex/lib/canonicalTrending.ts index d84e40b5..30d2d89b 100644 --- a/convex/lib/canonicalTrending.ts +++ b/convex/lib/canonicalTrending.ts @@ -1,11 +1,21 @@ import { type Infer, v } from "convex/values"; import type { Doc } from "../_generated/dataModel"; -export const CANONICAL_TRENDING_RANKING_VERSION = "skills-trending-v1"; +export const CANONICAL_TRENDING_RANKING_VERSION = "skills-trending-v2"; export const CANONICAL_TRENDING_WINDOW_HOURS = 24; export const CANONICAL_TRENDING_FIRST_PAGE_SIZE = 20; export const CANONICAL_TRENDING_PUBLISHER_CAP = 2; +export function isFreshExternalTrendingRun( + run: { runId: string | null; completedAt: number | null }, + now: number, + maxAgeMs: number, +) { + return Boolean( + run.runId && run.completedAt !== null && run.completedAt > now - Math.max(0, maxAgeMs), + ); +} + const canonicalTrendingUpstreamScannerValidator = v.object({ status: v.string(), sourceCheckedAt: v.optional(v.string()), diff --git a/convex/lib/retentionPolicy.ts b/convex/lib/retentionPolicy.ts index 21d0e1cf..8e092ce7 100644 --- a/convex/lib/retentionPolicy.ts +++ b/convex/lib/retentionPolicy.ts @@ -192,6 +192,18 @@ export const RETENTION_POLICIES = { curatedSkillSearchDigest: derived("Curated search projection of skill state.", "skills"), skillTopicSearchDigest: derived("Topic search projection of skill state.", "skills"), skillDailyStats: permanent("Daily aggregate stats are product analytics."), + skillHourlyStats: ephemeral( + "Hourly skill activity is retained only long enough to build rolling Trending snapshots.", + { + expirationField: "expiresAt", + expirationIndex: "by_expires_at", + prune: "skillHourlyStats.pruneExpiredInternal", + retention: "Seventy-two hours after the hourly bucket completes.", + }, + ), + skillHourlyStatStates: permanent( + "Hourly aggregation rollout boundary and backfill completion state.", + ), skillLeaderboards: derived("Leaderboard snapshots can be rebuilt from stats.", "skillDailyStats"), skillStatBackfillState: permanent("Backfill cursor state."), globalStats: derived("Global stats aggregate can be recalculated.", "skills/packages"), diff --git a/convex/lib/skillHourlyStats.test.ts b/convex/lib/skillHourlyStats.test.ts new file mode 100644 index 00000000..b954e04f --- /dev/null +++ b/convex/lib/skillHourlyStats.test.ts @@ -0,0 +1,380 @@ +/// +/* @vitest-environment edge-runtime */ +import { convexTest } from "convex-test"; +import { describe, expect, it, vi } from "vitest"; +import { internal } from "../_generated/api"; +import schema from "../schema"; +import { + HOUR_MS, + bumpHistoricalHourlySkillStats, + bumpLiveHourlySkillStats, + getHistoricalEventHourlyDelta, + getHistoricalStarHourlyDelta, + getCompletedRolling24HourWindow, + sumRollingHourlyStats, +} from "./skillHourlyStats"; + +const modules = import.meta.glob("../**/*.ts"); + +describe("rolling 24-hour skill metrics", () => { + it("uses exactly the latest 24 complete hourly buckets", () => { + const now = 100 * HOUR_MS + 37 * 60 * 1_000; + + expect(getCompletedRolling24HourWindow(now)).toEqual({ + startHour: 76, + endHour: 99, + startAt: 76 * HOUR_MS, + endAt: 100 * HOUR_MS, + }); + }); + + it("combines the historical seed and live deltas without exposing negative totals", () => { + expect( + sumRollingHourlyStats([ + { + skillId: "skills:one", + downloads: 6, + installs: 8, + bookmarks: 0, + updatedAt: 10, + }, + { + skillId: "skills:one", + downloads: 1, + installs: 2, + bookmarks: 0, + updatedAt: 20, + }, + ]), + ).toEqual( + new Map([ + [ + "skills:one", + { + downloads: 7, + installs: 10, + bookmarks: 0, + updatedAt: 20, + }, + ], + ]), + ); + }); + + it("keeps historical seed counts separate from concurrent live deltas", async () => { + const t = convexTest(schema, modules); + const hourAt = 90 * HOUR_MS; + const nowSpy = vi.spyOn(Date, "now").mockReturnValue(100 * HOUR_MS); + const rows = await t.run(async (ctx) => { + const now = Date.now(); + const userId = await ctx.db.insert("users", { + handle: "hourly-owner", + createdAt: now, + updatedAt: now, + }); + const skillId = await ctx.db.insert("skills", { + slug: "hourly-skill", + displayName: "Hourly skill", + ownerUserId: userId, + tags: {}, + stats: { downloads: 0, stars: 0, versions: 0, comments: 0 }, + createdAt: now, + updatedAt: now, + }); + + await bumpHistoricalHourlySkillStats(ctx, { + skillId, + occurredAt: hourAt, + downloads: 2, + installs: 3, + bookmarks: 1, + }); + await bumpLiveHourlySkillStats(ctx, { + skillId, + occurredAt: hourAt, + downloads: 4, + installs: 5, + bookmarks: -1, + }); + + return await ctx.db + .query("skillHourlyStats") + .withIndex("by_skill_and_hour_and_generation", (q) => + q.eq("skillId", skillId).eq("hour", 90), + ) + .collect(); + }); + nowSpy.mockRestore(); + + expect(rows).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + hour: 90, + generation: 0, + downloads: 2, + installs: 3, + bookmarks: 1, + }), + expect.objectContaining({ + hour: 90, + generation: 1, + downloads: 4, + installs: 5, + bookmarks: -1, + }), + ]), + ); + }); + + it("captures the existing event cursor before the first live hourly write", async () => { + const t = convexTest(schema, modules); + const now = 200 * HOUR_MS; + const nowSpy = vi.spyOn(Date, "now").mockReturnValue(now); + const state = await t.run(async (ctx) => { + const userId = await ctx.db.insert("users", { + handle: "hourly-state-owner", + createdAt: now, + updatedAt: now, + }); + const skillId = await ctx.db.insert("skills", { + slug: "hourly-state-skill", + displayName: "Hourly state skill", + ownerUserId: userId, + tags: {}, + stats: { downloads: 0, stars: 0, versions: 0, comments: 0 }, + createdAt: now, + updatedAt: now, + }); + await ctx.db.insert("skillStatUpdateCursors", { + key: "skill_stat_events", + cursorCreationTime: 500, + updatedAt: now - 1, + }); + + await bumpLiveHourlySkillStats(ctx, { + skillId, + occurredAt: now, + downloads: 1, + }); + return await ctx.db + .query("skillHourlyStatStates") + .withIndex("by_key", (q) => q.eq("key", "canonical_trending")) + .unique(); + }); + nowSpy.mockRestore(); + + expect(state).toMatchObject({ + key: "canonical_trending", + liveStartedAt: now, + eventBackfillThroughCreationTime: 500, + activeGeneration: 1, + }); + }); + + it("seeds only retained activity on the historical side of the live boundary", () => { + const now = 200 * HOUR_MS; + const state = { + liveStartedAt: 190 * HOUR_MS, + eventBackfillThroughCreationTime: 500, + }; + + expect( + getHistoricalEventHourlyDelta( + { kind: "download", occurredAt: 128 * HOUR_MS, _creationTime: 500 }, + state, + now, + ), + ).toEqual({ downloads: 1 }); + expect( + getHistoricalEventHourlyDelta( + { kind: "install_new", occurredAt: 128 * HOUR_MS, _creationTime: 501 }, + state, + now, + ), + ).toBeNull(); + expect( + getHistoricalEventHourlyDelta( + { kind: "download", occurredAt: 127 * HOUR_MS, _creationTime: 499 }, + state, + now, + ), + ).toBeNull(); + expect(getHistoricalStarHourlyDelta({ createdAt: 189 * HOUR_MS }, state, now)).toEqual({ + bookmarks: 1, + }); + expect(getHistoricalStarHourlyDelta({ createdAt: 191 * HOUR_MS }, state, now)).toBeNull(); + expect( + getHistoricalStarHourlyDelta( + { createdAt: 189 * HOUR_MS, hourlyStatsRecordedAt: 195 * HOUR_MS }, + state, + now, + ), + ).toBeNull(); + }); + + it("moves live writes to a new generation when a snapshot seals its cutoff", async () => { + const t = convexTest(schema, modules); + const now = Date.now(); + const skillId = await t.run(async (ctx) => { + const userId = await ctx.db.insert("users", { + handle: "hourly-generation-owner", + createdAt: now, + updatedAt: now, + }); + const id = await ctx.db.insert("skills", { + slug: "hourly-generation-skill", + displayName: "Hourly generation skill", + ownerUserId: userId, + tags: {}, + stats: { downloads: 0, stars: 0, versions: 0, comments: 0 }, + createdAt: now, + updatedAt: now, + }); + await ctx.db.insert("skillHourlyStatStates", { + key: "canonical_trending", + liveStartedAt: now - HOUR_MS, + eventBackfillThroughCreationTime: 100, + activeGeneration: 1, + backfillCompletedAt: now - HOUR_MS, + lastAggregationCompletedAt: now - 1, + updatedAt: now - 1, + }); + return id; + }); + + await expect( + t.mutation(internal.skillHourlyStats.sealForSnapshotInternal, { now }), + ).resolves.toMatchObject({ sealedGeneration: 1 }); + await t.run(async (ctx) => { + await bumpLiveHourlySkillStats(ctx, { + skillId, + occurredAt: now - HOUR_MS, + downloads: 1, + }); + }); + + await expect( + t.run( + async (ctx) => + await ctx.db + .query("skillHourlyStats") + .withIndex("by_skill_and_hour_and_generation", (q) => q.eq("skillId", skillId)) + .collect(), + ), + ).resolves.toEqual([expect.objectContaining({ generation: 2, downloads: 1 })]); + }); + + it("records processed downloads and first installs in their original hourly buckets", async () => { + const t = convexTest(schema, modules); + const nowSpy = vi.spyOn(Date, "now").mockReturnValue(100 * HOUR_MS); + const userId = await t.run(async (ctx) => { + const now = Date.now(); + return await ctx.db.insert("users", { + handle: "hourly-processor-owner", + createdAt: now, + updatedAt: now, + }); + }); + const skillId = await t.run(async (ctx) => { + const now = Date.now(); + return await ctx.db.insert("skills", { + slug: "hourly-processor-skill", + displayName: "Hourly processor skill", + ownerUserId: userId, + tags: {}, + stats: { downloads: 0, stars: 0, versions: 0, comments: 0 }, + createdAt: now, + updatedAt: now, + }); + }); + + await t.mutation(internal.skillStatEvents.applyAggregatedStatsAndUpdateCursor, { + skillDeltas: [ + { + skillId, + downloads: 1, + stars: 0, + installsAllTime: 1, + installsCurrent: 1, + downloadEvents: [80 * HOUR_MS + 1], + installNewEvents: [81 * HOUR_MS + 1], + }, + ], + newCursor: 123, + }); + + const rows = await t.run( + async (ctx) => + await ctx.db + .query("skillHourlyStats") + .withIndex("by_skill_and_hour_and_generation", (q) => q.eq("skillId", skillId)) + .collect(), + ); + nowSpy.mockRestore(); + expect(rows.map(({ hour, downloads, installs }) => ({ hour, downloads, installs }))).toEqual([ + { hour: 80, downloads: 1, installs: 0 }, + { hour: 81, downloads: 0, installs: 1 }, + ]); + }); + + it("prunes only expired hourly buckets in bounded batches", async () => { + const t = convexTest(schema, modules); + const now = Date.now(); + const nowSpy = vi.spyOn(Date, "now").mockReturnValue(now); + const { expiredId, freshId } = await t.run(async (ctx) => { + const userId = await ctx.db.insert("users", { + handle: "hourly-prune-owner", + createdAt: now, + updatedAt: now, + }); + const skillId = await ctx.db.insert("skills", { + slug: "hourly-prune-skill", + displayName: "Hourly prune skill", + ownerUserId: userId, + tags: {}, + stats: { downloads: 0, stars: 0, versions: 0, comments: 0 }, + createdAt: now, + updatedAt: now, + }); + const base = { + skillId, + generation: 1, + downloads: 1, + installs: 0, + bookmarks: 0, + updatedAt: now, + }; + return { + expiredId: await ctx.db.insert("skillHourlyStats", { + ...base, + hour: 1, + expiresAt: now - 1, + }), + freshId: await ctx.db.insert("skillHourlyStats", { + ...base, + hour: 2, + expiresAt: now + 1, + }), + }; + }); + + await expect( + t.mutation(internal.skillHourlyStats.pruneExpiredInternal, { batchSize: 10 }), + ).resolves.toEqual({ deleted: 1, hasMore: false }); + await expect( + t.run(async (ctx) => ({ + expired: await ctx.db.get(expiredId), + fresh: await ctx.db.get(freshId), + })), + ).resolves.toMatchObject({ expired: null, fresh: expect.objectContaining({ _id: freshId }) }); + nowSpy.mockRestore(); + }); + + it("requires an explicit confirmation token before applying the historical seed", async () => { + const t = convexTest(schema, modules); + + await expect( + t.action(internal.migrations.runSkillHourlyStatsBackfill, { dryRun: false }), + ).rejects.toThrow('Pass confirm="apply-skill-hourly-stats-backfill" to apply.'); + }); +}); diff --git a/convex/lib/skillHourlyStats.ts b/convex/lib/skillHourlyStats.ts new file mode 100644 index 00000000..2dd157bb --- /dev/null +++ b/convex/lib/skillHourlyStats.ts @@ -0,0 +1,188 @@ +import type { Id } from "../_generated/dataModel"; +import type { MutationCtx } from "../_generated/server"; + +export const HOUR_MS = 60 * 60 * 1_000; +export const ROLLING_TRENDING_HOURS = 24; +export const HOURLY_STATS_RETENTION_MS = 72 * HOUR_MS; +export const HOURLY_STATS_STATE_KEY = "canonical_trending"; +const SKILL_STAT_EVENT_CURSOR_KEY = "skill_stat_events"; + +export function toHourKey(timestamp: number) { + return Math.floor(timestamp / HOUR_MS); +} + +export function getHourlyStatExpiresAt(timestamp: number) { + return (toHourKey(timestamp) + 1) * HOUR_MS + HOURLY_STATS_RETENTION_MS; +} + +export function getCompletedRolling24HourWindow(now: number) { + const currentHour = toHourKey(now); + const endHour = currentHour - 1; + const startHour = endHour - (ROLLING_TRENDING_HOURS - 1); + return { + startHour, + endHour, + startAt: startHour * HOUR_MS, + endAt: currentHour * HOUR_MS, + }; +} + +type HourlySkillStat = { + skillId: string; + downloads: number; + installs: number; + bookmarks: number; + updatedAt: number; +}; + +export function sumRollingHourlyStats(rows: readonly HourlySkillStat[]) { + const totals = new Map< + string, + { downloads: number; installs: number; bookmarks: number; updatedAt: number } + >(); + for (const row of rows) { + const current = totals.get(row.skillId) ?? { + downloads: 0, + installs: 0, + bookmarks: 0, + updatedAt: 0, + }; + current.downloads += row.downloads; + current.installs += row.installs; + current.bookmarks += row.bookmarks; + current.updatedAt = Math.max(current.updatedAt, row.updatedAt); + totals.set(row.skillId, current); + } + for (const total of totals.values()) { + total.downloads = Math.max(0, total.downloads); + total.installs = Math.max(0, total.installs); + total.bookmarks = Math.max(0, total.bookmarks); + } + return totals; +} + +type HourlyStatDeltas = { + skillId: Id<"skills">; + occurredAt: number; + downloads?: number; + installs?: number; + bookmarks?: number; +}; + +type HourlyStatsBackfillState = { + liveStartedAt: number; + eventBackfillThroughCreationTime: number; +}; + +export function getHistoricalEventHourlyDelta( + event: { kind: string; occurredAt: number; _creationTime: number }, + state: HourlyStatsBackfillState, + now: number, +) { + if ( + event._creationTime > state.eventBackfillThroughCreationTime || + getHourlyStatExpiresAt(event.occurredAt) <= now + ) { + return null; + } + if (event.kind === "download") return { downloads: 1 }; + if (event.kind === "install_new") return { installs: 1 }; + return null; +} + +export function getHistoricalStarHourlyDelta( + star: { createdAt: number; hourlyStatsRecordedAt?: number }, + state: HourlyStatsBackfillState, + now: number, +) { + if ( + star.hourlyStatsRecordedAt !== undefined || + star.createdAt >= state.liveStartedAt || + getHourlyStatExpiresAt(star.createdAt) <= now + ) { + return null; + } + return { bookmarks: 1 }; +} + +export async function ensureHourlyStatsState(ctx: Pick) { + const existing = await ctx.db + .query("skillHourlyStatStates") + .withIndex("by_key", (q) => q.eq("key", HOURLY_STATS_STATE_KEY)) + .unique(); + if (existing) return existing; + + const cursor = await ctx.db + .query("skillStatUpdateCursors") + .withIndex("by_key", (q) => q.eq("key", SKILL_STAT_EVENT_CURSOR_KEY)) + .unique(); + const now = Date.now(); + const stateId = await ctx.db.insert("skillHourlyStatStates", { + key: HOURLY_STATS_STATE_KEY, + liveStartedAt: now, + eventBackfillThroughCreationTime: cursor?.cursorCreationTime ?? 0, + activeGeneration: 1, + updatedAt: now, + }); + const created = await ctx.db.get(stateId); + if (!created) throw new Error("Failed to initialize hourly skill stat state"); + return created; +} + +async function bumpHourlySkillStats( + ctx: Pick, + generation: number, + params: HourlyStatDeltas, +) { + const hour = toHourKey(params.occurredAt); + const existing = await ctx.db + .query("skillHourlyStats") + .withIndex("by_skill_and_hour_and_generation", (q) => + q.eq("skillId", params.skillId).eq("hour", hour).eq("generation", generation), + ) + .unique(); + const now = Date.now(); + const expiresAt = getHourlyStatExpiresAt(params.occurredAt); + if (expiresAt <= now) return null; + const downloads = params.downloads ?? 0; + const installs = params.installs ?? 0; + const bookmarks = params.bookmarks ?? 0; + + if (!existing) { + return await ctx.db.insert("skillHourlyStats", { + skillId: params.skillId, + hour, + generation, + downloads, + installs, + bookmarks, + updatedAt: now, + expiresAt, + }); + } + + await ctx.db.patch(existing._id, { + downloads: existing.downloads + downloads, + installs: existing.installs + installs, + bookmarks: existing.bookmarks + bookmarks, + updatedAt: now, + expiresAt, + }); + return existing._id; +} + +export async function bumpHistoricalHourlySkillStats( + ctx: Pick, + params: HourlyStatDeltas, +) { + return await bumpHourlySkillStats(ctx, 0, params); +} + +export async function bumpLiveHourlySkillStats( + ctx: Pick, + params: HourlyStatDeltas, + options?: { state?: { activeGeneration: number } }, +) { + const state = options?.state ?? (await ensureHourlyStatsState(ctx)); + return await bumpHourlySkillStats(ctx, state.activeGeneration, params); +} diff --git a/convex/migrations.ts b/convex/migrations.ts index 92f792e6..7dc25bcf 100644 --- a/convex/migrations.ts +++ b/convex/migrations.ts @@ -20,6 +20,12 @@ import { DOWNLOAD_BACKFILL_MODEL_VERSION, NVIDIA_GITHUB_DOWNLOAD_BACKFILL_SOURCE_REPO, } from "./lib/skillDownloadBackfill"; +import { + bumpHistoricalHourlySkillStats, + getHistoricalEventHourlyDelta, + getHistoricalStarHourlyDelta, + HOURLY_STATS_STATE_KEY, +} from "./lib/skillHourlyStats"; import { buildSkillInstallBackfillPatch, INSTALL_BACKFILL_CLEAN_WINDOW, @@ -36,6 +42,7 @@ const APPLY_SKILL_INSTALL_BACKFILL_CONFIRM = "apply-skill-install-backfill"; const APPLY_NVIDIA_GITHUB_DOWNLOAD_BACKFILL_CONFIRM = "apply-nvidia-github-download-backfill"; const BACKFILL_PLUGIN_MANIFEST_SUMMARIES_CONFIRM = "backfill-plugin-manifest-summaries"; const RECOVER_SUSPICIOUS_PUBLISH_ATTEMPTS_CONFIRM = "recover-suspicious-publish-attempts"; +const APPLY_SKILL_HOURLY_STATS_BACKFILL_CONFIRM = "apply-skill-hourly-stats-backfill"; const SKILL_STAT_EVENTS_CURSOR_KEY = "skill_stat_events"; const MAX_PENDING_SKILL_STAT_EVENTS_PER_SKILL = 1_000; const PLUGIN_PACKAGE_FAMILIES = ["code-plugin", "bundle-plugin"] as const; @@ -82,6 +89,48 @@ export const migrations = new Migrations(components.migrations, { defaultBatchSize: 25, }); +async function requireSkillHourlyStatsBackfillState(ctx: Pick) { + const state = await ctx.db + .query("skillHourlyStatStates") + .withIndex("by_key", (q) => q.eq("key", HOURLY_STATS_STATE_KEY)) + .unique(); + if (!state) throw new ConvexError("Initialize hourly skill stats before running the backfill."); + return state; +} + +export const backfillSkillHourlyStatsFromEvents = migrations.define({ + table: "skillStatEvents", + batchSize: 100, + migrateOne: async (ctx, event) => { + const state = await requireSkillHourlyStatsBackfillState(ctx); + const delta = getHistoricalEventHourlyDelta(event, state, Date.now()); + if (!delta) return; + await bumpHistoricalHourlySkillStats(ctx, { + skillId: event.skillId, + occurredAt: event.occurredAt, + ...delta, + }); + }, +}); + +export const backfillSkillHourlyStatsFromStars = migrations.define({ + table: "stars", + batchSize: 100, + migrateOne: async (ctx, star) => { + const state = await requireSkillHourlyStatsBackfillState(ctx); + const delta = getHistoricalStarHourlyDelta(star, state, Date.now()); + if (!delta) return; + const hourlyStatId = await bumpHistoricalHourlySkillStats(ctx, { + skillId: star.skillId, + occurredAt: star.createdAt, + ...delta, + }); + if (hourlyStatId) { + await ctx.db.patch(star._id, { hourlyStatsRecordedAt: Date.now() }); + } + }, +}); + type SuspiciousPublishAttemptRecoveryClassification = | "replay_missing" | "replay_identical" @@ -1296,6 +1345,72 @@ export const runPluginManifestSummaryBackfillPage = internalAction({ export const run = migrations.runner(); +type SkillHourlyStatsBackfillRunResult = { + ok: true; + dryRun: boolean; + confirmRequired?: string; + liveStartedAt: number; + eventBackfillThroughCreationTime: number; + backfillCompletedAt?: number; +}; + +export const runSkillHourlyStatsBackfill: ReturnType = internalAction({ + args: { + dryRun: v.optional(v.boolean()), + confirm: v.optional(v.string()), + }, + handler: async (ctx, args): Promise => { + const dryRun = args.dryRun !== false; + if (!dryRun && args.confirm !== APPLY_SKILL_HOURLY_STATS_BACKFILL_CONFIRM) { + throw new ConvexError( + `Pass confirm="${APPLY_SKILL_HOURLY_STATS_BACKFILL_CONFIRM}" to apply.`, + ); + } + + const initialized: Doc<"skillHourlyStatStates"> = await ctx.runMutation( + internal.skillHourlyStats.initializeInternal, + {}, + ); + if (dryRun) { + for (const fn of [ + "migrations:backfillSkillHourlyStatsFromEvents", + "migrations:backfillSkillHourlyStatsFromStars", + ]) { + await ctx.runMutation(internal.migrations.run, { + fn, + dryRun: true, + reset: true, + }); + } + } else { + await runToCompletion( + ctx, + components.migrations, + internal.migrations.backfillSkillHourlyStatsFromEvents, + ); + await runToCompletion( + ctx, + components.migrations, + internal.migrations.backfillSkillHourlyStatsFromStars, + ); + await ctx.runMutation(internal.skillHourlyStats.markBackfillCompletedInternal, {}); + } + const state: Doc<"skillHourlyStatStates"> | null = await ctx.runQuery( + internal.skillHourlyStats.getStateInternal, + {}, + ); + + return { + ok: true as const, + dryRun, + confirmRequired: dryRun ? APPLY_SKILL_HOURLY_STATS_BACKFILL_CONFIRM : undefined, + liveStartedAt: initialized.liveStartedAt, + eventBackfillThroughCreationTime: initialized.eventBackfillThroughCreationTime, + backfillCompletedAt: state?.backfillCompletedAt, + }; + }, +}); + export const runSuspiciousPublishAttemptRecovery: ReturnType = internalAction({ args: { diff --git a/convex/rolloutCapabilities.test.ts b/convex/rolloutCapabilities.test.ts index 730820d6..8e53ae35 100644 --- a/convex/rolloutCapabilities.test.ts +++ b/convex/rolloutCapabilities.test.ts @@ -32,6 +32,7 @@ describe("getPublicCapabilitiesHandler", () => { environment: "unknown", catalogDiscovery: { apiVersion: 1, + canonicalTrendingEnabled: true, }, skillsSh: { mode: "off", @@ -71,6 +72,7 @@ describe("getPublicCapabilitiesHandler", () => { environment: "test", catalogDiscovery: { apiVersion: 1, + canonicalTrendingEnabled: true, }, skillsSh: { mode: "test", diff --git a/convex/rolloutCapabilities.ts b/convex/rolloutCapabilities.ts index 29c433a4..99aa6863 100644 --- a/convex/rolloutCapabilities.ts +++ b/convex/rolloutCapabilities.ts @@ -22,6 +22,10 @@ export async function getPublicCapabilitiesHandler( environment: runtime.environment, catalogDiscovery: { apiVersion: 1, + // This advertises the canonical API contract, not row availability. + // Clients must use its unavailable/empty states instead of falling back + // to legacy seven-day or lifetime popularity under a 24-hour label. + canonicalTrendingEnabled: true, }, skillsSh: { mode: runtime.skillsSh.mode, diff --git a/convex/schema.ts b/convex/schema.ts index bfa4f8af..edbafd4f 100644 --- a/convex/schema.ts +++ b/convex/schema.ts @@ -2618,6 +2618,31 @@ const skillDailyStats = defineTable({ .index("by_skill_day", ["skillId", "day"]) .index("by_day", ["day"]); +const skillHourlyStats = defineTable({ + skillId: v.id("skills"), + hour: v.number(), + generation: v.number(), + downloads: v.number(), + installs: v.number(), + bookmarks: v.number(), + updatedAt: v.number(), + expiresAt: v.number(), +}) + .index("by_skill_and_hour_and_generation", ["skillId", "hour", "generation"]) + .index("by_hour", ["hour"]) + .index("by_expires_at", ["expiresAt"]); + +const skillHourlyStatStates = defineTable({ + key: v.string(), + liveStartedAt: v.number(), + eventBackfillThroughCreationTime: v.number(), + activeGeneration: v.number(), + backfillCompletedAt: v.optional(v.number()), + lastAggregationCompletedAt: v.optional(v.number()), + lastProcessedEventCreationTime: v.optional(v.number()), + updatedAt: v.number(), +}).index("by_key", ["key"]); + const skillLeaderboards = defineTable({ kind: v.string(), generatedAt: v.number(), @@ -2678,6 +2703,8 @@ const canonicalTrendingSnapshots = defineTable({ windowHours: v.number(), windowStartDay: v.number(), windowEndDay: v.number(), + windowStartHour: v.optional(v.number()), + windowEndHour: v.optional(v.number()), writtenItems: v.number(), totalItems: v.optional(v.number()), sourceCounts: v.optional( @@ -2923,6 +2950,7 @@ const stars = defineTable({ skillId: v.id("skills"), userId: v.id("users"), createdAt: v.number(), + hourlyStatsRecordedAt: v.optional(v.number()), }) .index("by_skill", ["skillId"]) .index("by_user", ["userId"]) @@ -4184,6 +4212,8 @@ export default defineSchema({ curatedSkillSearchDigest, skillTopicSearchDigest, skillDailyStats, + skillHourlyStats, + skillHourlyStatStates, skillLeaderboards, skillStatBackfillState, globalStats, diff --git a/convex/skillHourlyStats.ts b/convex/skillHourlyStats.ts new file mode 100644 index 00000000..664fe28e --- /dev/null +++ b/convex/skillHourlyStats.ts @@ -0,0 +1,107 @@ +import { v } from "convex/values"; +import { internal } from "./_generated/api"; +import { internalMutation, internalQuery } from "./functions"; +import { RETENTION_STANDARD_BATCH_SIZE } from "./lib/retentionPolicy"; +import { + ensureHourlyStatsState, + getCompletedRolling24HourWindow, + HOURLY_STATS_STATE_KEY, +} from "./lib/skillHourlyStats"; + +const MAX_PRUNE_BATCH_SIZE = 1_000; +const HOURLY_AGGREGATION_MAX_AGE_MS = 2 * 60 * 60 * 1_000; + +function normalizeBatchSize(value: number | undefined) { + if (!Number.isFinite(value)) return RETENTION_STANDARD_BATCH_SIZE; + return Math.max( + 1, + Math.min(Math.trunc(value ?? RETENTION_STANDARD_BATCH_SIZE), MAX_PRUNE_BATCH_SIZE), + ); +} + +export const initializeInternal = internalMutation({ + args: {}, + handler: async (ctx) => await ensureHourlyStatsState(ctx), +}); + +export const getStateInternal = internalQuery({ + args: {}, + handler: async (ctx) => + await ctx.db + .query("skillHourlyStatStates") + .withIndex("by_key", (q) => q.eq("key", HOURLY_STATS_STATE_KEY)) + .unique(), +}); + +export const sealForSnapshotInternal = internalMutation({ + args: { now: v.number() }, + handler: async (ctx, args) => { + const state = await ctx.db + .query("skillHourlyStatStates") + .withIndex("by_key", (q) => q.eq("key", HOURLY_STATS_STATE_KEY)) + .unique(); + if ( + !state?.backfillCompletedAt || + !state.lastAggregationCompletedAt || + state.lastAggregationCompletedAt <= args.now - HOURLY_AGGREGATION_MAX_AGE_MS + ) { + return null; + } + + const sealedGeneration = state.activeGeneration; + await ctx.db.patch(state._id, { + activeGeneration: sealedGeneration + 1, + updatedAt: args.now, + }); + return { + ...getCompletedRolling24HourWindow(Math.min(args.now, state.lastAggregationCompletedAt)), + lastAggregationCompletedAt: state.lastAggregationCompletedAt, + sealedGeneration, + }; + }, +}); + +export const markBackfillCompletedInternal = internalMutation({ + args: {}, + handler: async (ctx) => { + const state = await ensureHourlyStatsState(ctx); + const now = Date.now(); + await ctx.db.patch(state._id, { backfillCompletedAt: now, updatedAt: now }); + return { backfillCompletedAt: now }; + }, +}); + +export const markAggregationCompletedInternal = internalMutation({ + args: { cursorCreationTime: v.optional(v.number()) }, + handler: async (ctx, args) => { + const state = await ensureHourlyStatsState(ctx); + const now = Date.now(); + await ctx.db.patch(state._id, { + lastAggregationCompletedAt: now, + lastProcessedEventCreationTime: + args.cursorCreationTime ?? state.lastProcessedEventCreationTime, + updatedAt: now, + }); + return { completedAt: now }; + }, +}); + +export const pruneExpiredInternal = internalMutation({ + args: { batchSize: v.optional(v.number()) }, + handler: async (ctx, args) => { + const batchSize = normalizeBatchSize(args.batchSize); + const rows = await ctx.db + .query("skillHourlyStats") + .withIndex("by_expires_at", (q) => q.lte("expiresAt", Date.now())) + .take(batchSize); + for (const row of rows) await ctx.db.delete(row._id); + + const hasMore = rows.length === batchSize; + if (hasMore) { + await ctx.scheduler.runAfter(0, internal.skillHourlyStats.pruneExpiredInternal, { + batchSize, + }); + } + return { deleted: rows.length, hasMore }; + }, +}); diff --git a/convex/skillStatEvents.test.ts b/convex/skillStatEvents.test.ts index a9466da0..ef3065e4 100644 --- a/convex/skillStatEvents.test.ts +++ b/convex/skillStatEvents.test.ts @@ -7,6 +7,7 @@ const apiRefs = vi.hoisted(() => ({ getStatEventCursor: Symbol("getStatEventCursor"), getUnprocessedEventBatch: Symbol("getUnprocessedEventBatch"), kickProcessedSkillStatEventPruneInternal: Symbol("kickProcessedSkillStatEventPruneInternal"), + markHourlyAggregationCompletedInternal: Symbol("markHourlyAggregationCompletedInternal"), processSkillStatEventBatchInternal: Symbol("processSkillStatEventBatchInternal"), processSkillStatEventsAction: Symbol("processSkillStatEventsAction"), processSkillStatEventsInternal: Symbol("processSkillStatEventsInternal"), @@ -36,6 +37,9 @@ vi.mock("./_generated/api", () => ({ pruneProcessedSkillStatEventBatchInternal: apiRefs.pruneProcessedSkillStatEventBatchInternal, releaseSkillStatDocSyncLeaseInternal: apiRefs.releaseSkillStatDocSyncLeaseInternal, }, + skillHourlyStats: { + markAggregationCompletedInternal: apiRefs.markHourlyAggregationCompletedInternal, + }, }, })); @@ -175,6 +179,9 @@ describe("skill stat events", () => { ], newCursor: 456, }); + expect(runMutation).toHaveBeenCalledWith(apiRefs.markHourlyAggregationCompletedInternal, { + cursorCreationTime: 456, + }); expect(scheduler.runAfter).not.toHaveBeenCalled(); }); diff --git a/convex/skillStatEvents.ts b/convex/skillStatEvents.ts index 460d4fd3..7392f8f9 100644 --- a/convex/skillStatEvents.ts +++ b/convex/skillStatEvents.ts @@ -23,6 +23,11 @@ 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 { + bumpLiveHourlySkillStats, + ensureHourlyStatsState, + toHourKey, +} from "./lib/skillHourlyStats"; import { applySkillStatDeltas, bumpDailySkillStats } from "./lib/skillStats"; import { adjustUserSkillStatsForSkillChange } from "./lib/userSkillStats"; @@ -1153,6 +1158,10 @@ export const applyAggregatedStatsAndUpdateCursor = internalMutation({ string, { skillId: Id<"skills">; occurredAt: number; downloads: number; installs: number } >(); + const hourlyStats = new Map< + string, + { skillId: Id<"skills">; occurredAt: number; downloads: number; installs: number } + >(); for (const delta of args.skillDeltas) { for (const occurredAt of delta.downloadEvents) { @@ -1165,6 +1174,16 @@ export const applyAggregatedStatsAndUpdateCursor = internalMutation({ }; current.downloads += 1; dailyStats.set(key, current); + + const hourlyKey = `${delta.skillId}:${toHourKey(occurredAt)}`; + const hourly = hourlyStats.get(hourlyKey) ?? { + skillId: delta.skillId, + occurredAt, + downloads: 0, + installs: 0, + }; + hourly.downloads += 1; + hourlyStats.set(hourlyKey, hourly); } for (const occurredAt of delta.installNewEvents) { const key = `${delta.skillId}:${toDayKey(occurredAt)}`; @@ -1176,6 +1195,16 @@ export const applyAggregatedStatsAndUpdateCursor = internalMutation({ }; current.installs += 1; dailyStats.set(key, current); + + const hourlyKey = `${delta.skillId}:${toHourKey(occurredAt)}`; + const hourly = hourlyStats.get(hourlyKey) ?? { + skillId: delta.skillId, + occurredAt, + downloads: 0, + installs: 0, + }; + hourly.installs += 1; + hourlyStats.set(hourlyKey, hourly); } } @@ -1188,6 +1217,11 @@ export const applyAggregatedStatsAndUpdateCursor = internalMutation({ }); } + const hourlyState = hourlyStats.size > 0 ? await ensureHourlyStatsState(ctx) : null; + for (const stat of hourlyStats.values()) { + await bumpLiveHourlySkillStats(ctx, stat, { state: hourlyState! }); + } + // Update cursor position (upsert) const existingCursor = await ctx.db .query("skillStatUpdateCursors") @@ -1335,6 +1369,9 @@ export const processSkillStatEventsAction = internalAction({ // If we have nothing to process, we're done if (aggregatedBySkill.size === 0 || maxCreationTime === undefined) { console.log("[STAT-AGG] No events to process, done"); + await ctx.runMutation(internal.skillHourlyStats.markAggregationCompletedInternal, { + cursorCreationTime: cursor, + }); return { processed: 0, skillsUpdated: 0, exhausted: true }; } @@ -1360,6 +1397,9 @@ export const processSkillStatEventsAction = internalAction({ await ctx.scheduler.runAfter(0, internal.skillStatEvents.processSkillStatEventsAction, {}); } else { console.log("[STAT-AGG] All events processed, done"); + await ctx.runMutation(internal.skillHourlyStats.markAggregationCompletedInternal, { + cursorCreationTime: maxCreationTime, + }); } return { diff --git a/convex/stars.test.ts b/convex/stars.test.ts index 1b914ba9..875b31f7 100644 --- a/convex/stars.test.ts +++ b/convex/stars.test.ts @@ -6,6 +6,11 @@ vi.mock("./skillStatEvents", () => ({ insertStatEvent: vi.fn(), })); +vi.mock("./lib/skillHourlyStats", () => ({ + bumpLiveHourlySkillStats: vi.fn(), + ensureHourlyStatsState: vi.fn(async () => ({ activeGeneration: 1 })), +})); + vi.mock("@convex-dev/auth/server", () => ({ getAuthUserId: vi.fn(), authTables: {}, @@ -18,6 +23,7 @@ vi.mock("./functions", () => ({ })); const { insertStatEvent } = await import("./skillStatEvents"); +const { bumpLiveHourlySkillStats, ensureHourlyStatsState } = await import("./lib/skillHourlyStats"); const { addStarInternal, isStarred, removeStarInternal, toggle } = await import("./stars"); type WrappedHandler = { @@ -103,6 +109,8 @@ describe("stars mutations", () => { afterEach(() => { vi.mocked(getAuthUserId).mockReset(); vi.mocked(insertStatEvent).mockReset(); + vi.mocked(bumpLiveHourlySkillStats).mockReset(); + vi.mocked(ensureHourlyStatsState).mockClear(); }); it("toggle inserts a star row and updates denormalized star counts synchronously", async () => { @@ -116,6 +124,7 @@ describe("stars mutations", () => { skillId: "skills:1", userId: "users:viewer", createdAt: expect.any(Number), + hourlyStatsRecordedAt: expect.any(Number), }); expect(db.patch).toHaveBeenCalledWith( "skills:1", @@ -129,6 +138,15 @@ describe("stars mutations", () => { expect.objectContaining({ totalStars: 3 }), ); expect(insertStatEvent).not.toHaveBeenCalled(); + expect(bumpLiveHourlySkillStats).toHaveBeenCalledWith( + ctx, + { + skillId: "skills:1", + occurredAt: expect.any(Number), + bookmarks: 1, + }, + { state: { activeGeneration: 1 } }, + ); }); it("toggle deletes a star row and decrements counts without going below zero", async () => { @@ -145,7 +163,13 @@ describe("stars mutations", () => { comments: 0, }, }), - existingStar: { _id: "stars:1", skillId: "skills:1", userId: "users:viewer" }, + existingStar: { + _id: "stars:1", + skillId: "skills:1", + userId: "users:viewer", + createdAt: 1234, + hourlyStatsRecordedAt: 1200, + }, }); const result = await toggleHandler(ctx, { skillId: "skills:1" }); @@ -160,6 +184,32 @@ describe("stars mutations", () => { }), ); expect(insertStatEvent).not.toHaveBeenCalled(); + expect(bumpLiveHourlySkillStats).toHaveBeenCalledWith( + ctx, + { + skillId: "skills:1", + occurredAt: 1234, + bookmarks: -1, + }, + { state: { activeGeneration: 1 } }, + ); + }); + + it("does not write an unmatched hourly decrement for an unbackfilled legacy star", async () => { + vi.mocked(getAuthUserId).mockResolvedValue("users:viewer" as never); + const { ctx } = makeCtx({ + existingStar: { + _id: "stars:legacy", + skillId: "skills:1", + userId: "users:viewer", + createdAt: 1234, + }, + }); + + await expect(toggleHandler(ctx, { skillId: "skills:1" })).resolves.toEqual({ + starred: false, + }); + expect(bumpLiveHourlySkillStats).not.toHaveBeenCalled(); }); it("addStarInternal is idempotent and increments only when inserting a row", async () => { diff --git a/convex/stars.ts b/convex/stars.ts index 2efe794e..222a12a8 100644 --- a/convex/stars.ts +++ b/convex/stars.ts @@ -4,14 +4,32 @@ import type { MutationCtx } from "./_generated/server"; import { internalMutation, mutation, query } from "./functions"; import { getOptionalActiveAuthUserId, requireUser } from "./lib/access"; import { toPublicSkill } from "./lib/public"; +import { bumpLiveHourlySkillStats, ensureHourlyStatsState } from "./lib/skillHourlyStats"; import { applySkillStatDeltas } from "./lib/skillStats"; import { adjustUserSkillStatsForSkillChange } from "./lib/userSkillStats"; -async function applyStarDelta(ctx: Pick, skill: Doc<"skills">, delta: 1 | -1) { +async function applyStarDelta( + ctx: Pick, + skill: Doc<"skills">, + delta: 1 | -1, + occurredAt: number, + hourlyState?: { activeGeneration: number }, +) { const patch = applySkillStatDeltas(skill, { stars: delta }); const nextSkill = { ...skill, ...patch }; await ctx.db.patch(skill._id, patch); await adjustUserSkillStatsForSkillChange(ctx, skill, nextSkill); + if (hourlyState) { + await bumpLiveHourlySkillStats( + ctx, + { + skillId: skill._id, + occurredAt, + bookmarks: delta, + }, + { state: hourlyState }, + ); + } } export const isStarred = query({ @@ -41,19 +59,26 @@ export const toggle = mutation({ if (existing) { await ctx.db.delete(existing._id); - await applyStarDelta(ctx, skill, -1); + const hourlyState = + existing.hourlyStatsRecordedAt === undefined + ? undefined + : await ensureHourlyStatsState(ctx); + await applyStarDelta(ctx, skill, -1, existing.createdAt, hourlyState); return { starred: false }; } if (skill.softDeletedAt) throw new Error("Skill not found"); + const hourlyState = await ensureHourlyStatsState(ctx); + const createdAt = Date.now(); await ctx.db.insert("stars", { skillId: args.skillId, userId, - createdAt: Date.now(), + createdAt, + hourlyStatsRecordedAt: createdAt, }); - await applyStarDelta(ctx, skill, 1); + await applyStarDelta(ctx, skill, 1, createdAt, hourlyState); return { starred: true }; }, @@ -90,13 +115,16 @@ export const addStarInternal = internalMutation({ .unique(); if (existing) return { ok: true as const, starred: true, alreadyStarred: true }; + const hourlyState = await ensureHourlyStatsState(ctx); + const createdAt = Date.now(); await ctx.db.insert("stars", { skillId: args.skillId, userId: args.userId, - createdAt: Date.now(), + createdAt, + hourlyStatsRecordedAt: createdAt, }); - await applyStarDelta(ctx, skill, 1); + await applyStarDelta(ctx, skill, 1, createdAt, hourlyState); return { ok: true as const, starred: true, alreadyStarred: false }; }, @@ -114,7 +142,9 @@ export const removeStarInternal = internalMutation({ if (!existing) return { ok: true as const, unstarred: false, alreadyUnstarred: true }; await ctx.db.delete(existing._id); - await applyStarDelta(ctx, skill, -1); + const hourlyState = + existing.hourlyStatsRecordedAt === undefined ? undefined : await ensureHourlyStatsState(ctx); + await applyStarDelta(ctx, skill, -1, existing.createdAt, hourlyState); return { ok: true as const, unstarred: true, alreadyUnstarred: false }; }, diff --git a/src/__tests__/home-listing-section.claw591.test.tsx b/src/__tests__/home-listing-section.claw591.test.tsx index 878e0cf4..c79225b6 100644 --- a/src/__tests__/home-listing-section.claw591.test.tsx +++ b/src/__tests__/home-listing-section.claw591.test.tsx @@ -110,6 +110,21 @@ describe("HomeListingSection", () => { expect(screen.queryByText("skills.sh")).toBeNull(); }); + it("shows canonical Trending as unavailable without substituting legacy skills", () => { + render(); + + expect(screen.getByText("24-hour Trending unavailable")).toBeTruthy(); + expect(screen.getByText(/canonical 24-hour feed isn't available/i)).toBeTruthy(); + expect(screen.queryByText("Quiet shelf")).toBeNull(); + }); + + it("labels an empty canonical 24-hour window honestly", () => { + render(); + + expect(screen.getByText("No 24-hour activity yet")).toBeTruthy(); + expect(screen.getByText(/eligible activity in the current 24-hour window/i)).toBeTruthy(); + }); + it("shows New, Featured, and Official for plugins but never plugin Trending", async () => { render(); @@ -175,6 +190,22 @@ describe("HomeListingSection", () => { ); }); + it("keeps loaded Trending rows when a later Load more page fails", async () => { + const first = makeTrending("first", "First Skill", 17, 9000); + fetchCanonicalTrendingPageMock + .mockResolvedValueOnce(canonicalPage([first], "opaque cursor 2")) + .mockRejectedValueOnce(new Error("second page unavailable")); + render(); + + fireEvent.click(screen.getByRole("button", { name: "Load more" })); + + await waitFor(() => { + expect(fetchCanonicalTrendingPageMock).toHaveBeenCalledTimes(2); + }); + expect(screen.getByTitle("First Skill")).toBeTruthy(); + expect(screen.queryByText("24-hour Trending unavailable")).toBeNull(); + }); + it("keeps search as a separate relevance-first interaction", async () => { convexActionMock.mockResolvedValue([ { @@ -212,7 +243,11 @@ describe("HomeListingSection", () => { }); }); -function initialTrending(items: ReturnType[], hasMore = false) { +function initialTrending( + items: ReturnType[], + hasMore = false, + trendingState: "available" | "empty" | "unavailable" = items.length ? "available" : "empty", +) { return { kind: "skills" as const, tab: "trending" as const, @@ -220,6 +255,7 @@ function initialTrending(items: ReturnType[], hasMore = fal fetchLimit: 20 as const, items: items.map((trending) => ({ trending })), hasMore, + trendingState, }; } diff --git a/src/__tests__/skills-index.claw591.test.tsx b/src/__tests__/skills-index.claw591.test.tsx index 62af4a9a..de1b04e3 100644 --- a/src/__tests__/skills-index.claw591.test.tsx +++ b/src/__tests__/skills-index.claw591.test.tsx @@ -134,24 +134,34 @@ describe("SkillsIndex", () => { expect(screen.queryByText(/Not scanned by ClawHub/i)).toBeNull(); }); - it("renders legacy Trending while the canonical rollout is disabled", async () => { + it("shows Trending unavailable without reading the legacy leaderboard", async () => { fetchCatalogDiscoveryCapabilitiesMock.mockResolvedValue({ apiVersion: 0, canonicalTrendingEnabled: false, }); - convexHttpMock.query.mockResolvedValue({ - items: [makeListResult("legacy-trending", "Legacy Trending")], - nextCursor: null, - }); render(); - expect(await screen.findByText("Legacy Trending")).toBeTruthy(); + expect(await screen.findByText("24-hour Trending unavailable")).toBeTruthy(); expect(fetchCanonicalTrendingPageMock).not.toHaveBeenCalled(); - expect(convexHttpMock.query).toHaveBeenCalledWith( - expect.anything(), - expect.objectContaining({ limit: 20 }), - ); + expect(convexHttpMock.query).not.toHaveBeenCalled(); + }); + + it("shows stale canonical Trending as unavailable without a legacy retry", async () => { + fetchCanonicalTrendingPageMock.mockRejectedValue(new Error("Trending snapshot expired")); + + render(); + + expect(await screen.findByText("24-hour Trending unavailable")).toBeTruthy(); + expect(convexHttpMock.query).not.toHaveBeenCalled(); + }); + + it("labels an empty canonical 24-hour window honestly", async () => { + render(); + + expect(await screen.findByText("No 24-hour activity yet")).toBeTruthy(); + expect(screen.getByText(/eligible activity in the current 24-hour window/i)).toBeTruthy(); + expect(screen.queryByRole("link", { name: /add a skill/i })).toBeNull(); }); it("loads New from the native 14-day chronological feed", async () => { diff --git a/src/components/HomeListingSection.tsx b/src/components/HomeListingSection.tsx index d62748c1..0303cc60 100644 --- a/src/components/HomeListingSection.tsx +++ b/src/components/HomeListingSection.tsx @@ -41,6 +41,7 @@ import { type HomeNativeSkillListingEntry, type HomeListingTab as ListingTab, type HomeSkillListingEntry as SkillPageEntry, + type TrendingFeedState, } from "../lib/homeListingData"; import { formatCompactStat } from "../lib/numberFormat"; import { fetchPluginCatalog, type PackageListItem } from "../lib/packageApi"; @@ -99,25 +100,38 @@ function HomeListingEmptyPanel({ query, onClearSearch, }: { - variant: "error" | "search" | "filter"; + variant: "error" | "search" | "filter" | "trendingEmpty" | "trendingUnavailable"; query?: string; onClearSearch?: () => void; }) { - const Icon = variant === "error" ? CloudOff : variant === "search" ? Binoculars : Moon; + const Icon = + variant === "error" || variant === "trendingUnavailable" + ? CloudOff + : variant === "search" + ? Binoculars + : Moon; const title = - variant === "error" - ? "Listings took a coffee break" - : variant === "search" - ? query - ? `No claws for “${query}”` - : "No claws in this view" - : "Quiet shelf"; + variant === "trendingUnavailable" + ? "24-hour Trending unavailable" + : variant === "trendingEmpty" + ? "No 24-hour activity yet" + : variant === "error" + ? "Listings took a coffee break" + : variant === "search" + ? query + ? `No claws for “${query}”` + : "No claws in this view" + : "Quiet shelf"; const body = - variant === "error" - ? "We couldn't load this slice of the catalog. Give it another try in a moment." - : variant === "search" - ? "Try another query or clear the search." - : "Nothing on this tab right now. Peek at another tab or widen the category."; + variant === "trendingUnavailable" + ? "The canonical 24-hour feed isn't available right now. Try another tab." + : variant === "trendingEmpty" + ? "No skills have eligible activity in the current 24-hour window." + : variant === "error" + ? "We couldn't load this slice of the catalog. Give it another try in a moment." + : variant === "search" + ? "Try another query or clear the search." + : "Nothing on this tab right now. Peek at another tab or widen the category."; return (
@@ -436,6 +450,7 @@ function createInitialListingCache(initialListing: HomeListingInitialData | null kind: "skills", items: initialListing.items, hasMore: initialListing.hasMore, + trendingState: initialListing.trendingState, } : { kind: "plugins", @@ -475,6 +490,9 @@ export function HomeListingSection({ initialListing = null }: HomeListingSection const [searchPlugins, setSearchPlugins] = useState([]); const [searchStatus, setSearchStatus] = useState<"idle" | "loading" | "error">("idle"); const [listingHasMore, setListingHasMore] = useState(initialListing?.hasMore ?? false); + const [trendingState, setTrendingState] = useState( + initialListing?.kind === "skills" ? initialListing.trendingState : undefined, + ); const trimmedSearch = searchQuery.trim(); const isSearchMode = trimmedSearch.length > 0; @@ -541,8 +559,13 @@ export function HomeListingSection({ initialListing = null }: HomeListingSection const cacheKey = listingCacheKey({ kind, tab, categorySlugs, fetchLimit }); const cached = listingCache.get(cacheKey); if (cached) { - if (cached.kind === "skills") setSkills(cached.items); - else setPlugins(cached.items); + if (cached.kind === "skills") { + setSkills(cached.items); + setTrendingState(cached.trendingState); + } else { + setPlugins(cached.items); + setTrendingState(undefined); + } setListingHasMore(cached.hasMore); setStatus("idle"); setLoadingMore(false); @@ -569,8 +592,10 @@ export function HomeListingSection({ initialListing = null }: HomeListingSection kind: "skills", items: result.page, hasMore: result.hasMore, + trendingState: result.trendingState, }); setSkills(result.page); + setTrendingState(result.trendingState); setListingHasMore(result.hasMore); setStatus("idle"); }) @@ -587,6 +612,7 @@ export function HomeListingSection({ initialListing = null }: HomeListingSection hasMore: result.hasMore, }); setPlugins(result.items); + setTrendingState(undefined); setListingHasMore(result.hasMore); setStatus("idle"); }); @@ -955,7 +981,15 @@ export function HomeListingSection({ initialListing = null }: HomeListingSection {isEmpty ? ( diff --git a/src/lib/catalogDiscoveryCapabilities.test.ts b/src/lib/catalogDiscoveryCapabilities.test.ts index e510d21d..68e158f7 100644 --- a/src/lib/catalogDiscoveryCapabilities.test.ts +++ b/src/lib/catalogDiscoveryCapabilities.test.ts @@ -21,10 +21,10 @@ describe("fetchCatalogDiscoveryCapabilities", () => { convexQueryMock.mockReset(); }); - it("enables canonical discovery only when the backend advertises it", async () => { + it("enables canonical Trending independently of the skills.sh rollout", async () => { convexQueryMock.mockResolvedValue({ - catalogDiscovery: { apiVersion: 1 }, - skillsSh: { runtimeEnabled: true }, + catalogDiscovery: { apiVersion: 1, canonicalTrendingEnabled: true }, + skillsSh: { runtimeEnabled: false }, }); await expect(fetchCatalogDiscoveryCapabilities()).resolves.toEqual({ diff --git a/src/lib/catalogDiscoveryCapabilities.ts b/src/lib/catalogDiscoveryCapabilities.ts index 6a59ca21..9a929a09 100644 --- a/src/lib/catalogDiscoveryCapabilities.ts +++ b/src/lib/catalogDiscoveryCapabilities.ts @@ -17,12 +17,13 @@ export async function fetchCatalogDiscoveryCapabilities(): Promise { fetchLimit: HOME_LISTING_PAGE_SIZE, items: [{ trending: item }], hasMore: true, + trendingState: "available", }); }); @@ -97,22 +98,53 @@ describe("homeListingData", () => { }); }); - it("falls back to legacy Trending when the canonical rollout is disabled", async () => { - const legacy = makeNative("legacy-trending", Date.now(), 10); + it("rejects a later canonical page failure instead of replacing loaded results", async () => { + const first = makeTrending("first", "First", 12); + fetchCanonicalTrendingPageMock + .mockResolvedValueOnce(canonicalPage([first], "opaque cursor 2")) + .mockRejectedValueOnce(new Error("second page unavailable")); + + await expect(fetchHomeSkillListing("trending", [], 2)).rejects.toThrow( + "second page unavailable", + ); + expect(convexQueryMock).not.toHaveBeenCalled(); + }); + + it("reports canonical Trending unavailable without reading the legacy leaderboard", async () => { fetchCatalogDiscoveryCapabilitiesMock.mockResolvedValue({ apiVersion: 0, canonicalTrendingEnabled: false, }); - convexQueryMock.mockResolvedValue({ items: [legacy], nextCursor: null }); await expect(fetchHomeSkillListing("trending", [], HOME_LISTING_PAGE_SIZE)).resolves.toEqual({ - page: [legacy], + page: [], hasMore: false, + trendingState: "unavailable", }); - expect(convexQueryMock).toHaveBeenCalledWith("skills:listPublicTrendingPage", { - limit: HOME_LISTING_PAGE_SIZE, - categorySlug: undefined, + expect(convexQueryMock).not.toHaveBeenCalled(); + expect(fetchCanonicalTrendingPageMock).not.toHaveBeenCalled(); + }); + + it("reports stale or unavailable canonical Trending without a legacy retry", async () => { + fetchCanonicalTrendingPageMock.mockRejectedValue(new Error("Trending snapshot expired")); + + await expect(fetchHomeSkillListing("trending", [], HOME_LISTING_PAGE_SIZE)).resolves.toEqual({ + page: [], + hasMore: false, + trendingState: "unavailable", }); + expect(convexQueryMock).not.toHaveBeenCalled(); + }); + + it("reports Trending unavailable when capability discovery fails", async () => { + fetchCatalogDiscoveryCapabilitiesMock.mockRejectedValue(new Error("capability outage")); + + await expect(fetchHomeSkillListing("trending", [], HOME_LISTING_PAGE_SIZE)).resolves.toEqual({ + page: [], + hasMore: false, + trendingState: "unavailable", + }); + expect(convexQueryMock).not.toHaveBeenCalled(); expect(fetchCanonicalTrendingPageMock).not.toHaveBeenCalled(); }); diff --git a/src/lib/homeListingData.ts b/src/lib/homeListingData.ts index a4e1ceef..9c34b772 100644 --- a/src/lib/homeListingData.ts +++ b/src/lib/homeListingData.ts @@ -4,10 +4,15 @@ import { fetchCatalogDiscoveryCapabilities } from "./catalogDiscoveryCapabilitie import { getSkillCategoriesForSkill } from "./categories"; import { fetchPluginCatalog, type PackageListItem } from "./packageApi"; import type { PublicSkill, PublicUser } from "./publicUser"; -import { fetchCanonicalTrendingPage, type CanonicalTrendingItem } from "./trendingApi"; +import { + fetchCanonicalTrendingPage, + type CanonicalTrendingItem, + type TrendingFeedState, +} from "./trendingApi"; export type HomeListingKind = "skills" | "plugins"; export type HomeListingTab = "trending" | "new" | "featured" | "official"; +export type { TrendingFeedState } from "./trendingApi"; export type HomeNativeSkillListingEntry = { skill: PublicSkill; @@ -28,7 +33,12 @@ export function isHomeTrendingSkillEntry( } export type HomeListingCacheEntry = - | { kind: "skills"; items: HomeSkillListingEntry[]; hasMore: boolean } + | { + kind: "skills"; + items: HomeSkillListingEntry[]; + hasMore: boolean; + trendingState?: TrendingFeedState; + } | { kind: "plugins"; items: PackageListItem[]; hasMore: boolean }; type HomeListingInitialDataBase = { @@ -42,6 +52,7 @@ export type HomeListingInitialData = | (HomeListingInitialDataBase & { kind: "skills"; items: HomeSkillListingEntry[]; + trendingState?: TrendingFeedState; }) | (HomeListingInitialDataBase & { kind: "plugins"; @@ -104,16 +115,16 @@ export async function fetchHomeSkillListing( signal?: AbortSignal, ) { if (tab === "trending") { - const capabilities = await fetchCatalogDiscoveryCapabilities(); + // Trending is one global canonical feed. The homepage clears and hides category + // controls before selecting it, so filtering here would silently rerank that feed. + let capabilities: Awaited>; + try { + capabilities = await fetchCatalogDiscoveryCapabilities(); + } catch { + return { page: [], hasMore: false, trendingState: "unavailable" as const }; + } if (!capabilities.canonicalTrendingEnabled) { - const result = await convexHttp.query(api.skills.listPublicTrendingPage, { - limit: numItems, - categorySlug: categorySlugs.length === 1 ? categorySlugs[0] : undefined, - }); - const page = ((result as { items?: HomeNativeSkillListingEntry[] }).items ?? []).filter( - (entry) => skillMatchesAnyHomeCategory(entry.skill, categorySlugs), - ); - return { page, hasMore: false }; + return { page: [], hasMore: false, trendingState: "unavailable" as const }; } const items: HomeTrendingSkillListingEntry[] = []; @@ -121,18 +132,28 @@ export async function fetchHomeSkillListing( let hasMore = false; const maxRequests = numItems < HOME_LISTING_PAGE_SIZE ? numItems : Math.ceil(numItems / HOME_LISTING_PAGE_SIZE); - for (let pageIndex = 0; pageIndex < maxRequests; pageIndex += 1) { - const result = await fetchCanonicalTrendingPage({ - cursor, - limit: Math.min(HOME_LISTING_PAGE_SIZE, numItems - items.length), - signal, - }); - items.push(...result.items.map((trending) => ({ trending }))); - cursor = result.nextCursor; - hasMore = cursor !== null; - if (!cursor || items.length >= numItems) break; + try { + for (let pageIndex = 0; pageIndex < maxRequests; pageIndex += 1) { + const result = await fetchCanonicalTrendingPage({ + cursor, + limit: Math.min(HOME_LISTING_PAGE_SIZE, numItems - items.length), + signal, + }); + items.push(...result.items.map((trending) => ({ trending }))); + cursor = result.nextCursor; + hasMore = cursor !== null; + if (!cursor || items.length >= numItems) break; + } + } catch (error) { + if (signal?.aborted) throw error; + if (items.length > 0) throw error; + return { page: [], hasMore: false, trendingState: "unavailable" as const }; } - return { page: items, hasMore }; + return { + page: items, + hasMore, + trendingState: items.length > 0 ? ("available" as const) : ("empty" as const), + }; } // highlightedOnly is a dedicated backend path ordered by skillBadges.by_kind_at; @@ -351,6 +372,7 @@ export async function fetchInitialHomeListing(): Promise fetchLimit: HOME_LISTING_PAGE_SIZE, items: result.page, hasMore: result.hasMore, + trendingState: result.trendingState ?? "unavailable", }; } diff --git a/src/lib/trendingApi.ts b/src/lib/trendingApi.ts index 8552d3c5..547a3793 100644 --- a/src/lib/trendingApi.ts +++ b/src/lib/trendingApi.ts @@ -1,5 +1,7 @@ import { publicApiUrl } from "./publicApiUrl"; +export type TrendingFeedState = "available" | "empty" | "unavailable"; + export type CanonicalTrendingItem = { id: string; source: "clawhub" | "skills-sh"; diff --git a/src/routes/skills/-SkillsResults.tsx b/src/routes/skills/-SkillsResults.tsx index 2b864430..d78b636c 100644 --- a/src/routes/skills/-SkillsResults.tsx +++ b/src/routes/skills/-SkillsResults.tsx @@ -11,6 +11,7 @@ import { Button } from "../../components/ui/button"; import { getSkillBadges } from "../../lib/badges"; import { formatCompactStat } from "../../lib/numberFormat"; import { timeAgo } from "../../lib/timeAgo"; +import type { TrendingFeedState } from "../../lib/trendingApi"; import { truncateText } from "../../lib/truncateText"; import { useMediaQuery } from "../../lib/useMediaQuery"; import { @@ -35,6 +36,7 @@ type SkillsResultsProps = { loadMoreRef: RefObject; loadMore: () => void; catalogTab: SkillsCatalogTab; + trendingState?: TrendingFeedState; }; function TrendingSkillListItem({ item }: { item: TrendingSkillListEntry }) { @@ -194,6 +196,7 @@ export function SkillsResults({ loadMoreRef, loadMore, catalogTab, + trendingState, }: SkillsResultsProps) { const isMobileBrowse = useMediaQuery("(max-width: 760px)"); const effectiveView = isMobileBrowse ? "list" : view; @@ -205,18 +208,30 @@ export function SkillsResults({ ) : sorted.length === 0 && listDoneLoading ? (
-

No skills found

+

+ {!hasQuery && catalogTab === "trending" + ? trendingState === "unavailable" + ? "24-hour Trending unavailable" + : "No 24-hour activity yet" + : "No skills found"} +

{hasQuery ? "Try a different search term or remove filters." - : "No skills have been published yet."} + : catalogTab === "trending" + ? trendingState === "unavailable" + ? "The canonical 24-hour feed isn't available right now. Try another tab." + : "No skills have eligible activity in the current 24-hour window." + : "No skills have been published yet."}

- + {!hasQuery && catalogTab === "trending" ? null : ( + + )}
) : effectiveView === "grid" ? (
diff --git a/src/routes/skills/-useSkillsBrowseModel.ts b/src/routes/skills/-useSkillsBrowseModel.ts index 10df084d..4ab69851 100644 --- a/src/routes/skills/-useSkillsBrowseModel.ts +++ b/src/routes/skills/-useSkillsBrowseModel.ts @@ -9,7 +9,7 @@ import { getSkillCategoryBySlug, getSkillCategoriesForSkill, } from "../../lib/categories"; -import { fetchCanonicalTrendingPage } from "../../lib/trendingApi"; +import { fetchCanonicalTrendingPage, type TrendingFeedState } from "../../lib/trendingApi"; import { parseDir, parseSort, toListSort, type SortDir, type SortKey } from "./-params"; import { isExternalSkillListEntry, @@ -165,6 +165,7 @@ export function useSkillsBrowseModel({ const [listResults, setListResults] = useState([]); const [listCursor, setListCursor] = useState(null); const [listStatus, setListStatus] = useState("loading"); + const [trendingState, setTrendingState] = useState(); const [, setListAutoLoadPaused] = useState(false); const fetchGeneration = useRef(0); const newCutoff = useMemo(() => Date.now() - newWindowMs, [catalogTab]); @@ -175,32 +176,30 @@ export function useSkillsBrowseModel({ let consecutiveEmptyPages = 0; try { if (catalogTab === "trending") { + // Trending selection clears category/topic URL state and hides those controls. + // Consume the one canonical order instead of constructing filtered variants. const capabilities = await fetchCatalogDiscoveryCapabilities(); - if (capabilities.canonicalTrendingEnabled) { - const result = await fetchCanonicalTrendingPage({ - cursor: pageCursor, - limit: pageSize, - }); + if (!capabilities.canonicalTrendingEnabled) { if (generation !== fetchGeneration.current) return; - const entries = result.items.map((trending) => ({ trending })); - setListResults((prev) => (cursor ? [...prev, ...entries] : entries)); - setListCursor(result.nextCursor); + setListResults([]); + setListCursor(null); setListAutoLoadPaused(false); - setListStatus(result.nextCursor ? "idle" : "done"); + setTrendingState("unavailable"); + setListStatus("done"); return; } - const result = await convexHttp.query(api.skills.listPublicTrendingPage, { + const result = await fetchCanonicalTrendingPage({ + cursor: pageCursor, limit: pageSize, - categorySlug: activeCategory?.slug, - topic: activeTopic, }); if (generation !== fetchGeneration.current) return; - const entries = (result as { items: SkillListEntry[] }).items; - setListResults(entries); - setListCursor(null); + const entries = result.items.map((trending) => ({ trending })); + setListResults((prev) => (cursor ? [...prev, ...entries] : entries)); + setListCursor(result.nextCursor); setListAutoLoadPaused(false); - setListStatus("done"); + setTrendingState(entries.length > 0 || result.nextCursor ? "available" : "empty"); + setListStatus(result.nextCursor ? "idle" : "done"); return; } const capabilities = @@ -264,6 +263,10 @@ export function useSkillsBrowseModel({ if (!isNavigationAbortError(err)) { console.error("Failed to fetch skills page:", err); } + if (catalogTab === "trending" && !pageCursor) { + setListResults([]); + setTrendingState("unavailable"); + } // Reset to idle so the user can retry via "Load more" setListCursor(pageCursor); setListAutoLoadPaused(Boolean(pageCursor)); @@ -294,6 +297,7 @@ export function useSkillsBrowseModel({ setListResults([]); setListCursor(null); setListAutoLoadPaused(false); + setTrendingState(undefined); setListStatus("loading"); void fetchPage(null, generation); return () => { @@ -634,6 +638,7 @@ export function useSkillsBrowseModel({ query, sort, sorted, + trendingState, view, }; } diff --git a/src/routes/skills/index.tsx b/src/routes/skills/index.tsx index 6d833400..68522fc3 100644 --- a/src/routes/skills/index.tsx +++ b/src/routes/skills/index.tsx @@ -312,6 +312,7 @@ export function SkillsIndex() { loadMoreRef={model.loadMoreRef} loadMore={model.loadMore} catalogTab={model.catalogTab} + trendingState={model.trendingState} />