feat(mongo): export custom command API

This allows to write type-safe high-performance custom mongo commands very easily.

This exports also all internal client commands we use in the ORM.

packages/bson/src/bson-serializer.ts

@@ -240,6 +240,9 @@ export class ValueWithBSONSerializer {
     }
 }
 
+export function wrapValue<T>(value: any, type?: ReceiveType<T>) {
+    return new ValueWithBSONSerializer(value, resolveReceiveType(type));
+}
 
 export class Writer {
     public dataView: DataView;

packages/bson/tests/bson-serialize.spec.ts

@@ -1,5 +1,5 @@
 import { expect, test } from '@jest/globals';
-import { getBSONSerializer, getBSONSizer, getValueSize, hexToByte, serializeBSONWithoutOptimiser, uuidStringToByte } from '../src/bson-serializer.js';
+import { getBSONSerializer, getBSONSizer, getValueSize, hexToByte, serializeBSONWithoutOptimiser, uuidStringToByte, wrapValue } from '../src/bson-serializer.js';
 import {
     BinaryBigInt,
     createReference,
@@ -1428,3 +1428,11 @@ test('undefined for required object', () => {
     const bson = serializeBSONWithoutOptimiser(user);
     expect(() => deserialize(bson)).toThrow('Cannot convert bson type UNDEFINED to {id: number}');
 });
+
+test('wrapValue', () => {
+    const objectId = wrapValue<MongoId>('507f191e810c19729de860ea');
+    const bson = serializeBSONWithoutOptimiser({v: objectId});
+    const back = deserialize(bson);
+    expect(back.v).toBeInstanceOf(OfficialObjectId);
+    expect(back.v.toHexString()).toBe('507f191e810c19729de860ea');
+});

packages/mongo/index.ts

@@ -14,9 +14,30 @@ export * from './src/persistence.js';
 export * from './src/query.model.js';
 export * from './src/query.resolver.js';
 export * from './src/query.js';
+
 export * from './src/client/client.js';
+export * from './src/client/config.js';
 export * from './src/client/error.js';
+export * from './src/client/options.js';
 
-export { UpdateCommand } from './src/client/command/update.js';
-export { AggregateCommand }  from './src/client/command/aggregate.js';
-export { FindAndModifyCommand }  from './src/client/command/findAndModify.js';
+export * from './src/client/command/auth/auth.js';
+export * from './src/client/command/auth/scram.js';
+export * from './src/client/command/auth/x509.js';
+export * from './src/client/command/abortTransaction.js';
+export * from './src/client/command/aggregate.js';
+export * from './src/client/command/command.js';
+export * from './src/client/command/commitTransaction.js';
+export * from './src/client/command/count.js';
+export * from './src/client/command/createCollection.js';
+export * from './src/client/command/createIndexes.js';
+export * from './src/client/command/delete.js';
+export * from './src/client/command/dropDatabase.js';
+export * from './src/client/command/dropIndexes.js';
+export * from './src/client/command/find';
+export * from './src/client/command/findAndModify.js';
+export * from './src/client/command/getMore.js';
+export * from './src/client/command/handshake.js';
+export * from './src/client/command/insert.js';
+export * from './src/client/command/ismaster.js';
+export * from './src/client/command/startSession.js';
+export * from './src/client/command/update.js';

packages/mongo/src/adapter.ts

@@ -53,7 +53,7 @@ class MongoRawCommandQuery<T> implements FindQuery<T> {
     constructor(
         protected session: DatabaseSession<MongoDatabaseAdapter>,
         protected client: MongoClient,
-        protected command: Command,
+        protected command: Command<any>,
     ) {
     }
 
@@ -90,23 +90,23 @@ class MongoRawCommandQuery<T> implements FindQuery<T> {
     }
 }
 
-export class MongoRawFactory implements RawFactory<[Command]> {
+export class MongoRawFactory implements RawFactory<[Command<any>]> {
     constructor(
         protected session: DatabaseSession<MongoDatabaseAdapter>,
         protected client: MongoClient,
     ) {
     }
 
     create<Entity = any, ResultSchema = Entity>(
-        commandOrPipeline: Command | any[],
+        commandOrPipeline: Command<ResultSchema> | any[],
         type?: ReceiveType<Entity>,
         resultType?: ReceiveType<ResultSchema>,
     ): MongoRawCommandQuery<ResultSchema> {
         type = resolveReceiveType(type);
         const resultSchema = resultType ? resolveReceiveType(resultType) : undefined;
 
         const command = isArray(commandOrPipeline) ? new AggregateCommand(ReflectionClass.from(type), commandOrPipeline, resultSchema) : commandOrPipeline;
-        return new MongoRawCommandQuery<ResultSchema>(this.session, this.client, command);
+        return new MongoRawCommandQuery<ResultSchema>(this.session, this.client, command as Command<any>);
     }
 }
 

packages/mongo/src/client/client.ts

@@ -71,7 +71,7 @@ export class MongoClient {
         return connection;
     }
 
-    public async execute<T extends Command>(command: T): Promise<ReturnType<T['execute']>> {
+    public async execute<T extends Command<unknown>>(command: T): Promise<ReturnType<T['execute']>> {
         const maxRetries = 10;
         const request = { readonly: !command.needsWritableHost() };
 

packages/mongo/src/client/command/abortTransaction.ts

@@ -21,7 +21,7 @@ interface Request {
     autocommit?: boolean,
 }
 
-export class AbortTransactionCommand extends Command {
+export class AbortTransactionCommand extends Command<BaseResponse> {
     needsWritableHost() {
         return false;
     }

packages/mongo/src/client/command/aggregate.ts

@@ -27,7 +27,7 @@ interface AggregateMessage {
     autocommit?: boolean,
 }
 
-export class AggregateCommand<T, R = BaseResponse> extends Command {
+export class AggregateCommand<T, R = BaseResponse> extends Command<R[]> {
     partial: boolean = false;
     batchSize: number = 1_000_000;
 

packages/mongo/src/client/command/auth/auth.ts

@@ -12,5 +12,5 @@ import { MongoClientConfig } from '../../config.js';
 import { Command } from '../command.js';
 
 export interface MongoAuth {
-    auth(command: Command, config: MongoClientConfig): Promise<void>;
+    auth(command: Command<unknown>, config: MongoClientConfig): Promise<void>;
 }

packages/mongo/src/client/command/auth/scram.ts

@@ -96,7 +96,7 @@ export abstract class ScramAuth implements MongoAuth {
         this.cryptoMethod = this.mechanism === 'SCRAM-SHA-1' ? 'sha1' : 'sha256';
     }
 
-    async auth(command: Command, config: MongoClientConfig): Promise<void> {
+    async auth(command: Command<unknown>, config: MongoClientConfig): Promise<void> {
         const username = cleanUsername(config.authUser || '');
         const password = config.authPassword || '';
 

packages/mongo/src/client/command/auth/x509.ts

@@ -23,7 +23,7 @@ interface AuthenticateResponse extends BaseResponse {
 }
 
 export class X509Auth implements MongoAuth {
-    async auth(command: Command, config: MongoClientConfig): Promise<void> {
+    async auth(command: Command<unknown>, config: MongoClientConfig): Promise<void> {
         await command.sendAndWait<AuthenticateCommand, AuthenticateResponse>({
             authenticate: 1,
             mechanism: 'MONGODB-X509',

packages/mongo/src/client/command/command.ts

@@ -13,33 +13,10 @@ import { handleErrorResponse, MongoDatabaseError, MongoError } from '../error.js
 import { MongoClientConfig } from '../config.js';
 import { Host } from '../host.js';
 import type { MongoDatabaseTransaction } from '../connection.js';
-import { ReceiveType, ReflectionClass, resolveReceiveType, SerializationError, stringifyType, Type, typeOf, typeSettings, UnpopulatedCheck, ValidationError } from '@deepkit/type';
+import { ReceiveType, resolveReceiveType, SerializationError, stringifyType, Type, typeOf, typeSettings, UnpopulatedCheck, ValidationError } from '@deepkit/type';
 import { BSONDeserializer, deserializeBSONWithoutOptimiser, getBSONDeserializer } from '@deepkit/bson';
 import { mongoBinarySerializer } from '../../mongo-serializer.js';
 
-export interface CommandMessageResponseCallbackResult<T> {
-    /**
-     * When the command is finished, set the `result`
-     */
-    result?: T;
-
-    /**
-     * When the command is not finished and another message should be sent, set the new CommandMessage
-     * as `next`.
-     */
-    next?: CommandMessage<any, any>;
-}
-
-export class CommandMessage<T, R> {
-    constructor(
-        public readonly schema: ReflectionClass<T>,
-        public readonly message: T,
-        public readonly responseSchema: ReflectionClass<R>,
-        public readonly responseCallback: (response: R) => { result?: any, next?: CommandMessage<any, any> },
-    ) {
-    }
-}
-
 export interface BaseResponse {
     ok: number;
     errmsg?: string;
@@ -48,12 +25,12 @@ export interface BaseResponse {
     writeErrors?: Array<{ index: number, code: number, errmsg: string }>;
 }
 
-export abstract class Command {
+export abstract class Command<T> {
     protected current?: { responseType?: Type, resolve: Function, reject: Function };
 
     public sender?: <T>(schema: Type, message: T) => void;
 
-    public sendAndWait<T, R = BaseResponse>(
+    public sendAndWait<T, R extends BaseResponse = BaseResponse>(
         message: T, messageType?: ReceiveType<T>, responseType?: ReceiveType<R>,
     ): Promise<R> {
         if (!this.sender) throw new Error(`No sender set in command ${getClassName(this)}`);
@@ -64,7 +41,7 @@ export abstract class Command {
         });
     }
 
-    abstract execute(config: MongoClientConfig, host: Host, transaction?: MongoDatabaseTransaction): Promise<any>;
+    abstract execute(config: MongoClientConfig, host: Host, transaction?: MongoDatabaseTransaction): Promise<T>;
 
     abstract needsWritableHost(): boolean;
 
@@ -110,17 +87,22 @@ export abstract class Command {
     }
 }
 
-// export class GenericCommand extends Command {
-//     constructor(protected classSchema: ReflectionClass<any>, protected cmd: { [name: string]: any }, protected _needsWritableHost: boolean) {
-//         super();
-//     }
-//
-//     async execute(config): Promise<number> {
-//         const res = await this.sendAndWait(this.classSchema, this.cmd);
-//         return res.n;
-//     }
-//
-//     needsWritableHost(): boolean {
-//         return this._needsWritableHost;
-//     }
-// }
+export function createCommand<Request extends {}, Response extends BaseResponse>(
+    request: Request,
+    options: Partial<{ needsWritableHost: boolean }> = {},
+    typeRequest?: ReceiveType<Request>,
+    typeResponse?: ReceiveType<Response>,
+): Command<Response> {
+    class DynamicCommand extends Command<Response> {
+        async execute(): Promise<Response> {
+            return this.sendAndWait(request, typeRequest, typeResponse);
+        }
+
+        needsWritableHost(): boolean {
+            return options.needsWritableHost || false;
+        }
+    }
+
+    return new DynamicCommand();
+}
+

packages/mongo/src/client/command/commitTransaction.ts

@@ -21,7 +21,7 @@ interface Request {
     autocommit?: boolean;
 }
 
-export class CommitTransactionCommand extends Command {
+export class CommitTransactionCommand extends Command<BaseResponse> {
     needsWritableHost() {
         return false;
     }

packages/mongo/src/client/command/count.ts

@@ -27,7 +27,7 @@ interface CountSchema {
     autocommit?: boolean;
 }
 
-export class CountCommand<T extends ReflectionClass<any>> extends Command {
+export class CountCommand<T extends ReflectionClass<any>> extends Command<number> {
     constructor(
         public schema: T,
         public query: { [name: string]: any } = {},

packages/mongo/src/client/command/createCollection.ts

@@ -16,7 +16,7 @@ interface RequestSchema {
     $db: string;
 }
 
-export class CreateCollectionCommand<T extends ReflectionClass<any>> extends Command {
+export class CreateCollectionCommand<T extends ReflectionClass<any>> extends Command<BaseResponse> {
     constructor(
         public schema: T,
     ) {

packages/mongo/src/client/command/createIndexes.ts

@@ -26,7 +26,7 @@ interface RequestSchema {
     indexes: CreateIndex[];
 }
 
-export class CreateIndexesCommand<T extends ReflectionClass<any>> extends Command {
+export class CreateIndexesCommand<T extends ReflectionClass<any>> extends Command<BaseResponse> {
     constructor(
         public schema: T,
         public indexes: CreateIndex[],

packages/mongo/src/client/command/delete.ts

@@ -25,7 +25,7 @@ interface DeleteSchema {
     startTransaction?: boolean;
 }
 
-export class DeleteCommand<T extends ReflectionClass<any>> extends Command {
+export class DeleteCommand<T extends ReflectionClass<any>> extends Command<number> {
 
     constructor(
         public schema: T,

packages/mongo/src/client/command/dropDatabase.ts

@@ -9,14 +9,13 @@
  */
 
 import { Command } from './command.js';
-import { ReflectionClass } from '@deepkit/type';
 
 interface DropDatabase {
     dropDatabase: 1;
     $db: string;
 }
 
-export class DropDatabaseCommand<T extends ReflectionClass<any>> extends Command {
+export class DropDatabaseCommand<T> extends Command<void> {
     constructor(protected dbName: any) {
         super();
     }

packages/mongo/src/client/command/dropIndexes.ts

@@ -17,7 +17,7 @@ interface RequestSchema {
     index: string[];
 }
 
-export class DropIndexesCommand<T extends ReflectionClass<any>> extends Command {
+export class DropIndexesCommand<T extends ReflectionClass<any>> extends Command<BaseResponse> {
     constructor(
         public schema: T,
         public names: string[]

packages/mongo/src/client/command/empty.ts

@@ -11,7 +11,7 @@
 import { Command } from './command.js';
 
 
-export class EmptyCommand extends Command {
+export class EmptyCommand extends Command<void> {
     execute(): Promise<any> {
         return Promise.resolve(undefined);
     }

packages/mongo/src/client/command/find.ts

@@ -30,7 +30,7 @@ interface FindSchema {
     autocommit?: boolean,
 }
 
-export class FindCommand<T> extends Command {
+export class FindCommand<T> extends Command<T[]> {
     batchSize: number = 1_000_000;
 
     constructor(

packages/mongo/src/client/command/findAndModify.ts

@@ -29,7 +29,7 @@ interface findAndModifySchema {
     startTransaction?: boolean;
 }
 
-export class FindAndModifyCommand<T extends ReflectionClass<any>> extends Command {
+export class FindAndModifyCommand<T extends ReflectionClass<any>> extends Command<FindAndModifyResponse> {
     public upsert = false;
     public fields: string[] = [];
     public returnNew: boolean = false;

packages/mongo/src/client/command/handshake.ts

@@ -74,7 +74,7 @@ function detectedAuthMechanismFromResponse(response: IsMasterResponse): AuthMech
  * It differs to regular IsMasterCommand in a way that it sends `client` data as well,
  * which is only allowed at the first message, and additionally sends auth data if necessary.
  */
-export class HandshakeCommand extends Command {
+export class HandshakeCommand extends Command<boolean> {
     needsWritableHost() {
         return false;
     }

packages/mongo/src/client/command/insert.ts

@@ -25,7 +25,7 @@ interface InsertSchema {
     startTransaction?: boolean;
 }
 
-export class InsertCommand<T> extends Command {
+export class InsertCommand<T> extends Command<number> {
     constructor(
         protected schema: ReflectionClass<T>,
         protected documents: T[]

packages/mongo/src/client/command/ismaster.ts

@@ -45,7 +45,7 @@ interface IsMasterSchema {
     $db: string;
 }
 
-export class IsMasterCommand extends Command {
+export class IsMasterCommand extends Command<IsMasterResponse> {
     needsWritableHost() {
         return false;
     }

packages/mongo/src/client/command/startSession.ts

@@ -25,7 +25,7 @@ interface SessionSchema {
     autocommit?: boolean;
 }
 
-export class StartSessionCommand extends Command {
+export class StartSessionCommand extends Command<SessionResponse> {
     needsWritableHost() {
         return false;
     }

packages/mongo/src/client/command/update.ts

@@ -30,7 +30,7 @@ interface UpdateSchema {
     startTransaction?: boolean;
 }
 
-export class UpdateCommand<T extends ReflectionClass<any>> extends Command {
+export class UpdateCommand<T extends ReflectionClass<any>> extends Command<number> {
     constructor(
         public schema: T,
         public updates: { q: any, u: any, multi: boolean }[] = [],