# Report and Abuse System Low-Level Design: Triage, Moderator Queue, and Auto-Enforcement

Source: https://www.techinterview.org/post/3233468391/lld-report-abuse/
Updated: 2026-04-17 · techinterview.org

A content reporting and abuse detection system allows users to flag harmful content — spam, harassment, hate speech, illegal material — and routes reports to human moderators or automated enforcement. Core challenges: preventing report spam (false flags), triaging high-severity reports, escalating urgent content (CSAM, imminent violence), tracking enforcement actions, and providing feedback to reporters.

## Core Data Model


```
CREATE TYPE report_category AS ENUM (
    'spam','harassment','hate_speech','misinformation',
    'violence','nudity','illegal_content','impersonation','other'
);
CREATE TYPE report_status AS ENUM (
    'pending','under_review','action_taken','dismissed','escalated'
);

CREATE TABLE Report (
    report_id       UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    reporter_id     UUID NOT NULL,
    content_type    TEXT NOT NULL,     -- 'post', 'comment', 'user', 'message'
    content_id      TEXT NOT NULL,     -- ID of the reported content
    category        report_category NOT NULL,
    description     TEXT,              -- reporter's explanation (optional)
    status          report_status NOT NULL DEFAULT 'pending',
    priority        SMALLINT NOT NULL DEFAULT 2,  -- 1=urgent, 2=high, 3=normal, 4=low
    assigned_to     UUID,              -- moderator user_id
    resolution      TEXT,
    reported_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    resolved_at     TIMESTAMPTZ
);
CREATE INDEX idx_report_status_priority ON Report (status, priority, reported_at) WHERE status = 'pending';
CREATE INDEX idx_report_content ON Report (content_type, content_id);

-- Enforcement actions taken on content/user
CREATE TABLE EnforcementAction (
    action_id       UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    report_id       UUID REFERENCES Report(report_id),
    target_type     TEXT NOT NULL,    -- 'content' or 'user'
    target_id       TEXT NOT NULL,
    action_type     TEXT NOT NULL,    -- 'remove','warn','restrict','suspend','ban'
    actor_id        UUID NOT NULL,    -- moderator or 'system'
    reason          TEXT,
    duration_hours  INT,              -- for temporary restrictions
    expires_at      TIMESTAMPTZ,
    applied_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    appealed        BOOLEAN NOT NULL DEFAULT FALSE
);
```


## Submitting a Report


```
from uuid import uuid4
from datetime import datetime, timezone
import psycopg2

# Auto-escalate categories that require immediate review
URGENT_CATEGORIES = {'illegal_content', 'violence'}
# Content that was already reported N times gets priority boost
HIGH_VOLUME_THRESHOLD = 10

def submit_report(conn, reporter_id: str, content_type: str, content_id: str,
                   category: str, description: str = "") -> dict:
    """
    Submit a content report. Determines priority based on category and
    prior report volume for the same content. Deduplicates per reporter.
    """
    # Dedup: one report per (reporter, content) per 24 hours
    with conn.cursor() as cur:
        cur.execute("""
            SELECT 1 FROM Report
            WHERE reporter_id=%s AND content_type=%s AND content_id=%s
              AND reported_at > NOW() - interval '24 hours'
        """, (reporter_id, content_type, content_id))
        if cur.fetchone():
            raise ValueError("You have already reported this content recently")

    # Count existing reports for this content (volume signal)
    with conn.cursor() as cur:
        cur.execute(
            "SELECT COUNT(*) FROM Report WHERE content_type=%s AND content_id=%s",
            (content_type, content_id)
        )
        report_count = cur.fetchone()[0]

    # Determine priority
    if category in URGENT_CATEGORIES:
        priority = 1  # urgent
    elif report_count >= HIGH_VOLUME_THRESHOLD:
        priority = 2  # high — many people are reporting this
    else:
        priority = 3  # normal

    report_id = str(uuid4())
    with conn.cursor() as cur:
        cur.execute("""
            INSERT INTO Report
            (report_id, reporter_id, content_type, content_id, category, description, priority)
            VALUES (%s,%s,%s,%s,%s,%s,%s)
        """, (report_id, reporter_id, content_type, content_id, category, description, priority))
    conn.commit()

    # Trigger auto-actions for urgent reports
    if priority == 1:
        auto_restrict_pending_review(conn, content_type, content_id, report_id)

    return {"report_id": report_id, "priority": priority}

def auto_restrict_pending_review(conn, content_type: str, content_id: str, report_id: str):
    """
    For urgent reports: immediately hide content while awaiting human review.
    This limits damage before moderation completes.
    """
    with conn.cursor() as cur:
        cur.execute("""
            INSERT INTO EnforcementAction
            (report_id, target_type, target_id, action_type, actor_id, reason)
            VALUES (%s, %s, %s, 'restrict', 'system', 'Pending urgent review')
        """, (report_id, content_type, content_id))
    conn.commit()
    hide_content(content_type, content_id)
```


