Add AI-powered dataset populate with web search and CRUD tools#26
Conversation
Introduces the "Clear & Populate" flow: an AI agent (Claude Sonnet 4.6 via OpenRouter) searches the web using TinyFish APIs, fetches page content, and inserts real data into datasets row by row. Backend: - Mastra populate workflow (clear rows → build prompt → run agent) - Populate agent with 7 tools: 5 database CRUD (insert, list, get, update, delete) + 2 web (search_web via TinyFish Search API, fetch_page via TinyFish Fetch API) - All tools return structured errors so the agent can self-correct - Data keys are sanitized to strip stray quotes/backticks from LLM output - Fetch responses capped at 15K chars to protect agent context window - Convex client uses anyApi to avoid cross-project imports in Docker - POST /populate route with Clerk JWT auth Frontend: - "Clear & Populate" button on dataset detail page - API client function in lib/backend.ts - Rows appear in realtime via Convex reactive queries Convex: - New internal functions: datasetRows.get (query) and datasetRows.remove (mutation) for single-row read/delete Infra: - TINYFISH_API_KEY wired through docker-compose.dev.yml to backend and mastra services Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
📝 WalkthroughWalkthroughThis PR implements a dataset population feature: Zod schemas and frontend types; Convex runtime proxies and new internal row mutations/queries; Mastra dataset CRUD and web tools that call TinyFish, a Claude-based populate agent, and a three-step populateWorkflow (clear → prompt → agent); a protected POST /populate endpoint that starts the workflow; frontend UI to trigger population with token auth and analytics; and documentation + env/docker updates adding TINYFISH_API_KEY. Sequence Diagram(s)sequenceDiagram
participant Client
participant Backend
participant Mastra
participant TinyFish
participant Convex
Client->>Backend: POST /populate (DatasetContext + Bearer token)
Backend->>Mastra: start populateWorkflow(inputData)
Mastra->>TinyFish: search_web / fetch_page (TinyFish API)
Mastra->>Convex: insert_row / update_row / clearByDataset
Mastra-->>Backend: workflow result (text / status)
Backend-->>Client: response (success/result or error)
Possibly related PRs
Suggested reviewers
🚥 Pre-merge checks | ✅ 4✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@backend/src/index.ts`:
- Around line 53-77: The route handler registered with
instance.post("/populate") currently trusts datasetId from parsed.data; before
creating the run via populateWorkflow.createRun() you must enforce ownership by
extracting the authenticated user id from req.auth.userId, resolving/validating
the target dataset (e.g., lookup dataset by parsed.data.datasetId or derive
datasetId from the user's account) and comparing its ownerId to req.auth.userId,
rejecting with reply.code(403) if they don’t match or replacing
parsed.data.datasetId with the authorized id; ensure this check happens before
calling populateWorkflow.createRun() or run.start() and use the same identifiers
(parsed.data.datasetId, req.auth.userId, populateWorkflow.createRun, run.start)
to locate and fix the code.
In `@backend/src/mastra/tools/dataset-tools.ts`:
- Line 33: The console.log in the insert_row flow is printing raw row payload
(JSON.stringify(cleanedData)) which can leak PII; change it to log only
metadata: datasetId, the row id (e.g., cleanedData.id or cleanedData.row_id if
present), and the number of keys (Object.keys(cleanedData).length). Update the
log call in the insert_row area where the string `[insert_row] Inserting row
into ${datasetId}:` is used and the similar occurrence later (around the other
occurrence) to remove payload printing and instead output the safe metadata;
prefer using the project's logger if one exists (replace console.log with
logger.info/logger.debug) to keep logs consistent.
In `@backend/src/mastra/tools/web-tools.ts`:
- Around line 33-35: The two outbound fetch calls in web-tools.ts (the search
and page requests where you create const res = await fetch(url, { headers: {
"X-API-Key": apiKey } })) need AbortController-based timeouts: create an
AbortController before the fetch, start a setTimeout that calls
controller.abort() after a safe timeout (e.g. 10s), pass controller.signal in
the fetch options, and clear the timeout after the fetch completes; also catch
and handle the abort error (check for error.name === 'AbortError') to return a
deterministic timeout error/path. Update both fetch sites in web-tools.ts
accordingly (the functions handling the search and page requests) so they use
this pattern.
In `@frontend/convex/datasetRows.ts`:
- Around line 75-86: The clearByDataset mutation currently collects all rows via
the by_dataset index and deletes them in one mutation which can exceed Convex
limits; change the handler in clearByDataset to page through the by_dataset
index using a bounded take (e.g., 1000) and only delete that page of row IDs,
then if the page was full schedule the next batch by calling
ctx.scheduler.runAfter(0, { mutation: "clearByDataset", args: { datasetId:
args.datasetId } }) (or similar scheduling API) so subsequent batches run in new
mutations; ensure you use the index-backed query with pagination (take/after or
cursor) and return the number of rows deleted for the current batch.
In `@frontend/lib/backend.ts`:
- Around line 26-29: The PopulateResult type is mismatched: the backend returns
{ success, result } not { success, rows }; update the frontend type
(PopulateResult) to match the backend response shape by renaming or replacing
the rows property with result: unknown[] (or Record<string, unknown>[] if items
are objects) and update any uses (e.g., populate() consumer code) to reference
PopulateResult.result; ensure the type change is applied consistently where
PopulateResult is imported or referenced in frontend/lib/backend.ts.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: dee9c5b6-b683-45a9-835b-0e6ab082f41f
📒 Files selected for processing (16)
CLAUDE.mdbackend/.env.examplebackend/CLAUDE.mdbackend/src/convex.tsbackend/src/index.tsbackend/src/mastra/agents/populate.tsbackend/src/mastra/index.tsbackend/src/mastra/tools/dataset-tools.tsbackend/src/mastra/tools/web-tools.tsbackend/src/mastra/workflows/populate.tsbackend/src/pipeline/populate.tsdocker-compose.dev.ymlfrontend/app/dataset/[id]/page.tsxfrontend/convex/datasetRows.tsfrontend/lib/analytics.tsfrontend/lib/backend.ts
| export const clearByDataset = internalMutation({ | ||
| args: { datasetId: v.id("datasets") }, | ||
| handler: async (ctx, args) => { | ||
| const rows = await ctx.db | ||
| .query("datasetRows") | ||
| .withIndex("by_dataset", (q) => q.eq("datasetId", args.datasetId)) | ||
| .collect(); | ||
| for (const row of rows) { | ||
| await ctx.db.delete(row._id); | ||
| } | ||
| return rows.length; | ||
| }, |
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
Convex documentation: what are the per-mutation read/write/execution limits, and what is the recommended pattern to delete many documents by index safely?
💡 Result:
Per-mutation read/write/execution limits (Convex) 1) Execution time (per query or mutation function) - Query/mutation execution time: 1 second (note: this limit applies only to user code and does not include database operations). [1] - Action execution time: 10 minutes (actions aren’t mutations/queries, but included in the same limits table). [1] 2) Data read/write limits (per query or mutation function) - Data read: 16 MiB. [1] - Data written: 16 MiB. [1] - Function return value size: 16 MiB. [1] 3) Document/index-operation limits (per query or mutation function) - Documents scanned: 32,000. [1] - Documents written: 16,000. [1] - Index ranges read: 4,096 (i.e., the number of calls to db.get and db.query). [1] 4) Mutation count limits (per plan / deployment-class columns) - Mutations: 16 (Starter), 256 (Professional), 512 (Business/Enterprise in the table). [1] How to delete many documents by index safely (recommended pattern) Core idea: don’t attempt “load/collect then delete everything” in a single mutation. Instead, delete in bounded batches using an index-backed query, and if you might exceed transaction limits, continue the work in subsequent transactions scheduled after commit. A) Use an index-backed query to find candidate IDs - Use ctx.db.query(table).withIndex(...).filter(...) - Then only process a limited amount per transaction using take or pagination. Convex’s limits are enforced per single query/mutation execution. [1][2] B) Delete by document ID (write within the same bounded transaction) - Within that mutation batch, delete each document via ctx.db.delete(table, id) (or ctx.db.delete(tableId)). [3] C) If the workload might exceed limits, split it into multiple transactions - Convex recommends using tooling to keep within read/write limits and explicitly states you can “make batches of work dynamically and handle each batch in a separate transaction, using the scheduler or having an action call multiple mutations.” [3] D) Use pagination and an index for progress-safe batched work - For large workloads, use pagination; Convex has a dedicated paginate flow returning a cursor for subsequent pages. [4][5] - When deleting many rows, a safe approach is: query the next page of matching docs via an index, delete those IDs, then schedule the next batch. E) Schedule the next batch after the current mutation commits (safe continuation) - Use ctx.scheduler.runAfter(0,...) to queue the next batch immediately after the current mutation commits. [6] - This makes the process robust: if your current batch hits limits, you can continue in the next scheduled mutation rather than failing the whole delete. [3][7] Concrete skeleton pattern (index + bounded batch + scheduler) - internalClearTable (internalMutation) - query with index + filter - take/paginate a bounded number of docs - loop over returned docs and ctx.db.delete("table", doc._id) - read transaction metrics and, if near limits, schedule self again using ctx.scheduler.runAfter(0,...) Convex docs include an example “clearTasks” pattern that: - iterates matching docs, - checks metrics like bytesRead/bytesWritten and remaining databaseQueries, - and when close to limits, schedules the same clearing mutation again with ctx.scheduler.runAfter(0,...). [3] Practical notes / guardrails - Mutations run transactionally, so within each batch your reads are consistent and your writes commit together; if a mutation throws, its writes are rolled back. [8] - If you expect “1000+ documents” or an unbounded number of results, Convex best practices recommend adding an index to filter further before collect/taking all rows. [9] If you tell me your table schema and which index you plan to use (index name + filter condition), I can translate the skeleton into a tailored Convex mutation + pagination/scheduler implementation that stays comfortably under the numeric limits above.
Citations:
- 1: https://docs.convex.dev/production/state/limits
- 2: https://docs.convex.dev/functions/error-handling/
- 3: https://docs.convex.dev/database/writing-data
- 4: https://docs.convex.dev/database/pagination.md
- 5: https://docs.convex.dev/using/database-queries
- 6: https://docs.convex.dev/api/interfaces/server.Scheduler
- 7: https://docs.convex.dev/scheduling/scheduled-functions
- 8: https://docs.convex.dev/functions/mutation-functions
- 9: https://docs.convex.dev/understanding/best-practices/
Batch dataset row deletion to stay within Convex per-mutation limits
clearByDataset currently .collect()s all datasetRows for the dataset and then deletes them sequentially in a single mutation (lines 78-84). As row counts grow this can hit Convex per-mutation limits (e.g., data read/write 16 MiB and documents scanned 32,000), making “Clear & Populate” brittle.
Switch to an index-backed, bounded batch pattern: query the next page of IDs via the by_dataset index (with take/pagination), delete only that page, and if more remain schedule the next batch (e.g., via ctx.scheduler.runAfter(0, ...)) rather than doing an unbounded load-and-delete in one mutation.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@frontend/convex/datasetRows.ts` around lines 75 - 86, The clearByDataset
mutation currently collects all rows via the by_dataset index and deletes them
in one mutation which can exceed Convex limits; change the handler in
clearByDataset to page through the by_dataset index using a bounded take (e.g.,
1000) and only delete that page of row IDs, then if the page was full schedule
the next batch by calling ctx.scheduler.runAfter(0, { mutation:
"clearByDataset", args: { datasetId: args.datasetId } }) (or similar scheduling
API) so subsequent batches run in new mutations; ensure you use the index-backed
query with pagination (take/after or cursor) and return the number of rows
deleted for the current batch.
There was a problem hiding this comment.
Acknowledged — valid concern for scale. Our datasets are currently 10-15 rows so we're well within Convex per-mutation limits. We'll revisit with batched deletion + ctx.scheduler.runAfter when we support larger datasets.
There was a problem hiding this comment.
@simantak-dabhade, that makes sense — at 10–15 rows per dataset you're nowhere near the 32,000-document-scan or 16 MiB limits. Deferring until larger dataset support is warranted is a reasonable call.
Would you like me to open a GitHub issue to track the future work of switching clearByDataset to a bounded batch + ctx.scheduler.runAfter pattern? That way it won't get lost when you revisit scale.
- Enforce dataset ownership on POST /populate by querying Convex for the dataset and comparing ownerId to req.auth.userId before running the workflow (fixes authz gap) - Remove raw row payloads from insert_row/update_row logs, log column count instead to avoid PII leakage - Add 30s AbortController timeouts to both TinyFish fetch calls in web-tools.ts so they can't hang indefinitely - Align PopulateResult type (rows → result) to match actual backend response shape Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@backend/src/index.ts`:
- Around line 63-69: Wrap the convex.query call for api.datasets.get (using
parsed.data.datasetId) in a try/catch before performing the dataset.ownerId
authorization check so upstream query failures or invalid IDs are handled as
controlled API responses; catch errors from convex.query(api.datasets.get, { id:
parsed.data.datasetId }) and return an appropriate reply (e.g.,
reply.code(400).send({ error: "Invalid dataset id" }) for bad input or
reply.code(500).send({ error: "Failed to fetch dataset" }) for server errors)
instead of letting exceptions escape, then proceed with the existing null check
and ownerId comparison.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: a7520fc1-bd16-4928-83df-c243eafa9976
📒 Files selected for processing (4)
backend/src/index.tsbackend/src/mastra/tools/dataset-tools.tsbackend/src/mastra/tools/web-tools.tsfrontend/lib/backend.ts
Convex query for dataset lookup can throw on invalid IDs — wrapping it in the existing try/catch ensures controlled 400 responses instead of unhandled 500s. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@backend/src/index.ts`:
- Around line 83-86: Only treat the error as a client-side "Invalid datasetId"
when it actually comes from the dataset lookup: change the condition that
currently checks msg.includes("Invalid") to a stricter check such as msg ===
"Invalid datasetId" or msg.includes("Invalid datasetId") and/or verify the error
originates from the dataset lookup function (e.g., getDatasetById) or has a
specific error.name/marker; leave other errors to fall through so they return
the appropriate server error status instead of reply.code(400).send({ error:
"Invalid datasetId" }).
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 1fdcbae0-e13b-4bdc-a17b-337e7fb99b45
📒 Files selected for processing (1)
backend/src/index.ts
| const msg = err instanceof Error ? err.message : String(err); | ||
| if (msg.includes("validator") || msg.includes("Invalid")) { | ||
| return reply.code(400).send({ error: "Invalid datasetId" }); | ||
| } |
There was a problem hiding this comment.
Narrow invalid-dataset detection to dataset lookup failures only.
Line 84 matches any error containing "Invalid", so downstream workflow/tool failures can be mislabeled as 400 Invalid datasetId. That can hide real server-side failures and return the wrong status code.
Suggested fix
- try {
- const dataset = await convex.query(api.datasets.get, { id: parsed.data.datasetId });
- if (!dataset) {
- return reply.code(404).send({ error: "Dataset not found" });
- }
- if (dataset.ownerId !== req.auth.userId) {
- return reply.code(403).send({ error: "Not authorized to populate this dataset" });
- }
-
- const run = await populateWorkflow.createRun();
- const result = await run.start({ inputData: parsed.data });
+ let dataset;
+ try {
+ dataset = await convex.query(api.datasets.get, { id: parsed.data.datasetId });
+ } 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, "Dataset lookup failed");
+ return reply.code(502).send({ error: "Failed to populate dataset. Please try again." });
+ }
+
+ if (!dataset) {
+ return reply.code(404).send({ error: "Dataset not found" });
+ }
+ if (dataset.ownerId !== req.auth.userId) {
+ return reply.code(403).send({ error: "Not authorized to populate this dataset" });
+ }
+
+ try {
+ const run = await populateWorkflow.createRun();
+ const result = await run.start({ inputData: parsed.data });
@@
- } 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" });
- }
+ } catch (err) {
req.log.error(err, "Populate failed");
return reply.code(502).send({ error: "Failed to populate dataset. Please try again." });
}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/src/index.ts` around lines 83 - 86, Only treat the error as a
client-side "Invalid datasetId" when it actually comes from the dataset lookup:
change the condition that currently checks msg.includes("Invalid") to a stricter
check such as msg === "Invalid datasetId" or msg.includes("Invalid datasetId")
and/or verify the error originates from the dataset lookup function (e.g.,
getDatasetById) or has a specific error.name/marker; leave other errors to fall
through so they return the appropriate server error status instead of
reply.code(400).send({ error: "Invalid datasetId" }).
2 participants
Summary
insert_row,list_rows,get_row,update_row,delete_row) and 2 web tools (search_web,fetch_page) powered by the TinyFish Search and Fetch APIs.Architecture
Key decisions
"Name"vs`Name`vsName); the tool normalizes all of them.anyApifor Convex client — the backend uses an untyped proxy instead of importing the frontend's generated types, which don't resolve in Docker containers.New files
backend/src/mastra/agents/populate.tsbackend/src/mastra/tools/dataset-tools.tsbackend/src/mastra/tools/web-tools.tsbackend/src/mastra/workflows/populate.tsbackend/src/pipeline/populate.tsfrontend/lib/backend.tspopulate()functionSetup
Add
TINYFISH_API_KEY=...to the root.envfile (get one at https://agent.tinyfish.ai/api-keys), thendocker compose up -dto recreate containers.Test plan
TINYFISH_API_KEYto root.envand runmake devmake convex-pushto deploy new Convex functions (get,remove)[search_web],[fetch_page], and[insert_row]entriesTINYFISH_API_KEYand confirm agent falls back to synthetic data🤖 Generated with Claude Code