import 'dart:async'; import 'dart:convert'; import 'package:relic/relic.dart'; import 'package:web_socket/web_socket.dart'; import '../auth/crypto_service.dart'; import '../state/app_state.dart'; final _gatewayRegistry = _GatewayRegistry(); /// WebSocket gateway with JWT auth and channel subscriptions. Result gatewayHandler(Request req) { final token = _extractGatewayToken(req); if (token == null) { return Response.unauthorized( body: Body.fromString('Bearer token required'), ); } final payload = AuthService.verifyToken(token); if (payload == null) { return Response.forbidden( body: Body.fromString('Invalid or expired token'), ); } final userId = payload['sub'] as String?; final username = payload['username'] as String?; if (userId == null || userId.isEmpty) { return Response.forbidden( body: Body.fromString('Invalid token payload'), ); } return WebSocketUpgrade((webSocket) { webSocket.pingInterval = const Duration(seconds: 30); final connection = _gatewayRegistry.register( socket: webSocket, userId: userId, username: username, ); _sendGatewayPayload( connection.id, { 'op': 'ready', 'connectionId': connection.id, 'userId': connection.userId, if (connection.username != null) 'username': connection.username, }, ); webSocket.events.listen( (event) { switch (event) { case TextDataReceived(text: final text): _handleGatewayTextMessage(connection.id, text); case BinaryDataReceived(): _sendProtocolError( connection.id, 'Binary payloads are not supported', ); case CloseReceived(): _gatewayRegistry.unregister(connection.id); } }, onError: (_, __) { _gatewayRegistry.unregister(connection.id); }, onDone: () { _gatewayRegistry.unregister(connection.id); }, ); }); } void broadcastMessageCreated({ required int channelId, required Map message, }) { _gatewayRegistry.broadcastToChannel(channelId, { 'op': 'event', 'event': 'message.created', 'channelId': channelId, 'message': message, 'emittedAt': DateTime.now().toUtc().toIso8601String(), }); } void broadcastMessageUpdated({ required int channelId, required Map message, }) { _gatewayRegistry.broadcastToChannel(channelId, { 'op': 'event', 'event': 'message.updated', 'channelId': channelId, 'message': message, 'emittedAt': DateTime.now().toUtc().toIso8601String(), }); } void broadcastMessageDeleted({ required int channelId, required int messageId, }) { _gatewayRegistry.broadcastToChannel(channelId, { 'op': 'event', 'event': 'message.deleted', 'channelId': channelId, 'messageId': messageId, 'emittedAt': DateTime.now().toUtc().toIso8601String(), }); } String? _extractGatewayToken(Request req) { final authHeader = req.headers.authorization; if (authHeader is BearerAuthorizationHeader) { final token = authHeader.token.trim(); if (token.isNotEmpty) { return token; } } final rawAuthorization = req.headers[Headers.authorizationHeader]; if (rawAuthorization != null && rawAuthorization.isNotEmpty) { final value = rawAuthorization.first.trim(); if (value.toLowerCase().startsWith('bearer ')) { final token = value.substring(7).trim(); if (token.isNotEmpty) { return token; } } } final queryToken = req.url.queryParameters['token']?.trim(); if (queryToken != null && queryToken.isNotEmpty) { return queryToken; } final queryAccessToken = req.url.queryParameters['access_token']?.trim(); if (queryAccessToken != null && queryAccessToken.isNotEmpty) { return queryAccessToken; } return null; } void _handleGatewayTextMessage(String connectionId, String text) { final connection = _gatewayRegistry.getConnection(connectionId); if (connection == null) { return; } final payload = _decodePayload(text); if (payload == null) { _sendProtocolError(connectionId, 'Invalid JSON payload'); return; } final rawOp = payload['op']; if (rawOp is! String || rawOp.trim().isEmpty) { _sendProtocolError(connectionId, '`op` is required'); return; } switch (rawOp.trim().toLowerCase()) { case 'subscribe': _handleSubscribe(connectionId, payload); case 'unsubscribe': _handleUnsubscribe(connectionId, payload); case 'ping': case 'heartbeat': _sendGatewayPayload(connectionId, { 'op': 'pong', 'ts': DateTime.now().toUtc().toIso8601String(), }); case 'pong': return; default: _sendProtocolError(connectionId, 'Unsupported op `$rawOp`'); } } void _handleSubscribe(String connectionId, Map payload) { final channelIds = _parseChannelIds(payload); if (channelIds.isEmpty) { _sendProtocolError( connectionId, 'subscribe requires `channelId` or `channelIds`', ); return; } final knownChannelIds = []; final unknownChannelIds = []; for (final channelId in channelIds) { if (channelBox.get(channelId) == null) { unknownChannelIds.add(channelId); } else { knownChannelIds.add(channelId); } } final subscribed = _gatewayRegistry.subscribe(connectionId, knownChannelIds); _sendGatewayPayload(connectionId, { 'op': 'subscribed', 'channelIds': subscribed, 'subscriptions': _gatewayRegistry.subscriptions(connectionId), if (unknownChannelIds.isNotEmpty) 'unknownChannelIds': unknownChannelIds, }); } void _handleUnsubscribe(String connectionId, Map payload) { final channelIds = _parseChannelIds(payload); if (channelIds.isEmpty) { _sendProtocolError( connectionId, 'unsubscribe requires `channelId` or `channelIds`', ); return; } final unsubscribed = _gatewayRegistry.unsubscribe(connectionId, channelIds); _sendGatewayPayload(connectionId, { 'op': 'unsubscribed', 'channelIds': unsubscribed, 'subscriptions': _gatewayRegistry.subscriptions(connectionId), }); } Map? _decodePayload(String text) { try { final decoded = jsonDecode(text); if (decoded is Map) { return decoded; } if (decoded is Map) { return decoded.map( (key, value) => MapEntry(key.toString(), value), ); } return null; } catch (_) { return null; } } Set _parseChannelIds(Map payload) { final channelIds = {}; void parseAndAdd(Object? value) { final parsed = switch (value) { int id when id > 0 => id, String id => int.tryParse(id), _ => null, }; if (parsed != null && parsed > 0) { channelIds.add(parsed); } } parseAndAdd(payload['channelId']); final channelIdList = payload['channelIds']; if (channelIdList is List) { for (final value in channelIdList) { parseAndAdd(value); } } return channelIds; } void _sendProtocolError(String connectionId, String message) { _sendGatewayPayload(connectionId, { 'op': 'error', 'message': message, }); } void _sendGatewayPayload(String connectionId, Map payload) { final connection = _gatewayRegistry.getConnection(connectionId); if (connection == null) { return; } final sent = _safeSend(connection, payload); if (!sent) { _gatewayRegistry.unregister(connectionId); } } bool _safeSend(_GatewayConnection connection, Map payload) { try { return connection.socket.trySendText(jsonEncode(payload)); } catch (_) { return false; } } class _GatewayRegistry { final Map _connections = {}; final Map> _subscribersByChannel = {}; int _nextConnectionId = 1; _GatewayConnection register({ required RelicWebSocket socket, required String userId, String? username, }) { final id = 'gw_${_nextConnectionId++}'; final connection = _GatewayConnection( id: id, socket: socket, userId: userId, username: username, ); _connections[id] = connection; return connection; } _GatewayConnection? getConnection(String connectionId) { return _connections[connectionId]; } List subscriptions(String connectionId) { final connection = _connections[connectionId]; if (connection == null) { return const []; } return connection.subscriptions.toList()..sort(); } List subscribe(String connectionId, Iterable channelIds) { final connection = _connections[connectionId]; if (connection == null) { return const []; } for (final channelId in channelIds) { connection.subscriptions.add(channelId); _subscribersByChannel.putIfAbsent(channelId, () => {}).add( connectionId, ); } return channelIds.toSet().toList()..sort(); } List unsubscribe(String connectionId, Iterable channelIds) { final connection = _connections[connectionId]; if (connection == null) { return const []; } final removed = []; for (final channelId in channelIds) { final didRemove = connection.subscriptions.remove(channelId); if (didRemove) { removed.add(channelId); } final subscribers = _subscribersByChannel[channelId]; if (subscribers != null) { subscribers.remove(connectionId); if (subscribers.isEmpty) { _subscribersByChannel.remove(channelId); } } } removed.sort(); return removed; } void unregister(String connectionId) { final connection = _connections.remove(connectionId); if (connection == null) { return; } for (final channelId in connection.subscriptions) { final subscribers = _subscribersByChannel[channelId]; if (subscribers == null) { continue; } subscribers.remove(connectionId); if (subscribers.isEmpty) { _subscribersByChannel.remove(channelId); } } if (!connection.socket.isClosed) { unawaited(connection.socket.tryClose(1000, 'Disconnected')); } } void broadcastToChannel(int channelId, Map payload) { final subscribers = _subscribersByChannel[channelId]; if (subscribers == null || subscribers.isEmpty) { return; } final deadConnections = []; final subscriberIds = subscribers.toList(growable: false); for (final connectionId in subscriberIds) { final connection = _connections[connectionId]; if (connection == null) { deadConnections.add(connectionId); continue; } final sent = _safeSend(connection, payload); if (!sent) { deadConnections.add(connectionId); } } for (final connectionId in deadConnections) { unregister(connectionId); } } } class _GatewayConnection { final String id; final RelicWebSocket socket; final String userId; final String? username; final Set subscriptions = {}; _GatewayConnection({ required this.id, required this.socket, required this.userId, required this.username, }); }