feat: check for duplicate files when importing books

Incoming files are matched against what is already stored, keyed on the
KOReader hash and the file size. Bulk uploads skip and report them, deliberate
creates are refused with a 409, the consume directory parks them aside, and
allow_duplicates overrides all three.

Also: books whose metadata generates a path another book already owns are moved
aside, so a forced copy cannot overwrite the original's files.

.env.prod-example

@@ -5,6 +5,15 @@ CHITAI_TOKEN_SECRET=secret
 CHITAI_DEFAULT_LIBRARY_NAME=Books
 CHITAI_DEFAULT_LIBRARY_PATH="libraries/books"
 
+# Duplicate detection when importing files (optional).
+# Scope: "library" compares against the library being uploaded to, "global" against
+# every library, "off" disables the check.
+CHITAI_DUPLICATE_SCOPE=library
+
+# Where the consume watcher parks files it refused as duplicates. Keep it outside
+# CHITAI_CONSUME_PATH, or the watcher picks them straight back up.
+CHITAI_DUPLICATE_PATH="duplicates"
+
 # You probably should not change these
 CHITAI_API_URL="http://backend:8000"
 CHITAI_API_DEBUG=false

backend/AGENTS.md

@@ -88,6 +88,14 @@ The backend owns files on disk, not just rows:
 - **Layout** — `services/filesystem_library.py` (`BookPathGenerator`) renders a Jinja2 template
   against book metadata to decide where a book lives under the library's `root_path`
   (default: `author/series/position - title/`).
+- **One directory per book, never shared.** The generated path is a pure function of the metadata,
+  so two books with the same author and title produce the same one — two editions, or an
+  `allow_duplicates` copy. `BookService._reserve_book_path` moves the later one to `title (2)`
+  before anything is written, and `update_book` reserves the same way so a rename cannot move a
+  book in on top of another. This matters because `book.path` is what deletes, moves and file
+  lookups act on: books sharing a directory means one overwrites the other's files, and deleting
+  either takes both. `_unused_path` does the same job for filenames within a directory.
+  A book that already has a `path` keeps it — `add_files` must follow the book, not the template.
 - **Metadata extraction** — `services/metadata_extractor.py` reads EPUB (ebooklib) and PDF
   (pypdfium2) files; extracted values fill only *empty* fields on the incoming payload.
 - **Covers** — converted to WebP with a UUID filename under `settings.book_cover_path`, served by a
@@ -100,6 +108,41 @@ The backend owns files on disk, not just rows:
   if it differs, moves the directory contents and prunes empty parents. Keep that in mind before
   changing metadata handling.
 
+## Duplicate detection
+
+Every ingest path screens incoming files against what is already stored, keyed on
+**`(hash, size)`** — never the hash alone, because it samples 12 KiB (see below) and
+EPUBs from one toolchain often share their first window. `FileMetadata.hash` carries a
+plain, deliberately **non-unique** index: a collision must not be able to fail an import,
+and older databases may already hold duplicates.
+
+Scope comes from `CHITAI_DUPLICATE_SCOPE` (`library`, the default | `global` | `off`).
+
+The policy differs by how deliberate the import is:
+
+| Path | Behaviour |
+| --- | --- |
+| `create_many_from_files` (browser bulk) | Skip per file, skip a whole group whose files are all known, report everything skipped in `ImportResult.duplicates`. Re-dropping a folder to pick up what is new is the case this serves. |
+| `create_book` (single, with metadata) | All-or-nothing: raises `DuplicateFilesError`, which `controllers/book.py` renders as a **409** carrying the refused files in `extra`. |
+| `add_files` | A file the book already carries is a no-op; one stored under another book raises `DuplicateFilesError`. |
+| `create_many_from_existing_files` (consume watcher) | Skips, and **moves the file to `CHITAI_DUPLICATE_PATH/<library slug>/`** — nothing is deleted, and it cannot stay put because `watchfiles` only reports additions. That path must stay outside `consume_path` or the watcher re-imports it and tries to read the directory name as a library slug. |
+
+`allow_duplicates=true` overrides all of it, on every endpoint. Keep that working — the
+hash is not proof of identity, so a false positive has to be recoverable.
+
+Two things to preserve when touching this code:
+
+- **Screening runs before anything is written.** `fingerprint_upload` reads the spooled
+  upload and rewinds it; the resulting fingerprints are handed to `_save_book_files`,
+  which skips its own `StreamingHasher` when it already has the answer. Passing them
+  through is what keeps the file from being read twice.
+- **`_screen_for_duplicates` extends the `known` dict as it goes**, so the same bytes
+  submitted twice in one request are caught. Those duplicates report `book_id: None` —
+  there is no row to point at yet.
+
+`POST /books/duplicates` answers the same question from fingerprints alone, for clients
+that want to ask before uploading anything.
+
 ## KOReader hashing
 
 `services/utils.py` reimplements KOReader's partial-MD5 document identifier: 1 KiB samples at

backend/migrations/versions/2026-08-13_add_file_hash_index_e9c2c7e875ae.py

@@ -0,0 +1,80 @@
+"""add file hash index
+
+Revision ID: e9c2c7e875ae
+Revises: 6d72d1bbc0ee
+Create Date: 2026-08-13 14:52:09.341906
+
+"""
+
+import warnings
+from typing import TYPE_CHECKING
+
+import sqlalchemy as sa
+from alembic import op
+from advanced_alchemy.types import EncryptedString, EncryptedText, GUID, ORA_JSONB, DateTimeUTC, StoredObject, PasswordHash, FernetBackend
+from advanced_alchemy.types.encrypted_string import PGCryptoBackend
+from advanced_alchemy.types.password_hash.argon2 import Argon2Hasher
+from advanced_alchemy.types.password_hash.passlib import PasslibHasher
+from advanced_alchemy.types.password_hash.pwdlib import PwdlibHasher
+from sqlalchemy import Text  # noqa: F401
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+__all__ = ["downgrade", "upgrade", "schema_upgrades", "schema_downgrades", "data_upgrades", "data_downgrades"]
+
+sa.GUID = GUID
+sa.DateTimeUTC = DateTimeUTC
+sa.ORA_JSONB = ORA_JSONB
+sa.EncryptedString = EncryptedString
+sa.EncryptedText = EncryptedText
+sa.StoredObject = StoredObject
+sa.PasswordHash = PasswordHash
+sa.Argon2Hasher = Argon2Hasher
+sa.PasslibHasher = PasslibHasher
+sa.PwdlibHasher = PwdlibHasher
+sa.FernetBackend = FernetBackend
+sa.PGCryptoBackend = PGCryptoBackend
+
+# revision identifiers, used by Alembic.
+revision = 'e9c2c7e875ae'
+down_revision = '6d72d1bbc0ee'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with warnings.catch_warnings():
+        warnings.filterwarnings("ignore", category=UserWarning)
+        with op.get_context().autocommit_block():
+            schema_upgrades()
+            data_upgrades()
+
+def downgrade() -> None:
+    with warnings.catch_warnings():
+        warnings.filterwarnings("ignore", category=UserWarning)
+        with op.get_context().autocommit_block():
+            data_downgrades()
+            schema_downgrades()
+
+def schema_upgrades() -> None:
+    """schema upgrade migrations go here."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('file_metadata', schema=None) as batch_op:
+        batch_op.create_index('ix_file_metadata_hash', ['hash'], unique=False)
+
+    # ### end Alembic commands ###
+
+def schema_downgrades() -> None:
+    """schema downgrade migrations go here."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('file_metadata', schema=None) as batch_op:
+        batch_op.drop_index('ix_file_metadata_hash')
+
+    # ### end Alembic commands ###
+
+def data_upgrades() -> None:
+    """Add any optional data upgrade migrations here!"""
+
+def data_downgrades() -> None:
+    """Add any optional data downgrade migrations here!"""

backend/src/chitai/config.py

