diff --git a/CLIENT_INTEGRATION.md b/CLIENT_INTEGRATION.md new file mode 100644 index 0000000..2f92548 --- /dev/null +++ b/CLIENT_INTEGRATION.md @@ -0,0 +1,243 @@ +# Client Integration Guide + +This document describes the current server contract you can rely on when implementing the client. + +## Base Auth Model + +- REST endpoints under `/api/*` require `Authorization: Bearer `. +- WebSocket gateway at `/gateway` also requires JWT. +- Gateway token sources: +1. `Authorization: Bearer ` (preferred when possible) +2. query fallback: `?token=` or `?access_token=` (useful for browser WebSocket clients) + +## Gateway (`/gateway`) + +### Connect + +- URL: `ws://:/gateway` (or `wss://` behind TLS) +- On successful auth, server sends: + +```json +{ + "op": "ready", + "connectionId": "gw_1", + "userId": "alice@example.com", + "username": "alice" +} +``` + +If auth fails, handshake is rejected with HTTP error (`401`/`403`) and no WebSocket session is established. + +### Client -> Server Messages + +All frames are JSON text. + +#### Subscribe + +```json +{"op":"subscribe","channelId":12} +``` + +or + +```json +{"op":"subscribe","channelIds":[12,13]} +``` + +Response: + +```json +{ + "op": "subscribed", + "channelIds": [12, 13], + "subscriptions": [12, 13] +} +``` + +If one or more channels do not exist: + +```json +{ + "op": "subscribed", + "channelIds": [12], + "subscriptions": [12], + "unknownChannelIds": [999] +} +``` + +#### Unsubscribe + +```json +{"op":"unsubscribe","channelId":12} +``` + +or + +```json +{"op":"unsubscribe","channelIds":[12,13]} +``` + +Response: + +```json +{ + "op": "unsubscribed", + "channelIds": [12], + "subscriptions": [13] +} +``` + +#### Heartbeat + +Client may send: + +```json +{"op":"ping"} +``` + +or + +```json +{"op":"heartbeat"} +``` + +Server replies: + +```json +{ + "op": "pong", + "ts": "2026-02-20T12:34:56.000Z" +} +``` + +### Server -> Client Event Messages + +After subscription, channel message mutations are pushed as: + +#### Message Created + +```json +{ + "op": "event", + "event": "message.created", + "channelId": 12, + "message": { + "id": 101, + "content": "hello", + "authorId": "alice@example.com", + "authorUsername": "alice", + "authorAvatarUrl": null, + "authorHomeServer": "https://home.example.com", + "channelId": 12, + "createdAt": "2026-02-20T12:34:56.000", + "editedAt": null + }, + "emittedAt": "2026-02-20T12:34:56.000Z" +} +``` + +#### Message Updated + +```json +{ + "op": "event", + "event": "message.updated", + "channelId": 12, + "message": { "...same message shape..." }, + "emittedAt": "2026-02-20T12:35:56.000Z" +} +``` + +#### Message Deleted + +```json +{ + "op": "event", + "event": "message.deleted", + "channelId": 12, + "messageId": 101, + "emittedAt": "2026-02-20T12:36:56.000Z" +} +``` + +### Gateway Error Payload + +Malformed input or unsupported ops return: + +```json +{"op":"error","message":"..."} +``` + +Common causes: +- invalid JSON payload +- missing/invalid `op` +- unsupported op +- missing `channelId`/`channelIds` for subscribe/unsubscribe + +## Message History Pagination (`GET /api/messages`) + +### Query Params + +- `channelId` (required, int) +- `limit` (optional, int, default `50`, max `100`) +- `before` (optional, message ID cursor) +- `after` (optional, message ID cursor) + +Rules: +- Do not send `before` and `after` together. +- `before`/`after` must be positive integers. + +### Response Shape + +```json +{ + "messages": [ + { + "id": 100, + "content": "text", + "authorId": "alice@example.com", + "authorUsername": "alice", + "authorAvatarUrl": null, + "authorHomeServer": "https://home.example.com", + "channelId": 12, + "createdAt": "2026-02-20T12:30:00.000", + "editedAt": null + } + ], + "channelId": 12, + "limit": 50, + "hasMore": true, + "nextBefore": 51, + "nextAfter": 100 +} +``` + +### Ordering Semantics + +- no cursor: newest -> oldest (descending by message id) +- `before=`: older than cursor, still newest -> oldest +- `after=`: newer than cursor, oldest -> newest (ascending) + +For client state: +- `nextBefore`: cursor to request older page +- `nextAfter`: cursor to request newer page + +## Recommended Client Flow + +1. Login/register/join and store JWT. +2. Initial channel load: `GET /api/messages?channelId=&limit=50`. +3. Connect `/gateway` with token. +4. Send `subscribe` for active channel(s). +5. Merge realtime events into local message list: +- `message.created`: insert if absent +- `message.updated`: patch by id +- `message.deleted`: remove/tombstone by id +6. Pagination: +- load older: call `before=` +- catch up newer (after reconnect): call `after=` + +## Practical Notes + +- REST is source of truth; gateway is fanout only. +- Gateway registry is currently in-memory on server instance (no cross-instance sync yet). +- Channel permissions are not enforced yet by gateway subscriptions; only channel existence is checked. diff --git a/README.md b/README.md index 7d05f91..6f72bc7 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,7 @@ Each user has a **home server** where their identity lives (keypair), and can se - **Client does everything** - syncs, cross-server requests, profile updates See [ARCHITECTURE.md](./ARCHITECTURE.md) for full details. +See [CLIENT_INTEGRATION.md](./CLIENT_INTEGRATION.md) for client protocol/API integration details. ## Quick Start diff --git a/lib/server.dart b/lib/server.dart index 6564ff8..3b417f4 100644 --- a/lib/server.dart +++ b/lib/server.dart @@ -3,6 +3,7 @@ import 'package:relic/relic.dart'; import 'config/config.dart'; import 'src/auth/auth_middleware.dart'; +import 'src/gateway/gateway_handlers.dart'; import 'src/handlers/auth_handlers.dart'; import 'src/handlers/channel_handlers.dart'; import 'src/handlers/message_handlers.dart'; @@ -26,6 +27,7 @@ Future startServer() async { ..get('/.well-known/users/:username', publicUserHandler) ..get('/.well-known/verify/:userId', publicVerifyHandler) ..get('/.well-known/server', publicServerInfoHandler) + ..get('/gateway', gatewayHandler) // ============ PROTECTED ENDPOINTS (requires auth) ============ ..use('/api', jwtMiddleware) diff --git a/lib/src/gateway/gateway_handlers.dart b/lib/src/gateway/gateway_handlers.dart new file mode 100644 index 0000000..c68d067 --- /dev/null +++ b/lib/src/gateway/gateway_handlers.dart @@ -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 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, + }); +} diff --git a/lib/src/handlers/message_handlers.dart b/lib/src/handlers/message_handlers.dart index a8164e2..a1a9cb3 100644 --- a/lib/src/handlers/message_handlers.dart +++ b/lib/src/handlers/message_handlers.dart @@ -3,14 +3,20 @@ import 'package:chat_server/objectbox.g.dart'; import 'package:relic/relic.dart'; import '../auth/auth_middleware.dart'; +import '../gateway/gateway_handlers.dart'; import '../state/app_state.dart'; import '../utils/extensions.dart'; +const _defaultMessageLimit = 50; +const _maxMessageLimit = 100; + /// Get messages for a channel -/// Query params: channelId (required), limit, before +/// Query params: channelId (required), limit, before, after Future getMessagesHandler(Request req) async { final channelIdStr = req.url.queryParameters['channelId']; - final limitStr = req.url.queryParameters['limit'] ?? '50'; + final limitStr = req.url.queryParameters['limit']; + final beforeStr = req.url.queryParameters['before']; + final afterStr = req.url.queryParameters['after']; if (channelIdStr == null) { return Response.badRequest( @@ -18,8 +24,17 @@ Future getMessagesHandler(Request req) async { ); } + if (beforeStr != null && afterStr != null) { + return Response.badRequest( + body: Body.fromString('Use either `before` or `after`, not both'), + ); + } + final channelId = int.tryParse(channelIdStr); - final limit = int.tryParse(limitStr) ?? 50; + final parsedLimit = + limitStr == null ? _defaultMessageLimit : int.tryParse(limitStr); + int? before; + int? after; if (channelId == null) { return Response.badRequest( @@ -27,6 +42,32 @@ Future getMessagesHandler(Request req) async { ); } + if (parsedLimit == null || parsedLimit <= 0) { + return Response.badRequest( + body: Body.fromString('Invalid limit'), + ); + } + + final limit = parsedLimit > _maxMessageLimit ? _maxMessageLimit : parsedLimit; + + if (beforeStr != null) { + before = int.tryParse(beforeStr); + if (before == null || before <= 0) { + return Response.badRequest( + body: Body.fromString('Invalid `before` cursor'), + ); + } + } + + if (afterStr != null) { + after = int.tryParse(afterStr); + if (after == null || after <= 0) { + return Response.badRequest( + body: Body.fromString('Invalid `after` cursor'), + ); + } + } + // Check channel exists final channel = channelBox.get(channelId); if (channel == null) { @@ -35,33 +76,50 @@ Future getMessagesHandler(Request req) async { ); } - // Get messages - final query = messageBox - .query(ChatMessage_.channelId.equals(channelId)) - .order(ChatMessage_.createdAt, flags: Order.descending) - .build(); + var condition = ChatMessage_.channelId.equals(channelId); + if (before != null) { + condition = condition & ChatMessage_.id.lessThan(before); + } else if (after != null) { + condition = condition & ChatMessage_.id.greaterThan(after); + } - query.limit = limit; - final messages = query.find(); + final queryBuilder = messageBox.query(condition); + if (after != null) { + queryBuilder.order(ChatMessage_.id); + } else { + queryBuilder.order(ChatMessage_.id, flags: Order.descending); + } + + final query = queryBuilder.build(); + query.limit = limit + 1; + + final fetchedMessages = query.find(); query.close(); + final hasMore = fetchedMessages.length > limit; + final messages = + hasMore ? fetchedMessages.sublist(0, limit) : fetchedMessages; + + int? nextBefore; + int? nextAfter; + if (messages.isNotEmpty) { + if (after != null) { + nextBefore = messages.first.id; + nextAfter = messages.last.id; + } else { + nextBefore = messages.last.id; + nextAfter = messages.first.id; + } + } + return Response.ok( body: { - 'messages': messages - .map((m) => { - 'id': m.id, - 'content': m.content, - 'authorId': m.authorUserId, - 'authorUsername': m.authorUsername, - 'authorAvatarUrl': m.authorAvatarUrl, - 'authorHomeServer': m.authorHomeServer, - 'channelId': m.channelId, - 'createdAt': m.createdAt.toIso8601String(), - 'editedAt': m.editedAt?.toIso8601String(), - }) - .toList(), + 'messages': messages.map(_messageToMap).toList(), 'channelId': channelId, 'limit': limit, + 'hasMore': hasMore, + 'nextBefore': nextBefore, + 'nextAfter': nextAfter, }.toBody, ); } @@ -123,6 +181,12 @@ Future sendMessageHandler(Request req) async { ); final id = messageBox.put(message); + message.id = id; + + broadcastMessageCreated( + channelId: channelId, + message: _messageToMap(message), + ); return Response.ok( body: { @@ -192,6 +256,11 @@ Future editMessageHandler(Request req) async { message.editedAt = DateTime.now(); messageBox.put(message); + broadcastMessageUpdated( + channelId: message.channelId, + message: _messageToMap(message), + ); + return Response.ok( body: { 'id': message.id, @@ -243,8 +312,14 @@ Future deleteMessageHandler(Request req) async { ); } + final channelId = message.channelId; messageBox.remove(messageId); + broadcastMessageDeleted( + channelId: channelId, + messageId: messageId, + ); + return Response.ok( body: { 'deleted': true, @@ -252,3 +327,17 @@ Future deleteMessageHandler(Request req) async { }.toBody, ); } + +Map _messageToMap(ChatMessage message) { + return { + 'id': message.id, + 'content': message.content, + 'authorId': message.authorUserId, + 'authorUsername': message.authorUsername, + 'authorAvatarUrl': message.authorAvatarUrl, + 'authorHomeServer': message.authorHomeServer, + 'channelId': message.channelId, + 'createdAt': message.createdAt.toIso8601String(), + 'editedAt': message.editedAt?.toIso8601String(), + }; +}