ReadaBook/apps/api/src/reading/reading.service.ts

import { createHash } from "node:crypto";
import { createReadStream } from "node:fs";
import { BadRequestException, HttpException, HttpStatus, Inject, Injectable, NotFoundException } from "@nestjs/common";
import { and, eq } from "drizzle-orm";
import { DatabaseService } from "../database/database.service.js";
import { books, libraries } from "../database/schema.js";
import { JobsService } from "../jobs/jobs.service.js";
import { ReadingCachePort, ReadingSourcePort } from "./reading.ports.js";
import { ReadingBook, ReadingResourcePayload, ReadingWindowInput } from "./reading.types.js";

@Injectable()
export class ReadingService {
  private readonly preparing = new Set<string>();

  constructor(
    private readonly database: DatabaseService,
    private readonly jobs: JobsService,
    @Inject(ReadingSourcePort) private readonly source: ReadingSourcePort,
    @Inject(ReadingCachePort) private readonly cache: ReadingCachePort
  ) {}

  async manifest(bookId: number) {
    return this.source.manifest(this.accessibleBook(bookId));
  }

  async resource(bookId: number, resourceId: string, range?: string): Promise<ReadingResourcePayload & { stream?: NodeJS.ReadableStream; start: number; end: number; partial: boolean }> {
    const book = this.accessibleBook(bookId);
    const cached = book.format === "epub" ? await this.cache.get(book, resourceId) : null;
    const manifestResource = cached ? (await this.source.manifest(book)).resources.find((resource) => resource.resourceId === resourceId) : null;
    if (cached && !manifestResource) throw new NotFoundException("Reading resource not found");
    const payload = cached && manifestResource
      ? { data: cached, contentType: manifestResource.mediaType, size: cached.length, etag: manifestVersionEtag(book, resourceId), supportsRanges: false }
      : await this.source.resource(book, resourceId);
    const byteRange = parseByteRange(range, payload.size, payload.supportsRanges);
    if (payload.filePath) {
      return { ...payload, stream: createReadStream(payload.filePath, { start: byteRange.start, end: byteRange.end }), ...byteRange };
    }
    const data = payload.data;
    if (!data) throw new NotFoundException("Reading resource not found");
    if (!cached) await this.cache.put(book, resourceId, data);
    return { ...payload, data: data.subarray(byteRange.start, byteRange.end + 1), ...byteRange };
  }

  async window(bookId: number, input: ReadingWindowInput) {
    const book = this.accessibleBook(bookId);
    const manifest = await this.source.manifest(book);
    const position = resolvePosition(input.locator, manifest.spine);
    const current = manifest.spine[position];
    const referenced = referencedResources(current.resourceId, manifest.resources);
    const ordered = priorityWindow(manifest.spine, position, referenced, input.networkHint === "slow" || input.mode === "speed" ? 1 : 2);
    const states = await Promise.all(ordered.map(async ({ id, distance }) => ({
      resourceId: id,
      relation: distance === 0 ? "current" : distance === null ? "resource" : distance === -1 ? "previous" : distance === 1 ? "next" : "neighbor",
      state: await this.resourceState(book, id)
    })));
    if (input.networkHint !== "offline") this.enqueuePreparation(book, ordered.map((entry) => entry.id));
    return {
      bookId,
      locator: input.locator,
      mode: input.mode,
      networkHint: input.networkHint,
      current: states.find((state) => state.relation === "current") ?? null,
      previous: states.find((state) => state.relation === "previous") ?? null,
      next: states.find((state) => state.relation === "next") ?? null,
      resources: states.filter((state) => state.relation === "resource"),
      neighbors: states.filter((state) => state.relation === "neighbor")
    };
  }

  async retry(bookId: number, resourceId: string) {
    const book = this.accessibleBook(bookId);
    const manifest = await this.source.manifest(book);
    if (!manifest.resources.some((resource) => resource.resourceId === resourceId)) throw new NotFoundException("Reading resource not found");
    await this.cache.remove(book, resourceId);
    this.enqueuePreparation(book, [resourceId], true);
    return { bookId, resourceId, state: "pending" as const };
  }

  private accessibleBook(bookId: number): ReadingBook {
    if (!Number.isInteger(bookId) || bookId < 1) throw new BadRequestException("Invalid book id");
    const row = this.database.db
      .select({ id: books.id, format: books.format, filePath: books.filePath, fileSize: books.fileSize, fileMtime: books.fileMtime })
      .from(books)
      .innerJoin(libraries, and(eq(libraries.id, books.libraryId), eq(libraries.enabled, true)))
      .where(eq(books.id, bookId))
      .get();
    if (!row) throw new NotFoundException("Book not found in an accessible library");
    return row;
  }

