WebSockets: Real-Time Bidirectional Communication

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

// ═══════════════════════════════════════════════════════════════
// CHAT SYSTEM MESSAGE FLOW
// ═══════════════════════════════════════════════════════════════
 
                    ┌─────────────────────────────────────────┐
                    │              Message Types               │
                    ├─────────────────────────────────────────┤
                    │ Client → Server:                        │
                    │   - chat.send_message                   │
                    │   - chat.typing_start                   │
                    │   - chat.typing_stop                    │
                    │   - chat.mark_read                      │
                    │   - presence.update                     │
                    │                                          │
                    │ Server → Client:                        │
                    │   - chat.new_message                    │
                    │   - chat.user_typing                    │
                    │   - chat.message_delivered              │
                    │   - chat.message_read                   │
                    │   - presence.user_online                │
                    │   - presence.user_offline               │
                    └─────────────────────────────────────────┘
 
 
// ═══════════════════════════════════════════════════════════════
// MESSAGE STRUCTURE
// ═══════════════════════════════════════════════════════════════
 
interface ChatMessage {
    type: 'chat.send_message';
    payload: {
        conversationId: string;
        clientMessageId: string;  // For deduplication
        content: string;
        replyTo?: string;
        attachments?: Attachment[];
    };
}
 
interface MessageDelivered {
    type: 'chat.message_delivered';
    payload: {
        clientMessageId: string;
        serverMessageId: string;
        timestamp: number;
    };
}
 
interface TypingIndicator {
    type: 'chat.user_typing';
    payload: {
        conversationId: string;
        userId: string;
        isTyping: boolean;
    };
}
 
 
// ═══════════════════════════════════════════════════════════════
// SERVER-SIDE HANDLING
// ═══════════════════════════════════════════════════════════════
 
class ChatServer {
    private connections = new Map<string, WebSocket>();      // userId -> socket
    private conversations = new Map<string, Set<string>>();  // convId -> userIds
    
    handleMessage(userId: string, message: ChatMessage) {
        switch (message.type) {
            case 'chat.send_message':
                return this.handleChatMessage(userId, message.payload);
                
            case 'chat.typing_start':
                return this.broadcastTyping(userId, message.payload.conversationId, true);
        }
    }
    
    private async handleChatMessage(fromUserId: string, payload: any) {
        // 1. Persist message
        const savedMessage = await db.messages.create({
            conversationId: payload.conversationId,
            senderId: fromUserId,
            content: payload.content,
            clientMessageId: payload.clientMessageId,
        });
        
        // 2. Acknowledge to sender
        this.sendToUser(fromUserId, {
            type: 'chat.message_delivered',
            payload: {
                clientMessageId: payload.clientMessageId,
                serverMessageId: savedMessage.id,
                timestamp: savedMessage.createdAt,
            }
        });
        
        // 3. Broadcast to other conversation participants
        const participants = this.conversations.get(payload.conversationId);
        for (const userId of participants) {
            if (userId !== fromUserId) {
                this.sendToUser(userId, {
                    type: 'chat.new_message',
                    payload: savedMessage
                });
            }
        }
    }
    
