mud_server.db.database

Database initialization and management for the MUD server.

This module provides all database operations for the MUD server using SQLite. It handles: - Database schema initialization - User account management (create, authentication, roles) - Character management (creation, locations, inventory) - Session tracking (login/logout, active users) - Chat message storage and retrieval

Database Design:

Tables: - users: Account identities (login, role, status) - characters: World-facing personas owned by users - character_locations: Per-character room state - sessions: Active login sessions with activity tracking - chat_messages: All chat messages with room and recipient info

Security Considerations:

• Passwords hashed with bcrypt (never plain text)

• Email stored as hashed value only (privacy-first)

• SQL injection prevented using parameterized queries

• Session IDs are UUIDs (hard to guess)

Performance Notes:

• SQLite handles basic concurrency (~50-100 players)

• No connection pooling (single file database)

• Suitable for small-medium deployments

Attributes

DEFAULT_WORLD_ID

Functions

init_database(*[, skip_superuser])	Initialize the SQLite database with required tables.
resolve_character_name(name, *[, world_id])	Public wrapper for resolving character names from usernames or character names.
get_connection()	Create a new SQLite connection to the database file.
create_user_with_password(username, password, *[, ...])	Create a new user account and (optionally) a default character.
create_character_for_user(user_id, name, *[, ...])	Create a character for an existing user.
user_exists(username)	Return True if a user account exists.
get_user_id(username)	Return user id for the given username, or None if not found.
get_username_by_id(user_id)	Return username for a user id, or None if not found.
get_user_role(username)	Return the role for a username, or None if not found.
get_user_account_origin(username)	Return account_origin for the given username.
set_user_role(username, role)	Update a user's role.
verify_password_for_user(username, password)	Verify a password against stored bcrypt hash.
is_user_active(username)	Return True if the user is active (not banned).
deactivate_user(username)	Deactivate (ban) a user account.
activate_user(username)	Activate (unban) a user account.
change_password_for_user(username, new_password)	Change a user's password (hashes with bcrypt).
tombstone_user(user_id)	Tombstone a user account without deleting rows.
delete_user(username)	Delete a user account while preserving character data.
character_exists(name)	Return True if a character with this name exists.
get_character_by_name(name)	Return character row by name.
get_character_by_id(character_id)	Return character row by id.
get_character_name_by_id(character_id)	Return character name for the given id, or None if not found.
get_user_characters(user_id, *[, world_id])	Return all characters owned by the given user for a world.
get_user_character_world_ids(user_id)	Return the set of world ids in which the user has characters.
unlink_characters_for_user(user_id)	Detach characters from a user (used when tombstoning guest accounts).
get_character_room(name, *[, world_id])	Return the current room for a character by name within a world.
set_character_room(name, room, *[, world_id])	Set the current room for a character by name within a world.
get_characters_in_room(room, *[, world_id])	Return character names in a room with active sessions for a world.
get_character_inventory(name)	Return the character inventory as a list of item ids.
set_character_inventory(name, inventory)	Set the character inventory.
add_chat_message(character_name, message, room[, ...])	Add a chat message for a character.
get_room_messages(room, *[, limit, character_name, ...])	Get recent messages from a room. Filters whispers based on character.
create_session(user_id, session_id, *[, client_type, ...])	Create a new session record for a user.
set_session_character(session_id, character_id, *[, ...])	Attach a character (and optionally world) to an existing session.
remove_session_by_id(session_id)	Remove a specific session by its session_id.
remove_sessions_for_user(user_id)	Remove all sessions for a user (used for forced logout/ban).
update_session_activity(session_id)	Update last_activity for a session and extend expiry when sliding is enabled.
get_session_by_id(session_id)	Return session record by session_id (or None if not found).
get_active_session_count()	Count active sessions within the configured activity window.
cleanup_expired_sessions()	Remove expired sessions based on expires_at timestamp.
clear_all_sessions()	Remove all sessions from the database.
get_active_characters(*[, world_id])	Return character names with active sessions for a world.
cleanup_expired_guest_accounts()	Delete expired guest accounts and unlink their characters.
get_world_by_id(world_id)	Return a world catalog entry by id.
list_worlds(*[, include_inactive])	Return all worlds in the catalog.
list_worlds_for_user(user_id, *[, role, include_inactive])	Return worlds accessible to the given user.
get_table_names()	Return a sorted list of user-defined table names (excludes sqlite_*).
list_tables()	Return table metadata for admin database browsing.
get_table_rows(table_name[, limit])	Return column names and rows for a given table.
get_all_users_detailed()	Return detailed user list for admin database viewer.
get_all_users()	Return basic user list for admin summaries.
get_character_locations(*[, world_id])	Return character location rows with names for admin display.
get_all_sessions(*[, world_id])	Return all active (non-expired) sessions.
get_active_connections(*[, world_id])	Return active sessions with activity age in seconds.
get_all_chat_messages([limit, world_id])	Return recent chat messages across all rooms.
player_exists(username)	Backward-compatible alias for user_exists().
get_player_role(username)	Backward-compatible alias for get_user_role().
get_player_account_origin(username)	Backward-compatible alias for get_user_account_origin().
set_player_role(username, role)	Backward-compatible alias for set_user_role().
is_player_active(username)	Backward-compatible alias for is_user_active().
deactivate_player(username)	Backward-compatible alias for deactivate_user().
activate_player(username)	Backward-compatible alias for activate_user().
create_player_with_password(username, password[, ...])	Backward-compatible alias for create_user_with_password().
get_player_room(username)	Backward-compatible alias for get_character_room().
set_player_room(username, room)	Backward-compatible alias for set_character_room().
get_player_inventory(username)	Backward-compatible alias for get_character_inventory().
set_player_inventory(username, inventory)	Backward-compatible alias for set_character_inventory().
get_active_players()	Backward-compatible alias for get_active_characters().
get_players_in_room(room)	Backward-compatible alias for get_characters_in_room().
get_all_players_detailed()	Backward-compatible alias for get_all_users_detailed().
get_all_players()	Backward-compatible alias for get_all_users().
get_player_locations()	Backward-compatible alias for get_character_locations().
delete_player(username)	Backward-compatible alias for delete_user().
cleanup_temporary_accounts([max_age_hours, origin])	Backward-compatible alias for cleanup_expired_guest_accounts().
remove_session(username)	Backward-compatible alias for removing sessions by username.

