greetings
 
 login
You are here:

bbsengine6.message — messaging subsystem

Status: canonical. Phase 11 (2026-09-01). Replaces the deleted bbsengine6.notify / bbsengine6.message_delivery packages (2026-07-22). See bbsengine6/TODO-message-migration.md for the full migration history.

The bbsengine6.message package is the single source of truth for channel-based pub/sub with PostgreSQL persistence, rate limiting, blocking, recipient groups, templating, and a per-member unread counter. Consumers (bed, casino, empyre, the bottombar F2 notification, the message CLI) call into this package; the old notify / message_delivery APIs are gone.

Contents

Layered architecture

Phase 11 (2026-09-01) split the package into four layers that mirror casino's four-layer architecture (see casino/SPEC.md §3):

Layer Module Role
Service bbsengine6.message.service Business orchestration. Enable/disable gate, rate-limit gating, blocking filter, recipient expansion, legacy send shim
bbsengine6.message.lib Public re-export facade + Message dataclass, MessageUrgency enum, DB helpers
DAL bbsengine6.message.dal.messages engine.__message, engine.__message_recipient I/O
bbsengine6.message.dal.recipients engine.__message_group_member expansion (@group, @everyone)
bbsengine6.message.dal.groups engine.__message_group[_member] I/O
bbsengine6.message.dal.blocking engine.__message_block I/O
bbsengine6.message.dal.ratelimit engine.__message_rate_limit, engine.__message_type reads
bbsengine6.message.dal.types engine.__message_type writes
bbsengine6.message.dal._pool CONN_POOL_PATTERN helper + information_schema.tables probe
State bbsengine6.message.cache In-memory local unread counter (no DB)
Domain bbsengine6.message.templates {var} / $var template rendering
bbsengine6.message.access (in __init__.py) Per-op authorization (subscribe / unsubscribe / list_pending)

The DAL never imports psycopg directly; all DB plumbing goes through bbsengine6.database. Services call DAL methods; DAL executes queries; the engine layer never makes direct database calls. The local unread cache in cache.py is not a DAL module because it has no DB I/O — it sits at the package root so the DAL contract stays "talks to Postgres only".

Public API

The full public surface is re-exported from bbsengine6.message.lib and wildcard-imported into bbsengine6.message.__init__. from bbsengine6.message import <name> resolves any name listed below.

Dataclasses and enums

Name Definition Notes
Message dataclass (id, channel, sender_moniker, content, data, urgency, template, template_vars, datestamp) timestamp and recipients are derived properties
MessageUrgency Enum(ROUTINE, IMPORTANT, URGENT, CRITICAL) Coerced to str via _coerce_urgency at the service boundary

Persistence and delivery

Function Signature Notes
send send(notification_type, recipients, template, template_vars=None, sender_moniker=None, data=None, urgency=None, should_persist=True, args=None, **kwargs) -> int Legacy message_delivery.send shim. Returns engine.__message.id, or 0 if disabled / rate-limited / no recipient survived expansion
store_message store_message(channel, sender_moniker, content, recipient_monikers=None, data=None, urgency="ROUTINE", template=None, template_vars=None, database=None) -> int Thin wrapper over store_message_with_checks returning the new message id
store_message_with_checks returns dict{message_id, rate_limit_ok, recipients_stored, recipients_blocked, recipients_skipped} The full-diagnostics variant used by net/integration.py and the bed MessageRouter
get_pending_messages get_pending_messages(moniker, limit=50, database=None) -> List[dict] Datestamp-DESC, status IN (pending, delivered)
get_pending_messages_prioritized get_pending_messages_prioritized(moniker, limit=50, database=None) -> List[dict] CRITICAL/URGENT/IMPORTANT/ROUTINE order then datestamp-DESC
deliver_pending_on_connect deliver_pending_on_connect(moniker, database=None) -> List[dict] Prioritized; calls mark_delivered on every row returned
mark_delivered mark_delivered(message_id, moniker, database=None) UPDATE engine.__message_recipient SET status='delivered', datedelivered=now()
mark_read mark_read(message_id, moniker, database=None) UPDATE engine.__message_recipient SET status='read', dateread=now()
expunge expunge(message_id, sender_moniker, database=None) -> bool Sender-side hard delete; FK CASCADE removes recipient rows
get_queue get_queue(moniker, database=None) -> List[dict] Legacy notify-era API; thin wrapper around get_pending_messages(..., limit=1000)
get_urgent get_urgent(moniker, limit=50, database=None) -> List[dict] urgency IN ('URGENT', 'CRITICAL'), ordered by urgency bucket then datestamp
get_unread_count get_unread_count(moniker, database=None, *, args=None, pool=None, conn=None) -> int DB-side COUNT; returns 0 and logs a warning when engine.__message_recipient is missing

