Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 12 additions & 3 deletions shelfmark/api/websocket.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@

from flask_socketio import SocketIO, join_room, leave_room

from shelfmark.core.viewer_redaction import viewer_download_id

if TYPE_CHECKING:
from collections.abc import Callable

Expand Down Expand Up @@ -167,7 +169,12 @@ def _broadcast_status_update_to_room(self, room: str) -> None:
logger.exception("Failed to send status update for room %s", room)

def broadcast_download_progress(
self, book_id: str, progress: float, status: str, user_id: int | None = None
self,
book_id: str,
progress: float,
status: str,
user_id: int | None = None,
request_id: int | None = None,
) -> None:
"""Broadcast download progress update for a specific book."""
socketio = self._get_socketio()
Expand All @@ -178,12 +185,14 @@ def broadcast_download_progress(
data = {"book_id": book_id, "progress": progress, "status": status}
# Admins always see all progress
socketio.emit("download_progress", data, to="admins")
# If task belongs to a specific user, send to their room too
# If task belongs to a specific user, send to their room too, under the id
# their status uses: an opaque one for a request-linked download (#1418).
if user_id is not None:
room = f"user_{user_id}"
with self._rooms_lock:
if room in self._user_rooms:
socketio.emit("download_progress", data, to=room)
user_data = {**data, "book_id": viewer_download_id(book_id, request_id)}
socketio.emit("download_progress", user_data, to=room)
logger.debug("Broadcasted progress for book %s: %s%%", book_id, progress)
except Exception:
logger.exception("Error broadcasting download progress")
Expand Down
130 changes: 128 additions & 2 deletions shelfmark/core/activity_routes.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,15 @@
populate_request_usernames,
)
from shelfmark.core.request_validation import RequestStatus
from shelfmark.core.viewer_redaction import (
is_public_download_id,
match_public_download_id,
public_download_id,
redact_download_payload,
redact_request_row,
redact_status,
viewer_download_id,
)

if TYPE_CHECKING:
from collections.abc import Callable
Expand Down Expand Up @@ -501,6 +510,77 @@ def _request_history_entry(
}


def _request_linked_task_ids(
actor: _ActorContext,
*,
queue_status: Callable[..., dict[str, dict[str, Any]]],
download_history_service: DownloadHistoryService,
) -> list[str]:
"""Task ids of the actor's request-linked downloads, which they see by opaque id."""
if actor.db_user_id is None:
return []
live_task_ids = [
task_id
for bucket in queue_status(user_id=actor.owner_scope).values()
if isinstance(bucket, dict)
for task_id, payload in bucket.items()
if isinstance(payload, dict) and payload.get("request_id") is not None
]
return [
*live_task_ids,
*download_history_service.list_request_task_ids(user_id=actor.db_user_id),
]


def _viewer_dismissed_entries(
entries: list[dict[str, str]],
*,
request_ids_by_task: dict[str, object],
) -> list[dict[str, str]]:
"""Rewrite a non-admin's dismissed download keys to the ids their snapshot uses.

Downloads outside the recent rows and the live queue never reach the snapshot,
so their entries are dropped rather than looked up one by one.
"""
viewer_entries: list[dict[str, str]] = []
for entry in entries:
if entry["item_type"] != "download":
viewer_entries.append(entry)
continue
task_id = _parse_item_key(entry["item_key"], "download")
if task_id is None or task_id not in request_ids_by_task:
continue
viewer_id = viewer_download_id(task_id, request_ids_by_task[task_id])
viewer_entries.append({"item_type": "download", "item_key": f"download:{viewer_id}"})
return viewer_entries


def _redact_history_entry(entry: dict[str, Any]) -> dict[str, Any]:
"""Return a history entry with the same redactions as the non-admin snapshot."""
redacted = dict(entry)
snapshot = entry.get("snapshot")
source_id = entry.get("source_id")
if entry.get("item_type") == "download":
if isinstance(source_id, str) and source_id:
viewer_id = viewer_download_id(source_id, entry.get("request_id"))
item_key = f"download:{viewer_id}"
redacted.update(id=item_key, item_key=item_key, source_id=viewer_id)
if isinstance(snapshot, dict) and isinstance(snapshot.get("download"), dict):
redacted["snapshot"] = {
**snapshot,
"download": redact_download_payload(snapshot["download"], task_id=source_id),
}
return redacted

# A request's source_id names the download it was fulfilled with, which the
# requester sees by its opaque id.
if isinstance(source_id, str) and source_id:
redacted["source_id"] = public_download_id(source_id)
if isinstance(snapshot, dict) and isinstance(snapshot.get("request"), dict):
redacted["snapshot"] = {**snapshot, "request": redact_request_row(snapshot["request"])}
return redacted


