Event-Driven Architecture Patterns

[Your Frontend]  ←→  [Your Backend + SDK]  ←→  [Magick Mind Platform]
  (You manage)         (SDK lives here)         (We provide)

# ✅ CORRECT: Subscribe to personal channel for specific end user
await client.realtime.subscribe(
    target_user_id="end-user-456",  # Specific end user
    on_publication=handle_event
)
# This subscribes to: personal:end-user-456#<your-service-id>

# ❌ INCORRECT: Don't subscribe to mindspace/room directly
await client.realtime.subscribe(
    target_user_id="mindspace-123",  # This is a room, not a user!
    on_publication=handle_event
)

# Robot/Device as both service and end user
client = MagickMind(
    email="robot-001@company.com",  # Service credentials
    password="robot-password"
)

# Subscribe to own channel (service_user == end_user)
await client.realtime.subscribe("robot-001")  # Same identity!
# Channel: personal:robot-001#robot-001

# Send as itself
response = client.v1.chat.send(
    sender_id="robot-001",  # Same as service user
    message="Analyze my sensor data"
)

# Receive own AI responses
@client.realtime.on("chat_message")
async def handle(event: ChatMessageEvent, ctx: EventContext):
    # ctx.target_user_id == "robot-001" (self)
    print(f"AI response: {event.payload.message}")

the Magick Mind API --event(full message)--> Your Backend --relay--> Your Frontend
                                      ↓
                                 Store in DB

"""
Pattern 1: Events as Source of Truth

Your backend receives complete data from the Magick Mind API and relays to your frontend.
"""

from magick_mind import MagickMind
from magick_mind.realtime.events import ChatMessageEvent, EventContext
from fastapi import FastAPI, WebSocket
import asyncio

# Your backend application
app = FastAPI()

class ChatBackend:
    def __init__(self, client: MagickMind):
        self.sdk_client = client
        self.your_db = YourDatabase()
        self.your_websocket_connections = set()  # Your frontend connections

    async def on_api_event(self, event: ChatMessageEvent, ctx: EventContext):
        """
        Receives event from the Magick Mind API with complete message data.
        ctx.target_user_id identifies which end-user this event is for.
        Backend processes and relays to YOUR frontend.
        """
        payload = event.payload

        print(f"📥 Received from API for {ctx.target_user_id}: {payload.message_id}")

        # Business logic: Store in YOUR database
        await self.your_db.messages.insert_one({
            "id": payload.message_id,
            "task_id": payload.task_id,
            "content": payload.message,
            "reply_to": payload.reply_to,
            "user_id": ctx.target_user_id,
            "received_at": datetime.now()
        })

        # Relay to YOUR frontend via YOUR WebSocket
        await self.broadcast_to_your_frontend({
            "type": "new_message",
            "data": {
                "id": payload.message_id,
                "content": payload.message,
                "task_id": payload.task_id
            }
        })

        print(f"✓ Stored and relayed to frontend")

    async def broadcast_to_your_frontend(self, data: dict):
        """Send to YOUR frontend's WebSocket connections"""
        for websocket in self.your_websocket_connections:
            try:
                await websocket.send_json(data)
            except:
                self.your_websocket_connections.remove(websocket)

    async def start(self, end_user_id: str):
        """Connect to the Magick Mind API and start listening for this end user"""
        # Register handler — EventContext provides user identity
        @self.sdk_client.realtime.on("chat_message")
        async def handle(event: ChatMessageEvent, ctx: EventContext):
            await self.on_api_event(event, ctx)

        # Connect SDK to the Magick Mind API
        await self.sdk_client.realtime.connect()

        # Subscribe to personal channel for this specific end user
        # Pattern: personal:<target_user_id>#<service_user_id>
        await self.sdk_client.realtime.subscribe(target_user_id=end_user_id)
        print(f"✓ Backend listening for user {end_user_id}")

# Your frontend connects to YOUR WebSocket endpoint
@app.websocket("/ws")
async def your_frontend_websocket(websocket: WebSocket):
    """Your frontend connects here, NOT to the Magick Mind API directly"""
    await websocket.accept()
    backend.your_websocket_connections.add(websocket)
    
    try:
        while True:
            # Handle messages from your frontend
            data = await websocket.receive_json()
            # Process frontend requests...
    finally:
        backend.your_websocket_connections.remove(websocket)

@client.realtime.on("chat_message")
async def on_activity(event: ChatMessageEvent, ctx: EventContext):
    activity = parse_activity(event.payload)

    # Store in Redis for dashboard queries
    await redis.lpush("activities", json.dumps(activity))

    # Push to your frontend Dashboard
    await broadcast_to_dashboard(activity)

