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
147
148
149
150
151
|
import "./MongoBigInt";
import mongoose, { Collection, Connection, LeanDocument } from "mongoose";
import { ChangeStream, ChangeEvent, Long } from "mongodb";
import EventEmitter from "events";
const uri = process.env.MONGO_URL || "mongodb://localhost:27017/fosscord?readPreference=secondaryPreferred";
import { URL } from "url";
const url = new URL(uri.replace("mongodb://", "http://"));
const connection = mongoose.createConnection(uri, {
autoIndex: true,
useNewUrlParser: true,
useUnifiedTopology: true,
useFindAndModify: false,
});
console.log(`[Database] connect: mongodb://${url.username}@${url.host}${url.pathname}${url.search}`);
export default <Connection>connection;
function transform<T>(document: T) {
// @ts-ignore
if (!document || !document.toObject) {
try {
// @ts-ignore
delete document._id;
// @ts-ignore
delete document.__v;
} catch (error) {}
return document;
}
// @ts-ignore
return document.toObject({ virtuals: true });
}
export function toObject<T>(document: T): LeanDocument<T> {
// @ts-ignore
return Array.isArray(document) ? document.map((x) => transform<T>(x)) : transform(document);
}
export interface MongooseCache {
on(event: "delete", listener: (id: string) => void): this;
on(event: "change", listener: (data: any) => void): this;
on(event: "insert", listener: (data: any) => void): this;
on(event: "close", listener: () => void): this;
}
export class MongooseCache extends EventEmitter {
public stream: ChangeStream;
public data: any;
public initalizing?: Promise<void>;
constructor(
public collection: Collection,
public pipeline: Array<Record<string, unknown>>,
public opts: {
onlyEvents: boolean;
array?: boolean;
}
) {
super();
if (this.opts.array == null) this.opts.array = true;
}
init = () => {
if (this.initalizing) return this.initalizing;
this.initalizing = new Promise(async (resolve, reject) => {
// @ts-ignore
this.stream = this.collection.watch(this.pipeline, { fullDocument: "updateLookup" });
this.stream.on("change", this.change);
this.stream.on("close", this.destroy);
this.stream.on("error", console.error);
if (!this.opts.onlyEvents) {
const arr = await this.collection.aggregate(this.pipeline).toArray();
if (this.opts.array) this.data = arr || [];
else this.data = arr?.[0];
}
resolve();
});
return this.initalizing;
};
changeStream = (pipeline: any) => {
this.pipeline = pipeline;
this.destroy();
this.init();
};
convertResult = (obj: any) => {
if (obj instanceof Long) return BigInt(obj.toString());
if (typeof obj === "object") {
Object.keys(obj).forEach((key) => {
obj[key] = this.convertResult(obj[key]);
});
}
return obj;
};
change = (doc: ChangeEvent) => {
try {
switch (doc.operationType) {
case "dropDatabase":
return this.destroy();
case "drop":
return this.destroy();
case "delete":
if (!this.opts.onlyEvents) {
if (this.opts.array) {
this.data = this.data.filter((x: any) => doc.documentKey?._id?.equals(x._id));
} else this.data = null;
}
return this.emit("delete", doc.documentKey._id.toHexString());
case "insert":
if (!this.opts.onlyEvents) {
if (this.opts.array) this.data.push(doc.fullDocument);
else this.data = doc.fullDocument;
}
return this.emit("insert", doc.fullDocument);
case "update":
case "replace":
if (!this.opts.onlyEvents) {
if (this.opts.array) {
const i = this.data.findIndex((x: any) => doc.fullDocument?._id?.equals(x._id));
if (i == -1) this.data.push(doc.fullDocument);
else this.data[i] = doc.fullDocument;
} else this.data = doc.fullDocument;
}
return this.emit("change", doc.fullDocument);
case "invalidate":
return this.destroy();
default:
return;
}
} catch (error) {
this.emit("error", error);
}
};
destroy = () => {
this.data = null;
this.stream?.off("change", this.change);
this.emit("close");
if (this.stream.isClosed()) return;
return this.stream.close();
};
}
|
