WebSocket Bus

// Connect with access token
const token = 'your_access_token';
const ws = new WebSocket(`wss://cl-o.example.com/ws/bus?token=${token}`);

// Or via Authorization header (if supported by client)
const ws = new WebSocket('wss://cl-o.example.com/ws/bus');
// Set Authorization header before connecting

Client                          Server
  |                               |
  |--- GET /ws/bus?token=... ---->|
  |                               |--- Validate token
  |                               |--- Create session
  |                               |
  |<-- 101 Switching Protocols ---|
  |                               |
  |<===== WebSocket Open ========>|
  |                               |
  |<-- welcome message ----------|
  |                               |
  |--- subscribe message -------->|
  |                               |
  |<===== Event Stream ==========>|

{
  "type": "message_type",
  "data": { ... },
  "timestamp": 1738483200
}

{
  "type": "subscribe",
  "events": ["action", "presence", "typing"]
}

{
  "type": "unsubscribe",
  "events": ["typing"]
}

{
  "type": "presence",
  "status": "online",
  "activity": "browsing"
}

{
  "type": "typing",
  "recipient": "bob.example.com",
  "conversation_id": "conv_123"
}

{
  "type": "ping"
}

{
  "type": "pong",
  "timestamp": 1738483200
}

{
  "type": "welcome",
  "session_id": "sess_abc123",
  "user": "alice.example.com",
  "timestamp": 1738483200
}

{
  "type": "action",
  "action_type": "POST",
  "action_id": "a1~xyz789...",
  "issuer": "bob.example.com",
  "content": "Check out this cool feature!",
  "timestamp": 1738483200
}

{
  "type": "presence",
  "user": "bob.example.com",
  "status": "online",
  "activity": "browsing",
  "timestamp": 1738483200
}

{
  "type": "typing",
  "user": "bob.example.com",
  "conversation_id": "conv_123",
  "timestamp": 1738483200
}

{
  "type": "message",
  "message_id": "a1~msg789...",
  "sender": "bob.example.com",
  "preview": "Hey, are you available for a call?",
  "timestamp": 1738483200
}

{
  "type": "notification",
  "notification_type": "connection_request",
  "from": "charlie.example.com",
  "message": "Charlie wants to connect with you",
  "action_url": "/connections/pending",
  "timestamp": 1738483200
}

{
  "type": "error",
  "code": "E-WS-AUTH",
  "message": "Invalid or expired token",
  "timestamp": 1738483200
}

class CloudilloBus {
  constructor(domain, token) {
    this.domain = domain;
    this.token = token;
    this.ws = null;
    this.handlers = new Map();
  }

  connect() {
    this.ws = new WebSocket(`wss://cl-o.${this.domain}/ws/bus?token=${this.token}`);

    this.ws.onopen = () => {
      console.log('Connected to WebSocket Bus');

      // Subscribe to events
      this.send({
        type: 'subscribe',
        events: ['action', 'presence', 'typing', 'message']
      });
    };

    this.ws.onmessage = (event) => {
      const message = JSON.parse(event.data);
      this.handleMessage(message);
    };

    this.ws.onerror = (error) => {
      console.error('WebSocket error:', error);
    };

    this.ws.onclose = () => {
      console.log('WebSocket closed');
      // Reconnect after 5 seconds
      setTimeout(() => this.connect(), 5000);
    };

    // Send ping every 30 seconds
    setInterval(() => {
      if (this.ws.readyState === WebSocket.OPEN) {
        this.send({ type: 'ping' });
      }
    }, 30000);
  }

  send(message) {
    if (this.ws.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify(message));
    }
  }

  on(eventType, handler) {
    if (!this.handlers.has(eventType)) {
      this.handlers.set(eventType, []);
    }
    this.handlers.get(eventType).push(handler);
  }

  handleMessage(message) {
    const handlers = this.handlers.get(message.type) || [];
    handlers.forEach(handler => handler(message.data || message));
  }

  updatePresence(status, activity) {
    this.send({
      type: 'presence',
      status,
      activity
    });
  }

  sendTyping(recipient, conversationId) {
    this.send({
      type: 'typing',
      recipient,
      conversation_id: conversationId
    });
  }

  disconnect() {
    if (this.ws) {
      this.ws.close();
    }
  }
}

// Usage
const bus = new CloudilloBus('example.com', accessToken);

bus.on('action', (data) => {
  console.log('New action:', data);
  // Update UI with new post/comment/etc
});

bus.on('presence', (data) => {
  console.log('Presence update:', data);
  // Update online status indicators
});

bus.on('typing', (data) => {
  console.log('User typing:', data);
  // Show typing indicator
});

bus.on('message', (data) => {
  console.log('New message:', data);
  // Show notification, update message list
});

bus.connect();

import { useEffect, useState, useRef } from 'react';

