fix(convex): reduce write conflicts across hot paths

- downloads:increment: remove unnecessary db.get that added skill doc
  to read set, causing conflicts with the stat processing cron
- users:ensure: only patch when there are real field changes, skip
  unconditional updatedAt bump that forced a write on every call
- comments: route stats through event sourcing (insertStatEvent) instead
  of synchronous read-modify-write on the skill doc
- rateLimits: split into query-first check + conditional mutation so
  denied requests are conflict-free reads
- skillStatEvents: reduce MAX_SKILLS_PER_RUN from 500 to 50 to shrink
  the write set and lower conflict probability with concurrent mutations

Co-Authored-By: theonejvo <theonejvo@users.noreply.github.com>

Author: theonejvo

convex/comments.ts

@@ -3,6 +3,7 @@ import type { Doc } from './_generated/dataModel'
 import { mutation, query } from './_generated/server'
 import { assertModerator, requireUser } from './lib/access'
 import { type PublicUser, toPublicUser } from './lib/public'
+import { insertStatEvent } from './skillStatEvents'
 
 export const listBySkill = query({
   args: { skillId: v.id('skills'), limit: v.optional(v.number()) },
@@ -43,10 +44,7 @@ export const add = mutation({
       deletedBy: undefined,
     })
 
-    await ctx.db.patch(skill._id, {
-      stats: { ...skill.stats, comments: skill.stats.comments + 1 },
-      updatedAt: Date.now(),
-    })
+    await insertStatEvent(ctx, { skillId: skill._id, kind: 'comment' })
   },
 })
 
@@ -68,13 +66,7 @@ export const remove = mutation({
       deletedBy: user._id,
     })
 
