diff --git a/src/ScadaLink.Communication/Protos/sitestream.proto b/src/ScadaLink.Communication/Protos/sitestream.proto index 43ffbe3..5ceb709 100644 --- a/src/ScadaLink.Communication/Protos/sitestream.proto +++ b/src/ScadaLink.Communication/Protos/sitestream.proto @@ -9,6 +9,7 @@ service SiteStreamService { rpc SubscribeInstance(InstanceStreamRequest) returns (stream SiteStreamEvent); rpc IngestAuditEvents(AuditEventBatch) returns (IngestAck); rpc IngestCachedTelemetry(CachedTelemetryBatch) returns (IngestAck); + rpc PullAuditEvents(PullAuditEventsRequest) returns (PullAuditEventsResponse); } message InstanceStreamRequest { @@ -119,3 +120,19 @@ message CachedTelemetryPacket { } message CachedTelemetryBatch { repeated CachedTelemetryPacket packets = 1; } + +// Audit Log (#23) M6 reconciliation pull: central→site request for any +// site-local AuditLog rows with OccurredAtUtc >= since_utc that have not yet +// been ingested centrally (ForwardState in {Pending, Forwarded}). The site +// flips returned rows to Reconciled after the response is on the wire. +// more_available signals batch_size was saturated so the caller knows to +// issue a follow-up pull with an advanced since_utc cursor. +message PullAuditEventsRequest { + google.protobuf.Timestamp since_utc = 1; + int32 batch_size = 2; +} + +message PullAuditEventsResponse { + repeated AuditEventDto events = 1; + bool more_available = 2; +} diff --git a/src/ScadaLink.Communication/SiteStreamGrpc/Sitestream.cs b/src/ScadaLink.Communication/SiteStreamGrpc/Sitestream.cs index 9639242..ccac2bb 100644 --- a/src/ScadaLink.Communication/SiteStreamGrpc/Sitestream.cs +++ b/src/ScadaLink.Communication/SiteStreamGrpc/Sitestream.cs @@ -68,21 +68,27 @@ namespace ScadaLink.Communication.Grpc { "bnREdG8SNwoLb3BlcmF0aW9uYWwYAiABKAsyIi5zaXRlc3RyZWFtLlNpdGVD", "YWxsT3BlcmF0aW9uYWxEdG8iSgoUQ2FjaGVkVGVsZW1ldHJ5QmF0Y2gSMgoH", "cGFja2V0cxgBIAMoCzIhLnNpdGVzdHJlYW0uQ2FjaGVkVGVsZW1ldHJ5UGFj", - "a2V0KlwKB1F1YWxpdHkSFwoTUVVBTElUWV9VTlNQRUNJRklFRBAAEhAKDFFV", - "QUxJVFlfR09PRBABEhUKEVFVQUxJVFlfVU5DRVJUQUlOEAISDwoLUVVBTElU", - "WV9CQUQQAypdCg5BbGFybVN0YXRlRW51bRIbChdBTEFSTV9TVEFURV9VTlNQ", - "RUNJRklFRBAAEhYKEkFMQVJNX1NUQVRFX05PUk1BTBABEhYKEkFMQVJNX1NU", - "QVRFX0FDVElWRRACKoUBCg5BbGFybUxldmVsRW51bRIUChBBTEFSTV9MRVZF", - "TF9OT05FEAASEwoPQUxBUk1fTEVWRUxfTE9XEAESFwoTQUxBUk1fTEVWRUxf", - "TE9XX0xPVxACEhQKEEFMQVJNX0xFVkVMX0hJR0gQAxIZChVBTEFSTV9MRVZF", - "TF9ISUdIX0hJR0gQBDKFAgoRU2l0ZVN0cmVhbVNlcnZpY2USVQoRU3Vic2Ny", - "aWJlSW5zdGFuY2USIS5zaXRlc3RyZWFtLkluc3RhbmNlU3RyZWFtUmVxdWVz", - "dBobLnNpdGVzdHJlYW0uU2l0ZVN0cmVhbUV2ZW50MAESRwoRSW5nZXN0QXVk", - "aXRFdmVudHMSGy5zaXRlc3RyZWFtLkF1ZGl0RXZlbnRCYXRjaBoVLnNpdGVz", - "dHJlYW0uSW5nZXN0QWNrElAKFUluZ2VzdENhY2hlZFRlbGVtZXRyeRIgLnNp", - "dGVzdHJlYW0uQ2FjaGVkVGVsZW1ldHJ5QmF0Y2gaFS5zaXRlc3RyZWFtLklu", - "Z2VzdEFja0IfqgIcU2NhZGFMaW5rLkNvbW11bmljYXRpb24uR3JwY2IGcHJv", - "dG8z")); + "a2V0IlsKFlB1bGxBdWRpdEV2ZW50c1JlcXVlc3QSLQoJc2luY2VfdXRjGAEg", + "ASgLMhouZ29vZ2xlLnByb3RvYnVmLlRpbWVzdGFtcBISCgpiYXRjaF9zaXpl", + "GAIgASgFIlwKF1B1bGxBdWRpdEV2ZW50c1Jlc3BvbnNlEikKBmV2ZW50cxgB", + "IAMoCzIZLnNpdGVzdHJlYW0uQXVkaXRFdmVudER0bxIWCg5tb3JlX2F2YWls", + "YWJsZRgCIAEoCCpcCgdRdWFsaXR5EhcKE1FVQUxJVFlfVU5TUEVDSUZJRUQQ", + "ABIQCgxRVUFMSVRZX0dPT0QQARIVChFRVUFMSVRZX1VOQ0VSVEFJThACEg8K", + "C1FVQUxJVFlfQkFEEAMqXQoOQWxhcm1TdGF0ZUVudW0SGwoXQUxBUk1fU1RB", + "VEVfVU5TUEVDSUZJRUQQABIWChJBTEFSTV9TVEFURV9OT1JNQUwQARIWChJB", + "TEFSTV9TVEFURV9BQ1RJVkUQAiqFAQoOQWxhcm1MZXZlbEVudW0SFAoQQUxB", + "Uk1fTEVWRUxfTk9ORRAAEhMKD0FMQVJNX0xFVkVMX0xPVxABEhcKE0FMQVJN", + "X0xFVkVMX0xPV19MT1cQAhIUChBBTEFSTV9MRVZFTF9ISUdIEAMSGQoVQUxB", + "Uk1fTEVWRUxfSElHSF9ISUdIEAQy4QIKEVNpdGVTdHJlYW1TZXJ2aWNlElUK", + "EVN1YnNjcmliZUluc3RhbmNlEiEuc2l0ZXN0cmVhbS5JbnN0YW5jZVN0cmVh", + "bVJlcXVlc3QaGy5zaXRlc3RyZWFtLlNpdGVTdHJlYW1FdmVudDABEkcKEUlu", + "Z2VzdEF1ZGl0RXZlbnRzEhsuc2l0ZXN0cmVhbS5BdWRpdEV2ZW50QmF0Y2ga", + "FS5zaXRlc3RyZWFtLkluZ2VzdEFjaxJQChVJbmdlc3RDYWNoZWRUZWxlbWV0", + "cnkSIC5zaXRlc3RyZWFtLkNhY2hlZFRlbGVtZXRyeUJhdGNoGhUuc2l0ZXN0", + "cmVhbS5Jbmdlc3RBY2sSWgoPUHVsbEF1ZGl0RXZlbnRzEiIuc2l0ZXN0cmVh", + "bS5QdWxsQXVkaXRFdmVudHNSZXF1ZXN0GiMuc2l0ZXN0cmVhbS5QdWxsQXVk", + "aXRFdmVudHNSZXNwb25zZUIfqgIcU2NhZGFMaW5rLkNvbW11bmljYXRpb24u", + "R3JwY2IGcHJvdG8z")); descriptor = pbr::FileDescriptor.FromGeneratedCode(descriptorData, new pbr::FileDescriptor[] { global::Google.Protobuf.WellKnownTypes.TimestampReflection.Descriptor, global::Google.Protobuf.WellKnownTypes.WrappersReflection.Descriptor, }, new pbr::GeneratedClrTypeInfo(new[] {typeof(global::ScadaLink.Communication.Grpc.Quality), typeof(global::ScadaLink.Communication.Grpc.AlarmStateEnum), typeof(global::ScadaLink.Communication.Grpc.AlarmLevelEnum), }, null, new pbr::GeneratedClrTypeInfo[] { @@ -95,7 +101,9 @@ namespace ScadaLink.Communication.Grpc { new pbr::GeneratedClrTypeInfo(typeof(global::ScadaLink.Communication.Grpc.IngestAck), global::ScadaLink.Communication.Grpc.IngestAck.Parser, new[]{ "AcceptedEventIds" }, null, null, null, null), new pbr::GeneratedClrTypeInfo(typeof(global::ScadaLink.Communication.Grpc.SiteCallOperationalDto), global::ScadaLink.Communication.Grpc.SiteCallOperationalDto.Parser, new[]{ "TrackedOperationId", "Channel", "Target", "SourceSite", "Status", "RetryCount", "LastError", "HttpStatus", "CreatedAtUtc", "UpdatedAtUtc", "TerminalAtUtc" }, null, null, null, null), new pbr::GeneratedClrTypeInfo(typeof(global::ScadaLink.Communication.Grpc.CachedTelemetryPacket), global::ScadaLink.Communication.Grpc.CachedTelemetryPacket.Parser, new[]{ "AuditEvent", "Operational" }, null, null, null, null), - new pbr::GeneratedClrTypeInfo(typeof(global::ScadaLink.Communication.Grpc.CachedTelemetryBatch), global::ScadaLink.Communication.Grpc.CachedTelemetryBatch.Parser, new[]{ "Packets" }, null, null, null, null) + new pbr::GeneratedClrTypeInfo(typeof(global::ScadaLink.Communication.Grpc.CachedTelemetryBatch), global::ScadaLink.Communication.Grpc.CachedTelemetryBatch.Parser, new[]{ "Packets" }, null, null, null, null), + new pbr::GeneratedClrTypeInfo(typeof(global::ScadaLink.Communication.Grpc.PullAuditEventsRequest), global::ScadaLink.Communication.Grpc.PullAuditEventsRequest.Parser, new[]{ "SinceUtc", "BatchSize" }, null, null, null, null), + new pbr::GeneratedClrTypeInfo(typeof(global::ScadaLink.Communication.Grpc.PullAuditEventsResponse), global::ScadaLink.Communication.Grpc.PullAuditEventsResponse.Parser, new[]{ "Events", "MoreAvailable" }, null, null, null, null) })); } #endregion @@ -3862,6 +3870,482 @@ namespace ScadaLink.Communication.Grpc { } + /// + /// Audit Log (#23) M6 reconciliation pull: central→site request for any + /// site-local AuditLog rows with OccurredAtUtc >= since_utc that have not yet + /// been ingested centrally (ForwardState in {Pending, Forwarded}). The site + /// flips returned rows to Reconciled after the response is on the wire. + /// more_available signals batch_size was saturated so the caller knows to + /// issue a follow-up pull with an advanced since_utc cursor. + /// + [global::System.Diagnostics.DebuggerDisplayAttribute("{ToString(),nq}")] + public sealed partial class PullAuditEventsRequest : pb::IMessage + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + , pb::IBufferMessage + #endif + { + private static readonly pb::MessageParser _parser = new pb::MessageParser(() => new PullAuditEventsRequest()); + private pb::UnknownFieldSet _unknownFields; + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public static pb::MessageParser Parser { get { return _parser; } } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public static pbr::MessageDescriptor Descriptor { + get { return global::ScadaLink.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[10]; } + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + pbr::MessageDescriptor pb::IMessage.Descriptor { + get { return Descriptor; } + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public PullAuditEventsRequest() { + OnConstruction(); + } + + partial void OnConstruction(); + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public PullAuditEventsRequest(PullAuditEventsRequest other) : this() { + sinceUtc_ = other.sinceUtc_ != null ? other.sinceUtc_.Clone() : null; + batchSize_ = other.batchSize_; + _unknownFields = pb::UnknownFieldSet.Clone(other._unknownFields); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public PullAuditEventsRequest Clone() { + return new PullAuditEventsRequest(this); + } + + /// Field number for the "since_utc" field. + public const int SinceUtcFieldNumber = 1; + private global::Google.Protobuf.WellKnownTypes.Timestamp sinceUtc_; + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public global::Google.Protobuf.WellKnownTypes.Timestamp SinceUtc { + get { return sinceUtc_; } + set { + sinceUtc_ = value; + } + } + + /// Field number for the "batch_size" field. + public const int BatchSizeFieldNumber = 2; + private int batchSize_; + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public int BatchSize { + get { return batchSize_; } + set { + batchSize_ = value; + } + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public override bool Equals(object other) { + return Equals(other as PullAuditEventsRequest); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public bool Equals(PullAuditEventsRequest other) { + if (ReferenceEquals(other, null)) { + return false; + } + if (ReferenceEquals(other, this)) { + return true; + } + if (!object.Equals(SinceUtc, other.SinceUtc)) return false; + if (BatchSize != other.BatchSize) return false; + return Equals(_unknownFields, other._unknownFields); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public override int GetHashCode() { + int hash = 1; + if (sinceUtc_ != null) hash ^= SinceUtc.GetHashCode(); + if (BatchSize != 0) hash ^= BatchSize.GetHashCode(); + if (_unknownFields != null) { + hash ^= _unknownFields.GetHashCode(); + } + return hash; + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public override string ToString() { + return pb::JsonFormatter.ToDiagnosticString(this); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public void WriteTo(pb::CodedOutputStream output) { + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + output.WriteRawMessage(this); + #else + if (sinceUtc_ != null) { + output.WriteRawTag(10); + output.WriteMessage(SinceUtc); + } + if (BatchSize != 0) { + output.WriteRawTag(16); + output.WriteInt32(BatchSize); + } + if (_unknownFields != null) { + _unknownFields.WriteTo(output); + } + #endif + } + + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + void pb::IBufferMessage.InternalWriteTo(ref pb::WriteContext output) { + if (sinceUtc_ != null) { + output.WriteRawTag(10); + output.WriteMessage(SinceUtc); + } + if (BatchSize != 0) { + output.WriteRawTag(16); + output.WriteInt32(BatchSize); + } + if (_unknownFields != null) { + _unknownFields.WriteTo(ref output); + } + } + #endif + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public int CalculateSize() { + int size = 0; + if (sinceUtc_ != null) { + size += 1 + pb::CodedOutputStream.ComputeMessageSize(SinceUtc); + } + if (BatchSize != 0) { + size += 1 + pb::CodedOutputStream.ComputeInt32Size(BatchSize); + } + if (_unknownFields != null) { + size += _unknownFields.CalculateSize(); + } + return size; + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public void MergeFrom(PullAuditEventsRequest other) { + if (other == null) { + return; + } + if (other.sinceUtc_ != null) { + if (sinceUtc_ == null) { + SinceUtc = new global::Google.Protobuf.WellKnownTypes.Timestamp(); + } + SinceUtc.MergeFrom(other.SinceUtc); + } + if (other.BatchSize != 0) { + BatchSize = other.BatchSize; + } + _unknownFields = pb::UnknownFieldSet.MergeFrom(_unknownFields, other._unknownFields); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public void MergeFrom(pb::CodedInputStream input) { + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + input.ReadRawMessage(this); + #else + uint tag; + while ((tag = input.ReadTag()) != 0) { + if ((tag & 7) == 4) { + // Abort on any end group tag. + return; + } + switch(tag) { + default: + _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, input); + break; + case 10: { + if (sinceUtc_ == null) { + SinceUtc = new global::Google.Protobuf.WellKnownTypes.Timestamp(); + } + input.ReadMessage(SinceUtc); + break; + } + case 16: { + BatchSize = input.ReadInt32(); + break; + } + } + } + #endif + } + + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + void pb::IBufferMessage.InternalMergeFrom(ref pb::ParseContext input) { + uint tag; + while ((tag = input.ReadTag()) != 0) { + if ((tag & 7) == 4) { + // Abort on any end group tag. + return; + } + switch(tag) { + default: + _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, ref input); + break; + case 10: { + if (sinceUtc_ == null) { + SinceUtc = new global::Google.Protobuf.WellKnownTypes.Timestamp(); + } + input.ReadMessage(SinceUtc); + break; + } + case 16: { + BatchSize = input.ReadInt32(); + break; + } + } + } + } + #endif + + } + + [global::System.Diagnostics.DebuggerDisplayAttribute("{ToString(),nq}")] + public sealed partial class PullAuditEventsResponse : pb::IMessage + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + , pb::IBufferMessage + #endif + { + private static readonly pb::MessageParser _parser = new pb::MessageParser(() => new PullAuditEventsResponse()); + private pb::UnknownFieldSet _unknownFields; + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public static pb::MessageParser Parser { get { return _parser; } } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public static pbr::MessageDescriptor Descriptor { + get { return global::ScadaLink.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[11]; } + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + pbr::MessageDescriptor pb::IMessage.Descriptor { + get { return Descriptor; } + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public PullAuditEventsResponse() { + OnConstruction(); + } + + partial void OnConstruction(); + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public PullAuditEventsResponse(PullAuditEventsResponse other) : this() { + events_ = other.events_.Clone(); + moreAvailable_ = other.moreAvailable_; + _unknownFields = pb::UnknownFieldSet.Clone(other._unknownFields); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public PullAuditEventsResponse Clone() { + return new PullAuditEventsResponse(this); + } + + /// Field number for the "events" field. + public const int EventsFieldNumber = 1; + private static readonly pb::FieldCodec _repeated_events_codec + = pb::FieldCodec.ForMessage(10, global::ScadaLink.Communication.Grpc.AuditEventDto.Parser); + private readonly pbc::RepeatedField events_ = new pbc::RepeatedField(); + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public pbc::RepeatedField Events { + get { return events_; } + } + + /// Field number for the "more_available" field. + public const int MoreAvailableFieldNumber = 2; + private bool moreAvailable_; + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public bool MoreAvailable { + get { return moreAvailable_; } + set { + moreAvailable_ = value; + } + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public override bool Equals(object other) { + return Equals(other as PullAuditEventsResponse); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public bool Equals(PullAuditEventsResponse other) { + if (ReferenceEquals(other, null)) { + return false; + } + if (ReferenceEquals(other, this)) { + return true; + } + if(!events_.Equals(other.events_)) return false; + if (MoreAvailable != other.MoreAvailable) return false; + return Equals(_unknownFields, other._unknownFields); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public override int GetHashCode() { + int hash = 1; + hash ^= events_.GetHashCode(); + if (MoreAvailable != false) hash ^= MoreAvailable.GetHashCode(); + if (_unknownFields != null) { + hash ^= _unknownFields.GetHashCode(); + } + return hash; + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public override string ToString() { + return pb::JsonFormatter.ToDiagnosticString(this); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public void WriteTo(pb::CodedOutputStream output) { + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + output.WriteRawMessage(this); + #else + events_.WriteTo(output, _repeated_events_codec); + if (MoreAvailable != false) { + output.WriteRawTag(16); + output.WriteBool(MoreAvailable); + } + if (_unknownFields != null) { + _unknownFields.WriteTo(output); + } + #endif + } + + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + void pb::IBufferMessage.InternalWriteTo(ref pb::WriteContext output) { + events_.WriteTo(ref output, _repeated_events_codec); + if (MoreAvailable != false) { + output.WriteRawTag(16); + output.WriteBool(MoreAvailable); + } + if (_unknownFields != null) { + _unknownFields.WriteTo(ref output); + } + } + #endif + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public int CalculateSize() { + int size = 0; + size += events_.CalculateSize(_repeated_events_codec); + if (MoreAvailable != false) { + size += 1 + 1; + } + if (_unknownFields != null) { + size += _unknownFields.CalculateSize(); + } + return size; + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public void MergeFrom(PullAuditEventsResponse other) { + if (other == null) { + return; + } + events_.Add(other.events_); + if (other.MoreAvailable != false) { + MoreAvailable = other.MoreAvailable; + } + _unknownFields = pb::UnknownFieldSet.MergeFrom(_unknownFields, other._unknownFields); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public void MergeFrom(pb::CodedInputStream input) { + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + input.ReadRawMessage(this); + #else + uint tag; + while ((tag = input.ReadTag()) != 0) { + if ((tag & 7) == 4) { + // Abort on any end group tag. + return; + } + switch(tag) { + default: + _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, input); + break; + case 10: { + events_.AddEntriesFrom(input, _repeated_events_codec); + break; + } + case 16: { + MoreAvailable = input.ReadBool(); + break; + } + } + } + #endif + } + + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + void pb::IBufferMessage.InternalMergeFrom(ref pb::ParseContext input) { + uint tag; + while ((tag = input.ReadTag()) != 0) { + if ((tag & 7) == 4) { + // Abort on any end group tag. + return; + } + switch(tag) { + default: + _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, ref input); + break; + case 10: { + events_.AddEntriesFrom(ref input, _repeated_events_codec); + break; + } + case 16: { + MoreAvailable = input.ReadBool(); + break; + } + } + } + } + #endif + + } + #endregion } diff --git a/src/ScadaLink.Communication/SiteStreamGrpc/SitestreamGrpc.cs b/src/ScadaLink.Communication/SiteStreamGrpc/SitestreamGrpc.cs index e7b9b33..d5fd944 100644 --- a/src/ScadaLink.Communication/SiteStreamGrpc/SitestreamGrpc.cs +++ b/src/ScadaLink.Communication/SiteStreamGrpc/SitestreamGrpc.cs @@ -55,6 +55,10 @@ namespace ScadaLink.Communication.Grpc { static readonly grpc::Marshaller __Marshaller_sitestream_IngestAck = grpc::Marshallers.Create(__Helper_SerializeMessage, context => __Helper_DeserializeMessage(context, global::ScadaLink.Communication.Grpc.IngestAck.Parser)); [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] static readonly grpc::Marshaller __Marshaller_sitestream_CachedTelemetryBatch = grpc::Marshallers.Create(__Helper_SerializeMessage, context => __Helper_DeserializeMessage(context, global::ScadaLink.Communication.Grpc.CachedTelemetryBatch.Parser)); + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + static readonly grpc::Marshaller __Marshaller_sitestream_PullAuditEventsRequest = grpc::Marshallers.Create(__Helper_SerializeMessage, context => __Helper_DeserializeMessage(context, global::ScadaLink.Communication.Grpc.PullAuditEventsRequest.Parser)); + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + static readonly grpc::Marshaller __Marshaller_sitestream_PullAuditEventsResponse = grpc::Marshallers.Create(__Helper_SerializeMessage, context => __Helper_DeserializeMessage(context, global::ScadaLink.Communication.Grpc.PullAuditEventsResponse.Parser)); [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] static readonly grpc::Method __Method_SubscribeInstance = new grpc::Method( @@ -80,6 +84,14 @@ namespace ScadaLink.Communication.Grpc { __Marshaller_sitestream_CachedTelemetryBatch, __Marshaller_sitestream_IngestAck); + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + static readonly grpc::Method __Method_PullAuditEvents = new grpc::Method( + grpc::MethodType.Unary, + __ServiceName, + "PullAuditEvents", + __Marshaller_sitestream_PullAuditEventsRequest, + __Marshaller_sitestream_PullAuditEventsResponse); + /// Service descriptor public static global::Google.Protobuf.Reflection.ServiceDescriptor Descriptor { @@ -108,6 +120,12 @@ namespace ScadaLink.Communication.Grpc { throw new grpc::RpcException(new grpc::Status(grpc::StatusCode.Unimplemented, "")); } + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + public virtual global::System.Threading.Tasks.Task PullAuditEvents(global::ScadaLink.Communication.Grpc.PullAuditEventsRequest request, grpc::ServerCallContext context) + { + throw new grpc::RpcException(new grpc::Status(grpc::StatusCode.Unimplemented, "")); + } + } /// Client for SiteStreamService @@ -187,6 +205,26 @@ namespace ScadaLink.Communication.Grpc { { return CallInvoker.AsyncUnaryCall(__Method_IngestCachedTelemetry, null, options, request); } + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + public virtual global::ScadaLink.Communication.Grpc.PullAuditEventsResponse PullAuditEvents(global::ScadaLink.Communication.Grpc.PullAuditEventsRequest request, grpc::Metadata headers = null, global::System.DateTime? deadline = null, global::System.Threading.CancellationToken cancellationToken = default(global::System.Threading.CancellationToken)) + { + return PullAuditEvents(request, new grpc::CallOptions(headers, deadline, cancellationToken)); + } + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + public virtual global::ScadaLink.Communication.Grpc.PullAuditEventsResponse PullAuditEvents(global::ScadaLink.Communication.Grpc.PullAuditEventsRequest request, grpc::CallOptions options) + { + return CallInvoker.BlockingUnaryCall(__Method_PullAuditEvents, null, options, request); + } + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + public virtual grpc::AsyncUnaryCall PullAuditEventsAsync(global::ScadaLink.Communication.Grpc.PullAuditEventsRequest request, grpc::Metadata headers = null, global::System.DateTime? deadline = null, global::System.Threading.CancellationToken cancellationToken = default(global::System.Threading.CancellationToken)) + { + return PullAuditEventsAsync(request, new grpc::CallOptions(headers, deadline, cancellationToken)); + } + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + public virtual grpc::AsyncUnaryCall PullAuditEventsAsync(global::ScadaLink.Communication.Grpc.PullAuditEventsRequest request, grpc::CallOptions options) + { + return CallInvoker.AsyncUnaryCall(__Method_PullAuditEvents, null, options, request); + } /// Creates a new instance of client from given ClientBaseConfiguration. [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] protected override SiteStreamServiceClient NewInstance(ClientBaseConfiguration configuration) @@ -203,7 +241,8 @@ namespace ScadaLink.Communication.Grpc { return grpc::ServerServiceDefinition.CreateBuilder() .AddMethod(__Method_SubscribeInstance, serviceImpl.SubscribeInstance) .AddMethod(__Method_IngestAuditEvents, serviceImpl.IngestAuditEvents) - .AddMethod(__Method_IngestCachedTelemetry, serviceImpl.IngestCachedTelemetry).Build(); + .AddMethod(__Method_IngestCachedTelemetry, serviceImpl.IngestCachedTelemetry) + .AddMethod(__Method_PullAuditEvents, serviceImpl.PullAuditEvents).Build(); } /// Register service method with a service binder with or without implementation. Useful when customizing the service binding logic. @@ -216,6 +255,7 @@ namespace ScadaLink.Communication.Grpc { serviceBinder.AddMethod(__Method_SubscribeInstance, serviceImpl == null ? null : new grpc::ServerStreamingServerMethod(serviceImpl.SubscribeInstance)); serviceBinder.AddMethod(__Method_IngestAuditEvents, serviceImpl == null ? null : new grpc::UnaryServerMethod(serviceImpl.IngestAuditEvents)); serviceBinder.AddMethod(__Method_IngestCachedTelemetry, serviceImpl == null ? null : new grpc::UnaryServerMethod(serviceImpl.IngestCachedTelemetry)); + serviceBinder.AddMethod(__Method_PullAuditEvents, serviceImpl == null ? null : new grpc::UnaryServerMethod(serviceImpl.PullAuditEvents)); } } diff --git a/tests/ScadaLink.Communication.Tests/Protos/PullAuditEventsProtoTests.cs b/tests/ScadaLink.Communication.Tests/Protos/PullAuditEventsProtoTests.cs new file mode 100644 index 0000000..ba9ae37 --- /dev/null +++ b/tests/ScadaLink.Communication.Tests/Protos/PullAuditEventsProtoTests.cs @@ -0,0 +1,83 @@ +using Google.Protobuf; +using Google.Protobuf.WellKnownTypes; +using ScadaLink.Communication.Grpc; + +namespace ScadaLink.Communication.Tests.Protos; + +/// +/// Wire-format round-trip tests for the Audit Log (#23) M6 reconciliation +/// pull proto messages (, +/// ). Locks the additive contract the +/// central→site reconciliation puller depends on. +/// +public class PullAuditEventsProtoTests +{ + private static AuditEventDto NewAuditDto(Guid? id = null) => new() + { + EventId = (id ?? Guid.NewGuid()).ToString(), + OccurredAtUtc = Timestamp.FromDateTimeOffset( + new DateTimeOffset(2026, 5, 20, 10, 15, 30, 123, TimeSpan.Zero)), + Channel = "ApiOutbound", + Kind = "ApiCall", + Status = "Delivered", + SourceSiteId = "site-1", + }; + + [Fact] + public void PullAuditEventsRequest_RoundTrip() + { + var sinceUtc = Timestamp.FromDateTimeOffset( + new DateTimeOffset(2026, 5, 20, 9, 0, 0, TimeSpan.Zero)); + + var original = new PullAuditEventsRequest + { + SinceUtc = sinceUtc, + BatchSize = 250, + }; + + var bytes = original.ToByteArray(); + var deserialized = PullAuditEventsRequest.Parser.ParseFrom(bytes); + + Assert.Equal(sinceUtc, deserialized.SinceUtc); + Assert.Equal(250, deserialized.BatchSize); + } + + [Fact] + public void PullAuditEventsResponse_RoundTrip_WithEvents_And_MoreAvailable() + { + var dtos = Enumerable.Range(0, 4).Select(_ => NewAuditDto()).ToList(); + + var original = new PullAuditEventsResponse + { + MoreAvailable = true, + }; + original.Events.AddRange(dtos); + + var bytes = original.ToByteArray(); + var deserialized = PullAuditEventsResponse.Parser.ParseFrom(bytes); + + Assert.True(deserialized.MoreAvailable); + Assert.Equal(4, deserialized.Events.Count); + for (int i = 0; i < dtos.Count; i++) + { + Assert.Equal(dtos[i].EventId, deserialized.Events[i].EventId); + Assert.Equal(dtos[i].Status, deserialized.Events[i].Status); + Assert.Equal(dtos[i].SourceSiteId, deserialized.Events[i].SourceSiteId); + Assert.Equal(dtos[i].OccurredAtUtc, deserialized.Events[i].OccurredAtUtc); + } + } + + [Fact] + public void PullAuditEventsResponse_Empty_Yields_EmptyEvents() + { + var original = new PullAuditEventsResponse(); + Assert.Empty(original.Events); + Assert.False(original.MoreAvailable); + + var bytes = original.ToByteArray(); + var deserialized = PullAuditEventsResponse.Parser.ParseFrom(bytes); + + Assert.Empty(deserialized.Events); + Assert.False(deserialized.MoreAvailable); + } +}