export function useCloudilloBus(domain, token) {
  const [connected, setConnected] = useState(false);
  const [events, setEvents] = useState([]);
  const busRef = useRef(null);

  useEffect(() => {
    if (!token) return;

    const bus = new CloudilloBus(domain, token);
    busRef.current = bus;

    bus.on('welcome', () => setConnected(true));
    bus.on('action', (data) => {
      setEvents(prev => [...prev, { type: 'action', data }]);
    });
    bus.on('message', (data) => {
      setEvents(prev => [...prev, { type: 'message', data }]);
    });

    bus.connect();

    return () => {
      bus.disconnect();
      setConnected(false);
    };
  }, [domain, token]);

  return {
    connected,
    events,
    bus: busRef.current
  };
}

// Usage in component
function MyComponent() {
  const { connected, events, bus } = useCloudilloBus('example.com', accessToken);

  const handleTyping = () => {
    bus?.sendTyping('bob.example.com', 'conv_123');
  };

  return (
    <div>
      <p>Status: {connected ? 'Connected' : 'Disconnected'}</p>
      <ul>
        {events.map((event, i) => (
          <li key={i}>{event.type}: {JSON.stringify(event.data)}</li>
        ))}
      </ul>
    </div>
  );
}

use axum::{
  extract::{Query, State, WebSocketUpgrade},
  response::Response,
};
use serde::{Deserialize, Serialize};

#[derive(Deserialize)]
struct BusQuery {
  token: String,
}

pub async fn ws_bus_handler(
  ws: WebSocketUpgrade,
  Query(query): Query<BusQuery>,
  State(app): State<Arc<App>>,
) -> Response {
  // Validate access token
  let auth = match validate_access_token(&app, &query.token).await {
    Ok(auth) => auth,
    Err(_) => return (StatusCode::UNAUTHORIZED).into_response(),
  };

  // Upgrade to WebSocket
  ws.on_upgrade(move |socket| handle_bus_socket(socket, auth, app))
}

async fn handle_bus_socket(
  mut socket: WebSocket,
  auth: Auth,
  app: Arc<App>,
) {
  // Generate session ID
  let session_id = generate_session_id();

  // Send welcome message
  let welcome = json!({
    "type": "welcome",
    "session_id": session_id,
    "user": auth.id_tag,
    "timestamp": Utc::now().timestamp(),
  });

  if socket.send(Message::Text(welcome.to_string())).await.is_err() {
    return;
  }

  // Register session in bus
  app.bus.add_session(session_id.clone(), auth.id_tag.clone()).await;

  // Handle incoming messages
  while let Some(msg) = socket.recv().await {
    match msg {
      Ok(Message::Text(text)) => {
        if let Ok(message) = serde_json::from_str::<BusMessage>(&text) {
          handle_bus_message(message, &auth, &app, &mut socket).await;
        }
      }
      Ok(Message::Close(_)) => break,
      _ => {}
    }
  }

  // Cleanup session
  app.bus.remove_session(&session_id).await;
}

#[derive(Deserialize)]
#[serde(tag = "type")]
enum BusMessage {
  Subscribe { events: Vec<String> },
  Unsubscribe { events: Vec<String> },
  Presence { status: String, activity: Option<String> },
  Typing { recipient: String, conversation_id: String },
  Ping,
}

async fn handle_bus_message(
  message: BusMessage,
  auth: &Auth,
  app: &App,
  socket: &mut WebSocket,
) {
  match message {
    BusMessage::Subscribe { events } => {
      // Subscribe to events
      app.bus.subscribe(&auth.id_tag, events).await;
    }
    BusMessage::Presence { status, activity } => {
      // Broadcast presence update
      let update = json!({
        "type": "presence",
        "user": auth.id_tag,
        "status": status,
        "activity": activity,
        "timestamp": Utc::now().timestamp(),
      });

      app.bus.broadcast(&auth.id_tag, update).await;
    }
    BusMessage::Typing { recipient, conversation_id } => {
      // Send typing indicator to recipient
      let typing = json!({
        "type": "typing",
        "user": auth.id_tag,
        "conversation_id": conversation_id,
        "timestamp": Utc::now().timestamp(),
      });

      app.bus.send_to(&recipient, typing).await;
    }
    BusMessage::Ping => {
      // Respond with pong
      let pong = json!({
        "type": "pong",
        "timestamp": Utc::now().timestamp(),
      });

      socket.send(Message::Text(pong.to_string())).await.ok();
    }
    _ => {}
  }
}

bus.on('action', (data) => {
  if (data.action_type === 'POST' && data.issuer !== currentUser) {
    showNotification(`New post from ${data.issuer}`);
    prependToFeed(data);
  }
});

const onlineUsers = new Set();

bus.on('presence', (data) => {
  if (data.status === 'online') {
    onlineUsers.add(data.user);
  } else {
    onlineUsers.delete(data.user);
  }

  updateOnlineIndicators();
});

const typingUsers = new Map();

