PartsInventorySystem/app/main.py

"""Parts inventory API.

JSON API plus a single self-contained frontend. Everything a bench lookup needs
lives behind /api; the browser app in static/ is the only consumer today, and
the future "what could I build with this?" feature is meant to be another
consumer of the same endpoints rather than a fork of them.
"""

import asyncio
import hashlib
import json
import math
import os
import secrets
import sqlite3
from contextlib import asynccontextmanager
from typing import Annotated, Any, Literal

from fastapi import (Depends, FastAPI, File, Form, HTTPException, Query, Request,
                     Response, UploadFile)
from fastapi.exceptions import RequestValidationError
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from pydantic import AfterValidator, BaseModel, Field

from . import auth, db

STATIC_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "static")

# A downscaled phone photo lands around 300KB; the cap is generous enough for an
# un-resized original from a browser that couldn't downscale, and small enough
# that nothing can fill the volume by accident.
MAX_IMAGE_BYTES = 8 * 1024 * 1024
MAX_IMAGES_PER_PART = 12
# Multipart framing adds boundaries and headers around the file, so the whole
# request is allowed slightly more than one image.
MAX_BODY_BYTES = MAX_IMAGE_BYTES + 1024 * 1024

CSP = (
    "default-src 'self'; "
    "img-src 'self' data: blob:; "
    "style-src 'self' 'unsafe-inline'; "   # the UI sets inline style attributes
    "script-src 'self'; "
    "connect-src 'self'; "
    "frame-ancestors 'none'; "
    "base-uri 'none'; "
    "form-action 'self'"
)


class BodySizeLimit:
    """Reject an over-large request before anything parses it.

    Starlette parses and spools an entire multipart body *before* the route's
    dependencies run, so a route-level cap — and the login check — both happen
    after the bytes have already landed in the container's temporary storage.
    A caller who doesn't know the password could still make us write them.

    This is plain ASGI so it sits outside routing. Caddy enforces the same
    ceiling at the edge; this is the backstop for anything that reaches the app
    directly.
    """

    def __init__(self, app, max_bytes: int):
        self.app = app
        self.max_bytes = max_bytes

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            return await self.app(scope, receive, send)

        for name, value in scope.get("headers") or []:
            if name == b"content-length":
                try:
                    declared = int(value)
                except ValueError:
                    break
                if declared > self.max_bytes:
                    return await self._too_large(send)
                break

        seen = 0

        async def limited_receive():
            nonlocal seen
            message = await receive()
            if message.get("type") == "http.request":
                seen += len(message.get("body", b""))
                if seen > self.max_bytes:
                    # Chunked upload with no declared length: cut the stream so
                    # the parser fails rather than letting it run unbounded.
                    return {"type": "http.request", "body": b"", "more_body": False}
            return message

        await self.app(scope, limited_receive, send)

    async def _too_large(self, send):
        body = json.dumps(
            {"detail": f"Request body must be {self.max_bytes // (1024 * 1024)}MB or smaller"}
        ).encode()
        await send({"type": "http.response.start", "status": 413,
                    "headers": [(b"content-type", b"application/json"),
                                (b"content-length", str(len(body)).encode())]})
        await send({"type": "http.response.body", "body": body})


@asynccontextmanager
async def lifespan(_app: FastAPI):
    db.init()
    yield


app = FastAPI(title="Parts Inventory", docs_url=None, redoc_url=None, lifespan=lifespan)
# Added last, so it wraps everything else and runs before routing or body parsing.
app.add_middleware(BodySizeLimit, max_bytes=MAX_BODY_BYTES)


# --- validated field types --------------------------------------------------

def _required(v: str) -> str:
    v = v.strip()
    if not v:
        raise ValueError("must not be blank")
    return v


RequiredText = Annotated[str, AfterValidator(_required)]
Text = Annotated[str, AfterValidator(lambda v: v.strip())]


class SpecIn(BaseModel):
    key: RequiredText = Field(max_length=80)
    value: Text = Field(default="", max_length=400)


class PartIn(BaseModel):
    name: RequiredText = Field(max_length=200)
    description: Text = Field(default="", max_length=2000)
    category_id: int | None = None
    location_id: int | None = None
    manufacturer: Text = Field(default="", max_length=200)
    mpn: Text = Field(default="", max_length=120)
    quantity: float = Field(default=0, ge=0, allow_inf_nan=False)
    unit: RequiredText = Field(default="pcs", max_length=20)
    min_quantity: float | None = Field(default=None, ge=0, allow_inf_nan=False)
    cost_each: float | None = Field(default=None, ge=0, allow_inf_nan=False)
    datasheet_url: Text = Field(default="", max_length=1000)
    product_url: Text = Field(default="", max_length=1000)
    notes: Text = Field(default="", max_length=4000)
    specs: list[SpecIn] = []
    tags: list[str] = []


