Skip to content

Module System

StarStreamer uses a modular architecture to organize functionality into self-contained, reusable components. Modules provide the business logic and features, while plugins handle platform integrations. This separation allows for clean code organization and makes it easy to add, remove, or modify features without affecting the core framework.

What is a Module?

A module in StarStreamer is a self-contained package that provides features and business logic: - Groups related functionality together (chat commands, alerts, economy) - Uses plugins to interact with external platforms (Twitch, OBS, Discord) - Can be dynamically loaded and unloaded - Has its own actions, services, and configuration - Integrates seamlessly with the event system

Module vs Plugin

  • Module = Business logic and features (what your bot does)
  • Plugin = Platform integration (how your bot connects to services)

For example: - The Chat Module provides commands like !hello and !uptime - The Twitch Plugin handles connecting to Twitch and sending messages - The Chat Module uses the Twitch Plugin to send responses

Module Structure

Each module follows a standard structure:

modules/
├── your_module/
│   ├── __init__.py       # Module exports
│   ├── module.py         # Module definition (extends BaseModule)
│   └── actions/          # Event handlers
│       ├── __init__.py
│       └── handlers.py   # Your action handlers

Creating a Module

Step 1: Define Your Module Class

Create a module.py file that extends BaseModule:

from modules.base import BaseModule

class YourModule(BaseModule):
    """Your module description"""

    @property
    def module_name(self) -> str:
        """Unique identifier for this module"""
        return "your_module"

    @property
    def description(self) -> str:
        """Human-readable description"""
        return "This module does amazing things"

    @property
    def version(self) -> str:
        """Module version"""
        return "1.0.0"

    @property
    def author(self) -> str:
        """Module author"""
        return "Your Name"

    async def initialize(self) -> None:
        """Initialize module resources"""
        # Set up any resources your module needs
        pass

    async def shutdown(self) -> None:
        """Clean up module resources"""
        # Clean up resources when module is unloaded
        pass

    async def register_actions(self) -> None:
        """Register action handlers"""
        # Import your actions to register them
        from modules.your_module.actions import handlers  # noqa: F401

Step 2: Create Action Handlers

Add your event handlers in actions/handlers.py:

from starstreamer import on_event
from starstreamer.triggers import trigger, CommandTrigger
from starstreamer.plugins.twitch import TwitchClient  # Plugin import
from starstreamer.runtime.types import Event

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!yourcommand"))
async def your_command(event: Event, twitch: TwitchClient) -> None:
    """Handle your custom command using Twitch plugin"""
    username = event.data['user']['display_name']
    await twitch.send_message(f"Hello {username} from YourModule!")

Step 3: Register Your Module

The module will be automatically discovered and loaded if placed in the modules/ directory. You can also manually register it:

from modules.registry import ModuleRegistry
from modules.your_module import YourModule

registry = ModuleRegistry()
module = YourModule()
await registry.register_module(module)

Core Modules

StarStreamer comes with several built-in modules:

Chat Module

Provides basic chat commands and interactions: - !hello - Greeting command - !ping - Bot responsiveness check - !commands - List available commands - !uptime - Stream uptime - !clip - Create a clip

Alerts Module

Handles stream event notifications: - Follow alerts - Subscription notifications - Raid announcements - Cheer (bits) alerts - Hype train events

RPG Module

Economy and gamification features: - !balance - Check currency balance - !work - Earn currency (5-minute cooldown) - !daily - Daily bonus - !give @user amount - Transfer currency - !gamble amount - Gamble currency - !leaderboard - Top users by balance

Module Lifecycle

Loading

  1. Module class is instantiated
  2. initialize() is called to set up resources
  3. register_actions() is called to register event handlers
  4. Module is added to the registry

Runtime

  • Event handlers respond to events
  • Module can access services via dependency injection
  • Module state is maintained across events

Unloading

  1. shutdown() is called to clean up resources
  2. Event handlers are unregistered
  3. Module is removed from the registry

Dependency Injection in Modules

Modules and their actions can use dependency injection to access services:

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!stats"))
async def stats_command(
    event: Event,
    twitch: TwitchClient,        # Twitch plugin injected
    economy: EconomyService,     # Economy service injected
    logger: logging.Logger       # Logger injected
) -> None:
    """Command that uses plugin and services via dependency injection"""
    user_id = event.data['user']['id']
    balance = await economy.get_balance(user_id)

    await twitch.send_message(f"Your balance: {balance}")  # Uses Twitch plugin
    logger.info(f"Stats requested by user {user_id}")

