feat(api): implement event-driven audit logging and activity streaming (fixes #699)

app/api/analytics/audit-trail/route.ts

@@ -0,0 +1,57 @@
+import { NextRequest, NextResponse } from "next/server";
+import { getServerSession } from "next-auth";
+import { authOptions } from "@/lib/auth";
+import prisma from "@/lib/prisma";
+
+export async function GET(req: NextRequest) {
+    try {
+        const session = await getServerSession(authOptions);
+
+        if (!session?.user?.email) {
+            return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
+        }
+
+        const user = await prisma.user.findUnique({
+            where: { email: session.user.email },
+        });
+
+        if (!user) {
+            return NextResponse.json({ error: "User not found" }, { status: 404 });
+        }
+
+        // Pagination
+        const { searchParams } = new URL(req.url);
+        const page = parseInt(searchParams.get("page") || "1", 10);
+        const limit = parseInt(searchParams.get("limit") || "10", 10);
+
+        const skip = (page - 1) * limit;
+
+        const [auditLogs, totalCount] = await Promise.all([
+            prisma.auditLog.findMany({
+                where: { actorId: user.id },
+                orderBy: { createdAt: "desc" },
+                skip,
+                take: limit,
+            }),
+            prisma.auditLog.count({
+                where: { actorId: user.id },
+            }),
+        ]);
+
+        return NextResponse.json({
+            auditLogs,
+            pagination: {
+                page,
+                limit,
+                totalCount,
+                totalPages: Math.ceil(totalCount / limit),
+            }
+        });
+    } catch (error) {
+        console.error("Failed to fetch audit trail:", error);
+        return NextResponse.json(
+            { error: "Internal server error" },
+            { status: 500 }
+        );
+    }
+}

app/api/links/[id]/route.ts

@@ -10,6 +10,7 @@ import { validateUrlBackend } from "@/lib/urlValidation";
 import { PLATFORM_ICONS } from "@/lib/platformIcons";
 import { checkRateLimit } from "@/lib/rateLimit";
 import { invalidateProfileCache } from "@/lib/profileCache";
+import { eventBus } from "@/lib/event-bus";
 
 const LINK_MUTATE_LIMIT = 20;
 const LINK_MUTATE_WINDOW_MS = 60 * 1000; // 20 updates/deletes per minute per user
@@ -222,6 +223,15 @@ export async function PUT(
     // The updated link may be rendered on the public profile — purge the cache.
     await invalidateProfileCache(link.userId);
 
+    eventBus.publish({
+        actorId: link.userId,
+        actionType: link.isGroup ? "UPDATE_GROUP" : "UPDATE_LINK",
+        resourceId: updatedLink.id,
+        oldState: link,
+        newState: updatedLink,
+        ipAddress: req.headers.get("x-forwarded-for") || undefined,
+    });
+
     return NextResponse.json({ success: true, link: updatedLink });
   } catch (err: unknown) {
     const error = err as { code?: string; proposedRoute?: string };
@@ -325,6 +335,14 @@ export async function DELETE(
     // Deleted links disappear from the public profile — purge the cache.
     await invalidateProfileCache(link.userId);
 
+    eventBus.publish({
+        actorId: link.userId,
+        actionType: "DELETE_GROUP",
+        resourceId: link.id,
+        oldState: link,
+        ipAddress: req.headers.get("x-forwarded-for") || undefined,
+    });
+
     return NextResponse.json({ success: true });
   }
 
@@ -336,6 +354,14 @@ export async function DELETE(
   // Deleted links disappear from the public profile — purge the cache.
   await invalidateProfileCache(link.userId);
 
+  eventBus.publish({
+      actorId: link.userId,
+      actionType: "DELETE_LINK",
+      resourceId: link.id,
+      oldState: link,
+      ipAddress: req.headers.get("x-forwarded-for") || undefined,
+  });
+
   return NextResponse.json({ success: true });
 }
 

app/api/links/route.ts

@@ -15,6 +15,7 @@
 import { PLATFORM_ICONS } from "@/lib/platformIcons";
 import { rateLimit } from "@/lib/rateLimit";
 import { invalidateProfileCache } from "@/lib/profileCache";
+import { eventBus } from "@/lib/event-bus";
 
 // Maximum number of links a single user can add to their profile.
 // Prevents unbounded database growth and degraded public profile performance.
@@ -88,6 +89,14 @@
             // New link is public — purge the cached public profile.
             await invalidateProfileCache(user.id);
 
+            eventBus.publish({
+                actorId: user.id,
+                actionType: "CREATE_GROUP",
+                resourceId: link.id,
+                newState: link,
+                ipAddress: req.headers.get("x-forwarded-for") || req.ip || undefined,
+            });
+
             return NextResponse.json({ link: { ...link, children: [] } });
         } catch (err: unknown) {
             const error = err as { code?: string };
@@ -247,6 +256,14 @@
         // New link is public — purge the cached public profile.
         await invalidateProfileCache(user.id);
 
+        eventBus.publish({
+            actorId: user.id,
+            actionType: "CREATE_LINK",
+            resourceId: link.id,
+            newState: link,
+            ipAddress: req.headers.get("x-forwarded-for") || req.ip || undefined,
+        });
+
         return NextResponse.json({ link });
     } catch (err: unknown) {
         const error = err as { code?: string };

app/api/profile/update/route.ts

@@ -3,6 +3,7 @@ import { upsertProfileDraft } from "@/lib/profileWorkflow";
 import { getServerSession } from "next-auth";
 import { authOptions } from "@/lib/auth";
 import { invalidateProfileCache } from "@/lib/profileCache";
+import { eventBus } from "@/lib/event-bus";
 
 export async function PATCH(req: NextRequest) {
   try {
@@ -29,6 +30,22 @@ export async function PATCH(req: NextRequest) {
     // so any published (live) version is never served stale.
     await invalidateProfileCache(userId);
 
+    eventBus.publish({
+      actorId: userId,
+      actionType: "UPDATE_PROFILE_DRAFT",
+      resourceId: draft.id,
+      newState: {
+        username,
+        name,
+        bio,
+        image,
+        themeType,
+        themeColor,
+        themeCustom,
+      },
+      ipAddress: req.headers.get("x-forwarded-for") || req.ip || undefined,
+    });
+
     return NextResponse.json({ success: true, draft }, { status: 200 });
 
   } catch (error: unknown) {

lib/event-bus.ts

@@ -0,0 +1,42 @@
+import prisma from "@/lib/prisma";
+
+export interface AuditEvent {
+    actorId: string;
+    actionType: string;
+    resourceId?: string;
+    oldState?: any;
+    newState?: any;
+    ipAddress?: string;
+}
+
+class EventBus {
+    /**
+     * Publishes an audit event to the database asynchronously.
+     * This uses a "fire and forget" pattern by intentionally not returning the promise
+     * (but wrapping it in a catch to prevent UnhandledPromiseRejection warnings),
+     * enabling CQRS-lite behavior where writes don't block the HTTP response.
+     */
+    publish(event: AuditEvent): void {
+        const payload = {
+            actorId: event.actorId,
+            actionType: event.actionType,
+            resourceId: event.resourceId,
+            oldState: event.oldState ? JSON.stringify(event.oldState) : undefined,
+            newState: event.newState ? JSON.stringify(event.newState) : undefined,
+            ipAddress: event.ipAddress,
+        };
+
+        // Fire and forget
+        Promise.resolve().then(async () => {
+            try {
+                await prisma.auditLog.create({
+                    data: payload,
+                });
+            } catch (error) {
+                console.error("[EventBus] Failed to publish audit log:", error);
+            }
+        });
+    }
+}
+
+export const eventBus = new EventBus();

prisma/schema.prisma

@@ -53,6 +53,7 @@ model User {
   subscribers   Subscriber[]
   isVerified    Boolean  @default(false)
   role          Role     @default(USER)
+  auditLogs     AuditLog[]
   createdAt     DateTime @default(now())
 
   theme         String   @default("default")
@@ -325,4 +326,18 @@ model Subscriber {
 
   @@unique([email, userId])
   @@index([userId])
+}
+
+model AuditLog {
+  id          String   @id @default(uuid())
+  actorId     String
+  actor       User     @relation(fields: [actorId], references: [id], onDelete: Cascade)
+  actionType  String
+  resourceId  String?
+  oldState    Json?
+  newState    Json?
+  ipAddress   String?
+  createdAt   DateTime @default(now())
+
+  @@index([actorId, createdAt])
 }