All files / src/store document.store.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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
 
 
1x
 
1x
1x
 
1x
1x
1x
1x
 
1x
 
 
1x
 
1x
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
'use strict';
 
import * as _ from 'lodash';
 
import {asyncScheduler} from 'rxjs';
import {filter, throttleTime} from 'rxjs/operators';
 
import AStore from './a.store';
import EStoreType from '../_enums/store.type.enum';
import observableModel from '../mongodb/functions/observable.model';
import getHrtimeAsNumber from '../functions/performance/get.hrtime.as.number';
 
require('json-circular-stringify');
 
// tslint:disable-next-line:variable-name
const _getIdFromQuery = (query: any): string => (_.isString(query) ? query : _.get(query, '_id', '').toString());
 
export default class DocumentStore extends AStore {
	constructor(model: any, target: string) {
		super(model, target);
		this._type = EStoreType.DOCUMENT;
		Object.setPrototypeOf(this, DocumentStore.prototype);
	}
 
	protected extractFromConfig(): void {
		super.extractFromConfig();
 
		const {skip = 0} = this._config;
		this._paging = skip ? {} : {skip, limit: 1};
	}
 
	public restartSubscription(): void {
		this.subscription = observableModel(this.model) //
			.pipe(throttleTime(this._delay, asyncScheduler, {leading: true, trailing: true}))
			.pipe(filter((change) => this._pipeFilter(change)))
			.subscribe({
				next: (change: any): void => {
					this.load(change) //
						.then((): null => null)
						.catch((e: any): void => this.error(e));
				},
				error: (e: any): void => this.error(e),
				complete: (): void => this.complete()
			});
	}
 
	protected shouldReload(change: any): boolean {
		Iif (this.isInitialSubscription(change)) return true;
 
		const id: string = _getIdFromQuery(this._query);
		const {operationType: type, documentKey, updateDescription: description} = change;
		const key = _.get(documentKey, '_id', '').toString();
 
		switch (type) {
			case 'delete':
				return true;
 
			case 'insert':
				return !id;
 
			case 'replace':
			case 'update': {
				Iif (id && id === key) return true;
				Iif (!description) return true;
 
				Iif (!this.shouldConsiderFields()) return true;
 
				const {updatedFields, removedFields} = description;
				const us: any[] = _.concat(removedFields, _.keys(updatedFields));
				const qs: string[] = _.keys(this._fields);
				return !_.isEmpty(_.intersection(qs, us));
			}
		}
 
		return false;
	}
 
	protected async load(change: any): Promise<void> {
		const startTime: number = getHrtimeAsNumber();
 
		Iif (_.isEmpty(this._config)) return this.emitOne(startTime, this._subscriptionId);
		Iif (!this.shouldReload(change)) return;
		// console.log('[@owservable] -> DocumentStore::load', JSON.stringify(change));
 
		const id: string = _getIdFromQuery(this._query);
		const {operationType: type, documentKey} = change;
		const key = _.get(documentKey, '_id', '').toString();
 
		Iif (type === 'delete' && id === key) return this.emitDelete(startTime, this._subscriptionId, key);
 
		// console.log('[@owservable] -> DB Reload Document for query:', this._query);
		try {
			let data;
			if (!_.isEmpty(this._sort)) data = await this._loadSortedFirstDocument();
			else data = id ? await this._loadDocumentById(id) : await this._loadDocument();
 
			Iif (!data) return this.emitOne(startTime, this._subscriptionId);
 
			for (const populate of this._populates) {
				Iif (data?.populate) await data.populate(populate);
			}
 
			Iif (_.isEmpty(this._virtuals)) return this.emitOne(startTime, this._subscriptionId, data.toJSON());
 
			const jsonData: any = _.cloneDeep(_.omit(data.toJSON(), this._virtuals));
			for (const virtual of this._virtuals) {
				jsonData[virtual] = await Promise.resolve(data[virtual]);
			}
			this.emitOne(startTime, this._subscriptionId, jsonData);
		} catch (error) {
			console.error('[@owservable] -> DocumentStore::load Error:', {change, error});
			this.emitError(startTime, this._subscriptionId, error);
		}
	}
 
	private _pipeFilter(change: any): boolean {
		Iif (!_.isEmpty(this._sort)) return true;
 
		const {operationType: type} = change;
		Iif ('delete' === type) return true;
 
		const {documentKey, fullDocument: document} = change;
		const key = _.get(documentKey, '_id', '').toString();
		Iif (key === _getIdFromQuery(this._query)) return true;
 
		return this.testDocument(document);
	}
 
	private async _loadDocumentById(id: string): Promise<any> {
		return this._model.findById(id, this._fields);
	}
 
	private async _loadSortedFirstDocument(): Promise<any> {
		const docs: any = await this._model //
			.find(this._query, this._fields, this._paging)
			// .collation({locale: 'en'})
			.sort(this._sort) // @ts-ignore
			.setOptions({allowDiskUse: true});
		return _.first(docs);
	}
 
	private async _loadDocument(): Promise<any> {
		return this._model.findOne(this._query, this._fields);
	}
}