Repository navigation
Expand file tree
/
Copy pathworker.ts
More file actions
86 lines (70 loc) · 2 KB
/
Copy pathworker.ts
File metadata and controls
86 lines (70 loc) · 2 KB
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
import { config } from "../util/config.js";
import { type Job, Worker } from "bullmq";
import webPush from "web-push";
import { PushSubscription } from "../entity/pushSubscription.js";
import type { ActorMention } from "../util/activitypub/constants.js";
import { splitQualifiedMention } from "../util/activitypub/util.js";
import { initDatabase } from "../util/database.js";
import { getOrFetchUser } from "../util/entity/user.js";
webPush.setVapidDetails(
config().federation.instance_url.origin,
config().notifications.publicKey,
config().notifications.privateKey,
);
export type PushNotificationJobData = {
user: ActorMention;
notification: {
id: string;
title: string;
body: string;
sent: number; // timestamp of when notification sent
image?: string; // url of image to display
channel?: ActorMention;
guild?: ActorMention;
author?: ActorMention;
};
};
const jobHandler = async (job: Job<PushNotificationJobData>) => {
console.log(`${new Date()} ${job.data.user}`);
if (splitQualifiedMention(job.data.user).domain !== config().federation.webapp_url.hostname)
return;
const user = await getOrFetchUser(job.data.user);
const subscriptions = await PushSubscription.find({
where: {
userId: user.id,
},
});
for (const sub of subscriptions) {
try {
const res = await webPush.sendNotification(
sub.asStandard(),
JSON.stringify(job.data.notification),
);
console.log(res);
} catch {
await sub.remove();
}
}
};
const worker = new Worker("notifications", jobHandler, {
connection: {
host: config().redis.host,
port: config().redis.port,
},
autorun: false,
});
worker.on("ready", () => {
console.log("ready");
});
worker.on("failed", (job) => {
console.warn(`${new Date()} Push notification failed ${job?.failedReason}`);
});
worker.on("error", (e) => {
if ("code" in e && e.code === "ECONNREFUSED")
return console.error("Failed to connect to redis", e.message);
console.error(e);
});
void (async () => {
await initDatabase();
await worker.run();
})();