class PartPatch(BaseModel):
    name: RequiredText | None = Field(default=None, max_length=200)
    description: Text | None = Field(default=None, max_length=2000)
    category_id: int | None = None
    location_id: int | None = None
    manufacturer: Text | None = Field(default=None, max_length=200)
    mpn: Text | None = Field(default=None, max_length=120)
    quantity: float | None = Field(default=None, ge=0, allow_inf_nan=False)
    unit: RequiredText | None = Field(default=None, max_length=20)
    min_quantity: float | None = Field(default=None, ge=0, allow_inf_nan=False)
    cost_each: float | None = Field(default=None, ge=0, allow_inf_nan=False)
    datasheet_url: Text | None = Field(default=None, max_length=1000)
    product_url: Text | None = Field(default=None, max_length=1000)
    notes: Text | None = Field(default=None, max_length=4000)
    specs: list[SpecIn] | None = None
    tags: list[str] | None = None
    reason: Text = Field(default="edited", max_length=300)


# Columns a PATCH may legitimately set back to NULL. Everything else is NOT NULL
# in the schema, so an explicit null there is a client error, not a 500.
NULLABLE_COLUMNS = {"category_id", "location_id", "min_quantity", "cost_each"}
PART_COLUMNS = [
    "name", "description", "category_id", "location_id", "manufacturer", "mpn",
    "quantity", "unit", "min_quantity", "cost_each", "datasheet_url", "product_url", "notes",
]


class ImagePatch(BaseModel):
    caption: Text | None = Field(default=None, max_length=300)
    position: int | None = Field(default=None, ge=0)


class AdjustIn(BaseModel):
    delta: float = Field(allow_inf_nan=False)
    reason: Text = Field(default="", max_length=300)


class NodeIn(BaseModel):
    name: RequiredText = Field(max_length=120)
    parent_id: int | None = None
    unit: RequiredText = Field(default="pcs", max_length=20)
    spec_template: list[SpecIn] = []
    notes: Text = Field(default="", max_length=1000)
    sort_order: int = 0


class LoginIn(BaseModel):
    password: str = Field(max_length=500)


class PasswordChangeIn(BaseModel):
    current_password: str = Field(max_length=500)
    new_password: str = Field(max_length=500)


# --- helpers ----------------------------------------------------------------

def _specs_for(conn, part_ids: list[int]) -> dict[int, list[dict]]:
    if not part_ids:
        return {}
    marks = ",".join("?" * len(part_ids))
    rows = conn.execute(
        f"SELECT part_id, key, value FROM part_specs WHERE part_id IN ({marks}) ORDER BY position, key",
        part_ids,
    ).fetchall()
    out: dict[int, list[dict]] = {}
    for r in rows:
        out.setdefault(r["part_id"], []).append({"key": r["key"], "value": r["value"]})
    return out


def _tags_for(conn, part_ids: list[int]) -> dict[int, list[str]]:
    if not part_ids:
        return {}
    marks = ",".join("?" * len(part_ids))
    rows = conn.execute(
        f"""SELECT pt.part_id, t.name FROM part_tags pt
            JOIN tags t ON t.id = pt.tag_id
            WHERE pt.part_id IN ({marks}) ORDER BY t.name""",
        part_ids,
    ).fetchall()
    out: dict[int, list[str]] = {}
    for r in rows:
        out.setdefault(r["part_id"], []).append(r["name"])
    return out


def _images_for(conn, part_ids: list[int]) -> dict[int, list[dict]]:
    if not part_ids:
        return {}
    marks = ",".join("?" * len(part_ids))
    rows = conn.execute(
        f"""SELECT part_id, token, mime, bytes, caption, position
            FROM part_images WHERE part_id IN ({marks})
            ORDER BY position, id""",
        part_ids,
    ).fetchall()
    out: dict[int, list[dict]] = {}
    for r in rows:
        out.setdefault(r["part_id"], []).append(dict(r))
    return out


def _serialise(rows, specs, tags, cat_paths, loc_paths, images=None) -> list[dict]:
    images = images or {}
    out = []
    for r in rows:
        d = dict(r)
        d["specs"] = specs.get(r["id"], [])
        d["tags"] = tags.get(r["id"], [])
        d["images"] = images.get(r["id"], [])
        d["category_path"] = cat_paths.get(r["category_id"]) if r["category_id"] else None
        d["location_path"] = loc_paths.get(r["location_id"]) if r["location_id"] else None
        d["low_stock"] = (
            r["min_quantity"] is not None and r["quantity"] <= r["min_quantity"]
        )
        out.append(d)
    return out


def _write_specs(conn, part_id: int, specs: list[SpecIn]):
    conn.execute("DELETE FROM part_specs WHERE part_id = ?", (part_id,))
    seen = set()
    position = 0
    for s in specs:
        if s.key.lower() in seen:
            continue
        seen.add(s.key.lower())
        conn.execute(
            "INSERT INTO part_specs(part_id, key, value, position) VALUES (?,?,?,?)",
            (part_id, s.key, s.value, position),
        )
        position += 1


def _write_tags(conn, part_id: int, tags: list[str]):
    conn.execute("DELETE FROM part_tags WHERE part_id = ?", (part_id,))
    for raw in tags:
        name = raw.strip()
        if not name:
            continue
        conn.execute("INSERT OR IGNORE INTO tags(name) VALUES (?)", (name,))
        row = conn.execute("SELECT id FROM tags WHERE name = ? COLLATE NOCASE", (name,)).fetchone()
        if row:
            conn.execute(
                "INSERT OR IGNORE INTO part_tags(part_id, tag_id) VALUES (?,?)",
                (part_id, row["id"]),
            )


