🏗️ Core Architecture

This document explains how the loglife.core framework handles messages, threading, and concurrency efficiently.

---

🔄 System Overview

The Framework is built on a Producer-Consumer architecture using Python's queue.Queue. This ensures that the Main Thread (the Web Server) is never blocked by slow operations like processing logic or making network calls.

graph LR A[User] -->|WhatsApp Webhook| B(Main Thread) B -->|Push| C{Receive Queue} C -->|Pop| D[Router Thread] D -->|Push| E{Send Queue} E -->|Pop| F[Sender Thread] F -->|Dispatch| G[Transport Adapter] G -->|API Call| H[WhatsApp API]

---

🧵 Threading Model

The framework automatically manages the critical background threads for you.

Thread	Role	Efficiency
MainThread	Web Server. Receives Webhook & pushes to Receive Queue. Returns 200 OK instantly.	⚡ High. Non-blocking.
RouterWorker	Logic. Consumes Receive Queue, processes logic (Text/Audio/VCard), and pushes replies to Send Queue.	🐢 Variable. Depends on your code speed.
SenderWorker	I/O Worker. Pops from Send Queue and delegates to the correct Transport.	🐢 Medium. Handles network latency.

Why this matters

Because the SenderWorker is separate, your logic loop can queue 10 messages instantly and go back to listening for new inputs, while the Sender Thread handles the slow task of actually delivering them one by one.

---

📨 Unified Messaging

The core unifies all inputs (WhatsApp, Emulator, Tests) into a single Message object.

Workflow

1. Ingestion: Webhook receives JSON $\rightarrow$ Wraps in Message $\rightarrow$ Pushes to Receive Queue.
2. Consumption: RouterWorker calls recv_msg(), which blocks until a message is available.
3. Production: Your logic calls send_msg(), which wraps your text in a Message $\rightarrow$ Pushes to Send Queue.
4. Delivery: SenderWorker wakes up $\rightarrow$ Pops message $\rightarrow$ Calls transports.send_whatsapp_message (or Emulator).

Transport Layer

The system decouples Threading from Protocols. * Threading: loglife.core.messaging handles queues and workers. * Transports: loglife.core.transports handles the actual API calls (Requests to WhatsApp, SSE to Emulator).

---

🔌 API Reference

Chat interface core package (transports, clients, shared protocols).

Message dataclass

Normalized representation of transport messages.

Source code in src/loglife/core/messaging.py

@dataclass(slots=True)
class Message:
    """Normalized representation of transport messages."""

    sender: str
    msg_type: str
    raw_payload: str
    client_type: str
    metadata: dict[str, Any] = field(default_factory=dict)
    attachments: dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_payload(cls, payload: Mapping[str, Any]) -> "Message":
        """Construct a message from a raw transport payload."""
        return cls(
            sender=payload["sender"],
            msg_type=payload["msg_type"],
            raw_payload=payload.get("raw_msg", ""),
            client_type=payload.get("client_type", "unknown"),
            metadata=dict(payload.get("metadata") or {}),
        )

from_payload(payload) classmethod

Construct a message from a raw transport payload.

Source code in src/loglife/core/messaging.py

@classmethod
def from_payload(cls, payload: Mapping[str, Any]) -> "Message":
    """Construct a message from a raw transport payload."""
    return cls(
        sender=payload["sender"],
        msg_type=payload["msg_type"],
        raw_payload=payload.get("raw_msg", ""),
        client_type=payload.get("client_type", "unknown"),
        metadata=dict(payload.get("metadata") or {}),
    )

init()

Initialize the core system (DB, Logging, Workers).

Call this at the start of your application.

Source code in src/loglife/core/interface.py

def init() -> None:
    """Initialize the core system (DB, Logging, Workers).

    Call this at the start of your application.
    """
    setup_logging()
    init_db()

    # Start the sender worker to handle outbound traffic
    start_sender_worker()

recv_msg(*, block=True, timeout=None)

Receive the next message from the inbound queue.

Blocks until a message is available unless block=False.

Source code in src/loglife/core/interface.py

def recv_msg(*, block: bool = True, timeout: float | None = None) -> Message:
    """Receive the next message from the inbound queue.

    Blocks until a message is available unless block=False.
    """
    return _inbound_queue.get(block=block, timeout=timeout)

send_msg(message, to=None)

Send a message to the outbound queue.

Parameters:

Name	Type	Description	Default
message	Message | str	A Message object OR a string text.	required
to	str | None	The phone number to send to (required if message is a string).	None

Source code in src/loglife/core/interface.py

def send_msg(message: Message | str, to: str | None = None) -> None:
    """Send a message to the outbound queue.

    Args:
        message: A Message object OR a string text.
        to: The phone number to send to (required if message is a string).

    """
    if isinstance(message, str):
        if not to:
            msg = "Target 'to' number is required when sending a string."
            raise ValueError(msg)
        queue_async_message(to, message)
    else:
        enqueue_outbound_message(message)