feat(cli): add runKtxMongoQuery runner

packages/cli/src/mongo-query.ts

@@ -0,0 +1,110 @@
+import { runMongoQuery } from './context/mcp/mongo-query-runner.js';
+import { resolveConfiguredConnection } from './context/connections/resolve-connection.js';
+import { loadKtxProject, type KtxLocalProject } from './context/project/project.js';
+import type { KtxCliIo } from './cli-runtime.js';
+import { type KtxOutputMode, resolveOutputMode } from './io/mode.js';
+import { type KtxResultTable, printResultTable } from './io/result-table.js';
+import { createKtxCliScanConnector } from './local-scan-connectors.js';
+import { profileMark } from './startup-profile.js';
+import { isDemoConnection } from './telemetry/demo-detect.js';
+import { emitTelemetryEvent, reportException } from './telemetry/index.js';
+import { collectTelemetryRedactionSecrets } from './telemetry/redaction-secrets.js';
+import { scrubErrorClass } from './telemetry/scrubber.js';
+
+profileMark('module:mongo-query');
+
+export type KtxMongoQueryArgs = {
+  projectDir: string;
+  connectionId: string;
+  collection: string;
+  database?: string;
+  pipeline: Record<string, unknown>[];
+  limit: number;
+  output?: KtxOutputMode;
+  json?: boolean;
+  cliVersion: string;
+};
+
+export interface KtxMongoQueryCliDeps {
+  loadProject?: typeof loadKtxProject;
+  createScanConnector?: typeof createKtxCliScanConnector;
+}
+
+export async function runKtxMongoQuery(
+  args: KtxMongoQueryArgs,
+  io: KtxCliIo = process,
+  deps: KtxMongoQueryCliDeps = {},
+): Promise<number> {
+  const startedAt = performance.now();
+  let driver = 'unknown';
+  let demoConnection = false;
+  let project: KtxLocalProject | undefined;
+  try {
+    project = await (deps.loadProject ?? loadKtxProject)({ projectDir: args.projectDir });
+    const connection = resolveConfiguredConnection(project.config, args.connectionId);
+    driver = String(connection?.driver ?? 'unknown').toLowerCase();
+    demoConnection = isDemoConnection(args.connectionId, connection);
+
+    const createScanConnector = deps.createScanConnector ?? createKtxCliScanConnector;
+    const result = await runMongoQuery((connectionId) => createScanConnector(project!, connectionId), {
+      connectionId: args.connectionId,
+      collection: args.collection,
+      database: args.database,
+      pipeline: args.pipeline,
+      limit: args.limit,
+    });
+
+    const output: KtxResultTable = {
+      connectionId: args.connectionId,
+      headers: result.headers,
+      rows: result.rows,
+      rowCount: result.rowCount,
+    };
+    const mode = resolveOutputMode({ explicit: args.output, json: args.json, io });
+    printResultTable(output, mode, io);
+    await emitTelemetryEvent({
+      name: 'mongo_query_completed',
+      projectDir: args.projectDir,
+      io,
+      fields: {
+        driver,
+        isDemoConnection: demoConnection,
+        stageCount: args.pipeline.length,
+        durationMs: Math.max(0, performance.now() - startedAt),
+        outcome: 'ok',
+      },
+    });
+    return 0;
+  } catch (error) {
+    const errorClass = scrubErrorClass(error);
+    await emitTelemetryEvent({
+      name: 'mongo_query_completed',
+      projectDir: args.projectDir,
+      io,
+      fields: {
+        driver,
+        isDemoConnection: demoConnection,
+        stageCount: args.pipeline.length,
+        durationMs: Math.max(0, performance.now() - startedAt),
+        outcome: 'error',
+        ...(errorClass ? { errorClass } : {}),
+      },
+    });
+    await reportException({
+      error,
+      context: { source: 'mongo-query run', handled: true, fatal: false },
+      projectDir: args.projectDir,
+      io,
+      redactionSecrets: await collectTelemetryRedactionSecrets({
+        project,
+        projectDir: args.projectDir,
+        connectionId: args.connectionId,
+        includeLlm: false,
+        includeEmbeddings: false,
+        env: process.env,
+      }),
+    });
+    io.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`);
+    return 1;
+  }
+}

packages/cli/test/mongo-query.test.ts

@@ -0,0 +1,69 @@
+import { describe, expect, it, vi } from 'vitest';
+import { runKtxMongoQuery } from '../src/mongo-query.js';
+import { KtxMongoDbScanConnector } from '../src/connectors/mongodb/connector.js';
+import type { KtxCliIo } from '../src/cli-runtime.js';
+import type { KtxLocalProject } from '../src/context/project/project.js';
+
+function captureIo(): { io: KtxCliIo; out: () => string; err: () => string } {
+  let out = '';
+  let err = '';
+  const io = {
+    stdout: { write: (s: string) => { out += s; return true; }, isTTY: false },
+    stderr: { write: (s: string) => { err += s; return true; } },
+  } as unknown as KtxCliIo;
+  return { io, out: () => out, err: () => err };
+}
+
+function mongoConnector(): KtxMongoDbScanConnector {
+  return new KtxMongoDbScanConnector({
+    connectionId: 'mongo',
+    connection: { driver: 'mongodb', url: 'mongodb://localhost:27017/app', databases: ['app'] },
+    clientFactory: {
+      create: () => ({
+        listCollections: vi.fn(async () => []),
+        estimatedDocumentCount: vi.fn(async () => 0),
+        find: vi.fn(async () => []),
+        aggregate: vi.fn(async () => [{ _id: 'a1', city: 'Indianapolis' }, { _id: 'a2', city: 'Indianapolis' }]),
+        ping: vi.fn(async () => undefined),
+        close: vi.fn(async () => undefined),
+      }),
+    },
+  });
+}
+
+const project = {
+  projectDir: '/tmp/ktx',
+  config: { connections: { mongo: { driver: 'mongodb', url: 'mongodb://localhost:27017/app', databases: ['app'] } } },
+} as unknown as KtxLocalProject;
+
+const baseArgs = {
+  projectDir: '/tmp/ktx',
+  connectionId: 'mongo',
+  collection: 'business',
+  pipeline: [{ $match: { city: 'Indianapolis' } }],
+  limit: 100,
+  cliVersion: '0.0.0-test',
+};
+
+describe('runKtxMongoQuery', () => {
+  it('prints rows in plain mode and returns exit code 0', async () => {
+    const { io, out } = captureIo();
+    const code = await runKtxMongoQuery({ ...baseArgs, output: 'plain' }, io, {
+      loadProject: async () => project,
+      createScanConnector: (async () => mongoConnector()) as never,
+    });
+    expect(code).toBe(0);
+    expect(out()).toContain('_id\tcity');
+    expect(out()).toContain('a1\tIndianapolis');
+  });
+
+  it('returns exit code 1 and writes the error message for an unconfigured connection', async () => {
+    const { io, err } = captureIo();
+    const code = await runKtxMongoQuery({ ...baseArgs, connectionId: 'nope' }, io, {
+      loadProject: async () => project,
+      createScanConnector: (async () => mongoConnector()) as never,
+    });
+    expect(code).toBe(1);
+    expect(err().length).toBeGreaterThan(0);
+  });
+});