def _log_stock(conn, part_id: int, delta: float, after: float, reason: str):
    conn.execute(
        "INSERT INTO stock_log(part_id, delta, quantity_after, reason) VALUES (?,?,?,?)",
        (part_id, delta, after, reason),
    )


def _fetch_part(conn, part_id: int) -> dict:
    row = conn.execute("SELECT * FROM parts WHERE id = ?", (part_id,)).fetchone()
    if row is None:
        raise HTTPException(404, "Part not found")
    return _serialise(
        [row], _specs_for(conn, [part_id]), _tags_for(conn, [part_id]),
        db.tree_paths(conn, "categories"), db.tree_paths(conn, "locations"),
        _images_for(conn, [part_id]),
    )[0]


def _require_node(conn, table: str, node_id: int, label: str):
    if conn.execute(f"SELECT 1 FROM {table} WHERE id = ?", (node_id,)).fetchone() is None:
        raise HTTPException(404, f"{label} not found")


def _check_parent(conn, table: str, node_id: int, parent_id: int | None, label: str):
    """Reject re-parenting a node under itself or any of its own descendants.

    A direct self-parent is the obvious case; the indirect one (A under B, where
    B is already a child of A) is the one that actually corrupts the tree.
    """
    if parent_id is None:
        return
    if parent_id == node_id:
        raise HTTPException(400, f"A {label} cannot be its own parent")
    _require_node(conn, table, parent_id, f"Parent {label}")
    if parent_id in db.descendants(conn, table, node_id):
        raise HTTPException(400, f"A {label} cannot be moved beneath itself")


# --- error handling ---------------------------------------------------------

def _json_safe(value):
    """Make a validation-error payload encodable.

    FastAPI echoes the rejected value back in the error detail. When that value
    is Infinity or NaN — the very thing being rejected — the JSON encoder
    refuses it and a clean 422 turns into a serialisation failure. Stringify
    anything the encoder can't represent.
    """
    if isinstance(value, float):
        return value if math.isfinite(value) else str(value)
    if isinstance(value, dict):
        return {str(k): _json_safe(v) for k, v in value.items()}
    if isinstance(value, (list, tuple, set)):
        return [_json_safe(v) for v in value]
    if isinstance(value, (str, int, bool)) or value is None:
        return value
    return str(value)


@app.exception_handler(RequestValidationError)
def validation_error(request: Request, exc: RequestValidationError):
    return JSONResponse({"detail": _json_safe(exc.errors())}, status_code=422)


# --- middleware -------------------------------------------------------------

@app.middleware("http")
async def security_headers(request: Request, call_next):
    response = await call_next(request)
    headers = response.headers
    headers.setdefault("X-Content-Type-Options", "nosniff")
    headers.setdefault("X-Frame-Options", "DENY")
    headers.setdefault("Referrer-Policy", "no-referrer")
    headers.setdefault("Content-Security-Policy", CSP)
    if auth.is_https(request):
        headers.setdefault("Strict-Transport-Security", "max-age=31536000; includeSubDomains")
    if request.url.path.startswith("/static/"):
        # Asset URLs carry a content hash (see _index_html), so a versioned URL
        # is safe to cache forever and a deploy simply changes the URL. Without
        # the hash — someone hitting the bare path — it must not be cached, or
        # an intermediary can pin stale frontend code against a new API.
        if request.query_params.get("v"):
            headers["Cache-Control"] = "public, max-age=31536000, immutable"
        else:
            headers["Cache-Control"] = "no-cache"
    elif request.url.path.startswith("/api/"):
        # setdefault, not assignment: the image route serves immutable,
        # token-addressed files and sets its own caching.
        headers.setdefault("Cache-Control", "no-store")
    return response


# --- auth routes ------------------------------------------------------------

def _set_session_cookie(response: Response, request: Request, conn):
    response.set_cookie(
        auth.COOKIE_NAME,
        auth.issue_token(conn),
        max_age=auth.session_days() * 86400,
        httponly=True,
        samesite="lax",
        secure=auth.is_https(request),
        path="/",
    )


@app.get("/api/me")
def me(request: Request, conn=Depends(db.get_db)):
    if not auth.auth_enabled():
        return {"authenticated": True, "auth_required": False, "using_bootstrap_password": False}
    token = request.cookies.get(auth.COOKIE_NAME)
    authenticated = bool(token and auth.token_valid(token, conn))
    return {
        "authenticated": authenticated,
        "auth_required": True,
        # Surfaced so the UI can nag until the handed-over password is replaced.
        "using_bootstrap_password": authenticated and auth.using_bootstrap_password(conn),
        "min_password_length": auth.MIN_PASSWORD_LENGTH,
    }


@app.post("/api/login")
async def login(body: LoginIn, response: Response, request: Request, conn=Depends(db.get_db)):
    if not auth.auth_enabled():
        return {"authenticated": True, "auth_required": False}

    retry_after = auth.login_retry_after()
    if retry_after:
        raise HTTPException(
            429, "Too many failed attempts", headers={"Retry-After": str(retry_after)}
        )

    if not auth.check_password(conn, body.password):
        auth.record_failure()
        # Async sleep, so a burst of guesses can't tie up the worker threadpool
        # that real requests need.
        await asyncio.sleep(0.5)
        raise HTTPException(401, "Incorrect password")

    auth.clear_failures()
    _set_session_cookie(response, request, conn)
    return {
        "authenticated": True,
        "auth_required": True,
        "using_bootstrap_password": auth.using_bootstrap_password(conn),
    }


