path: root/packages/meshbay-hub/src/meshbay_hub/api/moderation.py

"""
MeshBay Hub — moderation endpoints (docs/MESHBAY_DESIGN.md §7.5).

Reporting:
  POST /v1/reports — a member of a public group reports a file they saw there.

  Who may: a person's account (never a node's token), old enough
  (`reports.min_account_age_hours`), an active member of that public group,
  within a daily allowance (`reports.daily_per_account`) as well as the per-address
  rate limit. One report per account per hash.

  What it leads to: once `reports.review_threshold` distinct accounts have
  reported a hash, it is queued for an administrator (`content_reviews`), who is
  notified and blocks or dismisses it. With `reports.auto_block` on, it is blocked
  at once instead — the instance's choice, off by default, because a handful of
  accounts made for the purpose would then be enough to take a file down.

  The flow only runs while the hub brokers public content: with public groups
  switched off there is nothing here to report.

Admin endpoints:
  GET  /v1/admin/reports                 — hashes waiting for a decision
  POST /v1/admin/reports/{hash}/block    — block it, and tell the nodes
  POST /v1/admin/reports/{hash}/dismiss  — close it without blocking
  GET  /v1/admin/blocklist               — list blocked hashes
  POST /v1/admin/blocklist               — manually add a hash
  DELETE /v1/admin/blocklist/{hash}      — remove a hash

Node integration:
  GET  /v1/blocklist?after=<hash>  — the list, paged, for a node's own token
  WebSocket `blocklist_update`      — additions and removals, pushed to nodes
                                      hosting a public group
"""

import logging
from collections import Counter
from datetime import UTC, datetime, timedelta
from typing import Literal

from fastapi import APIRouter, Depends, HTTPException, Query, Request
from pydantic import BaseModel, Field
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from meshbay_hub import hub_settings
from meshbay_hub.api.deps import (
    require_admin,
    require_moderator,
    require_node_scope,
    require_user_scope,
    user_is_admin,
)
from meshbay_hub.api.middleware import limiter
from meshbay_hub.api.netutil import client_ip
from meshbay_hub.api.revocation import broadcast_blocklist_update
from meshbay_hub.db.engine import get_db
from meshbay_hub.db.models import (
    ContentBlocklist,
    ContentReport,
    ContentReview,
    Group,
    GroupMember,
    User,
)

log = logging.getLogger(__name__)

router = APIRouter(tags=["moderation"])

# One answer for every reason a report is not accepted from this account for this
# group, so the endpoint does not tell anyone which groups exist or who is in them.
_NOT_YOURS = "You can report a file only in a public group you are a member of."


def _is_hash(value: str) -> bool:
    return len(value) == 64 and all(c in "0123456789abcdef" for c in value)


# ── Models ────────────────────────────────────────────────────────────────────

class ReportRequest(BaseModel):
    content_hash: str = Field(max_length=64)   # blake3 hex (64 chars)
    group_id:     str = Field(max_length=36)
    reason:       Literal["illegal", "spam", "copyright", "other"] = "illegal"
    detail:       str | None = Field(default=None, max_length=256)


class BlocklistAddRequest(BaseModel):
    content_hash: str
    reason:       str


# ── Reporting ─────────────────────────────────────────────────────────────────