@@ -1,9 +1,25 @@
+from enum import StrEnum
+
 from pydantic import Field, PostgresDsn, computed_field
 from pydantic_settings import BaseSettings, SettingsConfigDict
 from advanced_alchemy.extensions.litestar import (
     SQLAlchemyAsyncConfig,
 )
 
+
+class DuplicateScope(StrEnum):
+    """How widely an incoming file is compared against what is already stored."""
+
+    LIBRARY = "library"
+    """Only files in the library being uploaded to count as duplicates."""
+
+    GLOBAL = "global"
+    """A file already held by any library counts as a duplicate."""
+
+    OFF = "off"
+    """No duplicate detection at all."""
+
+
 class Settings(BaseSettings):
     version: str = Field("0.0.1")
     project_name: str = Field("chitai")
@@ -33,6 +49,14 @@ class Settings(BaseSettings):
     # Path to consume directory
     consume_path: str = Field("./consume")
 
+    # Duplicate detection
+    duplicate_scope: DuplicateScope = Field(DuplicateScope.LIBRARY)
+
+    # Where the consume watcher parks files it refused as duplicates. Must sit
+    # outside `consume_path`, or the watcher picks them straight back up and
+    # tries to resolve the directory name as a library slug.
+    duplicate_path: str = Field("./duplicates")
+
     @computed_field
     @property
     def postgres_uri(self) -> PostgresDsn:

backend/src/chitai/controllers/book.py

@@ -12,7 +12,11 @@ from litestar.params import Dependency, Body
 from litestar.enums import RequestEncodingType
 from litestar.response import File, Stream
 from litestar.exceptions import HTTPException
-from litestar.status_codes import HTTP_400_BAD_REQUEST
+from litestar.status_codes import (
+    HTTP_200_OK,
+    HTTP_400_BAD_REQUEST,
+    HTTP_409_CONFLICT,
+)
 from litestar.datastructures import UploadFile
 from advanced_alchemy.service.pagination import OffsetPagination
 from advanced_alchemy.filters import CollectionFilter
@@ -23,6 +27,24 @@ from chitai.services import dependencies as deps
 from chitai import schemas as s
 from chitai.database import models as m
 from chitai.services import BookService, BookProgressService
+from chitai.services.book import DuplicateFilesError
+
+
+def _duplicate_conflict(exc: DuplicateFilesError) -> HTTPException:
+    """
+    Turn refused files into a 409 the caller can act on.
+
+    The files ride along in `extra` so the client can name them and offer to send them
+    again with `allow_duplicates`, rather than being told only that something clashed.
+    """
+    return HTTPException(
+        status_code=HTTP_409_CONFLICT,
+        detail="These files are already in the library",
+        extra=[
+            s.DuplicateFileRead.model_validate(duplicate).model_dump(mode="json")
+            for duplicate in exc.duplicates
+        ],
+    )
 
 
 class BookController(Controller):
@@ -63,6 +85,7 @@ class BookController(Controller):
         books_service: BookService,
         library: m.Library,
         data: Annotated[s.BookCreate, Body(media_type=RequestEncodingType.MULTI_PART)],
+        allow_duplicates: bool = False,
     ) -> s.BookRead:
         """
         Create a new book with metadata and files.
@@ -73,6 +96,10 @@ class BookController(Controller):
         Path Parameters:
             library_id: The ID of the library the book belongs to.
 
+        Query Parameters:
+            allow_duplicates: If True, store the files even if the library already
+                holds them.
+
         Request Body:
             data: Book creation data including metadata and files.
 
@@ -83,9 +110,17 @@ class BookController(Controller):
         Returns:
             The created book as a BookRead schema.
 
+        Raises:
+            HTTPException: 409 if any of the files is already in the library.
         """
 
-        result = await books_service.create_book(data, library)
+        try:
+            result = await books_service.create_book(
+                data, library, screen_duplicates=not allow_duplicates
+            )
+        except DuplicateFilesError as exc:
+            raise _duplicate_conflict(exc)
+
         book = await books_service.get(result.id)
         return books_service.to_schema(book, schema_type=s.BookRead)
 
@@ -97,13 +132,20 @@ class BookController(Controller):
         data: Annotated[
             s.BooksCreateFromFiles, Body(media_type=RequestEncodingType.MULTI_PART)
         ],
-    ) -> OffsetPagination[s.BookRead]:
+        allow_duplicates: bool = False,
+    ) -> s.BooksUploadResult:
         """
         Create multiple books from uploaded files.
 
         Groups files by directory and creates separate books for each group.
         Metadata is automatically extracted from the files.
 
+        Files the library already holds are skipped rather than refused, and reported
+        back so the caller can say which ones did not make it in and why.
+
+        Query Parameters:
+            allow_duplicates: If True, store every file, even one already held.
+
         Request Body:
             data: Container with list of uploaded files.
 
@@ -112,19 +154,79 @@ class BookController(Controller):
             library: The library the books belong to.
 
         Returns:
-            Paginated list of created books.
+            The books created, and the files skipped as duplicates.
         """
         try:
-            results = await books_service.create_many_from_files(data, library)
+            result = await books_service.create_many_from_files(
+                data, library, allow_duplicates=allow_duplicates
+            )
         except ValueError:
             raise HTTPException(
                 status_code=HTTP_400_BAD_REQUEST, detail="Must upload at least one file"
             )
 
-        books = await books_service.list(
+        books = (
-            CollectionFilter("id", [result.id for result in results])
+            await books_service.list(
+                CollectionFilter("id", [book.id for book in result.books])
             )
-        return books_service.to_schema(books, schema_type=s.BookRead)
+            if result.books
+            else []
+        )
+
+        return s.BooksUploadResult(
+            created=[
+                books_service.to_schema(book, schema_type=s.BookRead) for book in books
+            ],
+            skipped=[
+                s.DuplicateFileRead.model_validate(duplicate)
+                for duplicate in result.duplicates
+            ],
+        )
+
+    # A question, not a change: 200 rather than the 201 a POST would default to.
+    @post(path="duplicates", status_code=HTTP_200_OK)
+    async def check_duplicates(
+        self,
+        books_service: BookService,
+        library: m.Library,
+        data: list[s.FileFingerprint],
+    ) -> list[s.DuplicateFileRead]:
+        """
+        Report which of the given files the library already holds.
+
+        Lets a client ask before it uploads anything, which is the difference between
+        re-sending a folder of books and re-sending twelve kilobytes of hashes.
+
+        Query Parameters:
+            library_id: The library to check against.
+
+        Request Body:
+            data: Hash and size for each file, optionally with the name to echo back.
+
+        Injected Dependencies:
+            books_service: The book service for database operations.
+            library: The library to check against.
+
+        Returns:
+            One entry per submitted file that is already stored. Files that are not
+            are absent.
+        """
+        matches = await books_service.find_duplicate_files(
+            ((item.hash, item.size) for item in data), library
+        )
+
+        return [
+            s.DuplicateFileRead(
+                filename=item.filename or match.filename,
+                hash=match.hash,
+                size=match.size,
+                library_id=match.library_id,
+                book_id=match.book_id,
+                book_title=match.book_title,
+            )
+            for item in data
+            if (match := matches.get((item.hash, item.size))) is not None
+        ]
 
     @get(path="/{book_id:int}")
     async def get_book_by_id(
@@ -303,13 +405,20 @@ class BookController(Controller):
         ],
         library: m.Library,
         books_service: BookService,
+        allow_duplicates: bool = False,
     ) -> s.BookRead:
         """
         Add files to an existing book.
 
+        A file the book already carries is ignored, so re-sending one is harmless.
+
         Path Parameters:
             book_id: The ID of the book to modify
 
+        Query Parameters:
+            allow_duplicates: If True, store the files even if the library already
+                holds them.
+
         Request Body:
             files: The files to add to the book
 
@@ -320,9 +429,17 @@ class BookController(Controller):
         Returns:
             The modified book
 
+        Raises:
+            HTTPException: 409 if a file is already stored under a different book.
         """
 
