-
-
Notifications
You must be signed in to change notification settings - Fork 496
/
ChangeSetPersister.ts
126 lines (104 loc) · 5.74 KB
/
ChangeSetPersister.ts
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
import { MetadataStorage } from '../metadata';
import { AnyEntity, Dictionary, EntityData, EntityMetadata, EntityProperty, FilterQuery, IPrimaryKey } from '../typings';
import { EntityIdentifier } from '../entity';
import { ChangeSet, ChangeSetType } from './ChangeSet';
import { QueryResult, Transaction } from '../connections';
import { Utils, ValidationError } from '../utils';
import { IDatabaseDriver } from '../drivers';
import { Hydrator } from '../hydration';
export class ChangeSetPersister {
constructor(private readonly driver: IDatabaseDriver,
private readonly identifierMap: Map<string, EntityIdentifier>,
private readonly metadata: MetadataStorage,
private readonly hydrator: Hydrator) { }
async persistToDatabase<T extends AnyEntity<T>>(changeSet: ChangeSet<T>, ctx?: Transaction): Promise<void> {
const meta = this.metadata.find(changeSet.name)!;
// process references first
for (const prop of Object.values(meta.properties)) {
this.processReference(changeSet, prop);
}
// persist the entity itself
await this.persistEntity(changeSet, meta, ctx);
}
private async persistEntity<T extends AnyEntity<T>>(changeSet: ChangeSet<T>, meta: EntityMetadata<T>, ctx?: Transaction): Promise<void> {
let res: QueryResult | undefined;
const wrapped = changeSet.entity.__helper!;
if (changeSet.type === ChangeSetType.DELETE) {
await this.driver.nativeDelete(changeSet.name, wrapped.__primaryKey as Dictionary, ctx);
} else if (changeSet.type === ChangeSetType.UPDATE) {
res = await this.updateEntity(meta, changeSet, ctx);
this.mapReturnedValues(changeSet.entity, res, meta);
} else if (Utils.isDefined(wrapped.__primaryKey, true)) { // ChangeSetType.CREATE with primary key
res = await this.driver.nativeInsert(changeSet.name, changeSet.payload, ctx);
this.mapReturnedValues(changeSet.entity, res, meta);
wrapped.__initialized = true;
} else { // ChangeSetType.CREATE without primary key
res = await this.driver.nativeInsert(changeSet.name, changeSet.payload, ctx);
this.mapReturnedValues(changeSet.entity, res, meta);
this.mapPrimaryKey(meta, res.insertId, changeSet);
wrapped.__initialized = true;
}
await this.processOptimisticLock(meta, changeSet, res, ctx);
changeSet.persisted = true;
}
private mapPrimaryKey<T extends AnyEntity<T>>(meta: EntityMetadata<T>, value: IPrimaryKey, changeSet: ChangeSet<T>): void {
const prop = meta.properties[meta.primaryKeys[0]];
const insertId = prop.customType ? prop.customType.convertToJSValue(value, this.driver.getPlatform()) : value;
const wrapped = changeSet.entity.__helper!;
wrapped.__primaryKey = Utils.isDefined(wrapped.__primaryKey, true) ? wrapped.__primaryKey : insertId;
this.identifierMap.get(wrapped.__uuid)!.setValue(changeSet.entity[prop.name] as unknown as IPrimaryKey);
}
private async updateEntity<T extends AnyEntity<T>>(meta: EntityMetadata<T>, changeSet: ChangeSet<T>, ctx?: Transaction): Promise<QueryResult> {
if (!meta.versionProperty || !changeSet.entity[meta.versionProperty]) {
return this.driver.nativeUpdate(changeSet.name, changeSet.entity.__helper!.__primaryKey as Dictionary, changeSet.payload, ctx);
}
const cond = {
...Utils.getPrimaryKeyCond<T>(changeSet.entity, meta.primaryKeys),
[meta.versionProperty]: changeSet.entity[meta.versionProperty],
} as FilterQuery<T>;
return this.driver.nativeUpdate(changeSet.name, cond, changeSet.payload, ctx);
}
private async processOptimisticLock<T extends AnyEntity<T>>(meta: EntityMetadata<T>, changeSet: ChangeSet<T>, res: QueryResult | undefined, ctx?: Transaction) {
if (meta.versionProperty && changeSet.type === ChangeSetType.UPDATE && res && !res.affectedRows) {
throw ValidationError.lockFailed(changeSet.entity);
}
if (meta.versionProperty && [ChangeSetType.CREATE, ChangeSetType.UPDATE].includes(changeSet.type)) {
const e = await this.driver.findOne<T>(meta.name, changeSet.entity.__helper!.__primaryKey, {
populate: [{
field: meta.versionProperty,
}] as unknown as boolean,
}, ctx);
(changeSet.entity as T)[meta.versionProperty] = e![meta.versionProperty];
}
}
private processReference<T extends AnyEntity<T>>(changeSet: ChangeSet<T>, prop: EntityProperty<T>): void {
const value = changeSet.payload[prop.name];
if (value as unknown instanceof EntityIdentifier) {
changeSet.payload[prop.name] = value.getValue();
}
if (prop.onCreate && changeSet.type === ChangeSetType.CREATE) {
changeSet.entity[prop.name] = changeSet.payload[prop.name] = prop.onCreate(changeSet.entity);
if (prop.primary) {
this.mapPrimaryKey(changeSet.entity.__helper!.__meta, changeSet.entity[prop.name] as unknown as IPrimaryKey, changeSet);
}
}
if (prop.onUpdate && changeSet.type === ChangeSetType.UPDATE) {
changeSet.entity[prop.name] = changeSet.payload[prop.name] = prop.onUpdate(changeSet.entity);
}
}
/**
* Maps values returned via `returning` statement (postgres) or the inserted id (other sql drivers).
* No need to handle composite keys here as they need to be set upfront.
*/
private mapReturnedValues<T extends AnyEntity<T>>(entity: T, res: QueryResult, meta: EntityMetadata<T>): void {
if (res.row && Object.keys(res.row).length > 0) {
const data = Object.values<EntityProperty>(meta.properties).reduce((data, prop) => {
if (prop.fieldNames && res.row![prop.fieldNames[0]] && !Utils.isDefined(entity[prop.name], true)) {
data[prop.name] = res.row![prop.fieldNames[0]];
}
return data;
}, {} as Dictionary);
this.hydrator.hydrate<T>(entity, meta, data as EntityData<T>, false);
}
}
}