@router.post("/v1/reports", status_code=201)
@limiter.limit("10/hour")
async def report_content(
    body: ReportRequest,
    request: Request,
    current_user: User = Depends(require_user_scope),
    db: AsyncSession = Depends(get_db),
):
    """
    Report a file of a public group, as a member of that group.

    Every bound here answers what a report costs someone else: a file taken out
    of a group everyone else uses, and an administrator's time. So a report takes
    a person's account (a node's token is refused), one that has existed for a
    while, membership of the public group the file was seen in, and a daily
    allowance per account besides the rate limit per address — an address is one
    of thousands a subscriber holds. It never blocks anything by itself unless the
    instance chose automatic blocking: past the threshold, an administrator
    decides.
    """
    if not await hub_settings.public_groups_allowed(db):
        raise HTTPException(
            status_code=403,
            detail="This hub does not broker public content, so there is nothing to report here.")
    if not _is_hash(body.content_hash):
        raise HTTPException(status_code=422, detail="content_hash must be 64 hex chars (blake3)")

    limits = await hub_settings.report_limits(db)
    now = datetime.now(UTC)

    created = current_user.created_at
    if created is not None and created.tzinfo is None:
        created = created.replace(tzinfo=UTC)
    if created is not None and \
            now - created < timedelta(hours=limits["min_account_age_hours"]):
        raise HTTPException(status_code=403,
                            detail="This account is too new to report content yet.")

    group = await db.get(Group, body.group_id)
    member = await db.scalar(select(GroupMember.user_id).where(
        GroupMember.group_id == body.group_id,
        GroupMember.user_id == current_user.id))
    if group is None or group.visibility != "public" or group.status != "active" \
            or member is None:
        raise HTTPException(status_code=403, detail=_NOT_YOURS)

    today = await db.scalar(select(func.count(ContentReport.id)).where(
        ContentReport.reporter_id == current_user.id,
        ContentReport.reported_at > now - timedelta(days=1))) or 0
    if today >= limits["daily_per_account"]:
        raise HTTPException(status_code=429,
                            detail="You have reached today's number of reports.")

    # One vote per account per hash — a single reporter must not be able to walk
    # the threshold up on their own by posting repeatedly.
    already = await db.scalar(
        select(ContentReport.id).where(
            ContentReport.content_hash == body.content_hash,
            ContentReport.reporter_id == current_user.id))
    if already:
        return {"status": "already_reported"}

    db.add(ContentReport(
        content_hash=body.content_hash,
        reporter_id=current_user.id,
        group_id=body.group_id,
        reason=body.reason,
        detail=body.detail,
        ip_address=client_ip(request),
    ))
    await db.flush()

    distinct_reporters = await db.scalar(
        select(func.count(func.distinct(ContentReport.reporter_id)))
        .where(ContentReport.content_hash == body.content_hash)) or 0

    blocked_now = False
    if distinct_reporters >= limits["review_threshold"] \
            and await db.get(ContentBlocklist, body.content_hash) is None:
        review = await db.get(ContentReview, body.content_hash)
        if review is not None and review.status == "dismissed":
            pass   # an administrator's decision stands; more reports do not reopen it
        elif limits["auto_block"]:
            db.add(ContentBlocklist(content_hash=body.content_hash,
                                    reason=f"auto:{body.reason}", added_by="auto"))
            if review is None:
                db.add(ContentReview(content_hash=body.content_hash, status="blocked",
                                     decided_at=now, decided_by="auto"))
            else:
                review.status, review.decided_at, review.decided_by = "blocked", now, "auto"
            blocked_now = True
            log.warning("Content auto-blocked after %d distinct reporters: %s",
                        distinct_reporters, body.content_hash[:16])
        elif review is None:
            db.add(ContentReview(content_hash=body.content_hash, status="pending"))
            await _notify_admins(db, body.content_hash)
            log.warning("Content queued for review after %d distinct reporters: %s",
                        distinct_reporters, body.content_hash[:16])
    await db.commit()
    if blocked_now:
        await broadcast_blocklist_update(db, add=[body.content_hash])
    # The same answer whatever happened next: a reporter is not told how close a
    # file is to review, which is a count to aim at.
    return {"status": "logged"}


async def _notify_admins(db: AsyncSession, content_hash: str) -> None:
    from meshbay_hub.api.notifications import create_notification
    admins = [u for u in (await db.execute(select(User).where(
        User.status == "active"))).scalars().all() if user_is_admin(u)]
    for admin in admins:
        await create_notification(
            db, admin.id, "content_review",
            "Reported content is waiting for a decision",
            detail=content_hash[:16], link="#/admin", aggregate=False)


@router.get("/v1/blocklist")
async def get_blocklist(
    current_node: User = Depends(require_node_scope),
    db: AsyncSession = Depends(get_db),
    after: str = Query(default="", max_length=64),
    limit: int = Query(default=10000, ge=1, le=10000),
):
    """
    The content blocklist, a page at a time, for a node hosting a public group.

    Hashes only, never the reasons. Ordered by hash so `after` (the last hash of
    the previous page) is a stable cursor: a list longer than one page used to
    be cut at 10 000 with no way to ask for the rest, and the node applying it
    silently served everything past the cut. A node's own token, because this
    is what a node fetches on its own behalf and nothing else asks for it.
    """
    result = await db.execute(
        select(ContentBlocklist.content_hash)
        .where(ContentBlocklist.content_hash > after)
        .order_by(ContentBlocklist.content_hash)
        .limit(limit)
    )
    hashes = [row[0] for row in result.fetchall()]
    return {"hashes": hashes,
            "next": hashes[-1] if len(hashes) == limit else None}


# ── Admin endpoints ───────────────────────────────────────────────────────────