## Moderator Review Queue


```
def get_review_queue(conn, moderator_id: str, limit: int = 20) -> list[dict]:
    """
    Fetch the highest-priority pending reports.
    Claims them for this moderator using SELECT ... FOR UPDATE SKIP LOCKED.
    """
    with conn.cursor() as cur:
        cur.execute("""
            UPDATE Report
            SET assigned_to = %s
            WHERE report_id IN (
                SELECT report_id FROM Report
                WHERE status = 'pending'
                  AND (assigned_to IS NULL OR assigned_to = %s)
                ORDER BY priority ASC, reported_at ASC
                LIMIT %s
                FOR UPDATE SKIP LOCKED
            )
            RETURNING report_id, content_type, content_id, category, description,
                      priority, reported_at
        """, (moderator_id, moderator_id, limit))
        cols = [d[0] for d in cur.description]
        reports = [dict(zip(cols, row)) for row in cur.fetchall()]
    conn.commit()
    return reports

def resolve_report(conn, report_id: str, moderator_id: str,
                    decision: str, action_type: str | None = None) -> dict:
    """
    Moderator resolves a report with one of: action_taken, dismissed.
    Optionally applies an enforcement action.
    """
    VALID_DECISIONS = {'action_taken', 'dismissed'}
    if decision not in VALID_DECISIONS:
        raise ValueError(f"Invalid decision: {decision}")

    with conn.cursor() as cur:
        cur.execute(
            "SELECT content_type, content_id FROM Report WHERE report_id=%s AND assigned_to=%s",
            (report_id, moderator_id)
        )
        row = cur.fetchone()
    if not row:
        raise ValueError("Report not found or not assigned to this moderator")

    content_type, content_id = row

    with conn.cursor() as cur:
        cur.execute("""
            UPDATE Report SET status=%s, resolved_at=NOW() WHERE report_id=%s
        """, (decision, report_id))

        if action_type:
            cur.execute("""
                INSERT INTO EnforcementAction
                (report_id, target_type, target_id, action_type, actor_id)
                VALUES (%s,%s,%s,%s,%s)
            """, (report_id, content_type, content_id, action_type, moderator_id))
    conn.commit()

    # Notify reporter of outcome
    notify_reporter_of_outcome(report_id, decision)
    return {"report_id": report_id, "decision": decision}
```


## Key Interview Points

- **Reporter reputation scoring**: Not all reports are equal. Maintain a reporter_accuracy score: reports_upheld / reports_resolved. High-accuracy reporters (0.8+) have their reports auto-promoted to priority 2. Low-accuracy reporters (<0.2) are throttled — limit to 5 reports/day. This counters coordinated false-flag attacks (harassment campaigns where a target is mass-reported by a group).

- **Moderator queue assignment**: Use SKIP LOCKED so multiple moderators work the queue without contention. Priority ordering: priority ASC (1 = most urgent first), then reported_at ASC (FIFO within priority). Claim (assigned_to) prevents two moderators from reviewing the same report. Release stale assignments: if a report has been assigned but not resolved for 30 minutes, clear assigned_to and return it to the queue.

- **Auto-moderation for scale**: At 10M reports/day, human review of everything is impossible. Use ML classifiers for initial triage: high-confidence spam/nudity detections → auto-remove (human appeals later). Low-confidence cases → human review queue. Track false positive rate of auto-moderation and retrain quarterly. Always allow appeals for auto-removed content — wrongful auto-removal of legitimate content is a severe UX failure.

- **Legal holds and mandatory reporting**: CSAM (Child Sexual Abuse Material) must be reported to NCMEC (National Center for Missing & Exploited Children) within 24 hours under US law. Design a separate CSAM_Report table that triggers an immediate legal hold (never delete the content), notifies the Trust & Safety legal team, and auto-submits the CyberTipline report. This flow must bypass the normal report queue and be handled by a dedicated pipeline.

- **User feedback loop**: Reporters who never receive feedback stop reporting. Send notifications on report resolution: "We reviewed your report about [content] and [took action / found it doesn't violate our policies]." Metrics: report response time (target <24h for normal, <1h for urgent), appeal rate, moderator consistency rate (two moderators reviewing the same content should agree 90%+ of the time).

Content moderation and report abuse system design is discussed in [Meta system design interview questions](https://www.techinterview.org/post/3233460272/meta-interview-guide-2026-facebook-instagram-whatsapp-engineering/).

Content moderation and abuse reporting design is covered in [Twitter system design interview preparation](https://www.techinterview.org/post/3233460305/twitter-x-interview-guide-2026-timeline-algorithms-real-time-search-and-content-at-scale/).

Trust and safety reporting system design is discussed in [Airbnb system design interview guide](https://www.techinterview.org/post/3233460279/airbnb-interview-guide-2026-search-systems-trust-and-safety-and-full-stack-engineering/).