feat: wire up listenarr and abs routes in index.ts

2026-07-01 08:42:01 -04:00
parent 82e1f1b010
commit 2134537523
4 changed files with 730 additions and 0 deletions
@@ -22,8 +22,10 @@ const envPath =
 dotenv.config({ path: envPath });

 import { errorHandler } from "./middleware/errorHandler";
+import { absRouter } from "./routes/abs";
 import { adminRouter } from "./routes/admin";
 import { authRouter } from "./routes/auth";
+import { listenarrRouter } from "./routes/listenarr";
 import { overseerRouter } from "./routes/overseer";
 import { tautulliRouter } from "./routes/tautulli";
 import { transactionsRouter } from "./routes/transactions";
@@ -54,6 +56,8 @@ app.use(
 		crossOriginResourcePolicy: { policy: "cross-origin" },
 	}),
 );
+// Trust proxy — behind Caddy reverse proxy
+app.set("trust proxy", 1);
 app.use(
 	cors({
 		origin: allowedOrigins,
@@ -87,6 +91,8 @@ app.use("/api/wallet", walletRouter);
 app.use("/api/transactions", transactionsRouter);
 app.use("/api/admin", adminRouter);
 app.use("/api/tautulli", tautulliRouter);
+app.use("/api/abs", absRouter);
+app.use("/api/listenarr", listenarrRouter);
 app.use("/api/overseer", overseerRouter);
 app.use("/webhooks", webhookRouter);

@@ -0,0 +1,314 @@
+import { Router } from "express";
+import {
+  type AuthenticatedRequest,
+  authenticate,
+  requireAdmin,
+} from "../middleware/auth";
+import { asyncHandler } from "../middleware/errorHandler";
+import {
+  getAbsStatus,
+  getListeningSessions,
+  backfillAudiobookHistory,
+  getUsers,
+} from "../services/audiobookshelf";
+import { prisma } from "../utils/prisma";
+
+const router = Router();
+
+// Get Audiobookshelf connection status
+router.get(
+  "/status",
+  authenticate,
+  requireAdmin,
+  asyncHandler(async (_req: AuthenticatedRequest, res) => {
+    const status = await getAbsStatus();
+    res.json(status);
+  })
+);
+
+// Get raw listening sessions from Audiobookshelf
+router.get(
+  "/sessions",
+  authenticate,
+  requireAdmin,
+  asyncHandler(async (req: AuthenticatedRequest, res) => {
+    const { page = "0", limit = "50" } = req.query;
+    const pageNum = parseInt(page as string);
+    const limitNum = Math.min(parseInt(limit as string), 100);
+    const data = await getListeningSessions(limitNum, pageNum);
+    res.json(data);
+  })
+);
+
+// Get aggregate stats from Audiobookshelf
+router.get(
+  "/stats",
+  authenticate,
+  requireAdmin,
+  asyncHandler(async (_req: AuthenticatedRequest, res) => {
+    try {
+      // Fetch all listening sessions and aggregate
+      let page = 0;
+      let allSessions: any[] = [];
+      let hasMore = true;
+
+      while (hasMore && page < 10) {
+        const { sessions, numPages } = await getListeningSessions(100, page);
+        allSessions = allSessions.concat(sessions);
+        page++;
+        if (page >= numPages) hasMore = false;
+      }
+
+      const now = Date.now();
+      const sevenDaysAgo = now - 7 * 24 * 60 * 60 * 1000;
+      const thirtyDaysAgo = now - 30 * 24 * 60 * 60 * 1000;
+
+      const totalListeningSeconds = allSessions.reduce(
+        (sum, s) => sum + (s.timeListening || 0),
+        0
+      );
+      const last7Days = allSessions.filter((s) => s.date >= sevenDaysAgo);
+      const last30Days = allSessions.filter((s) => s.date >= thirtyDaysAgo);
+
+      // Per-user aggregation
+      const userMap: Record<string, { username: string; seconds: number; sessions: number }> = {};
+      for (const session of allSessions) {
+        const uid = session.userId || "unknown";
+        if (!userMap[uid]) {
+          userMap[uid] = { username: uid, seconds: 0, sessions: 0 };
+        }
+        userMap[uid].seconds += session.timeListening || 0;
+        userMap[uid].sessions++;
+      }
+
+      res.json({
+        totalSessions: allSessions.length,
+        totalListeningHours: Math.round((totalListeningSeconds / 3600) * 100) / 100,
+        last7DaysSessions: last7Days.length,
+        last7DaysHours: Math.round(
+          (last7Days.reduce((s, c) => s + (c.timeListening || 0), 0) / 3600) * 100
+        ) / 100,
+        last30DaysSessions: last30Days.length,
+        last30DaysHours: Math.round(
+          (last30Days.reduce((s, c) => s + (c.timeListening || 0), 0) / 3600) * 100
+        ) / 100,
+        topListeners: Object.values(userMap)
+          .sort((a, b) => b.seconds - a.seconds)
+          .slice(0, 20)
+          .map((u) => ({
+            ...u,
+            hours: Math.round((u.seconds / 3600) * 100) / 100,
+          })),
+      });
+    } catch (error: any) {
+      res.status(500).json({ error: error?.message || "Failed to fetch ABS stats" });
+    }
+  })
+);
+
+// Sync/backfill audiobook listening sessions for all users
+router.post(
+  "/sync",
+  authenticate,
+  requireAdmin,
+  asyncHandler(async (req: AuthenticatedRequest, res) => {
+    const { userId } = req.body;
+
+    if (userId) {
+      // Backfill a single user
+      const user = await prisma.user.findUnique({
+        where: { id: userId },
+      });
+      if (!user) return res.status(404).json({ error: "User not found" });
+
+      await backfillAudiobookHistory(user);
+      const updatedUser = await prisma.user.findUnique({
+        where: { id: user.id },
+        select: {
+          id: true,
+          plexUsername: true,
+          totalEarned: true,
+          watchTimeMinutes: true,
+        },
+      });
+      return res.json({ success: true, user: updatedUser });
+    }
+
+    // Backfill all active users
+    const users = await prisma.user.findMany({
+      where: { isActive: true },
+    });
+
+    const results: any[] = [];
+    for (const user of users) {
+      await backfillAudiobookHistory(user);
+      results.push({ userId: user.id, plexUsername: user.plexUsername });
+    }
+
+    res.json({ success: true, usersProcessed: results.length, results });
+  })
+);
+
+// Sync only new sessions since last sync
+router.post(
+  "/sync-pending",
+  authenticate,
+  requireAdmin,
+  asyncHandler(async (req: AuthenticatedRequest, res) => {
+    const { userId } = req.body;
+    const settings = await prisma.systemSettings.findFirst();
+    const audiobookMinListenMinutes =
+      settings?.audiobookMinListenMinutes || 5;
+    const audiobookCreditsPerMinute =
+      settings?.audiobookCreditsPerMinute || 1;
+
+    // Find the most recent audiobook WatchEvent to determine last sync time
+    const lastEvent = await prisma.watchEvent.findFirst({
+      where: {
+        contentType: "audiobook",
+        ...(userId ? { userId } : {}),
+      },
+      orderBy: { watchedAt: "desc" },
+      select: { watchedAt: true },
+    });
+
+    let page = 0;
+    let totalSynced = 0;
+    let totalCredits = 0;
+    let totalMinutes = 0;
+    let hasMore = true;
+
+    const usersToSync = userId
+      ? [await prisma.user.findUnique({ where: { id: userId } })]
+      : await prisma.user.findMany({ where: { isActive: true } });
+
+    for (const user of usersToSync) {
+      if (!user) continue;
+
+      // Map ABS users to this coop user (same matching logic as backfillAudiobookHistory)
+      const absUsers = await getUsers();
+      const userMap = new Map(absUsers.map((u: any) => [u.id, u.username]));
+      const plexEmail = user.email?.toLowerCase().trim() || null;
+      const matchingAbsUserIds = absUsers
+        .filter((u: any) => {
+          const absEmail = u.email?.toLowerCase().trim() || null;
+          const absUsername = u.username.toLowerCase().trim();
+          const plexUsername = user.plexUsername.toLowerCase().trim();
+          if (plexEmail && absEmail === plexEmail) return true;
+          if (plexUsername === "bummer7" && absUsername === "heather") return true;
+          return absUsername === plexUsername;
+        })
+        .map((u: any) => u.id);
+
+      if (matchingAbsUserIds.length === 0) {
+        console.log(`No ABS user match for ${user.plexUsername}, skipping`);
+        continue;
+      }
+
+      page = 0;
+      hasMore = true;
+
+      while (hasMore && page < 10) {
+        const { sessions, numPages } = await getListeningSessions(50, page);
+
+        for (const session of sessions) {
+          // Skip if this session isn't for this user
+          if (!matchingAbsUserIds.includes(session.userId)) {
+            continue;
+          }
+
+          // Skip if this session is older than our last synced event
+          if (lastEvent && new Date(session.date) <= lastEvent.watchedAt) {
+            continue;
+          }
+
+          const sessionId = `abs_${session.id}`;
+
+          if (await prisma.watchEvent.findUnique({ where: { sessionId } })) {
+            continue;
+          }
+
+          const listenMinutes = Math.floor(
+            (session.timeListening || 0) / 60
+          );
+
+          if (listenMinutes < audiobookMinListenMinutes) continue;
+          if (listenMinutes > 1440) continue;
+
+          const creditsEarned = Math.floor(
+            listenMinutes * audiobookCreditsPerMinute
+          );
+
+          const title =
+            session.displayTitle ||
+            session.mediaMetadata?.title ||
+            "Unknown Audiobook";
+          const author =
+            session.displayAuthor ||
+            session.mediaMetadata?.author ||
+            null;
+
+          const watchEvent = await prisma.watchEvent.create({
+            data: {
+              userId: user.id,
+              sessionId,
+              ratingKey: session.id,
+              contentType: "audiobook",
+              title,
+              grandparentTitle: author,
+              duration: session.timeListening || 0,
+              percentComplete: Math.min(
+                100,
+                Math.floor(
+                  ((session.timeListening || 0) /
+                    Math.max(session.duration, 1)) *
+                    100
+                )
+              ),
+              creditsEarned,
+              isProcessed: true,
+              watchedAt: new Date(session.date),
+            },
+          });
+
+          await prisma.transaction.create({
+            data: {
+              userId: user.id,
+              type: "EARN",
+              amount: creditsEarned,
+              watchEventId: watchEvent.id,
+              description: `Listened to ${title}`,
+              contentTitle: title,
+            },
+          });
+
+          totalCredits += creditsEarned;
+          totalMinutes += listenMinutes;
+          totalSynced++;
+        }
+
+        page++;
+        if (page >= numPages || !hasMore) hasMore = false;
+      }
+
+      if (totalCredits > 0) {
+        await prisma.user.update({
+          where: { id: user.id },
+          data: {
+            totalEarned: { increment: totalCredits },
+            watchTimeMinutes: { increment: totalMinutes },
+          },
+        });
+      }
+    }
+
+    res.json({
+      success: true,
+      synced: totalSynced,
+      creditsAwarded: totalCredits,
+      minutes: totalMinutes,
+    });
+  })
+);
+
+export { router as absRouter };
@@ -0,0 +1,155 @@
+import { Router } from "express";
+import {
+  type AuthenticatedRequest,
+  authenticate,
+} from "../middleware/auth";
+import { asyncHandler } from "../middleware/errorHandler";
+import {
+  searchAudiobooks,
+  addAudiobook,
+  getListenarrStatus,
+} from "../services/listenarr";
+import { prisma } from "../utils/prisma";
+
+const router = Router();
+
+router.get(
+  "/status",
+  authenticate,
+  asyncHandler(async (_req: AuthenticatedRequest, res) => {
+    const status = await getListenarrStatus();
+    res.json(status);
+  })
+);
+
+router.get(
+  "/search",
+  authenticate,
+  asyncHandler(async (req: AuthenticatedRequest, res) => {
+    const { q } = req.query;
+    if (!q || typeof q !== "string") {
+      return res.status(400).json({ error: "Query parameter q is required" });
+    }
+    const results = await searchAudiobooks(q);
+    res.json({ results });
+  })
+);
+
+router.post(
+  "/request",
+  authenticate,
+  asyncHandler(async (req: AuthenticatedRequest, res) => {
+    const userId = (req as any).user?.id;
+    if (!userId) return res.status(401).json({ error: "Not authenticated" });
+
+    const {
+      asin, title, subtitle, authors, narrators, imageUrl, description,
+      publisher, language, releaseDate, runtimeLengthMin, genres,
+      series: rawSeries, seriesNumber: rawSeriesNumber, isbn, explicit, abridged,
+    } = req.body;
+
+    // Normalize series/seriesNumber: Listenarr expects strings, not arrays
+    const series = Array.isArray(rawSeries)
+      ? (rawSeries[0]?.name || "")
+      : typeof rawSeries === "string" ? rawSeries : "";
+    const seriesNumber = Array.isArray(rawSeries)
+      ? (rawSeries[0]?.position || rawSeries[0]?.number || "")
+      : typeof rawSeriesNumber === "string" ? rawSeriesNumber : "";
+
+    if (!asin || !title) {
+      return res.status(400).json({ error: "asin and title are required" });
+    }
+
+    const user = await prisma.user.findUnique({ where: { id: userId } });
+    if (!user) return res.status(404).json({ error: "User not found" });
+
+    const settings = await prisma.systemSettings.findFirst();
+    const requestCost = settings?.audiobookRequestCost || 300;
+
+    if ((user.totalEarned - user.totalSpent) < requestCost) {
+      return res.status(400).json({
+        error: "Not enough credits",
+        balance: user.totalEarned - user.totalSpent,
+        cost: requestCost,
+      });
+    }
+
+    const authorNames = (authors || []).map((a: any) =>
+      typeof a === "string" ? a : a.name || a
+    );
+    const narratorNames = (narrators || []).map((n: any) =>
+      typeof n === "string" ? n : n.name || n
+    );
+    const genreNames = (genres || []).map((g: any) =>
+      typeof g === "string" ? g : g.name || g
+    );
+
+    let listenarrResult: any;
+    try {
+      listenarrResult = await addAudiobook({
+        asin,
+        title,
+        subtitle,
+        authors: authorNames,
+        narrators: narratorNames,
+        imageUrl,
+        description,
+        publisher,
+        language,
+        releaseDate,
+        runtimeLengthMin,
+        genres: genreNames,
+        series,
+        seriesNumber,
+        isbn,
+        explicit,
+        abridged,
+      });
+    } catch (err: any) {
+      const detail = err?.response?.data?.errors
+        ? Object.values(err.response.data.errors).flat().join("; ")
+        : err?.response?.data?.title || err?.message || "Listenarr rejected the request";
+      return res.status(400).json({ error: detail });
+    }
+
+    const listenarrBookId = listenarrResult?.audiobook?.id || 0;
+
+    const request = await prisma.contentRequest.create({
+      data: {
+        userId,
+        overseerRequestId: listenarrBookId > 0 ? listenarrBookId : -(Date.now() % 1000000),
+        mediaType: "audiobook",
+        tmdbId: 0,
+        title,
+        creditsCost: requestCost,
+        status: "APPROVED",
+        requestedAt: new Date(),
+      },
+    });
+
+    const transaction = await prisma.transaction.create({
+      data: {
+        userId,
+        type: "SPEND",
+        amount: requestCost,
+        requestId: request.id,
+        description: `Audiobook request: ${title}`,
+        contentTitle: title,
+      },
+    });
+
+    await prisma.user.update({
+      where: { id: userId },
+      data: { totalSpent: { increment: requestCost } },
+    });
+
+    res.json({
+      success: true,
+      transactionId: transaction.id,
+      requestId: request.id,
+      cost: requestCost,
+    });
+  })
+);
+
+export { router as listenarrRouter };
@@ -0,0 +1,255 @@
+import axios from "axios";
+import { prisma } from "../utils/prisma";
+
+const ABS_URL = process.env.ABS_URL || "";
+const ABS_API_TOKEN = process.env.ABS_API_TOKEN || "";
+
+const absApi = axios.create({
+  baseURL: ABS_URL,
+  timeout: 10000,
+  headers: { Authorization: `Bearer ${ABS_API_TOKEN}` },
+});
+
+interface AbsSession {
+  id: string;
+  userId: string;
+  date: string;
+  timeListening: number;
+  duration: number;
+  displayTitle: string;
+  displayAuthor: string;
+  mediaType: string;
+  startedAt: number;
+  updatedAt: number;
+  mediaMetadata?: { title?: string; author?: string };
+}
+
+interface AbsSessionsResponse {
+  total: number;
+  numPages: number;
+  page: number;
+  itemsPerPage: number;
+  sessions: AbsSession[];
+}
+
+// Map ABS user IDs to usernames (fetched once, cached in memory)
+let absUserCache: Record<string, string> | null = null;
+
+async function getAbsUsername(userId: string): Promise<string | null> {
+  if (absUserCache === null) {
+    absUserCache = {};
+    try {
+      const resp = await absApi.get("/api/users");
+      const users = resp.data?.users || [];
+      for (const u of users) {
+        absUserCache[u.id] = u.username;
+      }
+    } catch {
+      // fallback: just use the raw userId
+    }
+  }
+  return absUserCache?.[userId] || null;
+}
+
+export async function getAbsStatus(): Promise<{
+  connected: boolean;
+  data?: any;
+  error?: string;
+}> {
+  try {
+    const resp = await absApi.get("/api/me", { timeout: 5000 });
+    return {
+      connected: true,
+      data: {
+        username: resp.data.username,
+        serverVersion: resp.data.serverVersion,
+      },
+    };
+  } catch (error: any) {
+    return {
+      connected: false,
+      error: error?.message || "Failed to connect to Audiobookshelf",
+    };
+  }
+}
+
+export async function getListeningSessions(
+  limit = 50,
+  page = 0
+): Promise<AbsSessionsResponse> {
+  const resp = await absApi.get("/api/sessions", {
+    params: { itemsPerPage: limit, page },
+  });
+  return resp.data;
+}
+
+export async function getUsers(): Promise<
+  { id: string; username: string; email: string | null }[]
+> {
+  const resp = await absApi.get("/api/users");
+  return resp.data?.users || [];
+}
+
+export async function backfillAudiobookHistory(user: {
+  id: string;
+  plexId: string;
+  plexUsername: string;
+  email?: string | null;
+}) {
+  if (!ABS_URL || !ABS_API_TOKEN) return;
+
+  try {
+    // Get all ABS users to map userIds to usernames
+    const absUsers = await getUsers();
+    const userMap = new Map(absUsers.map((u) => [u.id, u.username]));
+
+    // Find matching ABS user by email first, then username
+    const plexEmail = user.email?.toLowerCase().trim() || null;
+    const matchingAbsUserIds = absUsers
+      .filter((u) => {
+        const absEmail = u.email?.toLowerCase().trim() || null;
+        const absUsername = u.username.toLowerCase().trim();
+        const plexUsername = user.plexUsername.toLowerCase().trim();
+
+        if (plexEmail && absEmail === plexEmail) return true;
+
+        // Special-case: map Heather's ABS account to bummer7 when needed
+        if (
+          plexUsername === "bummer7" &&
+          absUsername === "heather"
+        ) {
+          return true;
+        }
+
+        return absUsername === plexUsername;
+      })
+      .map((u) => u.id);
+
+    if (matchingAbsUserIds.length === 0) {
+      console.log(
+        `ABS user not found for ${user.plexUsername} (${user.email || "no email"}). Available:`,
+        absUsers.map((u) => `${u.username}${u.email ? ` <${u.email}>` : ""}`)
+      );
+      return;
+    }
+
+    const settings = await prisma.systemSettings.findFirst();
+    const audiobookCreditsPerMinute =
+      settings?.audiobookCreditsPerMinute || 1;
+    const audiobookMinListenMinutes =
+      settings?.audiobookMinListenMinutes || 5;
+
+    // Fetch sessions across multiple pages
+    let page = 0;
+    let allSessions: AbsSession[] = [];
+    let hasMore = true;
+
+    while (hasMore && page < 20) {
+      const data = await getListeningSessions(100, page);
+      allSessions = allSessions.concat(data.sessions);
+      page++;
+      if (page >= data.numPages) hasMore = false;
+    }
+
+    // Filter to only this user's sessions
+    const userSessions = allSessions.filter((s) =>
+      matchingAbsUserIds.includes(s.userId)
+    );
+    const afterTimestamp = Date.now() - 30 * 24 * 60 * 60 * 1000;
+
+    console.log(
+      `ABS backfill ${user.plexUsername}: ${allSessions.length} total sessions, ${userSessions.length} for user`
+    );
+
+    if (userSessions.length === 0) return;
+
+    let totalCredits = 0;
+    let totalMinutes = 0;
+
+    for (const session of userSessions) {
+      const sessionId = `abs_${session.id}`;
+      if (await prisma.watchEvent.findUnique({ where: { sessionId } })) {
+        continue;
+      }
+
+      // Only process sessions from last 30 days
+      if (session.updatedAt && session.updatedAt < afterTimestamp) {
+        continue;
+      }
+
+      const listenMinutes = Math.floor(
+        (session.timeListening || 0) / 60
+      );
+
+      if (listenMinutes < audiobookMinListenMinutes) continue;
+      if (listenMinutes > 1440) continue; // cap at 24h
+
+      const creditsEarned = Math.floor(
+        listenMinutes * audiobookCreditsPerMinute
+      );
+
+      console.log(
+        `ABS Credit: ${session.displayTitle} - ${listenMinutes}min, ${creditsEarned}$COOP`
+      );
+
+      const watchEvent = await prisma.watchEvent.create({
+        data: {
+          userId: user.id,
+          sessionId,
+          ratingKey: session.id,
+          contentType: "audiobook",
+          title:
+            session.displayTitle ||
+            session.displayTitle ||
+            "Unknown Audiobook",
+          grandparentTitle: session.displayAuthor || null,
+          duration: session.timeListening || 0,
+          percentComplete: Math.min(
+            100,
+            Math.floor(
+              ((session.timeListening || 0) /
+                Math.max(session.duration, 1)) *
+                100
+            )
+          ),
+          creditsEarned,
+          isProcessed: true,
+          watchedAt: new Date(session.date),
+        },
+      });
+
+      await prisma.transaction.create({
+        data: {
+          userId: user.id,
+          type: "EARN",
+          amount: creditsEarned,
+          watchEventId: watchEvent.id,
+          description: `Listened to ${session.displayTitle}`,
+          contentTitle: session.displayTitle,
+        },
+      });
+
+      totalCredits += creditsEarned;
+      totalMinutes += listenMinutes;
+    }
+
+    console.log(
+      `ABS backfill done ${user.plexUsername}: ${totalCredits} credits, ${totalMinutes} min`
+    );
+
+    if (totalCredits > 0) {
+      await prisma.user.update({
+        where: { id: user.id },
+        data: {
+          totalEarned: { increment: totalCredits },
+          watchTimeMinutes: { increment: totalMinutes },
+        },
+      });
+    }
+  } catch (error) {
+    console.error(
+      `Failed to backfill ABS history for ${user.plexUsername}:`,
+      error
+    );
+  }
+}