@client.realtime.on("chat_message")
async def on_notification(event: ChatMessageEvent, ctx: EventContext):
    notification = parse_notification(event.payload)

    # Store notification
    await db.notifications.insert(notification)

    # Send push notification via FCM/APNS — ctx identifies the user
    await send_push_notification(ctx.target_user_id, notification)

@client.realtime.on("chat_message")
async def on_feed(event: ChatMessageEvent, ctx: EventContext):
    feed_item = parse_feed_item(event.payload)

    # Add to feed cache
    await redis.zadd("feed", {feed_item.id: time.now()})

    # Relay to your frontend
    await broadcast_to_feed_subscribers(feed_item)

"""Real-world: Telegram bot receiving AI responses"""

from telegram import Bot
from magick_mind import MagickMind
from magick_mind.realtime.events import ChatMessageEvent, EventContext

class TelegramBotBackend:
    def __init__(self, sdk_client: MagickMind, bot: Bot):
        self.sdk = sdk_client
        self.telegram_bot = bot
        self.chat_mappings = {}  # Map the Magick Mind API task_id → Telegram chat_id

        @sdk_client.realtime.on("chat_message")
        async def on_ai_response(event: ChatMessageEvent, ctx: EventContext):
            """
            the Magick Mind API sends AI response →
            Your bot backend relays to Telegram →
            User sees message in Telegram
            """
            payload = event.payload

            # Get which Telegram chat this belongs to
            telegram_chat_id = self.chat_mappings.get(payload.task_id)

            if telegram_chat_id:
                # Relay to YOUR frontend (Telegram)
                await self.telegram_bot.send_message(
                    chat_id=telegram_chat_id,
                    text=payload.message,
                )

                print(f"✓ Relayed to Telegram chat {telegram_chat_id}")

the Magick Mind API --event("message 123 created")--> Your Backend
                                              ↓
                                    Fetch full data from the Magick Mind API
                                              ↓
                                         Store in DB
                                              ↓
                                    Relay to Your Frontend

"""
Pattern 2: Events as Notifications

Event triggers your backend to fetch authoritative data.
"""

from magick_mind import MagickMind, ChatPayload

class ChatBackendWithFetch:
    def __init__(self, client: MagickMind):
        self.sdk_client = client
        self.your_db = YourDatabase()
    
    async def on_api_notification(self, channel: str, data: dict):
        """
        Receives notification from the Magick Mind API (minimal data).
        Backend fetches full data, then relays to frontend.
        """
        # Event: Just says "message created" with ID
        message_id = data.get("message_id")
        
        print(f"🔔 Notification from the Magick Mind API: message {message_id} created")
        
        # Fetch complete, authoritative data from the Magick Mind API
        # Note: Future Feature - Individual message fetch endpoint will be added to SDK
        # For now, use HTTP client directly:
        response = self.sdk_client.http.get(f"/v1/messages/{message_id}")
        message = ChatPayload.model_validate(response.json())
        
        print(f"📥 Fetched from the Magick Mind API: {message.content[:50]}...")
        
        # Verify data integrity before storing
        if not message.message_id or not message.content:
            print("⚠️ Invalid message data, skipping")
            return
        
        # Store in YOUR database (authoritative for your frontend)
        await self.your_db.messages.insert_one({
            "id": message.message_id,
            "task_id": message.task_id,
            "content": message.content,
            "verified": True,  # We fetched this, we trust it
            "stored_at": datetime.now()
        })
        
        # Relay to YOUR frontend via YOUR API
        await self.broadcast_to_your_frontend({
            "type": "new_message",
            "data": message.model_dump()
        })
        
        print(f"✓ Verified, stored, and relayed")

async def on_payment_notification(self, channel, data):
    payment_id = data["payment_id"]
    
    # MUST fetch authoritative data - can't trust event for money!
    payment = await self.sdk_client.http.get(f"/v1/payments/{payment_id}")
    
    # Verify, store, process
    if payment.verified:
        await self.your_db.payments.insert(payment)
        await self.process_payment(payment)
        await self.notify_your_frontend(payment)

async def on_order_notification(self, channel, data):
    order_id = data["order_id"]
    
    # Fetch complete order details
    order = await self.sdk_client.http.get(f"/v1/orders/{order_id}")
    
    # Store in your database
    await self.your_db.orders.insert(order)
    
    # Trigger fulfillment workflow
    await self.fulfillment_service.process(order)
    
    # Update your frontend
    await self.broadcast_to_admin_dashboard(order)

async def on_patient_update_notification(self, channel, data):
    patient_id = data["patient_id"]
    
    # Fetch complete, verified patient record
    record = await self.sdk_client.http.get(f"/v1/patients/{patient_id}")
    
    # Store with audit trail
    await self.your_db.patient_records.insert({
        **record,
        "fetched_at": datetime.now(),
        "verified": True
    })
    
    # Update EHR system
    await self.ehr_system.update(record)

