From 952394c3edde7ddb1d02867a66d5c14f423bc22e Mon Sep 17 00:00:00 2001 From: kaihere14 Date: Sat, 22 Aug 2026 03:14:33 +0530 Subject: [PATCH] fix: fix README cleanup bypasses the queue runs inline in the HTTP request Implement reactive README cleanup progress UI Replace the static cleanup progress toast with a reactive system driven by Convex log messages. Added `cleanup-queue` worker support and logic to handle job retries and recovery via `sharedLogId`. --- client/src/components/repos/RepoCard.jsx | 142 ++++++++++++++--- client/src/lib/cleanupProgressToast.js | 54 ------- server/src/controllers/github.controller.js | 125 ++++----------- server/src/services/logRecovery.service.js | 31 +++- server/src/services/readmeCleanup.service.js | 2 +- server/src/utils/git.worker.js | 153 ++++++++++++++++++- 6 files changed, 332 insertions(+), 175 deletions(-) delete mode 100644 client/src/lib/cleanupProgressToast.js diff --git a/client/src/components/repos/RepoCard.jsx b/client/src/components/repos/RepoCard.jsx index 9a2361a..c0267ff 100644 --- a/client/src/components/repos/RepoCard.jsx +++ b/client/src/components/repos/RepoCard.jsx @@ -1,4 +1,4 @@ -import React, { useState } from "react"; +import React, { useEffect, useMemo, useRef, useState } from "react"; import { motion, useReducedMotion } from "framer-motion"; import { GitBranch, @@ -10,13 +10,42 @@ import { Loader, } from "lucide-react"; import { toast } from "sonner"; +import { useQuery } from "convex/react"; import { api, ENDPOINTS } from "@/lib/api"; -import { - startCleanupProgressToast, - completeCleanupProgressToast, -} from "@/lib/cleanupProgressToast"; +import { convexApi } from "@/lib/convexApi"; import { usePostHog } from "@posthog/react"; +// The worker's terminal messages, matched so the toast can settle instead of +// guessing on a timer. Kept in sync with cleanupHandler in git.worker.js. +const CLEANUP_SUCCESS_PREFIX = "✓ README committed"; +const CLEANUP_FAILURE_PREFIX = "✗ README cleanup failed"; +const CLEANUP_TOAST_DURATION_MS = 5000; + +const cleanupToastId = (logId) => `cleanup-progress-${logId}`; + +// liveUpdate fires Convex mutations without awaiting them, so the terminal +// message is not guaranteed to be the newest — scan instead of trusting the +// tail. Returns null while there is nothing to show yet. +const readCleanupOutcome = (messages) => { + if (!messages?.length) return null; + + for (let i = messages.length - 1; i >= 0; i -= 1) { + const { message } = messages[i]; + if (message.startsWith(CLEANUP_SUCCESS_PREFIX)) { + return { settled: true, succeeded: true, message }; + } + if (message.startsWith(CLEANUP_FAILURE_PREFIX)) { + return { settled: true, succeeded: false, message }; + } + } + + return { + settled: false, + succeeded: false, + message: messages[messages.length - 1].message, + }; +}; + const RepoCard = ({ repo, showToggle = true, @@ -30,7 +59,23 @@ const RepoCard = ({ const posthog = usePostHog(); const [isActive, setIsActive] = useState(repo.activated); const [loading, setLoading] = useState(false); - const [isCleaningUp, setIsCleaningUp] = useState(false); + const [isEnqueueingCleanup, setIsEnqueueingCleanup] = useState(false); + const [cleanupLogId, setCleanupLogId] = useState(null); + const settledCleanupRef = useRef(null); + const cleanupMessages = useQuery( + convexApi.logs.getLogMessages, + cleanupLogId ? { logId: cleanupLogId } : "skip", + ); + + // Progress is derived from the worker's log stream rather than mirrored into + // state, so the button stays spinning until the job actually reports back. + const cleanupOutcome = useMemo( + () => readCleanupOutcome(cleanupMessages), + [cleanupMessages], + ); + const isCleaningUp = + isEnqueueingCleanup || (Boolean(cleanupLogId) && !cleanupOutcome?.settled); + const ownerLabel = repo.owner || repo.full_name?.split("/")?.[0] || "Repository"; const branchLabel = repo.default_branch || "main"; @@ -77,31 +122,90 @@ const RepoCard = ({ } }; + // Cleanup runs on a queue, so the request only enqueues it. The toast is + // driven by the worker's own log messages, keyed by the logId in the 202. + useEffect(() => { + if (!cleanupLogId || !cleanupOutcome) return; + + const toastId = cleanupToastId(cleanupLogId); + + if (!cleanupOutcome.settled) { + toast.loading(cleanupOutcome.message, { id: toastId }); + return; + } + + // Terminal messages never change again, but a remount would replay them. + if (settledCleanupRef.current === cleanupLogId) return; + settledCleanupRef.current = cleanupLogId; + + // A loading toast has no duration, and sonner keeps whatever the toast was + // created with when an update reuses the id — pass one so it can close. + if (cleanupOutcome.succeeded) { + toast.success("Your README is now clean and tidy", { + id: toastId, + duration: CLEANUP_TOAST_DURATION_MS, + }); + posthog?.capture("readme_cleanup_completed", { + repo_name: repo.name, + repo_full_name: repo.full_name, + }); + return; + } + + const reason = cleanupOutcome.message + .slice(CLEANUP_FAILURE_PREFIX.length) + .replace(/^:\s*/, ""); + toast.error(reason || "Failed to clean up your README", { + id: toastId, + duration: CLEANUP_TOAST_DURATION_MS, + }); + posthog?.capture("readme_cleanup_failed", { + repo_name: repo.name, + repo_full_name: repo.full_name, + }); + }, [cleanupLogId, cleanupOutcome, posthog, repo.name, repo.full_name]); + + // A loading toast never auto-dismisses, so one left without an updater hangs + // on screen forever. Drop it if this card stops watching the job — unmounted, + // or superseded by a newer cleanup — unless it already settled. + useEffect(() => { + if (!cleanupLogId) return undefined; + + return () => { + if (settledCleanupRef.current !== cleanupLogId) { + toast.dismiss(cleanupToastId(cleanupLogId)); + } + }; + }, [cleanupLogId]); + const handleCleanUp = async (e) => { e.stopPropagation(); if (isCleaningUp) return; - setIsCleaningUp(true); - const progress = startCleanupProgressToast(); + setIsEnqueueingCleanup(true); try { - await api.post(ENDPOINTS.CLEAN_UP_README, { repoId: repo.id }); - completeCleanupProgressToast(progress, { - success: true, - message: "Your README is now clean and tidy", + const res = await api.post(ENDPOINTS.CLEAN_UP_README, { + repoId: repo.id, }); - posthog?.capture("readme_cleanup_completed", { + if (res.status !== 202 || !res.data?.logId) { + throw new Error("Cleanup could not be queued"); + } + + posthog?.capture("readme_cleanup_started", { repo_name: repo.name, repo_full_name: repo.full_name, }); - } catch (error) { - completeCleanupProgressToast(progress, { - success: false, - message: - error.response?.data?.message || "Failed to clean up your README", + toast.loading("Queued README cleanup", { + id: cleanupToastId(res.data.logId), }); + setCleanupLogId(res.data.logId); + } catch (error) { + toast.error( + error.response?.data?.message || "Failed to clean up your README", + ); } finally { - setIsCleaningUp(false); + setIsEnqueueingCleanup(false); } }; diff --git a/client/src/lib/cleanupProgressToast.js b/client/src/lib/cleanupProgressToast.js deleted file mode 100644 index fe25afd..0000000 --- a/client/src/lib/cleanupProgressToast.js +++ /dev/null @@ -1,54 +0,0 @@ -import { toast } from "sonner"; - -const MESSAGE_INTERVAL_MS = 5000; - -export const CLEANUP_PROGRESS_MESSAGES = [ - "Zapping clutter, reindexing context…", - "Convincing duplicate sections to merge…", - "Deleting vibes-only bullet points…", - "Asking your README to calm down…", - "Untangling feature lists from feature novels…", - "Negotiating with stale badges…", - "Removing changelog energy from 2019…", - "Teaching markdown to breathe again…", - 'Consolidating five ways we said "fast"…', - "Sweeping marketing fluff under the rug…", - 'Renaming "Overview" to something useful…', - "Your AI librarian is on duty…", - "Polishing headings until they behave…", - "Almost done — README therapy in session…", -]; - -const FINAL_MESSAGE_INDEX = CLEANUP_PROGRESS_MESSAGES.length - 1; - -export function startCleanupProgressToast() { - let index = 0; - const toastId = toast.loading(CLEANUP_PROGRESS_MESSAGES[0]); - - const intervalId = setInterval(() => { - if (index >= FINAL_MESSAGE_INDEX) return; - - index += 1; - toast.loading(CLEANUP_PROGRESS_MESSAGES[index], { id: toastId }); - - if (index >= FINAL_MESSAGE_INDEX) { - clearInterval(intervalId); - } - }, MESSAGE_INTERVAL_MS); - - return { - toastId, - stop() { - clearInterval(intervalId); - }, - }; -} - -export function completeCleanupProgressToast(progress, { success, message }) { - progress.stop(); - if (success) { - toast.success(message, { id: progress.toastId }); - } else { - toast.error(message, { id: progress.toastId }); - } -} diff --git a/server/src/controllers/github.controller.js b/server/src/controllers/github.controller.js index c455728..e051d27 100644 --- a/server/src/controllers/github.controller.js +++ b/server/src/controllers/github.controller.js @@ -2,7 +2,7 @@ import User from "../schema/user.schema.js"; import { decrypt } from "./oauthcontroller.js"; import ActiveRepo from "../schema/activeRepo.js"; import crypto from "node:crypto"; -import { readmeQueue } from "../utils/git.worker.js"; +import { cleanUpQueue, readmeQueue } from "../utils/git.worker.js"; import UserLogModel from "../schema/userLog.schema.js"; import { GITHUB_API_BASE, @@ -12,8 +12,6 @@ import { } from "../utils/githubApiClient.js"; import { RedisConnection } from "bullmq"; import { redis } from "../utils/redis.js"; -import { commitFile, getFileContent } from "../services/github.service.js"; -import { cleanReadmeWithAI } from "../services/readmeCleanup.service.js"; import { liveUpdate } from "../services/convex.service.js"; export function verifyGithubSignature(req) { @@ -579,8 +577,9 @@ export const fetchAdminUsers = async (req, res) => { }; export const cleanUpReadme = async (req, res) => { - let userLog = null; - let sharedLogId = null; + // Minted here, not in the worker, so the 202 can hand the client a log id to + // subscribe to and so a retry reuses the same log row. + const sharedLogId = crypto.randomUUID(); try { console.log("[cleanUpReadme] Started"); @@ -609,109 +608,37 @@ export const cleanUpReadme = async (req, res) => { return res.status(404).json({ message: "GitHub access token not found" }); } - const accessToken = decrypt(user.githubAccessToken); - - console.log("[cleanUpReadme] Fetching README.md"); - const readmeFile = await getFileContent( - accessToken, - activeRepo.repoOwner, - activeRepo.repoName, - "README.md", - activeRepo.defaultBranch, - ); - if (!readmeFile?.content?.trim()) { - console.log("[cleanUpReadme] README.md not found"); - return res - .status(404) - .json({ message: "README.md not found in repository" }); - } - - console.log("[cleanUpReadme] README fetched"); - sharedLogId = crypto.randomUUID(); - userLog = await UserLogModel.create({ - logId: sharedLogId, - userId, - repoName: activeRepo.repoName, - repoOwner: activeRepo.repoOwner, - action: "README_CLEANUP_STARTED", - status: "ongoing", - }); - await redis.del("admin_analytics"); - - liveUpdate( - sharedLogId, - `Starting README cleanup for ${activeRepo.repoOwner}/${activeRepo.repoName}`, - ); - console.log("[cleanUpReadme] Running AI cleanup"); - liveUpdate(sharedLogId, "Fetched existing README.md"); - liveUpdate(sharedLogId, "Cleaning README content with AI"); - const cleanedReadme = await cleanReadmeWithAI(readmeFile.content, (msg) => - liveUpdate(sharedLogId, msg), - ); - console.log("[cleanUpReadme] AI cleanup complete"); - liveUpdate(sharedLogId, `Cleanup complete (${cleanedReadme.length} chars)`); - - console.log("[cleanUpReadme] Committing README"); - liveUpdate(sharedLogId, "Committing cleaned README to GitHub"); - const commitResult = await commitFile( - accessToken, - activeRepo.repoOwner, - activeRepo.repoName, - "README.md", - cleanedReadme, - "chore: cleanup README [skip ci]", - activeRepo.defaultBranch, - readmeFile.sha, - ); - - console.log("[cleanUpReadme] README committed:", commitResult.commit.sha); - liveUpdate( - sharedLogId, - `✓ README committed: ${commitResult.commit.sha.slice(0, 7)}`, - ); - await UserLogModel.findByIdAndUpdate( - userLog._id, + await cleanUpQueue.add( + "cleanup-queue", { - action: "README_CLEANUP_SUCCESS", - status: "success", - commitId: commitResult.commit.sha, + userId, + repoName: activeRepo.repoName, + repoOwner: activeRepo.repoOwner, + defaultBranch: activeRepo.defaultBranch, + // Ciphertext only — the job payload sits in Redis for the lifetime of + // the job, so the worker does the decrypting. + encryptedAccessToken: { + iv: user.githubAccessToken.iv, + content: user.githubAccessToken.content, + tag: user.githubAccessToken.tag, + }, + sharedLogId, }, { - new: true, - runValidators: true, + attempts: 3, + backoff: { type: "exponential", delay: 5000 }, }, ); - await redis.del("admin_analytics"); - return res.status(200).json({ - message: "Readme cleaned up successfully", - commitSha: commitResult.commit.sha, + return res.status(202).json({ + message: "Readme cleanup initiated", + logId: sharedLogId, }); } catch (error) { console.error("[cleanUpReadme] Failed:", error.message); liveUpdate(sharedLogId, `✗ Failed: ${error.message}`); - - if (userLog) { - try { - await UserLogModel.findByIdAndUpdate( - userLog._id, - { - action: "README_CLEANUP_FAILED", - status: "failed", - }, - { - new: true, - runValidators: true, - }, - ); - await redis.del("admin_analytics"); - } catch (logError) { - console.error( - "[cleanUpReadme] Failed to update Mongo log:", - logError.message, - ); - } - } - return res.status(500).json({ message: "Error cleaning up readme" }); + return res + .status(500) + .json({ message: "Error cleaning up readme", logId: sharedLogId }); } }; diff --git a/server/src/services/logRecovery.service.js b/server/src/services/logRecovery.service.js index 85b4300..fe3a1e0 100644 --- a/server/src/services/logRecovery.service.js +++ b/server/src/services/logRecovery.service.js @@ -1,12 +1,41 @@ import { liveUpdate } from "./convex.service.js"; import UserLogModel from "../schema/userLog.schema.js"; +import { cleanUpQueue } from "../utils/git.worker.js"; + +// The worker starts consuming as soon as git.worker.js is imported, which +// happens before this runs. A job that stalled on the previous shutdown is +// re-queued and retried, so its log is legitimately `ongoing` again — only +// logs with no job left behind them were actually interrupted. +async function getLogIdsStillQueued() { + const jobs = await cleanUpQueue.getJobs([ + "waiting", + "waiting-children", + "prioritized", + "delayed", + "paused", + "active", + ]); + + return new Set( + jobs.map((job) => job?.data?.sharedLogId).filter((logId) => Boolean(logId)), + ); +} export async function recoverInterruptedCleanupLogs() { - const interruptedLogs = await UserLogModel.find({ + const ongoingLogs = await UserLogModel.find({ action: "README_CLEANUP_STARTED", status: "ongoing", }).select("_id logId"); + if (ongoingLogs.length === 0) { + return 0; + } + + const stillQueued = await getLogIdsStillQueued(); + const interruptedLogs = ongoingLogs.filter( + (log) => !stillQueued.has(log.logId), + ); + if (interruptedLogs.length === 0) { return 0; } diff --git a/server/src/services/readmeCleanup.service.js b/server/src/services/readmeCleanup.service.js index a55267e..4aa5b70 100644 --- a/server/src/services/readmeCleanup.service.js +++ b/server/src/services/readmeCleanup.service.js @@ -40,7 +40,7 @@ export async function cleanReadmeWithAI(existingReadme, onProgress = null) { } if (onProgress) { - onProgress(`Sending README to cleanup model ${CLEANUP_MODEL}`); + onProgress(`Sending README to cleanup model`); } const response = await fetch(OPENROUTER_URL, { diff --git a/server/src/utils/git.worker.js b/server/src/utils/git.worker.js index f74ecdf..b6106b3 100644 --- a/server/src/utils/git.worker.js +++ b/server/src/utils/git.worker.js @@ -1,6 +1,6 @@ import IORedis from "ioredis"; import { Queue } from "bullmq"; -import { Worker } from "bullmq"; +import { UnrecoverableError, Worker } from "bullmq"; import { redis } from "./redis.js"; import User from "../schema/user.schema.js"; import ActiveRepo from "../schema/activeRepo.js"; @@ -29,6 +29,7 @@ import { } from "./prompt.builder.js"; import UserLogModel from "../schema/userLog.schema.js"; import { liveUpdate } from "../services/convex.service.js"; +import { cleanReadmeWithAI } from "../services/readmeCleanup.service.js"; export const connection = new IORedis({ host: process.env.REDIS_HOST || "localhost", @@ -659,3 +660,153 @@ function getImportantFiles(tree) { return categorized.map((item) => item.path); } + +export const cleanUpQueue = new Queue("cleanup-queue", { connection }); + +new Worker("cleanup-queue", cleanupHandler, { + connection, + removeOnComplete: { count: 100 }, + removeOnFail: { count: 50 }, +}); + +// A retry reuses the sharedLogId minted by the controller, so upsert the row +// instead of creating one — a stalled job re-run would otherwise leave a +// second Mongo row for the same cleanup, and logRecovery would mark the +// orphan failed while the retry is still running. +async function startCleanupLog({ sharedLogId, userId, repoName, repoOwner }) { + const userLog = await UserLogModel.findOneAndUpdate( + { logId: sharedLogId }, + { + logId: sharedLogId, + userId, + repoName, + repoOwner, + action: "README_CLEANUP_STARTED", + status: "ongoing", + }, + { + new: true, + upsert: true, + setDefaultsOnInsert: true, + runValidators: true, + }, + ); + await redis.del("admin_analytics"); + return userLog; +} + +async function cleanupHandler(job) { + const { + userId, + repoName, + repoOwner, + defaultBranch, + encryptedAccessToken, + sharedLogId, + } = job.data; + + const userLog = await startCleanupLog({ + sharedLogId, + userId, + repoName, + repoOwner, + }); + + try { + const accessToken = decrypt(encryptedAccessToken); + + liveUpdate( + sharedLogId, + `Starting README cleanup for ${repoOwner}/${repoName}`, + ); + console.log("[cleanUpReadme] Fetching README.md"); + const readmeFile = await getFileContent( + accessToken, + repoOwner, + repoName, + "README.md", + defaultBranch, + ); + + if (!readmeFile?.content?.trim()) { + console.log("[cleanUpReadme] README.md not found"); + // Retrying cannot conjure a README — fail the job outright rather than + // burning every attempt plus its backoff on a job that cannot succeed. + throw new UnrecoverableError("README.md not found in repository"); + } + + console.log("[cleanUpReadme] README fetched"); + liveUpdate(sharedLogId, "Fetched existing README.md"); + liveUpdate(sharedLogId, "Cleaning README content with AI"); + console.log("[cleanUpReadme] Running AI cleanup"); + const cleanedReadme = await cleanReadmeWithAI(readmeFile.content, (msg) => + liveUpdate(sharedLogId, msg), + ); + console.log("[cleanUpReadme] AI cleanup complete"); + liveUpdate(sharedLogId, `Cleanup complete (${cleanedReadme.length} chars)`); + + console.log("[cleanUpReadme] Committing README"); + liveUpdate(sharedLogId, "Committing cleaned README to GitHub"); + const commitResult = await commitFile( + accessToken, + repoOwner, + repoName, + "README.md", + cleanedReadme, + "chore: cleanup README [skip ci]", + defaultBranch, + readmeFile.sha, + ); + + console.log("[cleanUpReadme] README committed:", commitResult.commit.sha); + liveUpdate( + sharedLogId, + `✓ README committed: ${commitResult.commit.sha.slice(0, 7)}`, + ); + await UserLogModel.findByIdAndUpdate( + userLog._id, + { + action: "README_CLEANUP_SUCCESS", + status: "success", + commitId: commitResult.commit.sha, + }, + { + new: true, + runValidators: true, + }, + ); + await redis.del("admin_analytics"); + } catch (error) { + console.error("[cleanUpReadme] Failed:", error.message); + + // Only settle the log as failed once no attempt is left, so a transient + // failure does not flash "failed" in the UI before the retry reopens it. + const attemptsAllowed = job.opts.attempts ?? 1; + const attemptsUsed = job.attemptsStarted ?? job.attemptsMade + 1; + const isLastAttempt = + error instanceof UnrecoverableError || attemptsUsed >= attemptsAllowed; + + if (isLastAttempt) { + liveUpdate(sharedLogId, `✗ README cleanup failed: ${error.message}`); + await UserLogModel.findByIdAndUpdate( + userLog._id, + { + action: "README_CLEANUP_FAILED", + status: "failed", + }, + { + new: true, + runValidators: true, + }, + ); + await redis.del("admin_analytics"); + } else { + liveUpdate( + sharedLogId, + `Attempt ${attemptsUsed}/${attemptsAllowed} failed (${error.message}) — retrying`, + ); + } + + throw error; + } +}