Module Configuration

Modules can have their own configuration:

class ConfigurableModule(BaseModule):
    def __init__(self, config: dict | None = None) -> None:
        self.config = config or {}
        self.prefix = self.config.get("command_prefix", "!")

    async def initialize(self) -> None:
        # Use configuration during initialization
        if self.config.get("auto_greet", False):
            # Set up auto-greeting
            pass

Best Practices

1. Keep Modules Focused

Each module should have a single, clear purpose. Don't try to do everything in one module.

2. Use Dependency Injection

Don't import plugins or services directly. Use DI for better testability:

# Good - Plugin and services injected
async def handler(event: Event, twitch: TwitchClient, economy: EconomyService):
    await twitch.send_message("Hello")

# Avoid - Direct imports create tight coupling
from starstreamer.plugins.twitch import twitch_client
from starstreamer.services.economy import economy_service
async def handler(event: Event):
    await twitch_client.send_message("Hello")

3. Handle Errors Gracefully

Always handle potential errors in your modules:

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!risky"))
async def risky_command(event: Event, twitch: TwitchClient, logger: logging.Logger):
    try:
        # Risky operation
        result = await do_something_risky()
        await twitch.send_message(f"Success: {result}")
    except Exception as e:
        logger.error(f"Error in risky command: {e}")
        await twitch.send_message("Something went wrong!")

4. Document Your Module

Always include docstrings and type hints:

class WellDocumentedModule(BaseModule):
    """
    A module that demonstrates good documentation practices.

    This module provides example functionality to show
    how modules should be documented.
    """

    async def initialize(self) -> None:
        """
        Initialize the module's resources.

        Sets up database connections, loads configuration,
        and prepares any required state.
        """
        pass

5. Test Your Module

Write tests for your module's functionality:

import pytest
from modules.your_module import YourModule

@pytest.mark.asyncio
async def test_module_initialization():
    module = YourModule()
    await module.initialize()
    assert module.module_name == "your_module"
    await module.shutdown()

Advanced Topics

Inter-Module Communication

Modules can communicate through the event system:

# In module A
from starstreamer.core.event_bus import get_event_bus

async def trigger_module_b():
    bus = get_event_bus()
    await bus.emit("custom.module_a.event", {"data": "hello"})

# In module B
@on_event("custom.module_a.event")
async def handle_module_a_event(event: Event):
    print(f"Received from module A: {event.data}")

Module Dependencies

Specify dependencies between modules:

class DependentModule(BaseModule):
    @property
    def dependencies(self) -> list[str]:
        """List of required module names"""
        return ["chat", "economy"]

Hot Reloading

StarStreamer provides intelligent hot reloading for rapid development without losing connections or state. The system uses a hybrid approach that optimizes for both safety and development speed.

Development Mode with Auto-Reload

Enable hot reloading during development:

# Start with automatic file watching
uv run python src/main.py --reload

# Monitor specific directories (default: modules/, src/starstreamer/)
uv run python src/main.py --reload --watch-dirs modules,src

Hybrid Reload Strategy

The system intelligently chooses between hot reload and full restart based on what changed:

Hot Reload (Fast, Preserves State) - Trigger: Changes in modules/ directory - Behavior: Reloads only the affected module using importlib.reload() - Preserves: - Twitch WebSocket connections - Database connections - Service container state - Event bus configuration - Use Cases: Adding/modifying commands, fixing module logic, adjusting triggers

Full Restart (Safe, Clean State) - Trigger: Changes in src/starstreamer/ directory - Behavior: Graceful shutdown and complete restart - Use Cases: Core framework changes, dependency injection modifications, event bus changes

File Watcher Details

The hot reload system monitors file changes with intelligent filtering:

# Monitored file patterns
*.py files in watched directories

# Ignored patterns  
**/test_*.py          # Test files
**/__pycache__/**     # Python cache
**/*.pyc             # Compiled Python

Debouncing: Changes are debounced with a 300ms delay to handle multiple rapid file modifications.

Web Interface Module Management

StarStreamer provides a modern web interface for module management at http://localhost:8888/modules:

