musicshare/apps/api/src/services/notifications.ts

import { and, desc, eq, gt, inArray, isNull, lt, or } from 'drizzle-orm';
import webpush from 'web-push';
import {
  notificationPath,
  notificationPrefs,
  notificationText,
  type NotificationDTO,
  type NotificationType,
} from '@musicshare/shared';
import { db, schema } from '../db';
import { env } from '../env';
import { redis } from '../lib/redis';
import { assertSafeServerUrl, serverFetchOptions } from '../lib/ssrf';
import { toUserSummary } from './nowPlaying';
import { enqueuePush } from './queues';
import { getTrackDTOs } from './tracks';

export const notifyChannel = (userId: string) => `notify:${userId}`;

/** A second follow from the same person within this long (unfollow, follow again) isn't news. */
const FOLLOW_REPEAT_MS = 24 * 3600_000;

type Row = typeof schema.notifications.$inferSelect;

/**
 * Record something that happened to `userId` and push it. Reactions replace that
 * person's earlier reaction to the same play rather than piling up.
 */
export async function notify(input: {
  userId: string;
  actorId: string;
  type: NotificationType;
  trackId?: string | null;
  listenId?: string | null;
  emoji?: string | null;
  message?: string | null;
  commentId?: string | null;
}): Promise<string | null> {
  if (input.userId === input.actorId) return null;
  const n = schema.notifications;
  const mine = and(eq(n.userId, input.userId), eq(n.actorId, input.actorId), eq(n.type, input.type));

  let row: Row | undefined;
  if (input.type === 'follow') {
    const [recent] = await db
      .select({ id: n.id })
      .from(n)
      .where(and(mine, gt(n.createdAt, new Date(Date.now() - FOLLOW_REPEAT_MS))));
    if (recent) return null;
  }
  if (input.type === 'reaction' && input.listenId) {
    [row] = await db
      .update(n)
      .set({ emoji: input.emoji, createdAt: new Date(), readAt: null })
      .where(and(mine, eq(n.listenId, input.listenId)))
      .returning();
  }
  if (!row) [row] = await db.insert(n).values(input).returning();

  await redis.publish(notifyChannel(input.userId), row!.id);
  await enqueuePush(row!.id);
  return row!.id;
}

/** After a block: neither sees notifications from the other any more. */
export async function removeNotificationsBetween(a: string, b: string) {
  const n = schema.notifications;
  await db.delete(n).where(or(and(eq(n.userId, a), eq(n.actorId, b)), and(eq(n.userId, b), eq(n.actorId, a))));
}

/** Take back a reaction's notification when the reaction is removed (if it hasn't been seen). */
export async function unnotifyReaction(userId: string, actorId: string, listenId: string) {
  const n = schema.notifications;
  await db
    .delete(n)
    .where(and(eq(n.userId, userId), eq(n.actorId, actorId), eq(n.type, 'reaction'), eq(n.listenId, listenId), isNull(n.readAt)));
}

export async function toNotificationDTOs(rows: Row[]): Promise<NotificationDTO[]> {
  const actorIds = [...new Set(rows.map((r) => r.actorId))];
  const actors = actorIds.length
    ? await db.select().from(schema.users).where(inArray(schema.users.id, actorIds))
    : [];
  const byId = new Map(actors.map((u) => [u.id, toUserSummary(u)]));
  const tracks = await getTrackDTOs([...new Set(rows.flatMap((r) => (r.trackId ? [r.trackId] : [])))]);
  return rows.flatMap((r) => {
    const actor = byId.get(r.actorId);
    if (!actor) return [];
    return [
      {
        id: r.id,
        type: r.type,
        actor,
        track: r.trackId ? (tracks.get(r.trackId) ?? null) : null,
        listenId: r.listenId,
        emoji: r.emoji,
        message: r.message,
        read: Boolean(r.readAt),
        createdAt: r.createdAt.toISOString(),
      },
    ];
  });
}