def register_activity_routes(
app: Flask,
user_db: UserDB,
Expand Down Expand Up @@ -577,6 +657,25 @@ def api_activity_snapshot() -> Response | tuple[Response, int]:
continue
visible_request_rows.append(row)

if not actor.is_admin:
# Redacted only now: the delivery-state sync above matches on real task ids.
request_ids_by_task: dict[str, object] = {
task_id: payload.get("request_id")
for bucket in live_queue.values()
if isinstance(bucket, dict)
for task_id, payload in bucket.items()
if isinstance(payload, dict)
}
request_ids_by_task.update(
(str(row.get("task_id") or "").strip(), row.get("request_id")) for row in db_rows
)
dismissed_entries = _viewer_dismissed_entries(
dismissed_entries,
request_ids_by_task=request_ids_by_task,
)
status = redact_status(status)
visible_request_rows = [redact_request_row(row) for row in visible_request_rows]

return jsonify(
{
"status": status,
Expand Down Expand Up @@ -622,6 +721,21 @@ def api_activity_dismiss() -> Response | tuple[Response, int]:
item_key=item_key,
)

# The id as this viewer knows it, which is what the response echoes.
viewer_task_id = task_id
if not actor.is_admin and is_public_download_id(task_id):
task_id = (
match_public_download_id(
task_id,
_request_linked_task_ids(
actor,
queue_status=queue_status,
download_history_service=download_history_service,
),
)
or task_id
)

existing = download_history_service.get_by_task_id(task_id)
if existing is None:
return _activity_error_response(
Expand Down Expand Up @@ -671,7 +785,7 @@ def api_activity_dismiss() -> Response | tuple[Response, int]:
)
dismissal_item = {
"item_type": "download",
"item_key": f"download:{task_id}",
"item_key": f"download:{viewer_task_id}",
}

elif item_type == "request":
Expand Down Expand Up @@ -796,6 +910,7 @@ def api_activity_dismiss_many() -> Response | tuple[Response, int]:
dismissal_items: list[dict[str, str]] = []
missing_item_keys: list[str] = []
live_queue_index: dict[str, tuple[str, dict[str, Any]]] | None = None
request_linked_task_ids: list[str] | None = None

for item in items:
if not isinstance(item, dict):
Expand Down Expand Up @@ -824,9 +939,18 @@ def api_activity_dismiss_many() -> Response | tuple[Response, int]:
item_key=item_key,
item_count=len(items),
)
viewer_task_id = task_id
if not actor.is_admin and is_public_download_id(task_id):
if request_linked_task_ids is None:
request_linked_task_ids = _request_linked_task_ids(
actor,
queue_status=queue_status,
download_history_service=download_history_service,
)
task_id = match_public_download_id(task_id, request_linked_task_ids) or task_id
existing = download_history_service.get_by_task_id(task_id)
if existing is None:
missing_item_keys.append(f"download:{task_id}")
missing_item_keys.append(f"download:{viewer_task_id}")
continue
if live_queue_index is None:
live_queue_index = _build_queue_index(queue_status(user_id=actor.owner_scope))
Expand Down Expand Up @@ -1060,6 +1184,8 @@ def api_activity_history() -> Response | tuple[Response, int]:
msg = f"Unknown activity history item_type: {item_type}"
raise RuntimeError(msg)

if not actor.is_admin:
payload = [_redact_history_entry(entry) for entry in payload]
return jsonify(payload)

@app.route("/api/activity/history", methods=["DELETE"])
Expand Down
17 changes: 17 additions & 0 deletions shelfmark/core/download_history_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -407,6 +407,23 @@ def get_by_task_id(self, task_id: str) -> dict[str, Any] | None:
finally:
conn.close()

def list_request_task_ids(self, *, user_id: int) -> list[str]:
"""Return the task ids of every request-linked download owned by ``user_id``.

A non-admin addresses these downloads by an opaque id, so routes resolve it
against this set, which also confines the lookup to the user's own downloads.
"""
normalized_user_id = normalize_optional_positive_int(user_id, "user_id")
conn = self._connect()
try:
rows = conn.execute(
"SELECT task_id FROM download_history WHERE user_id = ? AND request_id IS NOT NULL",
(normalized_user_id,),
).fetchall()
return [str(row["task_id"]) for row in rows]
finally:
conn.close()

def list_recent(
self,
*,
Expand Down
6 changes: 6 additions & 0 deletions shelfmark/core/request_routes.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
fulfil_request,
reject_request,
)
from shelfmark.core.viewer_redaction import redact_request_row

if TYPE_CHECKING:
from collections.abc import Callable
Expand Down Expand Up @@ -847,6 +848,9 @@ def api_list_requests() -> ResponseReturnValue:
)
except ValueError as exc:
return jsonify({"error": str(exc)}), 400
if not session.get("is_admin", False):
# A fulfilled request carries the release an admin picked (#1418).
rows = [redact_request_row(row) for row in rows]
return jsonify(rows)

@app.route("/api/requests/<int:request_id>", methods=["DELETE"])
Expand Down Expand Up @@ -901,6 +905,8 @@ def api_cancel_request(request_id: int) -> ResponseReturnValue:
room="admins",
)

if not session.get("is_admin", False):
updated = redact_request_row(updated)
return jsonify(updated)

@app.route("/api/admin/requests", methods=["GET"])
Expand Down
Loading
Loading