    private sendToUser(userId: string, message: any) {
        const socket = this.connections.get(userId);
        if (socket && socket.readyState === WebSocket.OPEN) {
            socket.send(JSON.stringify(message));
        } else {
            // User offline - send push notification
            pushNotificationService.send(userId, message);
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

// ═══════════════════════════════════════════════════════════════
// OPERATIONAL TRANSFORMATION (OT) CONCEPT
// ═══════════════════════════════════════════════════════════════
 
// Document: "Hello World"
// User A (at server version 5): Insert "!" at position 11
// User B (at server version 5): Delete "World" (positions 6-11)
 
// Without transformation:
// If A's op applied first: "Hello World!" 
//   then B's op: "Hello !" (deleted wrong characters)
 
// With transformation:
// Server receives A then B
// A applied: "Hello World!" (server version 6)
// B must be transformed against A:
//   B originally: delete(6, 11)
//   After transformation: delete(6, 11) but now "!" is at 11
//   Result needs adjustment based on A's insert
 
// This is complex! Libraries like ShareDB, Yjs handle this.
 
 
// ═══════════════════════════════════════════════════════════════
// CRDT APPROACH (SIMPLER CONCEPTUALLY)
// ═══════════════════════════════════════════════════════════════
 
// Each character has a unique ID and position reference
// Operations are commutative - order doesn't matter
 
interface CRDTCharacter {
    id: string;           // Globally unique ID
    value: string;        // The actual character  
    afterId: string;      // ID of character this comes after
    deleted: boolean;     // Tombstone for deleted chars
    authorId: string;
    timestamp: number;
}
 
// Insertion: Create new character referencing predecessor
// Deletion: Mark character as deleted (tombstone)
// Merge: Characters self-order based on afterId + tiebreakers
 
 
// ═══════════════════════════════════════════════════════════════
// WEBSOCKET INTEGRATION
// ═══════════════════════════════════════════════════════════════
 
class CollaborativeDocument {
    private doc: CRDT;
    private socket: WebSocket;
    
    constructor(documentId: string) {
        this.doc = new CRDT();
        this.socket = new WebSocket(`wss://collab.app/doc/${documentId}`);
        
        // Receive remote operations
        this.socket.onmessage = (event) => {
            const message = JSON.parse(event.data);
            
            switch (message.type) {
                case 'operation':
                    // Apply remote operation to local document
                    this.doc.applyRemote(message.operation);
                    this.renderDocument();
                    break;
                    
                case 'cursor':
                    // Show other user's cursor position
                    this.showRemoteCursor(message.userId, message.position);
                    break;
                    
                case 'presence':
                    // Update presence list
                    this.updatePresence(message.users);
                    break;
                    
                case 'snapshot':
                    // Initial document state on join
                    this.doc.loadSnapshot(message.snapshot);
                    this.renderDocument();
                    break;
            }
        };
    }
    
    handleLocalEdit(position: number, insert?: string, delete?: number) {
        // Generate operation
        const operation = this.doc.createOperation(position, insert, delete);
        
        // Apply locally immediately (optimistic update)
        this.doc.applyLocal(operation);
        this.renderDocument();
        
        // Send to server for broadcast
        this.socket.send(JSON.stringify({
            type: 'operation',
            operation: operation
        }));
    }
    
    handleCursorMove(position: number) {
        // Debounce cursor updates (every 50ms max)
        this.debouncedCursorSend({
            type: 'cursor',
            position: position
        });
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126

// ═══════════════════════════════════════════════════════════════
// SUBSCRIPTION-BASED DASHBOARD ARCHITECTURE
// ═══════════════════════════════════════════════════════════════
 
interface Subscription {
    type: 'subscribe' | 'unsubscribe';
    channels: string[];  // e.g., ['metrics:cpu', 'metrics:memory', 'alerts:critical']
}
 
interface DashboardUpdate {
    channel: string;
    data: any;
    timestamp: number;
}
 
class DashboardServer {
    // Channel -> Set of connection IDs
    private subscriptions = new Map<string, Set<string>>();
    
    // Connection ID -> WebSocket
    private connections = new Map<string, WebSocket>();
    
    // Handle subscription requests
    handleSubscription(connId: string, sub: Subscription) {
        for (const channel of sub.channels) {
            if (sub.type === 'subscribe') {
                if (!this.subscriptions.has(channel)) {
                    this.subscriptions.set(channel, new Set());
                }
                this.subscriptions.get(channel)!.add(connId);
            } else {
                this.subscriptions.get(channel)?.delete(connId);
            }
        }
    }
    
    // Publish update to all subscribers of a channel
    publishUpdate(channel: string, data: any) {
        const subscribers = this.subscriptions.get(channel);
        if (!subscribers) return;
        
        const message = JSON.stringify({
            channel,
            data,
            timestamp: Date.now()
        });
        
        for (const connId of subscribers) {
            const socket = this.connections.get(connId);
            if (socket?.readyState === WebSocket.OPEN) {
                socket.send(message);
            }
        }
    }
}
 
 
// ═══════════════════════════════════════════════════════════════
// CLIENT-SIDE DASHBOARD
// ═══════════════════════════════════════════════════════════════
 
class DashboardClient {
    private socket: WebSocket;
    private activeSubscriptions = new Set<string>();
    private handlers = new Map<string, (data: any) => void>();
    
    subscribe(channel: string, handler: (data: any) => void) {
        this.handlers.set(channel, handler);
        
        if (!this.activeSubscriptions.has(channel)) {
            this.socket.send(JSON.stringify({
                type: 'subscribe',
                channels: [channel]
            }));
            this.activeSubscriptions.add(channel);
        }
    }
    
    unsubscribe(channel: string) {
        this.handlers.delete(channel);
        
        if (this.activeSubscriptions.has(channel)) {
            this.socket.send(JSON.stringify({
                type: 'unsubscribe',
                channels: [channel]
            }));
            this.activeSubscriptions.delete(channel);
        }
    }
    
    private handleMessage(event: MessageEvent) {
        const update = JSON.parse(event.data) as DashboardUpdate;
        const handler = this.handlers.get(update.channel);
        
        if (handler) {
            handler(update.data);
        }
    }
}
 
 
// ═══════════════════════════════════════════════════════════════
// USAGE EXAMPLE
// ═══════════════════════════════════════════════════════════════
 
const dashboard = new DashboardClient();
 
// Subscribe to metrics
dashboard.subscribe('metrics:cpu', (data) => {
    cpuChart.update(data.values);
});
 
dashboard.subscribe('metrics:memory', (data) => {
    memoryGauge.setValue(data.percentUsed);
});
 
dashboard.subscribe('alerts:critical', (data) => {
    showAlertNotification(data);
});
 
// Dynamic subscription when user opens server detail view
function onServerSelected(serverId: string) {
    dashboard.subscribe(`server:${serverId}:detailed`, (data) => {
        serverDetailPanel.update(data);
    });
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

// ═══════════════════════════════════════════════════════════════
// AUTHORITATIVE SERVER MODEL
// ═══════════════════════════════════════════════════════════════
 
// Client sends INPUTS to server
interface PlayerInput {
    type: 'input';
    inputId: number;      // Sequence number for reconciliation
    timestamp: number;
    actions: {
        moveX: number;    // -1 to 1
        moveY: number;    // -1 to 1
        jump: boolean;
        shoot: boolean;
        aimX: number;
        aimY: number;
    };
}
 
// Server sends AUTHORITATIVE STATE to clients
interface WorldState {
    type: 'state';
    serverTick: number;
    lastProcessedInput: number;  // For client reconciliation
    players: {
        id: string;
        position: { x: number; y: number; z: number };
        velocity: { x: number; y: number; z: number };
        rotation: number;
        health: number;
        animation: string;
    }[];
    projectiles: { id: string; position: any; velocity: any }[];
    events: GameEvent[];
}
 
 
// ═══════════════════════════════════════════════════════════════
// CLIENT-SIDE PREDICTION AND RECONCILIATION
// ═══════════════════════════════════════════════════════════════
 
class GameClient {
    private socket: WebSocket;
    private inputSequence = 0;
    private pendingInputs: PlayerInput[] = [];  // Inputs not yet confirmed by server
    private localPlayerState: PlayerState;
    
    // Run every frame (60fps)
    gameLoop() {
        // 1. Process local input
        const input = this.captureInput();
        
        // 2. Apply input locally (prediction)
        this.localPlayerState = this.applyInput(this.localPlayerState, input);
        
        // 3. Send to server
        const inputMsg: PlayerInput = {
            type: 'input',
            inputId: ++this.inputSequence,
            timestamp: Date.now(),
            actions: input
        };
        this.pendingInputs.push(inputMsg);
        this.socket.send(JSON.stringify(inputMsg));
        
        // 4. Render predicted state
        this.render();
    }
    
    handleServerState(state: WorldState) {
        // Find our player in server state
        const serverState = state.players.find(p => p.id === this.playerId);
        
        // Remove inputs that server has processed
        this.pendingInputs = this.pendingInputs.filter(
            input => input.inputId > state.lastProcessedInput
        );
        
        // Reconcile: Start from server state, replay pending inputs
        let reconciledState = serverState;
        for (const input of this.pendingInputs) {
            reconciledState = this.applyInput(reconciledState, input.actions);
        }
        
        this.localPlayerState = reconciledState;
        
        // Interpolate other players (they're always slightly in the past)
        this.interpolateOtherPlayers(state.players);
    }
    
    // Smoothly interpolate between received states
    interpolateOtherPlayers(serverPlayers: PlayerState[]) {
        for (const serverPlayer of serverPlayers) {
            if (serverPlayer.id === this.playerId) continue;
            
            const localPlayer = this.otherPlayers.get(serverPlayer.id);
            if (localPlayer) {
                // Store position history for interpolation
                localPlayer.addPositionSnapshot(serverPlayer.position, Date.now());
                
                // Render position from ~100ms ago for smooth interpolation
                localPlayer.renderPosition = localPlayer.getInterpolatedPosition(
                    Date.now() - 100
                );
            }
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

// ═══════════════════════════════════════════════════════════════
// RELIABLE ORDERED MESSAGING
// ═══════════════════════════════════════════════════════════════
 
interface MarketDataMessage {
    sequence: number;        // Global sequence for gap detection
    timestamp: number;       // Server timestamp (microsecond precision)
    type: 'quote' | 'trade' | 'orderbook';
    symbol: string;
    data: any;
}
 
interface OrderMessage {
    orderId: string;
    clientOrderId: string;   // Client's reference
    type: 'new_order' | 'cancel' | 'modify';
    // ... order details
}
 
interface ExecutionReport {
    type: 'execution_report';
    orderId: string;
    clientOrderId: string;
    status: 'new' | 'filled' | 'partial' | 'canceled' | 'rejected';
    fillQty?: number;
    fillPrice?: number;
    timestamp: number;
}
 
 
// ═══════════════════════════════════════════════════════════════
// CLIENT WITH GAP DETECTION AND RECOVERY
// ═══════════════════════════════════════════════════════════════
 
class TradingClient {
    private socket: WebSocket;
    private expectedSequence = 0;
    private pendingOrders = new Map<string, OrderMessage>();
    private messageBuffer: MarketDataMessage[] = [];
    
    handleMessage(msg: MarketDataMessage) {
        // Check for gaps
        if (msg.sequence > this.expectedSequence) {
            console.warn(`Gap detected: expected ${this.expectedSequence}, got ${msg.sequence}`);
            
            // Buffer this message
            this.messageBuffer.push(msg);
            
            // Request missing messages
            this.requestReplay(this.expectedSequence, msg.sequence - 1);
            return;
        }
        
        if (msg.sequence < this.expectedSequence) {
            // Duplicate or old message (from replay), ignore
            return;
        }
        
        // Expected message
        this.processMessage(msg);
        this.expectedSequence = msg.sequence + 1;
        
        // Process buffered messages that are now in sequence
        this.drainBuffer();
    }
    
    requestReplay(fromSeq: number, toSeq: number) {
        this.socket.send(JSON.stringify({
            type: 'replay_request',
            fromSequence: fromSeq,
            toSequence: toSeq
        }));
    }
    
    drainBuffer() {
        // Sort buffer by sequence
        this.messageBuffer.sort((a, b) => a.sequence - b.sequence);
        
        // Process messages that are now in order
        while (this.messageBuffer.length > 0 && 
               this.messageBuffer[0].sequence === this.expectedSequence) {
            const msg = this.messageBuffer.shift()!;
            this.processMessage(msg);
            this.expectedSequence++;
        }
    }
    
    // Order management with correlation
    submitOrder(order: OrderMessage) {
        // Track pending order
        this.pendingOrders.set(order.clientOrderId, order);
        
        // Send with timestamp for latency measurement
        this.socket.send(JSON.stringify({
            ...order,
            clientTimestamp: Date.now()
        }));
    }
    
    handleExecutionReport(report: ExecutionReport) {
        const order = this.pendingOrders.get(report.clientOrderId);
        
        if (report.status === 'filled' || report.status === 'canceled') {
            this.pendingOrders.delete(report.clientOrderId);
        }
        
        // Calculate round-trip latency
        if (order) {
            const latency = Date.now() - order.clientTimestamp;
            this.latencyMetrics.record(latency);
        }
        
        // Update UI
        this.updateOrderStatus(report);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

// ═══════════════════════════════════════════════════════════════
// DEVICE-SIDE WEBSOCKET CLIENT
// ═══════════════════════════════════════════════════════════════
 
class IoTDevice {
    private socket: WebSocket | null = null;
    private deviceId: string;
    private authToken: string;
    private telemetryQueue: TelemetryReading[] = [];
    private reconnectAttempts = 0;
    
    connect() {
        this.socket = new WebSocket(`wss://iot.example.com/device`);
        
        this.socket.onopen = () => {
            // Authenticate immediately
            this.socket!.send(JSON.stringify({
                type: 'auth',
                deviceId: this.deviceId,
                token: this.authToken
            }));
            
            this.reconnectAttempts = 0;
            
            // Flush queued telemetry
            this.flushTelemetryQueue();
        };
        
        this.socket.onmessage = (event) => {
            const msg = JSON.parse(event.data);
            this.handleServerMessage(msg);
        };
        
        this.socket.onclose = () => {
            this.scheduleReconnect();
        };
    }
    
    handleServerMessage(msg: any) {
        switch (msg.type) {
            case 'command':
                // Execute command on device
                this.executeCommand(msg.command, msg.params);
                // Acknowledge execution
                this.send({
                    type: 'command_ack',
                    commandId: msg.commandId,
                    status: 'executed',
                    result: result
                });
                break;
                
            case 'config_update':
                // Update device configuration
                this.updateConfig(msg.config);
                break;
                
            case 'firmware_update':
                // Initiate firmware update procedure
                this.startFirmwareUpdate(msg.firmwareUrl, msg.checksum);
                break;
        }
    }
    
    // Collect and batch telemetry readings
    collectTelemetry() {
        const reading: TelemetryReading = {
            timestamp: Date.now(),
            temperature: this.sensors.readTemperature(),
            humidity: this.sensors.readHumidity(),
            batteryLevel: this.getBatteryLevel()
        };
        
        if (this.socket?.readyState === WebSocket.OPEN) {
            this.send({
                type: 'telemetry',
                readings: [reading]
            });
        } else {
            // Queue for later transmission
            this.telemetryQueue.push(reading);
            
            // Trim queue if too large (avoid memory exhaustion)
            if (this.telemetryQueue.length > 1000) {
                this.telemetryQueue = this.telemetryQueue.slice(-1000);
            }
        }
    }
    
    flushTelemetryQueue() {
        if (this.telemetryQueue.length > 0) {
            this.send({
                type: 'telemetry_batch',
                readings: this.telemetryQueue
            });
            this.telemetryQueue = [];
        }
    }
}
 
 
// ═══════════════════════════════════════════════════════════════
// SERVER-SIDE DEVICE MANAGEMENT
// ═══════════════════════════════════════════════════════════════
 
class IoTServer {
    private devices = new Map<string, WebSocket>();
    
    // Send command to specific device
    sendCommand(deviceId: string, command: string, params: any): Promise<any> {
        const socket = this.devices.get(deviceId);
        
        if (!socket || socket.readyState !== WebSocket.OPEN) {
            throw new Error('Device not connected');
        }
        
        const commandId = generateId();
        
        return new Promise((resolve, reject) => {
            // Set up response handler
            this.pendingCommands.set(commandId, { resolve, reject });
            
            // Send command
            socket.send(JSON.stringify({
                type: 'command',
                commandId,
                command,
                params
            }));
            
            // Timeout
            setTimeout(() => {
                if (this.pendingCommands.has(commandId)) {
                    this.pendingCommands.delete(commandId);
                    reject(new Error('Command timeout'));
                }
            }, 30000);
        });
    }
    
    // Broadcast to all devices of a type
    broadcastToDeviceType(deviceType: string, message: any) {
        for (const [deviceId, socket] of this.devices) {
            const device = this.deviceRegistry.get(deviceId);
            if (device?.type === deviceType && socket.readyState === WebSocket.OPEN) {
                socket.send(JSON.stringify(message));
            }
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138

// ═══════════════════════════════════════════════════════════════
// NOTIFICATION SERVICE
// ═══════════════════════════════════════════════════════════════
 
interface Notification {
    id: string;
    userId: string;
    type: string;                // 'mention', 'comment', 'system_alert', etc.
    priority: 'low' | 'normal' | 'high' | 'critical';
    title: string;
    body: string;
    data: any;                   // Additional payload
    read: boolean;
    createdAt: number;
}
 
class NotificationService {
    async sendNotification(notification: Notification) {
        // 1. Persist to database
        await db.notifications.create(notification);
        
        // 2. Check user preferences
        const prefs = await this.getUserPreferences(notification.userId);
        if (this.shouldMute(prefs, notification)) {
            return; // User has muted this type
        }
        
        // 3. Try real-time delivery
        const delivered = await this.deliverRealtime(notification);
        
        // 4. Fallback for offline users
        if (!delivered && notification.priority !== 'low') {
            if (prefs.pushEnabled) {
                await pushNotificationService.send(notification);
            }
            if (notification.priority === 'critical' && prefs.emailEnabled) {
                await emailService.sendNotificationEmail(notification);
            }
        }
        
        // 5. Broadcast to other devices for sync
        await this.syncAcrossDevices(notification);
    }
    
    async deliverRealtime(notification: Notification): Promise<boolean> {
        const connections = connectionManager.getConnectionsForUser(notification.userId);
        
        if (connections.length === 0) {
            return false;
        }
        
        const message = JSON.stringify({
            type: 'notification',
            payload: notification
        });
        
        for (const conn of connections) {
            conn.send(message);
        }
        
        return true;
    }
    
    async markAsRead(userId: string, notificationIds: string[]) {
        // Update database
        await db.notifications.updateMany({
            id: { in: notificationIds },
            userId: userId
        }, {
            read: true,
            readAt: Date.now()
        });
        
        // Sync to other devices
        const connections = connectionManager.getConnectionsForUser(userId);
        const message = JSON.stringify({
            type: 'notifications_read',
            notificationIds
        });
        
        for (const conn of connections) {
            conn.send(message);
        }
    }
}
 
 
// ═══════════════════════════════════════════════════════════════
// CLIENT-SIDE NOTIFICATION HANDLER
// ═══════════════════════════════════════════════════════════════
 
class NotificationClient {
    private socket: WebSocket;
    private unreadCount = 0;
    
    handleNotification(notification: Notification) {
        // Update unread count
        if (!notification.read) {
            this.unreadCount++;
            this.updateBadge(this.unreadCount);
        }
        
        // Show UI based on priority
        switch (notification.priority) {
            case 'critical':
                // Full-screen alert for critical
                this.showCriticalAlert(notification);
                break;
            case 'high':
                // Toast that requires dismissal
                this.showPersistentToast(notification);
                break;
            case 'normal':
                // Standard toast (auto-dismiss)
                this.showToast(notification);
                break;
            case 'low':
                // Silent, just update notification center
                break;
        }
        
        // Play sound if enabled
        if (this.soundEnabled && notification.priority !== 'low') {
            this.playNotificationSound(notification.priority);
        }
        
        // Add to notification list
        this.addToNotificationList(notification);
    }
    
    handleNotificationsRead(notificationIds: string[]) {
        // Another device marked as read—update our UI
        for (const id of notificationIds) {
            this.markNotificationReadInUI(id);
        }
        this.recalculateUnreadCount();
    }
}