From 644430921afb7c66720e7fe8ee94848e8b83d751 Mon Sep 17 00:00:00 2001 From: Flam3rboy <34555296+Flam3rboy@users.noreply.github.com> Date: Fri, 13 Aug 2021 13:00:31 +0200 Subject: [PATCH] :sparkles: event emit --- util/package-lock.json | Bin 47678 -> 45786 bytes util/package.json | 7 +-- util/src/models/index.ts | 3 ++ util/src/util/Config.ts | 3 +- util/src/util/Database.ts | 3 ++ util/src/util/Event.ts | 94 ++++++++++++++++++++++++++++++++++++++ util/src/util/index.ts | 1 + 7 files changed, 107 insertions(+), 4 deletions(-) create mode 100644 util/src/util/Event.ts diff --git a/util/package-lock.json b/util/package-lock.json index 4977468dc5b18a77ddc1d0cd28ac95acaf85febf..4412d90e1a2655c4ded59ec076525382fe46228f 100644 GIT binary patch delta 927 zcmchU&1=(e9L8z3I^Q=zH=T~!`T`P7nxtEsLg&)AHGP|D(@+DVF-eoQY16#)^&rl_ zVCBc=@aT0YY+(lviu(t|lc2YO-aUAq9pb@*;Ke<+FFf%1@_gR?fWCZ#)~1JZ_Hgss zj4D`k)3OCkD~e{dW&6TcZL=+J@4j-KB9Nf(_Q?~Y_`t*Z1?L%LF1Rq@x`cwr`ZU;E z^>Tsx0gX+j;whe&l$?-i=rub}X-2&xNoAa2qKUjegQClWp1R!CXwv4%g1@6au9;X| z7P~N;3>q0fpRF-0PU2-HRVXk@7rdtgXDWOh0^U3ZK5fj5ja~u!ZVyMr=}yJw;Y>wm zsceq-H$$mTRFeluRV+zCLS=*$6YnxGQHImJW+Vc^N~urw)g+G>Yb`r=?^p@Ndbxo5+yV{X|kJUD{(TqOTwp&&hMgJ@zqTWCodQ;JLO9l*g1$c571x($d6soWw~*Y3n4-|M*Yhxaon)X&mQ|6DM{O z$CZlMg#)M2uRU>T*``SdG$CI&A$CIuq;Ud)(8PiJ4xAZTssNnRZ%qwqqS_Hv$467Gv*CwJUwnJdGU{^Tw&juG-eZHswHfo!s(3AaI3W>m z`(Nvg9_Ux#{h^mRtey_UrF6T#S4RR{xI2_k>9&y(j8+rvcI8|?s$_#T*CrKY?Np>|gvv=4bzz-d+?CICgqlQF0-CM`J?^b2 ztCu9)AO*e|tMu!B|GGryyj2S6!bt^67tYIX53E2}wv>5H{^UUyksF@oyW(@#a3@Ft@m;e zz}S{Jr5PmXWX#^Cyp^m!!~|qCp|Uzg=~|iA%9{-9*S65QIm0yhpGT=bQ_6kn%H zQKd0JSD6>$dh_<&VC^ZULLPqmd};u9dYvAasH4w7Sv>Ig<3;{bNJ*hQ7x3);x&7r4Si3z1JKL|o YL-q&r@jcY~4^Xq00#!19`Si))Urc*v`Tzg` diff --git a/util/package.json b/util/package.json index 69af0f26..c2b08a3d 100644 --- a/util/package.json +++ b/util/package.json @@ -1,10 +1,11 @@ { - "name": "@fosscord/server-util", + "name": "@fosscord/util", "version": "1.3.52", - "description": "Utility functions for the all server repositories", + "description": "Utility functions and database models for fosscord", "main": "dist/index.js", "types": "dist/index.d.ts", "scripts": { + "link": "npm run build && npm link", "build": "tsc -b ." }, "repository": { @@ -36,7 +37,7 @@ "jsonwebtoken": "^8.5.1", "missing-native-js-functions": "^1.2.2", "mongodb": "^3.6.9", - "mongoose": "^5.12.3", + "mongoose": "^5.13.7", "mongoose-autopopulate": "^0.12.3", "typescript": "^4.1.3" }, diff --git a/util/src/models/index.ts b/util/src/models/index.ts index d0a46bf9..db74ef40 100644 --- a/util/src/models/index.ts +++ b/util/src/models/index.ts @@ -12,6 +12,9 @@ type UpdateAggregationStage = type EnforceDocument = T extends Document ? T : T & Document & TMethods; declare module "mongoose" { + interface SchemaOptions { + removeResponse?: string[]; + } interface Model { // removed null -> always return document -> throw error if it doesn't exist findOne( diff --git a/util/src/util/Config.ts b/util/src/util/Config.ts index 78b44315..b4648668 100644 --- a/util/src/util/Config.ts +++ b/util/src/util/Config.ts @@ -1,6 +1,6 @@ import { Schema, model, Types, Document } from "mongoose"; import "missing-native-js-functions"; -import db, { MongooseCache } from "./Database"; +import db from "./Database"; import { Snowflake } from "./Snowflake"; import crypto from "crypto"; @@ -15,6 +15,7 @@ export default { return config as DefaultOptions; }, set: function set(val: any) { + config = val.merge(config); return db.collection("config").updateOne({}, { $set: val }, { upsert: true }); }, }; diff --git a/util/src/util/Database.ts b/util/src/util/Database.ts index 8c6847a8..233152f1 100644 --- a/util/src/util/Database.ts +++ b/util/src/util/Database.ts @@ -14,6 +14,9 @@ const connection = mongoose.createConnection(uri, { useFindAndModify: false, }); console.log(`[Database] connect: mongodb://${url.username}@${url.host}${url.pathname}${url.search}`); +connection.once("open", () => { + console.log("[Database] connected"); +}); export default connection; diff --git a/util/src/util/Event.ts b/util/src/util/Event.ts new file mode 100644 index 00000000..13dd797a --- /dev/null +++ b/util/src/util/Event.ts @@ -0,0 +1,94 @@ +import { Channel, ConsumeMessage } from "amqplib"; +import { EVENT, Event, EventModel } from "../models"; +import { RabbitMQ } from "./RabbitMQ"; +import EventEmitter from "events"; +const events = new EventEmitter(); + +export async function emitEvent(payload: Omit) { + const id = (payload.channel_id || payload.user_id || payload.guild_id) as string; + if (!id) console.error("event doesn't contain any id", payload); + + if (RabbitMQ.connection) { + const data = typeof payload.data === "object" ? JSON.stringify(payload.data) : payload.data; // use rabbitmq for event transmission + await RabbitMQ.channel?.assertExchange(id, "fanout", { durable: false }); + + // assertQueue isn't needed, because a queue will automatically created if it doesn't exist + const successful = RabbitMQ.channel?.publish(id, "", Buffer.from(`${data}`), { type: payload.event }); + if (!successful) throw new Error("failed to send event"); + } else { + events.emit(id, payload); + } +} + +export async function initEvent() { + await RabbitMQ.init(); // does nothing if rabbitmq is not setup + if (RabbitMQ.connection) { + } else { + // use event emitter + } +} + +export interface EventOpts extends Event { + acknowledge?: Function; + channel?: Channel; + cancel: Function; +} + +export interface ListenEventOpts { + channel?: Channel; + acknowledge?: boolean; +} + +export async function listenEvent(event: string, callback: (event: EventOpts) => any, opts?: ListenEventOpts) { + if (RabbitMQ.connection) { + // @ts-ignore + return rabbitListen(opts?.channel || RabbitMQ.channel, event, callback, { acknowledge: opts?.acknowledge }); + } else { + const cancel = () => events.removeListener(event, callback); + events.addListener(event, (opts) => callback({ ...opts, cancel })); + + return cancel; + } +} + +async function rabbitListen( + channel: Channel, + id: string, + callback: (event: EventOpts) => any, + opts?: { acknowledge?: boolean } +) { + await channel.assertExchange(id, "fanout", { durable: false }); + const q = await channel.assertQueue("", { exclusive: true, autoDelete: true }); + + const cancel = () => { + channel.cancel(q.queue); + channel.unbindQueue(q.queue, id, ""); + }; + + channel.bindQueue(q.queue, id, ""); + channel.consume( + q.queue, + (opts) => { + if (!opts) return; + + const data = JSON.parse(opts.content.toString()); + const event = opts.properties.type as EVENT; + + callback({ + event, + data, + acknowledge() { + channel.ack(opts); + }, + channel, + cancel, + }); + // rabbitCh.ack(opts); + }, + { + noAck: !opts?.acknowledge, + } + ); + + return cancel; +} diff --git a/util/src/util/index.ts b/util/src/util/index.ts index 7523a6ad..0aad15f2 100644 --- a/util/src/util/index.ts +++ b/util/src/util/index.ts @@ -7,3 +7,4 @@ export * from "./Snowflake"; export * from "./UserFlags"; export * from "./toBigInt"; export * from "./RabbitMQ"; +export * from "./Event";