fix(db operations): Update transaction API to consistently pass optional session parameters across database operations
This commit is contained in:
@@ -11,8 +11,18 @@ import { SmartdataLuceneAdapter } from './classes.lucene.adapter.js';
|
||||
* - validate: post-fetch validator, return true to keep a doc
|
||||
*/
|
||||
export interface SearchOptions<T> {
|
||||
/**
|
||||
* Additional MongoDB filter to AND‐merge into the query
|
||||
*/
|
||||
filter?: Record<string, any>;
|
||||
/**
|
||||
* Post‐fetch validator; return true to keep each doc
|
||||
*/
|
||||
validate?: (doc: T) => Promise<boolean> | boolean;
|
||||
/**
|
||||
* Optional MongoDB session for transactional operations
|
||||
*/
|
||||
session?: plugins.mongodb.ClientSession;
|
||||
}
|
||||
|
||||
export type TDocCreation = 'db' | 'new' | 'mixed';
|
||||
@@ -193,8 +203,13 @@ export class SmartDataDbDoc<T extends TImplements, TImplements, TManager extends
|
||||
public static async getInstances<T>(
|
||||
this: plugins.tsclass.typeFest.Class<T>,
|
||||
filterArg: plugins.tsclass.typeFest.PartialDeep<T>,
|
||||
opts?: { session?: plugins.mongodb.ClientSession }
|
||||
): Promise<T[]> {
|
||||
const foundDocs = await (this as any).collection.findAll(convertFilterForMongoDb(filterArg));
|
||||
// Pass session through to findAll for transactional queries
|
||||
const foundDocs = await (this as any).collection.findAll(
|
||||
convertFilterForMongoDb(filterArg),
|
||||
{ session: opts?.session },
|
||||
);
|
||||
const returnArray = [];
|
||||
for (const foundDoc of foundDocs) {
|
||||
const newInstance: T = (this as any).createInstanceFromMongoDbNativeDoc(foundDoc);
|
||||
@@ -212,8 +227,13 @@ export class SmartDataDbDoc<T extends TImplements, TImplements, TManager extends
|
||||
public static async getInstance<T>(
|
||||
this: plugins.tsclass.typeFest.Class<T>,
|
||||
filterArg: plugins.tsclass.typeFest.PartialDeep<T>,
|
||||
opts?: { session?: plugins.mongodb.ClientSession }
|
||||
): Promise<T> {
|
||||
const foundDoc = await (this as any).collection.findOne(convertFilterForMongoDb(filterArg));
|
||||
// Retrieve one document, with optional session for transactions
|
||||
const foundDoc = await (this as any).collection.findOne(
|
||||
convertFilterForMongoDb(filterArg),
|
||||
{ session: opts?.session },
|
||||
);
|
||||
if (foundDoc) {
|
||||
const newInstance: T = (this as any).createInstanceFromMongoDbNativeDoc(foundDoc);
|
||||
return newInstance;
|
||||
@@ -236,14 +256,19 @@ export class SmartDataDbDoc<T extends TImplements, TImplements, TManager extends
|
||||
* get cursor
|
||||
* @returns
|
||||
*/
|
||||
/**
|
||||
* Get a cursor for streaming results, with optional session
|
||||
*/
|
||||
public static async getCursor<T>(
|
||||
this: plugins.tsclass.typeFest.Class<T>,
|
||||
filterArg: plugins.tsclass.typeFest.PartialDeep<T>,
|
||||
opts?: { session?: plugins.mongodb.ClientSession }
|
||||
) {
|
||||
const collection: SmartdataCollection<T> = (this as any).collection;
|
||||
const cursor: SmartdataDbCursor<T> = await collection.getCursor(
|
||||
convertFilterForMongoDb(filterArg),
|
||||
this as any as typeof SmartDataDbDoc,
|
||||
{ session: opts?.session },
|
||||
);
|
||||
return cursor;
|
||||
}
|
||||
@@ -339,7 +364,9 @@ export class SmartDataDbDoc<T extends TImplements, TImplements, TManager extends
|
||||
if (opts?.filter) {
|
||||
mongoFilter = { $and: [mongoFilter, opts.filter] };
|
||||
}
|
||||
let docs: T[] = await (this as any).getInstances(mongoFilter);
|
||||
// Fetch with optional session for transactions
|
||||
// Fetch within optional session
|
||||
let docs: T[] = await (this as any).getInstances(mongoFilter, { session: opts?.session });
|
||||
if (opts?.validate) {
|
||||
const out: T[] = [];
|
||||
for (const d of docs) {
|
||||
@@ -546,10 +573,9 @@ export class SmartDataDbDoc<T extends TImplements, TImplements, TManager extends
|
||||
constructor() {}
|
||||
|
||||
/**
|
||||
* saves this instance but not any connected items
|
||||
* may lead to data inconsistencies, but is faster
|
||||
* saves this instance (optionally within a transaction)
|
||||
*/
|
||||
public async save() {
|
||||
public async save(opts?: { session?: plugins.mongodb.ClientSession }) {
|
||||
// allow hook before saving
|
||||
if (typeof (this as any).beforeSave === 'function') {
|
||||
await (this as any).beforeSave();
|
||||
@@ -562,10 +588,10 @@ export class SmartDataDbDoc<T extends TImplements, TImplements, TManager extends
|
||||
// perform insert or update
|
||||
switch (this.creationStatus) {
|
||||
case 'db':
|
||||
dbResult = await this.collection.update(self);
|
||||
dbResult = await this.collection.update(self, { session: opts?.session });
|
||||
break;
|
||||
case 'new':
|
||||
dbResult = await this.collection.insert(self);
|
||||
dbResult = await this.collection.insert(self, { session: opts?.session });
|
||||
this.creationStatus = 'db';
|
||||
break;
|
||||
default:
|
||||
@@ -579,15 +605,15 @@ export class SmartDataDbDoc<T extends TImplements, TImplements, TManager extends
|
||||
}
|
||||
|
||||
/**
|
||||
* deletes a document from the database
|
||||
* deletes a document from the database (optionally within a transaction)
|
||||
*/
|
||||
public async delete() {
|
||||
public async delete(opts?: { session?: plugins.mongodb.ClientSession }) {
|
||||
// allow hook before deleting
|
||||
if (typeof (this as any).beforeDelete === 'function') {
|
||||
await (this as any).beforeDelete();
|
||||
}
|
||||
// perform deletion
|
||||
const result = await this.collection.delete(this);
|
||||
const result = await this.collection.delete(this, { session: opts?.session });
|
||||
// allow hook after delete
|
||||
if (typeof (this as any).afterDelete === 'function') {
|
||||
await (this as any).afterDelete();
|
||||
|
Reference in New Issue
Block a user