apps/web/app/(ee)/api/cron/projects/usage-updates-queue/route.ts (3)

81-88: Use array length check.

entries is an array; Object.keys(entries).length is unnecessary and confusing.

-      if (!entries || Object.keys(entries).length === 0) {
+      if (!entries || entries.length === 0) {

---

113-116: Reduce noisy batch logging.

Dumping full batches (including entryIds) can flood logs. Log counts only.

-        console.log(`BATCH: ${JSON.stringify(batch)}`);
+        console.log(`Processing sub-batch size=${batch.length}`);

---

121-124: Escape reserved column name usage.

If MySQL treats usage as reserved, quote it to avoid edge cases.

-              await conn.execute(
-                "UPDATE Project p SET p.usage = p.usage + ?, p.totalClicks = p.totalClicks + ? WHERE id = ?",
-                [update.usage, update.clicks, update.projectId],
-              );
+              await conn.execute(
+                "UPDATE Project p SET p.`usage` = p.`usage` + ?, p.totalClicks = p.totalClicks + ? WHERE id = ?",
+                [update.usage, update.clicks, update.projectId],
+              );

apps/web/lib/upstash/redis-streams.ts (1)

86-144: Use XLEN for O(1) length; avoid scanning entire stream.

Current implementation reads the whole stream to compute length. Replace with XLEN and single-entry range calls.

   async getStreamInfo(): Promise<{
     length: number;
     firstEntryId: string | null;
     lastEntryId: string | null;
   }> {
     try {
-      // Get stream length - check if stream exists first
-      let length = 0;
-      let firstEntryId: string | null = null;
-      let lastEntryId: string | null = null;
-
-      try {
-        // Try to get first entry to check if stream exists
-        const firstEntry = await redis.xrange(this.streamKey, "-", "+", 1);
-
-        if (firstEntry && Object.keys(firstEntry).length > 0) {
-          firstEntryId = Object.keys(firstEntry)[0];
-
-          // If stream exists, get its length and last entry
-          const entries = await redis.xrange(this.streamKey, "-", "+");
-          length = Object.keys(entries || {}).length;
-
-          if (length > 0) {
-            // Use "$" to get the last entry instead of "+"
-            const lastEntry = await redis.xrevrange(
-              this.streamKey,
-              "+",
-              "-",
-              1,
-            );
-
-            if (lastEntry && Object.keys(lastEntry).length > 0) {
-              const entryId = Object.keys(lastEntry)[0];
-
-              if (entryId !== firstEntryId) {
-                lastEntryId = entryId;
-              }
-            }
-          }
-        }
-      } catch (streamError) {
-        // Stream might not exist yet, which is fine
-        console.log("Stream does not exist or is empty:", streamError);
-        console.log(streamError.message);
-        console.log(streamError.type);
-        console.log(JSON.stringify(streamError, null, 2));
-      }
-
-      return {
-        length,
-        firstEntryId,
-        lastEntryId,
-      };
+      const length = await redis.xlen(this.streamKey).catch(() => 0);
+      let firstEntryId: string | null = null;
+      let lastEntryId: string | null = null;
+
+      if (length > 0) {
+        const first = await redis.xrange(this.streamKey, "-", "+", 1);
+        if (first && Object.keys(first).length > 0) {
+          firstEntryId = Object.keys(first)[0];
+        }
+        const last = await redis.xrevrange(this.streamKey, "+", "-", 1);
+        if (last && Object.keys(last).length > 0) {
+          lastEntryId = Object.keys(last)[0];
+        }
+      }
+
+      return { length, firstEntryId, lastEntryId };
     } catch (error) {
       console.error("Failed to get stream info:", error);
       throw error;
     }
   }

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

apps/web/app/(ee)/api/cron/usage/update-workspace-clicks/route.ts (1)

35-40: Entries without workspaceId are dropped and never deleted.

You continue on missing workspaceId, so these IDs aren't included in processedEntryIds and remain forever. This causes reprocessing loops.

Apply this diff to track invalid entries for deletion:

 const aggregateWorkspaceUsage = (
   entries: RedisStreamEntry<ClickEvent>[],
-): { updates: WorkspaceAggregateUsage[]; lastProcessedId: string | null } => {
+): {
+  updates: WorkspaceAggregateUsage[];
+  lastProcessedId: string | null;
+  invalidEntryIds: string[];
+} => {
   // Aggregate usage by workspaceId
   const aggregatedUsage = new Map<string, WorkspaceAggregateUsage>();
 
   let lastId: string | null = null;
+  const invalidEntryIds: string[] = [];
 
   for (const entry of entries) {
     const workspaceId = entry.data.workspaceId;
 
     if (!workspaceId) {
+      invalidEntryIds.push(entry.id);
       continue;
     }

And include invalid IDs in the return:

   return {
     updates: Array.from(aggregatedUsage.values()),
     lastProcessedId: lastId,
+    invalidEntryIds,
   };

Then update the caller at Line 82:

-      const { updates, lastProcessedId } = aggregateWorkspaceUsage(entries);
+      const { updates, lastProcessedId, invalidEntryIds } = aggregateWorkspaceUsage(entries);

And include invalid IDs in processedEntryIds at Line 103:

-      const processedEntryIds: string[] = [];
+      const processedEntryIds: string[] = [...invalidEntryIds];

apps/web/lib/upstash/redis-streams.ts (3)

79-82: Error message hardcoded for workspace usage in generic method.

The error message "Failed to read workspace usage updates from stream:" is specific to workspace usage, but processBatch is a generic method used for any stream type. Use a generic message or make the error context configurable.

Apply this diff:

       console.error(
-        "Failed to read workspace usage updates from stream:",
+        `Failed to read from stream ${this.streamKey}:`,
         error,
       );

---

101-136: Reduce excessive debug logging.

Lines 132-135 log the same error information multiple times (streamError, streamError.message, streamError.type, and JSON.stringify(streamError)). This clutters logs for a non-critical path (stream not existing is normal).

Apply this diff to simplify:

       } catch (streamError) {
         // Stream might not exist yet, which is fine
-        console.log("Stream does not exist or is empty:", streamError);
-        console.log(streamError.message);
-        console.log(streamError.type);
-        console.log(JSON.stringify(streamError, null, 2));
+        console.log("Stream does not exist or is empty");
       }

---

109-110: Fetching all entries to compute length may be inefficient.

Line 109 calls xrange(this.streamKey, "-", "+") without a count limit, loading all entries into memory just to count them. For large streams, this is inefficient. Consider using XINFO STREAM if available or fetching with a count parameter and computing approximate length.

If the Redis client supports XINFO STREAM, use that command to get stream metadata directly. Otherwise, document that this method is best-effort and may be slow for large streams.

apps/web/app/(ee)/api/cron/usage/update-workspace-clicks/route.ts (2)

106-106: Verbose logging of entire batch JSON.

Line 106 logs JSON.stringify(batch) for each sub-batch of up to 50 updates. This will produce very large log entries and clutter logs. Consider logging only a summary (e.g., count and first few workspace IDs).

Apply this diff:

-        console.log(`BATCH: ${JSON.stringify(batch)}`);
+        console.log(`Processing sub-batch of ${batch.length} updates`);

---

202-213: Consider getStreamInfo performance impact.

Line 202 calls workspaceUsageStream.getStreamInfo() after processing, which fetches all entries to compute length (as noted in the redis-streams.ts review). For large streams, this could add significant latency to the cron response. Consider whether stream info is essential for the response or if it can be computed asynchronously or cached.

If stream info is primarily for monitoring/logging, consider:

1. Making it optional and skipping on timeout risk
2. Computing it asynchronously and logging separately
3. Using approximate metrics instead of exact counts

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro