From e5c12a03bd8a12522353c1f007d40158b08a4c05 Mon Sep 17 00:00:00 2001 From: Eric Doughty-Papassideris Date: Mon, 17 Jun 2024 18:26:50 +0200 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20backend:=20Adding=20atomicCompareAn?= =?UTF-8?q?dSet=20operation=20to=20DB=20connector=20(#525)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../orm/connectors/abstract-connector.ts | 10 ++ .../database/services/orm/connectors/index.ts | 28 ++++ .../orm/connectors/mongodb/mongodb.ts | 59 ++++++++ .../postgres/postgres-query-builder.ts | 141 ++++++++++++------ .../orm/connectors/postgres/postgres.ts | 69 +++++++-- .../services/orm/repository/repository.ts | 36 +++++ .../postgres/postgres-query-builder.test.ts | 29 +++- 7 files changed, 315 insertions(+), 57 deletions(-) diff --git a/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/abstract-connector.ts b/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/abstract-connector.ts index ff6e05c3..51d703be 100644 --- a/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/abstract-connector.ts +++ b/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/abstract-connector.ts @@ -20,6 +20,16 @@ export abstract class AbstractConnector implements abstract upsert(entities: any[], _options: UpsertOptions): Promise; + abstract atomicCompareAndSet( + entity: Entity, + fieldName: keyof Entity, + previousValue: FieldValueType, + newValue: FieldValueType, + ): Promise<{ + didSet: boolean; + currentValue: FieldValueType | null; + }>; + abstract remove(entities: any[]): Promise; abstract find( diff --git a/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/index.ts b/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/index.ts index 8b01704d..3ef487e2 100644 --- a/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/index.ts +++ b/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/index.ts @@ -49,6 +49,34 @@ export interface Connector extends Initializable { */ upsert(entities: any[], _options: UpsertOptions): Promise; + /** + * Updates an entity's field `fieldName` to `newValue`, only if its + * current value in the DB is `previousValue`. Does so atomically + * (for whatever that means for the specific DB type - mostly a + * single query, with no special write concern etc.). + * + * Note that the `currentValue` returned may not be atomic depending + * on the DB type. + * + * A single entity for that primary key must already exist. This is not + * always tested, and a return without error should not be interpreted as + * such a row existing. + * + * @param entity Entity to try to update + * @param fieldName Name of the field to compare and possibly set (the node.js field name) + * @param previousValue Value of that field expected in the DB + * @param newValue New value to assign if the existing value matches `previousValue` + */ + atomicCompareAndSet( + entity: Entity, + fieldName: keyof Entity, + previousValue: FieldValueType | null, + newValue: FieldValueType | null, + ): Promise<{ + didSet: boolean; + currentValue: FieldValueType | null; + }>; + /** * Remove * returns true if the object was removed, false otherwise diff --git a/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/mongodb/mongodb.ts b/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/mongodb/mongodb.ts index 05755448..16d5dc29 100644 --- a/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/mongodb/mongodb.ts +++ b/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/mongodb/mongodb.ts @@ -154,6 +154,65 @@ export class MongoConnector extends AbstractConnector { }); } + async atomicCompareAndSet( + entity: Entity, + fieldName: keyof Entity, + previousValue: FieldValueType, + newValue: FieldValueType, + ): Promise<{ + didSet: boolean; + currentValue: FieldValueType | null; + }> { + const { columnsDefinition, entityDefinition } = getEntityDefinition(entity); + const primaryKey = unwrapPrimarykey(entityDefinition); + const db = await this.getDatabase(); + const collection = db.collection(`${entityDefinition.name}`); + const columnName = (Object.entries(columnsDefinition).filter( + ([, { nodename }]) => nodename === fieldName, + )[0] ?? [])[0]; + const transformValue = (key, value) => + transformValueToDbString(value, columnsDefinition[key].type, { + columns: columnsDefinition[key].options, + secret: this.secret, + disableSalts: true, + column: { key }, + }); + const where: any = {}; + primaryKey.forEach(key => { + where[key] = transformValue(key, entity[columnsDefinition[key].nodename]); + }); + where[columnName] = transformValue(columnName, previousValue); + const set = { [columnName]: newValue === undefined ? null : newValue }; + + const result = await collection.updateOne(where, { $set: set }); + if (result.modifiedCount > 1) + throw new Error( + `Unexpected modified count ${JSON.stringify(result)} on mongo update(${JSON.stringify( + where, + )}, ${JSON.stringify(set)})`, + ); + + let currentValue: FieldValueType; + if (result.modifiedCount > 0) { + currentValue = newValue; + } else { + delete where[columnName]; + const existingItem = await collection.findOne(where, {}); + if (!existingItem) + throw new Error( + `Error setting ${entityDefinition.name}'s ${ + fieldName as string + } atomically, no row matched PK: ${JSON.stringify(where)}`, + ); + currentValue = existingItem[fieldName as string]; + } + + return { + didSet: result.modifiedCount === 1, + currentValue, + }; + } + async remove(entities: any[]): Promise { return new Promise(async resolve => { const promises: Promise[] = []; diff --git a/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/postgres/postgres-query-builder.ts b/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/postgres/postgres-query-builder.ts index cde84651..bae3b044 100644 --- a/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/postgres/postgres-query-builder.ts +++ b/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/postgres/postgres-query-builder.ts @@ -1,5 +1,5 @@ import { comparisonType, FindOptions } from "../../repository/repository"; -import { ObjectType } from "../../types"; +import { ColumnDefinition, ObjectType } from "../../types"; import { getEntityDefinition, unwrapPrimarykey } from "../../utils"; import { PostgresDataTransformer } from "./postgres-data-transform"; import _ from "lodash"; @@ -24,7 +24,7 @@ export class PostgresQueryBuilder { const query = (whereClause: string, orderByClause: string, limit: number, offset: number) => { return `SELECT * FROM "${entityDefinition.name}" ${whereClause?.length ? "WHERE " + whereClause : ""} - ${orderByClause?.length ? "ORDER BY " + orderByClause : ""} + ${orderByClause?.length ? "ORDER BY " + orderByClause : ""} LIMIT ${limit} OFFSET ${offset}`; }; @@ -109,22 +109,34 @@ export class PostgresQueryBuilder { return [query(whereClause, orderByClause, limit, offset), values]; } + private toValueKeyDBStringPairFromValue( + columnsDefinition: { [key: string]: ColumnDefinition }, + key: string, + value: any, + ): [string, any] { + return [key, this.dataTransformer.toDbString(value, columnsDefinition[key].type)]; + } + + private toValueKeyDBStringPairFromEntity( + entity: Entity, + columnsDefinition: { [key: string]: ColumnDefinition }, + key: string, + ) { + return this.toValueKeyDBStringPairFromValue( + columnsDefinition, + key, + entity[columnsDefinition[key].nodename], + ); + } + buildDelete(entity: Entity): Query { const { columnsDefinition, entityDefinition } = getEntityDefinition(entity); const primaryKey = unwrapPrimarykey(entityDefinition); - const toValueKeyDBStringPair = (key: string) => { - return [ - key, - this.dataTransformer.toDbString( - entity[columnsDefinition[key].nodename], - columnsDefinition[key].type, - ), - ]; - }; - - const where = primaryKey.map(key => toValueKeyDBStringPair(key)); - const query = `DELETE FROM "${entityDefinition.name}" + const where = primaryKey.map(key => + this.toValueKeyDBStringPairFromEntity(entity, columnsDefinition, key), + ); + const query = `DELETE FROM "${entityDefinition.name}" WHERE ${where.map((e, idx) => `${e[0]} = $${idx + 1}`).join(" AND ")}`; return [query, where.map(f => f[1])]; @@ -132,54 +144,91 @@ export class PostgresQueryBuilder { buildInsert(entity: Entity): Query { const { columnsDefinition, entityDefinition } = getEntityDefinition(entity); - const toValueKeyDBStringPair = (key: string) => { - return [ - key, - this.dataTransformer.toDbString( - entity[columnsDefinition[key].nodename], - columnsDefinition[key].type, - ), - ]; - }; const fields = Object.keys(columnsDefinition) .filter(key => entity[columnsDefinition[key].nodename] !== undefined) - .map(key => toValueKeyDBStringPair(key)); - const query = `INSERT INTO "${entityDefinition.name}" (${fields.map(e => e[0]).join(", ")}) + .map(key => this.toValueKeyDBStringPairFromEntity(entity, columnsDefinition, key)); + const query = `INSERT INTO "${entityDefinition.name}" (${fields.map(e => e[0]).join(", ")}) VALUES (${fields.map((e, idx) => `$${idx + 1}`).join(", ")})`; return [query, fields.map(f => f[1])]; } + private buildWhereClause(startIndex: number, where: [string, any[]][]): Query { + const equalsOrIs = ([key, value]) => + [key, value === null ? "IS" : "=", value === null ? "NULL" : `$${startIndex++}`].join(" "); + return [ + `WHERE ${where.map(equalsOrIs).join(" AND ")} `, + where.filter(f => f[1] !== null).map(f => f[1]), + ]; + } + + private buildUpdateQueryString( + tableName: string, + set: [string, any][], + where: [string, any][], + returning?: string, + ): Query { + // if (!set || set.length == 0) + // throw new Error(`Cannot build UPDATE query with empty SET on ${tableName} with WHERE: ${where}`); + const [whereClause, whereParameters] = this.buildWhereClause(set.length + 1, where); + const query = `UPDATE "${tableName}" + SET ${set.map((e, idx) => `${e[0]} = $${idx + 1}`).join(", ")} + ${whereClause} + ${returning ? "RETURNING " + returning : ""} `; + const values = []; + values.push(...set.map(f => f[1])); + values.push(...whereParameters); + return [query, values]; + } + buildUpdate(entity: Entity): Query { const { columnsDefinition, entityDefinition } = getEntityDefinition(entity); const primaryKey = unwrapPrimarykey(entityDefinition); - const toValueKeyDBStringPair = (key: string) => { - return [ - key, - this.dataTransformer.toDbString( - entity[columnsDefinition[key].nodename], - columnsDefinition[key].type, - ), - ]; - }; - // Set updated content const set = Object.keys(columnsDefinition) .filter(key => primaryKey.indexOf(key) === -1) .filter(key => entity[columnsDefinition[key].nodename] !== undefined) - .map(key => toValueKeyDBStringPair(key)); + .map(key => this.toValueKeyDBStringPairFromEntity(entity, columnsDefinition, key)); //Set primary key - const where = primaryKey.map(key => toValueKeyDBStringPair(key)); - //Start index for where clause params - const whereIdx = set.length + 1; - const query = `UPDATE "${entityDefinition.name}" - SET ${set.map((e, idx) => `${e[0]} = $${idx + 1}`).join(", ")} - WHERE ${where.map((e, idx) => `${e[0]} = $${whereIdx + idx}`).join(" AND ")}`; - const values = []; - values.push(...set.map(f => f[1])); - values.push(...where.map(f => f[1])); + const where = primaryKey.map(key => + this.toValueKeyDBStringPairFromEntity(entity, columnsDefinition, key), + ); + return this.buildUpdateQueryString(entityDefinition.name, set, where); + } - return [query, values]; + buildatomicCompareAndSet( + entity: Entity, + fieldName: keyof Entity, + previousValue: FieldValueType, + newValue: FieldValueType, + ) { + const { columnsDefinition, entityDefinition } = getEntityDefinition(entity); + const primaryKey = unwrapPrimarykey(entityDefinition); + const columnName = (Object.entries(columnsDefinition).filter( + ([, { nodename }]) => nodename === fieldName, + )[0] ?? [])[0]; + if (!columnName) + throw new Error( + `Can't find field ${JSON.stringify(fieldName)} in ${JSON.stringify(columnsDefinition)}`, + ); + const where = primaryKey.map(key => + this.toValueKeyDBStringPairFromEntity(entity, columnsDefinition, key), + ); + const whereQueryWithoutField = this.buildWhereClause(0 + 1, where); + where.push( + this.toValueKeyDBStringPairFromValue(columnsDefinition, columnName as string, previousValue), + ); + return { + updateQuery: this.buildUpdateQueryString( + entityDefinition.name, + [this.toValueKeyDBStringPairFromValue(columnsDefinition, columnName as string, newValue)], + where, + ), + getValueQuery: [ + `SELECT "${columnName}" FROM "${entityDefinition.name}" ${whereQueryWithoutField[0]}`, + whereQueryWithoutField[1], + ] as Query, + }; } } diff --git a/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/postgres/postgres.ts b/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/postgres/postgres.ts index 94dcd6a4..16aa2698 100644 --- a/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/postgres/postgres.ts +++ b/tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/postgres/postgres.ts @@ -270,19 +270,70 @@ export class PostgresConnector extends AbstractConnector( + entity: Entity, + fieldName: keyof Entity, + previousValue: FieldValueType, + newValue: FieldValueType, + ): Promise<{ + didSet: boolean; + currentValue: FieldValueType | null; + }> { + const { updateQuery, getValueQuery } = this.queryBuilder.buildatomicCompareAndSet( + entity, + fieldName, + previousValue, + newValue, + ); + const result = await this.executeRaw(updateQuery); + if (!result) + throw new Error( + `Error updating ${JSON.stringify(fieldName)} field of ${JSON.stringify( + entity["id"], + )} from ${JSON.stringify(previousValue)} to ${JSON.stringify( + newValue, + )}. Search logs for 'services.database.orm.postgres - Error with SQL query'`, + ); + if (result.rowCount > 1) + throw new Error( + `Unexpected modified count ${JSON.stringify(result)} on postgres update(${JSON.stringify( + updateQuery, + )})`, + ); + let currentValue = newValue; + if (result.rowCount == 0) { + const readValue = await this.executeRaw(getValueQuery); + if (readValue.rows.length == 0) + throw new Error( + `Error setting ${fieldName as string} atomically, no row matched PK: ${JSON.stringify( + getValueQuery, + )}`, + ); + currentValue = readValue.rows[0][readValue.fields[0].name]; + } + return { + didSet: result.rowCount > 0, + currentValue, + }; + } + + private async executeRaw(query: Query): Promise { + logger.debug(`service.database.orm.postgres - Query: "${query[0]}"`); + try { + return await this.client.query(query[0] as string, query[1] as never[]); + } catch (err) { + logger.error({ err }, `services.database.orm.postgres - Error with SQL query: ${query[0]}`); + return null; } } private async execute(query: Query) { - logger.debug(`service.database.orm.postgres - Query: "${query[0]}"`); - try { - const result: QueryResult = await this.client.query(query[0] as string, query[1] as never[]); - return result.rowCount > 0; - } catch (err) { - logger.error({ err }, `services.database.orm.postgres - Error with SQL query: ${query[0]}`); - return false; - } + const result = await this.executeRaw(query); + return result && result.rowCount > 0; } async getTableDefinition(name: string): Promise { diff --git a/tdrive/backend/node/src/core/platform/services/database/services/orm/repository/repository.ts b/tdrive/backend/node/src/core/platform/services/database/services/orm/repository/repository.ts index 01a3b801..dea74959 100644 --- a/tdrive/backend/node/src/core/platform/services/database/services/orm/repository/repository.ts +++ b/tdrive/backend/node/src/core/platform/services/database/services/orm/repository/repository.ts @@ -96,6 +96,42 @@ export default class Repository { return (await this.find(filters, options, context)).getEntities()[0] || null; } + /** + * Updates an entity's field `fieldName` to `newValue`, only if its + * current value in the DB is `previousValue`. Does so atomically + * (for whatever that means for the specific DB type - mostly a + * single query, with no special write concern etc.). + * + * Note that the `currentValue` returned may not be atomic depending + * on the DB type. + * + * Unlike the identical on the `Connector`, the manager is flushed + * and reset before the operation is ran. + * + * A single entity for that primary key must already exist. This is not + * always tested, and a return without error should not be interpreted as + * such a row existing. + * + * @param entity Entity to try to update + * @param fieldName Name of the field to compare and possibly set (the node.js field name) + * @param previousValue Value of that field expected in the DB + * @param newValue New value to assign if the existing value matches `previousValue` + */ + async atomicCompareAndSet( + entity: EntityType, + fieldName: keyof EntityType, + previousValue: FieldValueType | null, + newValue: FieldValueType | null, + ): Promise<{ + didSet: boolean; + currentValue: FieldValueType | null; + }> { + if (previousValue === newValue) + throw new Error(`Previous and new values are identical: ${JSON.stringify(previousValue)}`); + await this.manager.flush().then(manager => manager.reset()); + return this.connector.atomicCompareAndSet(entity, fieldName, previousValue, newValue); + } + async save(entity: EntityType, _context?: ExecutionContext): Promise { this.manager.persist(entity); return this.manager.flush().then(manager => manager.reset()); diff --git a/tdrive/backend/node/test/unit/core/services/database/services/orm/connectors/postgres/postgres-query-builder.test.ts b/tdrive/backend/node/test/unit/core/services/database/services/orm/connectors/postgres/postgres-query-builder.test.ts index 4e3c0ee3..f72824cb 100644 --- a/tdrive/backend/node/test/unit/core/services/database/services/orm/connectors/postgres/postgres-query-builder.test.ts +++ b/tdrive/backend/node/test/unit/core/services/database/services/orm/connectors/postgres/postgres-query-builder.test.ts @@ -1,7 +1,7 @@ import { PostgresQueryBuilder } from "../../../../../../../../../src/core/platform/services/database/services/orm/connectors/postgres/postgres-query-builder"; -import { expect, jest } from "@jest/globals"; +import { expect, jest, test, describe, beforeEach, afterEach } from "@jest/globals"; import { randomInt, randomUUID } from "crypto"; import { newTestDbEntity, normalizeWhitespace, TestDbEntity } from "./utils"; import { @@ -127,6 +127,32 @@ describe('The PostgresQueryBuilder', () => { assertUpdateQueryParams(entity, query[1] as any[]) }); + test('buildatomicCompareAndSet to value', async () => { + const entity = newTestDbEntity(); + + const newValue = "new-tag-value"; + const [queryText, params] = subj.buildatomicCompareAndSet(entity, "tags", null, newValue); + + expect(normalizeWhitespace(queryText as string)).toBe(`UPDATE "test_table" SET tags = $1 WHERE company_id = $2 AND id = $3 AND tags = $4 RETURNING tags`); + expect(params[0]).toBe(JSON.stringify(newValue)); + expect(params[1]).toBe(entity.company_id); + expect(params[2]).toBe(entity.id); + expect(params[3]).toBe(null); + }); + + test('buildatomicCompareAndSet to null', async () => { + const entity = newTestDbEntity(); + + const previousValue = "new-tag-value"; + const [queryText, params] = subj.buildatomicCompareAndSet(entity, "tags", previousValue, null); + + expect(normalizeWhitespace(queryText as string)).toBe(`UPDATE "test_table" SET tags = $1 WHERE company_id = $2 AND id = $3 AND tags = $4 RETURNING tags`); + expect(params[0]).toBe(null); + expect(params[1]).toBe(entity.company_id); + expect(params[2]).toBe(entity.id); + expect(params[3]).toBe(JSON.stringify(previousValue)); + }); + const assertInsertQueryParams = (actual: TestDbEntity, expected: any[]) => { expect(expected[0]).toBe(actual.company_id); expect(expected[1]).toBe(actual.id); @@ -145,5 +171,4 @@ describe('The PostgresQueryBuilder', () => { expect(expected[5]).toBe(actual.id); } - });