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> _eventsByTable = []; private readonly Dictionary> _fieldsByTable = []; private ulong _nextSeriesId = 1; private ulong _nextTableId = 1; private ulong _nextEventId = 1; /// 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] = []; return Task.FromResult(new TimeRegisterSeriesResponse { Status = NexusStatuses.Ok, SeriesId = id }); } /// 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); } }