Module Contents

mud_server.db.database.DEFAULT_WORLD_ID = 'pipeworks_web'

mud_server.db.database.init_database(*, skip_superuser=False)

Initialize the SQLite database with required tables.

Creates all necessary tables if they don’t exist. If MUD_ADMIN_USER and MUD_ADMIN_PASSWORD environment variables are set and no users exist, creates a superuser with those credentials (unless skip_superuser=True).

Parameters:

skip_superuser (bool) – If True, skip superuser creation from env vars.

Side Effects:

• Creates data/mud.db file if it doesn’t exist

• Creates tables if they don’t exist

• Creates superuser if env vars set and no users exist

mud_server.db.database.resolve_character_name(name, *, world_id=None)

Public wrapper for resolving character names from usernames or character names.

This preserves legacy call sites that still supply usernames while the character model is being adopted across the codebase.

mud_server.db.database.get_connection()

Create a new SQLite connection to the database file.

Returns:

sqlite3.Connection object

Return type:

sqlite3.Connection

mud_server.db.database.create_user_with_password(username, password, *, role='player', account_origin='legacy', email_hash=None, is_guest=False, guest_expires_at=None, create_default_character=True, world_id=DEFAULT_WORLD_ID)

Create a new user account and (optionally) a default character.

Parameters:

• username (str) – Unique account username.

• password (str) – Plain text password (hashed with bcrypt).

• role (str) – Role string.

• account_origin (str) – Provenance marker for cleanup/auditing.

• email_hash (str | None) – Hashed email value (nullable during development).

• is_guest (bool) – Whether this is a guest account.

• guest_expires_at (str | None) – Expiration timestamp for guest accounts.

• create_default_character (bool) – If True, create a character with the same name.

• world_id (str) – World to bind the default character to.

Returns:

True if created successfully, False if username already exists.

Return type:

bool

mud_server.db.database.create_character_for_user(user_id, name, *, is_guest_created=False, room_id='spawn', world_id=DEFAULT_WORLD_ID)

