Chat Moderation System

A real-time chat moderation system using Rust (WebSocket), Python (FastAPI), Kafka, and PostgreSQL.

Architecture Overview

This system provides real-time chat moderation with the following components:

1. Rust WebSocket Service: Handles WebSocket connections from clients, publishes messages to Kafka, and forwards moderation results back to clients.
2. Python FastAPI Service: Consumes chat messages from Kafka, performs moderation (using keyword filtering or LLM), and publishes the results back to Kafka.
3. Apache Kafka: Message broker for asynchronous communication between services.
4. PostgreSQL: Persistent storage for chat messages and moderation results.

Message Flow

1. Client connects to the Rust WebSocket service
2. Client sends a chat message
3. Rust service:
   • Stores the message in PostgreSQL
   • Publishes the message to Kafka topic chat_messages
4. Python service:
   • Consumes messages from chat_messages topic
   • Performs moderation (keyword filtering or LLM-based)
   • Stores moderation results in PostgreSQL
   • Publishes moderation results to Kafka topic moderation_results
5. Rust service:
   • Consumes moderation results from moderation_results topic
   • Forwards the results back to the appropriate client

Project Structure

• /docker: Docker Compose configurations and SQL initialization scripts
• /rust-service: Rust WebSocket service implementation
• /python-service: Python FastAPI moderation service implementation
• /docs: Additional documentation (if any)
• /tests: End-to-end tests

Getting Started

Prerequisites

• Docker and Docker Compose
• Rust (if developing the WebSocket service)
• Python 3.11+ (if developing the moderation service)

Quick Start with Docker Compose

The easiest way to run the entire system is using Docker Compose:

# Clone the repository
git clone https://github.com/yourusername/chat-moderation.git
cd chat-moderation

# Start all services
docker compose -f docker/docker-compose.yml up --build

This will start:

• Zookeeper (port 2181)
• Kafka (port 9092)
• PostgreSQL (port 5432)
• Rust WebSocket Service (port 8080)
• Python FastAPI Service (port 8000)

Manual Setup & Development

Rust WebSocket Service

# Navigate to the Rust service directory
cd rust-service

# Create .env file (if not exists)
echo "DATABASE_URL=postgres://user:password@localhost:5432/chatdb" > .env
echo "RUST_LOG=debug" >> .env
echo "KAFKA_BOOTSTRAP_SERVERS=localhost:9092" >> .env

# Build the service
cargo build

# Run the service
cargo run

The Rust service will be available at ws://localhost:8080/ws.

Python FastAPI Service

# Navigate to the Python service directory
cd python-service

# Create and activate virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies
pip install -r requirements.txt

# Create .env file (if not exists) - basic version
echo "KAFKA_BOOTSTRAP_SERVERS=localhost:9092" > .env
echo "KAFKA_CONSUMER_TOPIC=chat_messages" >> .env
echo "KAFKA_PRODUCER_TOPIC=moderation_results" >> .env
echo "KAFKA_GROUP_ID=moderation-service" >> .env
echo "DATABASE_URL=postgresql+asyncpg://user:password@localhost:5432/chatdb" >> .env
echo "USE_LLM_MODERATION=false" >> .env

# Start the FastAPI server
uvicorn app.main:app --reload

The Python service API will be available at http://localhost:8000.

API Endpoints

Rust WebSocket Service

• GET /ws: WebSocket endpoint for chat clients
• GET /health: Health check endpoint

Python FastAPI Service

• GET /: API information
• GET /health: Health check endpoint
• POST /moderate: HTTP endpoint for manual moderation (useful for testing)

  {
    "user_id": "user123",
    "message_id": 1,
    "message": "Hello world"
  }

Running Tests

Rust Tests

# Navigate to the Rust service directory
cd rust-service

# Run all tests
cargo test

# Run specific test
cargo test test_name

# Run tests with output
cargo test -- --nocapture

Python Tests

# Navigate to the Python service directory
cd python-service
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Run all tests
pytest

# Run with coverage
pytest --cov=app

# Run specific test module
pytest tests/test_module.py

# Run tests with verbose output
pytest -v

End-to-End Testing

# Ensure all services are running
docker-compose -f docker/docker-compose.yml up -d

# Run the end-to-end tests
python tests/end_to_end_test.py

Error Handling

The system implements robust error handling:

1. Kafka Connectivity: Both services handle Kafka connection failures gracefully, with automatic reconnection and message buffering.
2. Database Connectivity: Connection pooling and retry mechanisms ensure database operations are reliable.
3. WebSocket Connections: The Rust service manages client disconnections properly, cleaning up resources.

Development Workflow

1. Start the Docker infrastructure (Kafka, Zookeeper, Postgres)
2. Run both services in development mode
3. Make code changes
4. Run tests to verify changes
5. Restart services to apply changes

License

MIT License

Contributing

1. Fork the repository
2. Create a feature branch
3. Make your changes
4. Run tests
5. Submit a pull request