Add gateway realtime fanout and cursor pagination docs
This commit is contained in:
Foxyz committed
1 parent
4399eafc90
commit
ccf966ffc1
5 files changed
+800
-23
No files matched your search
@@ -0,0 +1,442 @@
|
||||
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<String, dynamic> 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<String, dynamic> 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<String, dynamic> payload) {
|
||||
final channelIds = _parseChannelIds(payload);
|
||||
if (channelIds.isEmpty) {
|
||||
_sendProtocolError(
|
||||
connectionId,
|
||||
'subscribe requires `channelId` or `channelIds`',
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
final knownChannelIds = <int>[];
|
||||
final unknownChannelIds = <int>[];
|
||||
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<String, dynamic> 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<String, dynamic>? _decodePayload(String text) {
|
||||
try {
|
||||
final decoded = jsonDecode(text);
|
||||
if (decoded is Map<String, dynamic>) {
|
||||
return decoded;
|
||||
}
|
||||
if (decoded is Map) {
|
||||
return decoded.map(
|
||||
(key, value) => MapEntry(key.toString(), value),
|
||||
);
|
||||
}
|
||||
return null;
|
||||
} catch (_) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
Set<int> _parseChannelIds(Map<String, dynamic> payload) {
|
||||
final channelIds = <int>{};
|
||||
|
||||
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<String, dynamic> 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<String, dynamic> payload) {
|
||||
try {
|
||||
return connection.socket.trySendText(jsonEncode(payload));
|
||||
} catch (_) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
class _GatewayRegistry {
|
||||
final Map<String, _GatewayConnection> _connections = {};
|
||||
final Map<int, Set<String>> _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<int> subscriptions(String connectionId) {
|
||||
final connection = _connections[connectionId];
|
||||
if (connection == null) {
|
||||
return const <int>[];
|
||||
}
|
||||
return connection.subscriptions.toList()..sort();
|
||||
}
|
||||
|
||||
List<int> subscribe(String connectionId, Iterable<int> channelIds) {
|
||||
final connection = _connections[connectionId];
|
||||
if (connection == null) {
|
||||
return const <int>[];
|
||||
}
|
||||
|
||||
for (final channelId in channelIds) {
|
||||
connection.subscriptions.add(channelId);
|
||||
_subscribersByChannel.putIfAbsent(channelId, () => <String>{}).add(
|
||||
connectionId,
|
||||
);
|
||||
}
|
||||
|
||||
return channelIds.toSet().toList()..sort();
|
||||
}
|
||||
|
||||
List<int> unsubscribe(String connectionId, Iterable<int> channelIds) {
|
||||
final connection = _connections[connectionId];
|
||||
if (connection == null) {
|
||||
return const <int>[];
|
||||
}
|
||||
|
||||
final removed = <int>[];
|
||||
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<String, dynamic> payload) {
|
||||
final subscribers = _subscribersByChannel[channelId];
|
||||
if (subscribers == null || subscribers.isEmpty) {
|
||||
return;
|
||||
}
|
||||
|
||||
final deadConnections = <String>[];
|
||||
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<int> subscriptions = <int>{};
|
||||
|
||||
_GatewayConnection({
|
||||
required this.id,
|
||||
required this.socket,
|
||||
required this.userId,
|
||||
required this.username,
|
||||
});
|
||||
}
|
||||
Reference in new issue
Block a user