@app.post("/api/password", dependencies=[Depends(auth.require_auth)])
async def change_password(
    body: PasswordChangeIn, response: Response, request: Request, conn=Depends(db.get_db)
):
    """Change the password from inside the app.

    Requires the current password even though the caller already holds a valid
    session: a borrowed browser tab should not be enough to lock the owner out.
    """
    if not auth.auth_enabled():
        raise HTTPException(400, "Authentication is disabled, so there is no password to change")

    # This route verifies the same credential the login form does, so it has to
    # honour the same budget. It was recording failures without checking them,
    # which meant a borrowed session could guess the password without limit —
    # while still locking the owner out of /api/login.
    retry_after = auth.login_retry_after()
    if retry_after:
        raise HTTPException(
            429, "Too many failed attempts", headers={"Retry-After": str(retry_after)}
        )

    if not auth.check_password(conn, body.current_password):
        auth.record_failure()
        await asyncio.sleep(0.5)
        raise HTTPException(403, "Current password is incorrect")

    problem = auth.password_problem(body.new_password)
    if problem:
        raise HTTPException(422, problem)
    if body.new_password == body.current_password:
        raise HTTPException(422, "New password must be different from the current one")

    auth.clear_failures()
    # Bumps the session epoch, so every cookie issued before now stops working.
    auth.set_password(conn, body.new_password)
    # Re-issue for the caller, so changing the password doesn't log you out of
    # the tab you changed it in.
    _set_session_cookie(response, request, conn)
    return {"changed": True, "other_sessions_signed_out": True}


@app.post("/api/sessions/revoke", dependencies=[Depends(auth.require_auth)])
def revoke_sessions(response: Response, request: Request, conn=Depends(db.get_db)):
    """Sign out every other device, keeping this one signed in."""
    auth.bump_epoch(conn)
    _set_session_cookie(response, request, conn)
    return {"revoked": True}


@app.post("/api/logout")
def logout(response: Response):
    response.delete_cookie(auth.COOKIE_NAME, path="/")
    return {"authenticated": False}


# --- parts ------------------------------------------------------------------

@app.get("/api/parts", dependencies=[Depends(auth.require_auth)])
def list_parts(
    conn=Depends(db.get_db),
    q: str = "",
    category_id: int | None = None,
    location_id: int | None = None,
    tag: str = "",
    low_stock: bool = False,
    sort: Literal["relevance", "name", "quantity", "updated", "created", "location"] = "relevance",
    limit: int = Query(default=200, ge=1, le=1000),
    offset: int = Query(default=0, ge=0),
):
    where: list[str] = []
    params: list[Any] = []
    joins = ""
    order = "p.name COLLATE NOCASE ASC"

    q = q.strip()
    if q:
        match = db.fts_query(q)
        if match:
            joins += " JOIN parts_fts ON parts_fts.rowid = p.id "
            where.append("parts_fts MATCH ?")
            params.append(match)
            if sort == "relevance":
                order = "parts_fts.rank ASC, p.name COLLATE NOCASE ASC"

    if category_id is not None:
        ids = db.descendants(conn, "categories", category_id)
        where.append(f"p.category_id IN ({','.join('?' * len(ids))})")
        params.extend(ids)

    if location_id is not None:
        ids = db.descendants(conn, "locations", location_id)
        where.append(f"p.location_id IN ({','.join('?' * len(ids))})")
        params.extend(ids)

    if tag.strip():
        where.append(
            "p.id IN (SELECT pt.part_id FROM part_tags pt JOIN tags t ON t.id = pt.tag_id "
            "WHERE t.name = ? COLLATE NOCASE)"
        )
        params.append(tag.strip())

    if low_stock:
        where.append("p.min_quantity IS NOT NULL AND p.quantity <= p.min_quantity")

    if sort == "name":
        order = "p.name COLLATE NOCASE ASC"
    elif sort == "quantity":
        order = "p.quantity ASC, p.name COLLATE NOCASE ASC"
    elif sort == "updated":
        order = "p.updated_at DESC, p.id DESC"
    elif sort == "created":
        order = "p.created_at DESC, p.id DESC"
    elif sort == "location":
        order = "l.name COLLATE NOCASE ASC, p.name COLLATE NOCASE ASC"

    clause = (" WHERE " + " AND ".join(where)) if where else ""
    base = f"FROM parts p LEFT JOIN locations l ON l.id = p.location_id {joins} {clause}"

    total = conn.execute(f"SELECT COUNT(*) AS n {base}", params).fetchone()["n"]
    rows = conn.execute(
        f"SELECT p.* {base} ORDER BY {order} LIMIT ? OFFSET ?", params + [limit, offset]
    ).fetchall()

    ids = [r["id"] for r in rows]
    items = _serialise(
        rows, _specs_for(conn, ids), _tags_for(conn, ids),
        db.tree_paths(conn, "categories"), db.tree_paths(conn, "locations"),
        _images_for(conn, ids),
    )
    return {"total": total, "limit": limit, "offset": offset, "items": items}


