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
|
import "./MongoBigInt";
import mongoose, { Collection, Connection, LeanDocument } from "mongoose";
import { ChangeStream, ChangeEvent, Long } from "mongodb";
import EventEmitter from "events";
import { Document } from "mongoose";
const uri = process.env.MONGO_URL || "mongodb://localhost:27017/fosscord?readPreference=secondaryPreferred";
// TODO: auto throw error if findOne doesn't find anything
console.log(`[DB] connect: ${uri}`);
const connection = mongoose.createConnection(uri, {
autoIndex: true,
useNewUrlParser: true,
useUnifiedTopology: true,
useFindAndModify: false,
});
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;
constructor(
public collection: Collection,
public pipeline: Array<Record<string, unknown>>,
public opts: {
onlyEvents: boolean;
}
) {
super();
}
init = async () => {
// @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();
this.data = arr.length ? arr[0] : arr;
}
};
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 {
// @ts-ignore
if (doc.fullDocument) {
// @ts-ignore
if (!this.opts.onlyEvents) this.data = doc.fullDocument;
}
switch (doc.operationType) {
case "dropDatabase":
return this.destroy();
case "drop":
return this.destroy();
case "delete":
return this.emit("delete", doc.documentKey._id.toHexString());
case "insert":
return this.emit("insert", doc.fullDocument);
case "update":
case "replace":
return this.emit("change", doc.fullDocument);
case "invalidate":
return this.destroy();
default:
return;
}
} catch (error) {
this.emit("error", error);
}
};
destroy = () => {
this.stream?.off("change", this.change);
this.emit("close");
if (this.stream.isClosed()) return;
return this.stream.close();
};
}
|
