Notifications and sending songs to friends

Reactions to your songs, new followers, and songs sent to you are now
notifications: a bell on the feed (app and website) with an unread
count that updates live, and a list that marks them read. "Send ✉" on
a song sends it to people who follow you, with an optional note.
Settings picks which kinds to be notified about.

Push delivery uses no third-party accounts:
- Android: UnifiedPush through a distributor app (ntfy), pointed at
  our own ntfy server (docker-compose.ntfy.yml). Works the same in
  the F-Droid build; no Firebase.
- Browsers: standard Web Push from the existing service worker.
Both use Web Push encryption with a VAPID key pair the server makes
and keeps in the database. Push addresses are checked like any other
user-supplied URL. Setup is in docs/PUSH.md; the privacy page covers
the new data.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

apps/api/drizzle/0016_notifications.sql

@@ -0,0 +1,38 @@
+CREATE TABLE "notifications" (
+	"id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
+	"user_id" uuid NOT NULL,
+	"type" text NOT NULL,
+	"actor_id" uuid NOT NULL,
+	"track_id" uuid,
+	"listen_id" uuid,
+	"emoji" varchar(32),
+	"message" varchar(280),
+	"read_at" timestamp with time zone,
+	"created_at" timestamp with time zone DEFAULT now() NOT NULL
+);
+--> statement-breakpoint
+CREATE TABLE "push_subscriptions" (
+	"id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
+	"user_id" uuid NOT NULL,
+	"kind" text NOT NULL,
+	"endpoint" text NOT NULL,
+	"p256dh" text NOT NULL,
+	"auth" text NOT NULL,
+	"created_at" timestamp with time zone DEFAULT now() NOT NULL,
+	CONSTRAINT "push_subscriptions_endpoint_unique" UNIQUE("endpoint")
+);
+--> statement-breakpoint
+CREATE TABLE "server_keys" (
+	"name" text PRIMARY KEY NOT NULL,
+	"value" jsonb NOT NULL,
+	"created_at" timestamp with time zone DEFAULT now() NOT NULL
+);
+--> statement-breakpoint
+ALTER TABLE "users" ADD COLUMN "notify_prefs" jsonb DEFAULT '{}'::jsonb NOT NULL;--> statement-breakpoint
+ALTER TABLE "notifications" ADD CONSTRAINT "notifications_user_id_users_id_fk" FOREIGN KEY ("user_id") REFERENCES "public"."users"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
+ALTER TABLE "notifications" ADD CONSTRAINT "notifications_actor_id_users_id_fk" FOREIGN KEY ("actor_id") REFERENCES "public"."users"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
+ALTER TABLE "notifications" ADD CONSTRAINT "notifications_track_id_tracks_id_fk" FOREIGN KEY ("track_id") REFERENCES "public"."tracks"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
+ALTER TABLE "notifications" ADD CONSTRAINT "notifications_listen_id_listens_id_fk" FOREIGN KEY ("listen_id") REFERENCES "public"."listens"("id") ON DELETE set null ON UPDATE no action;--> statement-breakpoint
+ALTER TABLE "push_subscriptions" ADD CONSTRAINT "push_subscriptions_user_id_users_id_fk" FOREIGN KEY ("user_id") REFERENCES "public"."users"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
+CREATE INDEX "notifications_user_created_idx" ON "notifications" USING btree ("user_id","created_at" DESC NULLS LAST);--> statement-breakpoint
+CREATE INDEX "push_subscriptions_user_idx" ON "push_subscriptions" USING btree ("user_id");

apps/api/drizzle/meta/_journal.json

@@ -113,6 +113,13 @@
       "when": 1792025274882,
       "tag": "0015_lyrics",
       "breakpoints": true
+    },
+    {
+      "idx": 16,
+      "version": "7",
+      "when": 1792028149643,
+      "tag": "0016_notifications",
+      "breakpoints": true
     }
   ]
 }

apps/api/package.json