@app.post("/api/parts", dependencies=[Depends(auth.require_auth)], status_code=201)
def create_part(body: PartIn, conn=Depends(db.get_db)):
    if body.category_id is not None:
        _require_node(conn, "categories", body.category_id, "Category")
    if body.location_id is not None:
        _require_node(conn, "locations", body.location_id, "Location")
    cur = conn.execute(
        """INSERT INTO parts(name, description, category_id, location_id, manufacturer, mpn,
                             quantity, unit, min_quantity, cost_each, datasheet_url,
                             product_url, notes)
           VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""",
        (
            body.name, body.description, body.category_id, body.location_id,
            body.manufacturer, body.mpn, body.quantity, body.unit,
            body.min_quantity, body.cost_each, body.datasheet_url,
            body.product_url, body.notes,
        ),
    )
    part_id = cur.lastrowid
    _write_specs(conn, part_id, body.specs)
    _write_tags(conn, part_id, body.tags)
    if body.quantity:
        _log_stock(conn, part_id, body.quantity, body.quantity, "initial stock")
    db.reindex_part(conn, part_id)
    return _fetch_part(conn, part_id)


@app.get("/api/parts/{part_id}", dependencies=[Depends(auth.require_auth)])
def get_part(part_id: int, conn=Depends(db.get_db)):
    return _fetch_part(conn, part_id)


@app.patch("/api/parts/{part_id}", dependencies=[Depends(auth.require_auth)])
def update_part(part_id: int, body: PartPatch, conn=Depends(db.get_db)):
    fields = body.model_dump(exclude_unset=True)
    fields.pop("specs", None)
    fields.pop("tags", None)
    reason = fields.pop("reason", "edited") or "edited"

    for col, value in fields.items():
        if value is None and col not in NULLABLE_COLUMNS:
            raise HTTPException(422, f"'{col}' cannot be null")
    if fields.get("category_id") is not None:
        _require_node(conn, "categories", fields["category_id"], "Category")
    if fields.get("location_id") is not None:
        _require_node(conn, "locations", fields["location_id"], "Location")

    # Setting quantity is a read-modify-write like any adjustment, and has to be
    # logged the same way — "every change is recorded" has to mean every change,
    # or the history is worse than useless.
    db.begin_immediate(conn)
    row = conn.execute("SELECT quantity FROM parts WHERE id = ?", (part_id,)).fetchone()
    if row is None:
        raise HTTPException(404, "Part not found")

    sets, params = [], []
    for col in PART_COLUMNS:
        if col in fields:
            sets.append(f"{col} = ?")
            params.append(fields[col])
    if sets:
        sets.append("updated_at = datetime('now')")
        conn.execute(f"UPDATE parts SET {', '.join(sets)} WHERE id = ?", params + [part_id])

    if "quantity" in fields and fields["quantity"] != row["quantity"]:
        new_qty = fields["quantity"]
        _log_stock(conn, part_id, new_qty - row["quantity"], new_qty, reason)

    if body.specs is not None:
        _write_specs(conn, part_id, body.specs)
    if body.tags is not None:
        _write_tags(conn, part_id, body.tags)
    db.reindex_part(conn, part_id)
    return _fetch_part(conn, part_id)


@app.delete("/api/parts/{part_id}", dependencies=[Depends(auth.require_auth)])
def delete_part(part_id: int, conn=Depends(db.get_db)):
    # Collect the image rows first: the cascade removes them from the database,
    # but the files on disk would be left behind as orphans.
    images = conn.execute(
        "SELECT token, mime FROM part_images WHERE part_id = ?", (part_id,)
    ).fetchall()
    cur = conn.execute("DELETE FROM parts WHERE id = ?", (part_id,))
    if cur.rowcount == 0:
        raise HTTPException(404, "Part not found")
    conn.execute("DELETE FROM parts_fts WHERE rowid = ?", (part_id,))
    # Commit before unlinking. If the order were reversed and the commit failed,
    # the rows would come back pointing at files that no longer exist — and a
    # missing photo is unrecoverable, where a stranded file is just clutter.
    conn.commit()
    db.delete_image_files(images)
    return {"deleted": part_id}


@app.post("/api/parts/{part_id}/adjust", dependencies=[Depends(auth.require_auth)])
def adjust_part(part_id: int, body: AdjustIn, conn=Depends(db.get_db)):
    # Read and write under one write lock. Without it, concurrent adjustments
    # read the same starting quantity and silently overwrite each other, leaving
    # the stock log disagreeing with the stock.
    db.begin_immediate(conn)
    row = conn.execute("SELECT quantity FROM parts WHERE id = ?", (part_id,)).fetchone()
    if row is None:
        raise HTTPException(404, "Part not found")

    before = row["quantity"]
    after = round(max(0.0, before + body.delta), 4)
    # Log what actually happened. Asking for -999 against a stock of 7 removes
    # 7, and recording -999 would make the ledger unsummable.
    applied = round(after - before, 4)

    conn.execute(
        "UPDATE parts SET quantity = ?, updated_at = datetime('now') WHERE id = ?",
        (after, part_id),
    )
    _log_stock(conn, part_id, applied, after, body.reason)
    return _fetch_part(conn, part_id)


