DiscordClone/convex/typing.ts

import { query, mutation, internalMutation } from "./_generated/server";
import { v } from "convex/values";
import { internal } from "./_generated/api";

const TYPING_TTL_MS = 6000;

export const startTyping = mutation({
  args: {
    channelId: v.id("channels"),
    userId: v.id("userProfiles"),
    username: v.string(),
  },
  returns: v.null(),
  handler: async (ctx, args) => {
    const expiresAt = Date.now() + TYPING_TTL_MS;

    const existing = await ctx.db
      .query("typingIndicators")
      .withIndex("by_channel", (q) => q.eq("channelId", args.channelId))
      .collect();

    const userTyping = existing.find((t) => t.userId === args.userId);

    if (userTyping) {
      // Refreshing an existing row — the cleanup scheduled from the
      // original insert is still pending, so don't pile another copy
      // onto the scheduler. The audit flagged the old code (schedule
      // on every heartbeat) as spamming cleanExpired tasks that all
      // did the same work.
      await ctx.db.patch(userTyping._id, { expiresAt });
    } else {
      await ctx.db.insert("typingIndicators", {
        channelId: args.channelId,
        userId: args.userId,
        username: args.username,
        expiresAt,
      });
      await ctx.scheduler.runAfter(TYPING_TTL_MS, internal.typing.cleanExpired, {});
    }

    return null;
  },
});

export const stopTyping = mutation({
  args: {
    channelId: v.id("channels"),
    userId: v.id("userProfiles"),
  },
  returns: v.null(),
  handler: async (ctx, args) => {
    const indicators = await ctx.db
      .query("typingIndicators")
      .withIndex("by_channel", (q) => q.eq("channelId", args.channelId))
      .collect();

    const mine = indicators.find((t) => t.userId === args.userId);
    if (mine) {
      await ctx.db.delete(mine._id);
    }

    return null;
  },
});

export const getTyping = query({
  args: { channelId: v.id("channels") },
  returns: v.array(
    v.object({
      userId: v.id("userProfiles"),
      username: v.string(),
      displayName: v.union(v.string(), v.null()),
    })
  ),
  handler: async (ctx, args) => {
    const now = Date.now();
    const indicators = await ctx.db
      .query("typingIndicators")
      .withIndex("by_channel", (q) => q.eq("channelId", args.channelId))
      .collect();

    const active = indicators.filter((t) => t.expiresAt > now);
    const results = [];
    for (const t of active) {
      const user = await ctx.db.get(t.userId);
      results.push({
        userId: t.userId,
        username: t.username,
        displayName: user?.displayName || null,
      });
    }
    return results;
  },
});

export const cleanExpired = internalMutation({
  args: {},
  returns: v.null(),
  handler: async (ctx) => {
    const now = Date.now();
    // Range-scan the by_expires_at index instead of collecting the whole
    // table, so cleanup cost scales with number of *expired* rows rather
    // than total typing activity across every channel.
    const expired = await ctx.db
      .query("typingIndicators")
      .withIndex("by_expires_at", (q) => q.lte("expiresAt", now))
      .collect();
    for (const t of expired) {
      await ctx.db.delete(t._id);
    }
    return null;
  },
});