-        await books_service.add_files(book_id, data, library)
+        try:
+            await books_service.add_files(
+                book_id, data, library, allow_duplicates=allow_duplicates
+            )
+        except DuplicateFilesError as exc:
+            raise _duplicate_conflict(exc)
+
         book = await books_service.get(book_id)
         return books_service.to_schema(book, schema_type=s.BookRead)
 

backend/src/chitai/database/models/book.py

@@ -139,6 +139,15 @@ class Identifier(BigIntBase):
 class FileMetadata(BigIntBase):
     __tablename__ = "file_metadata"
 
+    __table_args__ = (
+        # Deliberately not unique. The hash is KOReader's partial MD5, which samples
+        # 12 KiB of the file, so two genuinely different files can collide — and an
+        # existing database may already hold duplicates, which a unique index would
+        # refuse to build over. Duplicate detection pairs it with `size` and treats a
+        # match as advisory, so this only has to make the lookup cheap.
+        Index("ix_file_metadata_hash", "hash"),
+    )
+
     book_id: Mapped[int] = mapped_column(ForeignKey("books.id", ondelete="cascade"))
     book: Mapped[Book] = relationship(back_populates="files")
     hash: Mapped[str]

backend/src/chitai/schemas/__init__.py

@@ -4,7 +4,10 @@ from .book import (
     BookProgressCreate,
     BookProgressRead,
     BooksCreateFromFiles,
+    BooksUploadResult,
     BookMetadataUpdate,
+    DuplicateFileRead,
+    FileFingerprint,
     FileMetadataRead,
     BookSeriesRead,
 )

backend/src/chitai/schemas/book.py

