feat: Simple / Aggregation / Shell query modes

src/main/ipc/channels.ts

@@ -33,6 +33,7 @@ export const Channels = {
   UsersDrop: 'users:drop',
 
   QueryFind: 'query:find',
+  QueryAggregate: 'query:aggregate',
   QueryCount: 'query:count',
   QueryReplaceOne: 'query:replaceOne',
   QueryInsertOne: 'query:insertOne',

src/main/ipc/router.ts

@@ -2,6 +2,7 @@ import { ipcMain, type IpcMainInvokeEvent } from 'electron'
 import log from 'electron-log/main'
 import type { ZodType } from 'zod'
 import {
+  AggregateRequestSchema,
   CollectionRefSchema,
   ConnectionIdSchema,
   ConnectionInputSchema,
@@ -191,6 +192,11 @@ export function registerIpcHandlers(services: Services): void {
     withResult(FindRequestSchema, (request) => queries.find(request))
   )
 
+  ipcMain.handle(
+    Channels.QueryAggregate,
+    withResult(AggregateRequestSchema, (request) => queries.aggregate(request))
+  )
+
   ipcMain.handle(
     Channels.QueryCount,
     withResult(CountRequestSchema, (request) => queries.count(request))

src/main/services/QueryService.ts

@@ -1,6 +1,8 @@
 import type { Document, Filter, Sort } from 'mongodb'
 import { EJSON } from 'bson'
 import type {
+  AggregateRequest,
+  AggregateResponse,
   DeleteManyRequest,
   DeleteManyResponse,
   DeleteOneRequest,
@@ -42,6 +44,19 @@ export class QueryService {
     return { documents, tookMs }
   }
 
+  async aggregate(req: AggregateRequest): Promise<AggregateResponse> {
+    const client = this.connections.getClient(req.connectionId)
+    const coll = client.db(req.db).collection(req.coll)
+
+    const pipeline = parsePipeline(req.pipeline)
+
+    const startedAt = Date.now()
+    const docs = await coll.aggregate(pipeline).toArray()
+    const tookMs = Date.now() - startedAt
+
+    return { documents: docs.map(toEnvelope), tookMs }
+  }
+
   async count(req: {
     connectionId: string
     db: string
@@ -155,3 +170,20 @@ function parseDocument(canonical: string): Document {
   }
   return parsed as Document
 }
+
+function parsePipeline(canonical: string): Document[] {
+  const parsed = EJSON.parse(canonical, { relaxed: false })
+  if (!Array.isArray(parsed)) {
+    const e = new Error('Pipeline must be an array of stage objects')
+    e.name = 'ValidationError'
+    throw e
+  }
+  for (const stage of parsed) {
+    if (typeof stage !== 'object' || stage === null || Array.isArray(stage)) {
+      const e = new Error('Each pipeline stage must be an object')
+      e.name = 'ValidationError'
+      throw e
+    }
+  }
+  return parsed as Document[]
+}

src/preload/index.ts

@@ -2,6 +2,8 @@ import { contextBridge, ipcRenderer } from 'electron'
 import type { Api } from '@shared/api'
 import type { Result } from '@shared/result'
 import type {
+  AggregateRequest,
+  AggregateResponse,
   CollectionInfo,
   CollectionStats,
   ConnectionConfig,
@@ -70,6 +72,7 @@ const api: Api = {
   },
   query: {
     find: (request: FindRequest) => invoke<FindResponse>('query:find', request),
+    aggregate: (request: AggregateRequest) => invoke<AggregateResponse>('query:aggregate', request),
     count: (request: CountRequest) => invoke<CountResponse>('query:count', request),
     replaceOne: (request: ReplaceOneRequest) =>
       invoke<ReplaceOneResponse>('query:replaceOne', request),