Recipient resolution

Function Signature Notes
resolve_recipients resolve_recipients(recipients, database=None) -> List[str] Expands @group_name and @everyone references; depth cap 10; dedups by first occurrence
member.group_exists bbsengine6.member.group_exists(args, group_name, **kwargs) -> bool|None Survives from the deleted notify package; validates format and queries engine.__message_group
member.get_group_members bbsengine6.member.get_group_members(args, group_name, **kwargs) -> List[str]|None Survives from the deleted notify package; recursive expansion with cycle detection

Enable / disable and unread cache

Function Signature Notes
is_enabled () -> bool Module-level boolean; defaults True
enable () -> None Set the module-level boolean to True
disable () -> None Set the module-level boolean to False (every service returns early/empty)
get_local_unread_count (moniker) -> int Process-local cache; -1 means "never read"
set_local_unread_count (moniker, count) -> None Write to the cache (clamped >= 0)
bump_local_unread_count (moniker, delta=1) -> None Atomic update
clear_local_unread_cache () -> None Reset the cache (used by tests)

Groups and blocking

Function Signature Notes
create_message_group create_message_group(name, createdby=None, description=None, database=None) -> int INSERT into engine.__message_group; returns new id
add_to_message_group add_to_message_group(group_id, member_moniker, addedby=None, database=None) -> bool Idempotent INSERT (ON CONFLICT DO NOTHING)
remove_from_group remove_from_group(group_id, member_moniker, database=None) -> bool DELETE; returns True if a row was removed
get_message_group_members get_message_group_members(group_id, database=None) -> List[str] SELECT member_moniker rows
get_user_groups get_user_groups(moniker, database=None) -> List[dict] Groups the user belongs to
block_sender block_sender(blocker_moniker, blocked_moniker, database=None) -> bool Idempotent INSERT into engine.__message_block
unblock_sender unblock_sender(blocker_moniker, blocked_moniker, database=None) -> bool DELETE
is_blocked is_blocked(blocker_moniker, blocked_moniker, database=None) -> bool True iff a row exists for the pair
get_blocked get_blocked(moniker, database=None) -> List[str] Inverse of is_blocked: blocker monikers who have blocked moniker

Rate limiting

Function Signature Notes
check_rate_limit check_rate_limit(sender_moniker, message_type, database=None) -> Tuple[bool, int] Returns (allowed, remaining); 0 per-hour limit means unlimited
record_message_sent record_message_sent(sender_moniker, message_type, database=None) -> bool UPSERT into engine.__message_rate_limit for the current hour bucket
set_rate_limit set_rate_limit(type_name, limit, database=None) -> bool Idempotent UPDATE-or-INSERT on engine.__message_type.rate_limit_per_hour
get_message_type_rate_limit get_message_type_rate_limit(message_type, database=None) -> int 0 if the type is not registered

Type registration (legacy shims)

