feat: add per-book reading journal notes

Claude-Session: https://claude.ai/code/session_01M9MLit5Ko3X4s7rzC5Kv7X

backend/notes.py

@@ -1,4 +1,77 @@
-"""Notes slice: a per-book reading journal."""
+"""Notes slice: a per-book reading journal. Ownership is always derived through books.user_id."""
-from flask import Blueprint
+from flask import Blueprint, g, jsonify
+
+from auth import login_required
+from db import query, query_one
+from validation import ApiError, json_body, text
 
 bp = Blueprint("notes", __name__, url_prefix="/api")
+
+MAX_BODY = 10_000
+NOTE_COLUMNS = "n.id, n.book_id, n.body, n.created_at, n.updated_at"
+
+
+def to_json(row: dict) -> dict:
+    return {
+        "id": row["id"],
+        "book_id": row["book_id"],
+        "body": row["body"],
+        "created_at": row["created_at"].isoformat(),
+        "updated_at": row["updated_at"].isoformat(),
+    }
+
+
+def _require_book(book_id: int) -> None:
+    if query_one("SELECT 1 FROM books WHERE id = %s AND user_id = %s", (book_id, g.user_id)) is None:
+        raise ApiError(404, "Book not found")
+
+
+@bp.get("/books/<int(max=2147483647):book_id>/notes")
+@login_required
+def list_notes(book_id: int):
+    _require_book(book_id)
+    rows = query(
+        f"SELECT {NOTE_COLUMNS} FROM notes n WHERE n.book_id = %s ORDER BY n.created_at DESC, n.id DESC",
+        (book_id,),
+    )
+    return jsonify([to_json(row) for row in rows])
+
+
+@bp.post("/books/<int(max=2147483647):book_id>/notes")
+@login_required
+def add_note(book_id: int):
+    _require_book(book_id)
+    body = text(json_body({"body"}), "body", MAX_BODY)
+    row = query_one(f"INSERT INTO notes AS n (book_id, body) VALUES (%s, %s) RETURNING {NOTE_COLUMNS}", (book_id, body))
+    return jsonify(to_json(row)), 201
+
+
+@bp.patch("/notes/<int(max=2147483647):note_id>")
+@login_required
+def update_note(note_id: int):
+    body = text(json_body({"body"}), "body", MAX_BODY)
+    row = query_one(
+        f"""
+        UPDATE notes n
+           SET body = %s, updated_at = now()
+          FROM books b
+         WHERE n.id = %s AND b.id = n.book_id AND b.user_id = %s
+        RETURNING {NOTE_COLUMNS}
+        """,
+        (body, note_id, g.user_id),
+    )
+    if row is None:
+        raise ApiError(404, "Note not found")
+    return jsonify(to_json(row))
+
+
+@bp.delete("/notes/<int(max=2147483647):note_id>")
+@login_required
+def delete_note(note_id: int):
+    row = query_one(
+        "DELETE FROM notes n USING books b WHERE n.id = %s AND b.id = n.book_id AND b.user_id = %s RETURNING n.id",
+        (note_id, g.user_id),
+    )
+    if row is None:
+        raise ApiError(404, "Note not found")
+    return "", 204

backend/tests/test_notes.py

@@ -0,0 +1,68 @@
+from tests.support import ApiTestCase, db_execute
+
+
+class NoteTests(ApiTestCase):
+    def setUp(self):
+        super().setUp()
+        self.register()
+        self.book = self.create_book()
+        self.notes_path = f"/api/books/{self.book['id']}/notes"
+
+    def add(self, body: str, client=None):
+        return self.call("POST", self.notes_path, {"body": body}, client)
+
+    def test_add_and_list_newest_first(self):
+        first = self.add("  Started reading  ")
+        self.assertEqual(first.status_code, 201)
+        self.assertEqual(first.get_json()["body"], "Started reading")
+        self.add("Chapter 3 was great")
+        bodies = [n["body"] for n in self.call("GET", self.notes_path).get_json()]
+        self.assertEqual(bodies, ["Chapter 3 was great", "Started reading"])
+
+    def test_edit_updates_body_and_timestamp(self):
+        note = self.add("Draft").get_json()
+        response = self.call("PATCH", f"/api/notes/{note['id']}", {"body": "Final"})
+        self.assertEqual(response.status_code, 200)
+        edited = response.get_json()
+        self.assertEqual(edited["body"], "Final")
+        self.assertGreater(edited["updated_at"], edited["created_at"])
+
+    def test_delete(self):
+        note = self.add("Temporary").get_json()
+        self.assertEqual(self.call("DELETE", f"/api/notes/{note['id']}").status_code, 204)
+        self.assertEqual(self.call("GET", self.notes_path).get_json(), [])
+        self.assertEqual(self.call("DELETE", f"/api/notes/{note['id']}").status_code, 404)
+
+    def test_validation(self):
+        for body, status in (("   ", 400), ("x" * 10_001, 400), ("a\x00b", 400)):
+            with self.subTest(length=len(body)):
+                self.assertEqual(self.add(body).status_code, status)
+        self.assertEqual(self.call("POST", self.notes_path, {"body": "ok", "book_id": 2}).status_code, 400)
+
+    def test_missing_book_and_out_of_range_ids_are_404(self):
+        self.assertEqual(self.call("GET", "/api/books/999/notes").status_code, 404)
+        self.assertEqual(self.call("PATCH", "/api/notes/99999999999", {"body": "x"}).status_code, 404)
+
+    def test_deleting_book_deletes_its_notes(self):
+        self.add("Will be gone")
+        self.call("DELETE", f"/api/books/{self.book['id']}")
+        self.assertEqual(db_execute("SELECT count(*) FROM notes")[0][0], 0)
+
+
+class NoteIsolationTests(ApiTestCase):
+    def test_other_user_cannot_read_add_edit_or_delete_notes(self):
+        self.register("alice")
+        book = self.create_book()
+        note = self.call("POST", f"/api/books/{book['id']}/notes", {"body": "Private"}).get_json()
+        bob = self.other_user("bob")
+        attempts = (
+            ("GET", f"/api/books/{book['id']}/notes", None),
+            ("POST", f"/api/books/{book['id']}/notes", {"body": "Intrusion"}),
+            ("PATCH", f"/api/notes/{note['id']}", {"body": "Hacked"}),
+            ("DELETE", f"/api/notes/{note['id']}", None),
+        )
+        for method, path, payload in attempts:
+            with self.subTest(method=method, path=path):
+                self.assertEqual(self.call(method, path, payload, client=bob).status_code, 404)
+        bodies = [n["body"] for n in self.call("GET", f"/api/books/{book['id']}/notes").get_json()]
+        self.assertEqual(bodies, ["Private"])