using Grpc.Core;
using Mishima.Nexus.Wire.V1;
namespace Robotico.Nexus.Client.Tests.TestSupport;
/// In-memory fake standing in for a real Nexus time engine host.
public sealed class FakeTimeEngineService : NexusTimeEngineService.NexusTimeEngineServiceBase
{
private readonly Dictionary> _samplesBySeries = [];
private readonly Dictionary Labels)> _seriesMeta = [];
private readonly Dictionary> _eventsByTable = [];
private readonly Dictionary> _fieldsByTable = [];
private ulong _nextSeriesId = 1;
private ulong _nextTableId = 1;
private ulong _nextEventId = 1;
private long _retentionRemoved;
///
public override Task Ping(TimePingRequest request, ServerCallContext context)
{
NexusStatus? failure = FakeNexusTriggers.TryStatusFailure(context);
return Task.FromResult(new TimePingResponse { Status = failure ?? NexusStatuses.Ok, EngineVersion = "fake" });
}
///
public override Task Doctor(TimeDoctorRequest request, ServerCallContext context) =>
Task.FromResult(new TimeDoctorResponse
{
Status = NexusStatuses.Ok,
ActiveSeriesCount = _samplesBySeries.Count,
TotalSampleCount = _samplesBySeries.Values.Sum(list => list.Count),
ActiveEventTableCount = _eventsByTable.Count,
TotalEventCount = _eventsByTable.Values.Sum(list => list.Count),
PrimaryLengthBytes = 4096,
WalLengthBytes = 0,
PrimaryFormatVersion = 1,
WalFormatVersion = 1,
});
///
public override Task RegisterSeries(TimeRegisterSeriesRequest request, ServerCallContext context)
{
ulong id = _nextSeriesId++;
_samplesBySeries[id] = [];
_seriesMeta[id] = (request.MetricName, [.. request.Labels]);
return Task.FromResult(new TimeRegisterSeriesResponse { Status = NexusStatuses.Ok, SeriesId = id });
}
///
public override Task ListSeries(TimeListSeriesRequest request, ServerCallContext context)
{
TimeListSeriesResponse response = new() { Status = NexusStatuses.Ok };
int limit = request.Limit is 0 ? int.MaxValue : request.Limit;
foreach ((ulong id, (string metric, IReadOnlyList labels)) in _seriesMeta.Take(limit))
{
TimeSeriesCatalogEntryWire entry = new()
{
SeriesId = id,
MetricName = metric,
SampleCount = _samplesBySeries.TryGetValue(id, out List? samples) ? samples.Count : 0,
};
entry.Labels.AddRange(labels);
response.Series.Add(entry);
}
return Task.FromResult(response);
}
///
public override Task GetCardinality(TimeGetCardinalityRequest request, ServerCallContext context)
{
Dictionary byMetric = new(StringComparer.Ordinal);
foreach ((string metric, _) in _seriesMeta.Values)
{
byMetric[metric] = byMetric.GetValueOrDefault(metric) + 1;
}
int topN = request.TopN is 0 ? 20 : request.TopN;
TimeGetCardinalityResponse response = new()
{
Status = NexusStatuses.Ok,
ActiveSeriesCount = _seriesMeta.Count,
TotalSampleCount = _samplesBySeries.Values.Sum(list => list.Count),
};
foreach ((string metric, int count) in byMetric.OrderByDescending(static p => p.Value).ThenBy(static p => p.Key, StringComparer.Ordinal).Take(topN))
{
response.TopMetricsBySeriesCount.Add(new TimeMetricCardinalityWire { MetricName = metric, SeriesCount = count });
}
return Task.FromResult(response);
}
///
public override Task ApplyRetention(TimeApplyRetentionRequest request, ServerCallContext context)
{
long removed = 0;
foreach (List samples in _samplesBySeries.Values)
{
int before = samples.Count;
samples.RemoveAll(s => s.TimestampUnixNanos < request.RetainAfterExclusiveUnixNanos);
removed += before - samples.Count;
}
foreach (List events in _eventsByTable.Values)
{
int before = events.Count;
events.RemoveAll(e => e.TimestampUnixNanos < request.RetainAfterExclusiveUnixNanos);
removed += before - events.Count;
}
_retentionRemoved = removed;
return Task.FromResult(new TimeApplyRetentionResponse { Status = NexusStatuses.Ok, RemovedCount = removed });
}
///
public override Task AppendSample(TimeAppendSampleRequest request, ServerCallContext context)
{
if (!_samplesBySeries.TryGetValue(request.SeriesId, out List? samples))
{
return Task.FromResult(new TimeAppendSampleResponse { Status = NexusStatuses.NotFound($"Series {request.SeriesId} not registered.") });
}
samples.Add(new TimeSampleWire { TimestampUnixNanos = request.TimestampUnixNanos, Value = request.Value });
return Task.FromResult(new TimeAppendSampleResponse { Status = NexusStatuses.Ok });
}
///
public override Task AppendSampleBatch(TimeAppendSampleBatchRequest request, ServerCallContext context)
{
if (!_samplesBySeries.TryGetValue(request.SeriesId, out List? samples))
{
return Task.FromResult(new TimeAppendSampleResponse { Status = NexusStatuses.NotFound($"Series {request.SeriesId} not registered.") });
}
samples.AddRange(request.Samples);
return Task.FromResult(new TimeAppendSampleResponse { Status = NexusStatuses.Ok });
}
///
public override Task QueryRange(TimeQueryRangeRequest request, ServerCallContext context)
{
TimeQueryRangeResponse response = new() { Status = NexusStatuses.Ok };
if (_samplesBySeries.TryGetValue(request.SeriesId, out List? samples))
{
IEnumerable matching = samples
.Where(s => s.TimestampUnixNanos >= request.FromInclusive && s.TimestampUnixNanos < request.ToExclusive)
.Take(request.Limit is 0 ? int.MaxValue : request.Limit);
response.Samples.AddRange(matching);
}
return Task.FromResult(response);
}
///
public override Task Last(TimeLastRequest request, ServerCallContext context)
{
if (_samplesBySeries.TryGetValue(request.SeriesId, out List? samples) && samples.Count > 0)
{
return Task.FromResult(new TimeLastResponse { Status = NexusStatuses.Ok, HasSample = true, Sample = samples[^1] });
}
return Task.FromResult(new TimeLastResponse { Status = NexusStatuses.Ok, HasSample = false });
}
///
public override Task RegisterEventTable(TimeRegisterEventTableRequest request, ServerCallContext context)
{
ulong id = _nextTableId++;
_eventsByTable[id] = [];
_fieldsByTable[id] = [.. request.Fields];
return Task.FromResult(new TimeRegisterEventTableResponse { Status = NexusStatuses.Ok, TableId = id });
}
///
public override Task AppendEvent(TimeAppendEventRequest request, ServerCallContext context)
{
if (!_eventsByTable.TryGetValue(request.TableId, out List? events))
{
return Task.FromResult(new TimeAppendEventResponse { Status = NexusStatuses.NotFound($"Table {request.TableId} not registered.") });
}
ulong eventId = _nextEventId++;
TimeWideEventWire wire = new() { EventId = eventId, TimestampUnixNanos = request.TimestampUnixNanos };
wire.Cells.AddRange(request.Cells);
events.Add(wire);
return Task.FromResult(new TimeAppendEventResponse { Status = NexusStatuses.Ok, EventId = eventId });
}
///
public override Task QueryEvents(TimeQueryEventsRequest request, ServerCallContext context)
{
TimeQueryEventsResponse response = new() { Status = NexusStatuses.Ok };
if (_eventsByTable.TryGetValue(request.TableId, out List? events))
{
IEnumerable matching = events
.Where(e => e.TimestampUnixNanos >= request.FromInclusive && e.TimestampUnixNanos < request.ToExclusive)
.Take(request.Limit is 0 ? int.MaxValue : request.Limit);
response.Events.AddRange(matching);
}
return Task.FromResult(response);
}
///
public override Task LookupInverted(TimeLookupInvertedRequest request, ServerCallContext context)
{
TimeQueryEventsResponse response = new() { Status = NexusStatuses.Ok };
if (_eventsByTable.TryGetValue(request.TableId, out List? events))
{
IEnumerable matching = events
.Where(e => request.FieldOrdinal < e.Cells.Count && string.Equals(e.Cells[(int)request.FieldOrdinal].Utf8, request.ExactValue, StringComparison.Ordinal))
.Take(request.Limit is 0 ? int.MaxValue : request.Limit);
response.Events.AddRange(matching);
}
return Task.FromResult(response);
}
///
public override Task SearchFullText(TimeSearchFullTextRequest request, ServerCallContext context)
{
TimeSearchFullTextResponse response = new() { Status = NexusStatuses.Ok };
if (_eventsByTable.TryGetValue(request.TableId, out List? events))
{
IEnumerable matching = events
.Where(e => request.FieldOrdinal < e.Cells.Count
&& e.Cells[(int)request.FieldOrdinal].Utf8.Contains(request.Query, StringComparison.OrdinalIgnoreCase))
.Select(e => new TimeSearchHitWire { EventId = e.EventId, TimestampUnixNanos = e.TimestampUnixNanos, Score = 1.0 })
.Take(request.Limit is 0 ? int.MaxValue : request.Limit);
response.Hits.AddRange(matching);
}
return Task.FromResult(response);
}
}