diff --git a/.github/workflows/secrets-scanner.yml b/.github/workflows/secrets-scanner.yml index d419edf..ad955d2 100644 --- a/.github/workflows/secrets-scanner.yml +++ b/.github/workflows/secrets-scanner.yml @@ -14,7 +14,7 @@ jobs: runs-on: ubuntu-latest steps: - name: Checkout code - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6 + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6 with: fetch-depth: 0 - name: TruffleHog OSS diff --git a/README.md b/README.md index e0c89e7..f63efee 100644 --- a/README.md +++ b/README.md @@ -269,8 +269,24 @@ We'd love your feedback, ideas, or help building β€” come say hi: - πŸ—£ **Twitter:** [@not_simantak](https://x.com/not_simantak) for the unfiltered version - πŸ› **GitHub Issues:** [Report bugs or request features](https://github.com/tinyfish-io/bigset/issues) + +## Star History + +

+ + Star History Chart + +

+ + ## 🀝 Contributing + + + + +^ This awesome team is behing BigSet! We'd love to have you on board :) + Contributions are very welcome β€” whether it's code, feedback, or just telling us what datasets you'd want to build. 1. Fork the repo diff --git a/backend/src/abort-registry.ts b/backend/src/abort-registry.ts new file mode 100644 index 0000000..3911ea3 --- /dev/null +++ b/backend/src/abort-registry.ts @@ -0,0 +1,53 @@ +/** + * Module-level registry mapping datasetId β†’ AbortController. + * + * Allows the /stop HTTP route to cancel an in-flight populate or update + * workflow. The AbortSignal is retrieved inside Mastra workflow steps + * (which receive no signal parameter from the framework) via `getSignal()` + * and passed explicitly to each `agent.generate()` call. + * + * Keyed by datasetId because: + * - The /stop route knows the datasetId (from the request body). + * - Convex's atomic claim guarantees at most one active run per dataset, + * so datasetId uniquely identifies the in-flight run. + * - Workflow steps have authorizedDatasetId in their inputData β€” no + * separate workflowRunId lookup needed. + * + * Design notes: + * - `abortDataset()` fires the signal but does NOT remove the entry. + * The background runner's `finally` block calls `deregisterDataset()` + * so the catch block can still read `controller.signal.aborted` to + * distinguish a user stop from a genuine failure. + * - All operations are synchronous and safe within a single Node.js process. + */ + +const controllers = new Map(); + +/** Register an active run for a dataset and return its AbortController. */ +export function registerDataset(datasetId: string): AbortController { + const controller = new AbortController(); + controllers.set(datasetId, controller); + return controller; +} + +/** Retrieve the AbortSignal for a dataset's active run (undefined if not running). */ +export function getSignal(datasetId: string): AbortSignal | undefined { + return controllers.get(datasetId)?.signal; +} + +/** + * Fire the abort signal for a dataset's active run. + * Does NOT remove the entry β€” deregisterDataset() handles that. + * Returns true if a run was found and aborted; false if the dataset is idle. + */ +export function abortDataset(datasetId: string): boolean { + const controller = controllers.get(datasetId); + if (!controller) return false; + controller.abort(); + return true; +} + +/** Remove a dataset from the registry. Call in the background runner's finally block. */ +export function deregisterDataset(datasetId: string): void { + controllers.delete(datasetId); +} diff --git a/backend/src/index.ts b/backend/src/index.ts index 57c75e9..26c72ac 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -13,6 +13,7 @@ import { sendTransactionalEmail } from "./email/send.js"; import { datasetReadyTemplate } from "./email/templates/dataset-ready.js"; import { capture, shutdown as shutdownAnalytics } from "./analytics/posthog.js"; import { EVENTS } from "./analytics/events.js"; +import { registerDataset, deregisterDataset, abortDataset } from "./abort-registry.js"; /** Domain part of an email, for analytics (we never log full addresses). */ function emailDomain(email: string): string { @@ -141,6 +142,45 @@ async function sendDatasetReadyNotification({ } } +/** + * Shared stop-success path: set the dataset live, send the ready email. + * + * Called by both background runners when the user presses Stop. Populate + * only emails when at least one row was collected (a stopped run with 0 + * rows is live but empty). Update always emails regardless of count. + */ +async function finaliseRunAsLive({ + logger, + clerk, + datasetId, + authorizedUserId, + workflowType = "populate", +}: { + logger: FastifyBaseLogger; + clerk: ClerkClient; + datasetId: string; + authorizedUserId: string; + workflowType?: "populate" | "update"; +}): Promise { + const currentDataset = await convex.query(internal.datasets.getInternal, { id: datasetId }); + if (!currentDataset) return; + + await setDatasetPopulateStatus(datasetId, "live"); + + const rowCount = await convex.query(internal.datasetRows.countByDataset, { datasetId }); + if (workflowType === "update" || rowCount > 0) { + await sendDatasetReadyNotification({ + logger, + clerk, + userId: authorizedUserId, + datasetId, + datasetName: currentDataset.name, + rowCount, + workflowType, + }); + } +} + async function beginDatasetUpdate( datasetId: string, ownerId: string, @@ -172,6 +212,9 @@ async function runUpdateWorkflowInBackground({ }; }): Promise { const datasetId = input.datasetId; + // registerDataset is called by the route handler before void-ing this + // function, so the registry entry is guaranteed visible the moment the + // 202 response is sent. No call needed here. try { const result = await run.start({ @@ -224,6 +267,11 @@ async function runUpdateWorkflowInBackground({ workflowType: "update", }); } catch (err) { + // Note: a user-triggered stop is NOT handled here. The update workflow's + // refreshRowsStep detects the abort internally, clears pending row + // statuses, and returns normally β€” so run.start() returns { status: + // "success" } and the success path above handles the live transition. + // This catch only fires on genuine failures. const lastStatusError = statusErrorMessage(err); logger.error({ err, datasetId }, "Update background workflow failed"); @@ -245,6 +293,8 @@ async function runUpdateWorkflowInBackground({ "Failed to transition dataset status to 'failed' after update", ); } + } finally { + deregisterDataset(datasetId); } } @@ -266,6 +316,7 @@ async function runScheduledUpdateWorkflowInBackground({ }; }): Promise { const datasetId = input.datasetId; + registerDataset(datasetId); try { const result = await run.start({ @@ -312,12 +363,15 @@ async function runScheduledUpdateWorkflowInBackground({ "Failed to record scheduled refresh failure", ); } + } finally { + deregisterDataset(datasetId); } } async function runPopulateWorkflowInBackground({ input, run, + controller, authorizedUserId, logger, clerk, @@ -325,6 +379,7 @@ async function runPopulateWorkflowInBackground({ }: { input: DatasetContext; run: PopulateWorkflowRun; + controller: AbortController; authorizedUserId: string; logger: FastifyBaseLogger; clerk: ClerkClient; @@ -389,6 +444,25 @@ async function runPopulateWorkflowInBackground({ rowCount, }); } catch (err) { + if (controller.signal.aborted) { + // User pressed Stop β€” treat whatever was collected as the final dataset. + logger.info({ datasetId }, "Populate workflow stopped by user; transitioning to live"); + try { + await finaliseRunAsLive({ logger, clerk, datasetId, authorizedUserId }); + } catch (stopErr) { + logger.error({ err: stopErr, datasetId }, "Failed to finalise stopped populate run; marking as failed"); + // Ensure the dataset always leaves "building" β€” without this fallback, + // a failed finalisation leaves the dataset with no active registry entry + // and no way for /stop to act on it again. + try { + await setDatasetPopulateStatus(datasetId, "failed", "Workflow stopped but could not be finalised"); + } catch (fallbackErr) { + logger.error({ err: fallbackErr, datasetId }, "Could not update dataset status after stop finalisation failure"); + } + } + return; + } + const lastStatusError = statusErrorMessage(err); logger.error( { err, datasetId }, @@ -414,6 +488,8 @@ async function runPopulateWorkflowInBackground({ "Failed to transition dataset status to 'failed'", ); } + } finally { + deregisterDataset(datasetId); } } @@ -492,6 +568,7 @@ function startLocalRefreshScheduler( datasetId: dataset.datasetId, datasetName: dataset.datasetName, description: dataset.description, + maxRowCount: dataset.maxRowCount ?? 100, columns: dataset.columns, }, run, @@ -692,6 +769,13 @@ await fastify.register(async (instance) => { throw new Error(`Unexpected populate claim outcome: ${populateOutcome}`); } + const dataset = await convex.query(internal.datasets.getInternal, { + id: parsed.data.datasetId, + }); + if (!dataset) { + return reply.code(404).send({ error: "Dataset not found" }); + } + const { getModelConfig } = await import("./config/models.js"); const modelConfig = await getModelConfig(auth.userId); @@ -704,9 +788,19 @@ await fastify.register(async (instance) => { return reply.code(502).send({ error: "Failed to populate dataset. Please try again." }); } + // Register before void-ing so the abort-registry entry is visible the + // instant the 202 is sent, closing the TOCTOU window where a /stop + // arriving before registerDataset runs inside the background function + // would incorrectly force-transition an active run to "failed". + const controller = registerDataset(parsed.data.datasetId); + void runPopulateWorkflowInBackground({ - input: parsed.data, + input: { + ...parsed.data, + maxRowCount: dataset.maxRowCount ?? parsed.data.maxRowCount, + }, run, + controller, authorizedUserId: auth.userId, logger: req.log, clerk: req.server.clerk, @@ -772,6 +866,12 @@ await fastify.register(async (instance) => { const { getModelConfig } = await import("./config/models.js"); const modelConfig = await getModelConfig(auth.userId); + // Register before void-ing so the abort-registry entry is visible the + // instant the 202 is sent, closing the TOCTOU window where a /stop + // arriving before registerDataset runs inside the background function + // would incorrectly force-transition an active run to "failed". + registerDataset(parsed.data.datasetId); + void runUpdateWorkflowInBackground({ input: parsed.data, run, @@ -791,6 +891,75 @@ await fastify.register(async (instance) => { return reply.code(502).send({ error: "Failed to update dataset. Please try again." }); } }); + + instance.post("/stop", async (req, reply) => { + const body = req.body as { datasetId?: string }; + if (!body?.datasetId || typeof body.datasetId !== "string") { + return reply.code(400).send({ error: "datasetId is required" }); + } + + const auth = req.auth; + if (!auth) { + return reply.code(401).send({ error: "Authentication required" }); + } + + try { + const dataset = await convex.query(internal.datasets.getInternal, { + id: body.datasetId, + }); + if (!dataset) { + return reply.code(404).send({ error: "Dataset not found" }); + } + if (dataset.ownerId !== auth.userId) { + return reply.code(403).send({ error: "Not authorized to stop this dataset" }); + } + if (dataset.status !== "building" && dataset.status !== "updating") { + return reply.code(409).send({ error: "Dataset is not currently running" }); + } + + const aborted = abortDataset(body.datasetId); + if (!aborted) { + // No registered signal despite the dataset being "building"/"updating". + // The normal finish path always sets a terminal status in Convex + // *before* calling deregisterDataset(), so if the status is still + // busy here, no running process owns this dataset β€” it was orphaned + // by a server restart. Force-transition to "failed" so the dataset + // is no longer stuck. + req.log.warn( + { datasetId: body.datasetId }, + "Stop requested for orphaned dataset (no active run registered); forcing to failed", + ); + try { + if (dataset.status === "updating") { + await convex.mutation(internal.datasetRows.clearAllPendingUpdateStatus, { + datasetId: body.datasetId, + }); + } + await setDatasetPopulateStatus( + body.datasetId, + "failed", + "Run interrupted: server restarted while building/updating", + ); + } catch (statusErr) { + req.log.error( + { err: statusErr, datasetId: body.datasetId }, + "Failed to force-transition orphaned dataset to failed", + ); + } + return reply.code(200).send({ success: true }); + } + req.log.info({ datasetId: body.datasetId }, "Stop requested"); + + return reply.code(202).send({ success: true }); + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + if (msg.includes("validator") || msg.includes("Invalid")) { + return reply.code(400).send({ error: "Invalid datasetId" }); + } + req.log.error(err, "Stop failed"); + return reply.code(502).send({ error: "Failed to stop dataset run. Please try again." }); + } + }); }); try { diff --git a/backend/src/mastra/agents/populate.ts b/backend/src/mastra/agents/populate.ts index 85edf53..155492a 100644 --- a/backend/src/mastra/agents/populate.ts +++ b/backend/src/mastra/agents/populate.ts @@ -10,9 +10,10 @@ const openrouter = createOpenRouter({ apiKey: process.env.OPENROUTER_API_KEY!, }); -const INSTRUCTIONS = `You are an expert dataset builder. You conduct research using your web tools. +function buildInstructions(maxRowCount: number): string { + return `You are an expert dataset builder. You conduct research using your web tools. You do broad research to see which rows to add, and then you spin up sub-agents that can do the deep research and fill in each row for you. -Your job is to make sure you dispatch and manage your army of sub agents to build up a dataset with 100 rows in it. Stop as soon as the dataset reaches 100 rows. +Your job is to make sure you dispatch and manage your army of sub agents to build up a dataset with ${maxRowCount} rows in it. Stop as soon as the dataset reaches ${maxRowCount} rows. WORKFLOW: 1. Understand the data that is is needed and do some research to find places on the web where this data may be obvious and easy to find, collect these links to see what the task of scraping the web is going to look like. @@ -22,12 +23,13 @@ If the dataset is to look at YC Companies, collect links for the YC Startup regi 3. See what the subagent reports back with, if all good and it gives you some information, use that to give better instuctions to subsequent sub agents. -Keep going until you have 100 rows, then finish immediately. If run_subagent reports ROW_LIMIT_REACHED, stop calling tools and finish the run. +Keep going until you have ${maxRowCount} rows, then finish immediately. If run_subagent reports ROW_LIMIT_REACHED, stop calling tools and finish the run. This process should become faster overtime as you just find new rows to go and build, and you keep invoking sub agents in parallel to fill them in. Duplicates are rejected automatically based on primary key columns. If a subagent reports a duplicate, don't re-investigate the same entity β€” move on to a new one. `; +} /** * Build the orchestrator Agent for a populate run. @@ -42,6 +44,7 @@ export function buildPopulateAgent( authorizedDatasetId: string, authContext: AuthContext, columns: PopulateColumn[], + maxRowCount: number, metrics?: RunMetrics, ): Agent { const modelSlug = authContext.modelConfig!.populateOrchestrator; @@ -49,7 +52,7 @@ export function buildPopulateAgent( return new Agent({ id: "populate-agent", name: "Dataset Populate Orchestrator", - instructions: INSTRUCTIONS, + instructions: buildInstructions(maxRowCount), model: openrouter(modelSlug), tools: { search_web: searchWebTool, @@ -58,6 +61,7 @@ export function buildPopulateAgent( authorizedDatasetId, authContext, columns, + maxRowCount, metrics, ), }, diff --git a/backend/src/mastra/tools/investigate-tool.ts b/backend/src/mastra/tools/investigate-tool.ts index 2be1016..c1d6b18 100644 --- a/backend/src/mastra/tools/investigate-tool.ts +++ b/backend/src/mastra/tools/investigate-tool.ts @@ -5,8 +5,7 @@ import { buildInvestigateAgent } from "../agents/investigate.js"; import type { AuthContext } from "../workflows/populate.js"; import type { PopulateColumn } from "../../pipeline/populate.js"; import type { RunMetrics } from "../run-metrics.js"; - -const MAX_DATASET_ROWS = 100; +import { getSignal } from "../../abort-registry.js"; const investigateInputSchema = z.object({ entity_hint: z @@ -76,6 +75,7 @@ export function buildSubagentTool( authorizedDatasetId: string, authContext: AuthContext, columns: PopulateColumn[], + maxRowCount: number, metrics?: RunMetrics, ) { return createTool({ @@ -89,10 +89,10 @@ export function buildSubagentTool( const rowCount = await convex.query(internal.datasetRows.countByDataset, { datasetId: authorizedDatasetId, }); - if (rowCount >= MAX_DATASET_ROWS) { + if (rowCount >= maxRowCount) { return { inserted: false, - reason: `ROW_LIMIT_REACHED: BigSet datasets are capped at ${MAX_DATASET_ROWS} rows. Stop calling run_subagent and finish the run.`, + reason: `ROW_LIMIT_REACHED: this BigSet dataset is capped at ${maxRowCount} rows. Stop calling run_subagent and finish the run.`, row_summary: undefined, clues: undefined, }; @@ -128,7 +128,8 @@ ${pkBlock} Context (partial data already found): ${context}${urlsBlock}${notesBlock}`; - const result = await agent.generate(prompt, { maxSteps: 10 }); + const abortSignal = getSignal(authorizedDatasetId); + const result = await agent.generate(prompt, { abortSignal, maxSteps: 25 }); if (metrics) { // Use result.toolCalls (the flat accumulated list across all steps) rather // than iterating result.steps[n].toolCalls. The per-step arrays are snapshots @@ -151,6 +152,12 @@ ${context}${urlsBlock}${notesBlock}`; ); return parsed; } catch (err) { + // Only propagate an AbortError if OUR signal was actually fired (i.e. + // the user pressed Stop). Network errors in Node.js can also surface as + // AbortError β€” re-throwing those would cause the orchestrator's + // agent.generate() to exit early and return a graceful empty result, + // producing a "0 rows" run without any user action. + if (err instanceof Error && err.name === "AbortError" && getSignal(authorizedDatasetId)?.aborted) throw err; const msg = err instanceof Error ? err.message : String(err); console.error(`[run_subagent] subagent error entity="${entity_hint}" err=${msg}`); return { diff --git a/backend/src/mastra/workflows/populate.ts b/backend/src/mastra/workflows/populate.ts index ea06e0e..35db3b1 100644 --- a/backend/src/mastra/workflows/populate.ts +++ b/backend/src/mastra/workflows/populate.ts @@ -8,6 +8,7 @@ import { DEFAULT_MODEL_IDS } from "../../config/models.js"; import { buildPopulateAgent } from "../agents/populate.js"; import { RunMetrics } from "../run-metrics.js"; import { saveRunMetrics } from "../save-run-metrics.js"; +import { getSignal } from "../../abort-registry.js"; /** * Server-set auth/run context threaded through every step. @@ -116,6 +117,7 @@ Respond with EXACTLY one word: scraper or search`; model: openrouter(modelSlug), prompt: classificationPrompt, maxOutputTokens: 10, + abortSignal: getSignal(inputData.datasetId), }); const answer = result.text.trim().toLowerCase(); if (answer === "scraper" || answer === "search") { @@ -124,6 +126,9 @@ Respond with EXACTLY one word: scraper or search`; console.warn(`[enumerate] Unexpected classification "${answer}", defaulting to "search"`); } } catch (err) { + // Only re-throw if OUR signal was actually fired. A spurious network + // AbortError should fall through and default to "search" as before. + if (err instanceof Error && err.name === "AbortError" && getSignal(inputData.datasetId)?.aborted) throw err; const msg = err instanceof Error ? err.message : String(err); console.error(`[enumerate] Classification failed: ${msg}, defaulting to "search"`); } @@ -152,6 +157,7 @@ const buildPromptOutputSchema = z.object({ authorizedDatasetId: z.string(), authContext: authContextSchema, columns: z.array(populateColumnSchema), + maxRowCount: z.number().int().min(1), }); const buildPromptStep = createStep({ @@ -198,7 +204,7 @@ ${columnsDesc}${pkNote}${manifestNote}${strategyNote} Search the web broadly to find real entities that fit this dataset topic. For each lead you find, call run_subagent with the primary key values and any context/URLs you have found. If run_subagent returns ROW_LIMIT_REACHED, stop immediately and do not make any more tool calls. -Stop the populate run as soon as the dataset reaches 100 rows.`; +Stop the populate run as soon as the dataset reaches ${inputData.maxRowCount} rows.`; console.log( `[build-prompt] Built prompt for ${inputData.datasetName} (${inputData.columns.length} columns, strategy=${inputData.enumerationStrategy})`, @@ -208,6 +214,7 @@ Stop the populate run as soon as the dataset reaches 100 rows.`; authorizedDatasetId: inputData.datasetId, authContext: inputData.authContext, columns: inputData.columns, + maxRowCount: inputData.maxRowCount, }; }, }); @@ -241,16 +248,25 @@ const agentStep = createStep({ inputData.authorizedDatasetId, inputData.authContext, inputData.columns, + inputData.maxRowCount, metrics, ); - const result = await agent.generate(inputData.prompt, { maxSteps: 80 }); + const abortSignal = getSignal(inputData.authorizedDatasetId); + const result = await agent.generate(inputData.prompt, { abortSignal, maxSteps: 80 }); metrics.addOrchestratorResult(result); // Use result.toolCalls (flat accumulated list) β€” same reasoning as investigate-tool.ts. metrics.countToolCalls(result.toolCalls ?? []); return { text: result.text }; } catch (err) { status = "error"; - errorMsg = err instanceof Error ? err.message : String(err); + // Label user-initiated stops clearly in runStats; treat spurious network + // AbortErrors (signal not fired) as regular failures so the error message + // doesn't mislead operators into thinking the user pressed Stop. + if (err instanceof Error && err.name === "AbortError" && getSignal(inputData.authorizedDatasetId)?.aborted) { + errorMsg = "Stopped by user"; + } else { + errorMsg = err instanceof Error ? err.message : String(err); + } console.error(`[populate-agent] agent.generate failed: ${errorMsg}`); throw err; } finally { diff --git a/backend/src/mastra/workflows/update.ts b/backend/src/mastra/workflows/update.ts index 70c8739..45e3421 100644 --- a/backend/src/mastra/workflows/update.ts +++ b/backend/src/mastra/workflows/update.ts @@ -6,6 +6,7 @@ import { buildRefreshAgent } from "../agents/refresh.js"; import { authContextSchema } from "./populate.js"; import { RunMetrics } from "../run-metrics.js"; import { saveRunMetrics } from "../save-run-metrics.js"; +import { getSignal } from "../../abort-registry.js"; export const updateInputSchema = datasetContextSchema.extend({ authContext: authContextSchema, @@ -132,7 +133,8 @@ ${sourcesBlock} ${row.rowSummary ? `\nPrevious summary: ${row.rowSummary}` : ""} ${row.howFound ? `\nPreviously found via: ${row.howFound}` : ""}`; - const result = await agent.generate(prompt, { maxSteps: 10 }); + const abortSignal = getSignal(datasetId); + const result = await agent.generate(prompt, { abortSignal, maxSteps: 10 }); // Accumulate token usage into the investigate tier (refresh agents map // to the investigate tier so the runStats schema needs no new columns). @@ -156,6 +158,10 @@ ${row.howFound ? `\nPreviously found via: ${row.howFound}` : ""}`; `[refresh-rows] Row ${row._id}: updated=${updated} steps=${result.steps?.length ?? "?"} toolCalls=${(result.toolCalls as any[])?.length ?? "?"}`, ); } catch (err) { + // Only re-throw if OUR signal was actually fired. Spurious network + // AbortErrors must not terminate a worker β€” they should be counted as + // row errors so the rest of the dataset continues refreshing. + if (err instanceof Error && err.name === "AbortError" && getSignal(datasetId)?.aborted) throw err; const msg = err instanceof Error ? err.message : String(err); console.error( `[refresh-rows] Row ${row._id} failed: ${msg}`, @@ -182,6 +188,22 @@ ${row.howFound ? `\nPreviously found via: ${row.howFound}` : ""}`; ); await processWithConcurrency(rows, processRow, MAX_CONCURRENT); const finishedAt = Date.now(); + + // If the run was stopped mid-update, workers exited early via AbortError. + // Rows that were never processed still have updateStatus:"pending". + // Clear them now so the UI doesn't show stale shimmer indicators. + const abortSignal = getSignal(datasetId); + if (abortSignal?.aborted) { + console.log(`[refresh-rows] Run was stopped β€” clearing remaining pending row statuses`); + try { + await convex.mutation(internal.datasetRows.clearAllPendingUpdateStatus, { + datasetId, + }); + } catch (cleanupErr) { + console.error(`[refresh-rows] Failed to clear pending update statuses: ${cleanupErr}`); + } + } + console.log( `[refresh-rows] Done: ${updatedCount} updated, ${errors} errors, ${rows.length - updatedCount - errors} unchanged`, ); diff --git a/backend/src/pipeline/populate.ts b/backend/src/pipeline/populate.ts index 55d37aa..589db1e 100644 --- a/backend/src/pipeline/populate.ts +++ b/backend/src/pipeline/populate.ts @@ -1,5 +1,7 @@ import { z } from "zod"; +const FREE_TIER_MONTHLY_QUOTA = 2500; + export const populateColumnSchema = z.object({ name: z.string(), type: z.enum(["text", "number", "boolean", "url", "date"]), @@ -12,6 +14,7 @@ export const datasetContextSchema = z.object({ datasetId: z.string().min(1), datasetName: z.string(), description: z.string(), + maxRowCount: z.number().int().min(1).max(FREE_TIER_MONTHLY_QUOTA).default(100), columns: z.array(populateColumnSchema).min(1), rowIds: z.array(z.string()).min(1).optional(), }); diff --git a/frontend/app/dataset/[id]/page.tsx b/frontend/app/dataset/[id]/page.tsx index adba9a0..d28abda 100644 --- a/frontend/app/dataset/[id]/page.tsx +++ b/frontend/app/dataset/[id]/page.tsx @@ -9,10 +9,12 @@ import { api } from "@/convex/_generated/api"; import type { Id } from "@/convex/_generated/dataModel"; import { DatasetTable } from "@/components/table"; import { useSelection } from "@/components/table/use-selection"; +import { SideSheet, CellDetail } from "@/components/SideSheet"; +import type { DatasetColumn } from "@/components/table/types"; import { useTheme } from "@/components/ThemeToggle"; import { StatusBadge } from "@/components/dataset/StatusBadge"; import { downloadCSV, downloadXLSX } from "@/lib/export"; -import { populate, update } from "@/lib/backend"; +import { populate, update, stopPopulation } from "@/lib/backend"; import { EVENTS, captureException, track } from "@/lib/analytics"; import { REFRESH_CADENCE_OPTIONS, @@ -23,17 +25,25 @@ import type { ProfileUser } from "@/lib/profile-user"; export default function DatasetPage() { const params = useParams(); - const { isLoading: authLoading } = useConvexAuth(); + const { isLoading: authLoading, isAuthenticated } = useConvexAuth(); const { userId, getToken } = useAuth(); const { user } = useUser(); const { signOut } = useClerk(); const [exporting, setExporting] = useState<"csv" | "xlsx" | null>(null); const [populating, setPopulating] = useState(false); const [updating, setUpdating] = useState(false); + const [stopping, setStopping] = useState(false); const [exportOpen, setExportOpen] = useState(false); const [settingsOpen, setSettingsOpen] = useState(false); const [confirmPopulate, setConfirmPopulate] = useState(false); const [savingRefreshCadence, setSavingRefreshCadence] = useState(false); + const [savingMaxRowCount, setSavingMaxRowCount] = useState(false); + const [maxRowCountSaveError, setMaxRowCountSaveError] = useState(null); + const [cellDetail, setCellDetail] = useState<{ + column: DatasetColumn; + value: unknown; + sources?: string[]; + } | null>(null); const datasetId = params.id as Id<"datasets">; const dataset = useQuery( @@ -45,6 +55,11 @@ export default function DatasetPage() { authLoading ? "skip" : { datasetId }, ); const updateRefreshSettings = useMutation(api.datasets.updateRefreshSettings); + const updateMaxRowCount = useMutation(api.datasets.updateMaxRowCount); + const usage = useQuery( + api.quota.getMy, + isAuthenticated ? {} : "skip", + ); const rowIds = useMemo(() => (rows ?? []).map((r) => r._id), [rows]); const selection = useSelection(rowIds); @@ -52,6 +67,8 @@ export default function DatasetPage() { const handlePopulate = useCallback(async () => { if (!dataset || populating || dataset.status === "building") return; + // A new run is starting β€” discard any lingering stop-latch from the previous run. + setStopping(false); setPopulating(true); try { const token = await getToken(); @@ -61,6 +78,7 @@ export default function DatasetPage() { dataset._id, dataset.name, dataset.description, + dataset.maxRowCount ?? 100, dataset.columns, token, ); @@ -80,6 +98,14 @@ export default function DatasetPage() { } }, [dataset, populating, getToken]); + const handleCellExpand = useCallback((columnName: string, value: unknown, rowId: string) => { + if (!dataset || !rows) return; + const col = dataset.columns.find((c) => c.name === columnName); + if (!col) return; + const row = rows.find((r) => r._id === rowId); + setCellDetail({ column: col, value, sources: row?.sources }); + }, [dataset, rows]); + const openedFired = useRef(null); const autoPopulateFired = useRef(null); useEffect(() => { @@ -143,6 +169,8 @@ export default function DatasetPage() { async function handleUpdate() { if (!dataset || updating || dataset.status === "building" || dataset.status === "updating") return; + // A new run is starting β€” discard any lingering stop-latch from the previous run. + setStopping(false); setUpdating(true); try { const token = await getToken(); @@ -195,6 +223,86 @@ export default function DatasetPage() { } } + async function handleMaxRowCountChange(maxRowCount: number) { + if (!dataset || savingMaxRowCount || userId !== dataset.ownerId) return; + if (!Number.isInteger(maxRowCount) || maxRowCount < 1) { + setMaxRowCountSaveError("Max rows must be a whole number greater than 0."); + captureException(new Error("Invalid max row count"), { + operation: "dataset_max_row_count_update", + datasetId: dataset._id, + }); + return; + } + if (usage && maxRowCount > usage.remaining) { + setMaxRowCountSaveError( + `Max rows cannot exceed your remaining monthly quota of ${usage.remaining.toLocaleString()} row operations.`, + ); + return; + } + + setMaxRowCountSaveError(null); + setSavingMaxRowCount(true); + try { + await updateMaxRowCount({ + id: dataset._id, + maxRowCount, + }); + setMaxRowCountSaveError(null); + } catch (err) { + console.error("[max rows] failed", err); + setMaxRowCountSaveError( + err instanceof Error ? err.message : "Failed to update max rows.", + ); + captureException(err, { + operation: "dataset_max_row_count_update", + datasetId: dataset._id, + }); + } finally { + setSavingMaxRowCount(false); + } + } + + async function handleStop() { + if (!dataset || stopping) return; + if (dataset.status !== "building" && dataset.status !== "updating") return; + setStopping(true); + try { + const token = await getToken(); + if (!token) throw new Error("Not authenticated"); + await stopPopulation(dataset._id, token); + track(EVENTS.DATASET_STOP_REQUESTED, { + datasetId: dataset._id, + status: dataset.status, + }); + // Do NOT clear stopping here. Keep it true until Convex confirms the + // dataset has left the busy state (via the isDatasetBusy effect below). + // This prevents the button from briefly re-enabling during the cleanup window. + } catch (err) { + console.error("[stop] failed", err); + captureException(err, { + operation: "dataset_stop", + datasetId: dataset._id, + }); + // Request failed β€” clear immediately so the user can retry. + setStopping(false); + } + } + + // Computed before the loading guard so the useEffect below can depend on it + // without hitting the TDZ. Optional chaining handles the pre-load undefined state. + const isDatasetBusy = dataset?.status === "building" || dataset?.status === "updating"; + + // Clear the stop latch once the dataset is confirmed not-busy by Convex. + // Using setTimeout defers the setState out of the synchronous effect body, + // satisfying the react-hooks/set-state-in-effect lint rule while preserving + // the same semantic: latch clears on the next tick after the status transition. + useEffect(() => { + if (!isDatasetBusy) { + const id = setTimeout(() => setStopping(false), 0); + return () => clearTimeout(id); + } + }, [isDatasetBusy]); + if (authLoading || dataset === undefined || rows === undefined) { return (
@@ -208,15 +316,17 @@ export default function DatasetPage() { // the "Dataset not found" UI. const exportDisabled = exporting !== null || rows.length === 0; - const isDatasetBusy = dataset.status === "building" || dataset.status === "updating"; const isOwner = userId === dataset.ownerId; const displayDataset = { ...dataset, refreshCadence: dataset.refreshCadence ?? "daily", refreshEnabled: dataset.refreshEnabled ?? true, + maxRowCount: dataset.maxRowCount ?? 100, }; const updateDisabled = updating || isDatasetBusy; const populateDisabled = populating || isDatasetBusy; + const stopDisabled = stopping || !isDatasetBusy; + const stopLabel = stopping ? "Stopping…" : "Stop"; const updateLabel = dataset.status === "updating" ? "Updating…" : dataset.status === "building" @@ -267,17 +377,34 @@ export default function DatasetPage() { onExport={(fmt) => { setExportOpen(false); handleExport(fmt); }} /> + {isDatasetBusy && ( + + )} + setSettingsOpen((o) => !o)} onClose={() => setSettingsOpen(false)} refreshCadence={displayDataset.refreshCadence} refreshCadenceDisabled={!isOwner || savingRefreshCadence} + maxRowCount={displayDataset.maxRowCount} + maxRowCountRemaining={usage?.remaining} + maxRowCountSaveError={maxRowCountSaveError} + maxRowCountDisabled={!isOwner || savingMaxRowCount} updateLabel={updateLabel} updateDisabled={updateDisabled} populateLabel={populateLabel} populateDisabled={populateDisabled} onRefreshCadenceChange={handleRefreshCadenceChange} + onMaxRowCountChange={handleMaxRowCountChange} onUpdate={() => { setSettingsOpen(false); handleUpdate(); }} onPopulate={() => { setSettingsOpen(false); @@ -326,8 +453,19 @@ export default function DatasetPage() { rows={rows} datasetId={datasetId} selection={selection} + onCellExpand={handleCellExpand} /> + setCellDetail(null)}> + {cellDetail && ( + + )} + + {confirmPopulate && ( void; refreshCadence: RefreshCadence; refreshCadenceDisabled: boolean; + maxRowCount: number; + maxRowCountRemaining?: number; + maxRowCountSaveError: string | null; + maxRowCountDisabled: boolean; updateLabel: string; updateDisabled: boolean; populateLabel: string; populateDisabled: boolean; onRefreshCadenceChange: (refreshCadence: RefreshCadence) => void; + onMaxRowCountChange: (maxRowCount: number) => void; onUpdate: () => void; onPopulate: () => void; }) { const ref = useRef(null); + const [maxRowCountInput, setMaxRowCountInput] = useState(String(maxRowCount)); + const parsedMaxRowCount = Number(maxRowCountInput); + const maxRowCountValidationError = + !maxRowCountInput.trim() + ? "Required" + : !Number.isInteger(parsedMaxRowCount) || parsedMaxRowCount < 1 + ? "Use a whole number" + : maxRowCountRemaining !== undefined && parsedMaxRowCount > maxRowCountRemaining + ? `Max ${maxRowCountRemaining.toLocaleString()}` + : null; + const maxRowCountChanged = + Number.isInteger(parsedMaxRowCount) && parsedMaxRowCount !== maxRowCount; useEffect(() => { if (!open) return; @@ -463,6 +623,10 @@ function SettingsDropdown({ return () => document.removeEventListener("mousedown", handleClick); }, [open, onClose]); + useEffect(() => { + if (!open) setMaxRowCountInput(String(maxRowCount)); + }, [maxRowCount, open]); + return (
+
+
+ Max rows +
+
+ setMaxRowCountInput(e.currentTarget.value)} + onBlur={() => { + if (!maxRowCountInput.trim()) return; + const value = Number(maxRowCountInput); + if (Number.isInteger(value) && value >= 1) { + setMaxRowCountInput(String(value)); + } + }} + className="min-w-0 flex-1 rounded-lg border border-border bg-background px-2 py-1.5 text-xs text-foreground outline-none transition-colors focus:border-foreground/30 disabled:opacity-50" + /> + +
+

