directus

1
import { Action } from '@directus/constants';
2
import { useEnv } from '@directus/env';
3
import { ErrorCode, ForbiddenError, InvalidPayloadError, isDirectusError } from '@directus/errors';
4
import { isSystemCollection } from '@directus/system-data';
5
import type {
6
	Accountability,
7
	Item as AnyItem,
8
	PermissionsAction,
9
	PrimaryKey,
10
	Query,
11
	SchemaOverview,
12
} from '@directus/types';
13
import type Keyv from 'keyv';
14
import type { Knex } from 'knex';
15
import { assign, clone, cloneDeep, omit, pick, without } from 'lodash-es';
16
import { getCache } from '../cache.js';
17
import { translateDatabaseError } from '../database/errors/translate.js';
18
import { getHelpers } from '../database/helpers/index.js';
19
import getDatabase from '../database/index.js';
20
import runAST from '../database/run-ast.js';
21
import emitter from '../emitter.js';
22
import type { AbstractService, AbstractServiceOptions, ActionEventParams, MutationOptions } from '../types/index.js';
23
import getASTFromQuery from '../utils/get-ast-from-query.js';
24
import { shouldClearCache } from '../utils/should-clear-cache.js';
25
import { transaction } from '../utils/transaction.js';
26
import { validateKeys } from '../utils/validate-keys.js';
27
import { AuthorizationService } from './authorization.js';
28
import { PayloadService } from './payload.js';
29

30
const env = useEnv();
31

32
export type QueryOptions = {
33
	stripNonRequested?: boolean;
34
	permissionsAction?: PermissionsAction;
35
	emitEvents?: boolean;
36
};
37

38
export type MutationTracker = {
39
	trackMutations: (count: number) => void;
40
	getCount: () => number;
41
};
42