the Magick Mind API --event(full data)--> Your Backend --quick relay--> Your Frontend
                                   ↓
                              Store in DB
                                   ↓
                         Periodic fetch (every 5min)
                                   ↓
                           Fill any gaps

"""
Pattern 3: Hybrid Approach (RECOMMENDED FOR PRODUCTION)

Fast path: Trust and relay events
Slow path: Periodic sync to catch gaps
"""

from magick_mind import MagickMind
from magick_mind.realtime.events import ChatMessageEvent, EventContext
import asyncio
from typing import Set

class ProductionChatBackend:
    """
    Production-ready backend with hybrid pattern.
    
    Used in real systems like Telegram bots, web apps, mobile backends.
    """
    
    def __init__(self, client: MagickMind):
        self.sdk_client = client
        self.your_db = YourDatabase()
        self.your_frontend_websockets = set()
        
        # Track what we've seen
        self.processed_message_ids: Set[str] = set()
        self.last_sync_cursor = None
    
    async def on_api_event(self, event: ChatMessageEvent, ctx: EventContext):
        """
        FAST PATH: Receive event from the Magick Mind API, process immediately.
        ctx.target_user_id identifies which end-user this event is for.
        """
        payload = event.payload
        
        # Deduplicate
        if payload.message_id in self.processed_message_ids:
            return
        
        print(f"⚡ Quick path [{ctx.target_user_id}]: {payload.message_id}")
        
        # Store in YOUR database
        await self.your_db.messages.insert_one({
            "id": payload.message_id,
            "content": payload.message,
            "task_id": payload.task_id,
            "user_id": ctx.target_user_id,
            "source": "realtime_event",
            "received_at": datetime.now()
        })
        
        # Relay to YOUR frontend immediately
        await self.broadcast_to_frontend({
            "type": "new_message",
            "data": payload.model_dump()
        })
        
        # Track it
        self.processed_message_ids.add(payload.message_id)
        
        print(f"✓ Stored and relayed")
    
    async def periodic_sync(self, mindspace_id: str):
        """
        RELIABLE PATH: Periodic sync to catch any missed events.
        
        Runs every 5 minutes in background.
        """
        while True:
            await asyncio.sleep(300)  # 5 minutes
            
            print("🔄 Running periodic sync...")
            
            try:
                # Fetch history from the Magick Mind API
                resp = await self.sdk_client.v1.magickspaces.get_messages(
                    mindspace_id,
                    cursor=self.last_sync_cursor,
                    limit=100,
                )
                
                messages = resp.data
                
                # Find any we missed
                gaps_found = 0
                for msg_data in messages:
                    msg_id = msg_data["id"]
                    
                    if msg_id not in self.processed_message_ids:
                        # We MISSED this event! Add it now
                        print(f"⚠️ GAP FOUND: {msg_id}")
                        gaps_found += 1
                        
                        message = ChatPayload(
                            message_id=msg_id,
                            task_id="",
                            content=msg_data["content"],
                            reply_to=msg_data.get("reply_to_message_id")
                        )
                        
                        # Store in YOUR database
                        await self.your_db.messages.insert_one({
                            "id": message.message_id,
                            "content": message.content,
                            "source": "sync_recovery",
                            "recovered_at": datetime.now()
                        })
                        
                        # Relay to YOUR frontend
                        await self.broadcast_to_frontend({
                            "type": "recovered_message",
                            "data": message.model_dump()
                        })
                        
                        self.processed_message_ids.add(msg_id)
                
                # Update cursor
                if messages:
                    self.last_sync_cursor = messages[-1]["id"]
                
                print(f"✓ Sync complete: {len(messages)} checked, {gaps_found} gaps filled")
                
            except Exception as e:
                print(f"✗ Sync failed: {e}")
    
    async def broadcast_to_frontend(self, data: dict):
        """Broadcast to YOUR frontend's WebSocket connections"""
        for ws in self.your_frontend_websockets:
            try:
                await ws.send_json(data)
            except:
                self.your_frontend_websockets.remove(ws)
    
    async def start(self, end_user_id: str, mindspace_id: str):
        """
        Start hybrid backend service for a specific end user.
        
        Args:
            end_user_id: End user to subscribe to (personal channel)
            mindspace_id: Mindspace context for history sync
        
        Sets up both fast (events) and reliable (sync) paths.
        """
        print("🚀 Starting hybrid backend...")
        
        # Register handler — EventContext provides user identity
        @self.sdk_client.realtime.on("chat_message")
        async def handle(event: ChatMessageEvent, ctx: EventContext):
            await self.on_api_event(event, ctx)
        
        # 1. Connect to the Magick Mind API realtime (fast path)
        await self.sdk_client.realtime.connect()
        
        # Subscribe to personal channel for this specific end user
        # Pattern: personal:<end_user_id>#<service_user_id>
        await self.sdk_client.realtime.subscribe(target_user_id=end_user_id)
        print(f"  ✓ Realtime connected for user {end_user_id}")
        
        # 2. Start periodic sync (reliability)
        asyncio.create_task(self.periodic_sync(mindspace_id))
        print("  ✓ Periodic sync started (safety net)")
        
        print("✓ Hybrid backend running!")
        print("  - Events: Instant relay to your frontend")
        print("  - Sync: Every 5min to catch gaps")