-    const skill = await ctx.db.get(comment.skillId)
-    if (skill) {
-      await ctx.db.patch(skill._id, {
-        stats: { ...skill.stats, comments: Math.max(0, skill.stats.comments - 1) },
-        updatedAt: Date.now(),
-      })
-    }
+    await insertStatEvent(ctx, { skillId: comment.skillId, kind: 'uncomment' })
 
     await ctx.db.insert('auditLogs', {
       actorUserId: user._id,

convex/downloads.ts

@@ -92,10 +92,11 @@ export const downloadZip = httpAction(async (ctx, request) => {
 export const increment = mutation({
   args: { skillId: v.id('skills') },
   handler: async (ctx, args) => {
-    const skill = await ctx.db.get(args.skillId)
-    if (!skill) return
+    // Skip db.get to avoid adding the skill doc to the read set.
+    // The calling HTTP action already validated the skill exists,
+    // and the stat processor handles deleted skills gracefully.
     await insertStatEvent(ctx, {
-      skillId: skill._id,
+      skillId: args.skillId,
       kind: 'download',
     })
   },

convex/httpApiV1.ts

@@ -827,11 +827,30 @@ async function checkRateLimit(
   key: string,
   limit: number,
 ): Promise<RateLimitResult> {
-  return (await ctx.runMutation(internal.rateLimits.checkRateLimitInternal, {
+  // Step 1: Read-only check — no write conflicts for denied requests
+  const status = (await ctx.runQuery(internal.rateLimits.getRateLimitStatusInternal, {
     key,
     limit,
     windowMs: RATE_LIMIT_WINDOW_MS,
   })) as RateLimitResult
+
+  if (!status.allowed) {
+    return status
+  }
+
+  // Step 2: Consume a token (only when allowed, with double-check for races)
+  const result = (await ctx.runMutation(internal.rateLimits.consumeRateLimitInternal, {
+    key,
+    limit,
+    windowMs: RATE_LIMIT_WINDOW_MS,
+  })) as { allowed: boolean; remaining: number }
+
+  return {
+    allowed: result.allowed,
+    remaining: result.remaining,
+    limit: status.limit,
+    resetAt: status.resetAt,
+  }
 }
 
 function pickMostRestrictive(primary: RateLimitResult, secondary: RateLimitResult | null) {

convex/lib/skillStats.ts

@@ -5,6 +5,7 @@ import { toDayKey } from './leaderboards'
 type SkillStatDeltas = {
   downloads?: number
   stars?: number
+  comments?: number
   installsCurrent?: number
   installsAllTime?: number
 }
@@ -22,8 +23,10 @@ export function applySkillStatDeltas(skill: Doc<'skills'>, deltas: SkillStatDelt
       ? skill.statsInstallsAllTime
       : (skill.stats.installsAllTime ?? 0)
 
+  const currentComments = skill.stats.comments
   const nextDownloads = Math.max(0, currentDownloads + (deltas.downloads ?? 0))
   const nextStars = Math.max(0, currentStars + (deltas.stars ?? 0))
+  const nextComments = Math.max(0, currentComments + (deltas.comments ?? 0))
   const nextInstallsCurrent = Math.max(0, currentInstallsCurrent + (deltas.installsCurrent ?? 0))
   const nextInstallsAllTime = Math.max(0, currentInstallsAllTime + (deltas.installsAllTime ?? 0))
 
@@ -36,6 +39,7 @@ export function applySkillStatDeltas(skill: Doc<'skills'>, deltas: SkillStatDelt
       ...skill.stats,
       downloads: nextDownloads,
       stars: nextStars,
+      comments: nextComments,
       installsCurrent: nextInstallsCurrent,
       installsAllTime: nextInstallsAllTime,
     },

convex/rateLimits.ts

@@ -1,7 +1,11 @@
 import { v } from 'convex/values'
-import { internalMutation } from './_generated/server'
+import { internalMutation, internalQuery } from './_generated/server'
 
-export const checkRateLimitInternal = internalMutation({
+/**
+ * Read-only rate limit check. Returns current status without writing anything.
+ * This eliminates write conflicts for denied requests entirely.
+ */
+export const getRateLimitStatusInternal = internalQuery({
   args: {
     key: v.string(),
     limit: v.number(),
@@ -20,6 +24,43 @@ export const checkRateLimitInternal = internalMutation({
       .withIndex('by_key_window', (q) => q.eq('key', args.key).eq('windowStart', windowStart))
       .unique()
 
+    const count = existing?.count ?? 0
+    const allowed = count < args.limit
+    return {
+      allowed,
+      remaining: Math.max(0, args.limit - count),
+      limit: args.limit,
+      resetAt,
+    }
+  },
+})
+
+/**
+ * Consume one rate limit token. Only call this after getRateLimitStatusInternal
+ * returns allowed=true. Includes a double-check to handle races between the
+ * query and this mutation.
+ */
+export const consumeRateLimitInternal = internalMutation({
+  args: {
+    key: v.string(),
+    limit: v.number(),
+    windowMs: v.number(),
+  },
+  handler: async (ctx, args) => {
+    const now = Date.now()
+    const windowStart = Math.floor(now / args.windowMs) * args.windowMs
+
+    const existing = await ctx.db
+      .query('rateLimits')
+      .withIndex('by_key_window', (q) => q.eq('key', args.key).eq('windowStart', windowStart))
+      .unique()
+
+    // Double-check: another request may have consumed the last token
+    // between our query and this mutation
+    if (existing && existing.count >= args.limit) {
+      return { allowed: false, remaining: 0 }
+    }
+
     if (!existing) {
       await ctx.db.insert('rateLimits', {
         key: args.key,
@@ -28,11 +69,7 @@ export const checkRateLimitInternal = internalMutation({
         limit: args.limit,
         updatedAt: now,
       })
-      return { allowed: true, remaining: Math.max(0, args.limit - 1), limit: args.limit, resetAt }
-    }
-
-    if (existing.count >= args.limit) {
-      return { allowed: false, remaining: 0, limit: args.limit, resetAt }
+      return { allowed: true, remaining: Math.max(0, args.limit - 1) }
     }
 
     await ctx.db.patch(existing._id, {
@@ -43,8 +80,6 @@ export const checkRateLimitInternal = internalMutation({
     return {
       allowed: true,
       remaining: Math.max(0, args.limit - existing.count - 1),
-      limit: args.limit,
-      resetAt,
     }
   },
 })

convex/schema.ts

@@ -294,6 +294,8 @@ const skillStatEvents = defineTable({
     v.literal('download'),
     v.literal('star'),
     v.literal('unstar'),
+    v.literal('comment'),
+    v.literal('uncomment'),
     v.literal('install_new'),
     v.literal('install_reactivate'),
     v.literal('install_deactivate'),

convex/skillStatEvents.ts

@@ -35,6 +35,8 @@ export type StatEventKind =
   | 'download'
   | 'star'
   | 'unstar'
+  | 'comment'
+  | 'uncomment'
   | 'install_new'
   | 'install_reactivate'
   | 'install_deactivate'
@@ -86,6 +88,7 @@ export async function insertStatEvent(
 type AggregatedDeltas = {
   downloads: number
   stars: number
+  comments: number
   installsAllTime: number
   installsCurrent: number
   /** Original timestamps for each download event (for daily stats bucketing) */
@@ -117,6 +120,7 @@ function aggregateEvents(events: Doc<'skillStatEvents'>[]): AggregatedDeltas {
   const result: AggregatedDeltas = {
     downloads: 0,
     stars: 0,
+    comments: 0,
     installsAllTime: 0,
     installsCurrent: 0,
     downloadEvents: [],
@@ -135,6 +139,12 @@ function aggregateEvents(events: Doc<'skillStatEvents'>[]): AggregatedDeltas {
       case 'unstar':
         result.stars -= 1
         break
+      case 'comment':
+        result.comments += 1
+        break
+      case 'uncomment':
+        result.comments -= 1
+        break
       case 'install_new':
         // New user installing for the first time: count toward both lifetime and current
         result.installsAllTime += 1
@@ -231,12 +241,14 @@ export const processSkillStatEventsInternal = internalMutation({
       if (
         deltas.downloads !== 0 ||
         deltas.stars !== 0 ||
+        deltas.comments !== 0 ||
         deltas.installsAllTime !== 0 ||
         deltas.installsCurrent !== 0
       ) {
         const patch = applySkillStatDeltas(skill, {
           downloads: deltas.downloads,
           stars: deltas.stars,
+          comments: deltas.comments,
           installsAllTime: deltas.installsAllTime,
           installsCurrent: deltas.installsCurrent,
         })
@@ -285,7 +297,7 @@ export const processSkillStatEventsInternal = internalMutation({
 
 const CURSOR_KEY = 'skill_stat_events'
 const EVENT_BATCH_SIZE = 500
-const MAX_SKILLS_PER_RUN = 500
+const MAX_SKILLS_PER_RUN = 50
 
 /**
  * Fetch a batch of events after the given cursor (by _creationTime).
@@ -332,6 +344,7 @@ const skillDeltaValidator = v.object({
   skillId: v.id('skills'),
   downloads: v.number(),
   stars: v.number(),
+  comments: v.number(),
   installsAllTime: v.number(),
   installsCurrent: v.number(),
   downloadEvents: v.array(v.number()),
@@ -438,6 +451,7 @@ export const processSkillStatEventsAction = internalAction({
       {
         downloads: number
         stars: number
+        comments: number
         installsAllTime: number
         installsCurrent: number
         downloadEvents: number[]
@@ -471,6 +485,7 @@ export const processSkillStatEventsAction = internalAction({
           skillDelta = {
             downloads: 0,
             stars: 0,
+            comments: 0,
             installsAllTime: 0,
             installsCurrent: 0,
             downloadEvents: [],
@@ -491,6 +506,12 @@ export const processSkillStatEventsAction = internalAction({
           case 'unstar':
             skillDelta.stars -= 1
             break
+          case 'comment':
+            skillDelta.comments += 1
+            break
+          case 'uncomment':
+            skillDelta.comments -= 1
+            break
           case 'install_new':
             skillDelta.installsAllTime += 1
             skillDelta.installsCurrent += 1

convex/users.ts

@@ -76,7 +76,6 @@ export const ensure = mutation({
   args: {},
   handler: async (ctx) => {
     const { userId, user } = await requireUser(ctx)
-    const now = Date.now()
     const updates: Record<string, unknown> = {}
 
     const handle = user.handle || user.name || user.email?.split('@')[0]
@@ -86,9 +85,9 @@ export const ensure = mutation({
       updates.role = handle === ADMIN_HANDLE ? 'admin' : DEFAULT_ROLE
     }
     if (!user.createdAt) updates.createdAt = user._creationTime
-    updates.updatedAt = now
 
     if (Object.keys(updates).length > 0) {
+      updates.updatedAt = Date.now()
       await ctx.db.patch(userId, updates)
     }