@app.get("/api/parts/{part_id}/history", dependencies=[Depends(auth.require_auth)])
def part_history(part_id: int, conn=Depends(db.get_db), limit: int = Query(default=50, ge=1, le=500)):
    if conn.execute("SELECT 1 FROM parts WHERE id = ?", (part_id,)).fetchone() is None:
        raise HTTPException(404, "Part not found")
    rows = conn.execute(
        "SELECT * FROM stock_log WHERE part_id = ? ORDER BY id DESC LIMIT ?", (part_id, limit)
    ).fetchall()
    return {"items": [dict(r) for r in rows]}


# --- part images ---------------------------------------------------------
#
# Photos of the bag a part came out of, the print on a chip, a wiring note. The
# file is written to the volume beside the database; only metadata is stored in
# SQLite.


def _image_row(conn, part_id: int, token: str):
    row = conn.execute(
        "SELECT * FROM part_images WHERE part_id = ? AND token = ?", (part_id, token)
    ).fetchone()
    if row is None:
        raise HTTPException(404, "Image not found")
    return row


@app.get("/api/parts/{part_id}/images", dependencies=[Depends(auth.require_auth)])
def list_images(part_id: int, conn=Depends(db.get_db)):
    if conn.execute("SELECT 1 FROM parts WHERE id = ?", (part_id,)).fetchone() is None:
        raise HTTPException(404, "Part not found")
    return {"items": _images_for(conn, [part_id]).get(part_id, [])}


@app.post("/api/parts/{part_id}/images", dependencies=[Depends(auth.require_auth)],
          status_code=201)
def upload_image(
    part_id: int,
    file: UploadFile = File(...),
    caption: str = Form(default=""),
    conn=Depends(db.get_db),
):
    """Store one photo against a part.

    Deliberately a synchronous endpoint. FastAPI runs these in the threadpool,
    whereas an async one runs on the event loop — and this does blocking SQLite
    work, so under contention it would stall every other request for the length
    of SQLite's busy timeout rather than just queueing this one.
    """
    # Read and sniff before taking any lock: the bytes are the slow part, and
    # holding SQLite's write lock across them would serialise unrelated writes.
    data = b""
    while len(data) <= MAX_IMAGE_BYTES:
        chunk = file.file.read(256 * 1024)
        if not chunk:
            break
        data += chunk
    if len(data) > MAX_IMAGE_BYTES:
        raise HTTPException(413, f"Images must be {MAX_IMAGE_BYTES // (1024 * 1024)}MB or smaller")
    if not data:
        raise HTTPException(422, "Empty upload")

    # The declared content type is ignored; the bytes decide. That is what stops
    # an HTML or SVG payload being stored and later served back from this origin.
    mime = db.sniff_image(data[:32])
    if mime is None:
        raise HTTPException(415, "Not a supported image (JPEG, PNG, WEBP or GIF)")

    token = secrets.token_urlsafe(16)
    path = db.image_path(token, mime)
    with open(path, "wb") as fh:
        fh.write(data)

    try:
        # Everything that reads-then-writes happens under the write lock: the
        # existence check, the ceiling, and the position. Without it, overlapping
        # uploads all observe the same count and sail past the limit together.
        db.begin_immediate(conn)
        if conn.execute("SELECT 1 FROM parts WHERE id = ?", (part_id,)).fetchone() is None:
            raise HTTPException(404, "Part not found")
        existing = conn.execute(
            "SELECT COUNT(*) AS n FROM part_images WHERE part_id = ?", (part_id,)
        ).fetchone()["n"]
        if existing >= MAX_IMAGES_PER_PART:
            raise HTTPException(409, f"A part can hold at most {MAX_IMAGES_PER_PART} images")
        position = conn.execute(
            "SELECT COALESCE(MAX(position), -1) + 1 AS n FROM part_images WHERE part_id = ?",
            (part_id,),
        ).fetchone()["n"]
        conn.execute(
            "INSERT INTO part_images(part_id, token, mime, bytes, caption, position) "
            "VALUES (?,?,?,?,?,?)",
            (part_id, token, mime, len(data), caption.strip()[:300], position),
        )
        conn.execute("UPDATE parts SET updated_at = datetime('now') WHERE id = ?", (part_id,))
        # Commit here rather than leaving it to dependency cleanup, so the write
        # lock is released before the response is built rather than after.
        conn.commit()
    except Exception:
        # Don't leave a file behind for a row that never landed.
        db.delete_image_files([{"token": token, "mime": mime}])
        raise

    return {"token": token, "mime": mime, "bytes": len(data),
            "caption": caption.strip()[:300], "position": position}


@app.get("/api/parts/{part_id}/images/{token}", dependencies=[Depends(auth.require_auth)])
def get_image(part_id: int, token: str, conn=Depends(db.get_db)):
    row = _image_row(conn, part_id, token)
    path = db.image_path(row["token"], row["mime"])
    if not os.path.exists(path):
        raise HTTPException(404, "Image file is missing")
    return FileResponse(
        path,
        media_type=row["mime"],
        headers={
            # The token addresses exactly these bytes and never changes, so the
            # browser may keep it. Private, because it sits behind the login.
            "Cache-Control": "private, max-age=31536000, immutable",
            "Content-Disposition": "inline",
        },
    )


