"""Per-user per-channel variables (v3)
Variables are stored as Redis hashes and auto-injected
into context for recent active users.
"""
from __future__ import annotations
import json
import logging
from datetime import datetime, timezone
from typing import Any, Dict, List, TYPE_CHECKING
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
USERVARS_KEY_PREFIX = "stargazer:uservars:channel"
TTL_PRESETS = {
"1hour": 3600,
"6hours": 21600,
"24hours": 86400,
"1week": 604800,
"5ever": None,
}
DEFAULT_TTL = "24hours"
def _var_key(channel_id: str, user_id: str) -> str:
"""Internal helper: var key.
Args:
channel_id (str): Discord/Matrix channel identifier.
user_id (str): Unique identifier for the user.
Returns:
str: Result string.
"""
return (
f"{USERVARS_KEY_PREFIX}:{channel_id}"
f":user:{user_id}"
)
async def _redis(ctx):
"""Internal helper: redis.
Args:
ctx: Tool execution context providing access to bot internals.
"""
r = getattr(ctx, "redis", None)
if r is None:
raise RuntimeError("Redis not available")
return r
async def _resolve_uservars_user_id(
ctx: ToolContext,
requested_user_id: str,
) -> tuple[str, str | None]:
"""Use ctx.user_id unless another id is supplied and caller has ALTER_PRIVILEGES."""
self_id = (getattr(ctx, "user_id", "") or "").strip()
req = (requested_user_id or "").strip()
if not req or req == self_id:
return self_id, None
try:
from tools.alter_privileges import has_privilege, PRIVILEGES
r = getattr(ctx, "redis", None)
config = getattr(ctx, "config", None)
if r is None or config is None:
return self_id, json.dumps({
"success": False,
"error": "Cross-user variables require Redis and config.",
})
if not await has_privilege(
r, self_id, PRIVILEGES["ALTER_PRIVILEGES"], config,
):
return self_id, json.dumps({
"success": False,
"error": (
"Cross-user variable access requires "
"ALTER_PRIVILEGES."
),
})
return req, None
except ImportError:
return self_id, json.dumps({
"success": False,
"error": "Privilege check unavailable.",
})
# ---------------------------------------------------------------
# Tool handlers
# ---------------------------------------------------------------
async def _set_user_variable(
variable_name: str,
value: str,
user_id: str = "",
ttl_preset: str = "24hours",
ctx: ToolContext | None = None,
) -> str:
"""Internal helper: set user variable.
Args:
variable_name (str): The variable name value.
value (str): Value to set.
user_id (str): Unique identifier for the user.
ttl_preset (str): The ttl preset value.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
if not variable_name or len(variable_name) > 64:
return json.dumps({
"success": False,
"error": (
"Variable name must be 1-64 characters"
),
})
if not all(c.isalnum() or c == "_"
for c in variable_name):
return json.dumps({
"success": False,
"error": (
"Variable name must contain only "
"alphanumeric characters and underscores"
),
})
if value and len(value) > 4096:
return json.dumps({
"success": False,
"error": (
"Variable value must not exceed "
"4096 characters"
),
})
if ttl_preset not in TTL_PRESETS:
return json.dumps({
"success": False,
"error": (
"Invalid TTL preset. Must be one of: "
+ ", ".join(TTL_PRESETS.keys())
),
})
ttl_seconds = TTL_PRESETS[ttl_preset]
r = await _redis(ctx)
uid, cross_err = await _resolve_uservars_user_id(ctx, user_id)
if cross_err:
return cross_err
cid = ctx.channel_id
key = _var_key(cid, uid)
ts = datetime.now(timezone.utc)
var_data = {
"value": value,
"set_at": ts.isoformat(),
"set_at_unix": int(ts.timestamp()),
"ttl_preset": ttl_preset,
"ttl_seconds": ttl_seconds,
}
await r.hset(key, variable_name, json.dumps(var_data))
if ttl_seconds is not None:
current_ttl = await r.ttl(key)
if current_ttl < 0 or ttl_seconds > current_ttl:
await r.expire(key, ttl_seconds)
else:
await r.persist(key)
return json.dumps({
"success": True,
"message": (
f"Variable '{variable_name}' set successfully"
),
"channel_id": cid,
"user_id": uid,
"variable_name": variable_name,
"ttl_preset": ttl_preset,
"expires_in_seconds": ttl_seconds,
})
async def _get_user_variable(
variable_name: str,
user_id: str = "",
ctx: ToolContext | None = None,
) -> str:
"""Internal helper: get user variable.
Args:
variable_name (str): The variable name value.
user_id (str): Unique identifier for the user.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
r = await _redis(ctx)
uid, cross_err = await _resolve_uservars_user_id(ctx, user_id)
if cross_err:
return cross_err
key = _var_key(ctx.channel_id, uid)
var_json = await r.hget(key, variable_name)
if not var_json:
return json.dumps({
"success": True,
"found": False,
"message": (
f"Variable '{variable_name}' not found "
"for this user in this channel"
),
})
vd = json.loads(var_json)
return json.dumps({
"success": True,
"found": True,
"variable_name": variable_name,
"value": vd.get("value"),
"set_at": vd.get("set_at"),
"ttl_preset": vd.get("ttl_preset"),
})
async def _delete_user_variable(
variable_name: str,
user_id: str = "",
ctx: ToolContext | None = None,
) -> str:
"""Internal helper: delete user variable.
Args:
variable_name (str): The variable name value.
user_id (str): Unique identifier for the user.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
r = await _redis(ctx)
uid, cross_err = await _resolve_uservars_user_id(ctx, user_id)
if cross_err:
return cross_err
key = _var_key(ctx.channel_id, uid)
deleted = await r.hdel(key, variable_name)
if deleted:
return json.dumps({
"success": True,
"deleted": True,
"message": (
f"Variable '{variable_name}' "
"deleted successfully"
),
})
return json.dumps({
"success": True,
"deleted": False,
"message": (
f"Variable '{variable_name}' was not found"
),
})
async def _list_user_variables(
user_id: str = "",
ctx: ToolContext | None = None,
) -> str:
"""Internal helper: list user variables.
Args:
user_id (str): Unique identifier for the user.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
r = await _redis(ctx)
uid, cross_err = await _resolve_uservars_user_id(ctx, user_id)
if cross_err:
return cross_err
key = _var_key(ctx.channel_id, uid)
all_vars = await r.hgetall(key)
if not all_vars:
return json.dumps({
"success": True,
"variables": {},
"count": 0,
"message": (
"No variables found for this user "
"in this channel"
),
})
variables: Dict[str, Any] = {}
for name, var_json in all_vars.items():
try:
vd = json.loads(var_json)
variables[name] = {
"value": vd.get("value"),
"set_at": vd.get("set_at"),
"ttl_preset": vd.get("ttl_preset"),
}
except json.JSONDecodeError:
variables[name] = {"value": var_json}
ttl = await r.ttl(key)
return json.dumps({
"success": True,
"variables": variables,
"count": len(variables),
"key_ttl_seconds": (
ttl if ttl > 0 else None
),
})
async def _clear_user_variables(
user_id: str = "",
ctx: ToolContext | None = None,
) -> str:
"""Internal helper: clear user variables.
Args:
user_id (str): Unique identifier for the user.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
r = await _redis(ctx)
uid, cross_err = await _resolve_uservars_user_id(ctx, user_id)
if cross_err:
return cross_err
key = _var_key(ctx.channel_id, uid)
deleted = await r.delete(key)
msg = (
"Cleared all variables for user in channel"
if deleted
else "No variables found to clear"
)
return json.dumps({
"success": True,
"deleted": deleted > 0,
"message": msg,
})
async def _dump_user_variables(
ctx: ToolContext | None = None,
) -> str:
"""Dump all variables for the calling user across all channels."""
r = await _redis(ctx)
uid = ctx.user_id
# Only scan keys belonging to this user
pattern = f"{USERVARS_KEY_PREFIX}:*:user:{uid}"
keys = await r.keys(pattern)
all_entries: list[dict] = []
total = 0
for key in keys:
key_str = (
key if isinstance(key, str)
else key.decode()
)
# key format: stargazer:uservars:channel:{cid}:user:{uid}
parts = key_str.split(":")
channel_id = parts[3] if len(parts) > 3 else "?"
all_vars = await r.hgetall(key)
if not all_vars:
continue
variables: Dict[str, Any] = {}
for name, var_json in all_vars.items():
try:
vd = json.loads(var_json)
variables[name] = {
"value": vd.get("value"),
"set_at": vd.get("set_at"),
"ttl_preset": vd.get("ttl_preset"),
}
except json.JSONDecodeError:
variables[name] = {"value": var_json}
all_entries.append({
"channel_id": channel_id,
"variables": variables,
"count": len(variables),
})
total += len(variables)
return json.dumps({
"success": True,
"user_id": uid,
"total_variables": total,
"entries": all_entries,
})
# ---------------------------------------------------------------
# Internal helpers (context injection, not tools)
# ---------------------------------------------------------------
[docs]
async def get_user_variables_for_context(
channel_id: str,
user_id: str,
*,
redis_client=None,
) -> Dict[str, Any]:
"""Return {name: value} for a user in a channel."""
try:
r = redis_client
if r is None:
return {}
all_vars = await r.hgetall(
_var_key(channel_id, user_id)
)
if not all_vars:
return {}
result: Dict[str, Any] = {}
for name, var_json in all_vars.items():
try:
vd = json.loads(var_json)
result[name] = vd.get("value")
except json.JSONDecodeError:
result[name] = var_json
return result
except Exception as e:
logger.error(
"Failed to get user variables for "
"context: %s", e, exc_info=True,
)
return {}
[docs]
async def get_recent_active_users(
channel_id: str,
*,
redis_client=None,
limit: int = 5,
) -> List[Dict[str, str]]:
"""Return recent unique users in a channel."""
try:
r = redis_client
if r is None:
return []
log_key = f"stargazer_logs:{channel_id}"
recent = await r.zrevrange(log_key, 0, 100)
if not recent:
return []
seen: set[str] = set()
active: List[Dict[str, str]] = []
for msg_json in recent:
try:
msg = json.loads(msg_json)
aid = str(
msg.get("author_id")
or msg.get("user_id")
or ""
)
if not aid or aid in (
"0", "0000000000000000000",
):
continue
if aid in seen:
continue
seen.add(aid)
name = (
msg.get("author_name")
or msg.get("username")
or f"User-{aid}"
)
active.append({
"user_id": aid,
"display_name": name,
})
if len(active) >= limit:
break
except json.JSONDecodeError:
continue
return active
except Exception as e:
logger.error(
"Failed to get recent active users: %s",
e, exc_info=True,
)
return []
[docs]
async def get_all_active_user_variables(
channel_id: str,
*,
redis_client=None,
limit: int = 5,
) -> List[Dict[str, Any]]:
"""Variables for the most recent active users."""
try:
users = await get_recent_active_users(
channel_id,
redis_client=redis_client,
limit=limit,
)
if not users:
return []
out: List[Dict[str, Any]] = []
for u in users:
variables = await get_user_variables_for_context(
channel_id,
u["user_id"],
redis_client=redis_client,
)
if variables:
out.append({
"user_id": u["user_id"],
"display_name": u["display_name"],
"variables": variables,
})
return out
except Exception as e:
logger.error(
"Failed to get all active user "
"variables: %s", e, exc_info=True,
)
return []
# ---------------------------------------------------------------
# Multi-tool registration
# ---------------------------------------------------------------
TOOLS = [
{
"name": "set_user_variable",
"description": (
"Set a per-channel, per-user variable "
"with configurable TTL. Use to track "
"user-specific state, preferences, or "
"context within a channel. TTL options: "
"'1hour', '6hours', '24hours' (default), "
"'1week', '5ever' (no expiration). "
"Variables are automatically injected "
"into context for the last 5 active users."
),
"parameters": {
"type": "object",
"properties": {
"variable_name": {
"type": "string",
"description": (
"Name of the variable "
"(alphanumeric + underscores, "
"max 64 chars)."
),
},
"value": {
"type": "string",
"description": (
"Value to store "
"(max 4096 chars)."
),
},
"user_id": {
"type": "string",
"description": (
"Target user ID (defaults to "
"the invoking user)."
),
},
"ttl_preset": {
"type": "string",
"enum": [
"1hour", "6hours", "24hours",
"1week", "5ever",
],
"description": (
"Time-to-live preset "
"(default: 24hours)."
),
},
},
"required": ["variable_name", "value"],
},
"handler": _set_user_variable,
},
{
"name": "get_user_variable",
"description": (
"Get a specific variable for a user "
"in the current channel."
),
"parameters": {
"type": "object",
"properties": {
"variable_name": {
"type": "string",
"description": (
"Name of the variable "
"to retrieve."
),
},
"user_id": {
"type": "string",
"description": (
"Target user ID (defaults "
"to invoking user)."
),
},
},
"required": ["variable_name"],
},
"handler": _get_user_variable,
},
{
"name": "delete_user_variable",
"description": (
"Delete a specific variable for a user "
"in the current channel."
),
"parameters": {
"type": "object",
"properties": {
"variable_name": {
"type": "string",
"description": (
"Name of the variable "
"to delete."
),
},
"user_id": {
"type": "string",
"description": (
"Target user ID (defaults "
"to invoking user)."
),
},
},
"required": ["variable_name"],
},
"handler": _delete_user_variable,
},
{
"name": "list_user_variables",
"description": (
"List all variables for a user "
"in the current channel."
),
"parameters": {
"type": "object",
"properties": {
"user_id": {
"type": "string",
"description": (
"Target user ID (defaults "
"to invoking user)."
),
},
},
},
"handler": _list_user_variables,
},
{
"name": "clear_user_variables",
"description": (
"Clear ALL variables for a user in the "
"current channel. Use with caution."
),
"parameters": {
"type": "object",
"properties": {
"user_id": {
"type": "string",
"description": (
"Target user ID (defaults "
"to invoking user)."
),
},
},
},
"handler": _clear_user_variables,
},
{
"name": "dump_user_variables",
"description": (
"Dump YOUR OWN user variables across "
"ALL channels. Returns every variable "
"belonging to the calling user, "
"grouped by channel."
),
"parameters": {
"type": "object",
"properties": {},
},
"handler": _dump_user_variables,
},
]