Create a character for an existing user.

Parameters:

• user_id (int) – Owning user id.

• name (str) – Character name (globally unique for now).

• is_guest_created (bool) – Marks characters created from guest flow.

• room_id (str) – Initial room id.

• world_id (str) – World the character belongs to.

Returns:

True if character created, False on constraint violation.

Return type:

bool

mud_server.db.database.user_exists(username)

Return True if a user account exists.

mud_server.db.database.get_user_id(username)

Return user id for the given username, or None if not found.

mud_server.db.database.get_username_by_id(user_id)

Return username for a user id, or None if not found.

mud_server.db.database.get_user_role(username)

Return the role for a username, or None if not found.

mud_server.db.database.get_user_account_origin(username)

Return account_origin for the given username.

mud_server.db.database.set_user_role(username, role)

Update a user’s role.

mud_server.db.database.verify_password_for_user(username, password)

Verify a password against stored bcrypt hash.

Uses a dummy hash for timing safety when user doesn’t exist.

mud_server.db.database.is_user_active(username)

Return True if the user is active (not banned).

mud_server.db.database.deactivate_user(username)

Deactivate (ban) a user account.

mud_server.db.database.activate_user(username)

Activate (unban) a user account.

mud_server.db.database.change_password_for_user(username, new_password)

Change a user’s password (hashes with bcrypt).

mud_server.db.database.tombstone_user(user_id)

Tombstone a user account without deleting rows.

mud_server.db.database.delete_user(username)

Delete a user account while preserving character data.

This performs:

• Unlink characters from the user (user_id -> NULL)

• Remove all sessions

• Tombstone the user row (soft delete)

mud_server.db.database.character_exists(name)

Return True if a character with this name exists.

mud_server.db.database.get_character_by_name(name)

Return character row by name.

mud_server.db.database.get_character_by_id(character_id)

Return character row by id.

mud_server.db.database.get_character_name_by_id(character_id)

Return character name for the given id, or None if not found.

mud_server.db.database.get_user_characters(user_id, *, world_id=None)

Return all characters owned by the given user for a world.

When world_id is omitted, the default world is used to keep legacy code paths functional during the migration.

mud_server.db.database.get_user_character_world_ids(user_id)

Return the set of world ids in which the user has characters.

This is used to enforce allow_multi_world_characters when creating new characters.

mud_server.db.database.unlink_characters_for_user(user_id)

Detach characters from a user (used when tombstoning guest accounts).

mud_server.db.database.get_character_room(name, *, world_id=None)

Return the current room for a character by name within a world.

mud_server.db.database.set_character_room(name, room, *, world_id=None)

Set the current room for a character by name within a world.

mud_server.db.database.get_characters_in_room(room, *, world_id=None)

Return character names in a room with active sessions for a world.

mud_server.db.database.get_character_inventory(name)

Return the character inventory as a list of item ids.

mud_server.db.database.set_character_inventory(name, inventory)

Set the character inventory.

mud_server.db.database.add_chat_message(character_name, message, room, recipient_character_name=None, recipient=None, *, world_id=None)

Add a chat message for a character.

Supports optional whisper recipient and uses world scoping. If world_id is omitted, the default world is used during migration.

mud_server.db.database.get_room_messages(room, *, limit=50, character_name=None, username=None, world_id=None)

Get recent messages from a room. Filters whispers based on character.

Messages are scoped to the provided world_id; default world is used when omitted to preserve legacy code paths during migration.

mud_server.db.database.create_session(user_id, session_id, *, client_type='unknown', character_id=None, world_id=None)

Create a new session record for a user.

Behavior depends on configuration:

• allow_multiple_sessions = False: remove existing sessions for the user

• allow_multiple_sessions = True: keep existing sessions

If world_id is omitted, the default world is used. This is temporary compatibility behavior until the API layer is updated to provide it.

mud_server.db.database.set_session_character(session_id, character_id, *, world_id=None)

Attach a character (and optionally world) to an existing session.

mud_server.db.database.remove_session_by_id(session_id)

Remove a specific session by its session_id.

mud_server.db.database.remove_sessions_for_user(user_id)

Remove all sessions for a user (used for forced logout/ban).

mud_server.db.database.update_session_activity(session_id)

