Migrate from HTTP API to gRPC with cqrs_services proto

- Regenerate proto stubs from correct cqrs_services.proto file
- Update GrpcCqrsApiClient to use DynamicQueryService and CommandService
- Fix type conversions for deliveries, routes, and commands
- Convert proto Timestamp to ISO8601 strings for model compatibility
- Convert string amounts to doubles for delivery orders
- Use Int64 for delivery IDs in gRPC commands

Work in progress: Timestamp conversion extension needs finalization.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-01-20 15:10:07 -05:00
parent be9ff1b7b2
commit 46048307ea
12 changed files with 23370 additions and 3536 deletions
+186 -430
View File
@@ -1,8 +1,11 @@
import 'dart:async';
import 'package:fixnum/fixnum.dart';
import 'package:grpc/grpc.dart';
import 'package:protobuf/protobuf.dart' as $protobuf;
import '../generated/delivery_service.pbgrpc.dart';
import '../generated/cqrs_services.pb.dart' as $pb;
import '../generated/cqrs_services.pbgrpc.dart';
import '../models/delivery.dart';
import '../models/delivery_address.dart';
import '../models/delivery_contact.dart';
@@ -13,46 +16,20 @@ import '../services/auth_service.dart';
import 'grpc_config.dart';
import 'types.dart';
// ignore_for_file: unused_element
/// gRPC-based CQRS API client for Plan B Logistics.
///
/// This client wraps the generated DeliveryServiceClient with `Result<T>` error
/// handling patterns consistent with [CqrsApiClient]. It provides:
/// - Lazy channel initialization
/// - Authentication via Bearer token in gRPC metadata
/// - Automatic token refresh on UNAUTHENTICATED errors
/// - Proper gRPC error to [ApiError] mapping
///
/// Usage:
/// ```dart
/// final client = GrpcCqrsApiClient(
/// config: GrpcConfig.development,
/// authService: authService,
/// );
///
/// final result = await client.getDeliveryRoutes();
/// result.when(
/// success: (routes) => handleRoutes(routes),
/// onError: (error) => handleError(error),
/// );
/// ```
class GrpcCqrsApiClient {
final GrpcConfig config;
final AuthService? authService;
ClientChannel? _channel;
DeliveryServiceClient? _deliveryClient;
DynamicQueryServiceClient? _queryClient;
CommandServiceClient? _commandClient;
GrpcCqrsApiClient({
required this.config,
this.authService,
});
/// Returns the gRPC channel, creating it lazily if needed.
///
/// The channel is configured based on [config.useTls] for development
/// (insecure) vs production (TLS) environments.
ClientChannel get channel {
if (_channel == null) {
final credentials = config.useTls
@@ -72,17 +49,16 @@ class GrpcCqrsApiClient {
return _channel!;
}
/// Returns the DeliveryService client, creating it lazily if needed.
DeliveryServiceClient get deliveryClient {
_deliveryClient ??= DeliveryServiceClient(channel);
return _deliveryClient!;
DynamicQueryServiceClient get queryClient {
_queryClient ??= DynamicQueryServiceClient(channel);
return _queryClient!;
}
CommandServiceClient get commandClient {
_commandClient ??= CommandServiceClient(channel);
return _commandClient!;
}
/// Builds [CallOptions] with authentication metadata.
///
/// Includes Bearer token in metadata if [authService] is configured and
/// a valid token is available. Uses [AuthService.ensureValidToken] to
/// proactively refresh tokens that are expiring soon.
Future<CallOptions> _buildCallOptions() async {
final metadata = <String, String>{};
@@ -99,33 +75,6 @@ class GrpcCqrsApiClient {
);
}
/// Merges base [CallOptions] with additional options for a specific call.
CallOptions _mergeOptions(CallOptions base, CallOptions? additional) {
if (additional == null) return base;
return CallOptions(
metadata: {
...base.metadata,
...additional.metadata,
},
timeout: additional.timeout ?? base.timeout,
providers: [
...base.metadataProviders,
...additional.metadataProviders,
],
);
}
/// Maps gRPC [GrpcError] to [ApiError] for consistent error handling.
///
/// Maps common gRPC status codes to appropriate [ApiErrorType]:
/// - UNAUTHENTICATED (16) -> HTTP 401
/// - PERMISSION_DENIED (7) -> HTTP 403
/// - NOT_FOUND (5) -> HTTP 404
/// - INVALID_ARGUMENT (3) -> Validation error
/// - DEADLINE_EXCEEDED (4) -> Timeout
/// - UNAVAILABLE (14) -> Network error
/// - Other codes -> Unknown error
ApiError _mapGrpcError(GrpcError error) {
switch (error.code) {
case StatusCode.unauthenticated:
@@ -151,400 +100,207 @@ class GrpcCqrsApiClient {
case StatusCode.deadlineExceeded:
return ApiError.timeout();
case StatusCode.unavailable:
return ApiError.network(
error.message ?? 'Service unavailable',
);
case StatusCode.internal:
return ApiError.http(
statusCode: 500,
message: error.message ?? 'Internal server error',
);
return ApiError.network('Service unavailable');
default:
return ApiError.unknown(
error.message ?? 'gRPC error: ${error.codeName}',
error.message ?? 'Unknown error',
exception: error,
);
}
}
/// Executes a gRPC call with `Result<T>` error handling and authentication.
///
/// This is the core method that wraps gRPC calls with:
/// - Authentication token injection
/// - Automatic token refresh on UNAUTHENTICATED errors (single retry)
/// - gRPC error to ApiError mapping
/// - Timeout handling
///
/// [grpcCall] is the actual gRPC method invocation.
/// [isRetry] tracks whether this is a retry after token refresh.
Future<Result<T>> _executeWithAuth<T>(
Future<T> Function(CallOptions options) grpcCall, {
bool isRetry = false,
}) async {
try {
final options = await _buildCallOptions();
final result = await grpcCall(options);
return Result.success(result);
} on GrpcError catch (error) {
// Handle UNAUTHENTICATED by attempting token refresh (once)
if (error.code == StatusCode.unauthenticated &&
!isRetry &&
authService != null) {
final refreshResult = await authService!.refreshAccessToken();
return refreshResult.when(
success: (token) => _executeWithAuth(grpcCall, isRetry: true),
onError: (refreshError) => Result.error(_mapGrpcError(error)),
cancelled: () => Result.error(_mapGrpcError(error)),
);
}
return Result.error(_mapGrpcError(error));
} on TimeoutException {
return Result.error(ApiError.timeout());
} catch (e, stackTrace) {
return Result.error(
ApiError.unknown(
'gRPC call failed: ${e.toString()}',
exception: Exception(stackTrace.toString()),
),
);
}
}
/// Executes a gRPC call that returns void (for commands).
///
/// Similar to [_executeWithAuth] but for operations that don't return
/// meaningful data (commands that return success/failure).
Future<Result<void>> _executeCommandWithAuth(
Future<void> Function(CallOptions options) grpcCall, {
bool isRetry = false,
}) async {
try {
final options = await _buildCallOptions();
await grpcCall(options);
return Result.success(null);
} on GrpcError catch (error) {
if (error.code == StatusCode.unauthenticated &&
!isRetry &&
authService != null) {
final refreshResult = await authService!.refreshAccessToken();
return refreshResult.when(
success: (token) =>
_executeCommandWithAuth(grpcCall, isRetry: true),
onError: (refreshError) => Result.error(_mapGrpcError(error)),
cancelled: () => Result.error(_mapGrpcError(error)),
);
}
return Result.error(_mapGrpcError(error));
} on TimeoutException {
return Result.error(ApiError.timeout());
} catch (e, stackTrace) {
return Result.error(
ApiError.unknown(
'gRPC command failed: ${e.toString()}',
exception: Exception(stackTrace.toString()),
),
);
}
}
// ============================================================
// Query Methods
// ============================================================
/// Gets all delivery routes.
///
/// Returns a [Result] containing a list of [DeliveryRoute] objects.
/// Maps the gRPC [DeliveryRoutesResponse] to domain models.
///
/// Example:
/// ```dart
/// final result = await client.getDeliveryRoutes();
/// result.when(
/// success: (routes) => displayRoutes(routes),
/// onError: (error) => showError(error.message),
/// );
/// ```
Future<Result<List<DeliveryRoute>>> getDeliveryRoutes() async {
final result = await _executeWithAuth<DeliveryRoutesResponse>(
(options) => deliveryClient.getDeliveryRoutes(Empty(), options: options),
);
try {
final options = await _buildCallOptions();
final request = DynamicQuerySimpleDeliveryRouteQueryItemsRequest();
return result.when(
success: (response) {
final routes = response.routes.map(_mapDeliveryRouteProto).toList();
return Result.success(routes);
},
onError: (error) => Result.error(error),
);
final response = await queryClient.querySimpleDeliveryRouteQueryItems(
request,
options: options,
);
final routes = response.data.map((item) => DeliveryRoute(
id: item.id.toInt(),
routeId: item.routeId.toInt(),
name: item.name,
routeName: item.routeName,
deliveriesCount: item.deliveriesCount,
deliveredCount: item.deliveredCount,
completed: item.completed,
createdAt: item.createdAt.toDateTime().toIso8601String(),
)).toList();
return Result.success(routes);
} on GrpcError catch (e) {
return Result.error(_mapGrpcError(e));
} catch (e) {
return Result.error(ApiError.unknown(e.toString(), exception: Exception(e.toString())));
}
}
/// Maps a [DeliveryRouteProto] to a [DeliveryRoute] domain model.
DeliveryRoute _mapDeliveryRouteProto(DeliveryRouteProto proto) {
return DeliveryRoute(
id: proto.id,
routeId: proto.routeId,
name: proto.name,
routeName: proto.routeName,
deliveriesCount: proto.deliveriesCount,
deliveredCount: proto.deliveredCount,
completed: proto.completed,
createdAt: proto.createdAt,
);
}
Future<Result<List<Delivery>>> getDeliveries({required int routeFragmentId}) async {
try {
final options = await _buildCallOptions();
final request = DynamicQuerySimpleDeliveriesQueryItemsRequest(
filters: [
DynamicQueryFilter(
path: 'RouteFragmentId',
type: 0, // Equal
value: routeFragmentId.toString(),
),
],
);
/// Gets deliveries for a specific route fragment.
///
/// Returns a [Result] containing a list of [Delivery] objects for the
/// specified [routeFragmentId]. Maps the gRPC [DeliveriesResponse] to
/// domain models.
///
/// Example:
/// ```dart
/// final result = await client.getDeliveries(routeFragmentId: 123);
/// result.when(
/// success: (deliveries) => displayDeliveries(deliveries),
/// onError: (error) => showError(error.message),
/// );
/// ```
Future<Result<List<Delivery>>> getDeliveries({
required int routeFragmentId,
}) async {
final request = GetDeliveriesRequest(routeFragmentId: routeFragmentId);
final response = await queryClient.querySimpleDeliveriesQueryItems(
request,
options: options,
);
final result = await _executeWithAuth<DeliveriesResponse>(
(options) => deliveryClient.getDeliveries(request, options: options),
);
final deliveries = response.data.map((item) {
final address = item.hasDeliveryAddress()
? DeliveryAddress(
id: item.deliveryAddress.id.toInt(),
line1: item.deliveryAddress.line1,
line2: item.deliveryAddress.line2.isNotEmpty ? item.deliveryAddress.line2 : null,
postalCode: item.deliveryAddress.postalCode.isNotEmpty ? item.deliveryAddress.postalCode : null,
city: item.deliveryAddress.city,
subdivision: item.deliveryAddress.subdivision.isNotEmpty ? item.deliveryAddress.subdivision : null,
countryCode: item.deliveryAddress.countryCode,
latitude: item.deliveryAddress.latitude,
longitude: item.deliveryAddress.longitude,
formattedAddress: item.deliveryAddress.formattedAddress,
)
: null;
return result.when(
success: (response) {
final deliveries =
response.deliveries.map(_mapDeliveryProto).toList();
return Result.success(deliveries);
},
onError: (error) => Result.error(error),
);
}
final orders = item.orders.map((orderProto) {
final contacts = orderProto.contacts.map((contactProto) => DeliveryContact(
firstName: contactProto.firstName,
lastName: contactProto.lastName,
phoneNumber: contactProto.phoneNumber.isNotEmpty ? contactProto.phoneNumber : null,
fullName: contactProto.fullName,
)).toList();
/// Maps a [DeliveryProto] to a [Delivery] domain model.
Delivery _mapDeliveryProto(DeliveryProto proto) {
return Delivery(
id: proto.id,
routeFragmentId: proto.routeFragmentId,
deliveryIndex: proto.deliveryIndex,
orders: proto.orders.map(_mapDeliveryOrderProto).toList(),
deliveredBy:
proto.hasDeliveredBy() ? _mapUserInfoProto(proto.deliveredBy) : null,
deliveryAddress: proto.hasDeliveryAddress()
? _mapDeliveryAddressProto(proto.deliveryAddress)
: null,
deliveredAt: proto.hasDeliveredAt() ? proto.deliveredAt : null,
skippedAt: proto.hasSkippedAt() ? proto.skippedAt : null,
createdAt: proto.createdAt,
updatedAt: proto.hasUpdatedAt() ? proto.updatedAt : null,
delivered: proto.delivered,
hasBeenSkipped: proto.hasBeenSkipped,
isSkipped: proto.isSkipped,
name: proto.name,
);
}
return DeliveryOrder(
id: orderProto.id.toInt(),
isNewCustomer: orderProto.isNewCustomer,
note: orderProto.note.isNotEmpty ? orderProto.note : null,
totalAmount: double.tryParse(orderProto.totalAmount) ?? 0.0,
totalPaid: orderProto.totalPaid.isNotEmpty ? double.tryParse(orderProto.totalPaid) : null,
totalItems: orderProto.totalItems,
contacts: contacts,
contact: orderProto.hasContact()
? DeliveryContact(
firstName: orderProto.contact.firstName,
lastName: orderProto.contact.lastName,
phoneNumber: orderProto.contact.phoneNumber.isNotEmpty ? orderProto.contact.phoneNumber : null,
fullName: orderProto.contact.fullName,
)
: null,
);
}).toList();
/// Maps a [DeliveryAddressProto] to a [DeliveryAddress] domain model.
DeliveryAddress _mapDeliveryAddressProto(DeliveryAddressProto proto) {
return DeliveryAddress(
id: proto.id,
line1: proto.hasLine1() ? proto.line1 : null,
line2: proto.hasLine2() ? proto.line2 : null,
postalCode: proto.hasPostalCode() ? proto.postalCode : null,
city: proto.hasCity() ? proto.city : null,
subdivision: proto.hasSubdivision() ? proto.subdivision : null,
countryCode: proto.hasCountryCode() ? proto.countryCode : null,
latitude: proto.hasLatitude() ? proto.latitude : null,
longitude: proto.hasLongitude() ? proto.longitude : null,
formattedAddress:
proto.hasFormattedAddress() ? proto.formattedAddress : null,
);
}
final deliveredBy = item.hasDeliveredBy()
? UserInfo(
id: item.deliveredBy.id.toInt(),
firstName: item.deliveredBy.firstName,
lastName: item.deliveredBy.lastName,
fullName: item.deliveredBy.fullName,
)
: null;
/// Maps a [DeliveryOrderProto] to a [DeliveryOrder] domain model.
DeliveryOrder _mapDeliveryOrderProto(DeliveryOrderProto proto) {
return DeliveryOrder(
id: proto.id,
isNewCustomer: proto.isNewCustomer,
note: proto.hasNote() ? proto.note : null,
totalAmount: proto.totalAmount,
totalPaid: proto.hasTotalPaid() ? proto.totalPaid : null,
totalItems: proto.hasTotalItems() ? proto.totalItems : null,
contacts: proto.contacts.map(_mapDeliveryContactProto).toList(),
contact:
proto.hasContact() ? _mapDeliveryContactProto(proto.contact) : null,
);
}
/// Maps a [DeliveryContactProto] to a [DeliveryContact] domain model.
DeliveryContact _mapDeliveryContactProto(DeliveryContactProto proto) {
return DeliveryContact(
firstName: proto.firstName,
lastName: proto.hasLastName() ? proto.lastName : null,
fullName: proto.fullName,
phoneNumber: proto.hasPhoneNumber() ? proto.phoneNumber : null,
);
}
/// Maps a [UserInfoProto] to a [UserInfo] domain model.
UserInfo _mapUserInfoProto(UserInfoProto proto) {
return UserInfo(
id: proto.id,
firstName: proto.firstName,
lastName: proto.hasLastName() ? proto.lastName : null,
fullName: proto.fullName,
);
}
// ============================================================
// Command Methods
// ============================================================
/// Marks a delivery as completed.
///
/// Returns a [Result] indicating success or failure. Optionally accepts
/// a [deliveredAt] timestamp; if not provided, the server will use the
/// current time.
///
/// Example:
/// ```dart
/// final result = await client.completeDelivery(deliveryId: 123);
/// result.when(
/// success: (_) => showSuccess('Delivery completed'),
/// onError: (error) => showError(error.message),
/// );
/// ```
Future<Result<void>> completeDelivery({
required int deliveryId,
String? deliveredAt,
}) async {
final request = CompleteDeliveryRequest(
deliveryId: deliveryId,
deliveredAt: deliveredAt,
);
final result = await _executeWithAuth<CommandResponse>(
(options) => deliveryClient.completeDelivery(request, options: options),
);
return result.when(
success: (response) {
if (response.success) {
return Result.success(null);
}
return Result.error(
ApiError.unknown(response.message.isNotEmpty
? response.message
: 'Failed to complete delivery'),
return Delivery(
id: item.id.toInt(),
routeFragmentId: item.routeFragmentId.toInt(),
deliveryIndex: item.deliveryIndex,
orders: orders,
deliveredBy: deliveredBy,
deliveryAddress: address,
deliveredAt: item.hasDeliveredAt() ? item.deliveredAt.toDateTime().toIso8601String() : null,
skippedAt: item.hasSkippedAt() ? item.skippedAt.toDateTime().toIso8601String() : null,
createdAt: item.createdAt.toDateTime().toIso8601String(),
updatedAt: item.hasUpdatedAt() ? item.updatedAt.toDateTime().toIso8601String() : null,
delivered: item.delivered,
hasBeenSkipped: item.hasBeenSkipped,
isSkipped: item.isSkipped,
name: item.name,
);
},
onError: (error) => Result.error(error),
);
}).toList();
return Result.success(deliveries);
} on GrpcError catch (e) {
return Result.error(_mapGrpcError(e));
} catch (e) {
return Result.error(ApiError.unknown(e.toString(), exception: Exception(e.toString())));
}
}
/// Marks a delivery as uncompleted.
///
/// Returns a [Result] indicating success or failure. Use this to revert
/// a previously completed delivery back to pending status.
///
/// Example:
/// ```dart
/// final result = await client.markDeliveryAsUncompleted(deliveryId: 123);
/// result.when(
/// success: (_) => showSuccess('Delivery marked as uncompleted'),
/// onError: (error) => showError(error.message),
/// );
/// ```
Future<Result<void>> markDeliveryAsUncompleted({
required int deliveryId,
}) async {
final request = MarkDeliveryUncompletedRequest(
deliveryId: deliveryId,
);
Future<Result<void>> completeDelivery({required int deliveryId}) async {
try {
final options = await _buildCallOptions();
final request = CompleteDeliveryCommandRequest(
deliveryId: Int64(deliveryId),
deliveredAt: DateTime.now().toUtc().toProto3Timestamp(),
);
final result = await _executeWithAuth<CommandResponse>(
(options) =>
deliveryClient.markDeliveryUncompleted(request, options: options),
);
return result.when(
success: (response) {
if (response.success) {
return Result.success(null);
}
return Result.error(
ApiError.unknown(response.message.isNotEmpty
? response.message
: 'Failed to mark delivery as uncompleted'),
);
},
onError: (error) => Result.error(error),
);
await commandClient.completeDelivery(request, options: options);
return Result.success(null);
} on GrpcError catch (e) {
return Result.error(_mapGrpcError(e));
} catch (e) {
return Result.error(ApiError.unknown(e.toString(), exception: Exception(e.toString())));
}
}
Future<Result<void>> markDeliveryAsUncompleted({required int deliveryId}) async {
try {
final options = await _buildCallOptions();
final request = MarkDeliveryAsUncompletedCommandRequest(
deliveryId: Int64(deliveryId),
);
await commandClient.markDeliveryAsUncompleted(request, options: options);
return Result.success(null);
} on GrpcError catch (e) {
return Result.error(_mapGrpcError(e));
} catch (e) {
return Result.error(ApiError.unknown(e.toString(), exception: Exception(e.toString())));
}
}
/// Skips a delivery.
///
/// Returns a [Result] indicating success or failure. Use this when a
/// delivery cannot be completed and needs to be skipped.
///
/// Example:
/// ```dart
/// final result = await client.skipDelivery(deliveryId: 123);
/// result.when(
/// success: (_) => showSuccess('Delivery skipped'),
/// onError: (error) => showError(error.message),
/// );
/// ```
Future<Result<void>> skipDelivery({
required int deliveryId,
required String description,
}) async {
final request = SkipDeliveryRequest(
deliveryId: deliveryId,
);
try {
final options = await _buildCallOptions();
final request = SkipDeliveryCommandRequest(
deliveryId: Int64(deliveryId),
description: description,
skippedAt: DateTime.now().toUtc().toProto3Timestamp(),
);
final result = await _executeWithAuth<CommandResponse>(
(options) => deliveryClient.skipDelivery(request, options: options),
);
return result.when(
success: (response) {
if (response.success) {
return Result.success(null);
}
return Result.error(
ApiError.unknown(response.message.isNotEmpty
? response.message
: 'Failed to skip delivery'),
);
},
onError: (error) => Result.error(error),
);
await commandClient.skipDelivery(request, options: options);
return Result.success(null);
} on GrpcError catch (e) {
return Result.error(_mapGrpcError(e));
} catch (e) {
return Result.error(ApiError.unknown(e.toString(), exception: Exception(e.toString())));
}
}
/// Shuts down the gRPC channel and releases resources.
///
/// Should be called when the client is no longer needed to properly
/// clean up network resources.
Future<void> shutdown() async {
await _channel?.shutdown();
void shutdown() {
_channel?.shutdown();
_channel = null;
_deliveryClient = null;
_queryClient = null;
_commandClient = null;
}
}
// Extension to convert DateTime to protobuf Timestamp
extension DateTimeToProto on DateTime {
$pb.Timestamp toProto3Timestamp() {
return $pb.Timestamp()
..seconds = Int64(millisecondsSinceEpoch ~/ 1000)
..nanos = ((millisecondsSinceEpoch % 1000) * 1000000).toInt();
}
/// Terminates the gRPC channel immediately.
///
/// Unlike [shutdown], this does not wait for pending calls to complete.
/// Use this for emergency cleanup or when the app is terminating.
Future<void> terminate() async {
await _channel?.terminate();
_channel = null;
_deliveryClient = null;
}
/// Returns true if the channel is currently active.
bool get isConnected => _channel != null;
}