+ {maxRowCountValidationError ?? + maxRowCountSaveError ?? + (maxRowCountRemaining !== undefined + ? `${maxRowCountRemaining.toLocaleString()} row operations available` + : "Applies to the next populate run")} +

+
Refresh cadence diff --git a/frontend/app/dataset/new/page.tsx b/frontend/app/dataset/new/page.tsx index 285cbde..bb4dded 100644 --- a/frontend/app/dataset/new/page.tsx +++ b/frontend/app/dataset/new/page.tsx @@ -4,7 +4,7 @@ import { useEffect, useState, useRef } from "react"; import { useRouter } from "next/navigation"; import Link from "next/link"; import { useAuth } from "@clerk/nextjs"; -import { useMutation, useConvexAuth } from "convex/react"; +import { useMutation, useQuery, useConvexAuth } from "convex/react"; import { api } from "@/convex/_generated/api"; import { EVENTS, track } from "@/lib/analytics"; import { inferSchema, type InferredColumn } from "@/lib/backend"; @@ -43,6 +43,8 @@ const BACKEND_TYPE_MAP: Record = { boolean: "boolean", }; +const DEFAULT_MAX_ROW_COUNT = 100; + function mapBackendColumn(col: InferredColumn, index: number): ProposedColumn { return { id: String(index + 1), @@ -81,6 +83,9 @@ export default function NewDatasetPage() { const [step, setStep] = useState("describe"); const [prompt, setPrompt] = useState(""); const [refreshCadence, setRefreshCadence] = useState("daily"); + const [maxRowCountInput, setMaxRowCountInput] = useState( + String(DEFAULT_MAX_ROW_COUNT), + ); const [columns, setColumns] = useState([]); const [datasetName, setDatasetName] = useState(""); const [isCreating, setIsCreating] = useState(false); @@ -92,6 +97,10 @@ export default function NewDatasetPage() { const { getToken } = useAuth(); const createDataset = useMutation(api.datasets.create); + const usage = useQuery( + api.quota.getMy, + isAuthenticated ? {} : "skip", + ); // Page-view event: fires once when the wizard becomes visible (after // auth resolves and the user is authenticated; we don't want to fire @@ -163,6 +172,17 @@ export default function NewDatasetPage() { async function handleConfirm() { if (isCreating) return; + const maxRowCount = Number(maxRowCountInput); + if (!Number.isInteger(maxRowCount) || maxRowCount < 1) { + setError("Max rows must be a whole number greater than 0."); + return; + } + if (usage && maxRowCount > usage.remaining) { + setError( + `Max rows cannot exceed your remaining monthly quota of ${usage.remaining.toLocaleString()} row operations.`, + ); + return; + } setIsCreating(true); setError(null); let datasetId: string; @@ -171,6 +191,7 @@ export default function NewDatasetPage() { name: datasetName, description: prompt, refreshCadence, + maxRowCount, columns: columns.map((c) => ({ name: c.name, type: c.type, @@ -195,6 +216,7 @@ export default function NewDatasetPage() { datasetId, column_count: columns.length, refreshCadence, + maxRowCount, }); } catch {} router.push(`/dataset/${datasetId}`); @@ -320,6 +342,34 @@ export default function NewDatasetPage() { ))}
+ +
+ + setMaxRowCountInput(e.currentTarget.value)} + onBlur={() => { + if (!maxRowCountInput.trim()) return; + const value = Number(maxRowCountInput); + if (Number.isInteger(value) && value >= 1) { + setMaxRowCountInput(String(value)); + } + }} + className="w-36 rounded-lg border border-border bg-surface px-4 py-2.5 text-sm font-medium outline-none focus:border-foreground/30 transition-colors" + /> + {usage && ( +

+ Up to {usage.remaining.toLocaleString()} row operations available this month. +

+ )} +
diff --git a/frontend/components/SideSheet.tsx b/frontend/components/SideSheet.tsx new file mode 100644 index 0000000..215d477 --- /dev/null +++ b/frontend/components/SideSheet.tsx @@ -0,0 +1,228 @@ +"use client"; + +import { useEffect, useRef, useState, useCallback } from "react"; +import type { DatasetColumn } from "@/components/table/types"; + +function IconX() { + return ( + + ); +} +function IconCopy() { + return ( + + ); +} +function IconCheck() { + return ( + + ); +} +function IconExternalLink() { + return ( + + ); +} + +/* ------------------------------------------------------------------ */ +/* Shell */ +/* ------------------------------------------------------------------ */ + +interface SideSheetProps { + open: boolean; + onClose: () => void; + children: React.ReactNode; +} + +export function SideSheet({ open, onClose, children }: SideSheetProps) { + const panelRef = useRef(null); + + // Lock body scroll while open. + useEffect(() => { + if (!open) return; + const prev = document.body.style.overflow; + document.body.style.overflow = "hidden"; + return () => { document.body.style.overflow = prev; }; + }, [open]); + + // Keyboard: Escape closes, Tab stays trapped inside. + useEffect(() => { + if (!open || !panelRef.current) return; + const panel = panelRef.current; + + const focusable = panel.querySelectorAll( + 'button, [href], input, select, textarea, [tabindex]:not([tabindex="-1"])', + ); + focusable[0]?.focus(); + + function onKey(e: KeyboardEvent) { + if (e.key === "Escape") { onClose(); return; } + if (e.key !== "Tab" || focusable.length === 0) return; + const first = focusable[0]; + const last = focusable[focusable.length - 1]; + if (e.shiftKey && document.activeElement === first) { + e.preventDefault(); last.focus(); + } else if (!e.shiftKey && document.activeElement === last) { + e.preventDefault(); first.focus(); + } + } + + panel.addEventListener("keydown", onKey); + return () => panel.removeEventListener("keydown", onKey); + }, [open, onClose]); + + if (!open) return null; + + return ( +
+ {/* Backdrop */} +
+ {/* Modal */} +
+
+

Cell Detail

+ +
+
{children}
+
+
+ ); +} + +/* ------------------------------------------------------------------ */ +/* Content */ +/* ------------------------------------------------------------------ */ + +interface CellDetailProps { + column: DatasetColumn; + value: unknown; + /** Row-level sources stored by the populate agent. */ + sources?: string[]; +} + +function isValidHttpUrl(src: string): boolean { + try { + const { protocol } = new URL(src); + return protocol === "http:" || protocol === "https:"; + } catch { + return false; + } +} + +export function CellDetail({ column, value, sources }: CellDetailProps) { + const [copied, setCopied] = useState(false); + const copyTimerRef = useRef | null>(null); + const displayValue = value == null || value === "" ? "β€”" : String(value); + + // Clear any pending timer when the component unmounts to avoid calling + // setCopied on an already-gone component. + useEffect(() => { + return () => { + if (copyTimerRef.current != null) clearTimeout(copyTimerRef.current); + }; + }, []); + + const handleCopy = useCallback(async () => { + try { + await navigator.clipboard.writeText(value == null ? "" : String(value)); + setCopied(true); + if (copyTimerRef.current != null) clearTimeout(copyTimerRef.current); + copyTimerRef.current = setTimeout(() => setCopied(false), 2000); + } catch { + // Clipboard API unavailable (e.g. non-HTTPS dev); silently ignore. + } + }, [value]); + + return ( +
+ {/* Column name + description */} +
+

{column.name}

+ {column.description && ( +

{column.description}

+ )} +
+ + {/* Value */} +
+
+

+ Value ({column.type}) +

+ +
+
+

+ {displayValue} +

+
+
+ + {/* Sources */} + {sources && sources.length > 0 && ( +
+

+ Sources +

+
    + {sources.map((src, i) => ( +
  • + {isValidHttpUrl(src) ? ( + + + {src} + + ) : ( + + + {src} + + )} +
    • + ))} +
+
+ )} +
+ ); +} diff --git a/frontend/components/ThemeToggle.tsx b/frontend/components/ThemeToggle.tsx index 346a509..c185991 100644 --- a/frontend/components/ThemeToggle.tsx +++ b/frontend/components/ThemeToggle.tsx @@ -32,11 +32,19 @@ function applyTheme(theme: Theme): void { function subscribeToThemeChange(onThemeChange: () => void): () => void { if (typeof window === "undefined") return () => {}; - window.addEventListener("storage", onThemeChange); - window.addEventListener(THEME_CHANGED_EVENT, onThemeChange); + const mediaQuery = window.matchMedia("(prefers-color-scheme: dark)"); + function syncTheme() { + applyTheme(readEffectiveTheme()); + onThemeChange(); + } + + window.addEventListener("storage", syncTheme); + window.addEventListener(THEME_CHANGED_EVENT, syncTheme); + mediaQuery.addEventListener("change", syncTheme); return () => { - window.removeEventListener("storage", onThemeChange); - window.removeEventListener(THEME_CHANGED_EVENT, onThemeChange); + window.removeEventListener("storage", syncTheme); + window.removeEventListener(THEME_CHANGED_EVENT, syncTheme); + mediaQuery.removeEventListener("change", syncTheme); }; } diff --git a/frontend/components/table/DataRow.tsx b/frontend/components/table/DataRow.tsx index 1fe0974..2d54661 100644 --- a/frontend/components/table/DataRow.tsx +++ b/frontend/components/table/DataRow.tsx @@ -7,12 +7,21 @@ import type { DatasetRow, DatasetColumn } from "./types"; import { CellValue } from "./CellValue"; import { floorWidth } from "./utils"; +function IconMaximize2() { + return ( + + ); +} + export interface DataRowData { rows: Row[]; columns: DatasetColumn[]; columnWidths: number[]; isSelected: (id: string) => boolean; toggleRow: (id: string, shiftKey: boolean) => void; + onCellExpand: (columnName: string, value: unknown, rowId: string) => void; isBuilding: boolean; pendingRowIds: Set; flashingCells: Set; @@ -27,7 +36,7 @@ function DataRowImpl({ index: number; style: CSSProperties; }) { - const { rows, columns, columnWidths, isSelected, toggleRow, isBuilding, pendingRowIds, flashingCells } = data; + const { rows, columns, columnWidths, isSelected, toggleRow, onCellExpand, isBuilding, pendingRowIds, flashingCells } = data; const row = rows[index]; if (!row) { @@ -112,7 +121,7 @@ function DataRowImpl({
+ {isPending &&
}
); diff --git a/frontend/components/table/DatasetTable.tsx b/frontend/components/table/DatasetTable.tsx index f921875..e74c2bb 100644 --- a/frontend/components/table/DatasetTable.tsx +++ b/frontend/components/table/DatasetTable.tsx @@ -85,11 +85,13 @@ export function DatasetTable({ rows, datasetId, selection, + onCellExpand, }: { dataset: DatasetMeta; rows: DatasetRow[]; datasetId: string; selection: Selection; + onCellExpand: (columnName: string, value: unknown, rowId: string) => void; }) { const tableContainerRef = useRef(null); const previousResizingColumnIdRef = useRef(false); @@ -168,11 +170,12 @@ export function DatasetTable({ columnWidths, isSelected: selection.has, toggleRow, + onCellExpand, isBuilding, pendingRowIds, flashingCells, }), - [tableRows, dataset.columns, columnWidths, selection.has, toggleRow, isBuilding, pendingRowIds, flashingCells], + [tableRows, dataset.columns, columnWidths, selection.has, toggleRow, onCellExpand, isBuilding, pendingRowIds, flashingCells], ); return ( diff --git a/frontend/components/table/types.ts b/frontend/components/table/types.ts index 5fc380c..6eff009 100644 --- a/frontend/components/table/types.ts +++ b/frontend/components/table/types.ts @@ -28,5 +28,6 @@ export interface DatasetRow { _id: string; _creationTime: number; data: Record; + sources?: string[]; updateStatus?: "pending"; } diff --git a/frontend/convex/datasetRows.ts b/frontend/convex/datasetRows.ts index db609f7..eac39ea 100644 --- a/frontend/convex/datasetRows.ts +++ b/frontend/convex/datasetRows.ts @@ -5,7 +5,7 @@ import type { Id } from "./_generated/dataModel.js"; import { assertRowInDataset, loadReadableDataset } from "./lib/authz.js"; import { consumeQuotaForDataset } from "./lib/quota.js"; -const MAX_DATASET_ROWS = 100; +const DEFAULT_MAX_DATASET_ROWS = 100; /** * Authoritative row count for a dataset. O(N), so use only on the slow @@ -77,9 +77,10 @@ export const insert = internalMutation({ typeof dataset.rowCount === "number" ? dataset.rowCount : await actualRowCount(ctx, args.datasetId); - if (previousCount >= MAX_DATASET_ROWS) { + const maxRowCount = dataset.maxRowCount ?? DEFAULT_MAX_DATASET_ROWS; + if (previousCount >= maxRowCount) { throw new Error( - `Row limit reached: BigSet datasets are capped at ${MAX_DATASET_ROWS} rows. Stop inserting rows and finish the run.`, + `Row limit reached: this BigSet dataset is capped at ${maxRowCount} rows. Stop inserting rows and finish the run.`, ); } @@ -311,6 +312,34 @@ export const clearUpdateStatus = internalMutation({ }, }); +/** + * Bulk-clear all pending update statuses for a dataset. + * + * Called when a user stops an in-flight update workflow. Workers exit early + * via AbortError, so rows they never reached still have `updateStatus: + * "pending"`. This clears them so the UI doesn't show stale shimmer + * indicators after the run is marked live. + * + * Uses the `by_dataset_update_status` compound index so only the pending + * rows are scanned β€” the query never touches rows that have already been + * processed (updateStatus === undefined). + */ +export const clearAllPendingUpdateStatus = internalMutation({ + args: { datasetId: v.id("datasets") }, + handler: async (ctx, args) => { + const pendingRows = await ctx.db + .query("datasetRows") + .withIndex("by_dataset_update_status", (q) => + q.eq("datasetId", args.datasetId).eq("updateStatus", "pending"), + ) + .collect(); + for (const row of pendingRows) { + await ctx.db.patch(row._id, { updateStatus: undefined }); + } + return pendingRows.length; + }, +}); + /** * Admin-only row listing for a dataset. Used by the populate agent's * `list_rows` tool to see what's already been inserted in the dataset diff --git a/frontend/convex/datasets.ts b/frontend/convex/datasets.ts index 4f34619..d295327 100644 --- a/frontend/convex/datasets.ts +++ b/frontend/convex/datasets.ts @@ -13,7 +13,7 @@ import { loadReadableDataset, requireIdentity, } from "./lib/authz.js"; -import { requireQuotaRemaining } from "./lib/quota.js"; +import { FREE_TIER_MONTHLY_QUOTA, requireQuotaRemaining } from "./lib/quota.js"; import { nextRefreshAtFor, refreshCadenceValidator, @@ -62,6 +62,19 @@ function refreshCadenceFromLegacyLabel( } const PREVIEW_ROW_COUNT = 5; +const DEFAULT_MAX_ROW_COUNT = 100; + +function validateMaxRowCount(maxRowCount: number): void { + if ( + !Number.isInteger(maxRowCount) || + maxRowCount < 1 || + maxRowCount > FREE_TIER_MONTHLY_QUOTA + ) { + throw new Error( + `Max row count must be a whole number between 1 and ${FREE_TIER_MONTHLY_QUOTA}.`, + ); + } +} async function attachPreview(ctx: QueryCtx, dataset: Doc<"datasets">) { // Mini-table preview: just the first N rows. `.take` keeps the @@ -270,6 +283,7 @@ export const claimScheduledRefreshInternal = internalMutation({ description: dataset.description, columns: dataset.columns, ownerId: dataset.ownerId, + maxRowCount: dataset.maxRowCount ?? DEFAULT_MAX_ROW_COUNT, }, }; }, @@ -360,6 +374,7 @@ export const create = mutation({ name: v.string(), description: v.string(), refreshCadence: refreshCadenceValidator, + maxRowCount: v.number(), columns: v.array(columnValidator), retrievalStrategy: v.optional( v.union( @@ -373,10 +388,11 @@ export const create = mutation({ handler: async (ctx, args) => { const identity = await requireIdentity(ctx); assertNotReservedOwner(identity.subject); + validateMaxRowCount(args.maxRowCount); // Block dataset creation at full exhaustion β€” a dataset you can't // populate is just clutter. Row generation later will re-check, so // this is a UX safeguard, not the only line of defense. - await requireQuotaRemaining(ctx, identity.subject, 1); + await requireQuotaRemaining(ctx, identity.subject, args.maxRowCount); return await ctx.db.insert("datasets", { ...args, @@ -408,6 +424,23 @@ export const updateRefreshSettings = mutation({ }, }); +export const updateMaxRowCount = mutation({ + args: { + id: v.id("datasets"), + maxRowCount: v.number(), + }, + handler: async (ctx, args) => { + const dataset = await loadOwnedDataset(ctx, args.id); + validateMaxRowCount(args.maxRowCount); + const currentRowCount = dataset.rowCount ?? 0; + const additionalRowsNeeded = Math.max(0, args.maxRowCount - currentRowCount); + await requireQuotaRemaining(ctx, dataset.ownerId, additionalRowsNeeded); + await ctx.db.patch(dataset._id, { + maxRowCount: args.maxRowCount, + }); + }, +}); + export const backfillRefreshSettings = internalMutation({ args: { defaultCadence: v.optional(refreshCadenceValidator), diff --git a/frontend/convex/schema.ts b/frontend/convex/schema.ts index 7907d6c..4e7b5f4 100644 --- a/frontend/convex/schema.ts +++ b/frontend/convex/schema.ts @@ -50,6 +50,9 @@ export default defineSchema({ // with rows created before this field existed β€” write paths self-heal // on first hit, and `datasets.backfillRowCounts` migrates all at once. rowCount: v.optional(v.number()), + // User-selected target/limit for populate runs. Optional so existing + // datasets keep the legacy 100-row behavior until touched. + maxRowCount: v.optional(v.number()), columns: v.array( v.object({ name: v.string(), @@ -86,7 +89,11 @@ export default defineSchema({ howFound: v.optional(v.string()), updateStatus: v.optional(v.literal("pending")), scrapeScript: v.optional(v.string()), - }).index("by_dataset", ["datasetId"]), + }) + .index("by_dataset", ["datasetId"]) + // Compound index used by clearAllPendingUpdateStatus to scan only the rows + // that need clearing without a full-dataset read. + .index("by_dataset_update_status", ["datasetId", "updateStatus"]), datasetHistory: defineTable({ datasetRowId: v.id("datasetRows"), diff --git a/frontend/lib/analytics.ts b/frontend/lib/analytics.ts index bb2cb4a..a2a2757 100644 --- a/frontend/lib/analytics.ts +++ b/frontend/lib/analytics.ts @@ -34,6 +34,7 @@ export const EVENTS = { DATASET_EXPORTED: "dataset_exported", DATASET_POPULATE_STARTED: "dataset_populate_started", DATASET_UPDATE_STARTED: "dataset_update_started", + DATASET_STOP_REQUESTED: "dataset_stop_requested", // Creation flow DATASET_CREATION_STARTED: "dataset_creation_started", diff --git a/frontend/lib/backend.ts b/frontend/lib/backend.ts index 9406a18..f023a35 100644 --- a/frontend/lib/backend.ts +++ b/frontend/lib/backend.ts @@ -211,6 +211,7 @@ export async function populate( datasetId: string, datasetName: string, description: string, + maxRowCount: number, columns: PopulateColumn[], token: string, ): Promise { @@ -220,7 +221,7 @@ export async function populate( "Content-Type": "application/json", Authorization: `Bearer ${token}`, }, - body: JSON.stringify({ datasetId, datasetName: datasetName, description, columns }), + body: JSON.stringify({ datasetId, datasetName, description, maxRowCount, columns }), }); if (!res.ok) { @@ -259,3 +260,25 @@ export async function update( return res.json(); } + +export async function stopPopulation( + datasetId: string, + token: string, +): Promise<{ success: boolean }> { + const res = await fetch(`${BACKEND_URL}/stop`, { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: `Bearer ${token}`, + }, + body: JSON.stringify({ datasetId }), + }); + + if (!res.ok) { + const body = await res.json().catch(() => null); + const message = body?.error || `Backend error (${res.status})`; + throw new Error(message); + } + + return res.json(); +} diff --git a/makefiles/Makefile b/makefiles/Makefile index 81c3d6b..2eb758a 100644 --- a/makefiles/Makefile +++ b/makefiles/Makefile @@ -31,10 +31,11 @@ dev: validate-dev-env install-deps validate-dev-env: @test -f .env || { echo "Error: .env not found. Run: cp .env.example .env"; exit 1; } - @missing=""; \ + @env_value() { grep "^$$1=" .env | tail -n1 | cut -d= -f2- | tr -d '\r'; }; \ + missing=""; \ check_env() { \ key="$$1"; placeholder="$$2"; \ - value="$$(grep "^$$key=" .env | cut -d= -f2-)"; \ + value="$$(env_value "$$key")"; \ if [[ -z "$$value" || "$$value" == "$$placeholder" || "$$value" == *"..."* ]]; then \ missing="$$missing $$key"; \ fi; \ @@ -84,7 +85,8 @@ validate-dev-env: fi ensure-admin-key: - @admin_key="$$(grep '^CONVEX_SELF_HOSTED_ADMIN_KEY=' .env | cut -d= -f2-)"; \ + @env_value() { grep "^$$1=" .env | tail -n1 | cut -d= -f2- | tr -d '\r'; }; \ + admin_key="$$(env_value CONVEX_SELF_HOSTED_ADMIN_KEY)"; \ needs_gen=false; \ if [[ -z "$$admin_key" ]]; then \ needs_gen=true; \ @@ -115,8 +117,9 @@ ensure-admin-key: convex-env: @test -f .env || { echo "Error: .env not found. Run: cp .env.example .env"; exit 1; } - @issuer="$$(grep '^CLERK_JWT_ISSUER_DOMAIN=' .env | cut -d= -f2-)"; \ - admin_key="$$(grep '^CONVEX_SELF_HOSTED_ADMIN_KEY=' .env | cut -d= -f2-)"; \ + @env_value() { grep "^$$1=" .env | tail -n1 | cut -d= -f2- | tr -d '\r'; }; \ + issuer="$$(env_value CLERK_JWT_ISSUER_DOMAIN)"; \ + admin_key="$$(env_value CONVEX_SELF_HOSTED_ADMIN_KEY)"; \ if [[ -z "$$issuer" || "$$issuer" == "https://your-app.clerk.accounts.dev" ]]; then \ echo "Error: CLERK_JWT_ISSUER_DOMAIN must be your Clerk issuer URL in root .env"; \ exit 1; \ @@ -132,7 +135,8 @@ convex-env: convex-push: @test -f .env || { echo "Error: .env not found. Run: cp .env.example .env"; exit 1; } - @admin_key="$$(grep '^CONVEX_SELF_HOSTED_ADMIN_KEY=' .env | cut -d= -f2-)"; \ + @env_value() { grep "^$$1=" .env | tail -n1 | cut -d= -f2- | tr -d '\r'; }; \ + admin_key="$$(env_value CONVEX_SELF_HOSTED_ADMIN_KEY)"; \ if [[ -z "$$admin_key" ]]; then \ echo "Error: CONVEX_SELF_HOSTED_ADMIN_KEY is missing in root .env"; \ echo "Run make dev to generate it automatically."; \