Update last_activity for a session and extend expiry when sliding is enabled.

mud_server.db.database.get_session_by_id(session_id)

Return session record by session_id (or None if not found).

mud_server.db.database.get_active_session_count()

Count active sessions within the configured activity window.

mud_server.db.database.cleanup_expired_sessions()

Remove expired sessions based on expires_at timestamp.

mud_server.db.database.clear_all_sessions()

Remove all sessions from the database.

mud_server.db.database.get_active_characters(*, world_id=None)

Return character names with active sessions for a world.

mud_server.db.database.cleanup_expired_guest_accounts()

Delete expired guest accounts and unlink their characters.

Returns:

Number of guest users deleted.

Return type:

int

mud_server.db.database.get_world_by_id(world_id)

Return a world catalog entry by id.

Parameters:

world_id (str) – World identifier (primary key).

Returns:

Dict with world fields or None if not found.

Return type:

dict[str, Any] | None

mud_server.db.database.list_worlds(*, include_inactive=False)

Return all worlds in the catalog.

Parameters:

include_inactive (bool) – When False, only active worlds are returned.

mud_server.db.database.list_worlds_for_user(user_id, *, role=None, include_inactive=False)

Return worlds accessible to the given user.

Admins and superusers have implicit access to all worlds. Other roles must have a world_permissions row with can_access=1.

mud_server.db.database.get_table_names()

Return a sorted list of user-defined table names (excludes sqlite_*).

mud_server.db.database.list_tables()

Return table metadata for admin database browsing.

mud_server.db.database.get_table_rows(table_name, limit=100)

Return column names and rows for a given table.

mud_server.db.database.get_all_users_detailed()

Return detailed user list for admin database viewer.

mud_server.db.database.get_all_users()

Return basic user list for admin summaries.

mud_server.db.database.get_character_locations(*, world_id=None)

Return character location rows with names for admin display.

mud_server.db.database.get_all_sessions(*, world_id=None)

Return all active (non-expired) sessions.

mud_server.db.database.get_active_connections(*, world_id=None)

Return active sessions with activity age in seconds.

mud_server.db.database.get_all_chat_messages(limit=100, *, world_id=None)

Return recent chat messages across all rooms.

mud_server.db.database.player_exists(username)

Backward-compatible alias for user_exists().

mud_server.db.database.get_player_role(username)

Backward-compatible alias for get_user_role().

mud_server.db.database.get_player_account_origin(username)

Backward-compatible alias for get_user_account_origin().

mud_server.db.database.set_player_role(username, role)

Backward-compatible alias for set_user_role().

mud_server.db.database.is_player_active(username)

Backward-compatible alias for is_user_active().

mud_server.db.database.deactivate_player(username)

Backward-compatible alias for deactivate_user().

mud_server.db.database.activate_player(username)

Backward-compatible alias for activate_user().

mud_server.db.database.create_player_with_password(username, password, role='player', account_origin='legacy')

Backward-compatible alias for create_user_with_password().

mud_server.db.database.get_player_room(username)

Backward-compatible alias for get_character_room().

mud_server.db.database.set_player_room(username, room)

Backward-compatible alias for set_character_room().

mud_server.db.database.get_player_inventory(username)

Backward-compatible alias for get_character_inventory().

mud_server.db.database.set_player_inventory(username, inventory)

Backward-compatible alias for set_character_inventory().

mud_server.db.database.get_active_players()

Backward-compatible alias for get_active_characters().

mud_server.db.database.get_players_in_room(room)

Backward-compatible alias for get_characters_in_room().

mud_server.db.database.get_all_players_detailed()

Backward-compatible alias for get_all_users_detailed().

mud_server.db.database.get_all_players()

Backward-compatible alias for get_all_users().

mud_server.db.database.get_player_locations()

Backward-compatible alias for get_character_locations().

mud_server.db.database.delete_player(username)

Backward-compatible alias for delete_user().

mud_server.db.database.cleanup_temporary_accounts(max_age_hours=24, origin='visitor')

Backward-compatible alias for cleanup_expired_guest_accounts().

Args are ignored because guest expiry is timestamp-driven.

mud_server.db.database.remove_session(username)

Backward-compatible alias for removing sessions by username.