/** Newest first, 30 at a time; the cursor is the last one's time. */
export async function listNotifications(userId: string, cursor: string | null) {
  const n = schema.notifications;
  const before = cursor ? new Date(cursor) : null;
  const rows = await db
    .select()
    .from(n)
    .where(and(eq(n.userId, userId), before && !isNaN(before.getTime()) ? lt(n.createdAt, before) : undefined))
    .orderBy(desc(n.createdAt))
    .limit(31);
  const page = rows.slice(0, 30);
  return {
    notifications: await toNotificationDTOs(page),
    unread: await unreadCount(userId),
    nextCursor: rows.length > 30 ? page.at(-1)!.createdAt.toISOString() : null,
  };
}

export async function unreadCount(userId: string): Promise<number> {
  const n = schema.notifications;
  const rows = await db
    .select({ id: n.id })
    .from(n)
    .where(and(eq(n.userId, userId), isNull(n.readAt)))
    .limit(100);
  return rows.length;
}

// --- Push -------------------------------------------------------------------------------------

let vapid: { publicKey: string; privateKey: string } | null = null;

/**
 * The VAPID key pair push services use to know messages come from us. Set
 * VAPID_PUBLIC_KEY / VAPID_PRIVATE_KEY, or one is made on first use and kept in
 * the database (changing it would break every existing subscription).
 */
export async function vapidKeys() {
  if (vapid) return vapid;
  if (env.VAPID_PUBLIC_KEY && env.VAPID_PRIVATE_KEY) return (vapid = { publicKey: env.VAPID_PUBLIC_KEY, privateKey: env.VAPID_PRIVATE_KEY });
  const generated = webpush.generateVAPIDKeys();
  await db.insert(schema.serverKeys).values({ name: 'vapid', value: { ...generated } }).onConflictDoNothing();
  const [row] = await db.select().from(schema.serverKeys).where(eq(schema.serverKeys.name, 'vapid'));
  return (vapid = row!.value as { publicKey: string; privateKey: string });
}

/** Send one notification to all of its user's browsers and phones, if they want this kind. */
export async function sendPush(notificationId: string) {
  const [row] = await db.select().from(schema.notifications).where(eq(schema.notifications.id, notificationId));
  if (!row || row.readAt) return;
  const [user] = await db.select().from(schema.users).where(eq(schema.users.id, row.userId));
  if (!user || !notificationPrefs(user.notifyPrefs)[row.type]) return;
  const subs = await db.select().from(schema.pushSubscriptions).where(eq(schema.pushSubscriptions.userId, row.userId));
  if (subs.length === 0) return;
  const [dto] = await toNotificationDTOs([row]);
  if (!dto) return;

  const { title, body } = notificationText(dto);
  const payload = JSON.stringify({
    id: dto.id,
    type: dto.type,
    title,
    body,
    path: notificationPath(dto),
    icon: dto.actor.avatarUrl,
    unread: await unreadCount(row.userId),
  });
  const keys = await vapidKeys();
  await Promise.all(
    subs.map(async (sub) => {
      try {
        // web-push encrypts and signs; we send it ourselves so the address is checked like any user-supplied URL.
        const req = webpush.generateRequestDetails({ endpoint: sub.endpoint, keys: { p256dh: sub.p256dh, auth: sub.auth } }, payload, {
          vapidDetails: { subject: env.WEB_URL.startsWith('https:') ? env.WEB_URL : 'mailto:app@musicshare.binarygnome.com', ...keys },
          TTL: 24 * 3600,
          urgency: 'normal',
          contentEncoding: 'aes128gcm',
        });
        await assertSafeServerUrl(req.endpoint);
        const res = await fetch(req.endpoint, {
          method: req.method,
          headers: req.headers as Record<string, string>,
          body: req.body as Buffer<ArrayBuffer>,
          signal: AbortSignal.timeout(10_000),
          ...serverFetchOptions(),
        });
        // Gone: the browser unsubscribed or the app was uninstalled.
        if (res.status === 404 || res.status === 410) {
          await db.delete(schema.pushSubscriptions).where(eq(schema.pushSubscriptions.id, sub.id));
        } else if (!res.ok) {
          console.warn(`push to ${new URL(sub.endpoint).host} failed: ${res.status}`);
        }
      } catch (err) {
        console.warn(`push to ${new URL(sub.endpoint).host} failed:`, (err as Error).message);
      }
    }),
  );
}