@@ -129,6 +129,36 @@ class BooksCreateFromFiles(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
 
+class FileFingerprint(BaseModel):
+    """What a client can say about a file it has not uploaded yet."""
+
+    hash: str
+    size: int
+    filename: str = ""
+
+
+class DuplicateFileRead(BaseModel):
+    """A file that was not stored because the library already holds its bytes."""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    filename: str
+    hash: str
+    size: int
+    library_id: int
+
+    # Null when the match was another file in the same upload, which has no row yet.
+    book_id: int | None = None
+    book_title: str | None = None
+
+
+class BooksUploadResult(BaseModel):
+    """The outcome of a multi-file upload: what was created, and what was skipped."""
+
+    created: list["BookRead"]
+    skipped: list[DuplicateFileRead]
+
+
 class BookMetadataUpdate(BaseModel):
     title: str | None = None
     subtitle: str | None = None

backend/src/chitai/services/book.py

@@ -5,12 +5,13 @@ from __future__ import annotations
 from collections import defaultdict
 import mimetypes
 
-from collections.abc import Callable
+from collections.abc import Callable, Iterable, Sequence
+from dataclasses import dataclass, field, replace
 from io import BytesIO, RawIOBase
 from pathlib import Path
 import uuid
 import zipfile
-from typing import TYPE_CHECKING, Any, AsyncIterator
+from typing import TYPE_CHECKING, Any, AsyncIterator, TypeVar
 
 # Third-party libraries
 from advanced_alchemy.extensions.litestar import service
@@ -22,7 +23,7 @@ from advanced_alchemy.service import (
 )
 from advanced_alchemy.repository import SQLAlchemyAsyncRepository
 from advanced_alchemy.filters import CollectionFilter
-from sqlalchemy import inspect
+from sqlalchemy import inspect, select, tuple_
 from litestar.response import File
 from litestar.datastructures import UploadFile
 import aiofiles
@@ -30,7 +31,7 @@ from aiofiles import os as aios
 from PIL import Image
 
 # Local imports
-from chitai.config import settings
+from chitai.config import DuplicateScope, settings
 from chitai.database.models import (
     Book,
     Author,
@@ -47,9 +48,10 @@ from chitai.schemas.book import BooksCreateFromFiles
 from chitai.services.filesystem_library import BookPathGenerator
 from chitai.services.metadata_extractor import Extractor as MetadataExtractor
 from chitai.services.utils import (
-    calculate_koreader_hash,
     cleanup_empty_parent_directories,
     delete_file,
+    fingerprint_file,
+    fingerprint_upload,
     move_dir_contents,
     move_file,
     save_image,
@@ -57,6 +59,81 @@ from chitai.services.utils import (
 )
 
 
+# An incoming file, either uploaded or already sitting in the consume directory.
+FileT = TypeVar("FileT", UploadFile, Path)
+
+# A file's identity for duplicate detection: its hash and its byte size.
+#
+# The hash on its own is not proof. It is KOReader's partial MD5, which samples
+# twelve 1 KiB windows, so two different files can produce the same one — EPUBs are
+# especially prone at offset 0, where the `mimetype` entry and `META-INF/container.xml`
+# are often byte-identical across everything a given tool produced. Pairing it with the
+# size makes a false match require both.
+Fingerprint = tuple[str, int]
+
+
+@dataclass(frozen=True)
+class DuplicateFile:
+    """
+    An incoming file that was not stored because the library already holds its bytes.
+
+    `book_id` and `book_title` are `None` when the match was another file in the same
+    import rather than something already in the database — there is no row to name yet.
+    """
+
+    filename: str
+    hash: str
+    size: int
+    library_id: int
+    book_id: int | None = None
+    book_title: str | None = None
+
+
+@dataclass
+class ImportResult:
+    """What an import produced: the books created, and the files left out of them."""
+
+    books: list[Book] = field(default_factory=list)
+    duplicates: list[DuplicateFile] = field(default_factory=list)
+
+
+class DuplicateFilesError(Exception):
+    """Raised when an import would store files that are already in the library."""
+
+    def __init__(self, duplicates: list[DuplicateFile]) -> None:
+        self.duplicates = duplicates
+        super().__init__(f"{len(duplicates)} file(s) are already in the library")
+
+
+def _submitted_name(file: UploadFile | Path) -> str:
+    """The name an incoming file is tracked under while it is being screened."""
+    return file.filename if isinstance(file, UploadFile) else file.name
+
+
+def _unused_path(path: Path) -> Path:
+    """
+    The given path, or the first free `name (n).ext` beside it.
+
+    Filenames are generated from metadata, so two books that share an author and a
+    title produce the same one. Opening it for writing would replace the bytes another
+    `FileMetadata` row describes, leaving that row quietly lying about its own file.
+
+    Args:
+        path: Where the file would go.
+
+    Returns:
+        A path that nothing occupies.
+    """
+    candidate = path
+    suffix = 1
+
+    while candidate.exists():
+        suffix += 1
+        candidate = path.with_name(f"{path.stem} ({suffix}){path.suffix}")
+
+    return candidate
+
+
 class _ZipStream(RawIOBase):
     """
     A write-only sink that hands whatever `ZipFile` writes back to the caller.
@@ -106,7 +183,15 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
 
     
 
-    async def create_book(self, data: ModelDictT[Book], library: Library, **kwargs) -> Book:
+    async def create_book(
+        self,
+        data: ModelDictT[Book],
+        library: Library,
+        *,
+        screen_duplicates: bool = True,
+        fingerprints: dict[str, Fingerprint] | None = None,
+        **kwargs,
+    ) -> Book:
         """
         Create a new book entity.
 
@@ -116,37 +201,221 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
         Args:
             data: Book data as a dictionary or model.
             library: The library the book belongs to.
-            *args: Additional positional arguments passed to parent create.
+            screen_duplicates: Refuse the book if any of its files is already stored.
+            fingerprints: Pre-computed `(hash, size)` per submitted filename, so files
+                that have already been screened are not read a second time.
             **kwargs: Additional keyword arguments passed to parent create.
 
         Returns:
             The created Book entity.
+
+        Raises:
+            DuplicateFilesError: If any file is already stored. Creating one book is a
+                deliberate act naming specific files, so this is all-or-nothing:
+                dropping one of them silently would be worse than refusing the lot.
         """
         data = schema_dump(data)
+
+        files = data.get("files") or []
+        if screen_duplicates and files:
+            fingerprints = await self._fingerprint_uploads(files)
+            known = await self.find_duplicate_files(fingerprints.values(), library)
+            _, rejected = self._screen_for_duplicates(
+                files, fingerprints, known, library
+            )
+
+            if rejected:
+                raise DuplicateFilesError([duplicate for _, duplicate in rejected])
+
         await self._parse_metadata_from_files(data)
 
         await self._save_cover_image(data)
-        await self._save_book_files(library, data)
+        await self._save_book_files(library, data, fingerprints)
 
         return await super().create(data, **kwargs)
 
+    async def find_duplicate_files(
+        self, fingerprints: Iterable[Fingerprint], library: Library
+    ) -> dict[Fingerprint, DuplicateFile]:
+        """
+        Look up which of the given fingerprints are already stored.
+
+        Args:
+            fingerprints: The `(hash, size)` pairs to look for.
+            library: The library being imported into. The search is scoped to it
+                unless `duplicate_scope` widens or disables it.
+
+        Returns:
+            A mapping from each matched fingerprint to the book already holding it.
+            Fingerprints with no match are absent.
+        """
+        wanted = set(fingerprints)
+
+        if not wanted or settings.duplicate_scope == DuplicateScope.OFF:
+            return {}
+
+        statement = (
+            select(
+                FileMetadata.hash,
+                FileMetadata.size,
+                FileMetadata.path,
+                Book.id,
+                Book.title,
+                Book.library_id,
+            )
+            .join(Book, FileMetadata.book_id == Book.id)
+            .where(tuple_(FileMetadata.hash, FileMetadata.size).in_(wanted))
+            # Databases predating duplicate detection can hold the same file twice;
+            # order so that the book reported for it is at least a stable one.
+            .order_by(Book.id)
+        )
+
+        if settings.duplicate_scope == DuplicateScope.LIBRARY:
+            statement = statement.where(Book.library_id == library.id)
+
+        rows = await self.repository.session.execute(statement)
+
+        matches: dict[Fingerprint, DuplicateFile] = {}
+        for file_hash, size, path, book_id, title, library_id in rows:
+            matches.setdefault(
+                (file_hash, size),
+                DuplicateFile(
+                    filename=Path(path).name,
+                    hash=file_hash,
+                    size=size,
+                    library_id=library_id,
+                    book_id=book_id,
+                    book_title=title,
+                ),
+            )
+
+        return matches
+
+    async def _reserve_book_path(
+        self, parent: Path, book_id: int | None = None
+    ) -> Path:
+        """
+        Find a directory for a book that no other book is already using.
+
+        The path is generated from metadata alone, so two books that share an author
+        and title land on the same one — an `allow_duplicates` copy, or simply two
+        editions. Letting them share it is not a cosmetic problem: `book.path` is what
+        deletes, moves and file lookups act on, so one book's files overwrite the
+        other's and deleting either takes both.
+
+        Args:
+            parent: The path the template generated.
+            book_id: The book being placed, if it already exists. Its own directory is
+                not a collision with itself.
+
+        Returns:
+            `parent`, or the first free `parent (n)` beside it.
+        """
+        candidate = parent
+        suffix = 1
+
+        while await self._path_belongs_to_another_book(candidate, book_id):
+            suffix += 1
+            candidate = parent.with_name(f"{parent.name} ({suffix})")
+
+        return candidate
+
+    async def _path_belongs_to_another_book(
+        self, path: Path, book_id: int | None
+    ) -> bool:
+        """Whether a book other than `book_id` already claims this directory."""
+        statement = select(Book.id).where(Book.path == str(path)).limit(1)
+
+        if book_id is not None:
+            statement = statement.where(Book.id != book_id)
+
+        return (await self.repository.session.execute(statement)).first() is not None
+
+    async def _fingerprint_uploads(
+        self, files: Sequence[UploadFile]
+    ) -> dict[str, Fingerprint]:
+        """
+        Fingerprint uploaded files, keyed by the name each was submitted under.
+
+        Names carry the browser's relative path, so they are unique within a request
+        the same way the files themselves are — two entries under one name would
+        already be overwriting each other on disk.
+        """
+        return {file.filename: await fingerprint_upload(file) for file in files}
+
+    def _screen_for_duplicates(
+        self,
+        files: Sequence[FileT],
+        fingerprints: dict[str, Fingerprint],
+        known: dict[Fingerprint, DuplicateFile],
+        library: Library,
+    ) -> tuple[list[FileT], list[tuple[FileT, DuplicateFile]]]:
+        """
+        Split incoming files into the ones to store and the ones already held.
+
+        `known` is extended as it goes: an accepted file is recorded, so a second copy
+        of it later in the same import is refused too. The database cannot answer for
+        rows that do not exist yet.
+
+        Args:
+            files: The incoming files, in the order they should be considered.
+            fingerprints: `(hash, size)` per submitted name, covering every file.
+            known: Fingerprints already accounted for, from `find_duplicate_files`.
+            library: The library being imported into.
+
+        Returns:
+            The files to store, and the refused ones paired with a record of where
+            their bytes already live.
+        """
+        if settings.duplicate_scope == DuplicateScope.OFF:
+            return list(files), []
+
+        accepted: list[FileT] = []
+        rejected: list[tuple[FileT, DuplicateFile]] = []
+
+        for file in files:
+            name = _submitted_name(file)
+            fingerprint = fingerprints[name]
+
+            if (match := known.get(fingerprint)) is not None:
+                rejected.append((file, replace(match, filename=name)))
+                continue
+
+            accepted.append(file)
+            known[fingerprint] = DuplicateFile(
+                filename=name,
+                hash=fingerprint[0],
+                size=fingerprint[1],
+                library_id=library.id,
+            )
+
+        return accepted, rejected
+
     async def create_many_from_files(
-        self, data: BooksCreateFromFiles, library: Library, **kwargs
+        self,
-    ) -> list[Book]:
+        data: BooksCreateFromFiles,
+        library: Library,
+        allow_duplicates: bool = False,
+        **kwargs,
+    ) -> ImportResult:
         """
         Create multiple books from uploaded files.
 
         Groups files by their parent directory to organize books. Files in the root
         directory are treated as separate individual books.
 
+        Files the library already holds are skipped rather than refused: re-dropping a
+        folder to pick up the few books that are new to it is the thing this endpoint
+        exists for, so one already-stored file must not cost the rest of the import.
+
         Args:
             data: Container with list of uploaded files.
             library: The library the books belong to.
-            *args: Additional positional arguments passed to create.
+            allow_duplicates: Store every file, even one already held.
             **kwargs: Additional keyword arguments passed to create.
 
         Returns:
-            List of created Book entities.
+            The books created, and the files skipped along the way.
         """
 
         if not data.files:
@@ -171,26 +440,72 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
             else:
                 books[(filepath.parent, filepath.stem)].append(file)
 
-        return [
+        # Hashed once for the whole request. Screening has to happen before anything is
+        # written, and `_save_book_files` then reuses these rather than reading every
+        # file a second time on its way to disk.
+        fingerprints = await self._fingerprint_uploads(data.files)
+        known = (
+            {}
+            if allow_duplicates
+            else await self.find_duplicate_files(fingerprints.values(), library)
+        )
+
+        result = ImportResult()
+        for files in books.values():
+            if allow_duplicates:
+                accepted, rejected = list(files), []
+            else:
+                accepted, rejected = self._screen_for_duplicates(
+                    files, fingerprints, known, library
+                )
+
+            result.duplicates.extend(duplicate for _, duplicate in rejected)
+
+            # Nothing new in this folder. A book record with no files, and a directory
+            # to match, is worse than not importing it at all.
+            if not accepted:
+                continue
+
+            result.books.append(
                 await self.create_book(
-                {"files": [file for file in files], "library_id": library.id},
+                    {"files": accepted, "library_id": library.id},
                     library,
+                    screen_duplicates=False,
+                    fingerprints=fingerprints,
                     **kwargs,
                 )
-            for files in books.values()
+            )
-        ]
+
+        return result
 
     async def create_many_from_existing_files(
         self,
         file_paths: list[Path],
         consume_path: Path,
         library: Library,
+        allow_duplicates: bool = False,
         **kwargs
-    ) -> list[Book]:
+    ) -> ImportResult:
+        """
+        Import files that are already on disk, from the consume directory.
+
+        Files the library already holds are moved aside rather than imported — see
+        `_quarantine_duplicate`, since there is nobody to ask what to do with them.
+
+        Args:
+            file_paths: Absolute paths of the files to import.
+            consume_path: The library's consume directory, which the paths sit under.
+            library: The library the books belong to.
+            allow_duplicates: Import every file, even one already held.
+            **kwargs: Additional keyword arguments passed to create.
+
+        Returns:
+            The books created, and the files moved aside as duplicates.
+        """
 
         # Group up files if they are in the same leaf directory
 
-        books = []
+        result = ImportResult()
         file_groups: dict[Path, list[Path]] = defaultdict(list)
 
 
@@ -211,22 +526,39 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
 
         # For each grouping
         for group, files in file_groups.items():
-            data: dict[str, Any] = {'files': files}
+            # Fingerprinted before anything moves, since the paths are about to change.
+            fingerprints = {file.name: await fingerprint_file(file) for file in files}
+
+            if allow_duplicates:
+                accepted, rejected = list(files), []
+            else:
+                known = await self.find_duplicate_files(fingerprints.values(), library)
+                accepted, rejected = self._screen_for_duplicates(
+                    files, fingerprints, known, library
+                )
+
+            for file, duplicate in rejected:
+                await self._quarantine_duplicate(file, consume_path, library)
+                result.duplicates.append(duplicate)
+
+            if not accepted:
+                cleanup_empty_parent_directories(consume_path / group, consume_path)
+                continue
+
+            data: dict[str, Any] = {'files': accepted}
             await self._parse_metadata_from_files(data, root_path=consume_path)
             await self._save_cover_image(data)
 
             # Get info from files
             path_gen = BookPathGenerator(library.root_path)
-            parent = path_gen.generate_path(data)
+            parent = await self._reserve_book_path(path_gen.generate_path(data))
             data["path"] = str(parent)
             data["library_id"] = library.id
             file_metadata = []
 
-            for file in files:
+            for file in accepted:
-                stats = await aios.stat(file)
+                file_hash, file_size = fingerprints[file.name]
-                file_size = stats.st_size
                 content_type, _ = mimetypes.guess_type(file)
-                file_hash = await calculate_koreader_hash(file)
 
                 filename = path_gen.generate_filename(data, Path(file.name))
 
@@ -241,21 +573,46 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
 
             data["files"] = file_metadata
 
-            # Move files to appropriate directory
+            # Move files to appropriate directory. A single file is moved by name
-            if len(files) > 1:
+            # rather than by group: the group key is a directory for files that
+            # arrived in one and the file's own relative path for files dropped at the
+            # top level, and moving the directory would nest the file one level below
+            # where its `FileMetadata.path` says it is.
+            if len(accepted) > 1:
                 await move_dir_contents(consume_path / group, parent)
             else:
-                if len(group.parts) > 1:
+                await move_file(accepted[0], parent / accepted[0].name)
-                    await move_file(consume_path / group / files[0].name, parent / files[0].name)
-                else:
-                    await move_file(consume_path / group, parent / group)
 
             cleanup_empty_parent_directories(consume_path / group, consume_path)
 
-            books.append(await super().create(data))
+            result.books.append(await super().create(data))
             await self.repository.session.commit()
 
-        return books
+        return result
+
+    @staticmethod
+    async def _quarantine_duplicate(
+        file: Path, consume_path: Path, library: Library
+    ) -> None:
+        """
+        Move a file the library already holds out of the consume directory.
+
+        Nothing is deleted — the watcher has nobody to ask, and the file is the user's.
+        It cannot stay where it is either: `watchfiles` only reports additions, so a
+        file left behind is never looked at again and only accumulates.
+
+        Args:
+            file: The refused file.
+            consume_path: The library's consume directory, which the file sits under.
+            library: The library it was offered to.
+        """
+        destination = (
+            Path(settings.duplicate_path)
+            / library.slug
+            / file.relative_to(consume_path)
+        )
+
+        await move_file(file, destination)
 
     async def delete_books(
         self,
@@ -417,7 +774,11 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
         # Check if file path must be updated (only for books with files)
         if book.path is not None:
             path_gen = BookPathGenerator(library.root_path)
-            updated_path = path_gen.generate_path(book.to_dict() | data)
+            # Reserved rather than taken: an edit that renames this book onto another
+            # book's path would otherwise move its files in on top of theirs.
+            updated_path = await self._reserve_book_path(
+                path_gen.generate_path(book.to_dict() | data), book_id
+            )
             if str(updated_path) != book.path:
                 # TODO: Move only the files associated with the book instead of the whole directory
                 await move_dir_contents(book.path, updated_path)
@@ -427,7 +788,11 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
         return await super().update(data, item_id=book_id, execution_options={"populate_existing": True})
 
     async def add_files(
-        self, book_id: int, files: list[UploadFile], library: Library
+        self,
+        book_id: int,
+        files: list[UploadFile],
+        library: Library,
+        allow_duplicates: bool = False,
     ) -> None:
         """
         Add additional files to an existing book.
@@ -436,11 +801,36 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
             book_id: The ID of the book.
             files: List of files to add.
             library: The library containing the book.
+            allow_duplicates: Store every file, even one already held.
+
+        Raises:
+            DuplicateFilesError: If a file is already stored under a different book.
         """
         book = await self.get(book_id)
+        fingerprints = await self._fingerprint_uploads(files)
+
+        if not allow_duplicates:
+            # A file this book already carries is not a conflict — adding it again
+            # asks for a state that already holds, so there is simply nothing to do.
+            own = {(file.hash, file.size) for file in book.files}
+            files = [file for file in files if fingerprints[file.filename] not in own]
+
+            known = await self.find_duplicate_files(
+                (fingerprints[file.filename] for file in files), library
+            )
+            files, rejected = self._screen_for_duplicates(
+                files, fingerprints, known, library
+            )
+
+            if rejected:
+                raise DuplicateFilesError([duplicate for _, duplicate in rejected])
+
+        if not files:
+            return
+
         data = book.to_dict()
         data["files"] = files
-        new_files = await self._save_book_files(library, data)
+        new_files = await self._save_book_files(library, data, fingerprints)
         book.files.extend(new_files)
         await self.update_book(book.id, {"files": [file for file in book.files]}, library)
 
@@ -664,7 +1054,12 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
 
         book.identifiers[:] = reconciled
 
-    async def _save_book_files(self, library: Library, data: dict) -> list[FileMetadata]:
+    async def _save_book_files(
+        self,
+        library: Library,
+        data: dict,
+        fingerprints: dict[str, Fingerprint] | None = None,
+    ) -> list[FileMetadata]:
         """
         Save uploaded book files to the filesystem.
 
@@ -674,13 +1069,24 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
         Args:
             library: The library containing the book.
             data: Book data with files to save.
+            fingerprints: `(hash, size)` per submitted filename, for files that have
+                already been hashed by duplicate screening. Anything not covered here
+                is hashed on its way to disk instead.
 
         Returns:
             The data with file paths and metadata populated.
         """
         # Use the library template path with the book's data to generate a filepath
         path_gen = BookPathGenerator(library.root_path)
-        parent = path_gen.generate_path(data)
+
+        # A book that already has a directory keeps it. `add_files` has to land beside
+        # the files that are already there, which is not necessarily where the template
+        # points now — the directory may have been renamed, or moved aside for a book
+        # that generated the same one.
+        if data.get("path"):
+            parent = Path(data["path"])
+        else:
+            parent = await self._reserve_book_path(path_gen.generate_path(data))
             data["path"] = str(parent)
 
         file_metadata = []
@@ -694,14 +1100,21 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
 
             # Store files in the correct directory and return the file's metadata
             await file.seek(0)
-            path = parent / filename
+            parent.mkdir(parents=True, exist_ok=True)
-            path.parent.mkdir(parents=True, exist_ok=True)
+
+            path = _unused_path(parent / filename)
+            filename = path.name
+
+            # Screening has usually hashed the file already; hashing it again on the
+            # way past would read every byte a second time for an answer we hold.
+            fingerprint = fingerprints.get(file.filename) if fingerprints else None
 
             hasher = StreamingHasher()
             async with aiofiles.open(path, "wb") as dest:
                 # Read spooled file and save it to the local filesystem
                 while chunk := await file.read(CHUNK_SIZE):
                     await dest.write(chunk)
+                    if fingerprint is None:
                         hasher.update(chunk)
 
             stats = await aios.stat(path)
@@ -711,7 +1124,7 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
                 FileMetadata(
                     path=str(filename),
                     size=file_size,
-                    hash=hasher.hexdigest(),
+                    hash=fingerprint[0] if fingerprint else hasher.hexdigest(),
                     content_type=file.content_type,
                 )
             )

backend/src/chitai/services/consume.py

@@ -1,6 +1,7 @@
 import asyncio
 from pathlib import Path
 from collections import defaultdict
+from chitai.config import settings
 from chitai.database.models.library import Library
 from chitai.services import BookService, LibraryService
 from chitai.services.metadata_extractor import Extractor
@@ -111,13 +112,19 @@ class ConsumeDirectoryWatcher:
         """Process a batch of files."""
         try:
 
-            books = await self.book_service.create_many_from_existing_files(
+            result = await self.book_service.create_many_from_existing_files(
                 list(file_paths),
                 self.watch_path / Path(library_slug),
                 library=await self._get_library(library_slug),
             )
 
-            print(f"Created {len(books)} books!")
+            print(f"Created {len(result.books)} books!")
+
+            if result.duplicates:
+                print(
+                    f"Moved {len(result.duplicates)} already-stored file(s) "
+                    f"to {settings.duplicate_path}"
+                )
             
         except Exception as e:
             print(f"Error processing batch: {e}")

backend/src/chitai/services/utils.py

@@ -3,6 +3,7 @@
 # Standard library
 from __future__ import annotations
 
+import errno
 import hashlib
 from pathlib import Path
 import shutil
@@ -32,6 +33,9 @@ KO_STEP = 1024
 KO_SAMPLE_SIZE = 1024
 KO_INDICES = range(-1, 11)  # -1 to 10 inclusive
 
+# How much is read at a time while hashing.
+HASH_CHUNK_SIZE = 262144  # 256 KiB
+
 
 def _lshift32(val: int, shift: int) -> int:
     """
@@ -100,10 +104,9 @@ async def calculate_koreader_hash(file_path: Path) -> str:
     offsets = _get_koreader_offsets()
 
     file_pos = 0
-    chunk_size = 262144  # 256 KiB
 
     async with aiofiles.open(file_path, "rb") as f:
-        while chunk := await f.read(chunk_size):
+        while chunk := await f.read(HASH_CHUNK_SIZE):
             _partial_md5_from_chunk(chunk, hasher, offsets, file_pos)
             file_pos += len(chunk)
 
@@ -132,6 +135,49 @@ class StreamingHasher:
         """Return the final hash."""
         return self.hasher.hexdigest()
 
+    @property
+    def size(self) -> int:
+        """Total number of bytes fed in so far."""
+        return self.position
+
+
+async def fingerprint_upload(file: UploadFile) -> tuple[str, int]:
+    """
+    Calculate the hash and byte size of an uploaded file without storing it.
+
+    Duplicate detection has to answer before anything is written to the library, so
+    the file is read here and rewound for whoever writes it afterwards.
+
+    Args:
+        file: The uploaded file to read.
+
+    Returns:
+        The file's `(hash, size)` pair.
+    """
+    hasher = StreamingHasher()
+
+    await file.seek(0)
+    while chunk := await file.read(HASH_CHUNK_SIZE):
+        hasher.update(chunk)
+    await file.seek(0)
+
+    return hasher.hexdigest(), hasher.size
+
+
+async def fingerprint_file(file_path: Path) -> tuple[str, int]:
+    """
+    Calculate the hash and byte size of a file already on disk.
+
+    Args:
+        file_path: Path to the file to read.
+
+    Returns:
+        The file's `(hash, size)` pair.
+    """
+    stats = await aios.stat(file_path)
+    return await calculate_koreader_hash(file_path), stats.st_size
+
+
 ##################################
 #  Filesystem related utilities  #
 ##################################
@@ -184,7 +230,16 @@ async def move_file(src_path: Path, dest_path: Path, create_dirs=True) -> None:
         if dest_dir:  # Only create if there's a directory path
             await aios.makedirs(dest_dir, exist_ok=True)
 
+    try:
         await aios.rename(src_path, dest_path)
+    except OSError as exc:
+        if exc.errno != errno.EXDEV:
+            raise
+
+        # Source and destination are on different filesystems, which rename cannot
+        # cross. Libraries, the consume directory and the duplicates directory are
+        # all configured separately, so they can easily be separate mounts.
+        shutil.move(str(src_path), str(dest_path))
 
 async def move_dir_contents(source_dir: Path | str, target_dir: Path | str) -> None:
     """

backend/tests/conftest.py

@@ -40,7 +40,13 @@ pytest_plugins = [
 
 @pytest.fixture(autouse=True)
 def _patch_settings(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr(settings, "book_cover_path", f"{tmp_path}/covers")
+    # PIL will not create the directory it is asked to save into, so anything that
+    # imports a file carrying a cover needs it to exist first.
+    covers = tmp_path / "covers"
+    covers.mkdir()
+
+    monkeypatch.setattr(settings, "book_cover_path", str(covers))
+    monkeypatch.setattr(settings, "duplicate_path", str(tmp_path / "duplicates"))
 
 
 @pytest.fixture(name="engine")

backend/tests/integration/test_book.py

@@ -366,7 +366,8 @@ async def test_create_multiple_books_from_directory(
 
     assert response.status_code == 201
     data = response.json()
-    assert len(data.get("items") or data.get("data")) >= 1
+    assert len(data["created"]) == 2
+    assert data["skipped"] == []
 
 
 async def test_create_books_from_parent_directory_keeps_embedded_title(
@@ -396,7 +397,7 @@ async def test_create_books_from_parent_directory_keeps_embedded_title(
 
     assert response.status_code == 201
 
-    books = response.json()["items"]
+    books = response.json()["created"]
     assert len(books) == 1
     assert books[0]["title"] == "Metamorphosis"
 
@@ -419,11 +420,176 @@ async def test_create_books_groups_formats_within_one_folder(
 
     assert response.status_code == 201
 
-    books = response.json()["items"]
+    books = response.json()["created"]
     assert len(books) == 1
     assert len(books[0]["files"]) == 2
 
 
+class TestDuplicateHandling:
+    """A file the library already holds must not be stored a second time."""
+
+    epub_path = Path("tests/data_files/Metamorphosis - Franz Kafka.epub")
+
+    def upload(self, name: str | None = None) -> list[tuple[str, tuple]]:
+        return [
+            (
+                "files",
+                (
+                    name or self.epub_path.name,
+                    self.epub_path.read_bytes(),
+                    "application/epub+zip",
+                ),
+            )
+        ]
+
+    async def test_bulk_upload_reports_skipped_files(
+        self, authenticated_client: AsyncClient
+    ) -> None:
+        """Re-dropping a folder must import what is new and name what was not."""
+        first = await authenticated_client.post(
+            "/books/fromFiles?library_id=1", files=self.upload()
+        )
+        assert first.status_code == 201
+        created = first.json()["created"][0]
+
+        second = await authenticated_client.post(
+            "/books/fromFiles?library_id=1", files=self.upload()
+        )
+
+        assert second.status_code == 201
+        result = second.json()
+        assert result["created"] == []
+        assert len(result["skipped"]) == 1
+
+        skipped = result["skipped"][0]
+        assert skipped["filename"] == self.epub_path.name
+        assert skipped["book_id"] == created["id"]
+        assert skipped["book_title"] == created["title"]
+
+    async def test_bulk_upload_can_be_forced(
+        self, authenticated_client: AsyncClient
+    ) -> None:
+        await authenticated_client.post(
+            "/books/fromFiles?library_id=1", files=self.upload()
+        )
+
+        response = await authenticated_client.post(
+            "/books/fromFiles?library_id=1&allow_duplicates=true", files=self.upload()
+        )
+
+        assert response.status_code == 201
+        assert len(response.json()["created"]) == 1
+        assert response.json()["skipped"] == []
+
+    async def test_single_book_create_conflicts(
+        self, authenticated_client: AsyncClient
+    ) -> None:
+        """Naming files deliberately earns a refusal rather than a silent drop."""
+        await authenticated_client.post(
+            "/books/fromFiles?library_id=1", files=self.upload()
+        )
+
+        response = await authenticated_client.post(
+            "/books?library_id=1", files=self.upload(), data={"library_id": 1}
+        )
+
+        assert response.status_code == 409
+        assert response.json()["extra"][0]["filename"] == self.epub_path.name
+
+        forced = await authenticated_client.post(
+            "/books?library_id=1&allow_duplicates=true",
+            files=self.upload(),
+            data={"library_id": 1},
+        )
+        assert forced.status_code == 201
+
+    async def test_adding_another_books_file_conflicts(
+        self, authenticated_client: AsyncClient
+    ) -> None:
+        created = await authenticated_client.post(
+            "/books/fromFiles?library_id=1", files=self.upload()
+        )
+        book_id = created.json()["created"][0]["id"]
+
+        other = await authenticated_client.post(
+            "/books?library_id=1",
+            files=[
+                (
+                    "files",
+                    (
+                        "war.epub",
+                        Path("tests/data_files/The Art of War - Sun Tzu.epub").read_bytes(),
+                        "application/epub+zip",
+                    ),
+                )
+            ],
+            data={"library_id": 1},
+        )
+        other_id = other.json()["id"]
+
+        response = await authenticated_client.post(
+            f"/books/{book_id}/files",
+            files=[
+                (
+                    "files",
+                    (
+                        "war.epub",
+                        Path("tests/data_files/The Art of War - Sun Tzu.epub").read_bytes(),
+                        "application/epub+zip",
+                    ),
+                )
+            ],
+        )
+
+        assert response.status_code == 409
+        assert response.json()["extra"][0]["book_id"] == other_id
+
+    async def test_resending_a_books_own_file_changes_nothing(
+        self, authenticated_client: AsyncClient
+    ) -> None:
+        created = await authenticated_client.post(
+            "/books/fromFiles?library_id=1", files=self.upload()
+        )
+        book_id = created.json()["created"][0]["id"]
+
+        response = await authenticated_client.post(
+            f"/books/{book_id}/files", files=self.upload()
+        )
+
+        assert response.status_code == 201
+        assert len(response.json()["files"]) == 1
+
+    async def test_duplicates_can_be_checked_before_uploading(
+        self, authenticated_client: AsyncClient
+    ) -> None:
+        """The pre-flight check answers from hashes alone, with no file sent."""
+        created = await authenticated_client.post(
+            "/books/fromFiles?library_id=1", files=self.upload()
+        )
+        book = created.json()["created"][0]
+        stored = book["files"][0]
+
+        response = await authenticated_client.post(
+            "/books/duplicates?library_id=1",
+            json=[
+                {
+                    "hash": stored["hash"],
+                    "size": stored["size"],
+                    "filename": "local-copy.epub",
+                },
+                {"hash": stored["hash"], "size": stored["size"] + 1},
+                {"hash": "0" * 32, "size": 1234},
+            ],
+        )
+
+        assert response.status_code == 200
+
+        matches = response.json()
+        assert len(matches) == 1
+        assert matches[0]["filename"] == "local-copy.epub"
+        assert matches[0]["book_id"] == book["id"]
+
+
 # NOTE: the multi-book ZIP download is covered at the service level, in
 # tests/unit/test_services/test_book_service.py. Driving `/books/download` through
 # AsyncTestClient hangs in fixture teardown: it is the only `Stream` endpoint in the

backend/tests/unit/test_services/test_book_service.py

@@ -6,11 +6,29 @@ from pathlib import Path
 
 import pytest
 import aiofiles.os as aios
+from litestar.datastructures import UploadFile
+from sqlalchemy.ext.asyncio import AsyncSession
 
-from chitai.schemas import BookCreate
+from chitai.config import DuplicateScope, settings
+from chitai.schemas import BookCreate, BooksCreateFromFiles
 from chitai.services import BookService
+from chitai.services.book import DuplicateFilesError
 from chitai.database import models as m
 
+DATA_FILES = Path("tests/data_files")
+EPUB = DATA_FILES / "Metamorphosis - Franz Kafka.epub"
+OTHER_EPUB = DATA_FILES / "The Art of War - Sun Tzu.epub"
+PDF = DATA_FILES / "Calculus Made Easy - Silvanus Thompson.pdf"
+
+
+def upload(path: Path, name: str | None = None) -> UploadFile:
+    """An uploaded file carrying the bytes of one of the test fixtures."""
+    return UploadFile(
+        content_type="application/pdf" if path.suffix == ".pdf" else "application/epub+zip",
+        filename=name or path.name,
+        file_data=path.read_bytes(),
+    )
+
 
 @pytest.mark.asyncio
 class TestBookServiceCRUD:
@@ -202,3 +220,378 @@ class TestBookServiceCRUD:
             for name, payload in contents.items():
                 entry = next(n for n in names if Path(n).name == name)
                 assert archive.read(entry) == payload
+
+
+@pytest.mark.asyncio
+class TestBookServiceDuplicates:
+    """Files that are already stored must not be stored again."""
+
+    async def test_duplicate_upload_is_skipped_and_reported(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """The second import of a file creates nothing and names where it already is."""
+        first = await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]), test_library
+        )
+
+        assert len(first.books) == 1
+        assert first.duplicates == []
+
+        second = await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]), test_library
+        )
+
+        assert second.books == []
+        assert len(second.duplicates) == 1
+
+        duplicate = second.duplicates[0]
+        assert duplicate.filename == EPUB.name
+        assert duplicate.book_id == first.books[0].id
+        assert duplicate.book_title == first.books[0].title
+
+    async def test_new_format_beside_a_duplicate_is_still_imported(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """A folder is not all-or-nothing: the file that is new must still land."""
+        await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB, "Metamorphosis/book.epub")]),
+            test_library,
+        )
+
+        result = await books_service.create_many_from_files(
+            BooksCreateFromFiles(
+                files=[
+                    upload(EPUB, "Metamorphosis/book.epub"),
+                    upload(PDF, "Metamorphosis/book.pdf"),
+                ]
+            ),
+            test_library,
+        )
+
+        assert len(result.books) == 1
+        assert [d.filename for d in result.duplicates] == ["Metamorphosis/book.epub"]
+
+        book = await books_service.get(result.books[0].id)
+        assert [file.path for file in book.files] == ["book.pdf"]
+
+    async def test_duplicate_within_one_upload_is_caught(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """The same bytes twice in one request has no row to match against yet."""
+        result = await books_service.create_many_from_files(
+            BooksCreateFromFiles(
+                files=[upload(EPUB, "first.epub"), upload(EPUB, "second.epub")]
+            ),
+            test_library,
+        )
+
+        assert len(result.books) == 1
+        assert len(result.duplicates) == 1
+        assert result.duplicates[0].filename == "second.epub"
+
+        # Nothing in the database holds it yet, so there is no book to point at.
+        assert result.duplicates[0].book_id is None
+
+    async def test_allow_duplicates_stores_the_file_anyway(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """The escape hatch has to work: the hash is not proof of identity."""
+        await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]), test_library
+        )
+
+        result = await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]),
+            test_library,
+            allow_duplicates=True,
+        )
+
+        assert len(result.books) == 1
+        assert result.duplicates == []
+
+    async def test_a_different_file_is_not_a_duplicate(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        result = await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB), upload(OTHER_EPUB)]), test_library
+        )
+
+        assert len(result.books) == 2
+        assert result.duplicates == []
+
+    async def test_matching_hash_with_a_different_size_is_not_a_duplicate(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """The hash samples 12 KiB, so the size is what makes a match trustworthy."""
+        created = await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]), test_library
+        )
+        stored = (await books_service.get(created.books[0].id)).files[0]
+
+        matches = await books_service.find_duplicate_files(
+            [(stored.hash, stored.size), (stored.hash, stored.size + 1)], test_library
+        )
+
+        assert (stored.hash, stored.size) in matches
+        assert (stored.hash, stored.size + 1) not in matches
+
+    async def test_create_book_refuses_and_writes_nothing(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """A single-book create names its files, so it is refused rather than trimmed."""
+        await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]), test_library
+        )
+
+        before = sorted(Path(test_library.root_path).rglob("*"))
+
+        with pytest.raises(DuplicateFilesError) as excinfo:
+            await books_service.create_book(
+                BookCreate(
+                    library_id=test_library.id,
+                    title="Metamorphosis",
+                    authors=["Franz Kafka"],
+                    files=[upload(EPUB)],
+                ).model_dump(),
+                test_library,
+            )
+
+        assert len(excinfo.value.duplicates) == 1
+        assert sorted(Path(test_library.root_path).rglob("*")) == before
+
+    async def test_scope_decides_whether_libraries_share(
+        self,
+        books_service: BookService,
+        test_library: m.Library,
+        session: AsyncSession,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """A second library is a separate collection by default, and not under `global`."""
+        other = m.Library(
+            name="Second Library",
+            slug="second-library",
+            root_path=str(tmp_path / "second"),
+            path_template=test_library.path_template,
+            read_only=False,
+        )
+        session.add(other)
+        await session.commit()
+
+        await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]), test_library
+        )
+
+        result = await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]), other
+        )
+        assert len(result.books) == 1
+        assert result.duplicates == []
+
+        monkeypatch.setattr(settings, "duplicate_scope", DuplicateScope.GLOBAL)
+
+        result = await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]), other
+        )
+        assert result.books == []
+        assert len(result.duplicates) == 1
+
+    async def test_scope_off_disables_detection(
+        self,
+        books_service: BookService,
+        test_library: m.Library,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        monkeypatch.setattr(settings, "duplicate_scope", DuplicateScope.OFF)
+
+        await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]), test_library
+        )
+        result = await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB), upload(EPUB, "copy.epub")]),
+            test_library,
+        )
+
+        assert len(result.books) == 2
+        assert result.duplicates == []
+
+    async def test_re_adding_a_file_to_its_own_book_does_nothing(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """Asking for a state that already holds is not a conflict."""
+        created = await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]), test_library
+        )
+        book_id = created.books[0].id
+
+        await books_service.add_files(book_id, [upload(EPUB)], test_library)
+
+        book = await books_service.get(book_id)
+        assert len(book.files) == 1
+
+    async def test_adding_another_books_file_is_refused(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        created = await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB), upload(OTHER_EPUB)]), test_library
+        )
+        first, second = created.books
+
+        with pytest.raises(DuplicateFilesError) as excinfo:
+            await books_service.add_files(first.id, [upload(OTHER_EPUB)], test_library)
+
+        assert excinfo.value.duplicates[0].book_id == second.id
+        assert len((await books_service.get(first.id)).files) == 1
+
+    async def test_consume_duplicate_is_moved_aside(
+        self,
+        books_service: BookService,
+        test_library: m.Library,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """The watcher cannot ask, so a refused file is parked rather than dropped."""
+        quarantine = tmp_path / "duplicates"
+        monkeypatch.setattr(settings, "duplicate_path", str(quarantine))
+
+        await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]), test_library
+        )
+
+        consume = tmp_path / "consume"
+        await aios.makedirs(consume)
+        dropped = consume / EPUB.name
+        dropped.write_bytes(EPUB.read_bytes())
+
+        result = await books_service.create_many_from_existing_files(
+            [dropped], consume, test_library
+        )
+
+        assert result.books == []
+        assert len(result.duplicates) == 1
+        assert not dropped.exists()
+        assert (quarantine / test_library.slug / EPUB.name).is_file()
+
+    async def test_consume_imports_what_is_new(
+        self,
+        books_service: BookService,
+        test_library: m.Library,
+        tmp_path: Path,
+    ) -> None:
+        """The screening must not disturb the ordinary consume import."""
+        consume = tmp_path / "consume"
+        await aios.makedirs(consume / "Metamorphosis")
+        dropped = consume / "Metamorphosis" / EPUB.name
+        dropped.write_bytes(EPUB.read_bytes())
+
+        result = await books_service.create_many_from_existing_files(
+            [dropped], consume, test_library
+        )
+
+        assert len(result.books) == 1
+        assert result.duplicates == []
+
+        book = await books_service.get(result.books[0].id)
+        assert (Path(book.path) / book.files[0].path).is_file()
+
+
+@pytest.mark.asyncio
+class TestBookPathCollisions:
+    """Two books must never share a directory, whatever their metadata says."""
+
+    async def test_forced_duplicate_gets_its_own_copy(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """`allow_duplicates` must add a book, not overwrite the one already there.
+
+        The path comes from the metadata alone, so a forced duplicate generates the
+        same directory and the same filename. Writing it lands on top of the original:
+        one file on disk, two books pointing at it, and deleting either takes both.
+        """
+        first = await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]), test_library
+        )
+        second = await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]),
+            test_library,
+            allow_duplicates=True,
+        )
+
+        original = await books_service.get(first.books[0].id)
+        forced = await books_service.get(second.books[0].id)
+
+        assert original.path != forced.path
+
+        paths = {
+            Path(book.path) / book.files[0].path for book in (original, forced)
+        }
+        assert len(paths) == 2
+        assert all(path.is_file() for path in paths)
+
+    async def test_deleting_a_forced_duplicate_keeps_the_original(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        first = await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]), test_library
+        )
+        second = await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]),
+            test_library,
+            allow_duplicates=True,
+        )
+
+        original = await books_service.get(first.books[0].id)
+        kept = Path(original.path) / original.files[0].path
+
+        await books_service.delete_books(
+            [second.books[0].id], test_library, delete_files=True
+        )
+
+        assert kept.is_file()
+
+    async def test_editing_metadata_cannot_merge_into_another_book(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """A rename that collides must step aside rather than move in on top."""
+        first = await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]), test_library
+        )
+        second = await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(OTHER_EPUB)]), test_library
+        )
+
+        original = await books_service.get(first.books[0].id)
+
+        # Renamed onto the first book's author and title.
+        await books_service.update_book(
+            second.books[0].id,
+            {"title": original.title, "authors": [author.name for author in original.authors]},
+            test_library,
+        )
+
+        moved = await books_service.get(second.books[0].id)
+
+        assert moved.path != original.path
+        assert (Path(original.path) / original.files[0].path).is_file()
+        assert (Path(moved.path) / moved.files[0].path).is_file()
+
+    async def test_adding_a_file_lands_in_the_books_own_directory(
+        self, books_service: BookService, test_library: m.Library
+    ) -> None:
+        """`add_files` must follow the book, not the template."""
+        await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]), test_library
+        )
+        forced = await books_service.create_many_from_files(
+            BooksCreateFromFiles(files=[upload(EPUB)]),
+            test_library,
+            allow_duplicates=True,
+        )
+        book_id = forced.books[0].id
+
+        await books_service.add_files(book_id, [upload(PDF)], test_library)
+
+        book = await books_service.get(book_id)
+        assert len(book.files) == 2
+        for file in book.files:
+            assert (Path(book.path) / file.path).is_file()