Function Signature Notes
register_type register_type(type_name, description="", rate_limit_per_hour=0, requires_approval=False, database=None) -> bool UPSERT into engine.__message_type
register_type_compat register_type_compat(type_name, urgency=None, max_per_user_per_hour=0, persist_by_default=True, args=None, **kwargs) -> bool Adapter that accepts the legacy message_delivery.register_type positional signature
get_types get_types(database=None) -> List[dict] Every registered type, sorted by type_name

Templates

Function Signature Notes
render_template render_template(template, variables) -> str {var} and $var substitution
render_message_content render_message_content(content, template, template_vars) -> str Picks template+template_vars when both are given
parse_variables_from_content parse_variables_from_content(content) -> List[str] Extracts every {var} / $var name used
get_builtin_variables () -> Dict[str, Any] year, month, day, hour, minute, timestamp, date, time
validate_template validate_template(template) -> Tuple[bool, List[str]] Returns (is_valid, errors) for brace and $var mismatches

DAL contract

bbsengine6.message.dal.* is pure Postgres I/O. Modules never import psycopg; every connection goes through bbsengine6.database (via _connect_ctx(args, pool) in dal/_pool.py). The DAL also provides a generic table_exists(cur, schema, table) helper that runs a single information_schema.tables SELECT and never raises — it returns False on any exception so a probe failure cannot mask the caller's intent. This is what get_unread_count uses to distinguish "zero unread" from "missing schema" when the bootstrap hasn't run yet.

Module Owns
messages.py engine.__message, engine.__message_recipient I/O
recipients.py engine.__member (for @everyone expansion) and engine.__message_group (for @group lookups)
groups.py engine.__message_group, engine.__message_group_member I/O
blocking.py engine.__message_block I/O
ratelimit.py engine.__message_rate_limit reads/writes; engine.__message_type reads
types.py engine.__message_type writes (upsert, set_rate_limit, list_all)
_pool.py _connect_ctx(args, pool) + table_exists(cur, schema, table)

Async DAL is not yet provided. The current implementation is fully sync (via bbsengine6.database.getpool / pool.connection()); an async counterpart under bbsengine6.message.dal.aio/ would mirror casino/dal/aiosql/ once a consumer needs it.

Recipient resolution

resolve_recipients(recipients, database=None) is the canonical recipient expansion path. It runs at the service boundary (in store_message_with_checks) so callers can pass @group_name or @everyone and get notify-style expansion transparently:

@everyone   -> SELECT moniker FROM engine.__member WHERE approved = TRUE
@group_name -> SELECT id FROM engine.__message_group WHERE name = ...
            -> SELECT member_moniker FROM engine.__message_group_member
               WHERE group_id = ...
alice       -> "alice"  (literal)

Expansion is recursive (depth cap 10) so nested groups work; the @everyone token is case-insensitive (@Everyone works the same as @everyone). The returned list preserves the first-seen order with duplicates removed.

The surviving member.group_exists and member.get_group_members functions in py/src/bbsengine6/member/lib.py cover member validation and recursive group expansion at the bbsengine6.member package boundary. They are not a transport layer — they only read engine.__message_group and engine.__message_group_member. resolve_recipients is the message-system path; member.get_group_members is the validation/lookup path. They share the same table family but have different consumers.

Rate limiting and blocking

Every store_message_with_checks call (and therefore every store_message / send) runs through _check_blocking_and_ratelimit in bbsengine6.message.service:

  1. Rate limit — when sender_moniker is not None, check_rate_limit(sender_moniker, channel, database=...) is called. A 0 rate_limit_per_hour (or an unregistered type) means "unlimited" ((True, 999)). On deny the message is rejected before any DB writes and the result dict's message_id is 0.
  2. Blocking — for every recipient, is_blocked(recipient, sender_moniker) is consulted. Blocked pairs are skipped silently and accumulated into recipients_blocked / recipients_skipped in the diagnostics dict.
  3. Recording — after a successful insert, record_message_sent(sender_moniker, channel, ...) bumps the current hour bucket's engine.__message_rate_limit.message_count.

