mishima-nexus-client-csharp/tests/Robotico.Nexus.Client.Tests/TestSupport/FakeTimeEngineService.cs

using Grpc.Core;
using Mishima.Nexus.Wire.V1;

namespace Robotico.Nexus.Client.Tests.TestSupport;

/// <summary>In-memory fake standing in for a real Nexus time engine host.</summary>
public sealed class FakeTimeEngineService : NexusTimeEngineService.NexusTimeEngineServiceBase
{
    private readonly Dictionary<ulong, List<TimeSampleWire>> _samplesBySeries = [];
    private readonly Dictionary<ulong, (string MetricName, IReadOnlyList<TimeLabelPair> Labels)> _seriesMeta = [];
    private readonly Dictionary<ulong, List<TimeWideEventWire>> _eventsByTable = [];
    private readonly Dictionary<ulong, List<TimeEventFieldWire>> _fieldsByTable = [];
    private ulong _nextSeriesId = 1;
    private ulong _nextTableId = 1;
    private ulong _nextEventId = 1;
    private long _retentionRemoved;

    /// <inheritdoc />
    public override Task<TimePingResponse> Ping(TimePingRequest request, ServerCallContext context)
    {
        NexusStatus? failure = FakeNexusTriggers.TryStatusFailure(context);
        return Task.FromResult(new TimePingResponse { Status = failure ?? NexusStatuses.Ok, EngineVersion = "fake" });
    }

    /// <inheritdoc />
    public override Task<TimeDoctorResponse> 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,
        });

    /// <inheritdoc />
    public override Task<TimeRegisterSeriesResponse> 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 });
    }

    /// <inheritdoc />
    public override Task<TimeListSeriesResponse> 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<TimeLabelPair> labels)) in _seriesMeta.Take(limit))
        {
            TimeSeriesCatalogEntryWire entry = new()
            {
                SeriesId = id,
                MetricName = metric,
                SampleCount = _samplesBySeries.TryGetValue(id, out List<TimeSampleWire>? samples) ? samples.Count : 0,
            };
            entry.Labels.AddRange(labels);
            response.Series.Add(entry);
        }

        return Task.FromResult(response);
    }

    /// <inheritdoc />
    public override Task<TimeGetCardinalityResponse> GetCardinality(TimeGetCardinalityRequest request, ServerCallContext context)
    {
        Dictionary<string, int> 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);
    }

    /// <inheritdoc />
    public override Task<TimeApplyRetentionResponse> ApplyRetention(TimeApplyRetentionRequest request, ServerCallContext context)
    {
        long removed = 0;
        foreach (List<TimeSampleWire> samples in _samplesBySeries.Values)
        {
            int before = samples.Count;
            samples.RemoveAll(s => s.TimestampUnixNanos < request.RetainAfterExclusiveUnixNanos);
            removed += before - samples.Count;
        }

        foreach (List<TimeWideEventWire> 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 });
    }

    /// <inheritdoc />
    public override Task<TimeAppendSampleResponse> AppendSample(TimeAppendSampleRequest request, ServerCallContext context)
    {
        if (!_samplesBySeries.TryGetValue(request.SeriesId, out List<TimeSampleWire>? 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 });
    }

    /// <inheritdoc />
    public override Task<TimeAppendSampleResponse> AppendSampleBatch(TimeAppendSampleBatchRequest request, ServerCallContext context)
    {
        if (!_samplesBySeries.TryGetValue(request.SeriesId, out List<TimeSampleWire>? 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 });
    }

    /// <inheritdoc />
    public override Task<TimeQueryRangeResponse> QueryRange(TimeQueryRangeRequest request, ServerCallContext context)
    {
        TimeQueryRangeResponse response = new() { Status = NexusStatuses.Ok };
        if (_samplesBySeries.TryGetValue(request.SeriesId, out List<TimeSampleWire>? samples))
        {
            IEnumerable<TimeSampleWire> 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);
    }

    /// <inheritdoc />
    public override Task<TimeLastResponse> Last(TimeLastRequest request, ServerCallContext context)
    {
        if (_samplesBySeries.TryGetValue(request.SeriesId, out List<TimeSampleWire>? 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 });
    }

    /// <inheritdoc />
    public override Task<TimeRegisterEventTableResponse> RegisterEventTable(TimeRegisterEventTableRequest request, ServerCallContext context)
    {
        ulong id = _nextTableId++;
        _eventsByTable[id] = [];
        _fieldsByTable[id] = [.. request.Fields];
        return Task.FromResult(new TimeRegisterEventTableResponse { Status = NexusStatuses.Ok, TableId = id });
    }

    /// <inheritdoc />
    public override Task<TimeAppendEventResponse> AppendEvent(TimeAppendEventRequest request, ServerCallContext context)
    {
        if (!_eventsByTable.TryGetValue(request.TableId, out List<TimeWideEventWire>? 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 });
    }

    /// <inheritdoc />
    public override Task<TimeQueryEventsResponse> QueryEvents(TimeQueryEventsRequest request, ServerCallContext context)
    {
        TimeQueryEventsResponse response = new() { Status = NexusStatuses.Ok };
        if (_eventsByTable.TryGetValue(request.TableId, out List<TimeWideEventWire>? events))
        {
            IEnumerable<TimeWideEventWire> 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);
    }

    /// <inheritdoc />
    public override Task<TimeQueryEventsResponse> LookupInverted(TimeLookupInvertedRequest request, ServerCallContext context)
    {
        TimeQueryEventsResponse response = new() { Status = NexusStatuses.Ok };
        if (_eventsByTable.TryGetValue(request.TableId, out List<TimeWideEventWire>? events))
        {
            IEnumerable<TimeWideEventWire> 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);
    }

    /// <inheritdoc />
    public override Task<TimeSearchFullTextResponse> SearchFullText(TimeSearchFullTextRequest request, ServerCallContext context)
    {
        TimeSearchFullTextResponse response = new() { Status = NexusStatuses.Ok };
        if (_eventsByTable.TryGetValue(request.TableId, out List<TimeWideEventWire>? events))
        {
            IEnumerable<TimeSearchHitWire> 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);
    }
}