bus.on('typing', (data) => {
  typingUsers.set(data.user, Date.now());
  showTypingIndicator(data.user, data.conversation_id);

  // Clear after 5 seconds
  setTimeout(() => {
    if (Date.now() - typingUsers.get(data.user) >= 5000) {
      hideTypingIndicator(data.user);
      typingUsers.delete(data.user);
    }
  }, 5000);
});

// Send typing indicator when user types
messageInput.addEventListener('input', () => {
  bus.sendTyping(recipientId, conversationId);
});

bus.on('message', (data) => {
  playNotificationSound();
  showNotification(`New message from ${data.sender}`, data.preview);

  if (currentConversation === data.sender) {
    fetchAndDisplayMessage(data.message_id);
  } else {
    incrementUnreadCount(data.sender);
  }
});

Outbound Action Forwarding:

1. User creates action via POST /api/actions
2. Action is processed and stored
3. ActionCreatorTask completes
4. Forward to WebSocket Bus:
   a. Find all WebSocket sessions for user's tenant
   b. Send action notification to each session
5. Client receives real-time update

Audience Targeting:
- POST: Broadcast to creator's sessions
- MSG: Forward to all conversation participant sessions
- CMNT/REACT: Forward to parent action owner's sessions

{
  "type": "action",
  "action_type": "POST",
  "action_id": "a1~xyz789...",
  "issuer": "alice.example.com",
  "content": "New post content",
  "timestamp": 1738483200,
  "source": "local"
}

Inbound Action Forwarding:

1. Remote action arrives at POST /api/inbox
2. ActionVerifierTask verifies and stores action
3. After successful processing:
   a. Identify target tenant from action audience
   b. Find all WebSocket sessions for that tenant
   c. Broadcast action to all sessions
4. All connected clients see the new action

Broadcast Scope:
- All clients for the target tenant receive notification
- Client-side filtering determines what to display
- Enables real-time feed updates without polling

{
  "type": "action",
  "action_type": "POST",
  "action_id": "a1~abc123...",
  "issuer": "bob.remote.com",
  "content": "Federated post",
  "timestamp": 1738483200,
  "source": "federated"
}

Local Action Creation:
┌─────────┐    ┌──────────┐    ┌─────────────┐    ┌──────────────┐
│ Client  │───►│ API      │───►│ Task        │───►│ WebSocket    │
│ (POST)  │    │ Handler  │    │ Scheduler   │    │ Bus          │
└─────────┘    └──────────┘    └─────────────┘    └──────┬───────┘
                                                         │
                                    ┌────────────────────┼────────────────────┐
                                    ▼                    ▼                    ▼
                              ┌──────────┐        ┌──────────┐        ┌──────────┐
                              │ Session 1│        │ Session 2│        │ Session 3│
                              │ (Tab 1)  │        │ (Tab 2)  │        │ (Mobile) │
                              └──────────┘        └──────────┘        └──────────┘

Federated Action Reception:
┌─────────────┐    ┌──────────┐    ┌─────────────┐    ┌──────────────┐
│ Remote      │───►│ /api/    │───►│ Verifier    │───►│ WebSocket    │
│ Instance    │    │ inbox    │    │ Task        │    │ Bus          │
└─────────────┘    └──────────┘    └─────────────┘    └──────┬───────┘
                                                              │
                                                    Broadcast to all
                                                    tenant sessions

Push Notification Decision:

1. Action is processed successfully
   │
2. Check if forwarded to WebSocket
   ├─ Yes → User is online, skip push notification
   └─ No  → User is offline, continue...
         │
3. Check action type eligibility
   ├─ MSG (Message) → Eligible
   ├─ CONN (Connection request) → Eligible
   ├─ FSHR (File share) → Eligible
   ├─ CMNT (Comment on user's content) → Eligible
   └─ Other → Not eligible, skip
         │
4. Check user notification preferences
   ├─ Notifications disabled → Skip
   └─ Notifications enabled → Continue...
         │
5. Check notification throttling
   ├─ Too many recent notifications → Skip (prevent spam)
   └─ Within limits → Continue...
         │
6. Schedule push notification task

schedule_push_notification(action, recipient):
    # Check eligibility
    if NOT is_push_eligible(action.type):
        return

    # Check if user is online
    if websocket_bus.has_active_session(recipient):
        return  # User will see via WebSocket

    # Check user preferences
    prefs = load_notification_preferences(recipient)
    if NOT prefs.push_enabled:
        return

    if NOT prefs.allows_type(action.type):
        return

    # Check throttling
    recent_count = count_recent_pushes(recipient, window: 1 hour)
    if recent_count > MAX_PUSHES_PER_HOUR:
        return

    # Schedule the push
    task = PushNotificationTask {
        recipient: recipient,
        action_type: action.type,
        sender: action.issuer,
        preview: generate_preview(action),
        action_url: generate_deep_link(action),
    }

    scheduler.task(task).schedule()

// During WebSocket setup
bus.send({
  type: 'register_push',
  token: pushNotificationToken,  // From FCM/APNs
  platform: 'android'  // or 'ios', 'web'
});

// Server stores token for offline delivery