"""
Shared character provisioning service used by admin and player APIs.
Why this module exists:
The codebase previously duplicated character creation logic across route
handlers. This service centralizes deterministic name generation, DB create
retries, and optional entity-state seeding so every caller follows one
canonical behavior.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from pathlib import Path
from secrets import randbelow
from typing import Any
import requests
from mud_server.config import config
from mud_server.db import facade as database
from mud_server.services.condition_axis_service import (
ConditionAxisServiceError,
)
from mud_server.services.condition_axis_service import (
generate_condition_axis as service_generate_condition_axis,
)
logger = logging.getLogger(__name__)
# Keep retries intentionally small to avoid extending admin provisioning latency
# too much during degraded upstream conditions.
_NAMEGEN_MAX_ATTEMPTS = 3
_NAMEGEN_RETRYABLE_HTTP_STATUS_CODES = frozenset({408, 429, 500, 502, 503, 504})
[docs]
@dataclass(slots=True)
class CharacterProvisioningResult:
"""
Result payload for generated-character provisioning.
Attributes:
success: True when provisioning completed and DB rows were persisted.
reason: Stable reason key for caller-level error mapping.
message: Human-readable status message.
character_id: Created character id when successful.
character_name: Created character name when successful.
world_id: Target world id for the created character.
seed: Deterministic provisioning seed used for name/entity generation.
entity_state: Raw entity payload when fetched successfully.
entity_state_error: Non-fatal warning when entity seeding fails.
"""
success: bool
reason: str
message: str
character_id: int | None = None
character_name: str | None = None
world_id: str | None = None
seed: int | None = None
entity_state: dict[str, Any] | None = None
entity_state_error: str | None = None
[docs]
def generate_provisioning_seed() -> int:
"""
Generate a replayable, non-zero seed for provisioning flows.
We intentionally use ``secrets.randbelow`` rather than module-global RNG
state to avoid coupling name/entity generation to gameplay randomness.
"""
return randbelow(2_147_483_647) + 1
def _fetch_generated_name(
seed: int,
*,
class_key: str = "first_name",
) -> tuple[str | None, str | None]:
"""
Request one name token from the external name-generation API.
Retry policy:
We retry only transient upstream failures:
1. Request-layer exceptions (timeout/connection/reset).
2. Retryable HTTP statuses (408/429/5xx).
We intentionally do not retry schema/validation errors because those
indicate contract drift rather than transient transport issues.
Returns:
Tuple ``(name, error_message)``. On success, ``error_message`` is None.
"""
if not config.integrations.namegen_enabled:
return None, "Name generation integration is disabled."
base_url = config.integrations.namegen_base_url.strip().rstrip("/")
if not base_url:
return None, "Name generation integration is enabled but no base URL is configured."
endpoint = f"{base_url}/api/generate"
payload = {
"class_key": class_key,
"package_id": 2,
"syllable_key": "all",
"generation_count": 1,
"unique_only": True,
"output_format": "json",
"render_style": "title",
"seed": seed,
}
last_status_code: int | None = None
last_request_error: Exception | None = None
for attempt in range(1, _NAMEGEN_MAX_ATTEMPTS + 1):
try:
response = requests.post(
endpoint,
json=payload,
timeout=config.integrations.namegen_timeout_seconds,
)
except requests.exceptions.RequestException as exc:
last_request_error = exc
if attempt < _NAMEGEN_MAX_ATTEMPTS:
logger.warning(
"Name generation API retry attempt=%s/%s reason=request_exception error=%s",
attempt,
_NAMEGEN_MAX_ATTEMPTS,
exc,
)
continue
logger.warning(
"Name generation API request failed after %s attempts: %s",
attempt,
exc,
)
return None, "Name generation API unavailable."
if response.status_code != 200:
last_status_code = response.status_code
if (
response.status_code in _NAMEGEN_RETRYABLE_HTTP_STATUS_CODES
and attempt < _NAMEGEN_MAX_ATTEMPTS
):
logger.warning(
"Name generation API retry attempt=%s/%s reason=http_status status_code=%s",
attempt,
_NAMEGEN_MAX_ATTEMPTS,
response.status_code,
)
continue
return None, f"Name generation API returned HTTP {response.status_code}."
try:
body = response.json()
except ValueError:
logger.warning("Name generation API returned invalid JSON.")
return None, "Name generation API returned invalid JSON."
if not isinstance(body, dict):
return None, "Name generation API returned a non-object payload."
names = body.get("names")
if not isinstance(names, list) or not names or not isinstance(names[0], str):
return None, "Name generation API did not return a valid name."
generated_name = names[0].strip()
if not generated_name:
return None, "Name generation API returned an empty name."
return generated_name, None
if last_status_code is not None:
return None, f"Name generation API returned HTTP {last_status_code}."
if last_request_error is not None:
logger.warning("Name generation API request failed: %s", last_request_error)
return None, "Name generation API unavailable."
return None, "Name generation API unavailable."
[docs]
def fetch_generated_full_name(seed: int) -> tuple[str | None, str | None]:
"""
Generate a deterministic ``first last`` character name for provisioning.
The same base seed is used for both lookups with an offset for the surname
so retry attempts remain deterministic across deployments.
"""
first_name, first_error = _fetch_generated_name(seed, class_key="first_name")
if first_name is None:
return None, first_error or "Unable to generate first name."
last_name, last_error = _fetch_generated_name(seed + 1, class_key="last_name")
if last_name is None:
return None, last_error or "Unable to generate last name."
full_name = f"{first_name.strip()} {last_name.strip()}".strip()
if " " not in full_name:
return None, "Name generation API returned an invalid full name."
return full_name, None
def _resolve_world_root(world_id: str) -> Path:
"""Resolve on-disk world root for canonical service calls.
Args:
world_id: Target world identifier.
Returns:
Absolute or relative path to the configured world package root.
"""
return Path(config.worlds.worlds_root) / world_id
def _map_condition_axis_error_to_entity_state_error(exc: ConditionAxisServiceError) -> str:
"""Map canonical condition-axis service errors to provisioning-friendly text.
Provisioning intentionally treats entity-state generation as a best-effort
enrichment step. This mapper preserves the existing non-fatal UX contract
while still surfacing deterministic failure messaging for logs and API
responses.
Args:
exc: Canonical service-layer exception from condition-axis generation.
Returns:
Stable human-readable error message for provisioning payloads.
"""
if (
exc.code == "CONDITION_AXIS_UPSTREAM_UNSUPPORTED"
and not config.integrations.entity_state_base_url.strip()
):
return "Entity state integration is enabled but no base URL is configured."
if exc.code == "CONDITION_AXIS_UPSTREAM_TIMEOUT":
return "Entity state API unavailable."
if exc.code in {
"CONDITION_AXIS_UPSTREAM_GENERATION_FAILED",
"CONDITION_AXIS_UPSTREAM_UNSUPPORTED",
}:
return "Entity state API unavailable."
if exc.code == "CONDITION_AXIS_VALIDATION_ERROR":
return "Entity state generation request was invalid."
return exc.detail or "Entity state API unavailable."
[docs]
def fetch_entity_state_for_seed(
seed: int,
*,
world_id: str | None = None,
) -> tuple[dict[str, Any] | None, str | None]:
"""
Fetch an optional entity-state payload for a provisioning seed.
Entity-state integration is non-fatal: character creation continues even if
the upstream call fails.
"""
if not config.integrations.entity_state_enabled:
return None, None
if not config.integrations.entity_state_base_url.strip():
return None, "Entity state integration is enabled but no base URL is configured."
resolved_world_id = (world_id or config.worlds.default_world_id).strip()
if not resolved_world_id:
return None, "Entity state generation request was invalid."
world_root = _resolve_world_root(resolved_world_id)
try:
result = service_generate_condition_axis(
world_id=resolved_world_id,
world_root=world_root,
seed=seed,
strict_inputs=False,
)
return result.entity_state, None
except ConditionAxisServiceError as exc:
logger.warning(
"Condition-axis service failed during provisioning (world=%s seed=%s): %s (%s)",
resolved_world_id,
seed,
exc.code,
exc.detail,
)
return None, _map_condition_axis_error_to_entity_state_error(exc)
[docs]
def get_world_slot_capacity(user_id: int, world_id: str) -> tuple[int, int]:
"""
Return ``(current_count, slot_limit)`` for account ownership in one world.
"""
policy = config.resolve_world_character_policy(world_id)
slot_limit = max(0, int(policy.slot_limit_per_account))
current_count = database.get_user_character_count_for_world(user_id, world_id)
return current_count, slot_limit
[docs]
def provision_generated_character_for_user(
*,
user_id: int,
world_id: str,
max_attempts: int = 8,
) -> CharacterProvisioningResult:
"""
Create a generated-name character with optional entity-state axis seeding.
The flow is intentionally deterministic per retry sequence:
1. Resolve slot capacity early and fail fast if world budget is full.
2. Generate ``first last`` names from deterministic seeds.
3. Retry on unique-name collisions.
4. Apply optional entity-state deltas as a non-fatal post-step.
"""
current_count, slot_limit = get_world_slot_capacity(user_id, world_id)
if current_count >= slot_limit:
return CharacterProvisioningResult(
success=False,
reason="slot_limit_reached",
message=(
"No character slots are available in this world. "
f"{current_count}/{slot_limit} already used."
),
)
base_seed = generate_provisioning_seed()
chosen_name: str | None = None
chosen_seed: int | None = None
for attempt in range(max_attempts):
candidate_seed = base_seed + attempt
generated_name, name_error = fetch_generated_full_name(candidate_seed)
if generated_name is None:
return CharacterProvisioningResult(
success=False,
reason="name_generation_failed",
message=name_error or "Unable to generate character name.",
)
if database.create_character_for_user(
user_id,
generated_name,
world_id=world_id,
state_seed=candidate_seed,
):
chosen_name = generated_name
chosen_seed = candidate_seed
break
if chosen_name is None or chosen_seed is None:
return CharacterProvisioningResult(
success=False,
reason="name_collision_exhausted",
message="Unable to allocate a unique generated character name. Try again.",
)
character = database.get_character_by_name(chosen_name)
if character is None:
return CharacterProvisioningResult(
success=False,
reason="character_lookup_failed",
message="Character creation did not persist correctly.",
)
character_id = int(character["id"])
entity_state, entity_state_error = fetch_entity_state_for_seed(
chosen_seed,
world_id=world_id,
)
if entity_state is not None:
try:
database.apply_entity_state_to_character(
character_id=character_id,
world_id=world_id,
entity_state=entity_state,
seed=chosen_seed,
)
except Exception: # nosec B110 - caller receives controlled error payload
logger.exception(
"Failed to apply entity-state payload for character %s",
character_id,
)
entity_state_error = "Entity state axis seeding failed."
return CharacterProvisioningResult(
success=True,
reason="ok",
message="Character created successfully.",
character_id=character_id,
character_name=chosen_name,
world_id=world_id,
seed=chosen_seed,
entity_state=entity_state,
entity_state_error=entity_state_error,
)