43
export class ItemsService<Item extends AnyItem = AnyItem> implements AbstractService {
44
	collection: string;
45
	knex: Knex;
46
	accountability: Accountability | null;
47
	eventScope: string;
48
	schema: SchemaOverview;
49
	cache: Keyv<any> | null;
50

51
	constructor(collection: string, options: AbstractServiceOptions) {
52
		this.collection = collection;
53
		this.knex = options.knex || getDatabase();
54
		this.accountability = options.accountability || null;
55
		this.eventScope = isSystemCollection(this.collection) ? this.collection.substring(9) : 'items';
56
		this.schema = options.schema;
57
		this.cache = getCache().cache;
58

59
		return this;
60
	}
61

62
	/**
63
	 * Create a fork of the current service, allowing instantiation with different options.
64
	 */
65
	private fork(options?: Partial<AbstractServiceOptions>): ItemsService<AnyItem> {
66
		const Service = this.constructor;
67

68
		// ItemsService expects `collection` and `options` as parameters,
69
		// while the other services only expect `options`
70
		const isItemsService = Service.length === 2;
71

72
		const newOptions = { knex: this.knex, accountability: this.accountability, schema: this.schema, ...options };
73

74
		if (isItemsService) {
75
			return new ItemsService(this.collection, newOptions);
76
		}
77

78
		return new (Service as new (options: AbstractServiceOptions) => this)(newOptions);
79
	}
80

81
	createMutationTracker(initialCount = 0): MutationTracker {
82
		const maxCount = Number(env['MAX_BATCH_MUTATION']);
83
		let mutationCount = initialCount;
84
		return {
85
			trackMutations(count: number) {
86
				mutationCount += count;
87

88
				if (mutationCount > maxCount) {
89
					throw new InvalidPayloadError({ reason: `Exceeded max batch mutation limit of ${maxCount}` });
90
				}
91
			},
92
			getCount() {
93
				return mutationCount;
94
			},
95
		};
96
	}
97

98
	async getKeysByQuery(query: Query): Promise<PrimaryKey[]> {
99
		const primaryKeyField = this.schema.collections[this.collection]!.primary;
100
		const readQuery = cloneDeep(query);
101
		readQuery.fields = [primaryKeyField];
102

103
		// Allow unauthenticated access
104
		const itemsService = new ItemsService(this.collection, {
105
			knex: this.knex,
106
			schema: this.schema,
107
		});
108

109
		// We read the IDs of the items based on the query, and then run `updateMany`. `updateMany` does it's own
110
		// permissions check for the keys, so we don't have to make this an authenticated read
111
		const items = await itemsService.readByQuery(readQuery);
112
		return items.map((item: AnyItem) => item[primaryKeyField]).filter((pk) => pk);
113
	}
114

115
	/**
116
	 * Create a single new item.
117
	 */
118
	async createOne(data: Partial<Item>, opts: MutationOptions = {}): Promise<PrimaryKey> {
119
		if (!opts.mutationTracker) opts.mutationTracker = this.createMutationTracker();
120

121
		if (!opts.bypassLimits) {
122
			opts.mutationTracker.trackMutations(1);
123
		}
124

125
		const { ActivityService } = await import('./activity.js');
126
		const { RevisionsService } = await import('./revisions.js');
127

128
		const primaryKeyField = this.schema.collections[this.collection]!.primary;
129
		const fields = Object.keys(this.schema.collections[this.collection]!.fields);
130

131
		const aliases = Object.values(this.schema.collections[this.collection]!.fields)
132
			.filter((field) => field.alias === true)
133
			.map((field) => field.field);
134

135
		const payload: AnyItem = cloneDeep(data);
136
		const nestedActionEvents: ActionEventParams[] = [];
137

138
		// By wrapping the logic in a transaction, we make sure we automatically roll back all the
139
		// changes in the DB if any of the parts contained within throws an error. This also means
140
		// that any errors thrown in any nested relational changes will bubble up and cancel the whole
141
		// update tree
142
		const primaryKey: PrimaryKey = await transaction(this.knex, async (trx) => {
143
			// We're creating new services instances so they can use the transaction as their Knex interface
144
			const payloadService = new PayloadService(this.collection, {
145
				accountability: this.accountability,
146
				knex: trx,
147
				schema: this.schema,
148
			});
149

150
			const authorizationService = new AuthorizationService({
151
				accountability: this.accountability,
152
				knex: trx,
153
				schema: this.schema,
154
			});
155

156
			// Run all hooks that are attached to this event so the end user has the chance to augment the
157
			// item that is about to be saved
158
			const payloadAfterHooks =
159
				opts.emitEvents !== false
160
					? await emitter.emitFilter(
161
							this.eventScope === 'items'
162
								? ['items.create', `${this.collection}.items.create`]
163
								: `${this.eventScope}.create`,
164
							payload,
165
							{
166
								collection: this.collection,
167
							},
168
							{
169
								database: trx,
170
								schema: this.schema,
171
								accountability: this.accountability,
172
							},
173
					  )
174
					: payload;
175

176
			const payloadWithPresets = this.accountability
177
				? authorizationService.validatePayload('create', this.collection, payloadAfterHooks)
178
				: payloadAfterHooks;
179

180
			if (opts.preMutationError) {
181
				throw opts.preMutationError;
182
			}
183

184
			const {
185
				payload: payloadWithM2O,
186
				revisions: revisionsM2O,
187
				nestedActionEvents: nestedActionEventsM2O,
188
			} = await payloadService.processM2O(payloadWithPresets, opts);
189

190
			const {
191
				payload: payloadWithA2O,
192
				revisions: revisionsA2O,
193
				nestedActionEvents: nestedActionEventsA2O,
194
			} = await payloadService.processA2O(payloadWithM2O, opts);
195

196
			const payloadWithoutAliases = pick(payloadWithA2O, without(fields, ...aliases));
197
			const payloadWithTypeCasting = await payloadService.processValues('create', payloadWithoutAliases);
198

199
			// The primary key can already exist in the payload.
200
			// In case of manual string / UUID primary keys it's always provided at this point.
201
			// In case of an (big) integer primary key, it might be provided as the user can specify the value manually.
202
			let primaryKey: undefined | PrimaryKey = payloadWithTypeCasting[primaryKeyField];
203

204
			if (primaryKey) {
205
				validateKeys(this.schema, this.collection, primaryKeyField, primaryKey);
206
			}
207

208
			// If a PK of type number was provided, although the PK is set the auto_increment,
209
			// depending on the database, the sequence might need to be reset to protect future PK collisions.
210
			let autoIncrementSequenceNeedsToBeReset = false;
211

212
			const pkField = this.schema.collections[this.collection]!.fields[primaryKeyField];
213

214
			if (
215
				primaryKey &&
216
				pkField &&
217
				!opts.bypassAutoIncrementSequenceReset &&
218
				['integer', 'bigInteger'].includes(pkField.type) &&
219
				pkField.defaultValue === 'AUTO_INCREMENT'
220
			) {
221
				autoIncrementSequenceNeedsToBeReset = true;
222
			}
223

224
			try {
225
				const result = await trx
226
					.insert(payloadWithoutAliases)
227
					.into(this.collection)
228
					.returning(primaryKeyField)
229
					.then((result) => result[0]);
230

231
				const returnedKey = typeof result === 'object' ? result[primaryKeyField] : result;
232

233
				if (pkField!.type === 'uuid') {
234
					primaryKey = getHelpers(trx).schema.formatUUID(primaryKey ?? returnedKey);
235
				} else {
236
					primaryKey = primaryKey ?? returnedKey;
237
				}
238
			} catch (err: any) {
239
				const dbError = await translateDatabaseError(err);
240

241
				if (isDirectusError(dbError, ErrorCode.RecordNotUnique) && dbError.extensions.primaryKey) {
242
					// This is a MySQL specific thing we need to handle here, since MySQL does not return the field name
243
					// if the unique constraint is the primary key
244
					dbError.extensions.field = pkField?.field ?? null;
245

246
					delete dbError.extensions.primaryKey;
247
				}
248

249
				throw dbError;
250
			}
251

252
			// Most database support returning, those who don't tend to return the PK anyways
253
			// (MySQL/SQLite). In case the primary key isn't know yet, we'll do a best-attempt at
254
			// fetching it based on the last inserted row
255
			if (!primaryKey) {
256
				// Fetching it with max should be safe, as we're in the context of the current transaction
257
				const result = await trx.max(primaryKeyField, { as: 'id' }).from(this.collection).first();
258
				primaryKey = result.id;
259
				// Set the primary key on the input item, in order for the "after" event hook to be able
260
				// to read from it
261
				payload[primaryKeyField] = primaryKey;
262
			}
263

264
			// At this point, the primary key is guaranteed to be set.
265
			primaryKey = primaryKey as PrimaryKey;
266

267
			const { revisions: revisionsO2M, nestedActionEvents: nestedActionEventsO2M } = await payloadService.processO2M(
268
				payloadWithPresets,
269
				primaryKey,
270
				opts,
271
			);
272

273
			nestedActionEvents.push(...nestedActionEventsM2O);
274
			nestedActionEvents.push(...nestedActionEventsA2O);
275
			nestedActionEvents.push(...nestedActionEventsO2M);
276

277
			// If this is an authenticated action, and accountability tracking is enabled, save activity row
278
			if (this.accountability && this.schema.collections[this.collection]!.accountability !== null) {
279
				const activityService = new ActivityService({
280
					knex: trx,
281
					schema: this.schema,
282
				});
283

284
				const activity = await activityService.createOne({
285
					action: Action.CREATE,
286
					user: this.accountability!.user,
287
					collection: this.collection,
288
					ip: this.accountability!.ip,
289
					user_agent: this.accountability!.userAgent,
290
					origin: this.accountability!.origin,
291
					item: primaryKey,
292
				});
293

294
				// If revisions are tracked, create revisions record
295
				if (this.schema.collections[this.collection]!.accountability === 'all') {
296
					const revisionsService = new RevisionsService({
297
						knex: trx,
298
						schema: this.schema,
299
					});
300

301
					const revisionDelta = await payloadService.prepareDelta(payloadAfterHooks);
302

303
					const revision = await revisionsService.createOne({
304
						activity: activity,
305
						collection: this.collection,
306
						item: primaryKey,
307
						data: revisionDelta,
308
						delta: revisionDelta,
309
					});
310

311
					// Make sure to set the parent field of the child-revision rows
312
					const childrenRevisions = [...revisionsM2O, ...revisionsA2O, ...revisionsO2M];
313

314
					if (childrenRevisions.length > 0) {
315
						await revisionsService.updateMany(childrenRevisions, { parent: revision });
316
					}
317

318
					if (opts.onRevisionCreate) {
319
						opts.onRevisionCreate(revision);
320
					}
321
				}
322
			}
323

324
			if (autoIncrementSequenceNeedsToBeReset) {
325
				await getHelpers(trx).sequence.resetAutoIncrementSequence(this.collection, primaryKeyField);
326
			}
327

328
			return primaryKey;
329
		});
330

331
		if (opts.emitEvents !== false) {
332
			const actionEvent = {
333
				event:
334
					this.eventScope === 'items'
335
						? ['items.create', `${this.collection}.items.create`]
336
						: `${this.eventScope}.create`,
337
				meta: {
338
					payload,
339
					key: primaryKey,
340
					collection: this.collection,
341
				},
342
				context: {
343
					database: getDatabase(),
344
					schema: this.schema,
345
					accountability: this.accountability,
346
				},
347
			};
348

349
			if (opts.bypassEmitAction) {
350
				opts.bypassEmitAction(actionEvent);
351
			} else {
352
				emitter.emitAction(actionEvent.event, actionEvent.meta, actionEvent.context);
353
			}
354

355
			for (const nestedActionEvent of nestedActionEvents) {
356
				if (opts.bypassEmitAction) {
357
					opts.bypassEmitAction(nestedActionEvent);
358
				} else {
359
					emitter.emitAction(nestedActionEvent.event, nestedActionEvent.meta, nestedActionEvent.context);
360
				}
361
			}
362
		}
363

364
		if (shouldClearCache(this.cache, opts, this.collection)) {
365
			await this.cache.clear();
366
		}
367

368
		return primaryKey;
369
	}
370

371
	/**
372
	 * Create multiple new items at once. Inserts all provided records sequentially wrapped in a transaction.
373
	 *
374
	 * Uses `this.createOne` under the hood.
375
	 */
376
	async createMany(data: Partial<Item>[], opts: MutationOptions = {}): Promise<PrimaryKey[]> {
377
		if (!opts.mutationTracker) opts.mutationTracker = this.createMutationTracker();
378

379
		const { primaryKeys, nestedActionEvents } = await transaction(this.knex, async (knex) => {
380
			const service = this.fork({ knex });
381

382
			const primaryKeys: PrimaryKey[] = [];
383
			const nestedActionEvents: ActionEventParams[] = [];
384

385
			const pkField = this.schema.collections[this.collection]!.primary;
386

387
			for (const [index, payload] of data.entries()) {
388
				let bypassAutoIncrementSequenceReset = true;
389

390
				// the auto_increment sequence needs to be reset if the current item contains a manual PK and
391
				// if it's the last item of the batch or if the next item doesn't include a PK and hence one needs to be generated
392
				if (payload[pkField] && (index === data.length - 1 || !data[index + 1]?.[pkField])) {
393
					bypassAutoIncrementSequenceReset = false;
394
				}
395

396
				const primaryKey = await service.createOne(payload, {
397
					...(opts || {}),
398
					autoPurgeCache: false,
399
					bypassEmitAction: (params) => nestedActionEvents.push(params),
400
					mutationTracker: opts.mutationTracker,
401
					bypassAutoIncrementSequenceReset,
402
				});
403

404
				primaryKeys.push(primaryKey);
405
			}
406

407
			return { primaryKeys, nestedActionEvents };
408
		});
409

410
		if (opts.emitEvents !== false) {
411
			for (const nestedActionEvent of nestedActionEvents) {
412
				if (opts.bypassEmitAction) {
413
					opts.bypassEmitAction(nestedActionEvent);
414
				} else {
415
					emitter.emitAction(nestedActionEvent.event, nestedActionEvent.meta, nestedActionEvent.context);
416
				}
417
			}
418
		}
419

420
		if (shouldClearCache(this.cache, opts, this.collection)) {
421
			await this.cache.clear();
422
		}
423

424
		return primaryKeys;
425
	}
426

427
	/**
428
	 * Get items by query.
429
	 */
430
	async readByQuery(query: Query, opts?: QueryOptions): Promise<Item[]> {
431
		const updatedQuery =
432
			opts?.emitEvents !== false
433
				? await emitter.emitFilter(
434
						this.eventScope === 'items'
435
							? ['items.query', `${this.collection}.items.query`]
436
							: `${this.eventScope}.query`,
437
						query,
438
						{
439
							collection: this.collection,
440
						},
441
						{
442
							database: this.knex,
443
							schema: this.schema,
444
							accountability: this.accountability,
445
						},
446
				  )
447
				: query;
448

449
		let ast = await getASTFromQuery(this.collection, updatedQuery, this.schema, {
450
			accountability: this.accountability,
451
			// By setting the permissions action, you can read items using the permissions for another
452
			// operation's permissions. This is used to dynamically check if you have update/delete
453
			// access to (a) certain item(s)
454
			action: opts?.permissionsAction || 'read',
455
			knex: this.knex,
456
		});
457

458
		if (this.accountability && this.accountability.admin !== true) {
459
			const authorizationService = new AuthorizationService({
460
				accountability: this.accountability,
461
				knex: this.knex,
462
				schema: this.schema,
463
			});
464

465
			ast = await authorizationService.processAST(ast, opts?.permissionsAction);
466
		}
467

468
		const records = await runAST(ast, this.schema, {
469
			knex: this.knex,
470
			// GraphQL requires relational keys to be returned regardless
471
			stripNonRequested: opts?.stripNonRequested !== undefined ? opts.stripNonRequested : true,
472
		});
473

474
		if (records === null) {
475
			throw new ForbiddenError();
476
		}
477

478
		const filteredRecords =
479
			opts?.emitEvents !== false
480
				? await emitter.emitFilter(
481
						this.eventScope === 'items' ? ['items.read', `${this.collection}.items.read`] : `${this.eventScope}.read`,
482
						records,
483
						{
484
							query: updatedQuery,
485
							collection: this.collection,
486
						},
487
						{
488
							database: this.knex,
489
							schema: this.schema,
490
							accountability: this.accountability,
491
						},
492
				  )
493
				: records;
494

495
		if (opts?.emitEvents !== false) {
496
			emitter.emitAction(
497
				this.eventScope === 'items' ? ['items.read', `${this.collection}.items.read`] : `${this.eventScope}.read`,
498
				{
499
					payload: filteredRecords,
500
					query: updatedQuery,
501
					collection: this.collection,
502
				},
503
				{
504
					database: this.knex || getDatabase(),
505
					schema: this.schema,
506
					accountability: this.accountability,
507
				},
508
			);
509
		}
510

511
		return filteredRecords as Item[];
512
	}
513

514
	/**
515
	 * Get single item by primary key.
516
	 *
517
	 * Uses `this.readByQuery` under the hood.
518
	 */
519
	async readOne(key: PrimaryKey, query: Query = {}, opts?: QueryOptions): Promise<Item> {
520
		const primaryKeyField = this.schema.collections[this.collection]!.primary;
521
		validateKeys(this.schema, this.collection, primaryKeyField, key);
522

523
		const filterWithKey = assign({}, query.filter, { [primaryKeyField]: { _eq: key } });
524
		const queryWithKey = assign({}, query, { filter: filterWithKey });
525

526
		const results = await this.readByQuery(queryWithKey, opts);
527

528
		if (results.length === 0) {
529
			throw new ForbiddenError();
530
		}
531

532
		return results[0]!;
533
	}
534

535
	/**
536
	 * Get multiple items by primary keys.
537
	 *
538
	 * Uses `this.readByQuery` under the hood.
539
	 */
540
	async readMany(keys: PrimaryKey[], query: Query = {}, opts?: QueryOptions): Promise<Item[]> {
541
		const primaryKeyField = this.schema.collections[this.collection]!.primary;
542
		validateKeys(this.schema, this.collection, primaryKeyField, keys);
543

544
		const filterWithKey = { _and: [{ [primaryKeyField]: { _in: keys } }, query.filter ?? {}] };
545
		const queryWithKey = assign({}, query, { filter: filterWithKey });
546

547
		// Set query limit as the number of keys
548
		if (Array.isArray(keys) && keys.length > 0 && !queryWithKey.limit) {
549
			queryWithKey.limit = keys.length;
550
		}
551

552
		const results = await this.readByQuery(queryWithKey, opts);
553

554
		return results;
555
	}
556

557
	/**
558
	 * Update multiple items by query.
559
	 *
560
	 * Uses `this.updateMany` under the hood.
561
	 */
562
	async updateByQuery(query: Query, data: Partial<Item>, opts?: MutationOptions): Promise<PrimaryKey[]> {
563
		const keys = await this.getKeysByQuery(query);
564

565
		const primaryKeyField = this.schema.collections[this.collection]!.primary;
566
		validateKeys(this.schema, this.collection, primaryKeyField, keys);
567

568
		return keys.length ? await this.updateMany(keys, data, opts) : [];
569
	}
570

571
	/**
572
	 * Update a single item by primary key.
573
	 *
574
	 * Uses `this.updateMany` under the hood.
575
	 */
576
	async updateOne(key: PrimaryKey, data: Partial<Item>, opts?: MutationOptions): Promise<PrimaryKey> {
577
		const primaryKeyField = this.schema.collections[this.collection]!.primary;
578
		validateKeys(this.schema, this.collection, primaryKeyField, key);
579

580
		await this.updateMany([key], data, opts);
581
		return key;
582
	}
583

584
	/**
585
	 * Update multiple items in a single transaction.
586
	 *
587
	 * Uses `this.updateOne` under the hood.
588
	 */
589
	async updateBatch(data: Partial<Item>[], opts: MutationOptions = {}): Promise<PrimaryKey[]> {
590
		if (!Array.isArray(data)) {
591
			throw new InvalidPayloadError({ reason: 'Input should be an array of items' });
592
		}
593

594
		if (!opts.mutationTracker) opts.mutationTracker = this.createMutationTracker();
595

596
		const primaryKeyField = this.schema.collections[this.collection]!.primary;
597

598
		const keys: PrimaryKey[] = [];
599

600
		try {
601
			await transaction(this.knex, async (knex) => {
602
				const service = this.fork({ knex });
603

604
				for (const item of data) {
605
					const primaryKey = item[primaryKeyField];
606
					if (!primaryKey) throw new InvalidPayloadError({ reason: `Item in update misses primary key` });
607

608
					const combinedOpts = Object.assign({ autoPurgeCache: false }, opts);
609
					keys.push(await service.updateOne(primaryKey, omit(item, primaryKeyField), combinedOpts));
610
				}
611
			});
612
		} finally {
613
			if (shouldClearCache(this.cache, opts, this.collection)) {
614
				await this.cache.clear();
615
			}
616
		}
617

618
		return keys;
619
	}
620

621
	/**
622
	 * Update many items by primary key, setting all items to the same change.
623
	 */
624
	async updateMany(keys: PrimaryKey[], data: Partial<Item>, opts: MutationOptions = {}): Promise<PrimaryKey[]> {
625
		if (!opts.mutationTracker) opts.mutationTracker = this.createMutationTracker();
626

627
		if (!opts.bypassLimits) {
628
			opts.mutationTracker.trackMutations(keys.length);
629
		}
630

631
		const { ActivityService } = await import('./activity.js');
632
		const { RevisionsService } = await import('./revisions.js');
633

634
		const primaryKeyField = this.schema.collections[this.collection]!.primary;
635
		validateKeys(this.schema, this.collection, primaryKeyField, keys);
636

637
		const fields = Object.keys(this.schema.collections[this.collection]!.fields);
638

639
		const aliases = Object.values(this.schema.collections[this.collection]!.fields)
640
			.filter((field) => field.alias === true)
641
			.map((field) => field.field);
642

643
		const payload: Partial<AnyItem> = cloneDeep(data);
644
		const nestedActionEvents: ActionEventParams[] = [];
645

646
		const authorizationService = new AuthorizationService({
647
			accountability: this.accountability,
648
			knex: this.knex,
649
			schema: this.schema,
650
		});
651

652
		// Run all hooks that are attached to this event so the end user has the chance to augment the
653
		// item that is about to be saved
654
		const payloadAfterHooks =
655
			opts.emitEvents !== false
656
				? await emitter.emitFilter(
657
						this.eventScope === 'items'
658
							? ['items.update', `${this.collection}.items.update`]
659
							: `${this.eventScope}.update`,
660
						payload,
661
						{
662
							keys,
663
							collection: this.collection,
664
						},
665
						{
666
							database: this.knex,
667
							schema: this.schema,
668
							accountability: this.accountability,
669
						},
670
				  )
671
				: payload;
672

673
		// Sort keys to ensure that the order is maintained
674
		keys.sort();
675

676
		if (this.accountability) {
677
			await authorizationService.checkAccess('update', this.collection, keys);
678
		}
679

680
		const payloadWithPresets = this.accountability
681
			? authorizationService.validatePayload('update', this.collection, payloadAfterHooks)
682
			: payloadAfterHooks;
683

684
		if (opts.preMutationError) {
685
			throw opts.preMutationError;
686
		}
687

688
		await transaction(this.knex, async (trx) => {
689
			const payloadService = new PayloadService(this.collection, {
690
				accountability: this.accountability,
691
				knex: trx,
692
				schema: this.schema,
693
			});
694

695
			const {
696
				payload: payloadWithM2O,
697
				revisions: revisionsM2O,
698
				nestedActionEvents: nestedActionEventsM2O,
699
			} = await payloadService.processM2O(payloadWithPresets, opts);
700

701
			const {
702
				payload: payloadWithA2O,
703
				revisions: revisionsA2O,
704
				nestedActionEvents: nestedActionEventsA2O,
705
			} = await payloadService.processA2O(payloadWithM2O, opts);
706

707
			const payloadWithoutAliasAndPK = pick(payloadWithA2O, without(fields, primaryKeyField, ...aliases));
708
			const payloadWithTypeCasting = await payloadService.processValues('update', payloadWithoutAliasAndPK);
709

710
			if (Object.keys(payloadWithTypeCasting).length > 0) {
711
				try {
712
					await trx(this.collection).update(payloadWithTypeCasting).whereIn(primaryKeyField, keys);
713
				} catch (err: any) {
714
					throw await translateDatabaseError(err);
715
				}
716
			}
717

718
			const childrenRevisions = [...revisionsM2O, ...revisionsA2O];
719

720
			nestedActionEvents.push(...nestedActionEventsM2O);
721
			nestedActionEvents.push(...nestedActionEventsA2O);
722

723
			for (const key of keys) {
724
				const { revisions, nestedActionEvents: nestedActionEventsO2M } = await payloadService.processO2M(
725
					payloadWithA2O,
726
					key,
727
					opts,
728
				);
729

730
				childrenRevisions.push(...revisions);
731
				nestedActionEvents.push(...nestedActionEventsO2M);
732
			}
733

734
			// If this is an authenticated action, and accountability tracking is enabled, save activity row
735
			if (this.accountability && this.schema.collections[this.collection]!.accountability !== null) {
736
				const activityService = new ActivityService({
737
					knex: trx,
738
					schema: this.schema,
739
				});
740

741
				const activity = await activityService.createMany(
742
					keys.map((key) => ({
743
						action: Action.UPDATE,
744
						user: this.accountability!.user,
745
						collection: this.collection,
746
						ip: this.accountability!.ip,
747
						user_agent: this.accountability!.userAgent,
748
						origin: this.accountability!.origin,
749
						item: key,
750
					})),
751
					{ bypassLimits: true },
752
				);
753

754
				if (this.schema.collections[this.collection]!.accountability === 'all') {
755
					const itemsService = new ItemsService(this.collection, {
756
						knex: trx,
757
						schema: this.schema,
758
					});
759

760
					const snapshots = await itemsService.readMany(keys);
761

762
					const revisionsService = new RevisionsService({
763
						knex: trx,
764
						schema: this.schema,
765
					});
766

767
					const revisions = (
768
						await Promise.all(
769
							activity.map(async (activity, index) => ({
770
								activity: activity,
771
								collection: this.collection,
772
								item: keys[index],
773
								data:
774
									snapshots && Array.isArray(snapshots) ? JSON.stringify(snapshots[index]) : JSON.stringify(snapshots),
775
								delta: await payloadService.prepareDelta(payloadWithTypeCasting),
776
							})),
777
						)
778
					).filter((revision) => revision.delta);
779

780
					const revisionIDs = await revisionsService.createMany(revisions);
781

782
					for (let i = 0; i < revisionIDs.length; i++) {
783
						const revisionID = revisionIDs[i]!;
784

785
						if (opts.onRevisionCreate) {
786
							opts.onRevisionCreate(revisionID);
787
						}
788

789
						if (i === 0) {
790
							// In case of a nested relational creation/update in a updateMany, the nested m2o/a2o
791
							// creation is only done once. We treat the first updated item as the "main" update,
792
							// with all other revisions on the current level as regular "flat" updates, and
793
							// nested revisions as children of this first "root" item.
794
							if (childrenRevisions.length > 0) {
795
								await revisionsService.updateMany(childrenRevisions, { parent: revisionID });
796
							}
797
						}
798
					}
799
				}
800
			}
801
		});
802

803
		if (shouldClearCache(this.cache, opts, this.collection)) {
804
			await this.cache.clear();
805
		}
806

807
		if (opts.emitEvents !== false) {
808
			const actionEvent = {
809
				event:
810
					this.eventScope === 'items'
811
						? ['items.update', `${this.collection}.items.update`]
812
						: `${this.eventScope}.update`,
813
				meta: {
814
					payload: payloadWithPresets,
815
					keys,
816
					collection: this.collection,
817
				},
818
				context: {
819
					database: getDatabase(),
820
					schema: this.schema,
821
					accountability: this.accountability,
822
				},
823
			};
824

825
			if (opts.bypassEmitAction) {
826
				opts.bypassEmitAction(actionEvent);
827
			} else {
828
				emitter.emitAction(actionEvent.event, actionEvent.meta, actionEvent.context);
829
			}
830

831
			for (const nestedActionEvent of nestedActionEvents) {
832
				if (opts.bypassEmitAction) {
833
					opts.bypassEmitAction(nestedActionEvent);
834
				} else {
835
					emitter.emitAction(nestedActionEvent.event, nestedActionEvent.meta, nestedActionEvent.context);
836
				}
837
			}
838
		}
839

840
		return keys;
841
	}
842

843
	/**
844
	 * Upsert a single item.
845
	 *
846
	 * Uses `this.createOne` / `this.updateOne` under the hood.
847
	 */
848
	async upsertOne(payload: Partial<Item>, opts?: MutationOptions): Promise<PrimaryKey> {
849
		const primaryKeyField = this.schema.collections[this.collection]!.primary;
850
		const primaryKey: PrimaryKey | undefined = payload[primaryKeyField];
851

852
		if (primaryKey) {
853
			validateKeys(this.schema, this.collection, primaryKeyField, primaryKey);
854
		}
855

856
		const exists =
857
			primaryKey &&
858
			!!(await this.knex
859
				.select(primaryKeyField)
860
				.from(this.collection)
861
				.where({ [primaryKeyField]: primaryKey })
862
				.first());
863

864
		if (exists) {
865
			return await this.updateOne(primaryKey as PrimaryKey, payload, opts);
866
		} else {
867
			return await this.createOne(payload, opts);
868
		}
869
	}
870

871
	/**
872
	 * Upsert many items.
873
	 *
874
	 * Uses `this.upsertOne` under the hood.
875
	 */
876
	async upsertMany(payloads: Partial<Item>[], opts: MutationOptions = {}): Promise<PrimaryKey[]> {
877
		if (!opts.mutationTracker) opts.mutationTracker = this.createMutationTracker();
878

879
		const primaryKeys = await transaction(this.knex, async (knex) => {
880
			const service = this.fork({ knex });
881

882
			const primaryKeys: PrimaryKey[] = [];
883

884
			for (const payload of payloads) {
885
				const primaryKey = await service.upsertOne(payload, { ...(opts || {}), autoPurgeCache: false });
886
				primaryKeys.push(primaryKey);
887
			}
888

889
			return primaryKeys;
890
		});
891

892
		if (shouldClearCache(this.cache, opts, this.collection)) {
893
			await this.cache.clear();
894
		}
895

896
		return primaryKeys;
897
	}
898

899
	/**
900
	 * Delete multiple items by query.
901
	 *
902
	 * Uses `this.deleteMany` under the hood.
903
	 */
904
	async deleteByQuery(query: Query, opts?: MutationOptions): Promise<PrimaryKey[]> {
905
		const keys = await this.getKeysByQuery(query);
906

907
		const primaryKeyField = this.schema.collections[this.collection]!.primary;
908
		validateKeys(this.schema, this.collection, primaryKeyField, keys);
909

910
		return keys.length ? await this.deleteMany(keys, opts) : [];
911
	}
912

913
	/**
914
	 * Delete a single item by primary key.
915
	 *
916
	 * Uses `this.deleteMany` under the hood.
917
	 */
918
	async deleteOne(key: PrimaryKey, opts?: MutationOptions): Promise<PrimaryKey> {
919
		const primaryKeyField = this.schema.collections[this.collection]!.primary;
920
		validateKeys(this.schema, this.collection, primaryKeyField, key);
921

922
		await this.deleteMany([key], opts);
923
		return key;
924
	}
925

926
	/**
927
	 * Delete multiple items by primary key.
928
	 */
929
	async deleteMany(keys: PrimaryKey[], opts: MutationOptions = {}): Promise<PrimaryKey[]> {
930
		if (!opts.mutationTracker) opts.mutationTracker = this.createMutationTracker();
931

932
		if (!opts.bypassLimits) {
933
			opts.mutationTracker.trackMutations(keys.length);
934
		}
935

936
		const { ActivityService } = await import('./activity.js');
937

938
		const primaryKeyField = this.schema.collections[this.collection]!.primary;
939
		validateKeys(this.schema, this.collection, primaryKeyField, keys);
940

941
		if (this.accountability && this.accountability.admin !== true) {
942
			const authorizationService = new AuthorizationService({
943
				accountability: this.accountability,
944
				schema: this.schema,
945
				knex: this.knex,
946
			});
947

948
			await authorizationService.checkAccess('delete', this.collection, keys);
949
		}
950

951
		if (opts.preMutationError) {
952
			throw opts.preMutationError;
953
		}
954

955
		if (opts.emitEvents !== false) {
956
			await emitter.emitFilter(
957
				this.eventScope === 'items' ? ['items.delete', `${this.collection}.items.delete`] : `${this.eventScope}.delete`,
958
				keys,
959
				{
960
					collection: this.collection,
961
				},
962
				{
963
					database: this.knex,
964
					schema: this.schema,
965
					accountability: this.accountability,
966
				},
967
			);
968
		}
969

970
		await transaction(this.knex, async (trx) => {
971
			await trx(this.collection).whereIn(primaryKeyField, keys).delete();
972

973
			if (this.accountability && this.schema.collections[this.collection]!.accountability !== null) {
974
				const activityService = new ActivityService({
975
					knex: trx,
976
					schema: this.schema,
977
				});
978

979
				await activityService.createMany(
980
					keys.map((key) => ({
981
						action: Action.DELETE,
982
						user: this.accountability!.user,
983
						collection: this.collection,
984
						ip: this.accountability!.ip,
985
						user_agent: this.accountability!.userAgent,
986
						origin: this.accountability!.origin,
987
						item: key,
988
					})),
989
					{ bypassLimits: true },
990
				);
991
			}
992
		});
993

994
		if (shouldClearCache(this.cache, opts, this.collection)) {
995
			await this.cache.clear();
996
		}
997

998
		if (opts.emitEvents !== false) {
999
			const actionEvent = {
1000
				event:
1001
					this.eventScope === 'items'
1002
						? ['items.delete', `${this.collection}.items.delete`]
1003
						: `${this.eventScope}.delete`,
1004
				meta: {
1005
					payload: keys,
1006
					keys: keys,
1007
					collection: this.collection,
1008
				},
1009
				context: {
1010
					database: getDatabase(),
1011
					schema: this.schema,
1012
					accountability: this.accountability,
1013
				},
1014
			};
1015

1016
			if (opts.bypassEmitAction) {
1017
				opts.bypassEmitAction(actionEvent);
1018
			} else {
1019
				emitter.emitAction(actionEvent.event, actionEvent.meta, actionEvent.context);
1020
			}
1021
		}
1022

1023
		return keys;
1024
	}
1025

1026
	/**
1027
	 * Read/treat collection as singleton.
1028
	 */
1029
	async readSingleton(query: Query, opts?: QueryOptions): Promise<Partial<Item>> {
1030
		query = clone(query);
1031

1032
		query.limit = 1;
1033

1034
		const records = await this.readByQuery(query, opts);
1035
		const record = records[0];
1036

1037
		if (!record) {
1038
			let fields = Object.entries(this.schema.collections[this.collection]!.fields);
1039
			const defaults: Record<string, any> = {};
1040

1041
			if (query.fields && query.fields.includes('*') === false) {
1042
				fields = fields.filter(([name]) => {
1043
					return query.fields!.includes(name);
1044
				});
1045
			}
1046

1047
			for (const [name, field] of fields) {
1048
				if (this.schema.collections[this.collection]!.primary === name) {
1049
					defaults[name] = null;
1050
					continue;
1051
				}
1052

1053
				if (field.defaultValue !== null) defaults[name] = field.defaultValue;
1054
			}
1055

1056
			return defaults as Partial<Item>;
1057
		}
1058

1059
		return record;
1060
	}
1061

1062
	/**
1063
	 * Upsert/treat collection as singleton.
1064
	 *
1065
	 * Uses `this.createOne` / `this.updateOne` under the hood.
1066
	 */
1067
	async upsertSingleton(data: Partial<Item>, opts?: MutationOptions): Promise<PrimaryKey> {
1068
		const primaryKeyField = this.schema.collections[this.collection]!.primary;
1069

1070
		const record = await this.knex.select(primaryKeyField).from(this.collection).limit(1).first();
1071

1072
		if (record) {
1073
			return await this.updateOne(record[primaryKeyField], data, opts);
1074
		}
1075

1076
		return await this.createOne(data, opts);
1077
	}
1078
}
1079