diff --git a/apps/journal/app/components/ClientDate.tsx b/apps/journal/app/components/ClientDate.tsx index af210ae..300b376 100644 --- a/apps/journal/app/components/ClientDate.tsx +++ b/apps/journal/app/components/ClientDate.tsx @@ -3,10 +3,14 @@ import { useLocale } from "./LocaleContext"; /** * Renders a date formatted with the server-detected locale, * ensuring SSR and client output match (no hydration flicker). + * Pass `withTime` for surfaces where the hour/minute matter + * (e.g. notifications), keeping plain dates as the default. */ -export function ClientDate({ iso }: { iso: string }) { +export function ClientDate({ iso, withTime = false }: { iso: string; withTime?: boolean }) { const locale = useLocale(); - return ( - - ); + const d = new Date(iso); + const text = withTime + ? d.toLocaleString(locale, { dateStyle: "short", timeStyle: "short" }) + : d.toLocaleDateString(locale); + return ; } diff --git a/apps/journal/app/hooks/useUnreadNotifications.ts b/apps/journal/app/hooks/useUnreadNotifications.ts new file mode 100644 index 0000000..7286262 --- /dev/null +++ b/apps/journal/app/hooks/useUnreadNotifications.ts @@ -0,0 +1,39 @@ +import { useEffect, useState } from "react"; + +/** + * Subscribes to /api/events for live `notifications.unread` updates. + * Returns the live unread count, seeded with the loader-provided + * baseline so the badge renders correctly before the SSE handshake + * completes. Native EventSource handles reconnects with the + * server-suggested `retry:` interval. + */ +export function useUnreadNotifications(initialCount: number, signedIn: boolean): number { + const [count, setCount] = useState(initialCount); + + // Keep the in-component count in sync if the loader-provided baseline + // changes (e.g., on navigation). + useEffect(() => { + setCount(initialCount); + }, [initialCount]); + + useEffect(() => { + if (!signedIn) return; + if (typeof EventSource === "undefined") return; + const es = new EventSource("/api/events"); + const onUnread = (e: MessageEvent) => { + try { + const parsed = JSON.parse(e.data) as { count: number }; + if (typeof parsed.count === "number") setCount(parsed.count); + } catch { + // Malformed payload — ignore. + } + }; + es.addEventListener("notifications.unread", onUnread as EventListener); + return () => { + es.removeEventListener("notifications.unread", onUnread as EventListener); + es.close(); + }; + }, [signedIn]); + + return count; +} diff --git a/apps/journal/app/jobs/notifications-fanout.integration.test.ts b/apps/journal/app/jobs/notifications-fanout.integration.test.ts new file mode 100644 index 0000000..84a60c1 --- /dev/null +++ b/apps/journal/app/jobs/notifications-fanout.integration.test.ts @@ -0,0 +1,112 @@ +import { describe, it, expect, beforeAll, afterEach } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { randomUUID } from "node:crypto"; +import { getDb } from "../lib/db.ts"; +import { activities, follows, users } from "@trails-cool/db/schema/journal"; +import { fanout } from "./notifications-fanout.ts"; +import { listForUser } from "../lib/notifications.server.ts"; + +// Same opt-in flag as the rest of the notifications integration tests. +const runIntegration = process.env.NOTIFICATIONS_INTEGRATION === "1"; + +async function makeUser(opts: { username: string; profileVisibility?: "public" | "private" }) { + const db = getDb(); + const id = randomUUID(); + await db.insert(users).values({ + id, + email: `${opts.username}@example.test`, + username: opts.username, + domain: "test.local", + profileVisibility: opts.profileVisibility ?? "public", + }); + return id; +} + +async function makeFollow(followerId: string, followedId: string, opts: { accepted: boolean } = { accepted: true }) { + const db = getDb(); + const followedUsername = (await db.select({ u: users.username }).from(users).where(eq(users.id, followedId)))[0]!.u; + await db.insert(follows).values({ + id: randomUUID(), + followerId, + followedActorIri: `https://test.local/users/${followedUsername}`, + followedUserId: followedId, + acceptedAt: opts.accepted ? new Date() : null, + }); +} + +async function makeActivity(ownerId: string, visibility: "public" | "private" | "unlisted" = "public", name = "Walk") { + const db = getDb(); + const id = randomUUID(); + await db.insert(activities).values({ + id, + ownerId, + name, + visibility, + }); + return id; +} + +async function wipe() { + const db = getDb(); + await db.execute(sql`DELETE FROM journal.notifications WHERE recipient_user_id IN (SELECT id FROM journal.users WHERE email LIKE '%@example.test')`); + await db.execute(sql`DELETE FROM journal.activities WHERE owner_id IN (SELECT id FROM journal.users WHERE email LIKE '%@example.test')`); + await db.execute(sql`DELETE FROM journal.follows WHERE follower_id IN (SELECT id FROM journal.users WHERE email LIKE '%@example.test')`); + await db.execute(sql`DELETE FROM journal.users WHERE email LIKE '%@example.test'`); +} + +describe.skipIf(!runIntegration)("notifications-fanout integration", () => { + beforeAll(async () => { + const db = getDb(); + await db.execute(sql`SELECT 1 FROM journal.notifications LIMIT 0`); + }); + afterEach(wipe); + + it("inserts exactly one row per accepted follower; pending followers are skipped", async () => { + const owner = await makeUser({ username: `nf_o_${Date.now()}` }); + const a1 = await makeUser({ username: `nf_a1_${Date.now()}` }); + const a2 = await makeUser({ username: `nf_a2_${Date.now()}` }); + const p1 = await makeUser({ username: `nf_p1_${Date.now()}` }); + const p2 = await makeUser({ username: `nf_p2_${Date.now()}` }); + await makeFollow(a1, owner, { accepted: true }); + await makeFollow(a2, owner, { accepted: true }); + await makeFollow(p1, owner, { accepted: false }); + await makeFollow(p2, owner, { accepted: false }); + + const activityId = await makeActivity(owner, "public", "Public Walk"); + await fanout(activityId); + + expect((await listForUser(a1)).length).toBe(1); + expect((await listForUser(a2)).length).toBe(1); + expect((await listForUser(p1)).length).toBe(0); + expect((await listForUser(p2)).length).toBe(0); + + const a1Rows = await listForUser(a1); + expect(a1Rows[0]?.type).toBe("activity_published"); + expect((a1Rows[0]?.payload as { activityName?: string })?.activityName).toBe("Public Walk"); + }); + + it("skips fan-out for non-public activities (defense in depth)", async () => { + const owner = await makeUser({ username: `nf_np_o_${Date.now()}` }); + const f = await makeUser({ username: `nf_np_f_${Date.now()}` }); + await makeFollow(f, owner, { accepted: true }); + + const privateAct = await makeActivity(owner, "private"); + const unlistedAct = await makeActivity(owner, "unlisted"); + await fanout(privateAct); + await fanout(unlistedAct); + + expect((await listForUser(f)).length).toBe(0); + }); + + it("is idempotent under retry — second fanout doesn't double-insert", async () => { + const owner = await makeUser({ username: `nf_id_o_${Date.now()}` }); + const f = await makeUser({ username: `nf_id_f_${Date.now()}` }); + await makeFollow(f, owner, { accepted: true }); + + const activityId = await makeActivity(owner, "public"); + await fanout(activityId); + await fanout(activityId); + + expect((await listForUser(f)).length).toBe(1); + }); +}); diff --git a/apps/journal/app/jobs/notifications-fanout.ts b/apps/journal/app/jobs/notifications-fanout.ts new file mode 100644 index 0000000..8d9581e --- /dev/null +++ b/apps/journal/app/jobs/notifications-fanout.ts @@ -0,0 +1,98 @@ +import type { JobDefinition } from "@trails-cool/jobs"; +import { and, eq, isNotNull } from "drizzle-orm"; +import { getDb } from "../lib/db.ts"; +import { activities, follows, users } from "@trails-cool/db/schema/journal"; +import { createNotification } from "../lib/notifications.server.ts"; +import { logger } from "../lib/logger.server.ts"; + +interface FanoutData { + activityId: string; +} + +/** + * Fan out an `activity_published` notification to every accepted + * follower of the activity's owner. Idempotent at the DB level via the + * `(recipient_user_id, type, subject_id)` unique partial index — a + * retry after partial failure won't double-insert. + */ +export const notificationsFanoutJob: JobDefinition = { + name: "notifications-fanout", + retryLimit: 3, + expireInSeconds: 300, + async handler(job) { + // pg-boss v12: `job` may be an array (batch) per its docs; we + // process whichever shape we get. + const batch = Array.isArray(job) ? job : [job]; + for (const item of batch) { + const data = item.data as FanoutData; + await fanout(data.activityId); + } + }, +}; + +export async function fanout(activityId: string): Promise { + const db = getDb(); + + // Load the activity + owner info needed for the payload snapshot. + const [row] = await db + .select({ + id: activities.id, + name: activities.name, + visibility: activities.visibility, + ownerId: activities.ownerId, + ownerUsername: users.username, + ownerDisplayName: users.displayName, + }) + .from(activities) + .innerJoin(users, eq(activities.ownerId, users.id)) + .where(eq(activities.id, activityId)); + + if (!row) { + logger.warn({ activityId }, "fanout: activity not found, skipping"); + return; + } + // Defense in depth — the create-side guard already filters this, but + // recheck here so the job doesn't mistakenly fan out a row that was + // later flipped to private/unlisted before the job ran. + if (row.visibility !== "public") { + logger.info({ activityId, visibility: row.visibility }, "fanout: skipping non-public activity"); + return; + } + + // Find every accepted follower of the owner. + const recipients = await db + .select({ followerId: follows.followerId }) + .from(follows) + .where( + and( + eq(follows.followedUserId, row.ownerId), + isNotNull(follows.acceptedAt), + ), + ); + + let inserted = 0; + for (const r of recipients) { + // Don't notify the owner about their own activity if they happen + // to follow themselves (shouldn't happen — followUser refuses + // self-follow — but defense in depth). + if (r.followerId === row.ownerId) continue; + const created = await createNotification({ + type: "activity_published", + recipientUserId: r.followerId, + actorUserId: row.ownerId, + subjectId: row.id, + payload: { + activityId: row.id, + activityName: row.name, + ownerUsername: row.ownerUsername, + ownerDisplayName: row.ownerDisplayName, + }, + }); + if (created) inserted += 1; + } + + logger.info( + { activityId, recipients: recipients.length, inserted }, + "notifications-fanout completed", + ); +} diff --git a/apps/journal/app/jobs/notifications-purge.ts b/apps/journal/app/jobs/notifications-purge.ts new file mode 100644 index 0000000..cf87fa9 --- /dev/null +++ b/apps/journal/app/jobs/notifications-purge.ts @@ -0,0 +1,21 @@ +import type { JobDefinition } from "@trails-cool/jobs"; +import { purgeReadOlderThan } from "../lib/notifications.server.ts"; +import { logger } from "../lib/logger.server.ts"; + +/** + * Daily retention pass. Drops notifications whose `read_at` is older + * than 90 days; unread rows are kept indefinitely so users never miss + * an event. + */ +export const notificationsPurgeJob: JobDefinition = { + name: "notifications-purge", + cron: "30 3 * * *", // daily at 03:30 UTC (offset from demo-bot-prune to spread load) + retryLimit: 1, + expireInSeconds: 60, + async handler() { + const days = 90; + const purged = await purgeReadOlderThan(days); + logger.info({ days, purged }, "notifications-purge"); + return { days, purged }; + }, +}; diff --git a/apps/journal/app/lib/activities.server.ts b/apps/journal/app/lib/activities.server.ts index 5b3562a..0dfd7f2 100644 --- a/apps/journal/app/lib/activities.server.ts +++ b/apps/journal/app/lib/activities.server.ts @@ -28,7 +28,18 @@ export async function updateActivityVisibility( .set({ visibility }) .where(and(eq(activities.id, id), eq(activities.ownerId, ownerId))) .returning({ id: activities.id }); - return result.length > 0; + if (result.length === 0) return false; + + // Notify followers when an activity becomes public. The unique + // (recipient, type, subject_id) partial index makes the fan-out + // idempotent, so toggling private→public→private→public won't spam + // followers (only the first transition per activity emits). + if (visibility === "public") { + const { enqueueOptional } = await import("./boss.server.ts"); + await enqueueOptional("notifications-fanout", { activityId: id }, { source: "updateActivityVisibility" }); + } + + return true; } export async function createActivity(ownerId: string, input: ActivityInput) { @@ -67,12 +78,21 @@ export async function createActivity(ownerId: string, input: ActivityInput) { elevationGain, elevationLoss, startedAt, + ...(input.visibility ? { visibility: input.visibility } : {}), }); if (input.gpx) { await setGeomFromGpx(id, "activities", input.gpx); } + // Public activities at creation also fan out (matches the + // updateActivityVisibility path for the case where visibility is set + // up-front rather than flipped later). + if (input.visibility === "public") { + const { enqueueOptional } = await import("./boss.server.ts"); + await enqueueOptional("notifications-fanout", { activityId: id }, { source: "createActivity" }); + } + return id; } diff --git a/apps/journal/app/lib/auth.server.ts b/apps/journal/app/lib/auth.server.ts index cc006bc..2fbcbf5 100644 --- a/apps/journal/app/lib/auth.server.ts +++ b/apps/journal/app/lib/auth.server.ts @@ -154,7 +154,7 @@ export async function registerWithMagicLink( email: string, username: string, termsVersion: string, -): Promise { +): Promise<{ token: string; code: string }> { const db = getDb(); const [existingEmail] = await db.select().from(users).where(eq(users.email, email)); @@ -175,18 +175,21 @@ export async function registerWithMagicLink( termsVersion, }); - // Create magic token for verification + // Same shape as login's createMagicToken — token for the click-through + // link, 6-digit code for paste-from-email/SMS flows (mobile). const token = randomBytes(32).toString("base64url"); + const code = generateLoginCode(); const expiresAt = new Date(Date.now() + 15 * 60 * 1000); await db.insert(magicTokens).values({ id: randomUUID(), email, token, + code, expiresAt, }); - return token; + return { token, code }; } // --- Passkey Login --- diff --git a/apps/journal/app/lib/boss.server.ts b/apps/journal/app/lib/boss.server.ts new file mode 100644 index 0000000..7e4b654 --- /dev/null +++ b/apps/journal/app/lib/boss.server.ts @@ -0,0 +1,51 @@ +// Module-level pg-boss singleton. server.ts initializes the boss + starts +// the worker; feature code (e.g., activities.server.ts) calls `getBoss()` +// to enqueue jobs against the same instance. The singleton's lifecycle +// is bound to the Node process — startWorker calls boss.start(); the +// SIGTERM handler stops it. + +// Structurally typed (we only need `send`) so we don't have to pull +// pg-boss into the journal app's dep graph just for the typedef. +interface BossLike { + send(queueName: string, data: unknown): Promise; +} + +let _boss: BossLike | null = null; + +/** Set by server.ts once the boss is created + started. */ +export function setBoss(boss: BossLike): void { + _boss = boss; +} + +/** + * Get the started pg-boss instance. Throws if called before + * server.ts has initialized it (i.e., outside a running Journal + * server context). + */ +export function getBoss(): BossLike { + if (!_boss) { + throw new Error("pg-boss not initialized — getBoss called before server bootstrap"); + } + return _boss; +} + +/** + * Best-effort enqueue: log + swallow errors so a downstream queue + * outage doesn't fail the user-visible request that triggered the + * fan-out. Use this for "fire and forget" notifications work. + */ +export async function enqueueOptional( + queue: string, + data: unknown, + ctx: Record = {}, +): Promise { + try { + const boss = getBoss(); + await boss.send(queue, data); + } catch (err) { + // Lazy import to avoid cycles in test environments where logger.server + // pulls in env-dependent setup. + const { logger } = await import("./logger.server.ts"); + logger.warn({ err, queue, ...ctx }, "boss.send failed; continuing"); + } +} diff --git a/apps/journal/app/lib/events.server.ts b/apps/journal/app/lib/events.server.ts new file mode 100644 index 0000000..4f649da --- /dev/null +++ b/apps/journal/app/lib/events.server.ts @@ -0,0 +1,48 @@ +// In-process Server-Sent Events broker. Generation hooks call +// `emitTo(userId, event, data)` after they commit; any open SSE +// connection for that user gets the event written to its stream. +// +// Single-process today. When the Journal goes multi-process, swap the +// in-memory `Map` for a Redis pub/sub adapter behind the same +// emitTo/register interface — no caller changes needed. + +interface Connection { + send: (event: string, data: unknown) => void; + close: () => void; +} + +const connections = new Map>(); + +export function register(userId: string, conn: Connection): () => void { + let set = connections.get(userId); + if (!set) { + set = new Set(); + connections.set(userId, set); + } + set.add(conn); + return () => { + const s = connections.get(userId); + if (!s) return; + s.delete(conn); + if (s.size === 0) connections.delete(userId); + }; +} + +export function emitTo(userId: string, event: string, data: unknown): void { + const set = connections.get(userId); + if (!set) return; + for (const conn of set) { + try { + conn.send(event, data); + } catch { + // Broken pipe — connection will get cleaned up on its own + // teardown path; defensively close here too. + try { conn.close(); } catch { /* ignore */ } + } + } +} + +/** Test/diagnostic helper: how many connections does `userId` have? */ +export function connectionCount(userId: string): number { + return connections.get(userId)?.size ?? 0; +} diff --git a/apps/journal/app/lib/follow.server.ts b/apps/journal/app/lib/follow.server.ts index 5165e2c..48351d2 100644 --- a/apps/journal/app/lib/follow.server.ts +++ b/apps/journal/app/lib/follow.server.ts @@ -3,6 +3,8 @@ import { eq, and, count, desc, isNull, isNotNull } from "drizzle-orm"; import { getDb } from "./db.ts"; import { users, follows } from "@trails-cool/db/schema/journal"; import { localActorIri } from "./actor-iri.ts"; +import { createNotification } from "./notifications.server.ts"; +import { logger } from "./logger.server.ts"; export class FollowError extends Error { readonly code: "self_follow" | "user_not_found" | "not_found" | "forbidden"; @@ -52,18 +54,57 @@ export async function followUser(followerId: string, targetUsername: string): Pr .from(follows) .where(and(eq(follows.followerId, followerId), eq(follows.followedActorIri, followedActorIri))); if (existing) { + // Idempotent re-follow: do NOT emit a notification (the recipient + // was notified when the row was first created). return { following: existing.acceptedAt !== null, pending: existing.acceptedAt === null }; } const acceptedAt = target.profileVisibility === "public" ? new Date() : null; + const followId = randomUUID(); await db.insert(follows).values({ - id: randomUUID(), + id: followId, followerId, followedActorIri, followedUserId: target.id, acceptedAt, }); + // Notify the target. follow_received for auto-accepted public follows; + // follow_request_received for Pending follows against private profiles. + // Errors are logged but don't fail the follow — the user-visible + // follow already succeeded. + try { + const [follower] = await db + .select({ username: users.username, displayName: users.displayName }) + .from(users) + .where(eq(users.id, followerId)); + if (follower) { + const payload = { + followerUsername: follower.username, + followerDisplayName: follower.displayName, + }; + await createNotification( + acceptedAt !== null + ? { + type: "follow_received", + recipientUserId: target.id, + actorUserId: followerId, + subjectId: followId, + payload, + } + : { + type: "follow_request_received", + recipientUserId: target.id, + actorUserId: followerId, + subjectId: followId, + payload, + }, + ); + } + } catch (err) { + logger.warn({ err, followId }, "followUser: notification emit failed"); + } + return { following: acceptedAt !== null, pending: acceptedAt === null, @@ -183,8 +224,36 @@ export async function approveFollowRequest(ownerId: string, followId: string): P isNull(follows.acceptedAt), ), ) - .returning({ id: follows.id }); - return result.length > 0; + .returning({ id: follows.id, followerId: follows.followerId }); + + if (result.length === 0) return false; + + // Notify the requester that their request landed. Lookup target's + // display info for the payload. Errors logged but don't undo the + // approve — the follow row is already accepted. + try { + const followerId = result[0]!.followerId; + const [target] = await db + .select({ username: users.username, displayName: users.displayName }) + .from(users) + .where(eq(users.id, ownerId)); + if (target) { + await createNotification({ + type: "follow_request_approved", + recipientUserId: followerId, + actorUserId: ownerId, + subjectId: followId, + payload: { + targetUsername: target.username, + targetDisplayName: target.displayName, + }, + }); + } + } catch (err) { + logger.warn({ err, followId }, "approveFollowRequest: notification emit failed"); + } + + return true; } /** diff --git a/apps/journal/app/lib/notifications.integration.test.ts b/apps/journal/app/lib/notifications.integration.test.ts new file mode 100644 index 0000000..62229fc --- /dev/null +++ b/apps/journal/app/lib/notifications.integration.test.ts @@ -0,0 +1,233 @@ +import { describe, it, expect, beforeAll, afterEach } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { randomUUID } from "node:crypto"; +import { getDb } from "./db.ts"; +import { users, notifications } from "@trails-cool/db/schema/journal"; +import { + createNotification, + listForUser, + countUnread, + markRead, + markAllRead, + purgeReadOlderThan, +} from "./notifications.server.ts"; +import { + followUser, + approveFollowRequest, + listPendingFollowRequests, +} from "./follow.server.ts"; + +// Opt-in: hits real Postgres. Same convention as demo-bot / follow integration. +const runIntegration = process.env.NOTIFICATIONS_INTEGRATION === "1"; + +async function makeUser(opts: { username: string; profileVisibility?: "public" | "private" }) { + const db = getDb(); + const id = randomUUID(); + await db.insert(users).values({ + id, + email: `${opts.username}@example.test`, + username: opts.username, + domain: "test.local", + profileVisibility: opts.profileVisibility ?? "public", + }); + return id; +} + +async function wipe() { + const db = getDb(); + await db.execute(sql`DELETE FROM journal.notifications WHERE recipient_user_id IN (SELECT id FROM journal.users WHERE email LIKE '%@example.test')`); + await db.execute(sql`DELETE FROM journal.follows WHERE follower_id IN (SELECT id FROM journal.users WHERE email LIKE '%@example.test')`); + await db.execute(sql`DELETE FROM journal.users WHERE email LIKE '%@example.test'`); +} + +describe.skipIf(!runIntegration)("notifications.server integration", () => { + beforeAll(async () => { + const db = getDb(); + await db.execute(sql`SELECT 1 FROM journal.notifications LIMIT 0`); + }); + afterEach(wipe); + + it("createNotification persists payload + version and is idempotent on (recipient,type,subject)", async () => { + const a = await makeUser({ username: `n_a_${Date.now()}` }); + const b = await makeUser({ username: `n_b_${Date.now()}` }); + const subject = randomUUID(); + + const first = await createNotification({ + type: "activity_published", + recipientUserId: a, + actorUserId: b, + subjectId: subject, + payload: { activityId: subject, activityName: "Walk", ownerUsername: "b", ownerDisplayName: null }, + }); + expect(first).toBe(true); + + // Re-insert with same (recipient, type, subject_id) is a no-op. + const second = await createNotification({ + type: "activity_published", + recipientUserId: a, + actorUserId: b, + subjectId: subject, + payload: { activityId: subject, activityName: "Walk", ownerUsername: "b", ownerDisplayName: null }, + }); + expect(second).toBe(false); + + const rows = await listForUser(a); + expect(rows.length).toBe(1); + expect(rows[0]?.payloadVersion).toBe(1); + expect(rows[0]?.payload).toMatchObject({ activityId: subject }); + }); + + it("countUnread excludes read; markRead is owner-bound and idempotent", async () => { + const a = await makeUser({ username: `n_mr_a_${Date.now()}` }); + const c = await makeUser({ username: `n_mr_c_${Date.now()}` }); + + await createNotification({ + type: "follow_received", + recipientUserId: a, + actorUserId: null, + subjectId: null, + payload: { followerUsername: "x", followerDisplayName: null }, + }); + expect(await countUnread(a)).toBe(1); + + const all = await listForUser(a); + const id = all[0]!.id; + + // Foreign user can't mark a's notification read. + expect(await markRead(c, id)).toBe(false); + expect(await countUnread(a)).toBe(1); + + // Owner can. + expect(await markRead(a, id)).toBe(true); + expect(await countUnread(a)).toBe(0); + + // Idempotent — already read, but row still belongs to a, so returns true. + expect(await markRead(a, id)).toBe(true); + }); + + it("markAllRead clears every unread row of the caller and only the caller", async () => { + const a = await makeUser({ username: `n_mar_a_${Date.now()}` }); + const b = await makeUser({ username: `n_mar_b_${Date.now()}` }); + + await createNotification({ + type: "follow_received", + recipientUserId: a, + actorUserId: null, + subjectId: null, + payload: { followerUsername: "x", followerDisplayName: null }, + }); + await createNotification({ + type: "follow_received", + recipientUserId: a, + actorUserId: null, + subjectId: null, + payload: { followerUsername: "y", followerDisplayName: null }, + }); + await createNotification({ + type: "follow_received", + recipientUserId: b, + actorUserId: null, + subjectId: null, + payload: { followerUsername: "z", followerDisplayName: null }, + }); + + const updated = await markAllRead(a); + expect(updated).toBe(2); + expect(await countUnread(a)).toBe(0); + expect(await countUnread(b)).toBe(1); + }); + + it("purgeReadOlderThan removes only read rows past the window", async () => { + const a = await makeUser({ username: `n_pr_a_${Date.now()}` }); + + const db = getDb(); + // Old read (should be purged), old unread (kept), recent read (kept). + await db.insert(notifications).values([ + { + id: randomUUID(), + recipientUserId: a, + type: "follow_received", + actorUserId: null, + subjectId: null, + payload: { followerUsername: "old-read" }, + payloadVersion: 1, + readAt: new Date(Date.now() - 100 * 24 * 60 * 60 * 1000), + createdAt: new Date(Date.now() - 120 * 24 * 60 * 60 * 1000), + }, + { + id: randomUUID(), + recipientUserId: a, + type: "follow_received", + actorUserId: null, + subjectId: null, + payload: { followerUsername: "old-unread" }, + payloadVersion: 1, + readAt: null, + createdAt: new Date(Date.now() - 120 * 24 * 60 * 60 * 1000), + }, + { + id: randomUUID(), + recipientUserId: a, + type: "follow_received", + actorUserId: null, + subjectId: null, + payload: { followerUsername: "recent-read" }, + payloadVersion: 1, + readAt: new Date(Date.now() - 1 * 24 * 60 * 60 * 1000), + createdAt: new Date(Date.now() - 5 * 24 * 60 * 60 * 1000), + }, + ]); + + const purged = await purgeReadOlderThan(90); + expect(purged).toBe(1); + + const survivors = await db + .select({ payload: notifications.payload }) + .from(notifications) + .where(eq(notifications.recipientUserId, a)); + const names = survivors + .map((s) => (s.payload as { followerUsername?: string } | null)?.followerUsername) + .sort(); + expect(names).toEqual(["old-unread", "recent-read"]); + }); + + it("followUser → follow_received emitted for public-target auto-accept; idempotent re-follow does NOT double-emit", async () => { + const a = await makeUser({ username: `n_fr_a_${Date.now()}` }); + const b = await makeUser({ username: `n_fr_b_${Date.now()}` }); + const bRow = (await getDb().select().from(users).where(eq(users.id, b)))[0]!; + + await followUser(a, bRow.username); + let bRows = await listForUser(b); + expect(bRows.length).toBe(1); + expect(bRows[0]?.type).toBe("follow_received"); + + // Re-follow is idempotent at the follow layer — and must NOT emit again. + await followUser(a, bRow.username); + bRows = await listForUser(b); + expect(bRows.length).toBe(1); + }); + + it("followUser private-target → follow_request_received; approveFollowRequest → follow_request_approved", async () => { + const a = await makeUser({ username: `n_fp_a_${Date.now()}` }); + const b = await makeUser({ username: `n_fp_b_${Date.now()}`, profileVisibility: "private" }); + const bRow = (await getDb().select().from(users).where(eq(users.id, b)))[0]!; + + await followUser(a, bRow.username); + const bRows = await listForUser(b); + expect(bRows.length).toBe(1); + expect(bRows[0]?.type).toBe("follow_request_received"); + + const reqs = await listPendingFollowRequests(b); + expect(reqs.length).toBe(1); + + await approveFollowRequest(b, reqs[0]!.id); + const aRows = await listForUser(a); + expect(aRows.length).toBe(1); + expect(aRows[0]?.type).toBe("follow_request_approved"); + + // Idempotent re-approval must not double-emit. + await approveFollowRequest(b, reqs[0]!.id); + const aRowsAfter = await listForUser(a); + expect(aRowsAfter.length).toBe(1); + }); +}); diff --git a/apps/journal/app/lib/notifications.server.ts b/apps/journal/app/lib/notifications.server.ts new file mode 100644 index 0000000..63e1fef --- /dev/null +++ b/apps/journal/app/lib/notifications.server.ts @@ -0,0 +1,190 @@ +import { randomUUID } from "node:crypto"; +import { and, count, desc, eq, isNull, lt, sql } from "drizzle-orm"; +import { getDb } from "./db.ts"; +import { notifications } from "@trails-cool/db/schema/journal"; +import type { NotificationType } from "@trails-cool/db/schema/journal"; +import { emitTo } from "./events.server.ts"; +import { + PAYLOAD_VERSION, + type ActivityPayloadV1, + type ApprovalPayloadV1, + type FollowPayloadV1, +} from "./notifications/payload.ts"; + +/** + * Push the current unread count to all of `userId`'s open SSE + * connections. Called after any state change that could affect the + * count: createNotification (new row), markRead (one less unread), + * markAllRead (zero unread). + */ +async function emitUnreadCount(userId: string): Promise { + const c = await countUnread(userId); + emitTo(userId, "notifications.unread", { count: c }); +} + +// Discriminated union of (type, payload) so callers can't pass the +// wrong payload shape for a given event type. PAYLOAD_VERSION is set +// internally; callers don't pick a version. +type CreateInput = + | { type: "follow_request_received"; recipientUserId: string; actorUserId: string | null; subjectId: string | null; payload: FollowPayloadV1 } + | { type: "follow_received"; recipientUserId: string; actorUserId: string | null; subjectId: string | null; payload: FollowPayloadV1 } + | { type: "follow_request_approved"; recipientUserId: string; actorUserId: string | null; subjectId: string | null; payload: ApprovalPayloadV1 } + | { type: "activity_published"; recipientUserId: string; actorUserId: string | null; subjectId: string | null; payload: ActivityPayloadV1 }; + +/** + * Create a notification. When `subjectId` is set, idempotent against + * the partial unique index `(recipient_user_id, type, subject_id) WHERE + * subject_id IS NOT NULL` so fan-out job retries can't double-insert. + * When `subjectId` is null (1:1 follow events) idempotency is enforced + * upstream by the follow hooks (they only emit on a *new* row), so a + * plain insert is safe and avoids ON CONFLICT inference against a + * partial index, which Postgres rejects. + */ +export async function createNotification(input: CreateInput): Promise { + const db = getDb(); + const values = { + id: randomUUID(), + recipientUserId: input.recipientUserId, + type: input.type, + actorUserId: input.actorUserId, + subjectId: input.subjectId, + payload: input.payload as unknown as Record, + payloadVersion: PAYLOAD_VERSION, + }; + + const result = input.subjectId + ? await db + .insert(notifications) + .values(values) + .onConflictDoNothing({ + target: [notifications.recipientUserId, notifications.type, notifications.subjectId], + // Match the partial unique index's predicate so Postgres can + // infer the arbiter index (`WHERE subject_id IS NOT NULL`). + // `onConflictDoNothing` only exposes `where` (no targetWhere). + where: sql`${notifications.subjectId} IS NOT NULL`, + }) + .returning({ id: notifications.id }) + : await db.insert(notifications).values(values).returning({ id: notifications.id }); + + if (result.length > 0) { + // Push refreshed count to any open SSE connections for this + // recipient. Emit happens after the row commits so the count + // reflects reality. + await emitUnreadCount(input.recipientUserId); + return true; + } + return false; +} + +export interface NotificationRow { + id: string; + type: NotificationType; + actorUserId: string | null; + subjectId: string | null; + payload: Record | null; + payloadVersion: number; + readAt: Date | null; + createdAt: Date; +} + +const PAGE_SIZE = 50; + +export async function listForUser( + userId: string, + opts: { page?: number } = {}, +): Promise { + const db = getDb(); + const page = Math.max(1, opts.page ?? 1); + const rows = await db + .select({ + id: notifications.id, + type: notifications.type, + actorUserId: notifications.actorUserId, + subjectId: notifications.subjectId, + payload: notifications.payload, + payloadVersion: notifications.payloadVersion, + readAt: notifications.readAt, + createdAt: notifications.createdAt, + }) + .from(notifications) + .where(eq(notifications.recipientUserId, userId)) + .orderBy(desc(notifications.createdAt)) + .limit(PAGE_SIZE) + .offset((page - 1) * PAGE_SIZE); + return rows; +} + +export async function countUnread(userId: string): Promise { + const db = getDb(); + const [row] = await db + .select({ n: count() }) + .from(notifications) + .where( + and( + eq(notifications.recipientUserId, userId), + isNull(notifications.readAt), + ), + ); + return row?.n ?? 0; +} + +/** + * Mark a single notification read. Owner-bound: only the recipient can + * mark their own. Returns true on success, false if the row doesn't + * exist or doesn't belong to the caller. Idempotent for already-read. + */ +export async function markRead(ownerId: string, id: string): Promise { + const db = getDb(); + const result = await db + .update(notifications) + .set({ readAt: new Date() }) + .where( + and( + eq(notifications.id, id), + eq(notifications.recipientUserId, ownerId), + ), + ) + .returning({ id: notifications.id }); + if (result.length === 0) return false; + await emitUnreadCount(ownerId); + return true; +} + +/** + * Mark all of `ownerId`'s unread notifications read. Returns the + * number of rows touched. + */ +export async function markAllRead(ownerId: string): Promise { + const db = getDb(); + const result = await db + .update(notifications) + .set({ readAt: new Date() }) + .where( + and( + eq(notifications.recipientUserId, ownerId), + isNull(notifications.readAt), + ), + ) + .returning({ id: notifications.id }); + if (result.length > 0) await emitUnreadCount(ownerId); + return result.length; +} + +/** + * Retention: drop read notifications older than `days`. Unread rows + * survive indefinitely. Called by the daily pg-boss job. + */ +export async function purgeReadOlderThan(days: number): Promise { + const db = getDb(); + const cutoff = new Date(Date.now() - days * 24 * 60 * 60 * 1000); + const result = await db + .delete(notifications) + .where( + and( + sql`${notifications.readAt} IS NOT NULL`, + lt(notifications.readAt, cutoff), + ), + ) + .returning({ id: notifications.id }); + return result.length; +} diff --git a/apps/journal/app/lib/notifications/link-for.test.ts b/apps/journal/app/lib/notifications/link-for.test.ts new file mode 100644 index 0000000..0b501ce --- /dev/null +++ b/apps/journal/app/lib/notifications/link-for.test.ts @@ -0,0 +1,75 @@ +import { describe, it, expect } from "vitest"; +import { linkFor } from "./link-for.ts"; + +describe("linkFor", () => { + it("follow_received uses payload.followerUsername", () => { + const link = linkFor({ + type: "follow_received", + subjectId: null, + payloadVersion: 1, + payload: { followerUsername: "alice", followerDisplayName: null }, + }); + expect(link.web).toBe("/users/alice"); + expect(link.mobile).toBe("trails:/users/alice"); + expect(link.email).toMatch(/^https?:\/\//); + expect(link.email).toMatch(/\/users\/alice$/); + }); + + it("follow_received falls back to / when payload is missing", () => { + const link = linkFor({ + type: "follow_received", + subjectId: null, + payloadVersion: 1, + payload: null, + }); + expect(link.web).toBe("/"); + }); + + it("follow_request_received always points at the requests page", () => { + const link = linkFor({ + type: "follow_request_received", + subjectId: null, + payloadVersion: 1, + payload: { followerUsername: "bob", followerDisplayName: null }, + }); + expect(link.web).toBe("/follows/requests"); + }); + + it("follow_request_approved uses payload.targetUsername", () => { + const link = linkFor({ + type: "follow_request_approved", + subjectId: null, + payloadVersion: 1, + payload: { targetUsername: "carol", targetDisplayName: null }, + }); + expect(link.web).toBe("/users/carol"); + }); + + it("activity_published prefers subjectId, falls back to payload.activityId", () => { + const fromSubject = linkFor({ + type: "activity_published", + subjectId: "sub-id-1", + payloadVersion: 1, + payload: { activityId: "payload-id", activityName: "x", ownerUsername: "o", ownerDisplayName: null }, + }); + expect(fromSubject.web).toBe("/activities/sub-id-1"); + + const fromPayload = linkFor({ + type: "activity_published", + subjectId: null, + payloadVersion: 1, + payload: { activityId: "payload-id", activityName: "x", ownerUsername: "o", ownerDisplayName: null }, + }); + expect(fromPayload.web).toBe("/activities/payload-id"); + }); + + it("activity_published falls back to / when both subjectId and payload are missing", () => { + const link = linkFor({ + type: "activity_published", + subjectId: null, + payloadVersion: 1, + payload: null, + }); + expect(link.web).toBe("/"); + }); +}); diff --git a/apps/journal/app/lib/notifications/link-for.ts b/apps/journal/app/lib/notifications/link-for.ts new file mode 100644 index 0000000..e551854 --- /dev/null +++ b/apps/journal/app/lib/notifications/link-for.ts @@ -0,0 +1,65 @@ +// Single source of truth for "what URL does this notification link to?" +// Every renderer (web loader, future mobile push formatter, future email +// formatter) calls linkFor; the type→URL mapping lives here, in one +// place, in three flavors so the same logical destination renders +// correctly per platform. + +import type { NotificationType } from "@trails-cool/db/schema/journal"; +import { readPayload } from "./payload.ts"; + +export interface LinkBundle { + web: string; // app-relative path for in-app navigation + mobile?: string; // trails:// scheme for native deep linking + email?: string; // absolute https:// URL for email click-through +} + +export interface NotificationForLink { + type: NotificationType; + subjectId: string | null; + payload: Record | null; + payloadVersion: number; +} + +export function linkFor(n: NotificationForLink): LinkBundle { + const origin = process.env.ORIGIN ?? "https://trails.cool"; + const p = readPayload(n.type, n.payloadVersion, n.payload); + + switch (n.type) { + case "follow_received": + case "follow_request_received": { + // The actionable surface for a request lives at /follows/requests + // (where Approve/Reject is); a received auto-accept notification + // links to the new follower's profile. Both fall back to the + // payload's followerUsername if the actor account is gone. + const username = + p && "followerUsername" in p ? p.followerUsername : null; + const path = + n.type === "follow_request_received" + ? "/follows/requests" + : username + ? `/users/${username}` + : "/"; + return buildBundle(origin, path); + } + case "follow_request_approved": { + const username = + p && "targetUsername" in p ? p.targetUsername : null; + const path = username ? `/users/${username}` : "/"; + return buildBundle(origin, path); + } + case "activity_published": { + const activityId = + n.subjectId ?? (p && "activityId" in p ? p.activityId : null); + const path = activityId ? `/activities/${activityId}` : "/"; + return buildBundle(origin, path); + } + } +} + +function buildBundle(origin: string, path: string): LinkBundle { + return { + web: path, + mobile: `trails:/${path.startsWith("/") ? "" : "/"}${path.replace(/^\//, "")}`, + email: `${origin.replace(/\/$/, "")}${path}`, + }; +} diff --git a/apps/journal/app/lib/notifications/payload.ts b/apps/journal/app/lib/notifications/payload.ts new file mode 100644 index 0000000..f7e4742 --- /dev/null +++ b/apps/journal/app/lib/notifications/payload.ts @@ -0,0 +1,70 @@ +// Per-type payload snapshots stored at notification creation time. +// Lets future mobile push / email / RSS renderers display the +// notification without a fresh DB lookup, and lets the web renderer +// fall back to the snapshot when the live record is gone (subject +// deleted, gone private, etc.). +// +// Versioning: every payload type carries an associated `payloadVersion` +// (stored in a separate column on the row, not embedded in the JSON). +// To evolve a payload schema, bump the version and update renderers to +// handle both shapes. Retention naturally ages out old versions. + +import type { NotificationType } from "@trails-cool/db/schema/journal"; + +export const PAYLOAD_VERSION = 1; + +export interface FollowPayloadV1 { + followerUsername: string; + followerDisplayName: string | null; +} + +export interface ApprovalPayloadV1 { + targetUsername: string; + targetDisplayName: string | null; +} + +export interface ActivityPayloadV1 { + activityId: string; + activityName: string; + ownerUsername: string; + ownerDisplayName: string | null; +} + +export type NotificationPayload = + | { type: "follow_request_received"; v: 1; payload: FollowPayloadV1 } + | { type: "follow_received"; v: 1; payload: FollowPayloadV1 } + | { type: "follow_request_approved"; v: 1; payload: ApprovalPayloadV1 } + | { type: "activity_published"; v: 1; payload: ActivityPayloadV1 }; + +// Narrow the `payload` JSONB to the matching shape based on `type` + `v`. +// Renderers should call this rather than indexing into the loose Record. +export function readPayload( + type: NotificationType, + version: number, + payload: Record | null, +): NotificationPayload["payload"] | null { + if (!payload || version !== 1) return null; + switch (type) { + case "follow_request_received": + case "follow_received": + return { + followerUsername: String(payload.followerUsername ?? ""), + followerDisplayName: + (payload.followerDisplayName as string | null | undefined) ?? null, + }; + case "follow_request_approved": + return { + targetUsername: String(payload.targetUsername ?? ""), + targetDisplayName: + (payload.targetDisplayName as string | null | undefined) ?? null, + }; + case "activity_published": + return { + activityId: String(payload.activityId ?? ""), + activityName: String(payload.activityName ?? ""), + ownerUsername: String(payload.ownerUsername ?? ""), + ownerDisplayName: + (payload.ownerDisplayName as string | null | undefined) ?? null, + }; + } +} diff --git a/apps/journal/app/root.tsx b/apps/journal/app/root.tsx index 3b52242..6101cad 100644 --- a/apps/journal/app/root.tsx +++ b/apps/journal/app/root.tsx @@ -8,6 +8,7 @@ import { detectLocale } from "@trails-cool/i18n"; import { getSessionUser } from "~/lib/auth.server"; import { LocaleProvider } from "~/components/LocaleContext"; import { AlphaBanner } from "~/components/AlphaBanner"; +import { useUnreadNotifications } from "~/hooks/useUnreadNotifications"; import { Footer } from "~/components/Footer"; import { initSentryClient, stopSentryClient } from "~/lib/sentry.client"; import { TERMS_VERSION } from "~/lib/legal"; @@ -67,27 +68,38 @@ export async function loader({ request }: Route.LoaderArgs) { // users. Hidden behind a dynamic import so the root layout doesn't // pull in the follow module on anonymous renders. let pendingFollowRequests = 0; + let unreadNotifications = 0; if (user) { const { countPendingFollowRequests } = await import("./lib/follow.server.ts"); - pendingFollowRequests = await countPendingFollowRequests(user.id); + const { countUnread } = await import("./lib/notifications.server.ts"); + [pendingFollowRequests, unreadNotifications] = await Promise.all([ + countPendingFollowRequests(user.id), + countUnread(user.id), + ]); } return { user: user ? { id: user.id, username: user.username } : null, locale, pendingFollowRequests, + unreadNotifications, }; } function NavBar({ user, pendingFollowRequests, + unreadNotifications, }: { user: { id: string; username: string } | null; pendingFollowRequests: number; + unreadNotifications: number; }) { const { t } = useTranslation("journal"); const location = useLocation(); + // Live-updating unread count for the navbar badge. Loader value is + // the SSR baseline; SSE pushes overrides after first event. + const liveUnread = useUnreadNotifications(unreadNotifications, user !== null); const isActive = (path: string) => location.pathname === path || location.pathname.startsWith(path + "/"); @@ -123,6 +135,33 @@ function NavBar({
{user ? ( <> + + + {liveUnread > 0 && ( + + {liveUnread} + + )} + { if (user) { initSentryClient(); @@ -189,7 +229,11 @@ export default function App({ loaderData }: Route.ComponentProps) { return ( - +