Save Direct updated

Author: ackava

src/drivers/base/BaseDriver.ts

@@ -4,9 +4,10 @@ import EntityAccessError from "../../common/EntityAccessError.js";
 import "../../common/IDisposable.js";
 
 import QueryCompiler from "../../compiler/QueryCompiler.js";
+import { IColumn } from "../../decorators/IColumn.js";
 import EntityType from "../../entity-query/EntityType.js";
 import Migrations from "../../migrations/Migrations.js";
-import ChangeEntry from "../../model/changes/ChangeEntry.js";
+import ChangeEntry, { IChange } from "../../model/changes/ChangeEntry.js";
 import { BinaryExpression, Constant, DeleteStatement, ExistsExpression, Expression, ExpressionAs, Identifier, InsertStatement, NotExits, NumberLiteral, ReturnUpdated, SelectStatement, TableLiteral, UnionAllStatement, UpdateStatement, UpsertStatement, ValuesStatement } from "../../query/ast/Expressions.js";
 
 export interface IRecord {
@@ -229,6 +230,97 @@ export abstract class BaseDriver {
     /** Must dispose ObjectPools */
     abstract dispose();
 
+    abstract insertQuery(type: EntityType, entity): { text: string, values: any[] };
+
+    updateQuery(type: EntityType, entity: any, changes?: Map<IColumn, IChange>): { text: string; values: any[]; } {
+        let where = "";
+        let setParams = "";
+        let returning = "";
+        const values = [];
+        let i = 1;
+        if (changes) {
+            for (const [iterator, value] of changes.entries()) {
+                if (iterator.computed) {
+                    if (returning) {
+                        returning += ",";
+                    }
+                    returning += iterator.columnName + " as " + this.compiler.quote(iterator.name);
+                    continue;
+                }
+                if (setParams) {
+                    setParams += ",\r\n\t\t";
+                }
+                setParams += `${iterator.columnName} = $${i++}`;
+                values.push(value.newValue);
+            }
+            for (const iterator of type.keys) {
+                if(where) {
+                    where += "\r\n\t\tAND ";
+                }
+                where += `${iterator.columnName} = $${i++}`;
+                values.push(entity[iterator.name]);
+                continue;
+
+            }
+        } else {
+            for (const iterator of type.columns) {
+                if (iterator.key) {
+                    if(where) {
+                        where += "\r\n\t\tAND ";
+                    }
+                    where += `${iterator.columnName} = $${i++}`;
+                    values.push(entity[iterator.name]);
+                    continue;
+                }
+                if (setParams) {
+                    setParams += ",\r\n\t\t";
+                }
+                setParams += `${iterator.columnName} = $${i++}`;
+                values.push(entity[iterator.name]);
+            }
+        }
+        const text = `UPDATE ${type.fullyQualifiedTableName}\r\n\tSET ${setParams}\r\n\tWHERE ${where}`;
+        return { text, values };
+    }
+
+    deleteQuery(type: EntityType, entity: any): { text: string; values: any[]; } {
+        let where = "";
+        const values = [];
+        let i = 1;
+        for (const iterator of type.keys) {
+            if(where) {
+                where += "\r\n\t\tAND ";
+            }
+            where += `${iterator.columnName} = $${i++}`;
+            values.push(entity[iterator.name]);
+        }
+        const text = `DELETE FROM ${type.fullyQualifiedTableName}\r\n\tWHERE ${where}`;
+        return { text, values };
+    }
+
+    selectQueryWithKeys(type: EntityType, entity) {
+        let where = "";
+        let columns = "";
+        const values = [];
+        let i = 1;
+        for (const iterator of type.columns) {
+            if (iterator.key) {
+                if(where) {
+                    where += "\r\n\t\tAND ";
+                }
+                where += `${iterator.columnName} = $${i++}`;
+                values.push(entity[iterator.name]);
+                continue;
+            }
+            if (columns) {
+                columns += "\r\n\t\t";
+            }
+            columns += `${iterator.columnName} as ${this.compiler.escapeLiteral(iterator.name)}`;
+        }
+        const text = `SELECT ${columns}\r\n\tFROM ${type.fullyQualifiedTableName}\r\n\tWHERE ${where}`;
+        return { text, values };
+    }
+
     createSelectWithKeysExpression(type: EntityType, check: any, returnFields: Expression[] ) {
         let where = null as Expression;
         for (const key in check) {

src/drivers/postgres/PostgreSqlDriver.ts

@@ -1,6 +1,7 @@
 /* eslint-disable no-console */
 import ObjectPool, { IPooledObject } from "../../common/ObjectPool.js";
 import QueryCompiler from "../../compiler/QueryCompiler.js";
+import EntityType from "../../entity-query/EntityType.js";
 import Migrations from "../../migrations/Migrations.js";
 import PostgresAutomaticMigrations from "../../migrations/postgres/PostgresAutomaticMigrations.js";
 import { BaseConnection, BaseDriver, EntityTransaction, IDbConnectionString, IDbReader, IQuery, toQuery } from "../base/BaseDriver.js";
@@ -109,6 +110,38 @@ export default class PostgreSqlDriver extends BaseDriver {
         });
     }
 
+    insertQuery(type: EntityType, entity: any): { text: string; values: any[]; } {
+        let fields = "";
+        let valueParams = "";
+        let returning = "";
+        const values = [];
+        let i = 1;
+        for (const iterator of type.columns) {
+            if (iterator.generated || iterator.computed) {
+                if (returning) {
+                    returning += ",";
+                } else {
+                    returning = "RETURNING ";
+                }
+                returning += iterator.columnName + " as " + this.compiler.quote(iterator.name);
+                continue;
+            }
+            const value = entity[iterator.name];
+            if (value === void 0) {
+                continue;
+            }
+            if (fields) {
+                fields += ",";
+                valueParams += ",";
+            }
+            fields += iterator.columnName;
+            valueParams += `$${i++}`;
+            values.push(value);
+        }
+        const text = `INSERT INTO ${type.fullyQualifiedTableName}(${fields}) VALUES (${valueParams}) ${returning}`;
+        return { text, values };
+    }
+
     dispose() {
         this.pool?.dispose().catch(console.error);
     }

src/drivers/sql-server/SqlServerDriver.ts

@@ -7,6 +7,7 @@ import SqlServerQueryCompiler from "./SqlServerQueryCompiler.js";
 import SqlServerAutomaticMigrations from "../../migrations/sql-server/SqlServerAutomaticMigrations.js";
 import { SqlServerLiteral } from "./SqlServerLiteral.js";
 import TimedCache from "../../common/cache/TimedCache.js";
+import EntityType from "../../entity-query/EntityType.js";
 
 export type ISqlServerConnectionString = IDbConnectionString & sql.config;
 
@@ -25,6 +26,39 @@ export default class SqlServerDriver extends BaseDriver {
         config.server = config.host;
     }
 
+    insertQuery(type: EntityType, entity: any): { text: string; values: any[]; } {
+        let fields = "";
+        let valueParams = "";
+        const values = [];
+        let returning = "";
+        let i = 1;
+        for (const iterator of type.columns) {
+            if (iterator.generated || iterator.computed) {
+                if (returning) {
+                    returning += ",";
+                } else {
+                    returning = "OUTPUT ";
+                }
+                returning += "INSERTED." + iterator.columnName + " as " + this.compiler.quote(iterator.name);
+                continue;
+            }
+            const field = iterator.columnName;
+            const value = entity[iterator.name];
+            if (value === void 0) {
+                continue;
+            }
+            values.push(value);
+            if (fields) {
+                fields += ",";
+                valueParams += ",";
+            }
+            fields += field;
+            valueParams += `$${i++}`;
+        }
+        const text = `INSERT INTO ${type.fullyQualifiedTableName}(${fields}) ${returning} VALUES (${valueParams})`;
+        return { text, values };
+    }
+
     dispose() {
         // do nothing
     }

src/entity-query/EntityType.ts

@@ -40,6 +40,7 @@ export default class EntityType {
     public readonly keys: IColumn[] = [];
 
     public readonly nonKeys: IColumn[] = [];
+    public readonly fullyQualifiedTableName: string;
 
     @InstanceCache
     public get fullyQualifiedName() {
@@ -67,6 +68,7 @@ export default class EntityType {
         this.schema = original.schema ? (namingConvention ? namingConvention(original.schema) : original.schema) : original.schema;
         this.entityName = original.entityName;
         this.doNotCreate = original.doNotCreate;
+        this.fullyQualifiedTableName = this.schema ? `${this.schema}.${this.name}` : this.name;
     }
 
     [addOrCreateColumnSymbol](name: string): IColumn {

src/model/EntityContext.ts

@@ -230,31 +230,36 @@ export default class EntityContext {
     }
 
     private async saveChangesInternalWithoutEvents(options?: ISaveOptions) {
+        const signal = options?.signal;
         const copy = Array.from(this.changeSet.getChanges()) as ChangeEntry[];
+        const { connection } = this;
         for (const iterator of copy) {
             switch (iterator.status) {
                 case "inserted":
                     // const insert = iterator.type.getInsertStatement();
                     // we are choosing not to create one complicated SQL as generation
                     // of insert requires checking if value is supplied or not, if not, we have to choose
                     // default value
-                    const insert = this.driver.createInsertExpression(iterator.type, iterator.entity);
-                    const r = await this.executeExpression(insert, options);
-                    iterator.apply(r);
+                    // const insert = this.driver.createInsertExpression(iterator.type, iterator.entity);
+                    // const r = await this.executeExpression(insert, options);
+                    const insert = this.driver.insertQuery(iterator.type, iterator.entity);
+                    const r = await connection.executeQuery(insert, signal);
+                    iterator.apply(r.rows[0]);
                     break;
                 case "modified":
                     // this will update the modified map
                     iterator.detect();
                     if (iterator.modified.size > 0) {
-                        const update = this.driver.createUpdateExpression(iterator);
-                        const r1 = await this.executeExpression(update, options);
-                        iterator.apply(r1 ?? {});
+                        // const update = this.driver.createUpdateExpression(iterator);
+                        const update = this.driver.updateQuery(iterator.type, iterator.entity, iterator.modified);
+                        const r1 = await connection.executeQuery(update, signal);
+                        iterator.apply(r1.rows?.[0] ?? {});
                     }
                     break;
                 case "deleted":
-                    const deleteQuery = this.driver.createDeleteExpression(iterator.type, iterator.entity);
+                    const deleteQuery = this.driver.deleteQuery(iterator.type, iterator.entity);
                     if (deleteQuery) {
-                        await this.executeExpression(deleteQuery, options);
+                        await connection.executeQuery(deleteQuery, signal);
                     }
                     iterator.apply({});
                     break;

src/model/EntitySource.ts

@@ -55,11 +55,111 @@ export type ISaveDirect<T> = {
     changes?: never
 };
 
-export class EntitySource<T = any> {
+export class EntityStatements<T = any> {
+
+    private readonly model: EntityType;
+    private readonly context: EntityContext;
+
+    constructor(private source: EntitySource<T>) {
+        this.model = source[modelSymbol];
+        this.context = source[contextSymbol];
+    }
+
+    async select(keys: Partial<T>, loadChangeEntry = false) {
+        const q = this.context.driver.selectQueryWithKeys(this.model, keys);
+        const r = await this.context.connection.executeQuery(q);
+        const result = r.rows[0];
+        if (loadChangeEntry) {
+            const ce = this.context.changeSet.getEntry(result);
+            ce.apply(result);
+            return ce.entity;
+        }
+        return result;
+    }
+
+    async insert(entity: Partial<T>, loadChangeEntry = false) {
+        const q = this.context.driver.insertQuery(this.model, entity);
+        const r = await this.context.connection.executeQuery(q);
+        const result = r.rows[0];
+        if (loadChangeEntry) {
+            const ce = this.context.changeSet.getEntry(result);
+            ce.apply(result);
+            return ce.entity;
+        }
+        return r.rows[0];
+    }
+
+    async update(entity: Partial<T>, loadChangeEntry = false) {
+        const q = this.context.driver.updateQuery(this.model, entity);
+        const r = await this.context.connection.executeQuery(q);
+        const result = r.rows?.[0];
+        if (loadChangeEntry) {
+            const ce = this.context.changeSet.getEntry(result);
+            ce.apply(result ?? {});
+            return ce.entity;
+        }
+        return r.rows[0];
+    }
+
+    async selectOrInsert(entity: Partial<T>, retry = 3) {
+        const tx = this.context.connection.currentTransaction;
+        let tid: string;
+        if (tx) {
+            tid = `txp_${Date.now()}`;
+            await tx.save(tid);
+        }
+
+        try {
+            const r = await this.select(entity);
+            if (r) {
+                return r;
+            }
+            return await this.insert(entity);
+        } catch (error) {
+            retry --;
+            if(retry > 0) {
+                if (tid) {
+                    await tx.rollbackTo(tid);
+                }
+                await sleep(300);
+                return await this.selectOrInsert(entity, retry);
+            }
+            throw error;
+        }
+    }
+
+    async upsert(entity: Partial<T>, retry = 3) {
+
+        const tx = this.context.connection.currentTransaction;
+        let tid: string;
+        if (tx) {
+            tid = `txp_${Date.now()}`;
+            await tx.save(tid);
+        }
 
-    public statements = {
+        try {
+            const r = await this.update(entity);
+            if (r) {
+                return r;
+            }
+            return await this.insert(entity);
+        } catch (error) {
+            retry --;
+            if(retry > 0) {
+                if (tid) {
+                    await tx.rollbackTo(tid);
+                }
+                await sleep(300);
+                return await this.upsert(entity, retry);
+            }
+            throw error;
+        }
+    }
+}
+
+export class EntitySource<T = any> {
 
-    };
+    public statements = new EntityStatements<T>(this);
 
     get [modelSymbol]() {
         return this.model;