Features: - Real-time module statistics - View total, enabled, and loaded module counts - Interactive module list - See all discovered modules with their current status - One-click actions - Enable, disable, and reload modules without restarting - Status indicators - Visual feedback for module states (loaded, enabled, disabled) - Hot reload support - Safely reload module code during development

Module Actions: - Load - Load an unloaded module - Enable - Activate a loaded but disabled module - Disable - Deactivate a module while keeping it loaded - Reload - Hot reload module code (useful for development)

The web interface provides a user-friendly alternative to programmatic module management and is especially useful during development and for non-technical users.

Manual Module Reloading

You can also reload modules programmatically:

from modules.registry import ModuleRegistry

# Reload a specific module
success = await registry.reload_module("chat")
if success:
    print("Chat module reloaded successfully")

# Discover and load new modules
await registry.discover_modules()

# Get reload statistics
stats = await registry.get_reload_stats()
print(f"Successful reloads: {stats['successful']}")
print(f"Failed reloads: {stats['failed']}")

Development Workflow Example

  1. Start development mode: 

    uv run python src/main.py --reload

  2. Modify a command in modules/chat/actions/basic_commands.py: 

    @on_event("twitch.chat.message")
@trigger(CommandTrigger("!hello"))
async def hello_command(event: Event, twitch: TwitchClient):
    username = event.data['user']['display_name']
    # ✨ Change this message
    await twitch.send_message(f"Hey there {username}! 🎉")

  3. Save the file → Module hot reloads automatically → Test immediately in chat

  4. Modify core framework in src/starstreamer/core/event_bus.py → Full restart occurs → All modules reload cleanly

Hot Reload Safety

The hot reload system includes several safety mechanisms:

  • Exception Isolation: Failed reloads don't crash the application
  • State Validation: Module state is validated after reload
  • Rollback Support: Failed reloads restore the previous module version
  • Handler Cleanup: Old event handlers are properly unregistered before reload
# Example of safe hot reload with error handling
try:
    await registry.reload_module("problematic_module")
except ImportError as e:
    logger.error(f"Module has syntax errors: {e}")
    # Previous version continues running
except Exception as e:
    logger.error(f"Reload failed: {e}")
    # Module remains in previous working state

Performance Benefits

Hot reloading provides significant development speed improvements:

  • No Connection Loss: Twitch EventSub connections remain active
  • Preserved State: Database connections and service state maintained
  • Fast Feedback: Module changes are live in ~200ms
  • Reduced Downtime: No need to restart entire application for module changes

Best Practices for Hot Reloading

  1. Structure modules for reloadability: 

    # ✅ Good: Stateless handlers
@on_event("twitch.chat.message")
async def command(event: Event, twitch: TwitchClient):
    # No module-level state

# ❌ Avoid: Module-level mutable state
user_cache = {}  # This won't reset on hot reload

  2. Use dependency injection instead of global imports: 

    # ✅ Good: Services injected
async def handler(event: Event, economy: EconomyService):
    pass

# ❌ Avoid: Global service imports
from starstreamer.services.economy import economy_service

  3. Test hot reloads during development: 

    # Make a small change and verify it works
echo "# timestamp: $(date)" >> modules/chat/actions/basic_commands.py

Examples

Simple Greeting Module

# modules/greeter/module.py
from modules.base import BaseModule

class GreeterModule(BaseModule):
    @property
    def module_name(self) -> str:
        return "greeter"

    async def register_actions(self) -> None:
        from modules.greeter.actions import greetings  # noqa: F401

# modules/greeter/actions/greetings.py
from starstreamer import on_event
from starstreamer.triggers import trigger, CommandTrigger
from starstreamer.plugins.twitch import TwitchClient  # Import Twitch plugin
from starstreamer.runtime.types import Event

@on_event("twitch.follow")
async def greet_follower(event: Event, twitch: TwitchClient):
    """Handle follow events using Twitch plugin"""
    username = event.data['user']['display_name']
    await twitch.send_message(f"Welcome {username}! Thanks for following!")

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!greet"))
async def greet_command(event: Event, twitch: TwitchClient):
    """Handle greet command using Twitch plugin"""
    username = event.data['user']['display_name']
    await twitch.send_message(f"Hello {username}! 👋")

Next Steps