@app.patch("/api/parts/{part_id}/images/{token}", dependencies=[Depends(auth.require_auth)])
def update_image(part_id: int, token: str, body: ImagePatch, conn=Depends(db.get_db)):
    _image_row(conn, part_id, token)
    if body.caption is not None:
        conn.execute(
            "UPDATE part_images SET caption = ? WHERE part_id = ? AND token = ?",
            (body.caption, part_id, token),
        )
    if body.position is not None:
        conn.execute(
            "UPDATE part_images SET position = ? WHERE part_id = ? AND token = ?",
            (body.position, part_id, token),
        )
    return {"token": token}


@app.delete("/api/parts/{part_id}/images/{token}", dependencies=[Depends(auth.require_auth)])
def delete_image(part_id: int, token: str, conn=Depends(db.get_db)):
    row = _image_row(conn, part_id, token)
    conn.execute("DELETE FROM part_images WHERE part_id = ? AND token = ?", (part_id, token))
    conn.execute("UPDATE parts SET updated_at = datetime('now') WHERE id = ?", (part_id,))
    # Commit first: an orphaned file is recoverable, a row without its file is not.
    conn.commit()
    db.delete_image_files([row])
    return {"deleted": token}


# --- categories & locations -------------------------------------------------

def _node_payload(conn, table: str) -> list[dict]:
    paths = db.tree_paths(conn, table)
    rows = conn.execute(
        f"SELECT * FROM {table} ORDER BY sort_order, name COLLATE NOCASE"
    ).fetchall()
    column = "category_id" if table == "categories" else "location_id"
    direct = {
        r["k"]: r["n"]
        for r in conn.execute(
            f"SELECT {column} AS k, COUNT(*) AS n FROM parts GROUP BY k"
        ).fetchall()
        if r["k"] is not None
    }
    # Counts roll up: "Electronics" reports everything filed under its children
    # too, matching what clicking it as a filter actually returns.
    children: dict[int, list[int]] = {}
    for r in rows:
        children.setdefault(r["parent_id"], []).append(r["id"])

    def rollup(node_id: int, seen: set[int]) -> int:
        total = direct.get(node_id, 0)
        for child in children.get(node_id, []):
            if child not in seen:
                total += rollup(child, seen | {node_id})
        return total

    out = []
    for r in rows:
        d = dict(r)
        d["path"] = paths.get(r["id"], r["name"])
        d["part_count"] = rollup(r["id"], set())
        d["direct_count"] = direct.get(r["id"], 0)
        if table == "categories":
            try:
                d["spec_template"] = json.loads(r["spec_template"] or "[]")
            except (TypeError, ValueError):
                d["spec_template"] = []
        out.append(d)
    out.sort(key=lambda d: d["path"].lower())
    return out


def _template_json(specs: list[SpecIn]) -> str:
    return json.dumps([{"key": s.key} for s in specs])


@app.get("/api/categories", dependencies=[Depends(auth.require_auth)])
def list_categories(conn=Depends(db.get_db)):
    return {"items": _node_payload(conn, "categories")}


@app.post("/api/categories", dependencies=[Depends(auth.require_auth)], status_code=201)
def create_category(body: NodeIn, conn=Depends(db.get_db)):
    if body.parent_id is not None:
        _require_node(conn, "categories", body.parent_id, "Parent category")
    try:
        cur = conn.execute(
            "INSERT INTO categories(name, parent_id, unit, spec_template, sort_order) VALUES (?,?,?,?,?)",
            (body.name, body.parent_id, body.unit, _template_json(body.spec_template), body.sort_order),
        )
    except sqlite3.IntegrityError:
        raise HTTPException(409, "A category with that name already exists here")
    return {"id": cur.lastrowid}


@app.patch("/api/categories/{cat_id}", dependencies=[Depends(auth.require_auth)])
def update_category(cat_id: int, body: NodeIn, conn=Depends(db.get_db)):
    # Before the first read: the affected-parts scan and the reindex that
    # follows it have to see one consistent tree, or a concurrent write can slip
    # between them and leave the search index describing the old taxonomy.
    db.begin_immediate(conn)
    _require_node(conn, "categories", cat_id, "Category")
    _check_parent(conn, "categories", cat_id, body.parent_id, "category")
    affected = db.parts_under(conn, "categories", cat_id)
    try:
        conn.execute(
            "UPDATE categories SET name=?, parent_id=?, unit=?, spec_template=?, sort_order=? WHERE id=?",
            (body.name, body.parent_id, body.unit, _template_json(body.spec_template),
             body.sort_order, cat_id),
        )
    except sqlite3.IntegrityError:
        raise HTTPException(409, "A category with that name already exists here")
    # The category path is part of every affected part's search text.
    db.reindex_parts(conn, affected)
    return {"id": cat_id}


@app.delete("/api/categories/{cat_id}", dependencies=[Depends(auth.require_auth)])
def delete_category(cat_id: int, conn=Depends(db.get_db)):
    # Before the first read: the affected-parts scan and the reindex that
    # follows it have to see one consistent tree, or a concurrent write can slip
    # between them and leave the search index describing the old taxonomy.
    db.begin_immediate(conn)
    _require_node(conn, "categories", cat_id, "Category")
    # Capture the parts first: ON DELETE SET NULL means they survive as
    # uncategorised rather than vanish, but their indexed path must be rebuilt.
    affected = db.parts_under(conn, "categories", cat_id)
    conn.execute("DELETE FROM categories WHERE id = ?", (cat_id,))
    db.reindex_parts(conn, affected)
    return {"deleted": cat_id}


