mirror of
https://github.com/t8y2/dbx.git
synced 2026-10-19 16:16:45 +00:00
feat(mongodb): support createIndex shell command (#2187)
This commit is contained in:
@@ -375,6 +375,7 @@ export const documentFindDocuments = forward("documentFindDocuments");
|
||||
export const mongoFindDocuments = forward("mongoFindDocuments");
|
||||
export const mongoServerVersion = forward("mongoServerVersion");
|
||||
export const mongoAggregateDocuments = forward("mongoAggregateDocuments");
|
||||
export const mongoCreateIndex = forward("mongoCreateIndex");
|
||||
export const mongoInsertDocument = forward("mongoInsertDocument");
|
||||
export const mongoInsertDocuments = forward("mongoInsertDocuments");
|
||||
export const mongoUpdateDocument = forward("mongoUpdateDocument");
|
||||
|
||||
@@ -1756,6 +1756,10 @@ export async function mongoAggregateDocuments(connectionId: string, database: st
|
||||
return post("/api/mongo/aggregate-documents", { connectionId, database, collection, pipelineJson, maxRows, executionId });
|
||||
}
|
||||
|
||||
export async function mongoCreateIndex(connectionId: string, database: string, collection: string, keysJson: string, optionsJson?: string): Promise<{ name: string }> {
|
||||
return post("/api/mongo/create-index", { connectionId, database, collection, keysJson, optionsJson });
|
||||
}
|
||||
|
||||
export async function mongoInsertDocument(connectionId: string, database: string, collection: string, docJson: string): Promise<string> {
|
||||
return post("/api/mongo/insert-document", { connectionId, database, collection, docJson });
|
||||
}
|
||||
|
||||
@@ -31,7 +31,11 @@ export interface MongoVersionCommand {
|
||||
kind: "version";
|
||||
}
|
||||
|
||||
export type MongoWriteCommand = { kind: "insert"; collection: string; docsJson: string } | { kind: "update"; collection: string; filter: string; update: string; many: boolean } | { kind: "delete"; collection: string; filter: string; many: boolean };
|
||||
export type MongoWriteCommand =
|
||||
| { kind: "insert"; collection: string; docsJson: string }
|
||||
| { kind: "update"; collection: string; filter: string; update: string; many: boolean }
|
||||
| { kind: "delete"; collection: string; filter: string; many: boolean }
|
||||
| { kind: "createIndex"; collection: string; keys: string; options?: string };
|
||||
|
||||
export interface MongoAggregateSafetyOptions {
|
||||
allowWrites?: boolean;
|
||||
@@ -201,6 +205,21 @@ export function parseMongoWriteCommand(input: string): MongoWriteCommand | null
|
||||
return { kind: "delete", collection: target.collection, filter, many: method === "deleteMany" };
|
||||
}
|
||||
|
||||
const createIndex = parseCollectionMethodTarget(source, "createIndex");
|
||||
if (createIndex) {
|
||||
const args = parseMethodArgs(source, createIndex.methodCallIndex);
|
||||
if (!args || args.length < 1 || args.length > 2) return null;
|
||||
const keys = normalizeJsonArgument(args[0]);
|
||||
if (!keys) return null;
|
||||
let options: string | undefined;
|
||||
if (args[1]?.trim()) {
|
||||
const parsedOptions = normalizeJsonArgument(args[1]);
|
||||
if (!parsedOptions) return null;
|
||||
options = parsedOptions;
|
||||
}
|
||||
return { kind: "createIndex", collection: createIndex.collection, keys, ...(options ? { options } : {}) };
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -282,6 +301,15 @@ export function mongoWriteToQueryResult(affectedRows: number, executionTimeMs: n
|
||||
};
|
||||
}
|
||||
|
||||
export function mongoCreateIndexToQueryResult(name: string, executionTimeMs: number): QueryResult {
|
||||
return {
|
||||
columns: ["name"],
|
||||
rows: [[name]],
|
||||
affected_rows: 1,
|
||||
execution_time_ms: Math.max(0, Math.round(executionTimeMs)),
|
||||
};
|
||||
}
|
||||
|
||||
export function mongoUseToQueryResult(database: string, executionTimeMs: number): QueryResult {
|
||||
return {
|
||||
columns: ["message"],
|
||||
|
||||
@@ -1486,6 +1486,10 @@ export async function mongoAggregateDocuments(connectionId: string, database: st
|
||||
return invoke("mongo_aggregate_documents", { connectionId, database, collection, pipelineJson, maxRows, executionId });
|
||||
}
|
||||
|
||||
export async function mongoCreateIndex(connectionId: string, database: string, collection: string, keysJson: string, optionsJson?: string): Promise<{ name: string }> {
|
||||
return invoke("mongo_create_index", { connectionId, database, collection, keysJson, optionsJson });
|
||||
}
|
||||
|
||||
export async function mongoInsertDocument(connectionId: string, database: string, collection: string, docJson: string): Promise<string> {
|
||||
return invoke("mongo_insert_document", { connectionId, database, collection, docJson });
|
||||
}
|
||||
|
||||
@@ -12,6 +12,7 @@ import { restoreOpenTabsState, serializeOpenTabs } from "@/lib/openTabsPersisten
|
||||
import {
|
||||
evaluateMongoAggregateSafety,
|
||||
mongoCountToQueryResult,
|
||||
mongoCreateIndexToQueryResult,
|
||||
mongoDocumentsToQueryResult,
|
||||
mongoIndexesToQueryResult,
|
||||
mongoUseToQueryResult,
|
||||
@@ -1756,6 +1757,29 @@ export const useQueryStore = defineStore("query", () => {
|
||||
} else if (mongoWrite.kind === "update") {
|
||||
const result = await api.mongoUpdateDocuments(tab.connectionId, tab.database, mongoWrite.collection, mongoWrite.filter, mongoWrite.update, mongoWrite.many);
|
||||
affectedRows = result.affected_rows;
|
||||
} else if (mongoWrite.kind === "createIndex") {
|
||||
const result = await api.mongoCreateIndex(tab.connectionId, tab.database, mongoWrite.collection, mongoWrite.keys, mongoWrite.options);
|
||||
console.info("[DBX][executeTabSql:mongo-write:done]", {
|
||||
traceId,
|
||||
indexName: result.name,
|
||||
elapsed: elapsed(),
|
||||
});
|
||||
const current = tabs.value.find((t) => t.id === id);
|
||||
if (current?.executionId === executionId) {
|
||||
current.results = undefined;
|
||||
current.activeResultIndex = undefined;
|
||||
current.result = markQueryResultRowsRaw(mongoCreateIndexToQueryResult(result.name, performance.now() - startedAt));
|
||||
touchResult(current);
|
||||
current.queryAnalysis = undefined;
|
||||
current.querySourceColumns = undefined;
|
||||
current.queryEditabilityReason = undefined;
|
||||
current.mongoEditTarget = undefined;
|
||||
current.tableMeta = undefined;
|
||||
current.resultBaseSql = options?.resultBaseSql ?? sql;
|
||||
current.resultSortedSql = options?.resultSortedSql;
|
||||
syncDisplayedResultRun(current, options?.resultBaseSql ?? sql);
|
||||
}
|
||||
return;
|
||||
} else {
|
||||
const result = await api.mongoDeleteDocuments(tab.connectionId, tab.database, mongoWrite.collection, mongoWrite.filter, mongoWrite.many);
|
||||
affectedRows = result.affected_rows;
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
use mongodb::{
|
||||
bson::{doc, oid::ObjectId, Bson, DateTime, Document},
|
||||
options::ClientOptions,
|
||||
options::{ClientOptions, IndexOptions},
|
||||
Client, IndexModel,
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
@@ -282,6 +282,36 @@ pub async fn aggregate_documents(
|
||||
Ok(MongoDocumentResult { documents, total })
|
||||
}
|
||||
|
||||
pub async fn create_index(
|
||||
client: &Client,
|
||||
database: &str,
|
||||
collection: &str,
|
||||
keys_json: &str,
|
||||
options_json: Option<&str>,
|
||||
) -> Result<String, String> {
|
||||
let keys_value: serde_json::Value =
|
||||
serde_json::from_str(keys_json).map_err(|e| format!("Invalid index keys JSON: {e}"))?;
|
||||
let keys = json_object_to_document(&keys_value).map_err(|e| format!("Invalid index keys: {e}"))?;
|
||||
if keys.is_empty() {
|
||||
return Err("Index keys are required".to_string());
|
||||
}
|
||||
|
||||
let options = match options_json.map(str::trim).filter(|json| !json.is_empty()) {
|
||||
Some(json) => {
|
||||
let value: serde_json::Value =
|
||||
serde_json::from_str(json).map_err(|e| format!("Invalid index options JSON: {e}"))?;
|
||||
let doc = json_object_to_document(&value).map_err(|e| format!("Invalid index options: {e}"))?;
|
||||
Some(mongodb::bson::from_document::<IndexOptions>(doc).map_err(|e| format!("Invalid index options: {e}"))?)
|
||||
}
|
||||
None => None,
|
||||
};
|
||||
|
||||
let col = client.database(database).collection::<Document>(collection);
|
||||
let result =
|
||||
col.create_index(IndexModel::builder().keys(keys).options(options).build()).await.map_err(|e| e.to_string())?;
|
||||
Ok(result.index_name)
|
||||
}
|
||||
|
||||
pub async fn insert_document(
|
||||
client: &Client,
|
||||
database: &str,
|
||||
@@ -689,7 +719,6 @@ fn expand_object_id_string_array(items: &[serde_json::Value]) -> Bson {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use mongodb::options::IndexOptions;
|
||||
|
||||
#[test]
|
||||
fn multi_seed_uri_removes_direct_connection_true_before_driver_parse() {
|
||||
|
||||
@@ -223,6 +223,25 @@ pub async fn mongo_aggregate_documents_core(
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn mongo_create_index_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
database: &str,
|
||||
collection: &str,
|
||||
keys_json: &str,
|
||||
options_json: Option<&str>,
|
||||
) -> Result<String, String> {
|
||||
ensure_document_pool(state, connection_id).await?;
|
||||
let connections = state.connections.read().await;
|
||||
match connections.get(connection_id).ok_or("Not found")? {
|
||||
PoolKind::MongoDb(client) => {
|
||||
mongo_driver::create_index(client, database, collection, keys_json, options_json).await
|
||||
}
|
||||
PoolKind::Agent(_) => Err("MongoDB legacy agent does not support createIndex".to_string()),
|
||||
_ => Err("Not a MongoDB connection".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn mongo_insert_document_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
|
||||
@@ -448,6 +448,7 @@ async fn main() {
|
||||
.route("/mongo/find-documents", post(routes::mongo::find_documents))
|
||||
.route("/mongo/server-version", post(routes::mongo::server_version))
|
||||
.route("/mongo/aggregate-documents", post(routes::mongo::aggregate_documents))
|
||||
.route("/mongo/create-index", post(routes::mongo::create_index))
|
||||
.route("/mongo/insert-document", post(routes::mongo::insert_document))
|
||||
.route("/mongo/insert-documents", post(routes::mongo::insert_documents))
|
||||
.route("/mongo/update-document", post(routes::mongo::update_document))
|
||||
|
||||
@@ -97,6 +97,16 @@ pub struct MongoAggregateRequest {
|
||||
pub execution_id: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct MongoCreateIndexRequest {
|
||||
pub connection_id: String,
|
||||
pub database: String,
|
||||
pub collection: String,
|
||||
pub keys_json: String,
|
||||
pub options_json: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct MongoInsertRequest {
|
||||
@@ -286,6 +296,24 @@ pub async fn aggregate_documents(
|
||||
Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?))
|
||||
}
|
||||
|
||||
pub async fn create_index(
|
||||
State(state): State<Arc<WebState>>,
|
||||
Json(req): Json<MongoCreateIndexRequest>,
|
||||
) -> Result<Json<serde_json::Value>, AppError> {
|
||||
ensure_writable(&state.app, &req.connection_id, "Create index").await?;
|
||||
let name = dbx_core::mongo_ops::mongo_create_index_core(
|
||||
&state.app,
|
||||
&req.connection_id,
|
||||
&req.database,
|
||||
&req.collection,
|
||||
&req.keys_json,
|
||||
req.options_json.as_deref(),
|
||||
)
|
||||
.await
|
||||
.map_err(AppError)?;
|
||||
Ok(Json(serde_json::json!({ "name": name })))
|
||||
}
|
||||
|
||||
pub async fn insert_document(
|
||||
State(state): State<Arc<WebState>>,
|
||||
Json(req): Json<MongoInsertRequest>,
|
||||
|
||||
@@ -105,6 +105,15 @@ test("parseMongoWriteCommand accepts unquoted insert and update commands", () =>
|
||||
});
|
||||
});
|
||||
|
||||
test("parseMongoWriteCommand parses createIndex with optional options", () => {
|
||||
assert.deepEqual(parseMongoWriteCommand("db.users.createIndex({email: 1}, {unique: true, name: 'users_email_unique'})"), {
|
||||
kind: "createIndex",
|
||||
collection: "users",
|
||||
keys: '{"email": 1}',
|
||||
options: '{"unique": true, "name": "users_email_unique"}',
|
||||
});
|
||||
});
|
||||
|
||||
test("parseMongoCountDocumentsCommand parses db collection countDocuments", () => {
|
||||
assert.deepEqual(parseMongoCountDocumentsCommand("db.products.countDocuments({})"), {
|
||||
collection: "products",
|
||||
|
||||
@@ -1842,6 +1842,59 @@ test("mongo aggregate execution uses editor page size when pagination plan has n
|
||||
}
|
||||
});
|
||||
|
||||
test("mongo createIndex execution uses the dedicated create-index endpoint", async () => {
|
||||
const restoreStorage = installMemoryStorage();
|
||||
setActivePinia(createPinia());
|
||||
const connectionStore = useConnectionStore();
|
||||
const store = useQueryStore();
|
||||
const originalFetch = globalThis.fetch;
|
||||
let createIndexBody: any;
|
||||
|
||||
connectionStore.addEphemeralConnection({
|
||||
...conn("mongo-1"),
|
||||
db_type: "mongodb",
|
||||
port: 27017,
|
||||
});
|
||||
|
||||
globalThis.fetch = withConnectionHealthMock(async (input, init) => {
|
||||
const url = String(input);
|
||||
if (url === "/api/query/prepare-pagination-plan") {
|
||||
const body = JSON.parse(String(init?.body ?? "{}"));
|
||||
return new Response(JSON.stringify({ sqlToExecute: body.options.sql, useAgentResultSession: false }), {
|
||||
status: 200,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
if (url === "/api/mongo/create-index") {
|
||||
createIndexBody = JSON.parse(String(init?.body ?? "{}"));
|
||||
return new Response(JSON.stringify({ name: "users_email_unique" }), {
|
||||
status: 200,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
return new Response("unexpected request", { status: 500 });
|
||||
});
|
||||
|
||||
try {
|
||||
const tabId = store.createTab("mongo-1", "accounting", "Query", "query", "");
|
||||
await store.executeTabSql(tabId, 'db.users.createIndex({email: 1}, {unique: true, name: "users_email_unique"})');
|
||||
const tab = store.tabs.find((item) => item.id === tabId);
|
||||
|
||||
assert.deepEqual(createIndexBody, {
|
||||
connectionId: "mongo-1",
|
||||
database: "accounting",
|
||||
collection: "users",
|
||||
keysJson: '{"email": 1}',
|
||||
optionsJson: '{"unique": true, "name": "users_email_unique"}',
|
||||
});
|
||||
assert.deepEqual(tab?.result?.columns, ["name"]);
|
||||
assert.deepEqual(tab?.result?.rows, [["users_email_unique"]]);
|
||||
} finally {
|
||||
globalThis.fetch = originalFetch;
|
||||
restoreStorage();
|
||||
}
|
||||
});
|
||||
|
||||
test("table data export fetches every filtered page", async () => {
|
||||
const restoreStorage = installMemoryStorage();
|
||||
setActivePinia(createPinia());
|
||||
|
||||
@@ -655,10 +655,17 @@ export async function executeQuery(config: ConnectionConfig, sql: string, option
|
||||
if (write) {
|
||||
const safety = evaluateMongoWriteSafety(write, sqlSafetyFromEnv());
|
||||
if (!safety.allowed) throw new Error(safety.reason);
|
||||
const affected = await withTimeout(executeMongoWrite(config, write), resolveTimeoutMs(options));
|
||||
return { columns: [], rows: [], row_count: affected };
|
||||
const result = await withTimeout(executeMongoWrite(config, write), resolveTimeoutMs(options));
|
||||
if (write.kind === "createIndex") {
|
||||
return {
|
||||
columns: ["name"],
|
||||
rows: [{ name: result.indexName ?? "" }],
|
||||
row_count: 1,
|
||||
};
|
||||
}
|
||||
return { columns: [], rows: [], row_count: result.affectedRows };
|
||||
}
|
||||
throw new Error("Use MongoDB shell-style commands, for example: db.projects.find({}).limit(100), db.version(), db.projects.countDocuments({}), db.projects.getIndexes(), db.projects.insertOne({...}), db.projects.updateOne({...}, {$set: {...}}), or db.projects.deleteOne({...})");
|
||||
throw new Error("Use MongoDB shell-style commands, for example: db.projects.find({}).limit(100), db.version(), db.projects.countDocuments({}), db.projects.getIndexes(), db.projects.createIndex({...}), db.projects.insertOne({...}), db.projects.updateOne({...}, {$set: {...}}), or db.projects.deleteOne({...})");
|
||||
}
|
||||
if (isDirectQueryType(config.db_type)) {
|
||||
return query(config, sql, undefined, options);
|
||||
@@ -873,7 +880,10 @@ async function mongoServerVersion(config: ConnectionConfig): Promise<string> {
|
||||
});
|
||||
}
|
||||
|
||||
async function executeMongoWrite(config: ConnectionConfig, command: MongoWriteCommand): Promise<number> {
|
||||
async function executeMongoWrite(
|
||||
config: ConnectionConfig,
|
||||
command: MongoWriteCommand,
|
||||
): Promise<{ affectedRows: number; indexName?: string }> {
|
||||
if (command.kind === "insert") {
|
||||
const result = await bridgeDataRequest<{ affected_rows: number }>("/data/mongo/insert-documents", {
|
||||
connection_name: config.name,
|
||||
@@ -881,7 +891,7 @@ async function executeMongoWrite(config: ConnectionConfig, command: MongoWriteCo
|
||||
collection: command.collection,
|
||||
docs_json: command.docsJson,
|
||||
});
|
||||
return result.affected_rows;
|
||||
return { affectedRows: result.affected_rows };
|
||||
}
|
||||
if (command.kind === "update") {
|
||||
const result = await bridgeDataRequest<{ affected_rows: number }>("/data/mongo/update-documents", {
|
||||
@@ -892,7 +902,17 @@ async function executeMongoWrite(config: ConnectionConfig, command: MongoWriteCo
|
||||
update_json: command.update,
|
||||
many: command.many,
|
||||
});
|
||||
return result.affected_rows;
|
||||
return { affectedRows: result.affected_rows };
|
||||
}
|
||||
if (command.kind === "createIndex") {
|
||||
const result = await bridgeDataRequest<{ name: string }>("/data/mongo/create-index", {
|
||||
connection_name: config.name,
|
||||
database: config.database || "",
|
||||
collection: command.collection,
|
||||
keys_json: command.keys,
|
||||
options_json: command.options,
|
||||
});
|
||||
return { affectedRows: 1, indexName: result.name };
|
||||
}
|
||||
const result = await bridgeDataRequest<{ affected_rows: number }>("/data/mongo/delete-documents", {
|
||||
connection_name: config.name,
|
||||
@@ -901,7 +921,7 @@ async function executeMongoWrite(config: ConnectionConfig, command: MongoWriteCo
|
||||
filter_json: command.filter,
|
||||
many: command.many,
|
||||
});
|
||||
return result.affected_rows;
|
||||
return { affectedRows: result.affected_rows };
|
||||
}
|
||||
|
||||
async function mongoAggregateDocuments(config: ConnectionConfig, collection: string, pipelineJson: string, maxRows: number): Promise<MongoDocumentResult> {
|
||||
@@ -984,7 +1004,11 @@ interface MongoGetIndexesCommand {
|
||||
collection: string;
|
||||
}
|
||||
|
||||
export type MongoWriteCommand = { kind: "insert"; collection: string; docsJson: string } | { kind: "update"; collection: string; filter: string; update: string; many: boolean } | { kind: "delete"; collection: string; filter: string; many: boolean };
|
||||
export type MongoWriteCommand =
|
||||
| { kind: "insert"; collection: string; docsJson: string }
|
||||
| { kind: "update"; collection: string; filter: string; update: string; many: boolean }
|
||||
| { kind: "delete"; collection: string; filter: string; many: boolean }
|
||||
| { kind: "createIndex"; collection: string; keys: string; options?: string };
|
||||
|
||||
export function parseMongoFindCommand(input: string): MongoFindCommand | null {
|
||||
const source = input.trim().replace(/;$/, "").trim();
|
||||
@@ -1111,6 +1135,21 @@ export function parseMongoWriteCommand(input: string): MongoWriteCommand | null
|
||||
return { kind: "delete", collection: target.collection, filter, many: method === "deleteMany" };
|
||||
}
|
||||
|
||||
const createIndex = parseCollectionMethodTarget(source, "createIndex");
|
||||
if (createIndex) {
|
||||
const args = parseMethodArgs(source, createIndex.methodCallIndex);
|
||||
if (!args || args.length < 1 || args.length > 2) return null;
|
||||
const keys = normalizeJsonArgument(args[0]);
|
||||
if (!keys) return null;
|
||||
let options: string | undefined;
|
||||
if (args[1]?.trim()) {
|
||||
const parsedOptions = normalizeJsonArgument(args[1]);
|
||||
if (!parsedOptions) return null;
|
||||
options = parsedOptions;
|
||||
}
|
||||
return { kind: "createIndex", collection: createIndex.collection, keys, ...(options ? { options } : {}) };
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
|
||||
@@ -210,10 +210,13 @@ export async function executeQuery(config: ConnectionConfig, sql: string, option
|
||||
if (write) {
|
||||
const safety = evaluateMongoWriteSafety(write, sqlSafetyFromEnv());
|
||||
if (!safety.allowed) throw new Error(safety.reason);
|
||||
const affected = await executeMongoWrite(config, write);
|
||||
return { columns: [], rows: [], row_count: affected };
|
||||
const result = await executeMongoWrite(config, write);
|
||||
if (write.kind === "createIndex") {
|
||||
return { columns: ["name"], rows: [{ name: result.indexName ?? "" }], row_count: 1 };
|
||||
}
|
||||
return { columns: [], rows: [], row_count: result.affectedRows };
|
||||
}
|
||||
throw new Error("Use MongoDB shell-style commands, for example: db.projects.find({}).limit(100), db.version(), db.projects.countDocuments({}), db.projects.getIndexes(), db.projects.insertOne({...}), db.projects.updateOne({...}, {$set: {...}}), or db.projects.deleteOne({...})");
|
||||
throw new Error("Use MongoDB shell-style commands, for example: db.projects.find({}).limit(100), db.version(), db.projects.countDocuments({}), db.projects.getIndexes(), db.projects.createIndex({...}), db.projects.insertOne({...}), db.projects.updateOne({...}, {$set: {...}}), or db.projects.deleteOne({...})");
|
||||
}
|
||||
const res = await apiFetch("/api/query/execute", {
|
||||
method: "POST",
|
||||
@@ -252,7 +255,10 @@ export async function executeRedisCommand(config: ConnectionConfig, db: number,
|
||||
return (await res.json()) as RedisCommandResult;
|
||||
}
|
||||
|
||||
async function executeMongoWrite(config: ConnectionConfig, command: MongoWriteCommand): Promise<number> {
|
||||
async function executeMongoWrite(
|
||||
config: ConnectionConfig,
|
||||
command: MongoWriteCommand,
|
||||
): Promise<{ affectedRows: number; indexName?: string }> {
|
||||
if (command.kind === "insert") {
|
||||
const res = await apiFetch("/api/mongo/insert-documents", {
|
||||
method: "POST",
|
||||
@@ -264,7 +270,7 @@ async function executeMongoWrite(config: ConnectionConfig, command: MongoWriteCo
|
||||
}),
|
||||
});
|
||||
const result = (await res.json()) as { affected_rows: number };
|
||||
return result.affected_rows;
|
||||
return { affectedRows: result.affected_rows };
|
||||
}
|
||||
if (command.kind === "update") {
|
||||
const res = await apiFetch("/api/mongo/update-documents", {
|
||||
@@ -279,7 +285,21 @@ async function executeMongoWrite(config: ConnectionConfig, command: MongoWriteCo
|
||||
}),
|
||||
});
|
||||
const result = (await res.json()) as { affected_rows: number };
|
||||
return result.affected_rows;
|
||||
return { affectedRows: result.affected_rows };
|
||||
}
|
||||
if (command.kind === "createIndex") {
|
||||
const res = await apiFetch("/api/mongo/create-index", {
|
||||
method: "POST",
|
||||
body: JSON.stringify({
|
||||
connectionId: config.id,
|
||||
database: config.database || "",
|
||||
collection: command.collection,
|
||||
keysJson: command.keys,
|
||||
optionsJson: command.options,
|
||||
}),
|
||||
});
|
||||
const result = (await res.json()) as { name: string };
|
||||
return { affectedRows: 1, indexName: result.name };
|
||||
}
|
||||
const res = await apiFetch("/api/mongo/delete-documents", {
|
||||
method: "POST",
|
||||
@@ -292,5 +312,5 @@ async function executeMongoWrite(config: ConnectionConfig, command: MongoWriteCo
|
||||
}),
|
||||
});
|
||||
const result = (await res.json()) as { affected_rows: number };
|
||||
return result.affected_rows;
|
||||
return { affectedRows: result.affected_rows };
|
||||
}
|
||||
|
||||
@@ -135,6 +135,12 @@ test("parseMongoWriteCommand accepts supported write commands", () => {
|
||||
filter: '{"stale":true}',
|
||||
many: true,
|
||||
});
|
||||
assert.deepEqual(parseMongoWriteCommand('db.projects.createIndex({"email":1},{"unique":true,"name":"projects_email_unique"})'), {
|
||||
kind: "createIndex",
|
||||
collection: "projects",
|
||||
keys: '{"email":1}',
|
||||
options: '{"unique":true,"name":"projects_email_unique"}',
|
||||
});
|
||||
});
|
||||
|
||||
test("mongodb executeQuery blocks writes when writes are explicitly disabled", async () => {
|
||||
@@ -162,6 +168,31 @@ test("mongodb executeQuery blocks writes when writes are explicitly disabled", a
|
||||
else process.env.DBX_MCP_ALLOW_WRITES = oldAllowWrites;
|
||||
});
|
||||
|
||||
test("mongodb executeQuery treats createIndex as a write when writes are explicitly disabled", async () => {
|
||||
const oldAllowWrites = process.env.DBX_MCP_ALLOW_WRITES;
|
||||
process.env.DBX_MCP_ALLOW_WRITES = "0";
|
||||
await assert.rejects(
|
||||
executeQuery(
|
||||
{
|
||||
id: "mongo",
|
||||
name: "mongo",
|
||||
db_type: "mongodb",
|
||||
host: "127.0.0.1",
|
||||
port: 27017,
|
||||
username: "",
|
||||
password: "",
|
||||
database: "app",
|
||||
ssh_enabled: false,
|
||||
ssl: false,
|
||||
},
|
||||
'db.projects.createIndex({"email":1})',
|
||||
),
|
||||
/read-only/i,
|
||||
);
|
||||
if (oldAllowWrites === undefined) delete process.env.DBX_MCP_ALLOW_WRITES;
|
||||
else process.env.DBX_MCP_ALLOW_WRITES = oldAllowWrites;
|
||||
});
|
||||
|
||||
test("mongoDocumentsToQueryResult turns documents into rows", () => {
|
||||
assert.deepEqual(
|
||||
mongoDocumentsToQueryResult(
|
||||
|
||||
@@ -70,6 +70,15 @@ struct MongoAggregateDocumentsRequest {
|
||||
max_rows: Option<usize>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct MongoCreateIndexRequest {
|
||||
connection_name: String,
|
||||
database: Option<String>,
|
||||
collection: String,
|
||||
keys_json: String,
|
||||
options_json: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct MongoInsertDocumentsRequest {
|
||||
connection_name: String,
|
||||
@@ -168,6 +177,8 @@ pub fn start(app_handle: AppHandle, state: Arc<AppState>) {
|
||||
handle_mongo_server_version_data(&st, body, &mut stream).await;
|
||||
} else if first_line.starts_with("POST /data/mongo/aggregate-documents") {
|
||||
handle_mongo_aggregate_documents_data(&st, body, &mut stream).await;
|
||||
} else if first_line.starts_with("POST /data/mongo/create-index") {
|
||||
handle_mongo_create_index_data(&st, body, &mut stream).await;
|
||||
} else if first_line.starts_with("POST /data/mongo/insert-documents") {
|
||||
handle_mongo_insert_documents_data(&st, body, &mut stream).await;
|
||||
} else if first_line.starts_with("POST /data/mongo/update-documents") {
|
||||
@@ -476,6 +487,38 @@ async fn handle_mongo_aggregate_documents_data(state: &Arc<AppState>, body: &str
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_mongo_create_index_data(state: &Arc<AppState>, body: &str, stream: &mut tokio::net::TcpStream) {
|
||||
let req: MongoCreateIndexRequest = match serde_json::from_str(body) {
|
||||
Ok(r) => r,
|
||||
Err(_) => {
|
||||
respond_error(stream, "400 Bad Request", "Invalid JSON").await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
let Some((pool_key, database, connection_id)) =
|
||||
resolve_mongo_pool_key(state, &req.connection_name, req.database, stream).await
|
||||
else {
|
||||
return;
|
||||
};
|
||||
if let Err(e) = ensure_connection_writable(state, &connection_id, "Create index").await {
|
||||
respond_error(stream, "403 Forbidden", &e).await;
|
||||
return;
|
||||
}
|
||||
match dbx_core::mongo_ops::mongo_create_index_core(
|
||||
state,
|
||||
&pool_key,
|
||||
&database,
|
||||
&req.collection,
|
||||
&req.keys_json,
|
||||
req.options_json.as_deref(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(name) => respond_json(stream, &serde_json::json!({ "name": name })).await,
|
||||
Err(e) => respond_error(stream, "500 Internal Server Error", &e).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_mongo_insert_documents_data(state: &Arc<AppState>, body: &str, stream: &mut tokio::net::TcpStream) {
|
||||
let req: MongoInsertDocumentsRequest = match serde_json::from_str(body) {
|
||||
Ok(r) => r,
|
||||
|
||||
@@ -175,6 +175,28 @@ pub async fn mongo_aggregate_documents(
|
||||
.await
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn mongo_create_index(
|
||||
state: State<'_, Arc<AppState>>,
|
||||
connection_id: String,
|
||||
database: String,
|
||||
collection: String,
|
||||
keys_json: String,
|
||||
options_json: Option<String>,
|
||||
) -> Result<serde_json::Value, String> {
|
||||
ensure_connection_writable(&state, &connection_id, "Create index").await?;
|
||||
let name = dbx_core::mongo_ops::mongo_create_index_core(
|
||||
&state,
|
||||
&connection_id,
|
||||
&database,
|
||||
&collection,
|
||||
&keys_json,
|
||||
options_json.as_deref(),
|
||||
)
|
||||
.await?;
|
||||
Ok(serde_json::json!({ "name": name }))
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn mongo_insert_document(
|
||||
state: State<'_, Arc<AppState>>,
|
||||
|
||||
@@ -850,6 +850,7 @@ pub fn run() {
|
||||
commands::mongo_cmd::mongo_find_documents,
|
||||
commands::mongo_cmd::mongo_server_version,
|
||||
commands::mongo_cmd::mongo_aggregate_documents,
|
||||
commands::mongo_cmd::mongo_create_index,
|
||||
commands::mongo_cmd::mongo_insert_document,
|
||||
commands::mongo_cmd::mongo_insert_documents,
|
||||
commands::mongo_cmd::mongo_update_document,
|
||||
|
||||
Reference in New Issue
Block a user