block_sender / unblock_sender / is_blocked / get_blocked are direct UIs onto engine.__message_block and are exposed for caller-supplied blocks UI / CLI work.

Templates

bbsengine6.message.templates is pure rendering (no I/O, no policy). Two syntaxes are accepted in the same template:

render_template(template, variables) walks the variables dict once and applies both substitutions to a single result string. The dollar form is positional: $varname is replaced; $ varname (with a space) is not. Unknown variables are not an error — they simply do not match either pattern.

validate_template(template) returns (is_valid, errors) where errors lists brace mismatches, dollar mismatches, and any malformed { tokens (e.g. {1foo}).

get_builtin_variables() returns a fresh dict on every call with the current wall-clock time:

Key Value
year datetime.now().year
month datetime.now().month
day datetime.now().day
hour datetime.now().hour
minute datetime.now().minute
timestamp datetime.now().isoformat()
date datetime.now().strftime("%Y-%m-%d")
time datetime.now().strftime("%H:%M:%S")

CLI

python -m bbsengine6.message (or python -m bbsengine6.message.cli) runs the operator CLI. The default program name is bbsengine6-msg.

Subcommand Args Mutating?
list-types — no
pending moniker [--limit N] no
unread moniker no
mark-read moniker --message-id N --yes yes
mark-delivered moniker --message-id N --yes yes
expunge --message-id N --sender MONIKER --yes yes
register-type type_name [--description ...] [--rate-limit N] [--requires-approval] --yes yes
resolve --to TOKEN [--to TOKEN ...] no
send --to TOKEN [--to TOKEN ...] --type TYPE --body BODY [--sender MONIKER] [--urgency U] [--vars k=v] [--dry-run] --yes yes

Mutating subcommands require --yes; send also accepts --dry-run to preview the rendered output without writing. send --to @everyone is a privilege gate, not a confirmation: the CLI checks bbsengine6.backend.lib.issysop against the connected DB role and refuses with a clear error if current_user is not in the sysop pg role. --to accepts comma-separated tokens (alice,bob) and may be repeated.

--database overrides $BBSENGINE6_DBNAME (default zoid6).

Module API and access()

bbsengine6.message is a first-class module under the bbsengine6 module registry. bbsengine6.message.init(args) registers bbsengine6.message with apis={"access": access} so the loader can resolve the per-op authorization policy via bbsengine6.get_module_api("bbsengine6.message", "access").

access(args, op, /, **kwargs) authorizes op for the given session/message pair. Recognized op values:

op Required message keys Returns True when
subscribe moniker session.is_sysop is True, or session.moniker == target moniker
unsubscribe moniker (same as subscribe)
list_pending moniker (same as subscribe)

At module-load time (bbsengine6.module.check calls with op="run" and no session kwarg), access returns True for everyone; the per-op rules only fire when the caller passes a session kwarg. The function does NOT perform input validation (moniker present, non-empty); that's the caller's job, sitting next to the wire-envelope shape checks.

SQL surface

The schema lives under py/src/bbsengine6/sql/:

File Purpose
message.sql engine.__message, engine.__message_recipient, supporting indexes
message_groups.sql engine.__message_group, engine.__message_group_member
message_type.sql engine.__message_type
message_rate_limit.sql engine.__message_rate_limit
message_block.sql engine.__message_block
message_enum.sql urgency enum (ROUTINE / IMPORTANT / URGENT / CRITICAL)
messageview.sql engine.message, engine.message_unread, engine.message_urgent, engine.message_blocked
migrate_notify_to_message.sql One-time migration of legacy notify tables (already-migrated clusters skip)

Bootstrap verification is performed by py/src/bbsengine6/backend/checkmessage.py, which is called by backend.stage_one.main() during bbsengine6.startup.

Migration history

The full Phases 1-11 checklist is in bbsengine6/TODO-message-migration.md. The short version: