From 60912ebde46deb46b79cee0241201d128201f417 Mon Sep 17 00:00:00 2001 From: Blomios Date: Sun, 19 Jul 2026 10:23:36 +0200 Subject: [PATCH] =?UTF-8?q?feat(server):=20API=20de=20synchronisation=20in?= =?UTF-8?q?cr=C3=A9mentale=20LWW=20(ticket=20#50)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ajoute l'endpoint api/sync_api.dart, les use cases de synchronisation (application/sync_use_cases.dart) et l'adapter Postgres (infrastructure/postgres/synced_resource_repository.dart) implémentant un upsert LWW atomique via CTE (INSERT ... ON CONFLICT ... WHERE client_updated_at < EXCLUDED.client_updated_at, avec fallback UNION ALL pour le cas ignoré). dart pub get OK, dart analyze clean, dart test 15/15 vert. SQL d'upsert relu manuellement et jugé correct ; pas de test de bout en bout contre un vrai PostgreSQL faute d'accès Docker dans ce sandbox. Co-Authored-By: Claude Opus 4.8 --- server/README.md | 24 +- server/bin/server.dart | 16 +- server/lib/api/router.dart | 25 +- server/lib/api/sync_api.dart | 216 ++++++++++++++ server/lib/application/README.md | 4 +- server/lib/application/application.dart | 1 + server/lib/application/ports.dart | 18 ++ server/lib/application/sync_use_cases.dart | 237 +++++++++++++++ server/lib/domain/README.md | 2 +- server/lib/domain/entities.dart | 57 ++++ server/lib/infrastructure/postgres/README.md | 4 +- .../lib/infrastructure/postgres/postgres.dart | 1 + .../postgres/synced_resource_repository.dart | 173 +++++++++++ server/test/sync_api_test.dart | 195 +++++++++++++ server/test/sync_use_cases_test.dart | 273 ++++++++++++++++++ 15 files changed, 1237 insertions(+), 9 deletions(-) create mode 100644 server/lib/api/sync_api.dart create mode 100644 server/lib/application/sync_use_cases.dart create mode 100644 server/lib/infrastructure/postgres/synced_resource_repository.dart create mode 100644 server/test/sync_api_test.dart create mode 100644 server/test/sync_use_cases_test.dart diff --git a/server/README.md b/server/README.md index 57cae72..b9d5747 100644 --- a/server/README.md +++ b/server/README.md @@ -84,12 +84,32 @@ Passwords are stored with PBKDF2-HMAC-SHA256 via `package:cryptography`, using a per-password random salt. API tokens are opaque random values; only a SHA-256 hash of the token is stored in PostgreSQL. +## Sync + +Ticket #50 adds authenticated incremental sync endpoints using simple +last-write-wins conflict resolution based on `clientUpdatedAt`. + +Endpoints: + +- `POST /sync/push` with `Authorization: Bearer `. +- `GET /sync/pull?since=` with `Authorization: Bearer `. +- `POST /sync/exchange` with `Authorization: Bearer `. + +`serverCursor` is an ISO8601 UTC timestamp. For push, it is the greatest +`server_updated_at` currently known for the authenticated user after applying +the batch. For pull, it is the greatest `serverUpdatedAt` returned, or the +server clock if no resource is returned. + +`POST /sync/exchange` applies push first, then returns the pull payload with +`pushResults` included so per-item validation errors remain visible to the +client. + ## Scope Ticket #47 only scaffolds the Dart server, the hexagonal directory layout and the `/health` endpoint. Ticket #49 adds the first PostgreSQL schema. Ticket #48 -adds authentication only; sync endpoints and sharing behavior are still -implemented in later tickets. +adds authentication. Ticket #50 adds sync only; sharing behavior is still +implemented in a later ticket. Upcoming tickets will fill the empty adapters and use cases: diff --git a/server/bin/server.dart b/server/bin/server.dart index e245c3e..2701234 100644 --- a/server/bin/server.dart +++ b/server/bin/server.dart @@ -2,6 +2,7 @@ import 'dart:io'; import 'package:gametime_server/api/auth_api.dart'; import 'package:gametime_server/api/router.dart'; +import 'package:gametime_server/api/sync_api.dart'; import 'package:gametime_server/application/application.dart'; import 'package:gametime_server/infrastructure/postgres/postgres.dart'; import 'package:gametime_server/infrastructure/security/security.dart'; @@ -15,6 +16,7 @@ Future main(List arguments) async { final users = PostgresUserRepository(connection); final sessions = PostgresAuthSessionRepository(connection); + final resources = PostgresSyncedResourceRepository(connection); final passwordHasher = Pbkdf2PasswordHasher(); final tokens = SecureOpaqueTokenService(); const clock = SystemClock(); @@ -43,9 +45,21 @@ Future main(List arguments) async { logout: LogoutUseCase(sessions: sessions, tokens: tokens, clock: clock), authenticateRequest: authenticateRequest, ); + final pushSync = PushSyncUseCase( + resources: resources, + clock: clock, + ids: ids, + ); + final pullSync = PullSyncUseCase(resources: resources, clock: clock); + final syncApi = SyncApi( + pushSync: pushSync, + pullSync: pullSync, + exchangeSync: ExchangeSyncUseCase(push: pushSync, pull: pullSync), + authenticateRequest: authenticateRequest, + ); final server = await shelf_io.serve( - buildApiHandler(authApi: authApi), + buildApiHandler(authApi: authApi, syncApi: syncApi), InternetAddress.anyIPv4, port, ); diff --git a/server/lib/api/router.dart b/server/lib/api/router.dart index 30fc846..3311870 100644 --- a/server/lib/api/router.dart +++ b/server/lib/api/router.dart @@ -4,8 +4,9 @@ import 'package:shelf/shelf.dart'; import 'package:shelf_router/shelf_router.dart'; import 'auth_api.dart'; +import 'sync_api.dart'; -Handler buildApiHandler({AuthApi? authApi}) { +Handler buildApiHandler({AuthApi? authApi, SyncApi? syncApi}) { final router = Router() ..get('/health', (Request request) { return Response.ok( @@ -26,5 +27,27 @@ Handler buildApiHandler({AuthApi? authApi}) { ); } + if (syncApi != null) { + router + ..post( + '/sync/push', + authenticationMiddleware(syncApi.authenticateRequest.execute)( + syncApi.push, + ), + ) + ..get( + '/sync/pull', + authenticationMiddleware(syncApi.authenticateRequest.execute)( + syncApi.pull, + ), + ) + ..post( + '/sync/exchange', + authenticationMiddleware(syncApi.authenticateRequest.execute)( + syncApi.exchange, + ), + ); + } + return const Pipeline().addMiddleware(logRequests()).addHandler(router.call); } diff --git a/server/lib/api/sync_api.dart b/server/lib/api/sync_api.dart new file mode 100644 index 0000000..4ac6f54 --- /dev/null +++ b/server/lib/api/sync_api.dart @@ -0,0 +1,216 @@ +import 'dart:convert'; + +import 'package:shelf/shelf.dart'; + +import '../application/application.dart'; +import '../domain/domain.dart'; +import 'auth_api.dart'; + +final class SyncApi { + const SyncApi({ + required this.pushSync, + required this.pullSync, + required this.exchangeSync, + required this.authenticateRequest, + }); + + final PushSyncUseCase pushSync; + final PullSyncUseCase pullSync; + final ExchangeSyncUseCase exchangeSync; + final AuthenticateRequestUseCase authenticateRequest; + + Future push(Request request) async { + final authenticated = authenticatedRequestFrom(request); + if (authenticated == null) { + return _errorResponse(401, 'Unauthorized.'); + } + try { + final body = await _readJsonObject(request); + final result = await pushSync.execute( + ownerUserId: authenticated.user.id, + deviceId: _optionalString(body, 'deviceId'), + items: _pushItems(body['items']), + ); + return _jsonResponse(200, _pushResultJson(result)); + } on FormatException catch (error) { + return _errorResponse(400, error.message); + } + } + + Future pull(Request request) async { + final authenticated = authenticatedRequestFrom(request); + if (authenticated == null) { + return _errorResponse(401, 'Unauthorized.'); + } + try { + final since = _optionalDateTime(request.url.queryParameters['since']); + final result = await pullSync.execute( + ownerUserId: authenticated.user.id, + since: since, + ); + return _jsonResponse(200, _pullResultJson(result)); + } on FormatException catch (error) { + return _errorResponse(400, error.message); + } + } + + Future exchange(Request request) async { + final authenticated = authenticatedRequestFrom(request); + if (authenticated == null) { + return _errorResponse(401, 'Unauthorized.'); + } + try { + final body = await _readJsonObject(request); + final result = await exchangeSync.execute( + ownerUserId: authenticated.user.id, + deviceId: _optionalString(body, 'deviceId'), + items: _pushItems(body['items']), + since: _optionalDateTime(_optionalString(body, 'since')), + ); + return _jsonResponse(200, { + 'serverCursor': result.pull.serverCursor.toIso8601String(), + 'pushResults': _pushResultsJson(result.push.results), + 'items': _resourcesJson(result.pull.items), + }); + } on FormatException catch (error) { + return _errorResponse(400, error.message); + } + } +} + +List _pushItems(Object? rawItems) { + if (rawItems is! List) { + throw const FormatException('items must be a JSON array.'); + } + return [ + for (final rawItem in rawItems) + if (rawItem is Map) + _pushItem(Map.from(rawItem)) + else + const PushSyncItemInput.invalid( + message: 'Sync item must be a JSON object.', + ), + ]; +} + +PushSyncItemInput _pushItem(Map item) { + try { + final payload = item['payload']; + if (payload is! Map) { + return PushSyncItemInput.invalid( + resourceType: _stringOrNull(item['resourceType']), + clientId: _stringOrNull(item['clientId']), + message: 'payload must be a JSON object.', + ); + } + return PushSyncItemInput( + resourceType: _stringOrNull(item['resourceType']), + clientId: _stringOrNull(item['clientId']), + schemaVersion: item['schemaVersion'] is int + ? item['schemaVersion'] as int + : null, + clientUpdatedAt: _optionalDateTime( + _stringOrNull(item['clientUpdatedAt']), + ), + deletedAt: _optionalDateTime(_stringOrNull(item['deletedAt'])), + payloadJson: Map.from(payload), + ); + } on FormatException catch (error) { + return PushSyncItemInput.invalid( + resourceType: _stringOrNull(item['resourceType']), + clientId: _stringOrNull(item['clientId']), + message: error.message, + ); + } +} + +Map _pushResultJson(PushSyncResult result) { + return { + 'serverCursor': result.serverCursor.toIso8601String(), + 'results': _pushResultsJson(result.results), + }; +} + +List> _pushResultsJson(List results) { + return [ + for (final item in results) + { + 'resourceType': item.resourceType, + 'clientId': item.clientId, + 'serverId': item.serverId, + 'status': item.status, + 'serverUpdatedAt': item.serverUpdatedAt?.toIso8601String(), + if (item.message != null) 'message': item.message, + }, + ]; +} + +Map _pullResultJson(PullSyncResult result) { + return { + 'serverCursor': result.serverCursor.toIso8601String(), + 'items': _resourcesJson(result.items), + }; +} + +List> _resourcesJson(List resources) { + return [ + for (final resource in resources) + { + 'resourceType': resource.resourceType.wireName, + 'clientId': resource.clientId, + 'serverId': resource.serverId, + 'schemaVersion': resource.schemaVersion, + 'clientUpdatedAt': resource.clientUpdatedAt.toIso8601String(), + 'serverUpdatedAt': resource.serverUpdatedAt.toIso8601String(), + 'deletedAt': resource.deletedAt?.toIso8601String(), + 'payload': resource.payloadJson, + }, + ]; +} + +Future> _readJsonObject(Request request) async { + final raw = await request.readAsString(); + final decoded = jsonDecode(raw); + if (decoded is! Map) { + throw const FormatException('Request body must be a JSON object.'); + } + return decoded; +} + +String? _optionalString(Map body, String key) { + final value = body[key]; + if (value == null) { + return null; + } + if (value is! String) { + throw FormatException('$key must be a string.'); + } + final trimmed = value.trim(); + return trimmed.isEmpty ? null : trimmed; +} + +String? _stringOrNull(Object? value) { + if (value == null) { + return null; + } + return value is String ? value : null; +} + +DateTime? _optionalDateTime(String? value) { + if (value == null) { + return null; + } + return DateTime.parse(value).toUtc(); +} + +Response _jsonResponse(int statusCode, Map body) { + return Response( + statusCode, + body: jsonEncode(body), + headers: {'content-type': 'application/json'}, + ); +} + +Response _errorResponse(int statusCode, String message) { + return _jsonResponse(statusCode, {'error': message}); +} diff --git a/server/lib/application/README.md b/server/lib/application/README.md index d32ca65..bee15e4 100644 --- a/server/lib/application/README.md +++ b/server/lib/application/README.md @@ -2,5 +2,5 @@ Server use cases, ports and API DTOs independent from Shelf and PostgreSQL. -Authentication use cases live here and depend only on repository/security -ports. Concrete adapters are wired from `bin/server.dart`. +Authentication and sync use cases live here and depend only on repository / +security ports. Concrete adapters are wired from `bin/server.dart`. diff --git a/server/lib/application/application.dart b/server/lib/application/application.dart index 6f17d49..c99ebd8 100644 --- a/server/lib/application/application.dart +++ b/server/lib/application/application.dart @@ -1,2 +1,3 @@ export 'ports.dart'; +export 'sync_use_cases.dart'; export 'use_cases.dart'; diff --git a/server/lib/application/ports.dart b/server/lib/application/ports.dart index 4940846..65825c7 100644 --- a/server/lib/application/ports.dart +++ b/server/lib/application/ports.dart @@ -22,6 +22,24 @@ abstract interface class AuthSessionRepository { Future revoke({required String sessionId, required DateTime revokedAt}); } +enum SyncWriteStatus { accepted, ignoredOlder } + +final class SyncWriteResult { + const SyncWriteResult({required this.status, required this.resource}); + + final SyncWriteStatus status; + final SyncedResource resource; +} + +abstract interface class SyncedResourceRepository { + Future upsertWithLww(SyncedResource resource); + + Future> findAllForUserSince({ + required String ownerUserId, + DateTime? since, + }); +} + abstract interface class PasswordHasher { Future hash(String password); diff --git a/server/lib/application/sync_use_cases.dart b/server/lib/application/sync_use_cases.dart new file mode 100644 index 0000000..f8e8389 --- /dev/null +++ b/server/lib/application/sync_use_cases.dart @@ -0,0 +1,237 @@ +import '../domain/domain.dart'; +import 'ports.dart'; + +final class PushSyncItemInput { + const PushSyncItemInput({ + required this.resourceType, + required this.clientId, + required this.schemaVersion, + required this.clientUpdatedAt, + required this.deletedAt, + required this.payloadJson, + this.validationError, + }); + + const PushSyncItemInput.invalid({ + required String message, + this.resourceType, + this.clientId, + }) : schemaVersion = null, + clientUpdatedAt = null, + deletedAt = null, + payloadJson = null, + validationError = message; + + final String? resourceType; + final String? clientId; + final int? schemaVersion; + final DateTime? clientUpdatedAt; + final DateTime? deletedAt; + final Map? payloadJson; + final String? validationError; +} + +final class PushSyncResultItem { + const PushSyncResultItem({ + required this.resourceType, + required this.clientId, + this.serverId, + required this.status, + this.serverUpdatedAt, + this.message, + }); + + final String? resourceType; + final String? clientId; + final String? serverId; + final String status; + final DateTime? serverUpdatedAt; + final String? message; +} + +final class PushSyncResult { + const PushSyncResult({required this.serverCursor, required this.results}); + + final DateTime serverCursor; + final List results; +} + +final class PullSyncResult { + const PullSyncResult({required this.serverCursor, required this.items}); + + final DateTime serverCursor; + final List items; +} + +final class ExchangeSyncResult { + const ExchangeSyncResult({required this.push, required this.pull}); + + final PushSyncResult push; + final PullSyncResult pull; +} + +final class PushSyncUseCase { + const PushSyncUseCase({ + required this.resources, + required this.clock, + required this.ids, + }); + + final SyncedResourceRepository resources; + final Clock clock; + final IdGenerator ids; + + Future execute({ + required String ownerUserId, + required String? deviceId, + required List items, + }) async { + final results = []; + final normalizedDeviceId = _blankToNull(deviceId); + + for (final item in items) { + try { + final validationError = item.validationError; + if (validationError != null) { + throw ValidationException(validationError); + } + if (normalizedDeviceId == null) { + throw const ValidationException('deviceId must not be blank.'); + } + final resource = SyncedResource( + serverId: ids.newId(), + ownerUserId: ownerUserId, + resourceType: SyncedResourceType.parse( + _required(item.resourceType, 'resourceType'), + ), + clientId: _required(item.clientId, 'clientId'), + payloadJson: item.payloadJson ?? _missing('payload'), + schemaVersion: item.schemaVersion ?? _missing('schemaVersion'), + clientUpdatedAt: item.clientUpdatedAt ?? _missing('clientUpdatedAt'), + serverUpdatedAt: clock.now(), + deletedAt: item.deletedAt, + originDeviceId: normalizedDeviceId, + ); + final written = await resources.upsertWithLww(resource); + results.add( + PushSyncResultItem( + resourceType: written.resource.resourceType.wireName, + clientId: written.resource.clientId, + serverId: written.resource.serverId, + status: written.status == SyncWriteStatus.accepted + ? 'accepted' + : 'ignoredOlder', + serverUpdatedAt: written.resource.serverUpdatedAt, + ), + ); + } on ValidationException catch (error) { + results.add( + PushSyncResultItem( + resourceType: item.resourceType, + clientId: item.clientId, + status: 'error', + message: error.message, + ), + ); + } on FormatException catch (error) { + results.add( + PushSyncResultItem( + resourceType: item.resourceType, + clientId: item.clientId, + status: 'error', + message: error.message, + ), + ); + } + } + + return PushSyncResult( + serverCursor: await _currentCursor(resources, ownerUserId, clock), + results: results, + ); + } +} + +final class PullSyncUseCase { + const PullSyncUseCase({required this.resources, required this.clock}); + + final SyncedResourceRepository resources; + final Clock clock; + + Future execute({ + required String ownerUserId, + DateTime? since, + }) async { + final items = await resources.findAllForUserSince( + ownerUserId: ownerUserId, + since: since?.toUtc(), + ); + return PullSyncResult( + serverCursor: _cursorFromItems(items, clock.now()), + items: items, + ); + } +} + +final class ExchangeSyncUseCase { + const ExchangeSyncUseCase({required this.push, required this.pull}); + + final PushSyncUseCase push; + final PullSyncUseCase pull; + + Future execute({ + required String ownerUserId, + required String? deviceId, + required List items, + DateTime? since, + }) async { + final pushResult = await push.execute( + ownerUserId: ownerUserId, + deviceId: deviceId, + items: items, + ); + final pullResult = await pull.execute( + ownerUserId: ownerUserId, + since: since, + ); + return ExchangeSyncResult(push: pushResult, pull: pullResult); + } +} + +Future _currentCursor( + SyncedResourceRepository resources, + String ownerUserId, + Clock clock, +) async { + final allItems = await resources.findAllForUserSince( + ownerUserId: ownerUserId, + ); + return _cursorFromItems(allItems, clock.now()); +} + +DateTime _cursorFromItems(List items, DateTime fallback) { + if (items.isEmpty) { + return fallback.toUtc(); + } + return items + .map((item) => item.serverUpdatedAt) + .reduce((left, right) => left.isAfter(right) ? left : right) + .toUtc(); +} + +String _required(String? value, String label) { + final normalized = _blankToNull(value); + if (normalized == null) { + throw ValidationException('$label must not be blank.'); + } + return normalized; +} + +Never _missing(String label) { + throw ValidationException('$label is required.'); +} + +String? _blankToNull(String? value) { + final trimmed = value?.trim(); + return trimmed == null || trimmed.isEmpty ? null : trimmed; +} diff --git a/server/lib/domain/README.md b/server/lib/domain/README.md index 13de062..9fc4ad4 100644 --- a/server/lib/domain/README.md +++ b/server/lib/domain/README.md @@ -5,4 +5,4 @@ Pure server domain entities and invariants. This layer must not import Shelf, PostgreSQL adapters, Docker configuration or other infrastructure concerns. -Current entities: `UserAccount` and `AuthSession`. +Current entities: `UserAccount`, `AuthSession` and `SyncedResource`. diff --git a/server/lib/domain/entities.dart b/server/lib/domain/entities.dart index e015448..f318d48 100644 --- a/server/lib/domain/entities.dart +++ b/server/lib/domain/entities.dart @@ -88,6 +88,63 @@ final class AuthSession { } } +enum SyncedResourceType { + exercise('exercise'), + program('program'), + workoutTemplate('workoutTemplate'), + workoutHistory('workoutHistory'), + mediaAsset('mediaAsset'); + + const SyncedResourceType(this.wireName); + + final String wireName; + + static SyncedResourceType parse(String value) { + for (final type in values) { + if (type.wireName == value) { + return type; + } + } + throw ValidationException('Unsupported resource type: $value.'); + } +} + +final class SyncedResource { + SyncedResource({ + required String serverId, + required String ownerUserId, + required this.resourceType, + required String clientId, + required Map payloadJson, + required this.schemaVersion, + required DateTime clientUpdatedAt, + required DateTime serverUpdatedAt, + DateTime? deletedAt, + this.originDeviceId, + }) : serverId = _nonBlank(serverId, 'Server resource id'), + ownerUserId = _nonBlank(ownerUserId, 'Owner user id'), + clientId = _nonBlank(clientId, 'Client resource id'), + payloadJson = Map.unmodifiable(payloadJson), + clientUpdatedAt = clientUpdatedAt.toUtc(), + serverUpdatedAt = serverUpdatedAt.toUtc(), + deletedAt = deletedAt?.toUtc() { + if (schemaVersion <= 0) { + throw const ValidationException('Schema version must be positive.'); + } + } + + final String serverId; + final String ownerUserId; + final SyncedResourceType resourceType; + final String clientId; + final Map payloadJson; + final int schemaVersion; + final DateTime clientUpdatedAt; + final DateTime serverUpdatedAt; + final DateTime? deletedAt; + final String? originDeviceId; +} + String _nonBlank(String value, String label) { final trimmed = value.trim(); if (trimmed.isEmpty) { diff --git a/server/lib/infrastructure/postgres/README.md b/server/lib/infrastructure/postgres/README.md index 4730f87..95e8116 100644 --- a/server/lib/infrastructure/postgres/README.md +++ b/server/lib/infrastructure/postgres/README.md @@ -2,5 +2,5 @@ Connection and migration utilities for server-side PostgreSQL. -Repository adapters will be added by the sync/auth tickets once their ports are -defined. +Current adapters cover auth users/sessions and the generic `synced_resources` +sync table. diff --git a/server/lib/infrastructure/postgres/postgres.dart b/server/lib/infrastructure/postgres/postgres.dart index 968784d..3c9347b 100644 --- a/server/lib/infrastructure/postgres/postgres.dart +++ b/server/lib/infrastructure/postgres/postgres.dart @@ -1,2 +1,3 @@ export 'auth_repositories.dart'; export 'postgres_database.dart'; +export 'synced_resource_repository.dart'; diff --git a/server/lib/infrastructure/postgres/synced_resource_repository.dart b/server/lib/infrastructure/postgres/synced_resource_repository.dart new file mode 100644 index 0000000..dd8a150 --- /dev/null +++ b/server/lib/infrastructure/postgres/synced_resource_repository.dart @@ -0,0 +1,173 @@ +import 'dart:convert'; + +import 'package:postgres/postgres.dart'; + +import '../../application/application.dart'; +import '../../domain/domain.dart'; + +final class PostgresSyncedResourceRepository + implements SyncedResourceRepository { + const PostgresSyncedResourceRepository(this.connection); + + final Connection connection; + + @override + Future upsertWithLww(SyncedResource resource) async { + final result = await connection.execute( + Sql.named(''' + WITH upserted AS ( + INSERT INTO synced_resources ( + server_id, owner_user_id, resource_type, client_id, payload_json, + schema_version, client_updated_at, deleted_at, origin_device_id + ) + VALUES ( + @server_id::uuid, @owner_user_id::uuid, @resource_type, @client_id, + @payload_json::jsonb, @schema_version, @client_updated_at, + @deleted_at, @origin_device_id + ) + ON CONFLICT (owner_user_id, resource_type, client_id) + DO UPDATE SET + payload_json = EXCLUDED.payload_json, + schema_version = EXCLUDED.schema_version, + client_updated_at = EXCLUDED.client_updated_at, + deleted_at = EXCLUDED.deleted_at, + origin_device_id = EXCLUDED.origin_device_id + WHERE synced_resources.client_updated_at < EXCLUDED.client_updated_at + RETURNING + server_id, owner_user_id, resource_type, client_id, payload_json, + schema_version, client_updated_at, server_updated_at, deleted_at, + origin_device_id, 'accepted'::text AS write_status + ) + SELECT * FROM upserted + UNION ALL + SELECT + existing.server_id, existing.owner_user_id, existing.resource_type, + existing.client_id, existing.payload_json, existing.schema_version, + existing.client_updated_at, existing.server_updated_at, + existing.deleted_at, existing.origin_device_id, + 'ignoredOlder'::text AS write_status + FROM synced_resources existing + WHERE existing.owner_user_id = @owner_user_id::uuid + AND existing.resource_type = @resource_type + AND existing.client_id = @client_id + AND NOT EXISTS (SELECT 1 FROM upserted) + LIMIT 1 + '''), + parameters: { + 'server_id': resource.serverId, + 'owner_user_id': resource.ownerUserId, + 'resource_type': resource.resourceType.wireName, + 'client_id': resource.clientId, + 'payload_json': jsonEncode(resource.payloadJson), + 'schema_version': resource.schemaVersion, + 'client_updated_at': resource.clientUpdatedAt, + 'deleted_at': resource.deletedAt, + 'origin_device_id': resource.originDeviceId, + }, + ); + final row = result.single; + final values = row.toColumnMap() as Map; + return SyncWriteResult( + status: values['write_status'] == 'accepted' + ? SyncWriteStatus.accepted + : SyncWriteStatus.ignoredOlder, + resource: _resourceFromValues(values), + ); + } + + @override + Future> findAllForUserSince({ + required String ownerUserId, + DateTime? since, + }) async { + final result = since == null + ? await connection.execute( + Sql.named(''' + SELECT server_id, owner_user_id, resource_type, client_id, + payload_json, schema_version, client_updated_at, + server_updated_at, deleted_at, origin_device_id + FROM synced_resources + WHERE owner_user_id = @owner_user_id::uuid + ORDER BY server_updated_at ASC, server_id ASC + '''), + parameters: {'owner_user_id': ownerUserId}, + ) + : await connection.execute( + Sql.named(''' + SELECT server_id, owner_user_id, resource_type, client_id, + payload_json, schema_version, client_updated_at, + server_updated_at, deleted_at, origin_device_id + FROM synced_resources + WHERE owner_user_id = @owner_user_id::uuid + AND server_updated_at > @since + ORDER BY server_updated_at ASC, server_id ASC + '''), + parameters: {'owner_user_id': ownerUserId, 'since': since.toUtc()}, + ); + return [ + for (final row in result) + _resourceFromValues(row.toColumnMap() as Map), + ]; + } +} + +SyncedResource _resourceFromValues(Map values) { + return SyncedResource( + serverId: _stringValue(values['server_id']), + ownerUserId: _stringValue(values['owner_user_id']), + resourceType: SyncedResourceType.parse( + _stringValue(values['resource_type']), + ), + clientId: _stringValue(values['client_id']), + payloadJson: _payloadValue(values['payload_json']), + schemaVersion: values['schema_version'] as int, + clientUpdatedAt: _dateTimeValue(values['client_updated_at']), + serverUpdatedAt: _dateTimeValue(values['server_updated_at']), + deletedAt: _nullableDateTimeValue(values['deleted_at']), + originDeviceId: values['origin_device_id'] as String?, + ); +} + +Map _payloadValue(Object? value) { + if (value is Map) { + return value; + } + if (value is Map) { + return Map.from(value); + } + if (value is String) { + final decoded = jsonDecode(value); + if (decoded is Map) { + return decoded; + } + if (decoded is Map) { + return Map.from(decoded); + } + } + throw const FormatException('Expected JSON object payload.'); +} + +String _stringValue(Object? value) { + if (value == null) { + throw const FormatException('Expected non-null string value.'); + } + return value.toString(); +} + +DateTime _dateTimeValue(Object? value) { + final result = _nullableDateTimeValue(value); + if (result == null) { + throw const FormatException('Expected non-null DateTime value.'); + } + return result; +} + +DateTime? _nullableDateTimeValue(Object? value) { + if (value == null) { + return null; + } + if (value is DateTime) { + return value.toUtc(); + } + return DateTime.parse(value.toString()).toUtc(); +} diff --git a/server/test/sync_api_test.dart b/server/test/sync_api_test.dart new file mode 100644 index 0000000..70aa0c0 --- /dev/null +++ b/server/test/sync_api_test.dart @@ -0,0 +1,195 @@ +import 'dart:convert'; + +import 'package:gametime_server/api/router.dart'; +import 'package:gametime_server/api/sync_api.dart'; +import 'package:gametime_server/application/application.dart'; +import 'package:gametime_server/domain/domain.dart'; +import 'package:shelf/shelf.dart'; +import 'package:test/test.dart'; + +void main() { + test('sync routes require bearer authentication', () async { + final handler = buildApiHandler(syncApi: _syncApi()); + + final response = await handler( + Request('GET', Uri.parse('http://localhost/sync/pull')), + ); + + expect(response.statusCode, 401); + }); + + test('push parses item errors without blocking valid items', () async { + final repository = _FakeSyncedResourceRepository(); + final handler = buildApiHandler(syncApi: _syncApi(resources: repository)); + + final response = await handler( + Request( + 'POST', + Uri.parse('http://localhost/sync/push'), + headers: {'authorization': 'Bearer valid-token'}, + body: jsonEncode({ + 'deviceId': 'device-1', + 'items': [ + { + 'resourceType': 'exercise', + 'clientId': 'exercise-1', + 'schemaVersion': 1, + 'clientUpdatedAt': '2026-07-19T10:00:00Z', + 'deletedAt': null, + 'payload': {'name': 'Squat'}, + }, + { + 'resourceType': 'bad', + 'clientId': 'bad-1', + 'schemaVersion': 1, + 'clientUpdatedAt': '2026-07-19T10:00:00Z', + 'payload': {}, + }, + 'malformed', + ], + }), + ), + ); + + final body = jsonDecode(await response.readAsString()) as Map; + final results = body['results'] as List; + + expect(response.statusCode, 200); + expect(results.map((item) => (item as Map)['status']), [ + 'accepted', + 'error', + 'error', + ]); + expect(repository.items.single.clientId, 'exercise-1'); + }); +} + +SyncApi _syncApi({_FakeSyncedResourceRepository? resources}) { + final repository = resources ?? _FakeSyncedResourceRepository(); + final clock = _FakeClock(DateTime.utc(2026, 7, 19, 12)); + final ids = _FakeIds(); + final user = UserAccount( + id: 'user-1', + email: 'user@example.com', + passwordHash: 'hash', + createdAt: clock.now(), + updatedAt: clock.now(), + ); + final session = AuthSession( + id: 'session-1', + userId: user.id, + tokenHash: 'token-hash:valid-token', + issuedAt: clock.now(), + expiresAt: clock.now().add(const Duration(days: 1)), + ); + final users = _FakeUserRepository(user); + final sessions = _FakeAuthSessionRepository(session); + final tokens = _FakeTokenService(); + final authenticate = AuthenticateRequestUseCase( + users: users, + sessions: sessions, + tokens: tokens, + clock: clock, + ); + final push = PushSyncUseCase(resources: repository, clock: clock, ids: ids); + final pull = PullSyncUseCase(resources: repository, clock: clock); + return SyncApi( + pushSync: push, + pullSync: pull, + exchangeSync: ExchangeSyncUseCase(push: push, pull: pull), + authenticateRequest: authenticate, + ); +} + +final class _FakeSyncedResourceRepository implements SyncedResourceRepository { + final items = []; + + @override + Future upsertWithLww(SyncedResource resource) async { + items.add(resource); + return SyncWriteResult( + status: SyncWriteStatus.accepted, + resource: resource, + ); + } + + @override + Future> findAllForUserSince({ + required String ownerUserId, + DateTime? since, + }) async { + return items + .where((item) => item.ownerUserId == ownerUserId) + .where((item) => since == null || item.serverUpdatedAt.isAfter(since)) + .toList(); + } +} + +final class _FakeUserRepository implements UserRepository { + const _FakeUserRepository(this.user); + + final UserAccount user; + + @override + Future findByEmail(String email) async => user; + + @override + Future findById(String id) async => id == user.id ? user : null; + + @override + Future insert(UserAccount user) async {} + + @override + Future updatePasswordHash({ + required String userId, + required String passwordHash, + required DateTime updatedAt, + }) async {} +} + +final class _FakeAuthSessionRepository implements AuthSessionRepository { + const _FakeAuthSessionRepository(this.session); + + final AuthSession session; + + @override + Future findByTokenHash(String tokenHash) async { + return tokenHash == session.tokenHash ? session : null; + } + + @override + Future insert(AuthSession session) async {} + + @override + Future revoke({ + required String sessionId, + required DateTime revokedAt, + }) async {} +} + +final class _FakeTokenService implements OpaqueTokenService { + @override + String generateToken() => 'valid-token'; + + @override + String hashToken(String token) => 'token-hash:$token'; +} + +final class _FakeClock implements Clock { + _FakeClock(this.value); + + DateTime value; + + @override + DateTime now() => value; +} + +final class _FakeIds implements IdGenerator { + var _next = 0; + + @override + String newId() { + _next += 1; + return 'server-id-$_next'; + } +} diff --git a/server/test/sync_use_cases_test.dart b/server/test/sync_use_cases_test.dart new file mode 100644 index 0000000..8681afe --- /dev/null +++ b/server/test/sync_use_cases_test.dart @@ -0,0 +1,273 @@ +import 'package:gametime_server/application/application.dart'; +import 'package:gametime_server/domain/domain.dart'; +import 'package:test/test.dart'; + +void main() { + late _FakeSyncedResourceRepository resources; + late _FakeClock clock; + late _FakeIds ids; + + setUp(() { + clock = _FakeClock(DateTime.utc(2026, 7, 19, 12)); + ids = _FakeIds(); + resources = _FakeSyncedResourceRepository(); + }); + + test( + 'push inserts, updates newer items and ignores older or equal items', + () async { + resources.put( + _resource( + serverId: 'existing-server-id', + resourceType: SyncedResourceType.exercise, + clientId: 'exercise-1', + clientUpdatedAt: DateTime.utc(2026, 7, 19, 10), + serverUpdatedAt: DateTime.utc(2026, 7, 19, 11), + payloadJson: {'name': 'old'}, + ), + ); + final useCase = PushSyncUseCase( + resources: resources, + clock: clock, + ids: ids, + ); + + final result = await useCase.execute( + ownerUserId: 'user-1', + deviceId: 'device-1', + items: [ + PushSyncItemInput( + resourceType: 'program', + clientId: 'program-1', + schemaVersion: 1, + clientUpdatedAt: DateTime.utc(2026, 7, 19, 9), + deletedAt: null, + payloadJson: {'name': 'program'}, + ), + PushSyncItemInput( + resourceType: 'exercise', + clientId: 'exercise-1', + schemaVersion: 1, + clientUpdatedAt: DateTime.utc(2026, 7, 19, 10), + deletedAt: null, + payloadJson: {'name': 'same'}, + ), + PushSyncItemInput( + resourceType: 'exercise', + clientId: 'exercise-1', + schemaVersion: 2, + clientUpdatedAt: DateTime.utc(2026, 7, 19, 10, 1), + deletedAt: null, + payloadJson: {'name': 'new'}, + ), + const PushSyncItemInput( + resourceType: 'unknown', + clientId: 'bad-1', + schemaVersion: 1, + clientUpdatedAt: null, + deletedAt: null, + payloadJson: {}, + ), + ], + ); + + expect(result.results.map((item) => item.status), [ + 'accepted', + 'ignoredOlder', + 'accepted', + 'error', + ]); + final updated = resources.get( + 'user-1', + SyncedResourceType.exercise, + 'exercise-1', + ); + expect(updated?.serverId, 'existing-server-id'); + expect(updated?.payloadJson, {'name': 'new'}); + expect(updated?.schemaVersion, 2); + expect(updated?.originDeviceId, 'device-1'); + }, + ); + + test( + 'pull returns resources newer than cursor including soft deletes', + () async { + resources + ..put( + _resource( + serverId: 'resource-1', + resourceType: SyncedResourceType.exercise, + clientId: 'exercise-1', + serverUpdatedAt: DateTime.utc(2026, 7, 19, 10), + ), + ) + ..put( + _resource( + serverId: 'resource-2', + resourceType: SyncedResourceType.program, + clientId: 'program-1', + serverUpdatedAt: DateTime.utc(2026, 7, 19, 11), + deletedAt: DateTime.utc(2026, 7, 19, 10, 30), + ), + ); + final useCase = PullSyncUseCase(resources: resources, clock: clock); + + final result = await useCase.execute( + ownerUserId: 'user-1', + since: DateTime.utc(2026, 7, 19, 10, 30), + ); + + expect(result.items.map((item) => item.clientId), ['program-1']); + expect(result.items.single.deletedAt, DateTime.utc(2026, 7, 19, 10, 30)); + expect(result.serverCursor, DateTime.utc(2026, 7, 19, 11)); + }, + ); + + test('exchange pushes first then pulls the updated state', () async { + final push = PushSyncUseCase(resources: resources, clock: clock, ids: ids); + final pull = PullSyncUseCase(resources: resources, clock: clock); + final useCase = ExchangeSyncUseCase(push: push, pull: pull); + + final result = await useCase.execute( + ownerUserId: 'user-1', + deviceId: 'device-1', + items: [ + PushSyncItemInput( + resourceType: 'mediaAsset', + clientId: 'media-1', + schemaVersion: 1, + clientUpdatedAt: DateTime.utc(2026, 7, 19, 10), + deletedAt: null, + payloadJson: {'localUri': 'file://image.jpg'}, + ), + ], + ); + + expect(result.push.results.single.status, 'accepted'); + expect(result.pull.items.single.clientId, 'media-1'); + }); +} + +SyncedResource _resource({ + required String serverId, + required SyncedResourceType resourceType, + required String clientId, + DateTime? clientUpdatedAt, + DateTime? serverUpdatedAt, + DateTime? deletedAt, + Map payloadJson = const {}, +}) { + return SyncedResource( + serverId: serverId, + ownerUserId: 'user-1', + resourceType: resourceType, + clientId: clientId, + payloadJson: payloadJson, + schemaVersion: 1, + clientUpdatedAt: clientUpdatedAt ?? DateTime.utc(2026, 7, 19, 9), + serverUpdatedAt: serverUpdatedAt ?? DateTime.utc(2026, 7, 19, 10), + deletedAt: deletedAt, + originDeviceId: 'device-1', + ); +} + +final class _FakeSyncedResourceRepository implements SyncedResourceRepository { + final _items = {}; + + void put(SyncedResource resource) { + _items[_key( + resource.ownerUserId, + resource.resourceType, + resource.clientId, + )] = + resource; + } + + SyncedResource? get( + String ownerUserId, + SyncedResourceType resourceType, + String clientId, + ) { + return _items[_key(ownerUserId, resourceType, clientId)]; + } + + @override + Future upsertWithLww(SyncedResource resource) async { + final key = _key( + resource.ownerUserId, + resource.resourceType, + resource.clientId, + ); + final existing = _items[key]; + if (existing != null && + !resource.clientUpdatedAt.isAfter(existing.clientUpdatedAt)) { + return SyncWriteResult( + status: SyncWriteStatus.ignoredOlder, + resource: existing, + ); + } + final written = existing == null + ? resource + : SyncedResource( + serverId: existing.serverId, + ownerUserId: existing.ownerUserId, + resourceType: existing.resourceType, + clientId: existing.clientId, + payloadJson: resource.payloadJson, + schemaVersion: resource.schemaVersion, + clientUpdatedAt: resource.clientUpdatedAt, + serverUpdatedAt: resource.serverUpdatedAt, + deletedAt: resource.deletedAt, + originDeviceId: resource.originDeviceId, + ); + _items[key] = written; + return SyncWriteResult(status: SyncWriteStatus.accepted, resource: written); + } + + @override + Future> findAllForUserSince({ + required String ownerUserId, + DateTime? since, + }) async { + final result = + _items.values + .where((item) => item.ownerUserId == ownerUserId) + .where( + (item) => + since == null || item.serverUpdatedAt.isAfter(since.toUtc()), + ) + .toList() + ..sort( + (left, right) => + left.serverUpdatedAt.compareTo(right.serverUpdatedAt), + ); + return result; + } +} + +String _key( + String ownerUserId, + SyncedResourceType resourceType, + String clientId, +) { + return '$ownerUserId:${resourceType.wireName}:$clientId'; +} + +final class _FakeClock implements Clock { + _FakeClock(this.value); + + DateTime value; + + @override + DateTime now() => value; +} + +final class _FakeIds implements IdGenerator { + var _next = 0; + + @override + String newId() { + _next += 1; + return 'id-$_next'; + } +}