From c90a88481cbfaab803e4186d301a95f09ab090e1 Mon Sep 17 00:00:00 2001 From: flemming-it Date: Thu, 28 May 2026 16:54:51 +0200 Subject: [PATCH] feat(client): streamEvents() wraps server-streaming audit log MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds `HubClient.streamEvents(backfill:, types:)` returning a `Stream` that fronts the new `HubAdmin/StreamEvents` RPC. Each event arrives the instant the hub appends it; cancel the subscription to close the RPC. `backfill` defaults to 50 historical events (oldest-last for chronological reception). `RESOURCE_EXHAUSTED` from the server means the receiver fell behind — reconnect with a fresh backfill to resync. Bindings regenerated from fai_platform/proto. dart analyze + dart test green (4 tests). Signed-off-by: flemming-it --- lib/src/generated/fai/v1/hub.pb.dart | 58 ++++++++++++++++++++++++ lib/src/generated/fai/v1/hub.pbgrpc.dart | 25 ++++++++++ lib/src/generated/fai/v1/hub.pbjson.dart | 14 ++++++ lib/src/hub_client.dart | 19 ++++++++ 4 files changed, 116 insertions(+) diff --git a/lib/src/generated/fai/v1/hub.pb.dart b/lib/src/generated/fai/v1/hub.pb.dart index 5492dce..63f0a23 100644 --- a/lib/src/generated/fai/v1/hub.pb.dart +++ b/lib/src/generated/fai/v1/hub.pb.dart @@ -864,6 +864,64 @@ class EventLogResponse extends $pb.GeneratedMessage { $pb.PbList get events => $_getList(0); } +class StreamEventsRequest extends $pb.GeneratedMessage { + factory StreamEventsRequest({ + $core.int? backfill, + $core.Iterable<$core.String>? eventTypes, + }) { + final $result = create(); + if (backfill != null) { + $result.backfill = backfill; + } + if (eventTypes != null) { + $result.eventTypes.addAll(eventTypes); + } + return $result; + } + StreamEventsRequest._() : super(); + factory StreamEventsRequest.fromBuffer($core.List<$core.int> i, [$pb.ExtensionRegistry r = $pb.ExtensionRegistry.EMPTY]) => create()..mergeFromBuffer(i, r); + factory StreamEventsRequest.fromJson($core.String i, [$pb.ExtensionRegistry r = $pb.ExtensionRegistry.EMPTY]) => create()..mergeFromJson(i, r); + + static final $pb.BuilderInfo _i = $pb.BuilderInfo(_omitMessageNames ? '' : 'StreamEventsRequest', package: const $pb.PackageName(_omitMessageNames ? '' : 'fai.v1'), createEmptyInstance: create) + ..a<$core.int>(1, _omitFieldNames ? '' : 'backfill', $pb.PbFieldType.O3) + ..pPS(2, _omitFieldNames ? '' : 'eventTypes') + ..hasRequiredFields = false + ; + + @$core.Deprecated('See https://github.com/google/protobuf.dart/issues/998.') + StreamEventsRequest clone() => StreamEventsRequest()..mergeFromMessage(this); + @$core.Deprecated('See https://github.com/google/protobuf.dart/issues/998.') + StreamEventsRequest copyWith(void Function(StreamEventsRequest) updates) => super.copyWith((message) => updates(message as StreamEventsRequest)) as StreamEventsRequest; + + $pb.BuilderInfo get info_ => _i; + + @$core.pragma('dart2js:noInline') + static StreamEventsRequest create() => StreamEventsRequest._(); + StreamEventsRequest createEmptyInstance() => create(); + static $pb.PbList createRepeated() => $pb.PbList(); + @$core.pragma('dart2js:noInline') + static StreamEventsRequest getDefault() => _defaultInstance ??= $pb.GeneratedMessage.$_defaultFor(create); + static StreamEventsRequest? _defaultInstance; + + /// Backfill size before the stream switches to live updates. + /// 0 means "no backfill, only new events". Defaults to 50 when + /// unset; capped at 1000 server-side to bound the initial buffer. + @$pb.TagNumber(1) + $core.int get backfill => $_getIZ(0); + @$pb.TagNumber(1) + set backfill($core.int v) { $_setSignedInt32(0, v); } + @$pb.TagNumber(1) + $core.bool hasBackfill() => $_has(0); + @$pb.TagNumber(1) + void clearBackfill() => $_clearField(1); + + /// Optional event_type filter; when empty, all types are streamed. + /// The filter applies to both the backfill prefix and the live + /// tail. + @$pb.TagNumber(2) + $pb.PbList<$core.String> get eventTypes => $_getList(1); +} + class LoggedEvent extends $pb.GeneratedMessage { factory LoggedEvent({ $core.String? eventId, diff --git a/lib/src/generated/fai/v1/hub.pbgrpc.dart b/lib/src/generated/fai/v1/hub.pbgrpc.dart index 659fe01..33bd5fc 100644 --- a/lib/src/generated/fai/v1/hub.pbgrpc.dart +++ b/lib/src/generated/fai/v1/hub.pbgrpc.dart @@ -139,6 +139,10 @@ class HubAdminClient extends $grpc.Client { '/fai.v1.HubAdmin/EventLog', ($0.EventLogRequest value) => value.writeToBuffer(), ($core.List<$core.int> value) => $0.EventLogResponse.fromBuffer(value)); + static final _$streamEvents = $grpc.ClientMethod<$0.StreamEventsRequest, $0.LoggedEvent>( + '/fai.v1.HubAdmin/StreamEvents', + ($0.StreamEventsRequest value) => value.writeToBuffer(), + ($core.List<$core.int> value) => $0.LoggedEvent.fromBuffer(value)); static final _$saveFlow = $grpc.ClientMethod<$0.SaveFlowRequest, $0.SaveFlowResponse>( '/fai.v1.HubAdmin/SaveFlow', ($0.SaveFlowRequest value) => value.writeToBuffer(), @@ -315,6 +319,15 @@ class HubAdminClient extends $grpc.Client { return $createUnaryCall(_$eventLog, request, options: options); } + /// Server-streaming variant of EventLog. The hub first emits up + /// to `limit` historical events (newest-last so clients receive + /// them in chronological order), then keeps the stream open and + /// forwards every freshly-appended event in real time. Reads + /// require the Read scope when auth is on. + $grpc.ResponseStream<$0.LoggedEvent> streamEvents($0.StreamEventsRequest request, {$grpc.CallOptions? options}) { + return $createStreamingCall(_$streamEvents, $async.Stream.fromIterable([request]), options: options); + } + /// Saved-flow management. A saved flow is a YAML file under /// /flows/.yaml; it can be re-run by name later. $grpc.ResponseFuture<$0.SaveFlowResponse> saveFlow($0.SaveFlowRequest request, {$grpc.CallOptions? options}) { @@ -619,6 +632,13 @@ abstract class HubAdminServiceBase extends $grpc.Service { false, ($core.List<$core.int> value) => $0.EventLogRequest.fromBuffer(value), ($0.EventLogResponse value) => value.writeToBuffer())); + $addMethod($grpc.ServiceMethod<$0.StreamEventsRequest, $0.LoggedEvent>( + 'StreamEvents', + streamEvents_Pre, + false, + true, + ($core.List<$core.int> value) => $0.StreamEventsRequest.fromBuffer(value), + ($0.LoggedEvent value) => value.writeToBuffer())); $addMethod($grpc.ServiceMethod<$0.SaveFlowRequest, $0.SaveFlowResponse>( 'SaveFlow', saveFlow_Pre, @@ -903,6 +923,10 @@ abstract class HubAdminServiceBase extends $grpc.Service { return eventLog($call, await $request); } + $async.Stream<$0.LoggedEvent> streamEvents_Pre($grpc.ServiceCall $call, $async.Future<$0.StreamEventsRequest> $request) async* { + yield* streamEvents($call, await $request); + } + $async.Future<$0.SaveFlowResponse> saveFlow_Pre($grpc.ServiceCall $call, $async.Future<$0.SaveFlowRequest> $request) async { return saveFlow($call, await $request); } @@ -1059,6 +1083,7 @@ abstract class HubAdminServiceBase extends $grpc.Service { $async.Future<$0.ModuleInfoResponse> moduleInfo($grpc.ServiceCall call, $0.ModuleInfoRequest request); $async.Future<$0.CheckPermissionResponse> checkPermission($grpc.ServiceCall call, $0.CheckPermissionRequest request); $async.Future<$0.EventLogResponse> eventLog($grpc.ServiceCall call, $0.EventLogRequest request); + $async.Stream<$0.LoggedEvent> streamEvents($grpc.ServiceCall call, $0.StreamEventsRequest request); $async.Future<$0.SaveFlowResponse> saveFlow($grpc.ServiceCall call, $0.SaveFlowRequest request); $async.Future<$0.DeleteFlowResponse> deleteFlow($grpc.ServiceCall call, $0.DeleteFlowRequest request); $async.Future<$0.FlowList> listFlows($grpc.ServiceCall call, $1.Empty request); diff --git a/lib/src/generated/fai/v1/hub.pbjson.dart b/lib/src/generated/fai/v1/hub.pbjson.dart index 3e088f0..a28c9fd 100644 --- a/lib/src/generated/fai/v1/hub.pbjson.dart +++ b/lib/src/generated/fai/v1/hub.pbjson.dart @@ -245,6 +245,20 @@ final $typed_data.Uint8List eventLogResponseDescriptor = $convert.base64Decode( 'ChBFdmVudExvZ1Jlc3BvbnNlEisKBmV2ZW50cxgBIAMoCzITLmZhaS52MS5Mb2dnZWRFdmVudF' 'IGZXZlbnRz'); +@$core.Deprecated('Use streamEventsRequestDescriptor instead') +const StreamEventsRequest$json = { + '1': 'StreamEventsRequest', + '2': [ + {'1': 'backfill', '3': 1, '4': 1, '5': 5, '10': 'backfill'}, + {'1': 'event_types', '3': 2, '4': 3, '5': 9, '10': 'eventTypes'}, + ], +}; + +/// Descriptor for `StreamEventsRequest`. Decode as a `google.protobuf.DescriptorProto`. +final $typed_data.Uint8List streamEventsRequestDescriptor = $convert.base64Decode( + 'ChNTdHJlYW1FdmVudHNSZXF1ZXN0EhoKCGJhY2tmaWxsGAEgASgFUghiYWNrZmlsbBIfCgtldm' + 'VudF90eXBlcxgCIAMoCVIKZXZlbnRUeXBlcw=='); + @$core.Deprecated('Use loggedEventDescriptor instead') const LoggedEvent$json = { '1': 'LoggedEvent', diff --git a/lib/src/hub_client.dart b/lib/src/hub_client.dart index 8cf22f8..7e74a96 100644 --- a/lib/src/hub_client.dart +++ b/lib/src/hub_client.dart @@ -158,6 +158,25 @@ class HubClient { return r.events; } + /// Server-streaming audit-log subscription. The hub first + /// replays up to [backfill] historical events (oldest-last so + /// the receiver sees them in chronological order), then keeps + /// the stream open and forwards every newly-appended event as + /// soon as it lands. + /// + /// Cancel the returned [Stream]'s subscription to close the + /// RPC. If the server reports `RESOURCE_EXHAUSTED` ("stream + /// lagged"), reconnect with backfill to resync. + Stream streamEvents({ + int backfill = 50, + List types = const [], + }) { + final req = pb.StreamEventsRequest() + ..backfill = backfill + ..eventTypes.addAll(types); + return _admin.streamEvents(req); + } + /// Pending approvals filtered by status. Defaults to all. Future> listApprovals({ List statuses = const [],