feat(orm): new selector API, still work in progress 5

packages/bson/src/bson-parser.ts

@@ -15,7 +15,7 @@ import {
     digitByteSize,
     TWO_PWR_32_DBL_N,
 } from './utils.js';
-import { decodeUTF8 } from './strings.js';
+import { decodeUTF8 } from '@deepkit/core';
 import { nodeBufferToArrayBuffer, ReflectionKind, SerializationError, Type } from '@deepkit/type';
 import { hexTable } from './model.js';
 
@@ -316,7 +316,7 @@ export class BaseParser {
     }
 
     /**
-     * Size includes the \0. If not existend, increase by 1.
+     * Size includes the \0. If not existent, increase by 1.
      */
     eatString(size: number): string {
         this.offset += size;

packages/core/src/string.ts

@@ -17,3 +17,57 @@ export function indent(indentation: number, prefix: string = '') {
 export function capitalize(string: string): string {
     return string.charAt(0).toUpperCase() + string.slice(1)
 }
+
+const decoder = new TextDecoder('utf-8');
+
+const decodeUTF8Fast = 'undefined' !== typeof Buffer ? Buffer.prototype.utf8Slice : undefined;
+
+const decodeUTF8Big = decodeUTF8Fast ? (buffer: Uint8Array, off: number, end: number) => {
+    return decodeUTF8Fast.call(buffer, off, end);
+} : (buffer: Uint8Array, off: number, end: number) => {
+    return decoder.decode(buffer.slice(off, end));
+}
+
+export function decodeUTF8(buffer: Uint8Array, off: number = 0, end: number) {
+    if (end - off > 64) {
+        return decodeUTF8Big(buffer, off, end);
+    } else {
+        return decodeUTF8Short(buffer, off, end);
+    }
+}
+
+export function decodeUTF8Short(buffer: Uint8Array, off: number = 0, end: number) {
+    let s = '';
+    while (off < end) {
+        let c = buffer[off++];
+
+        if (c > 127) {
+            if (c > 191 && c < 224) {
+                if (off >= end)
+                    throw new Error('UTF-8 decode: incomplete 2-byte sequence');
+                c = (c & 31) << 6 | buffer[off++] & 63;
+            } else if (c > 223 && c < 240) {
+                if (off + 1 >= end)
+                    throw new Error('UTF-8 decode: incomplete 3-byte sequence');
+                c = (c & 15) << 12 | (buffer[off++] & 63) << 6 | buffer[off++] & 63;
+            } else if (c > 239 && c < 248) {
+                if (off + 2 >= end)
+                    throw new Error('UTF-8 decode: incomplete 4-byte sequence');
+                c = (c & 7) << 18 | (buffer[off++] & 63) << 12 | (buffer[off++] & 63) << 6 | buffer[off++] & 63;
+            } else throw new Error('UTF-8 decode: unknown multibyte start 0x' + c.toString(16) + ' at index ' + (off - 1));
+            if (c <= 0xffff) {
+                s += String.fromCharCode(c);
+            } else if (c <= 0x10ffff) {
+                c -= 0x10000;
+                s += String.fromCharCode(c >> 10 | 0xd800, c & 0x3FF | 0xdc00);
+            } else throw new Error('UTF-8 decode: code point 0x' + c.toString(16) + ' exceeds UTF-16 reach');
+        } else {
+            if (c === 0) {
+                return s;
+            }
+
+            s += String.fromCharCode(c);
+        }
+    }
+    return s;
+}

packages/core/tests/core.spec.ts

@@ -620,3 +620,20 @@ test('isGlobalClass', () => {
 
     expect(isGlobalClass(Uint8Array)).toBe(true);
 });
+
+test('typed array offset', () => {
+    const a = new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
+    expect(a.byteOffset).toBe(0);
+    expect(a.length).toBe(10);
+    expect(a[0]).toBe(1);
+
+    const b=  new Uint8Array(a.buffer, 1);
+    expect(b.byteOffset).toBe(1);
+    expect(b.length).toBe(9);
+    expect(b[0]).toBe(2);
+
+    const c = new Uint8Array(b.buffer, b.byteOffset + 1);
+    expect(c.byteOffset).toBe(2);
+    expect(c.length).toBe(8);
+    expect(c[0]).toBe(3);
+});

packages/mongo/src/client/client.ts

@@ -8,7 +8,13 @@
  * You should have received a copy of the MIT License along with this program.
  */
 
-import { ConnectionRequest, MongoConnection, MongoConnectionPool, MongoDatabaseTransaction, MongoStats } from './connection.js';
+import {
+    ConnectionRequest,
+    MongoConnection,
+    MongoConnectionPool,
+    MongoDatabaseTransaction,
+    MongoStats,
+} from './connection.js';
 import { isErrorRetryableRead, isErrorRetryableWrite, MongoError } from './error.js';
 import { sleep } from '@deepkit/core';
 import { Command } from './command/command.js';
@@ -27,9 +33,7 @@ export class MongoClient {
 
     protected serializer: BSONBinarySerializer = mongoBinarySerializer;
 
-    constructor(
-        connectionString: string
-    ) {
+    constructor(connectionString: string) {
         this.config = new MongoClientConfig(connectionString);
         this.connectionPool = new MongoConnectionPool(this.config, this.serializer, this.stats);
     }

packages/mongo/src/client/connection.ts

@@ -63,8 +63,6 @@ export class MongoConnectionPool {
 
     protected queue: { resolve: (connection: MongoConnection) => void, request: ConnectionRequest }[] = [];
 
-    protected nextConnectionClose: Promise<boolean> = Promise.resolve(true);
-
     protected lastError?: Error;
 
     constructor(
@@ -172,7 +170,7 @@ export class MongoConnectionPool {
             waiter.resolve(connection);
             //we don't set reserved/set cleanupTimeout,
             //since the connection is already reserved and the timeout
-            //is only set when the connection actually starting idling.
+            //is only set when the connection actually is starting idling.
             return;
         }
 
@@ -435,7 +433,7 @@ export class MongoConnection {
 
     /**
      * Puts a command on the queue and executes it when queue is empty.
-     * A promises is return that is resolved with the  when executed successfully, or rejected
+     * A promise is returned that is resolved when executed successfully, or rejected
      * when timed out, parser error, or any other error.
      */
     public async execute<T extends Command<unknown>>(command: T): Promise<ReturnType<T['execute']>> {
@@ -497,9 +495,9 @@ export class MongoConnection {
             writer.writeInt32(messageLength);
 
             //detect backPressure
+            this.socket.write(buffer);
             this.bytesSent += buffer.byteLength;
             this.onSent(buffer.byteLength);
-            this.socket.write(buffer);
         } catch (error) {
             console.log('failed sending message', message, 'for type', stringifyType(type));
             throw error;

packages/mongo/src/client/options.ts

@@ -27,7 +27,7 @@ export class ConnectionOptions {
     journal?: string;
 
     appName?: string;
-    retryWrites: boolean = true;
+    retryWrites: boolean = false;
     retryReads: boolean = true;
 
     readConcernLevel: 'local' | 'majority' | 'linearizable' | 'available' = 'majority';

packages/mongo/src/query.resolver.ts

@@ -8,7 +8,17 @@
  * You should have received a copy of the MIT License along with this program.
  */
 
-import { DatabaseAdapter, DatabaseDeleteError, DatabasePatchError, DatabaseSession, DeleteResult, Formatter, GenericQueryResolver, OrmEntity, PatchResult } from '@deepkit/orm';
+import {
+    DatabaseAdapter,
+    DatabaseDeleteError,
+    DatabasePatchError,
+    DatabaseSession,
+    DeleteResult,
+    Formatter,
+    GenericQueryResolver,
+    OrmEntity,
+    PatchResult,
+} from '@deepkit/orm';
 import {
     Changes,
     getPartialSerializeFunction,
@@ -314,7 +324,7 @@ export class MongoQueryResolver<T extends OrmEntity> extends GenericQueryResolve
     }
 
     public async find(model: MongoQueryModel<T>): Promise<T[]> {
-        const formatter = this.createFormatter(model.withIdentityMap);
+        const formatter9 = this.createFormatter(model.withIdentityMap);
         const connection = await this.client.getConnection(undefined, this.session.assignedTransaction);
 
         try {

packages/orm/src/database-session.ts

@@ -292,6 +292,7 @@ export abstract class DatabaseTransaction {
 export class DatabaseSession<ADAPTER extends DatabaseAdapter = DatabaseAdapter> {
     public readonly id = SESSION_IDS++;
     public withIdentityMap = true;
+    public withChangeDetection = true;
 
     /**
      * When this session belongs to a transaction, then this is set.

packages/orm/src/formatter.ts

@@ -63,9 +63,12 @@ export class Formatter {
         protected serializer: Serializer,
         protected hydrator?: HydratorFn,
         protected identityMap?: IdentityMap,
+        protected withChangeDetection: boolean = true,
     ) {
         this.deserialize = getSerializeFunction(rootClassSchema.type, serializer.deserializeRegistry);
         this.partialDeserialize = getPartialSerializeFunction(rootClassSchema.type, serializer.deserializeRegistry);
+        if (identityMap) throw new Error('nope');
+        if (withChangeDetection) throw new Error('nope');
     }
 
     protected getInstancePoolForClass(classType: ClassType): Map<PKHash, any> {
@@ -274,7 +277,7 @@ export class Formatter {
         const converted = this.createObject(model, classState, classSchema, dbRecord);
 
         if (!partial) {
-            if (model.withChangeDetection !== false) getInstanceState(classState, converted).markAsPersisted();
+            if (this.withChangeDetection) getInstanceState(classState, converted).markAsPersisted();
             if (pool) pool.set(pkHash, converted);
             if (this.identityMap) this.identityMap.store(classSchema, converted);
         }
@@ -336,7 +339,7 @@ export class Formatter {
             : (partial ? getPartialSerializeFunction(classSchema.type, this.serializer.deserializeRegistry)(dbRecord) : getSerializeFunction(classSchema.type, this.serializer.deserializeRegistry)(dbRecord));
 
         if (!partial) {
-            if (model.withChangeDetection !== false) getInstanceState(classState, converted).markAsFromDatabase();
+            if (this.withChangeDetection) getInstanceState(classState, converted).markAsFromDatabase();
         }
 
         // if (!partial && model.lazyLoad.size) {

packages/orm/src/select.ts

@@ -268,7 +268,12 @@ export type OpExpression = {
 export type Op = ((...args: any[]) => OpExpression) & { id: symbol };
 
 export function getStateCacheId(state: SelectorState): string {
-    const cacheId = state.schema.type.id + '_' + state.where?.[treeTag].id + '_' + state.orderBy?.map(v => v.a[treeTag].id).join(':');
+    const cacheId = state.schema.type.id
+        + '_' + state.where?.[treeTag].id
+        + '_' + state.limit
+        + '_' + state.offset
+        + '_' + state.select.map(v => v[treeTag].id).join(':')
+        + '_' + state.orderBy?.map(v => v.a[treeTag].id).join(':');
     //todo select also
     // todo join also
     return cacheId;
@@ -545,6 +550,20 @@ export class Query2<T extends object, R = any> {
         return this;
     }
 
+    /**
+     * When receiving full objects the change-detector is enabled by default
+     * to be able to calculate change sets for database.persist()/session.commit().
+     *
+     * If disabled, it is impossible to send updates via database.persist()/session.commit(),
+     * and patchOne/patchMany has to be used.
+     *
+     * This is disabled per default for partial results.
+     */
+    disableChangeDetection(): this {
+        this.state.withChangeDetection = false;
+        return this;
+    }
+
     protected async callOnFetchEvent(query: Query2<object>): Promise<void> {
         const hasEvents = this.session.eventDispatcher.hasListeners(onFind);
         if (!hasEvents) return;
@@ -607,24 +626,24 @@ export class Query2<T extends object, R = any> {
      * @throws DatabaseError
      */
     public async find(): Promise<T[]> {
-        const frame = this.session
-            .stopwatch?.start('Find:' + this.classSchema.getClassName(), FrameCategory.database);
+        // const frame = this.session
+        //     .stopwatch?.start('Find:' + this.classSchema.getClassName(), FrameCategory.database);
 
         try {
-            frame?.data({
-                collection: this.classSchema.getCollectionName(),
-                className: this.classSchema.getClassName(),
-            });
-            const eventFrame = this.session.stopwatch?.start('Events');
-            await this.callOnFetchEvent(this);
-            this.onQueryResolve(this);
-            eventFrame?.end();
+            // frame?.data({
+            //     collection: this.classSchema.getCollectionName(),
+            //     className: this.classSchema.getClassName(),
+            // });
+            // const eventFrame = this.session.stopwatch?.start('Events');
+            // await this.callOnFetchEvent(this);
+            // this.onQueryResolve(this);
+            // eventFrame?.end();
             return await this.resolver.find(this.state) as T[];
         } catch (error: any) {
             await this.session.eventDispatcher.dispatch(onDatabaseError, new DatabaseErrorEvent(error, this.session, this.state.schema, this));
             throw error;
         } finally {
-            frame?.end();
+            // frame?.end();
         }
     }