diff --git a/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditArchiveHelpers.cs b/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditArchiveHelpers.cs
new file mode 100644
index 00000000..20f053ef
--- /dev/null
+++ b/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditArchiveHelpers.cs
@@ -0,0 +1,117 @@
+using System.Globalization;
+using System.Text.Json;
+
+namespace ZB.MOM.WW.ScadaBridge.CLI.Commands;
+
+///
+/// Helpers for the audit archive list / audit archive verify subcommands
+/// (AuditLog Parquet cold archive, ScadaBridge#42). Both send a management command
+/// (ListAuditArchives, VerifyAuditArchive); the archives live on the central
+/// nodes' archive directory, not on the operator's machine.
+///
+///
+/// verify exit codes: 0 verified, when the
+/// files, the chain or the purge anchor disagree, 1 on a request error, 2 on an
+/// authorization failure.
+///
+public static class AuditArchiveHelpers
+{
+ /// Process exit code when an archive failed verification (tamper evidence).
+ public const int NotVerifiedExitCode = 3;
+
+ ///
+ /// Client deadline for verify: the server hashes and re-chains a whole month under
+ /// the long-running management Ask (5 min); the client waits a little longer.
+ ///
+ public static readonly TimeSpan VerifyRequestTimeout = TimeSpan.FromMinutes(6);
+
+ /// Renders a ListAuditArchives result.
+ /// The success body (an AuditArchiveListResult, camelCase).
+ /// Resolved output format.
+ /// Writer for the report.
+ /// Always 0.
+ public static int RenderList(string json, string format, TextWriter output)
+ {
+ if (!string.Equals(format, "table", StringComparison.OrdinalIgnoreCase))
+ {
+ output.WriteLine(json);
+ output.Flush();
+ return 0;
+ }
+
+ using var doc = JsonDocument.Parse(json);
+ var root = doc.RootElement;
+ var enabled = root.TryGetProperty("enabled", out var e) && e.ValueKind == JsonValueKind.True;
+ output.WriteLine($"Archiving: {(enabled ? "enabled" : "disabled")}");
+ if (!root.TryGetProperty("archives", out var archives) || archives.GetArrayLength() == 0)
+ {
+ output.WriteLine("No archives.");
+ output.Flush();
+ return 0;
+ }
+
+ output.WriteLine($"{"NAME",-42} {"PARTITION",-9} {"STATUS",-8} {"ROWS",12} {"CHAINED",12} {"BYTES",14}");
+ foreach (var a in archives.EnumerateArray())
+ {
+ output.WriteLine(string.Create(CultureInfo.InvariantCulture,
+ $"{Str(a, "name"),-42} {Str(a, "partition"),-9} {Str(a, "status"),-8} {Num(a, "rowCount"),12} {Num(a, "chainedRowCount"),12} {Num(a, "dataBytes"),14}"));
+ }
+
+ output.Flush();
+ return 0;
+ }
+
+ /// Renders a VerifyAuditArchive result and maps it to the exit code.
+ /// The success body (an AuditArchiveVerificationResult, camelCase).
+ /// Resolved output format.
+ /// Writer for the report.
+ /// 0 when verified, otherwise.
+ public static int RenderVerify(string json, string format, TextWriter output)
+ {
+ using var doc = JsonDocument.Parse(json);
+ var root = doc.RootElement;
+ var verified = root.TryGetProperty("verified", out var v) && v.ValueKind == JsonValueKind.True;
+
+ if (!string.Equals(format, "table", StringComparison.OrdinalIgnoreCase))
+ {
+ output.WriteLine(json);
+ output.Flush();
+ return verified ? 0 : NotVerifiedExitCode;
+ }
+
+ output.WriteLine($"Archive : {Str(root, "name")}");
+ output.WriteLine($"Partition : {Str(root, "partition")}");
+ output.WriteLine($"Manifest : {Str(root, "manifestStatus")}");
+ output.WriteLine($"Verified : {(verified ? "yes" : "NO")}");
+ output.WriteLine($"Files intact: {(root.TryGetProperty("filesIntact", out var f) && f.ValueKind == JsonValueKind.True ? "yes" : "NO")}");
+ if (Str(root, "fileProblem") is { Length: > 0 } problem)
+ {
+ output.WriteLine($" problem : {problem}");
+ }
+
+ output.WriteLine($"Rows read : {Num(root, "rowsRead")}");
+ if (root.TryGetProperty("chains", out var chains) && chains.ValueKind == JsonValueKind.Array)
+ {
+ foreach (var c in chains.EnumerateArray())
+ {
+ var chain = c.GetProperty("chain");
+ output.WriteLine($"Chain {Str(chain, "partition")}: {Str(chain, "status")}, {Num(chain, "rowsVerified")} rows verified, "
+ + $"{Num(chain, "tombstonesApplied")} tombstones, {Num(chain, "unchainedRows")} unchained; anchor {Str(c, "anchor")}");
+ output.WriteLine($" {Str(c, "anchorDetail")}");
+ if (chain.TryGetProperty("firstBreak", out var brk) && brk.ValueKind == JsonValueKind.Object)
+ {
+ output.WriteLine($" first break: chainSeq {Num(brk, "chainSeq")}, eventId {Str(brk, "eventId", "-")}, {Str(brk, "kind")} — {Str(brk, "detail")}");
+ }
+ }
+ }
+
+ output.Flush();
+ return verified ? 0 : NotVerifiedExitCode;
+ }
+
+ private static string Str(JsonElement e, string name, string missing = "")
+ => e.TryGetProperty(name, out var v) && v.ValueKind == JsonValueKind.String ? v.GetString() ?? missing : missing;
+
+ private static string Num(JsonElement e, string name, string missing = "0")
+ => e.TryGetProperty(name, out var v) && v.ValueKind == JsonValueKind.Number ? v.GetInt64().ToString(CultureInfo.InvariantCulture) : missing;
+}
diff --git a/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditCommands.cs b/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditCommands.cs
index 8ec96f77..91976085 100644
--- a/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditCommands.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditCommands.cs
@@ -9,7 +9,9 @@ namespace ZB.MOM.WW.ScadaBridge.CLI.Commands;
/// the centralized append-only Audit Log via the REST endpoints
/// (GET /api/audit/query, GET /api/audit/export,
/// GET /api/audit/tree), plus verify-chain, which sends the
-/// VerifyAuditChain management command (hash-chain tamper evidence, ScadaBridge#41).
+/// VerifyAuditChain management command (hash-chain tamper evidence, ScadaBridge#41),
+/// and archive list / archive verify, which send ListAuditArchives /
+/// VerifyAuditArchive (Parquet cold archive, ScadaBridge#42).
///
public static class AuditCommands
{
@@ -31,6 +33,7 @@ public static class AuditCommands
command.Add(BuildTree(urlOption, formatOption, usernameOption, passwordOption));
command.Add(BuildVerifyChain(urlOption, formatOption, usernameOption, passwordOption));
command.Add(BuildBackfillSourceNode(urlOption, formatOption, usernameOption, passwordOption));
+ command.Add(BuildArchive(urlOption, formatOption, usernameOption, passwordOption));
return command;
}
@@ -316,6 +319,49 @@ public static class AuditCommands
return cmd;
}
+ ///
+ /// Builds the audit archive group (ScadaBridge#42): list and verify
+ /// over the AuditLog Parquet cold archives on the central archive directory.
+ ///
+ private static Command BuildArchive(Option urlOption, Option formatOption, Option usernameOption, Option passwordOption)
+ {
+ var group = new Command("archive") { Description = "List and verify the AuditLog Parquet cold archives" };
+
+ var list = new Command("list") { Description = "List the AuditLog Parquet archives written by the retention purge" };
+ list.SetAction(async (ParseResult result) =>
+ {
+ var format = AuditCommandHelpers.ResolveFormat(result, formatOption);
+ return await CommandHelpers.ExecuteCommandAsync(
+ result, urlOption, formatOption, usernameOption, passwordOption,
+ new ListAuditArchivesCommand(),
+ onSuccess: json => AuditArchiveHelpers.RenderList(json, format, Console.Out));
+ });
+ group.Add(list);
+
+ var nameOption = new Option("--name")
+ {
+ Description = "Archive name, as 'audit archive list' prints it (auditlog-YYYY-MM-yyyyMMddTHHmmssfffZ)",
+ Required = true,
+ };
+ var verify = new Command("verify")
+ {
+ Description = "Verify an archive's files, its hash chain and its purge anchor; exit 3 if it does not verify",
+ };
+ verify.Add(nameOption);
+ verify.SetAction(async (ParseResult result) =>
+ {
+ var format = AuditCommandHelpers.ResolveFormat(result, formatOption);
+ return await CommandHelpers.ExecuteCommandAsync(
+ result, urlOption, formatOption, usernameOption, passwordOption,
+ new VerifyAuditArchiveCommand(result.GetValue(nameOption)!),
+ timeout: AuditArchiveHelpers.VerifyRequestTimeout,
+ onSuccess: json => AuditArchiveHelpers.RenderVerify(json, format, Console.Out));
+ });
+ group.Add(verify);
+
+ return group;
+ }
+
///
/// Builds the audit backfill-source-node sub-command.
/// Sets SourceNode on historical pre-feature rows whose SourceNode IS NULL
diff --git a/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditExportHelpers.cs b/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditExportHelpers.cs
index b57a82c7..3a5e46e3 100644
--- a/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditExportHelpers.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditExportHelpers.cs
@@ -134,8 +134,8 @@ public static class AuditExportHelpers
/// Executes the export: GETs /api/audit/export and copies the response body
/// stream directly to . The body is never fully
/// buffered — the copy streams in fixed-size chunks via .
- /// A 501 Not Implemented (parquet not yet supported server-side) prints the
- /// server message and returns a non-zero exit code.
+ /// A 501 Not Implemented (a central older than ScadaBridge#42 does not serve
+ /// format=parquet) prints the server message and returns a non-zero exit code.
///
/// The management HTTP client for API communication.
/// The export arguments containing filters and output file path.
diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Management/AuditLogCommands.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Management/AuditLogCommands.cs
index 8a024ae2..5fd94e17 100644
--- a/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Management/AuditLogCommands.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Management/AuditLogCommands.cs
@@ -31,3 +31,19 @@ public record AuditLogQueryResult(IReadOnlyList Entries, int Tota
///
/// The month to verify, YYYY-MM.
public record VerifyAuditChainCommand(string Month);
+
+///
+/// Lists the AuditLog Parquet cold archives under the configured archive directory (audit
+/// roadmap T2, ScadaBridge#42). Read-only; OperationalAudit roles, refused for a site-scoped
+/// caller because an archive holds every site's rows.
+///
+public record ListAuditArchivesCommand;
+
+///
+/// Verifies one AuditLog Parquet archive (ScadaBridge#42): file integrity against its
+/// manifest, the hash chain recomputed over the archived rows, and the chain end against the
+/// checkpoint anchor the partition purge left. Read-only; same gate as
+/// .
+///
+/// The archive name, as audit archive list reports it.
+public record VerifyAuditArchiveCommand(string Name);
diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/AuditEndpoints.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/AuditEndpoints.cs
index 1c5f983b..01c44dce 100644
--- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/AuditEndpoints.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/AuditEndpoints.cs
@@ -22,8 +22,8 @@ namespace ZB.MOM.WW.ScadaBridge.ManagementService;
///
/// - GET /api/audit/query — keyset-paged JSON page, gated on the
/// permission.
-/// - GET /api/audit/export — streamed bulk export (csv / jsonl;
-/// parquet returns HTTP 501), gated on the
+///
- GET /api/audit/export — streamed bulk export (csv / jsonl /
+/// parquet), gated on the
/// permission.
/// - GET /api/audit/tree — execution-chain tree rooted at the
/// topmost ancestor of a given executionId, returned as a JSON array
@@ -47,11 +47,8 @@ namespace ZB.MOM.WW.ScadaBridge.ManagementService;
///
///
///
-/// Parquet. No Parquet library is referenced by the solution and
-/// Component-AuditLog.md explicitly defers Parquet archival to v1.x, so
-/// format=parquet returns 501 Not Implemented with a guidance
-/// message rather than a half-built binary stream. csv + jsonl are fully
-/// implemented.
+/// Parquet. format=parquet (ScadaBridge#42) streams the same 24 columns as
+/// csv / jsonl as one Parquet file (), a row group at a time.
///
///
public static class AuditEndpoints
@@ -65,6 +62,9 @@ public static class AuditEndpoints
/// Repository round-trip size used while streaming an export.
public const int ExportPageSize = 1000;
+ /// Content type of a format=parquet export (ScadaBridge#42).
+ public const string ParquetContentType = "application/vnd.apache.parquet";
+
///
/// Serializer for individual rows (query results +
/// jsonl export lines). Drops null columns so a sparse audit row stays
@@ -185,10 +185,10 @@ public static class AuditEndpoints
// ─────────────────────────────────────────────────────────────────────
///
- /// Handles GET /api/audit/export: authenticates, checks the AuditExport permission, and streams the matching rows as CSV or JSONL.
+ /// Handles GET /api/audit/export: authenticates, checks the AuditExport permission, and streams the matching rows as CSV, JSONL or Parquet.
///
/// The HTTP context for the current request.
- /// A task that resolves to the HTTP result (streaming CSV/JSONL, 400, 401, 403, or 501).
+ /// A task that resolves to the HTTP result (streaming CSV/JSONL/Parquet, 400, 401, or 403).
internal static async Task HandleExport(HttpContext context)
{
var auth = await AuthenticateAsync(context);
@@ -208,17 +208,7 @@ public static class AuditEndpoints
format = "csv";
}
- if (format == "parquet")
- {
- // Deferred to v1.x (Component-AuditLog.md §"Deferred"): no Parquet
- // library is in the solution. Return a 501 with actionable guidance
- // rather than a partial binary stream.
- return Results.Json(
- new { error = "Parquet export deferred to v1.x; use csv or jsonl.", code = "NOT_IMPLEMENTED" },
- statusCode: 501);
- }
-
- if (format != "csv" && format != "jsonl")
+ if (format != "csv" && format != "jsonl" && format != "parquet")
{
return Results.Json(
new { error = $"Unsupported export format '{format}'. Use csv, jsonl, or parquet.", code = "BAD_REQUEST" },
@@ -243,8 +233,12 @@ public static class AuditEndpoints
filter = restricted;
var repo = context.RequestServices.GetRequiredService();
- var contentType = format == "csv" ? "text/csv; charset=utf-8" : "application/x-ndjson";
- var extension = format == "csv" ? "csv" : "jsonl";
+ var (contentType, extension) = format switch
+ {
+ "csv" => ("text/csv; charset=utf-8", "csv"),
+ "parquet" => (ParquetContentType, "parquet"),
+ _ => ("application/x-ndjson", "jsonl"),
+ };
var fileName = $"audit-log-{DateTime.UtcNow:yyyyMMdd-HHmmss}.{extension}";
context.Response.ContentType = contentType;
@@ -256,6 +250,12 @@ public static class AuditEndpoints
{
await StreamCsvAsync(repo, filter, context.Response.Body, context.RequestAborted);
}
+ else if (format == "parquet")
+ {
+ // ScadaBridge#42: same 24 columns as csv/jsonl, one row group per 10k rows.
+ await AuditExportParquet.WriteAsync(
+ ExportRowPagesAsync(repo, filter, context.RequestAborted), context.Response.Body, context.RequestAborted);
+ }
else
{
await StreamJsonlAsync(repo, filter, context.Response.Body, context.RequestAborted);
@@ -494,6 +494,18 @@ public static class AuditEndpoints
}
}
+ /// Projects pages onto the flat export row shape.
+ private static async IAsyncEnumerable> ExportRowPagesAsync(
+ IAuditLogRepository repo,
+ AuditLogQueryFilter filter,
+ [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct)
+ {
+ await foreach (var page in PageAsync(repo, filter, ct))
+ {
+ yield return page.Select(AuditExportRow.From).ToList();
+ }
+ }
+
///
/// Lazily yields full repository pages, advancing the keyset cursor until a
/// short page signals the end. Shared by the csv + jsonl streamers.
diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/AuditExportParquet.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/AuditExportParquet.cs
new file mode 100644
index 00000000..1a829b82
--- /dev/null
+++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/AuditExportParquet.cs
@@ -0,0 +1,219 @@
+using Parquet;
+using Parquet.Schema;
+
+namespace ZB.MOM.WW.ScadaBridge.ManagementService;
+
+///
+/// GET /api/audit/export?format=parquet (ScadaBridge#42): the same 24 columns as the
+/// csv / jsonl export (, in
+/// order) as one Parquet file, one row group per rows.
+///
+///
+///
+/// Types: ids are D-format strings, timestamps are UTC Parquet timestamps
+/// (microseconds), enums are their names, HttpStatus/DurationMs are int32,
+/// PayloadTruncated is boolean, everything else is a UTF-8 string.
+///
+///
+/// Streaming. Kestrel forbids synchronous writes to the response body and the Parquet
+/// writer issues some, so each row group is encoded into a small in-memory buffer and then
+/// copied to the response asynchronously. Memory is bounded by one row group, and the client
+/// starts receiving bytes after the first one. A Parquet file's footer is written last, so a
+/// response cut off mid-stream is not a readable file — the CLI treats it as a failed export.
+///
+///
+/// This is the ad-hoc export. The retention archive (alog.md §15.2) is a different layout
+/// that carries the hash-chain columns; see AuditArchiveParquet.
+///
+///
+public static class AuditExportParquet
+{
+ /// Rows per Parquet row group.
+ public const int RowGroupRows = 10_000;
+
+ private static readonly DataField EventId = new(nameof(AuditExportRow.EventId), false);
+ private static readonly DateTimeDataField OccurredAtUtc = Timestamp(nameof(AuditExportRow.OccurredAtUtc), false);
+ private static readonly DateTimeDataField IngestedAtUtc = Timestamp(nameof(AuditExportRow.IngestedAtUtc), true);
+ private static readonly DataField Channel = new(nameof(AuditExportRow.Channel), false);
+ private static readonly DataField Kind = new(nameof(AuditExportRow.Kind), false);
+ private static readonly DataField CorrelationId = Text(nameof(AuditExportRow.CorrelationId));
+ private static readonly DataField ExecutionId = Text(nameof(AuditExportRow.ExecutionId));
+ private static readonly DataField ParentExecutionId = Text(nameof(AuditExportRow.ParentExecutionId));
+ private static readonly DataField SourceSiteId = Text(nameof(AuditExportRow.SourceSiteId));
+ private static readonly DataField SourceNode = Text(nameof(AuditExportRow.SourceNode));
+ private static readonly DataField SourceInstanceId = Text(nameof(AuditExportRow.SourceInstanceId));
+ private static readonly DataField SourceScript = Text(nameof(AuditExportRow.SourceScript));
+ private static readonly DataField Actor = Text(nameof(AuditExportRow.Actor));
+ private static readonly DataField Target = Text(nameof(AuditExportRow.Target));
+ private static readonly DataField Status = new(nameof(AuditExportRow.Status), false);
+ private static readonly DataField HttpStatus = new(nameof(AuditExportRow.HttpStatus), true);
+ private static readonly DataField DurationMs = new(nameof(AuditExportRow.DurationMs), true);
+ private static readonly DataField ErrorMessage = Text(nameof(AuditExportRow.ErrorMessage));
+ private static readonly DataField ErrorDetail = Text(nameof(AuditExportRow.ErrorDetail));
+ private static readonly DataField RequestSummary = Text(nameof(AuditExportRow.RequestSummary));
+ private static readonly DataField ResponseSummary = Text(nameof(AuditExportRow.ResponseSummary));
+ private static readonly DataField PayloadTruncated = new(nameof(AuditExportRow.PayloadTruncated), false);
+ private static readonly DataField Extra = Text(nameof(AuditExportRow.Extra));
+ private static readonly DataField ForwardState = Text(nameof(AuditExportRow.ForwardState));
+
+ /// The export schema; column names and order equal .
+ public static ParquetSchema Schema { get; } = new(
+ EventId, OccurredAtUtc, IngestedAtUtc, Channel, Kind, CorrelationId, ExecutionId, ParentExecutionId,
+ SourceSiteId, SourceNode, SourceInstanceId, SourceScript, Actor, Target, Status, HttpStatus, DurationMs,
+ ErrorMessage, ErrorDetail, RequestSummary, ResponseSummary, PayloadTruncated, Extra, ForwardState);
+
+ /// Writes every row of to as one Parquet file.
+ /// Export rows, page by page.
+ /// The destination (e.g. the HTTP response body); written asynchronously only.
+ /// Cancellation token.
+ /// The number of rows written.
+ public static async Task WriteAsync(
+ IAsyncEnumerable> pages, Stream output, CancellationToken ct = default)
+ {
+ ArgumentNullException.ThrowIfNull(pages);
+ ArgumentNullException.ThrowIfNull(output);
+
+ // Non-seekable on purpose: the writer then tracks file offsets itself, so emptying the
+ // buffer after each drain cannot corrupt the column-chunk offsets in the footer.
+ var staging = new StagingStream();
+ long total = 0;
+ var buffer = new List(RowGroupRows);
+ await using (var writer = await ParquetWriter.CreateAsync(Schema, staging, cancellationToken: ct).ConfigureAwait(false))
+ {
+ await foreach (var page in pages.WithCancellation(ct).ConfigureAwait(false))
+ {
+ foreach (var row in page)
+ {
+ buffer.Add(row);
+ if (buffer.Count == RowGroupRows)
+ {
+ total += await WriteGroupAsync(writer, buffer, ct).ConfigureAwait(false);
+ await DrainAsync(staging, output, ct).ConfigureAwait(false);
+ }
+ }
+ }
+
+ total += await WriteGroupAsync(writer, buffer, ct).ConfigureAwait(false);
+ }
+
+ // The footer is written when the writer is disposed.
+ await DrainAsync(staging, output, ct).ConfigureAwait(false);
+ return total;
+ }
+
+ private static async Task WriteGroupAsync(ParquetWriter writer, List rows, CancellationToken ct)
+ {
+ if (rows.Count == 0)
+ {
+ return 0;
+ }
+
+ using (var group = writer.CreateRowGroup())
+ {
+ await group.WriteAsync(EventId, Map(rows, r => r.EventId.ToString("D"))).ConfigureAwait(false);
+ await group.WriteAsync(OccurredAtUtc, rows.Select(r => DateTime.SpecifyKind(r.OccurredAtUtc, DateTimeKind.Utc)).ToArray(), cancellationToken: ct).ConfigureAwait(false);
+ await group.WriteAsync(IngestedAtUtc, rows.Select(r => r.IngestedAtUtc is { } i ? DateTime.SpecifyKind(i, DateTimeKind.Utc) : (DateTime?)null).ToArray(), cancellationToken: ct).ConfigureAwait(false);
+ await group.WriteAsync(Channel, Map(rows, r => r.Channel.ToString())).ConfigureAwait(false);
+ await group.WriteAsync(Kind, Map(rows, r => r.Kind.ToString())).ConfigureAwait(false);
+ await group.WriteAsync(CorrelationId, Map(rows, r => r.CorrelationId?.ToString("D"))!).ConfigureAwait(false);
+ await group.WriteAsync(ExecutionId, Map(rows, r => r.ExecutionId?.ToString("D"))!).ConfigureAwait(false);
+ await group.WriteAsync(ParentExecutionId, Map(rows, r => r.ParentExecutionId?.ToString("D"))!).ConfigureAwait(false);
+ await group.WriteAsync(SourceSiteId, Map(rows, r => r.SourceSiteId)!).ConfigureAwait(false);
+ await group.WriteAsync(SourceNode, Map(rows, r => r.SourceNode)!).ConfigureAwait(false);
+ await group.WriteAsync(SourceInstanceId, Map(rows, r => r.SourceInstanceId)!).ConfigureAwait(false);
+ await group.WriteAsync(SourceScript, Map(rows, r => r.SourceScript)!).ConfigureAwait(false);
+ await group.WriteAsync(Actor, Map(rows, r => r.Actor)!).ConfigureAwait(false);
+ await group.WriteAsync(Target, Map(rows, r => r.Target)!).ConfigureAwait(false);
+ await group.WriteAsync(Status, Map(rows, r => r.Status.ToString())).ConfigureAwait(false);
+ await group.WriteAsync(HttpStatus, rows.Select(r => r.HttpStatus).ToArray(), cancellationToken: ct).ConfigureAwait(false);
+ await group.WriteAsync(DurationMs, rows.Select(r => r.DurationMs).ToArray(), cancellationToken: ct).ConfigureAwait(false);
+ await group.WriteAsync(ErrorMessage, Map(rows, r => r.ErrorMessage)!).ConfigureAwait(false);
+ await group.WriteAsync(ErrorDetail, Map(rows, r => r.ErrorDetail)!).ConfigureAwait(false);
+ await group.WriteAsync(RequestSummary, Map(rows, r => r.RequestSummary)!).ConfigureAwait(false);
+ await group.WriteAsync(ResponseSummary, Map(rows, r => r.ResponseSummary)!).ConfigureAwait(false);
+ await group.WriteAsync(PayloadTruncated, rows.Select(r => r.PayloadTruncated).ToArray(), cancellationToken: ct).ConfigureAwait(false);
+ await group.WriteAsync(Extra, Map(rows, r => r.Extra)!).ConfigureAwait(false);
+ await group.WriteAsync(ForwardState, Map(rows, r => r.ForwardState?.ToString())!).ConfigureAwait(false);
+ }
+
+ var count = rows.Count;
+ rows.Clear();
+ return count;
+ }
+
+ private static async Task DrainAsync(StagingStream staging, Stream output, CancellationToken ct)
+ {
+ var pending = staging.Pending;
+ if (pending.Length == 0)
+ {
+ return;
+ }
+
+ await output.WriteAsync(pending, ct).ConfigureAwait(false);
+ await output.FlushAsync(ct).ConfigureAwait(false);
+ staging.Clear();
+ }
+
+ private static string?[] Map(List rows, Func select)
+ => rows.Select(select).ToArray();
+
+ private static DataField Text(string name) => new(name, true);
+
+ private static DateTimeDataField Timestamp(string name, bool nullable)
+ => new(name, DateTimeFormat.Timestamp, isAdjustedToUTC: true, unit: DateTimeTimeUnit.Micros, isNullable: nullable);
+
+ ///
+ /// Forward-only, non-seekable buffer the Parquet writer writes into (synchronously or
+ /// not); moves its bytes to the real output asynchronously.
+ ///
+ private sealed class StagingStream : Stream
+ {
+ private readonly MemoryStream _buffer = new();
+
+ public ReadOnlyMemory Pending => _buffer.GetBuffer().AsMemory(0, (int)_buffer.Length);
+
+ public override bool CanRead => false;
+
+ public override bool CanSeek => false;
+
+ public override bool CanWrite => true;
+
+ public override long Length => throw new NotSupportedException();
+
+ public override long Position
+ {
+ get => throw new NotSupportedException();
+ set => throw new NotSupportedException();
+ }
+
+ public void Clear() => _buffer.SetLength(0);
+
+ public override void Write(byte[] buffer, int offset, int count) => _buffer.Write(buffer, offset, count);
+
+ public override void Write(ReadOnlySpan buffer) => _buffer.Write(buffer);
+
+ public override ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default)
+ {
+ _buffer.Write(buffer.Span);
+ return ValueTask.CompletedTask;
+ }
+
+ public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+ {
+ _buffer.Write(buffer, offset, count);
+ return Task.CompletedTask;
+ }
+
+ public override void Flush()
+ {
+ }
+
+ public override Task FlushAsync(CancellationToken cancellationToken) => Task.CompletedTask;
+
+ public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException();
+
+ public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
+
+ public override void SetLength(long value) => throw new NotSupportedException();
+ }
+}
diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs
index cafd61c4..d4175567 100644
--- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs
@@ -393,6 +393,11 @@ public class ManagementActor : ReceiveActor
// spans every site.
VerifyAuditChainCommand => OperationalAuditReaders,
+ // AuditLog Parquet archives (ScadaBridge#42): audit reads with verify-chain's gate —
+ // OperationalAudit roles, site-scoped callers refused in the handler.
+ ListAuditArchivesCommand => OperationalAuditReaders,
+ VerifyAuditArchiveCommand => OperationalAuditReaders,
+
// Read-only queries -- any authenticated user
_ => null
};
@@ -564,6 +569,8 @@ public class ManagementActor : ReceiveActor
// Audit Log
QueryAuditLogCommand cmd => await HandleQueryAuditLog(sp, cmd),
VerifyAuditChainCommand cmd => await HandleVerifyAuditChain(sp, cmd, user),
+ ListAuditArchivesCommand => await HandleListAuditArchives(sp, user),
+ VerifyAuditArchiveCommand cmd => await HandleVerifyAuditArchive(sp, cmd, user),
// Health
GetHealthSummaryCommand => HandleGetHealthSummary(sp),
@@ -3256,6 +3263,59 @@ public class ManagementActor : ReceiveActor
}
}
+ ///
+ /// Lists the AuditLog Parquet archives (ScadaBridge#42). Refused for a site-scoped,
+ /// non-Administrator caller: an archive holds every site's rows.
+ ///
+ private static async Task HandleListAuditArchives(IServiceProvider sp, AuthenticatedUser user)
+ {
+ var archive = RequireAuditArchive(sp, user);
+ return new AuditArchiveListResult(archive.IsEnabled, await archive.ListAsync());
+ }
+
+ ///
+ /// Verifies one AuditLog Parquet archive (ScadaBridge#42): files against the manifest,
+ /// the chain recomputed from the archived rows, the chain end against the purge anchor.
+ ///
+ private static async Task HandleVerifyAuditArchive(
+ IServiceProvider sp, VerifyAuditArchiveCommand cmd, AuthenticatedUser user)
+ {
+ var archive = RequireAuditArchive(sp, user);
+ try
+ {
+ return await archive.VerifyAsync(cmd.Name);
+ }
+ catch (ArgumentException ex)
+ {
+ throw new ManagementCommandException(ex.Message, "INVALID_ARGUMENT");
+ }
+ catch (FileNotFoundException ex)
+ {
+ throw new ManagementCommandException(ex.Message, "NOT_FOUND");
+ }
+ }
+
+ private static IAuditLogArchive RequireAuditArchive(IServiceProvider sp, AuthenticatedUser user)
+ {
+ if (user.PermittedSiteIds.Length > 0
+ && !user.Roles.Contains(Roles.Administrator, StringComparer.OrdinalIgnoreCase))
+ {
+ throw new SiteScopeViolationException(
+ "Access denied: AuditLog archives hold every site's rows; they need an "
+ + "OperationalAudit role that is not scoped to specific sites.");
+ }
+
+ var archive = sp.GetService();
+ if (archive is null || !archive.IsConfigured)
+ {
+ throw new ManagementCommandException(
+ "No AuditLog archive directory is configured on this central node (AuditLog:Archive:Directory).",
+ "ARCHIVE_NOT_CONFIGURED");
+ }
+
+ return archive;
+ }
+
// ========================================================================
// Health handlers
// ========================================================================
diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementEndpoints.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementEndpoints.cs
index f0e12098..5e178854 100644
--- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementEndpoints.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementEndpoints.cs
@@ -63,6 +63,9 @@ public static class ManagementEndpoints
// AuditLog hash-chain verification (ScadaBridge#41) recomputes one hash per row of
// a whole month — millions of rows on a busy central.
typeof(VerifyAuditChainCommand),
+ // AuditLog archive verification (ScadaBridge#42) hashes and re-chains a whole month's
+ // Parquet archive.
+ typeof(VerifyAuditArchiveCommand),
];
///
diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ZB.MOM.WW.ScadaBridge.ManagementService.csproj b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ZB.MOM.WW.ScadaBridge.ManagementService.csproj
index 67db2fc2..4f76e22e 100644
--- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ZB.MOM.WW.ScadaBridge.ManagementService.csproj
+++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ZB.MOM.WW.ScadaBridge.ManagementService.csproj
@@ -12,6 +12,7 @@
+
diff --git a/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/CommandTreeTests.cs b/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/CommandTreeTests.cs
index c185c1ca..0d17ebc8 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/CommandTreeTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/CommandTreeTests.cs
@@ -69,7 +69,7 @@ public class CommandTreeTests
Assert.Contains("bundle", groupNames);
}
- /// Verifies that the audit command group exposes the query, export and verify-chain subcommands.
+ /// Verifies that the audit command group exposes the query, export, verify-chain and archive subcommands.
[Fact]
public void AuditCommandGroup_HasQueryExportAndVerifyChain()
{
@@ -80,6 +80,7 @@ public class CommandTreeTests
Assert.Contains("query", subNames);
Assert.Contains("export", subNames);
Assert.Contains("verify-chain", subNames);
+ Assert.Contains("archive", subNames);
}
/// Verifies that the bundle command group exposes the export, preview and import subcommands.
diff --git a/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/Commands/AuditArchiveCommandTests.cs b/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/Commands/AuditArchiveCommandTests.cs
new file mode 100644
index 00000000..8c0b4651
--- /dev/null
+++ b/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/Commands/AuditArchiveCommandTests.cs
@@ -0,0 +1,136 @@
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+using System.Text.Json;
+using ZB.MOM.WW.ScadaBridge.CLI.Commands;
+
+namespace ZB.MOM.WW.ScadaBridge.CLI.Tests.Commands;
+
+///
+/// Tests for scadabridge audit archive list / verify (ScadaBridge#42): each
+/// sends its management command (ListAuditArchives, VerifyAuditArchive), and
+/// verify maps the verdict onto the exit code (0 verified, 3 not verified).
+///
+[Collection("Console")]
+public class AuditArchiveCommandTests
+{
+ private const string Name = "auditlog-2025-10-20261109T120000123Z";
+
+ private const string ListBody =
+ "{\"enabled\":true,\"archives\":[{\"name\":\"" + Name + "\",\"partition\":\"2025-10\",\"status\":\"Purged\"," +
+ "\"rowCount\":11,\"chainedRowCount\":10,\"dataBytes\":4096}]}";
+
+ private const string VerifiedBody =
+ "{\"name\":\"" + Name + "\",\"partition\":\"2025-10\",\"manifestStatus\":\"Purged\",\"verified\":true,\"filesIntact\":true," +
+ "\"rowsRead\":11,\"chains\":[{\"chain\":{\"partition\":\"2025-10\",\"status\":\"Intact\",\"rowsVerified\":10," +
+ "\"tombstonesApplied\":2,\"unchainedRows\":1},\"anchor\":\"Matches\",\"anchorDetail\":\"ends at the anchor\"}]}";
+
+ private const string MismatchBody =
+ "{\"name\":\"" + Name + "\",\"partition\":\"2025-10\",\"manifestStatus\":\"Purged\",\"verified\":false,\"filesIntact\":true," +
+ "\"rowsRead\":11,\"chains\":[{\"chain\":{\"partition\":\"2025-10\",\"status\":\"Intact\",\"rowsVerified\":10," +
+ "\"tombstonesApplied\":0,\"unchainedRows\":0},\"anchor\":\"Mismatch\",\"anchorDetail\":\"hash differs\"}]}";
+
+ /// The audit group carries the archive group with list + verify.
+ [Fact]
+ public void AuditGroup_HasArchiveListAndVerify()
+ {
+ var root = AuditCommandTestHarness.BuildRoot();
+ var archive = root.Subcommands.Single(c => c.Name == "audit").Subcommands.Single(c => c.Name == "archive");
+ Assert.Equal(["list", "verify"], archive.Subcommands.Select(c => c.Name).Order());
+ }
+
+ /// list sends ListAuditArchives and prints the archives.
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task List_SendsListAuditArchives_AndPrintsTheArchives()
+ {
+ var (exit, output, request) = await RunAsync(ListBody, "audit", "archive", "list", "--format", "table");
+
+ Assert.Equal(0, exit);
+ Assert.Contains(Name, output);
+ Assert.Contains("Archiving: enabled", output);
+ using var payload = JsonDocument.Parse(request);
+ Assert.Equal("ListAuditArchives", payload.RootElement.GetProperty("command").GetString());
+ }
+
+ /// verify sends VerifyAuditArchive with the name and maps the verdict to the exit code.
+ /// The verification result the fake endpoint serves.
+ /// Output format.
+ /// Expected exit code.
+ /// Text the report must contain.
+ /// A task that represents the asynchronous operation.
+ [Theory]
+ [InlineData(VerifiedBody, "table", 0, "anchor Matches")]
+ [InlineData(VerifiedBody, "json", 0, "\"verified\":true")]
+ [InlineData(MismatchBody, "table", 3, "Verified : NO")]
+ [InlineData(MismatchBody, "json", 3, "\"anchor\":\"Mismatch\"")]
+ public async Task Verify_SendsVerifyAuditArchive_AndMapsVerdictToExitCode(
+ string body, string format, int expectedExit, string expectedText)
+ {
+ var (exit, output, request) = await RunAsync(body, "audit", "archive", "verify", "--name", Name, "--format", format);
+
+ Assert.Equal(expectedExit, exit);
+ Assert.Contains(expectedText, output);
+ using var payload = JsonDocument.Parse(request);
+ Assert.Equal("VerifyAuditArchive", payload.RootElement.GetProperty("command").GetString());
+ var name = payload.RootElement.GetProperty("payload").EnumerateObject()
+ .First(p => string.Equals(p.Name, "name", StringComparison.OrdinalIgnoreCase)).Value.GetString();
+ Assert.Equal(Name, name);
+ }
+
+ /// verify without --name is a usage error.
+ [Fact]
+ public void Verify_MissingName_ProducesRequiredFlagError()
+ {
+ var root = AuditCommandTestHarness.BuildRoot();
+ var (exit, _, err) = AuditCommandTestHarness.Invoke(root, "audit", "archive", "verify");
+
+ Assert.NotEqual(0, exit);
+ Assert.Contains("--name", err);
+ }
+
+ private static async Task<(int Exit, string Output, string Request)> RunAsync(string body, params string[] args)
+ {
+ var listener = new TcpListener(IPAddress.Loopback, 0);
+ listener.Start();
+ try
+ {
+ var port = ((IPEndPoint)listener.LocalEndpoint).Port;
+ string? requestBody = null;
+ var server = Task.Run(async () =>
+ {
+ using var client = await listener.AcceptTcpClientAsync();
+ await using var stream = client.GetStream();
+ using var reader = new StreamReader(stream, Encoding.UTF8, leaveOpen: true);
+ var length = 0;
+ while (await reader.ReadLineAsync() is { Length: > 0 } line)
+ if (line.StartsWith("Content-Length:", StringComparison.OrdinalIgnoreCase))
+ length = int.Parse(line[15..].Trim());
+ var buffer = new char[length];
+ var read = 0;
+ while (read < length)
+ {
+ var count = await reader.ReadAsync(buffer.AsMemory(read));
+ if (count == 0) throw new EndOfStreamException();
+ read += count;
+ }
+ requestBody = new string(buffer);
+ var bytes = Encoding.UTF8.GetBytes(body);
+ var headers = Encoding.ASCII.GetBytes(
+ $"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {bytes.Length}\r\nConnection: close\r\n\r\n");
+ await stream.WriteAsync(headers);
+ await stream.WriteAsync(bytes);
+ });
+
+ var root = AuditCommandTestHarness.BuildRoot();
+ var (exit, output, _) = AuditCommandTestHarness.Invoke(
+ root, [.. args, "--url", $"http://127.0.0.1:{port}", "--username", "u", "--password", "p"]);
+ await server.WaitAsync(TimeSpan.FromSeconds(5));
+ return (exit, output, requestBody!);
+ }
+ finally
+ {
+ listener.Stop();
+ }
+ }
+}
diff --git a/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/Commands/AuditCommandsScaffoldTests.cs b/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/Commands/AuditCommandsScaffoldTests.cs
index 8b93275e..a2540984 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/Commands/AuditCommandsScaffoldTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/Commands/AuditCommandsScaffoldTests.cs
@@ -27,14 +27,14 @@ public class AuditCommandsScaffoldTests
Assert.False(string.IsNullOrWhiteSpace(audit.Description));
}
- /// Verifies that the audit group exposes the query, export, tree, verify-chain and backfill-source-node subcommands.
+ /// Verifies that the audit group exposes the query, export, tree, verify-chain, backfill-source-node and archive subcommands.
[Fact]
- public void Audit_HasFiveSubcommands_QueryExportTreeVerifyChainBackfillSourceNode()
+ public void Audit_HasSixSubcommands_QueryExportTreeVerifyChainBackfillSourceNodeArchive()
{
var audit = BuildAudit();
var names = audit.Subcommands.Select(c => c.Name).OrderBy(n => n).ToArray();
Assert.Equal(
- new[] { "backfill-source-node", "export", "query", "tree", "verify-chain" },
+ new[] { "archive", "backfill-source-node", "export", "query", "tree", "verify-chain" },
names);
}
@@ -62,8 +62,11 @@ public class AuditCommandsScaffoldTests
[Fact]
public void Audit_EveryLeafCommand_HasAnAction()
{
+ // `archive` is a group (list / verify, ScadaBridge#42): check its leaves, not it.
var audit = BuildAudit();
- Assert.All(audit.Subcommands, sub =>
+ var leaves = audit.Subcommands.SelectMany(c => c.Subcommands.Count == 0 ? [c] : c.Subcommands).ToList();
+ Assert.Equal(7, leaves.Count);
+ Assert.All(leaves, sub =>
Assert.True(sub.Action != null, $"Leaf command '{sub.Name}' has no action."));
}
}
diff --git a/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/ManagementSurfaceParityTests.cs b/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/ManagementSurfaceParityTests.cs
index d65ee09e..0e3d17b4 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/ManagementSurfaceParityTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/ManagementSurfaceParityTests.cs
@@ -44,6 +44,9 @@ public class ManagementSurfaceParityTests
Assert.Contains("ListSharedScripts", sent.Keys);
}
+ private static readonly HashSet AuditManagementLeaves =
+ ["audit verify-chain", "audit archive list", "audit archive verify"];
+
private static async Task> ProbeProductionTree(string? omitGroup = null)
{
var root = CliApplication.BuildRoot();
@@ -93,9 +96,10 @@ public class ManagementSurfaceParityTests
foreach (var (leaf, path) in Leaves(root, []))
{
// Separate REST and persistent SignalR surfaces, not management contracts.
- // `audit verify-chain` is the one audit leaf that IS a management contract
- // (VerifyAuditChain, ScadaBridge#41), so it stays in the probe.
- if (path.Length == 0 || (path[0] == "audit" && string.Join(' ', path) != "audit verify-chain")
+ // `audit verify-chain` (VerifyAuditChain, ScadaBridge#41) and `audit archive
+ // list|verify` (ListAuditArchives / VerifyAuditArchive, ScadaBridge#42) are the
+ // audit leaves that ARE management contracts, so they stay in the probe.
+ if (path.Length == 0 || (path[0] == "audit" && !AuditManagementLeaves.Contains(string.Join(' ', path)))
|| path[0] == omitGroup
|| string.Join(' ', path) == "debug stream") continue;
currentPath = string.Join(' ', path);
diff --git a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditArchiveCommandTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditArchiveCommandTests.cs
new file mode 100644
index 00000000..d69a04ec
--- /dev/null
+++ b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditArchiveCommandTests.cs
@@ -0,0 +1,147 @@
+using System.Text.Json;
+using Akka.Actor;
+using Akka.TestKit.Xunit2;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging.Abstractions;
+using NSubstitute;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
+
+namespace ZB.MOM.WW.ScadaBridge.ManagementService.Tests;
+
+///
+/// The ListAuditArchives / VerifyAuditArchive management commands
+/// (ScadaBridge#42): role gate, site-scope refusal, an unconfigured archive directory, bad
+/// names, and the results reaching the wire with enums as names.
+///
+public class AuditArchiveCommandTests : TestKit
+{
+ private const string Name = "auditlog-2025-10-20261109T120000123Z";
+
+ private readonly IAuditLogArchive _archive = Substitute.For();
+
+ /// Configures the archive substitute as a configured, enabled archive.
+ public AuditArchiveCommandTests()
+ {
+ _archive.IsConfigured.Returns(true);
+ _archive.IsEnabled.Returns(true);
+ }
+
+ private IActorRef CreateActor(bool registerArchive = true)
+ {
+ var services = new ServiceCollection().AddTestConfigurationTransaction();
+ if (registerArchive)
+ {
+ services.AddScoped(_ => _archive);
+ }
+
+ return Sys.ActorOf(Props.Create(() => new ManagementActor(
+ services.BuildServiceProvider(), NullLogger.Instance)));
+ }
+
+ private static ManagementEnvelope Envelope(object command, string[] sites, params string[] roles) =>
+ new(new AuthenticatedUser("u", "U", roles, sites), command, Guid.NewGuid().ToString("N"));
+
+ private static AuditArchiveManifest Manifest() => new(
+ 1, Name, new DateTime(2025, 10, 1, 0, 0, 0, DateTimeKind.Utc), "2025-10",
+ new DateTime(2026, 11, 9, 12, 0, 0, DateTimeKind.Utc), AuditArchiveStatus.Purged,
+ new DateTime(2026, 11, 9, 12, 0, 1, DateTimeKind.Utc), 10, 9, 1, null, null,
+ Name + ".parquet", "ab", 1234, Name + ".tombstones.parquet", "cd", 0,
+ [new AuditArchiveChainState("2025-10", 9, new string('0', 64), null, null, 0)]);
+
+ /// A Viewer lists the archives; status renders as a name.
+ [Fact]
+ public void Viewer_ListsArchives()
+ {
+ _archive.ListAsync(Arg.Any()).Returns([Manifest()]);
+
+ CreateActor().Tell(Envelope(new ListAuditArchivesCommand(), [], "Viewer"));
+
+ var response = ExpectMsg(TimeSpan.FromSeconds(5));
+ using var json = JsonDocument.Parse(response.JsonData);
+ Assert.True(json.RootElement.GetProperty("enabled").GetBoolean());
+ var archive = json.RootElement.GetProperty("archives")[0];
+ Assert.Equal(Name, archive.GetProperty("name").GetString());
+ Assert.Equal("Purged", archive.GetProperty("status").GetString());
+ }
+
+ /// Verification results reach the wire with the anchor and chain status as names.
+ [Fact]
+ public void Viewer_GetsVerificationResult()
+ {
+ var chain = new AuditChainVerificationResult("2025-10", AuditChainStatus.Intact, 9, 0, null, 9, 1, null);
+ _archive.VerifyAsync(Name, Arg.Any()).Returns(new AuditArchiveVerificationResult(
+ Name, "2025-10", AuditArchiveStatus.Purged, true, true, null, 10,
+ [new AuditArchiveChainVerification(chain, AuditArchiveAnchorStatus.Matches, "ends at the anchor")]));
+
+ CreateActor().Tell(Envelope(new VerifyAuditArchiveCommand(Name), [], "Viewer"));
+
+ var response = ExpectMsg(TimeSpan.FromSeconds(5));
+ using var json = JsonDocument.Parse(response.JsonData);
+ Assert.True(json.RootElement.GetProperty("verified").GetBoolean());
+ var c = json.RootElement.GetProperty("chains")[0];
+ Assert.Equal("Matches", c.GetProperty("anchor").GetString());
+ Assert.Equal("Intact", c.GetProperty("chain").GetProperty("status").GetString());
+ }
+
+ /// Designer/Deployer without an OperationalAudit role are refused both commands.
+ [Fact]
+ public void NonAuditRole_IsUnauthorized()
+ {
+ var actor = CreateActor();
+ actor.Tell(Envelope(new ListAuditArchivesCommand(), [], "Designer", "Deployer"));
+ ExpectMsg(TimeSpan.FromSeconds(5));
+ actor.Tell(Envelope(new VerifyAuditArchiveCommand(Name), [], "Designer", "Deployer"));
+ ExpectMsg(TimeSpan.FromSeconds(5));
+ _archive.DidNotReceiveWithAnyArgs().ListAsync(default);
+ _archive.DidNotReceiveWithAnyArgs().VerifyAsync(default!, default);
+ }
+
+ /// A site-scoped Viewer is refused: an archive holds every site's rows.
+ [Fact]
+ public void SiteScopedViewer_IsRefused()
+ {
+ CreateActor().Tell(Envelope(new VerifyAuditArchiveCommand(Name), ["3"], "Viewer"));
+
+ var response = ExpectMsg(TimeSpan.FromSeconds(5));
+ Assert.Contains("every site", response.Message);
+ _archive.DidNotReceiveWithAnyArgs().VerifyAsync(default!, default);
+ }
+
+ /// No archive directory configured is an explicit error code, not an empty list.
+ [Fact]
+ public void UnconfiguredArchive_IsArchiveNotConfigured()
+ {
+ _archive.IsConfigured.Returns(false);
+
+ CreateActor().Tell(Envelope(new ListAuditArchivesCommand(), [], "Administrator"));
+
+ var response = ExpectMsg(TimeSpan.FromSeconds(5));
+ Assert.Equal("ARCHIVE_NOT_CONFIGURED", response.ErrorCode);
+ }
+
+ /// A malformed name is INVALID_ARGUMENT; an unknown one is NOT_FOUND.
+ [Fact]
+ public void BadNames_MapToCallerErrors()
+ {
+ _archive.VerifyAsync("../etc", Arg.Any())
+ .Returns>(_ => throw new ArgumentException("not an archive name"));
+ _archive.VerifyAsync(Name, Arg.Any())
+ .Returns>(_ => throw new FileNotFoundException("no archive"));
+
+ var actor = CreateActor();
+ actor.Tell(Envelope(new VerifyAuditArchiveCommand("../etc"), [], "Administrator"));
+ Assert.Equal("INVALID_ARGUMENT", ExpectMsg(TimeSpan.FromSeconds(5)).ErrorCode);
+ actor.Tell(Envelope(new VerifyAuditArchiveCommand(Name), [], "Administrator"));
+ Assert.Equal("NOT_FOUND", ExpectMsg(TimeSpan.FromSeconds(5)).ErrorCode);
+ }
+
+ /// Archive verification runs under the long-running Ask timeout.
+ [Fact]
+ public void VerifyAuditArchive_UsesLongRunningAskTimeout()
+ {
+ Assert.Equal(TimeSpan.FromMinutes(5),
+ ManagementEndpoints.ResolveAskTimeout(new ManagementServiceOptions(), typeof(VerifyAuditArchiveCommand)));
+ }
+}
diff --git a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditEndpointsTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditEndpointsTests.cs
index 4992b5d3..2155e718 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditEndpointsTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditEndpointsTests.cs
@@ -506,21 +506,43 @@ public class AuditEndpointsTests
Assert.Equal(propertyNames, AuditEndpoints.CsvHeader.Split(','));
}
- /// Verifies that a Parquet export returns 501.
+ ///
+ /// format=parquet (ScadaBridge#42; it used to be a 501) streams one readable Parquet
+ /// file whose columns are the csv/jsonl export's columns, in order, holding every row.
+ ///
/// A task that represents the asynchronous operation.
[Fact]
- public async Task Export_Parquet_Returns501()
+ public async Task Export_Parquet_StreamsReadableParquet_WithTheCsvColumns()
{
- // Parquet archival is deferred to v1.x (Component-AuditLog.md) — no
- // library is referenced, so the endpoint returns 501 with guidance.
- var (client, _, host) = await BuildHostAsync(roles: new[] { "Administrator" });
+ var (client, _, host) = await BuildHostAsync(
+ roles: new[] { "Administrator" },
+ queryPages: new[]
+ {
+ (IReadOnlyList)new[]
+ {
+ SampleEvent(Guid.Parse("cccccccc-0000-0000-0000-000000000001")),
+ SampleEvent(Guid.Parse("cccccccc-0000-0000-0000-000000000002")),
+ },
+ (IReadOnlyList)Array.Empty(),
+ });
using (host)
{
- var response = await client.SendAsync(Get("/api/audit/export?format=parquet"));
+ var response = await client.SendAsync(Get("/api/audit/export?format=Parquet"));
- Assert.Equal(HttpStatusCode.NotImplemented, response.StatusCode);
- var body = await response.Content.ReadAsStringAsync();
- Assert.Contains("Parquet export deferred to v1.x", body);
+ Assert.Equal(HttpStatusCode.OK, response.StatusCode);
+ Assert.Equal(AuditEndpoints.ParquetContentType, response.Content.Headers.ContentType!.MediaType);
+ Assert.Contains(".parquet", response.Content.Headers.ContentDisposition?.ToString()
+ ?? response.Headers.GetValues("Content-Disposition").FirstOrDefault()
+ ?? response.Content.Headers.GetValues("Content-Disposition").First());
+
+ var bytes = await response.Content.ReadAsByteArrayAsync();
+ await using var reader = await Parquet.ParquetReader.CreateAsync(new MemoryStream(bytes));
+ Assert.Equal(AuditEndpoints.CsvHeader.Split(','), reader.Schema.GetDataFields().Select(f => f.Name));
+ using var group = reader.OpenRowGroupReader(0);
+ Assert.Equal(2, group.RowCount);
+ var ids = new string?[2];
+ await group.ReadAsync(reader.Schema.GetDataFields()[0], ids!);
+ Assert.Equal(new[] { "cccccccc-0000-0000-0000-000000000001", "cccccccc-0000-0000-0000-000000000002" }, ids);
}
}
diff --git a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditExportParquetTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditExportParquetTests.cs
new file mode 100644
index 00000000..3c3fe68e
--- /dev/null
+++ b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditExportParquetTests.cs
@@ -0,0 +1,150 @@
+using Parquet;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
+
+namespace ZB.MOM.WW.ScadaBridge.ManagementService.Tests;
+
+///
+/// (ScadaBridge#42): the Parquet export round-trips every
+/// one of the 24 export columns, splits into row groups, and never writes synchronously to
+/// the output (Kestrel rejects synchronous response writes).
+///
+public class AuditExportParquetTests
+{
+ /// Rows written and read back are equal, column for column, across row groups.
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task RoundTrip_EveryColumn_AcrossRowGroups()
+ {
+ var rows = Enumerable.Range(0, AuditExportParquet.RowGroupRows + 7).Select(Row).ToList();
+ var output = new AsyncOnlyStream();
+
+ var written = await AuditExportParquet.WriteAsync(Pages(rows, 1000), output);
+
+ Assert.Equal(rows.Count, written);
+ await using var reader = await ParquetReader.CreateAsync(new MemoryStream(output.ToArray()));
+ Assert.Equal(2, reader.RowGroupCount);
+ var back = new List();
+ var fields = reader.Schema.GetDataFields();
+ for (var g = 0; g < reader.RowGroupCount; g++)
+ {
+ using var group = reader.OpenRowGroupReader(g);
+ var n = (int)group.RowCount;
+ async Task S(int i) { var a = new string?[n]; await group.ReadAsync(fields[i], a!); return a; }
+ async Task V(int i) where T : struct { var a = new T?[n]; await group.ReadAsync(fields[i], a); return a; }
+ var id = await S(0); var occurred = await V(1); var ingested = await V(2);
+ var channel = await S(3); var kind = await S(4); var corr = await S(5); var exec = await S(6); var parent = await S(7);
+ var site = await S(8); var node = await S(9); var inst = await S(10); var script = await S(11); var actor = await S(12);
+ var target = await S(13); var status = await S(14); var http = await V(15); var dur = await V(16);
+ var errMsg = await S(17); var errDetail = await S(18); var req = await S(19); var resp = await S(20);
+ var trunc = await V(21); var extra = await S(22); var fwd = await S(23);
+ for (var i = 0; i < n; i++)
+ {
+ back.Add(new AuditExportRow
+ {
+ EventId = Guid.Parse(id[i]!),
+ OccurredAtUtc = occurred[i]!.Value,
+ IngestedAtUtc = ingested[i],
+ Channel = Enum.Parse(channel[i]!),
+ Kind = Enum.Parse(kind[i]!),
+ CorrelationId = corr[i] is { } c ? Guid.Parse(c) : null,
+ ExecutionId = exec[i] is { } e ? Guid.Parse(e) : null,
+ ParentExecutionId = parent[i] is { } p ? Guid.Parse(p) : null,
+ SourceSiteId = site[i],
+ SourceNode = node[i],
+ SourceInstanceId = inst[i],
+ SourceScript = script[i],
+ Actor = actor[i],
+ Target = target[i],
+ Status = Enum.Parse(status[i]!),
+ HttpStatus = http[i],
+ DurationMs = dur[i],
+ ErrorMessage = errMsg[i],
+ ErrorDetail = errDetail[i],
+ RequestSummary = req[i],
+ ResponseSummary = resp[i],
+ PayloadTruncated = trunc[i]!.Value,
+ Extra = extra[i],
+ ForwardState = fwd[i] is { } f ? Enum.Parse(f) : null,
+ });
+ }
+ }
+
+ Assert.Equal(rows, back);
+ }
+
+ /// An export with no rows is still a valid (empty) Parquet file.
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task NoRows_IsAValidEmptyFile()
+ {
+ var output = new AsyncOnlyStream();
+ Assert.Equal(0, await AuditExportParquet.WriteAsync(Pages([], 10), output));
+ await using var reader = await ParquetReader.CreateAsync(new MemoryStream(output.ToArray()));
+ Assert.Equal(0, reader.RowGroupCount);
+ Assert.Equal(24, reader.Schema.GetDataFields().Length);
+ }
+
+ private static AuditExportRow Row(int i) => new()
+ {
+ EventId = Guid.NewGuid(),
+ // Microsecond precision: the export timestamp column is TIMESTAMP(MICROS).
+ OccurredAtUtc = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc).AddTicks(i * 10L),
+ IngestedAtUtc = i % 3 == 0 ? null : new DateTime(2026, 5, 20, 12, 0, 1, DateTimeKind.Utc),
+ Channel = (AuditChannel)(i % Enum.GetValues().Length),
+ Kind = AuditKind.ApiCall,
+ CorrelationId = i % 2 == 0 ? Guid.NewGuid() : null,
+ ExecutionId = Guid.NewGuid(),
+ ParentExecutionId = i % 5 == 0 ? Guid.NewGuid() : null,
+ SourceSiteId = i % 4 == 0 ? null : "plant-a",
+ SourceNode = "node-a",
+ SourceInstanceId = "Plant1.Boiler",
+ SourceScript = i % 2 == 0 ? "OnTick" : null,
+ Actor = i % 7 == 0 ? string.Empty : "api-key-1",
+ Target = "erp",
+ Status = AuditStatus.Delivered,
+ HttpStatus = i % 2 == 0 ? 200 : null,
+ DurationMs = i,
+ ErrorMessage = i % 9 == 0 ? "boom, \"quoted\"\nline" : null,
+ ErrorDetail = null,
+ RequestSummary = "{\"ü\":\"日本\"}",
+ ResponseSummary = null,
+ PayloadTruncated = i % 2 == 1,
+ Extra = null,
+ ForwardState = i % 11 == 0 ? AuditForwardState.Forwarded : null,
+ };
+
+ private static async IAsyncEnumerable> Pages(List rows, int size)
+ {
+ for (var i = 0; i < rows.Count; i += size)
+ {
+ await Task.Yield();
+ yield return rows.Skip(i).Take(size).ToList();
+ }
+ }
+
+ /// A response-body stand-in that fails any synchronous write, like Kestrel.
+ private sealed class AsyncOnlyStream : Stream
+ {
+ private readonly MemoryStream _inner = new();
+
+ public byte[] ToArray() => _inner.ToArray();
+
+ public override bool CanRead => false;
+ public override bool CanSeek => false;
+ public override bool CanWrite => true;
+ public override long Length => throw new NotSupportedException();
+ public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
+ public override void Flush() => throw new InvalidOperationException("Synchronous operations are disallowed.");
+ public override Task FlushAsync(CancellationToken cancellationToken) => Task.CompletedTask;
+ public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException();
+ public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
+ public override void SetLength(long value) => throw new NotSupportedException();
+ public override void Write(byte[] buffer, int offset, int count) => throw new InvalidOperationException("Synchronous operations are disallowed.");
+ public override ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default)
+ {
+ _inner.Write(buffer.Span);
+ return ValueTask.CompletedTask;
+ }
+ }
+}
diff --git a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/RequiredRoleMatrixTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/RequiredRoleMatrixTests.cs
index 1f44a349..afd85b6a 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/RequiredRoleMatrixTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/RequiredRoleMatrixTests.cs
@@ -171,6 +171,9 @@ public class RequiredRoleMatrixTests
// AuditLog hash-chain verification (ScadaBridge#41) carries audit query's
// OperationalAudit permission (alog.md §15.1); written out by hand to keep the freeze.
["VerifyAuditChain"] = [Roles.Administrator, Roles.Viewer],
+ // AuditLog Parquet archives (ScadaBridge#42): same audit-read gate as verify-chain.
+ ["ListAuditArchives"] = [Roles.Administrator, Roles.Viewer],
+ ["VerifyAuditArchive"] = [Roles.Administrator, Roles.Viewer],
// ---- Two-person secured write ----------------------------------------------
["SubmitSecuredWrite"] = [Roles.Operator],