@@ -36,11 +36,13 @@
     "postgres": "^3.4.9",
     "sharp": "^0.35.5",
     "undici": "^6.29.0",
+    "web-push": "^3.6.7",
     "zod": "^4.6.5"
   },
   "devDependencies": {
     "@types/node": "^26.6.4",
     "@types/nodemailer": "^8.0.2",
+    "@types/web-push": "^3.6.4",
     "@types/ws": "^8.18.2",
     "concurrently": "^10.0.5",
     "drizzle-kit": "^0.31.11",

apps/api/src/app.ts

@@ -18,6 +18,7 @@ import { connectionRoutes } from './routes/connections';
 import { deviceRoutes } from './routes/devices';
 import { meRoutes } from './routes/me';
 import { reactionRoutes } from './routes/reactions';
+import { notificationRoutes } from './routes/notifications';
 import { realtimeRoutes } from './routes/realtime';
 import { safetyRoutes } from './routes/safety';
 import { trackRoutes } from './routes/tracks';
@@ -76,6 +77,7 @@ export async function buildApp({ logger = true, rateLimits = true }: { logger?:
   await app.register(adminRoutes);
   await app.register(trackRoutes);
   await app.register(reactionRoutes);
+  await app.register(notificationRoutes);
   await app.register(connectionRoutes);
   await app.register(deviceRoutes);
   await app.register(realtimeRoutes);

apps/api/src/db/schema.ts

@@ -13,7 +13,7 @@ import {
   uuid,
   varchar,
 } from 'drizzle-orm/pg-core';
-import type { NowPlayingProvider, Plan, ReportReason, StreamingService } from '@musicshare/shared';
+import type { NotificationPrefs, NotificationType, NowPlayingProvider, Plan, ReportReason, StreamingService } from '@musicshare/shared';
 
 const createdAt = () => timestamp('created_at', { withTimezone: true }).notNull().defaultNow();
 
@@ -33,6 +33,8 @@ export const users = pgTable('users', {
   preferredService: text('preferred_service').$type<StreamingService>(),
   /** The services they use, so songs only offer those; null until they choose (offer all). */
   services: text('services').array().$type<StreamingService[]>(),
+  /** Which notifications to push; missing keys mean on. Everything still shows in the app's list. */
+  notifyPrefs: jsonb('notify_prefs').$type<Partial<NotificationPrefs>>().notNull().default({}),
   /** Billing plan. Kept in sync by the billing webhook (RevenueCat/Stripe) — see lib/entitlements.ts */
   /** The first account on a server becomes admin (moderation). */
   role: text('role').$type<'user' | 'admin'>().notNull().default('user'),
@@ -304,6 +306,54 @@ export const lyrics = pgTable('lyrics', {
   fetchedAt: timestamp('fetched_at', { withTimezone: true }).notNull().defaultNow(),
 });
 
+/** Things that happened to a user: a reaction, a new follower, a song sent to them. */
+export const notifications = pgTable(
+  'notifications',
+  {
+    id: uuid('id').primaryKey().defaultRandom(),
+    userId: uuid('user_id')
+      .notNull()
+      .references(() => users.id, { onDelete: 'cascade' }),
+    type: text('type').$type<NotificationType>().notNull(),
+    actorId: uuid('actor_id')
+      .notNull()
+      .references(() => users.id, { onDelete: 'cascade' }),
+    trackId: uuid('track_id').references(() => tracks.id, { onDelete: 'cascade' }),
+    /** The play a reaction was on (gone once the play is trimmed from history). */
+    listenId: uuid('listen_id').references(() => listens.id, { onDelete: 'set null' }),
+    emoji: varchar('emoji', { length: 32 }),
+    /** A note sent with a song. */
+    message: varchar('message', { length: 280 }),
+    readAt: timestamp('read_at', { withTimezone: true }),
+    createdAt: createdAt(),
+  },
+  (t) => [index('notifications_user_created_idx').on(t.userId, t.createdAt.desc())],
+);
+
+/** Where to push notifications: browsers (Web Push) and phones (UnifiedPush). Both use Web Push encryption. */
+export const pushSubscriptions = pgTable(
+  'push_subscriptions',
+  {
+    id: uuid('id').primaryKey().defaultRandom(),
+    userId: uuid('user_id')
+      .notNull()
+      .references(() => users.id, { onDelete: 'cascade' }),
+    kind: text('kind').$type<'web' | 'unifiedpush'>().notNull(),
+    endpoint: text('endpoint').notNull().unique(),
+    p256dh: text('p256dh').notNull(),
+    auth: text('auth').notNull(),
+    createdAt: createdAt(),
+  },
+  (t) => [index('push_subscriptions_user_idx').on(t.userId)],
+);
+
+/** Keys the server generates for itself once (e.g. the Web Push VAPID key pair). */
+export const serverKeys = pgTable('server_keys', {
+  name: text('name').primaryKey(),
+  value: jsonb('value').$type<Record<string, string>>().notNull(),
+  createdAt: createdAt(),
+});
+
 const bytea = customType<{ data: Buffer }>({ dataType: () => 'bytea' });
 
 /** Profile pictures, already resized and re-encoded (WebP, metadata stripped). */

apps/api/src/env.ts

@@ -26,6 +26,9 @@ const schema = z.object({
   /** Enables "Sign in with Last.fm" (web auth). Without it, users type their username. */
   LASTFM_SHARED_SECRET: optional,
   ODESLI_API_KEY: optional,
+  /** Web Push keys (`npx web-push generate-vapid-keys`). Optional: without them the server makes a pair and keeps it in the database. */
+  VAPID_PUBLIC_KEY: optional,
+  VAPID_PRIVATE_KEY: optional,
   /** Lyrics come from LRCLIB (open source, self-hostable). Point this at your own instance, or set it empty to turn lyrics off. */
   LRCLIB_URL: z
     .string()

apps/api/src/routes/me.ts

@@ -1,6 +1,6 @@
 import { eq } from 'drizzle-orm';
 import type { FastifyInstance } from 'fastify';
-import { setFeaturedSchema, setSharingSchema, updateMeSchema, type MeDTO } from '@musicshare/shared';
+import { notificationPrefs, setFeaturedSchema, setSharingSchema, updateMeSchema, type MeDTO } from '@musicshare/shared';
 import { db, schema } from '../db';
 import type { User } from '../db/schema';
 import { SESSION_COOKIE, requireUser } from '../lib/auth';
@@ -25,6 +25,7 @@ export const toMeDTO = (u: User): MeDTO => ({
   bio: u.bio,
   preferredService: u.preferredService,
   services: u.services,
+  notifyPrefs: notificationPrefs(u.notifyPrefs),
   role: u.role,
   onboardedAt: u.onboardedAt?.toISOString() ?? null,
   sharingPausedUntil: isSharingPaused(u) ? u.sharingPausedUntil!.toISOString() : null,
@@ -40,7 +41,12 @@ export async function meRoutes(app: FastifyInstance) {
     const { onboarded, ...input } = updateMeSchema.parse(req.body);
     // Clearing the display name falls back to the username.
     if ('displayName' in input && !input.displayName) input.displayName = user.username;
-    const changes = { ...input, ...(onboarded && !user.onboardedAt ? { onboardedAt: new Date() } : {}) };
+    const { notifyPrefs, ...rest } = input;
+    const changes = {
+      ...rest,
+      ...(notifyPrefs ? { notifyPrefs: { ...user.notifyPrefs, ...notifyPrefs } } : {}),
+      ...(onboarded && !user.onboardedAt ? { onboardedAt: new Date() } : {}),
+    };
     if (Object.keys(changes).length === 0) return toMeDTO(user); // nothing to update
     const [updated] = await db.update(schema.users).set(changes).where(eq(schema.users.id, user.id)).returning();
     return toMeDTO(updated!);

apps/api/src/routes/notifications.ts

@@ -0,0 +1,88 @@
+import { and, eq, inArray, isNull } from 'drizzle-orm';
+import type { FastifyInstance } from 'fastify';
+import { z } from 'zod';
+import { pushSubscriptionSchema, sendSongSchema } from '@musicshare/shared';
+import { db, schema } from '../db';
+import { requireUser } from '../lib/auth';
+import { HttpError } from '../lib/http';
+import { UnsafeUrlError, assertSafeServerUrl } from '../lib/ssrf';
+import { hiddenUserIds } from '../services/blocks';
+import { listNotifications, notify, vapidKeys } from '../services/notifications';
+import { enqueueResolve } from '../services/queues';
+
+/** Most browsers and phones someone might have at once. */
+const MAX_PUSH_SUBSCRIPTIONS = 10;
+
+export async function notificationRoutes(app: FastifyInstance) {
+  app.get<{ Querystring: { cursor?: string } }>('/notifications', async (req) => {
+    const user = requireUser(req);
+    return listNotifications(user.id, req.query.cursor ?? null);
+  });
+
+  app.post('/notifications/read', async (req) => {
+    const user = requireUser(req);
+    const n = schema.notifications;
+    await db.update(n).set({ readAt: new Date() }).where(and(eq(n.userId, user.id), isNull(n.readAt)));
+    return { ok: true };
+  });
+
+  /** Send a song to people who follow you (and haven't blocked you). Others are skipped. */
+  app.post('/shares', { config: { rateLimit: { max: 30, timeWindow: '1 minute' } } }, async (req) => {
+    const user = requireUser(req);
+    const { trackId, to, message } = sendSongSchema.parse(req.body);
+    const [track] = await db.select({ id: schema.tracks.id }).from(schema.tracks).where(eq(schema.tracks.id, trackId));
+    if (!track) throw new HttpError(404, 'Track not found');
+
+    const recipients = await db
+      .select({ id: schema.users.id, username: schema.users.username })
+      .from(schema.users)
+      .innerJoin(schema.follows, and(eq(schema.follows.followerId, schema.users.id), eq(schema.follows.followeeId, user.id)))
+      .where(and(inArray(schema.users.username, to), isNull(schema.users.suspendedAt)));
+    const hidden = await hiddenUserIds(user.id);
+    const sentTo: string[] = [];
+    for (const r of recipients) {
+      if (hidden.has(r.id)) continue;
+      await notify({ userId: r.id, actorId: user.id, type: 'share', trackId, message: message || null });
+      sentTo.push(r.username);
+    }
+    if (sentTo.length === 0) throw new HttpError(400, 'You can only send songs to people who follow you');
+    await enqueueResolve(trackId); // so it opens straight in their service
+    return { sentTo };
+  });
+
+  /** The server's public key, for subscribing browsers and phones to push. */
+  app.get('/push/key', async () => ({ publicKey: (await vapidKeys()).publicKey }));
+
+  app.put('/me/push', async (req) => {
+    const user = requireUser(req);
+    const input = pushSubscriptionSchema.parse(req.body);
+    try {
+      await assertSafeServerUrl(input.endpoint);
+    } catch (err) {
+      if (err instanceof UnsafeUrlError) throw new HttpError(400, `Can't send notifications there: ${err.message}`);
+      throw err;
+    }
+    const values = { userId: user.id, kind: input.kind, endpoint: input.endpoint, ...input.keys };
+    await db
+      .insert(schema.pushSubscriptions)
+      .values(values)
+      .onConflictDoUpdate({ target: schema.pushSubscriptions.endpoint, set: values });
+    // Keep the newest few.
+    const subs = await db
+      .select({ id: schema.pushSubscriptions.id, createdAt: schema.pushSubscriptions.createdAt })
+      .from(schema.pushSubscriptions)
+      .where(eq(schema.pushSubscriptions.userId, user.id));
+    const extra = subs.sort((a, b) => b.createdAt.getTime() - a.createdAt.getTime()).slice(MAX_PUSH_SUBSCRIPTIONS);
+    if (extra.length) await db.delete(schema.pushSubscriptions).where(inArray(schema.pushSubscriptions.id, extra.map((s) => s.id)));
+    return { ok: true };
+  });
+
+  app.delete('/me/push', async (req) => {
+    const user = requireUser(req);
+    const { endpoint } = z.object({ endpoint: z.string().max(2000) }).parse(req.body);
+    await db
+      .delete(schema.pushSubscriptions)
+      .where(and(eq(schema.pushSubscriptions.userId, user.id), eq(schema.pushSubscriptions.endpoint, endpoint)));
+    return { ok: true };
+  });
+}

apps/api/src/routes/reactions.ts

@@ -6,6 +6,7 @@ import { db, schema } from '../db';
 import { requireUser } from '../lib/auth';
 import { HttpError } from '../lib/http';
 import { blockStatus } from '../services/blocks';
+import { notify, unnotifyReaction } from '../services/notifications';
 import { getReactions, publishNowPlaying } from '../services/nowPlaying';
 
 const params = z.object({ id: z.uuid() });
@@ -42,6 +43,7 @@ export async function reactionRoutes(app: FastifyInstance) {
       .insert(schema.reactions)
       .values({ listenId: id, userId: user.id, emoji })
       .onConflictDoUpdate({ target: [schema.reactions.listenId, schema.reactions.userId], set: { emoji, createdAt: new Date() } });
+    await notify({ userId: listen.userId, actorId: user.id, type: 'reaction', listenId: id, trackId: listen.trackId, emoji });
     return afterChange(listen);
   });
 
@@ -50,6 +52,7 @@ export async function reactionRoutes(app: FastifyInstance) {
     const { id } = params.parse(req.params);
     const listen = await reactable(user.id, id);
     await db.delete(schema.reactions).where(and(eq(schema.reactions.listenId, id), eq(schema.reactions.userId, user.id)));
+    await unnotifyReaction(listen.userId, user.id, id);
     return afterChange(listen);
   });
 }

apps/api/src/routes/realtime.ts

@@ -12,6 +12,7 @@ import type { WebSocket } from 'ws';
 import { db, schema } from '../db';
 import { FOLLOWS_CHANGED_CHANNEL, createRedis } from '../lib/redis';
 import { watchUsers } from '../services/connections';
+import { unreadCount } from '../services/notifications';
 
 const watchers = new Map<string, Set<WebSocket>>(); // watched userId -> sockets
 const socketsByOwner = new Map<string, Set<WebSocket>>(); // viewer userId -> their sockets
@@ -42,10 +43,23 @@ async function subscribeSocket(socket: WebSocket, viewerId: string) {
 
 export async function realtimeRoutes(app: FastifyInstance) {
   const sub = createRedis();
-  await sub.psubscribe('np:*');
+  await sub.psubscribe('np:*', 'notify:*');
   await sub.subscribe(FOLLOWS_CHANGED_CHANNEL);
 
-  sub.on('pmessage', (_pattern, channel, message) => {
+  sub.on('pmessage', (pattern, channel, message) => {
+    if (pattern === 'notify:*') {
+      // Something happened to this user: tell their open apps the new unread count.
+      const userId = channel.slice('notify:'.length);
+      const sockets = socketsByOwner.get(userId);
+      if (!sockets) return;
+      unreadCount(userId)
+        .then((unread) => {
+          const payload = JSON.stringify({ type: 'notification', data: { unread } });
+          for (const socket of sockets) socket.send(payload);
+        })
+        .catch((e) => app.log.error(e));
+      return;
+    }
     const userId = channel.slice(3);
     const payload = `{"type":"now_playing","data":${message}}`;
     for (const socket of watchers.get(userId) ?? []) socket.send(payload);

apps/api/src/routes/users.ts

@@ -2,6 +2,7 @@ import { and, count, desc, eq, ilike, inArray, isNotNull, isNull, or } from 'dri
 import type { FastifyInstance } from 'fastify';
 import { sortFeed, sourceProfileUrl, type FeedItemDTO, type FollowListDTO, type ProfileDTO } from '@musicshare/shared';
 import { db, schema } from '../db';
+import { notify } from '../services/notifications';
 import { requireUser } from '../lib/auth';
 import { entitlementsFor } from '../lib/entitlements';
 import { HttpError } from '../lib/http';
@@ -107,8 +108,13 @@ export async function userRoutes(app: FastifyInstance) {
     const { user: target, viewerBlocked } = await findVisible(req.params.username, viewer.id);
     if (target.id === viewer.id) throw new HttpError(400, "You can't follow yourself");
     if (viewerBlocked) throw new HttpError(400, 'Unblock them first');
-    await db.insert(schema.follows).values({ followerId: viewer.id, followeeId: target.id }).onConflictDoNothing();
+    const added = await db
+      .insert(schema.follows)
+      .values({ followerId: viewer.id, followeeId: target.id })
+      .onConflictDoNothing()
+      .returning({ id: schema.follows.followerId });
     await redis.publish(FOLLOWS_CHANGED_CHANNEL, viewer.id);
+    if (added.length) await notify({ userId: target.id, actorId: viewer.id, type: 'follow' });
     return { ok: true };
   });
 

apps/api/src/services/notifications.ts

@@ -0,0 +1,194 @@
+import { and, desc, eq, gt, inArray, isNull, lt } 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;
+}): 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;
+}
+
+/** 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,
+        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);
+      }
+    }),
+  );
+}

apps/api/src/services/queues.ts

@@ -5,12 +5,14 @@ const connection = createRedis();
 
 export const POLL_QUEUE = 'poll';
 export const RESOLVE_QUEUE = 'resolve-track';
+export const PUSH_QUEUE = 'push';
 
 export const pollQueue = new Queue<{ connectionId: string }>(POLL_QUEUE, { connection });
 export const resolveQueue = new Queue<{ trackId: string }>(RESOLVE_QUEUE, { connection });
+export const pushQueue = new Queue<{ notificationId: string }>(PUSH_QUEUE, { connection });
 
 export async function closeQueues() {
-  await Promise.all([pollQueue.close(), resolveQueue.close()]);
+  await Promise.all([pollQueue.close(), resolveQueue.close(), pushQueue.close()]);
   await connection.quit();
 }
 
@@ -27,3 +29,8 @@ export function enqueueResolve(trackId: string) {
     },
   );
 }
+
+/** Push a notification to the user's devices (the worker sends it). */
+export function enqueuePush(notificationId: string) {
+  return pushQueue.add('push', { notificationId }, { attempts: 2, backoff: { type: 'fixed', delay: 30_000 }, removeOnComplete: true, removeOnFail: 1000 });
+}

apps/api/src/worker.ts

@@ -8,7 +8,8 @@ import { createRedis } from './lib/redis';
 import { POLLED_PROVIDERS, pollingProviders } from './providers';
 import { ProviderAuthError, ProviderRateLimitError } from './providers/types';
 import { applyPollResult, isSharingPaused } from './services/nowPlaying';
-import { POLL_QUEUE, RESOLVE_QUEUE, pollQueue } from './services/queues';
+import { sendPush } from './services/notifications';
+import { POLL_QUEUE, PUSH_QUEUE, RESOLVE_QUEUE, pollQueue } from './services/queues';
 import { resolveTrackLinks } from './services/resolver';
 import {
   IDLE_POLL_SECONDS,
@@ -116,7 +117,11 @@ const resolveWorker = new Worker<{ trackId: string }>(RESOLVE_QUEUE, (job) => re
   connection: createRedis(),
   concurrency: 5,
 });
-for (const w of [pollWorker, resolveWorker]) w.on('failed', (job, err) => log(`${w.name} job ${job?.id} failed`, err.message));
+const pushWorker = new Worker<{ notificationId: string }>(PUSH_QUEUE, (job) => sendPush(job.data.notificationId), {
+  connection: createRedis(),
+  concurrency: 10,
+});
+for (const w of [pollWorker, resolveWorker, pushWorker]) w.on('failed', (job, err) => log(`${w.name} job ${job?.id} failed`, err.message));
 
 await reconcile();
 const tickTimer = setInterval(tick, 1000);
@@ -126,7 +131,7 @@ log('worker started');
 async function shutdown() {
   clearInterval(tickTimer);
   clearInterval(reconcileTimer);
-  await Promise.all([pollWorker.close(), resolveWorker.close()]);
+  await Promise.all([pollWorker.close(), resolveWorker.close(), pushWorker.close()]);
   await closeConnections();
   process.exit(0);
 }

apps/api/test/notifications.test.ts

@@ -0,0 +1,126 @@
+import { createECDH, randomBytes } from 'node:crypto';
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { applyPollResult } from '../src/services/nowPlaying';
+import { sendPush } from '../src/services/notifications';
+import { bearer, createUser, follow, useTestApp } from './helpers';
+
+const app = useTestApp();
+afterEach(() => vi.unstubAllGlobals());
+
+const song = (title: string) => ({ track: { title, artist: 'Massive Attack' }, isPlaying: true, progressMs: null });
+const inbox = async (token: string) => (await app().inject({ url: '/notifications', headers: bearer(token) })).json();
+const react = (token: string, listenId: string, emoji: string) =>
+  app().inject({ method: 'PUT', url: `/listens/${listenId}/reaction`, headers: bearer(token), payload: { emoji } });
+
+/** Alice plays a song; Bob follows her. */
+async function setup() {
+  const alice = await createUser(app(), 'alice');
+  const bob = await createUser(app(), 'bob');
+  await follow(app(), bob.token, 'alice');
+  await applyPollResult(alice.user.id, 'lastfm', song('Teardrop'));
+  const feed = (await app().inject({ url: '/feed', headers: bearer(bob.token) })).json();
+  return { alice, bob, listenId: feed[0].listenId as string, trackId: feed[0].track.id as string };
+}
+
+describe('notifications', () => {
+  it('tells you about a new follower once, even if they unfollow and follow again', async () => {
+    const { alice, bob } = await setup();
+    await app().inject({ method: 'DELETE', url: '/users/alice/follow', headers: bearer(bob.token) });
+    await follow(app(), bob.token, 'alice');
+    const list = await inbox(alice.token);
+    expect(list.unread).toBe(1);
+    expect(list.notifications).toEqual([expect.objectContaining({ type: 'follow', actor: expect.objectContaining({ username: 'bob' }), read: false })]);
+  });
+
+  it('a reaction is one notification that changes with the emoji, and goes if taken back', async () => {
+    const { alice, bob, listenId } = await setup();
+    await app().inject({ method: 'POST', url: '/notifications/read', headers: bearer(alice.token), payload: {} });
+    await react(bob.token, listenId, '🔥');
+    await react(bob.token, listenId, '😂');
+    let list = await inbox(alice.token);
+    const reactions = list.notifications.filter((n: { type: string }) => n.type === 'reaction');
+    expect(reactions).toEqual([expect.objectContaining({ emoji: '😂', track: expect.objectContaining({ title: 'Teardrop' }) })]);
+    expect(list.unread).toBe(1);
+
+    await app().inject({ method: 'DELETE', url: `/listens/${listenId}/reaction`, headers: bearer(bob.token) });
+    list = await inbox(alice.token);
+    expect(list.notifications.filter((n: { type: string }) => n.type === 'reaction')).toEqual([]);
+  });
+
+  it('sends songs only to people who follow you', async () => {
+    const { alice, bob, trackId } = await setup();
+    const send = (token: string, to: string[]) =>
+      app().inject({ method: 'POST', url: '/shares', headers: bearer(token), payload: { trackId, to, message: 'you need this' } });
+
+    // Bob follows Alice, so Alice can send to Bob, but Bob can't send to Alice.
+    expect((await send(alice.token, ['bob', 'nobody'])).json()).toEqual({ sentTo: ['bob'] });
+    expect((await send(bob.token, ['alice'])).statusCode).toBe(400);
+
+    const list = await inbox(bob.token);
+    expect(list.notifications[0]).toMatchObject({
+      type: 'share',
+      message: 'you need this',
+      actor: { username: 'alice' },
+      track: { title: 'Teardrop' },
+    });
+    await app().inject({ method: 'POST', url: '/notifications/read', headers: bearer(bob.token), payload: {} });
+    expect((await inbox(bob.token)).unread).toBe(0);
+  });
+
+  it('saves notification settings', async () => {
+    const { alice } = await setup();
+    const me = (await app().inject({ method: 'PATCH', url: '/me', headers: bearer(alice.token), payload: { notifyPrefs: { follow: false } } })).json();
+    expect(me.notifyPrefs).toEqual({ reaction: true, follow: false, share: true });
+  });
+});
+
+describe('push', () => {
+  // A real P-256 key pair, as a browser or the UnifiedPush connector makes.
+  const ecdh = createECDH('prime256v1');
+  ecdh.generateKeys();
+  const keys = { p256dh: ecdh.getPublicKey('base64url'), auth: randomBytes(16).toString('base64url') };
+  const subscribe = (token: string, endpoint: string) =>
+    app().inject({ method: 'PUT', url: '/me/push', headers: bearer(token), payload: { kind: 'unifiedpush', endpoint, keys } });
+
+  it('refuses push endpoints on private networks', async () => {
+    const alice = await createUser(app(), 'alice');
+    expect((await subscribe(alice.token, 'http://127.0.0.1:8080/up/abc')).statusCode).toBe(400);
+    expect((await subscribe(alice.token, 'https://93.184.216.34/up/abc')).statusCode).toBe(200);
+  });
+
+  it('sends an encrypted, signed push, respects settings, and forgets endpoints that are gone', async () => {
+    const { alice } = await setup();
+    await subscribe(alice.token, 'https://93.184.216.34/up/abc');
+    const id = (await inbox(alice.token)).notifications[0].id;
+
+    const sent: { url: string; headers: Record<string, string>; body: Buffer }[] = [];
+    let status = 201;
+    vi.stubGlobal('fetch', async (url: string, init: RequestInit) => {
+      sent.push({ url, headers: init.headers as Record<string, string>, body: init.body as Buffer });
+      return new Response(null, { status });
+    });
+
+    await sendPush(id);
+    expect(sent).toHaveLength(1);
+    expect(sent[0]!.url).toBe('https://93.184.216.34/up/abc');
+    expect(sent[0]!.headers['Content-Encoding']).toBe('aes128gcm');
+    expect(sent[0]!.headers.Authorization).toMatch(/^vapid t=/);
+    expect(sent[0]!.body.toString()).not.toContain('bob'); // encrypted
+
+    await app().inject({ method: 'PATCH', url: '/me', headers: bearer(alice.token), payload: { notifyPrefs: { follow: false } } });
+    await sendPush(id);
+    expect(sent).toHaveLength(1);
+
+    await app().inject({ method: 'PATCH', url: '/me', headers: bearer(alice.token), payload: { notifyPrefs: { follow: true } } });
+    status = 410;
+    await sendPush(id);
+    expect(sent).toHaveLength(2);
+    await sendPush(id);
+    expect(sent).toHaveLength(2); // subscription removed
+  });
+
+  it('serves the public key', async () => {
+    const res = await app().inject({ url: '/push/key' });
+    expect(res.json().publicKey).toMatch(/^[A-Za-z0-9_-]{80,}$/);
+  });
+});