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
|
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) 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();
};
}
|