# Usage in your backend application
async def main():
    # Initialize SDK client
    sdk_client = MagickMind(
        email="your-service@example.com",
        password="your-password",
        base_url="https://api.magickmind.ai"
    )
    
    # Create your production backend
    backend = ProductionChatBackend(sdk_client)
    
    # Start it - subscribe to specific end user's personal channel
    await backend.start(
        end_user_id="end-user-456",  # The actual end user
        mindspace_id="mind-123"  # Mindspace context
    )
    
    # Keep running
    await asyncio.Future()

if __name__ == "__main__":
    asyncio.run(main())

# Events: Show messages instantly
# Sync: Catch any gaps due to network issues

# Events: Show updates immediately
# Sync: Verify document state periodically

# Events: Add posts to feed in real-time
# Sync: Refresh complete feed occasionally

# Events: Push notifications instantly
# Sync: Ensure app has complete data when it opens

Can you afford to lose data?
├─ YES → Need real-time speed for your frontend?
│         ├─ YES → Pattern 1: Events as Truth
│         └─ NO  → Pattern 2: Events as Notifications
│
└─ NO  → Is your frontend latency-sensitive?
          ├─ YES → Pattern 3: Hybrid ✅ (RECOMMENDED)
          └─ NO  → Pattern 2: Events as Notifications

"""
Real architecture of a Telegram bot using the Magick Mind API
"""

# Architecture:
# Telegram Chat (Your Frontend)
#     ↕
# Python Bot Backend + SDK (This is your code)
#     ↕  
# the Magick Mind API AI Service (SaaS)

class TelegramBotBackend:
    def setup_handlers(self, client):
        @client.realtime.on("chat_message")
        async def on_ai_response(event: ChatMessageEvent, ctx: EventContext):
            """the Magick Mind API sends AI response → Relay to Telegram"""
            payload = event.payload

            # Get Telegram chat to send to
            telegram_chat_id = self.task_to_chat_mapping[payload.task_id]

            # Relay to YOUR frontend (Telegram)
            await self.telegram_bot.send_message(
                chat_id=telegram_chat_id,
                text=payload.message,
            )

            # Store in your DB
            await self.db.messages.insert(payload.model_dump())

"""
React frontend ← FastAPI backend + SDK ← the Magick Mind API
"""

from fastapi import FastAPI, WebSocket

app = FastAPI()

class WebAppBackend:
    def __init__(self, client: MagickMind):
        self.frontend_connections = set()  # YOUR frontend's WebSockets

        @client.realtime.on("chat_message")
        async def on_api_event(event: ChatMessageEvent, ctx: EventContext):
            """the Magick Mind API → Your backend → Your React frontend"""
            payload = event.payload

            # Store in your DB
            await db.messages.insert(payload.model_dump())

            # Relay to YOUR frontend
            await self.broadcast_to_react_app({
                "type": "ai_message",
                "data": payload.model_dump(),
            })

@app.websocket("/ws")
async def your_frontend_connects_here(websocket: WebSocket):
    """Your React app connects to YOUR backend, not to the Magick Mind API"""
    await websocket.accept()
    backend.frontend_connections.add(websocket)

Flutter/React Native App (Your Frontend)
            ↕
     Django/FastAPI + SDK (Your Backend)
            ↕
        the Magick Mind API SaaS

@client.realtime.on("chat_message")
async def on_event_from_api(event: ChatMessageEvent, ctx: EventContext):
    # event.payload contains typed data, ctx.target_user_id identifies the user
    payload = event.payload
    ...

    # Store in YOUR database (PostgreSQL, MongoDB, etc.)
    await your_db.messages.insert({
        "id": message.message_id,
        "content": message.content,
        ...
    })

    # Send to YOUR frontend via YOUR communication channel
    # (WebSocket, REST API, GraphQL, Push Notifications, etc.)
    await your_websocket_server.broadcast(message)
    # or
    await your_rest_api.notify_clients(message)
    # or
    await send_push_notification(user_id, message)

    # Your custom logic
    if is_urgent(message):
        await trigger_alert()
    
    if contains_payment_info(message):
        await process_payment()
    
    await update_analytics(message)