feat(api): ✨ Finish push notification delivery

classes/database/pushsubscription.ts

@@ -3,8 +3,8 @@ import type {
     Alerts,
     PushSubscription as ApiPushSubscription,
 } from "@versia/client/types";
-import { type Token, db } from "@versia/kit/db";
-import { PushSubscriptions } from "@versia/kit/tables";
+import { type Token, type User, db } from "@versia/kit/db";
+import { PushSubscriptions, Users } from "@versia/kit/tables";
 import {
     type InferInsertModel,
     type InferSelectModel,
@@ -136,6 +136,27 @@ export class PushSubscription extends BaseInterface<
         );
     }
 
+    public static async manyFromUser(
+        user: User,
+        limit?: number,
+        offset?: number,
+    ): Promise<PushSubscription[]> {
+        const found = await db.query.PushSubscriptions.findMany({
+            where: (): SQL => eq(Users.id, user.id),
+            limit,
+            offset,
+            with: {
+                token: {
+                    with: {
+                        user: true,
+                    },
+                },
+            },
+        });
+
+        return found.map((s) => new PushSubscription(s));
+    }
+
     public static async fromSql(
         sql: SQL<unknown> | undefined,
         orderBy: SQL<unknown> | undefined = desc(PushSubscriptions.id),

classes/database/user.ts

@@ -21,7 +21,7 @@ import type {
     FollowReject as VersiaFollowReject,
     User as VersiaUser,
 } from "@versia/federation/types";
-import { Notification, db } from "@versia/kit/db";
+import { Notification, PushSubscription, db } from "@versia/kit/db";
 import {
     EmojiToUser,
     Likes,
@@ -54,6 +54,7 @@ import { searchManager } from "~/classes/search/search-manager";
 import { type Config, config } from "~/packages/config-manager";
 import type { KnownEntity } from "~/types/api.ts";
 import { DeliveryJobType, deliveryQueue } from "../queues/delivery.ts";
+import { PushJobType, pushQueue } from "../queues/push.ts";
 import { BaseInterface } from "./base.ts";
 import { Emoji } from "./emoji.ts";
 import { Instance } from "./instance.ts";
@@ -572,12 +573,40 @@ export class User extends BaseInterface<typeof Users, UserWithRelations> {
         relatedUser: User,
         note?: Note,
     ): Promise<void> {
-        await Notification.insert({
+        const notification = await Notification.insert({
             accountId: relatedUser.id,
             type,
             notifiedId: this.id,
             noteId: note?.id ?? null,
         });
+
+        // Also do push notifications
+        if (config.notifications.push.enabled) {
+            await this.notifyPush(notification.id, type, relatedUser, note);
+        }
+    }
+
+    private async notifyPush(
+        notificationId: string,
+        type: "mention" | "follow_request" | "follow" | "favourite" | "reblog",
+        relatedUser: User,
+        note?: Note,
+    ): Promise<void> {
+        // Fetch all push subscriptions
+        const ps = await PushSubscription.manyFromUser(this);
+
+        pushQueue.addBulk(
+            ps.map((p) => ({
+                data: {
+                    psId: p.id,
+                    type,
+                    relatedUserId: relatedUser.id,
+                    noteId: note?.id,
+                    notificationId,
+                },
+                name: PushJobType.Notify,
+            })),
+        );
     }
 
     public async clearAllNotifications(): Promise<void> {

classes/queues/push.ts

@@ -0,0 +1,18 @@
+import { Queue } from "bullmq";
+import { connection } from "~/utils/redis.ts";
+
+export enum PushJobType {
+    Notify = "notify",
+}
+
+export type PushJobData = {
+    psId: string;
+    type: string;
+    relatedUserId: string;
+    noteId?: string;
+    notificationId: string;
+};
+
+export const pushQueue = new Queue<PushJobData, void, PushJobType>("push", {
+    connection,
+});

classes/workers/push.ts

@@ -0,0 +1,128 @@
+import { Note, PushSubscription, Token, User } from "@versia/kit/db";
+import { Worker } from "bullmq";
+import { htmlToText } from "html-to-text";
+import { sendNotification } from "web-push";
+import { config } from "~/packages/config-manager";
+import { connection } from "~/utils/redis.ts";
+import {
+    type PushJobData,
+    type PushJobType,
+    pushQueue,
+} from "../queues/push.ts";
+
+export const getPushWorker = (): Worker<PushJobData, void, PushJobType> =>
+    new Worker<PushJobData, void, PushJobType>(
+        pushQueue.name,
+        async (job) => {
+            const {
+                data: { psId, relatedUserId, type, noteId, notificationId },
+            } = job;
+
+            const ps = await PushSubscription.fromId(psId);
+
+            if (!ps) {
+                throw new Error(
+                    `Could not resolve push subscription ID ${psId}`,
+                );
+            }
+
+            const token = await Token.fromId(ps.data.tokenId);
+
+            if (!token) {
+                throw new Error(
+                    `Could not resolve token ID ${ps.data.tokenId}`,
+                );
+            }
+
+            const relatedUser = await User.fromId(relatedUserId);
+
+            if (!relatedUser) {
+                throw new Error(
+                    `Could not resolve related user ID ${relatedUserId}`,
+                );
+            }
+
+            const note = noteId ? await Note.fromId(noteId) : null;
+
+            const truncate = (str: string, len: number): string => {
+                if (str.length <= len) {
+                    return str;
+                }
+
+                return `${str.slice(0, len)}...`;
+            };
+
+            const name = truncate(
+                relatedUser.data.displayName || relatedUser.data.username,
+                50,
+            );
+
+            let title = name;
+
+            switch (type) {
+                case "mention":
+                    title = `${name} mentioned you`;
+                    break;
+                case "reply":
+                    title = `${name} replied to you`;
+                    break;
+                case "like":
+                    title = `${name} liked your note`;
+                    break;
+                case "reblog":
+                    title = `${name} reblogged your note`;
+                    break;
+                case "follow":
+                    title = `${name} followed you`;
+                    break;
+                case "follow_request":
+                    title = `${name} requested to follow you`;
+                    break;
+                case "poll":
+                    title = "Poll ended";
+                    break;
+            }
+
+            const body = note
+                ? htmlToText(note.data.spoilerText || note.data.content)
+                : htmlToText(relatedUser.data.note);
+
+            sendNotification(
+                {
+                    endpoint: ps.data.endpoint,
+                    keys: {
+                        auth: ps.data.authSecret,
+                        p256dh: ps.data.publicKey,
+                    },
+                },
+                JSON.stringify({
+                    access_token: token.data.accessToken,
+                    // FIXME
+                    preferred_locale: "en-US",
+                    notification_id: notificationId,
+                    notification_type: type,
+                    icon: relatedUser.getAvatarUrl(config),
+                    title,
+                    body: truncate(body, 140),
+                }),
+                {
+                    vapidDetails: {
+                        subject:
+                            config.notifications.push.vapid.subject ||
+                            config.http.base_url,
+                        privateKey: config.notifications.push.vapid.private,
+                        publicKey: config.notifications.push.vapid.public,
+                    },
+                },
+            );
+        },
+        {
+            connection,
+            removeOnComplete: {
+                age: config.queues.push.remove_on_complete,
+            },
+            removeOnFail: {
+                age: config.queues.push.remove_on_failure,
+            },
+        },
+    );