@app.get("/api/locations", dependencies=[Depends(auth.require_auth)])
def list_locations(conn=Depends(db.get_db)):
    return {"items": _node_payload(conn, "locations")}


@app.post("/api/locations", dependencies=[Depends(auth.require_auth)], status_code=201)
def create_location(body: NodeIn, conn=Depends(db.get_db)):
    if body.parent_id is not None:
        _require_node(conn, "locations", body.parent_id, "Parent location")
    try:
        cur = conn.execute(
            "INSERT INTO locations(name, parent_id, notes, sort_order) VALUES (?,?,?,?)",
            (body.name, body.parent_id, body.notes, body.sort_order),
        )
    except sqlite3.IntegrityError:
        raise HTTPException(409, "A location with that name already exists here")
    return {"id": cur.lastrowid}


@app.patch("/api/locations/{loc_id}", dependencies=[Depends(auth.require_auth)])
def update_location(loc_id: int, body: NodeIn, conn=Depends(db.get_db)):
    # Before the first read: the affected-parts scan and the reindex that
    # follows it have to see one consistent tree, or a concurrent write can slip
    # between them and leave the search index describing the old taxonomy.
    db.begin_immediate(conn)
    _require_node(conn, "locations", loc_id, "Location")
    _check_parent(conn, "locations", loc_id, body.parent_id, "location")
    affected = db.parts_under(conn, "locations", loc_id)
    try:
        conn.execute(
            "UPDATE locations SET name=?, parent_id=?, notes=?, sort_order=? WHERE id=?",
            (body.name, body.parent_id, body.notes, body.sort_order, loc_id),
        )
    except sqlite3.IntegrityError:
        raise HTTPException(409, "A location with that name already exists here")
    db.reindex_parts(conn, affected)
    return {"id": loc_id}


@app.delete("/api/locations/{loc_id}", dependencies=[Depends(auth.require_auth)])
def delete_location(loc_id: int, conn=Depends(db.get_db)):
    # Before the first read: the affected-parts scan and the reindex that
    # follows it have to see one consistent tree, or a concurrent write can slip
    # between them and leave the search index describing the old taxonomy.
    db.begin_immediate(conn)
    _require_node(conn, "locations", loc_id, "Location")
    affected = db.parts_under(conn, "locations", loc_id)
    conn.execute("DELETE FROM locations WHERE id = ?", (loc_id,))
    db.reindex_parts(conn, affected)
    return {"deleted": loc_id}


@app.get("/api/tags", dependencies=[Depends(auth.require_auth)])
def list_tags(conn=Depends(db.get_db)):
    rows = conn.execute(
        """SELECT t.name, COUNT(pt.part_id) AS n FROM tags t
           LEFT JOIN part_tags pt ON pt.tag_id = t.id
           GROUP BY t.id ORDER BY t.name COLLATE NOCASE"""
    ).fetchall()
    return {"items": [dict(r) for r in rows]}


@app.get("/api/stats", dependencies=[Depends(auth.require_auth)])
def stats(conn=Depends(db.get_db)):
    row = conn.execute(
        """SELECT COUNT(*) AS parts,
                  COALESCE(SUM(quantity * COALESCE(cost_each, 0)), 0) AS value,
                  SUM(CASE WHEN min_quantity IS NOT NULL AND quantity <= min_quantity
                           THEN 1 ELSE 0 END) AS low
           FROM parts"""
    ).fetchone()
    return {
        "parts": row["parts"],
        "low_stock": row["low"] or 0,
        "estimated_value": round(row["value"] or 0, 2),
    }


@app.get("/healthz")
def healthz():
    return {"ok": True}


# --- frontend ---------------------------------------------------------------

app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")

_index_cache: str | None = None


def _index_html() -> str:
    """index.html with a content hash appended to each asset URL.

    Cloudflare sits in front of this and will happily serve a cached app.js for
    hours. Versioned URLs mean a deploy changes the URL itself, so a stale copy
    can never be paired with a newer API.
    """
    global _index_cache
    if _index_cache is None:
        with open(os.path.join(STATIC_DIR, "index.html"), encoding="utf-8") as fh:
            html = fh.read()
        for asset in ("app.css", "app.js"):
            with open(os.path.join(STATIC_DIR, asset), "rb") as fh:
                digest = hashlib.sha256(fh.read()).hexdigest()[:10]
            html = html.replace(f"/static/{asset}", f"/static/{asset}?v={digest}")
        _index_cache = html
    return _index_cache


def _index_response(status_code: int = 200) -> HTMLResponse:
    return HTMLResponse(
        _index_html(), status_code=status_code, headers={"Cache-Control": "no-store"}
    )


@app.get("/")
def index():
    return _index_response()


@app.exception_handler(404)
def not_found(request: Request, exc):
    path = request.url.path
    if path.startswith("/api/") or path.startswith("/static/") or path == "/healthz":
        return JSONResponse({"detail": "Not found"}, status_code=404)
    return _index_response()