@router.get("/v1/admin/blocklist")
async def admin_list_blocklist(
    current_user: User = Depends(require_admin),
    db: AsyncSession = Depends(get_db),
    limit: int = 500,
):
    result = await db.execute(
        select(ContentBlocklist)
        .order_by(ContentBlocklist.added_at.desc())
        .limit(limit)
    )
    entries = result.scalars().all()
    return {
        "entries": [
            {
                "hash":     e.content_hash,
                "reason":   e.reason,
                "added_at": e.added_at.isoformat(),
                "added_by": e.added_by,
            }
            for e in entries
        ]
    }


@router.post("/v1/admin/blocklist", status_code=201)
async def admin_add_blocklist(
    body: BlocklistAddRequest,
    current_user: User = Depends(require_admin),
    db: AsyncSession = Depends(get_db),
):
    if not _is_hash(body.content_hash):
        raise HTTPException(status_code=422, detail="content_hash must be 64 hex chars (blake3)")
    existing = await db.get(ContentBlocklist, body.content_hash)
    if existing:
        raise HTTPException(status_code=409, detail="Hash already blocked")

    db.add(ContentBlocklist(
        content_hash=body.content_hash,
        reason=body.reason[:64],
        added_by=current_user.username,
    ))
    await db.commit()
    await broadcast_blocklist_update(db, add=[body.content_hash])
    return {"status": "blocked", "hash": body.content_hash}


@router.delete("/v1/admin/blocklist/{content_hash}", status_code=200)
async def admin_remove_blocklist(
    content_hash: str,
    current_user: User = Depends(require_admin),
    db: AsyncSession = Depends(get_db),
):
    entry = await db.get(ContentBlocklist, content_hash)
    if not entry:
        raise HTTPException(status_code=404, detail="Hash not in blocklist")
    await db.delete(entry)
    await db.commit()
    await broadcast_blocklist_update(db, remove=[content_hash])
    return {"status": "unblocked", "hash": content_hash}


# ── Review queue ──────────────────────────────────────────────────────────────

@router.get("/v1/admin/reports")
async def admin_list_reports(
    current_user: User = Depends(require_moderator),
    db: AsyncSession = Depends(get_db),
    limit: int = Query(default=100, ge=1, le=500),
):
    """Hashes waiting for a decision, oldest first, with what was said about them."""
    reviews = (await db.execute(
        select(ContentReview).where(ContentReview.status == "pending")
        .order_by(ContentReview.opened_at).limit(limit))).scalars().all()
    out = []
    for r in reviews:
        reports = (await db.execute(select(ContentReport).where(
            ContentReport.content_hash == r.content_hash))).scalars().all()
        group_ids = sorted({x.group_id for x in reports if x.group_id})
        names = dict((await db.execute(select(Group.id, Group.name).where(
            Group.id.in_(group_ids)))).all()) if group_ids else {}
        out.append({
            "hash": r.content_hash,
            "opened_at": r.opened_at.isoformat(),
            "reporters": len({x.reporter_id for x in reports}),
            "reasons": dict(Counter(x.reason for x in reports)),
            "details": [x.detail for x in reports if x.detail][:10],
            "groups": [{"id": g, "name": names.get(g, "")} for g in group_ids],
        })
    return {"reports": out}


async def _decide(db: AsyncSession, content_hash: str, status: str, by: str) -> ContentReview:
    review = await db.get(ContentReview, content_hash)
    if review is None or review.status != "pending":
        raise HTTPException(status_code=404, detail="Nothing waiting for this hash")
    review.status, review.decided_at, review.decided_by = status, datetime.now(UTC), by
    return review


@router.post("/v1/admin/reports/{content_hash}/block")
async def admin_block_reported(
    content_hash: str,
    current_user: User = Depends(require_admin),
    db: AsyncSession = Depends(get_db),
):
    await _decide(db, content_hash, "blocked", current_user.username)
    reasons = Counter((await db.execute(select(ContentReport.reason).where(
        ContentReport.content_hash == content_hash))).scalars().all())
    if await db.get(ContentBlocklist, content_hash) is None:
        db.add(ContentBlocklist(
            content_hash=content_hash,
            reason=f"reported:{reasons.most_common(1)[0][0] if reasons else 'other'}",
            added_by=current_user.username))
    await db.commit()
    await broadcast_blocklist_update(db, add=[content_hash])
    return {"status": "blocked", "hash": content_hash}


@router.post("/v1/admin/reports/{content_hash}/dismiss")
async def admin_dismiss_reported(
    content_hash: str,
    current_user: User = Depends(require_admin),
    db: AsyncSession = Depends(get_db),
):
    await _decide(db, content_hash, "dismissed", current_user.username)
    await db.commit()
    return {"status": "dismissed", "hash": content_hash}