  private async resourceState(book: ReadingBook, resourceId: string) {
    if (book.format === "pdf" && resourceId === "original") return "ready" as const;
    return (await this.cache.get(book, resourceId)) ? "ready" as const : "pending" as const;
  }

  private enqueuePreparation(book: ReadingBook, resourceIds: string[], force = false): void {
    const pending = resourceIds.filter((id) => force || !this.preparing.has(`${book.id}:${id}`));
    if (!pending.length) return;
    const job = this.jobs.create("prepare-reading-window", JSON.stringify({ bookId: book.id, resourceIds: pending }));
    for (const id of pending) this.preparing.add(`${book.id}:${id}`);
    setImmediate(async () => {
      this.jobs.markRunning(job.id, JSON.stringify({ bookId: book.id, resourceIds: pending }));
      try {
        for (const resourceId of pending) {
          const payload = await this.source.resource(book, resourceId);
          if (payload.data) await this.cache.put(book, resourceId, payload.data);
        }
        this.jobs.markSucceeded(job.id, JSON.stringify({ bookId: book.id, prepared: pending }));
      } catch (error) {
        this.jobs.markFailed(job.id, error);
      } finally {
        for (const id of pending) this.preparing.delete(`${book.id}:${id}`);
      }
    });
  }
}

function manifestVersionEtag(book: ReadingBook, resourceId: string): string {
  return `\"${createHash("sha256").update(`${book.id}:${book.fileSize}:${book.fileMtime}:${resourceId}`).digest("base64url")}\"`;
}

function resolvePosition(locator: string, spine: Array<{ index: number; href: string; resourceId: string }>): number {
  if (!spine.length) throw new BadRequestException("Book has no readable spine");
  const direct = spine.findIndex((item) => item.resourceId === locator || item.href === locator);
  if (direct >= 0) return direct;
  const numeric = Number(locator.match(/(?:^|:)(\d+)(?:$|:)/)?.[1]);
  if (Number.isInteger(numeric) && numeric >= 0 && numeric < spine.length) return numeric;
  throw new BadRequestException("Locator does not identify a readable resource");
}

function priorityWindow(spine: Array<{ resourceId: string }>, position: number, referenced: string[], radius: number) {
  const result: Array<{ id: string; distance: number | null }> = [{ id: spine[position].resourceId, distance: 0 }];
  for (const resourceId of referenced) result.push({ id: resourceId, distance: null });
  for (let distance = 1; distance <= radius; distance += 1) {
    if (position - distance >= 0) result.push({ id: spine[position - distance].resourceId, distance: -distance });
    if (position + distance < spine.length) result.push({ id: spine[position + distance].resourceId, distance });
  }
  return result.filter((entry, index) => result.findIndex((candidate) => candidate.id === entry.id) === index);
}

function referencedResources(currentId: string, resources: Array<{ resourceId: string; references?: string[] }>): string[] {
  const byId = new Map(resources.map((resource) => [resource.resourceId, resource]));
  const result: string[] = [];
  const queue = [...(byId.get(currentId)?.references ?? [])];
  const seen = new Set([currentId]);
  while (queue.length) {
    const resourceId = queue.shift()!;
    if (seen.has(resourceId)) continue;
    seen.add(resourceId);
    result.push(resourceId);
    queue.push(...(byId.get(resourceId)?.references ?? []));
  }
  return result;
}

function parseByteRange(range: string | undefined, size: number, supported: boolean) {
  if (!range) return { start: 0, end: Math.max(size - 1, 0), partial: false };
  if (!supported || size <= 0) throw new HttpException("Requested range not satisfiable", HttpStatus.REQUESTED_RANGE_NOT_SATISFIABLE);
  const match = range.match(/^bytes=(\d*)-(\d*)$/);
  if (!match) throw new HttpException("Requested range not satisfiable", HttpStatus.REQUESTED_RANGE_NOT_SATISFIABLE);
  const suffix = !match[1] && match[2] ? Number(match[2]) : null;
  const start = suffix === null ? Number(match[1]) : Math.max(size - suffix, 0);
  const end = suffix === null && match[2] ? Number(match[2]) : size - 1;
  if (!Number.isInteger(start) || !Number.isInteger(end) || start < 0 || end < start || start >= size) {
    throw new HttpException("Requested range not satisfiable", HttpStatus.REQUESTED_RANGE_NOT_SATISFIABLE);
  }
  return { start, end: Math.min(end, size - 1), partial: true };
}