diff --git a/src/Junction.App/Controls/Sparkline.cs b/src/Junction.App/Controls/Sparkline.cs new file mode 100644 index 0000000..c2edf2f --- /dev/null +++ b/src/Junction.App/Controls/Sparkline.cs @@ -0,0 +1,121 @@ +using System; +using System.Collections.Generic; +using Avalonia; +using Avalonia.Controls; +using Avalonia.Media; + +namespace Junction.App.Controls +{ + /// + /// Lightweight per-item trend line: draws a numeric time-series (oldest → newest, left → right) + /// as a single polyline scaled to fill the control. No axes, no labels — a table-cell sparkline. + /// Rendering-only; the parsing/collection of trend values lives in the view-model layer, so this + /// control stays a pure with no data-shaping logic to unit-test. + /// net48-safe: uses only core Avalonia drawing APIs (StreamGeometry + Pen), no external + /// charting dependency. Blank when there is nothing meaningful to draw (< 2 points). + /// + public sealed class Sparkline : Control + { + /// Trend samples in ascending time order. < 2 points renders blank. + public static readonly StyledProperty?> PointsProperty = + AvaloniaProperty.Register?>(nameof(Points)); + + /// Line color. Defaults to Junction's accent blue; the view binds the themed brush. + public static readonly StyledProperty StrokeProperty = + AvaloniaProperty.Register( + nameof(Stroke), + new SolidColorBrush(Color.FromRgb(0x25, 0x63, 0xEB))); + + static Sparkline() + { + AffectsRender(PointsProperty, StrokeProperty); + AffectsMeasure(PointsProperty); + } + + public IReadOnlyList? Points + { + get => GetValue(PointsProperty); + set => SetValue(PointsProperty, value); + } + + public IBrush? Stroke + { + get => GetValue(StrokeProperty); + set => SetValue(StrokeProperty, value); + } + + protected override Size MeasureOverride(Size availableSize) + { + // Own no intrinsic width (stretch into the column); pick a sensible default height + // when the parent imposes none. + double height = double.IsInfinity(availableSize.Height) ? 24 : availableSize.Height; + return new Size(0, height); + } + + public override void Render(DrawingContext context) + { + base.Render(context); + + var points = Points; + if (points == null || points.Count < 2) + { + return; // nothing meaningful to draw + } + + double width = Bounds.Width; + double height = Bounds.Height; + if (width <= 0 || height <= 0) + { + return; + } + + double min = points[0]; + double max = points[0]; + for (int i = 1; i < points.Count; i++) + { + double v = points[i]; + if (v < min) min = v; + if (v > max) max = v; + } + double range = max - min; + + // Small vertical inset so the stroke never clips at the top/bottom edge. + const double pad = 1.5; + double usableHeight = Math.Max(0, height - (2 * pad)); + double stepX = width / (points.Count - 1); + + var pen = new Pen(Stroke ?? Brushes.Gray, 1.5) + { + LineCap = PenLineCap.Round, + LineJoin = PenLineJoin.Round + }; + + var geometry = new StreamGeometry(); + using (var ctx = geometry.Open()) + { + for (int i = 0; i < points.Count; i++) + { + double x = i * stepX; + // All-equal series → flat mid line; else normalize into [0,1] and invert + // so a higher value sits higher on screen. + double norm = range > 0 ? (points[i] - min) / range : 0.5; + double y = pad + ((1 - norm) * usableHeight); + + var p = new Point(x, y); + if (i == 0) + { + ctx.BeginFigure(p, false); + } + else + { + ctx.LineTo(p); + } + } + + ctx.EndFigure(false); + } + + context.DrawGeometry(null, pen, geometry); + } + } +} diff --git a/src/Junction.App/ViewModels/DataItemRowViewModel.cs b/src/Junction.App/ViewModels/DataItemRowViewModel.cs index 34b6bc2..b2c5119 100644 --- a/src/Junction.App/ViewModels/DataItemRowViewModel.cs +++ b/src/Junction.App/ViewModels/DataItemRowViewModel.cs @@ -1,20 +1,33 @@ +using System.Collections.Generic; +using System.Globalization; +using CommunityToolkit.Mvvm.ComponentModel; using Junction.Domain.Models; namespace Junction.App.ViewModels { /// - /// One row of the machine-detail data-item table. Immutable projection of a - /// ; rebuilt (not mutated) whenever a fresh snapshot arrives, - /// so it needs no change-notification. + /// One row of the machine-detail data-item table. The string fields are a fixed projection of a + /// (set in the ctor; rebuilt, not mutated, per snapshot). Observable only + /// for , which is populated asynchronously after the row is created (history + /// query) and drives the per-item sparkline. /// - public sealed class DataItemRowViewModel + public sealed partial class DataItemRowViewModel : ViewModelBase { + /// History item-id key (matches / snapshot item id). public string Id { get; } public string Name { get; } public string Value { get; } public string Category { get; } public string Timestamp { get; } + /// Numeric trend samples (ascending time order) for the sparkline; null when none. + [ObservableProperty] + [NotifyPropertyChangedFor(nameof(HasTrend))] + private IReadOnlyList? _trend; + + /// True when there are enough numeric samples to draw a trend line. + public bool HasTrend => Trend != null && Trend.Count >= 2; + public DataItemRowViewModel(DataItem item) { Id = item.Id; @@ -23,5 +36,24 @@ namespace Junction.App.ViewModels Category = string.IsNullOrWhiteSpace(item.Category) ? "—" : item.Category; Timestamp = item.Timestamp.LocalDateTime.ToString("HH:mm:ss"); } + + /// + /// Culture-invariant numeric parse of a datum value. Trims; returns false for null/blank or + /// non-numeric text (e.g. "ON", "AVAILABLE"). Pure and Avalonia-free — unit-testable. + /// + public static bool TryParseNumeric(string? value, out double result) + { + result = 0; + if (string.IsNullOrWhiteSpace(value)) + { + return false; + } + + return double.TryParse( + value!.Trim(), + NumberStyles.Float | NumberStyles.AllowThousands, + CultureInfo.InvariantCulture, + out result); + } } } diff --git a/src/Junction.App/ViewModels/MachineDetailViewModel.cs b/src/Junction.App/ViewModels/MachineDetailViewModel.cs index 3ef1c2f..cf1236a 100644 --- a/src/Junction.App/ViewModels/MachineDetailViewModel.cs +++ b/src/Junction.App/ViewModels/MachineDetailViewModel.cs @@ -1,5 +1,7 @@ using System; +using System.Collections.Generic; using System.Collections.ObjectModel; +using System.Linq; using System.Threading; using System.Threading.Tasks; using Avalonia.Threading; @@ -23,6 +25,12 @@ namespace Junction.App.ViewModels /// public sealed partial class MachineDetailViewModel : ViewModelBase, IDisposable { + /// How far back the per-item trend sparkline looks. + private static readonly TimeSpan TrendWindow = TimeSpan.FromHours(1); + + /// Cap on samples pulled per item for a sparkline (most-recent kept). + private const int TrendMaxPoints = 60; + private readonly IMachineRepository _repository; private readonly IMachineMonitor _monitor; private readonly ILogger _logger; @@ -30,6 +38,12 @@ namespace Junction.App.ViewModels private bool _subscribed; private bool _disposed; + /// + /// Handle on the most recently kicked (fire-and-forget) trend load. Production ignores it; + /// tests await it to observe the populated values. + /// + public Task? TrendLoadTask { get; private set; } + /// Shell reference used by to return to the dashboard. public MainWindowViewModel? Navigator { get; set; } @@ -163,6 +177,7 @@ namespace Junction.App.ViewModels if (snapshot != null) { ApplySnapshot(snapshot); + KickTrendLoad(); } _logger.LogInformation("Machine detail opened for {Name} ({MachineId}); {Count} item(s).", @@ -210,9 +225,93 @@ namespace Junction.App.ViewModels } ApplySnapshot(snapshot); + KickTrendLoad(); }); } + /// + /// Fires (fire-and-forget) a per-item history query for every current row and, for items + /// whose samples parse as numeric, populates the row's + /// so its sparkline renders. Rebuilt rows carry no stale trend. A per-item failure never + /// aborts the others and never propagates. + /// + private void KickTrendLoad() + { + TrendLoadTask = LoadTrendsAsync(); + } + + private async Task LoadTrendsAsync() + { + // Snapshot the rows: the collection may be rebuilt by a later snapshot while we await. + var rows = Items.ToArray(); + var since = DateTimeOffset.UtcNow - TrendWindow; + + for (int r = 0; r < rows.Length; r++) + { + if (_disposed) + { + return; + } + + var row = rows[r]; + try + { + var result = await _repository + .GetHistoryAsync(_machineId, row.Id, since, TrendMaxPoints, CancellationToken.None) + .ConfigureAwait(true); + + if (_disposed) + { + return; + } + + if (!result.IsSuccess || result.Value == null) + { + continue; + } + + var history = result.Value; + var values = new List(history.Count); + for (int i = 0; i < history.Count; i++) + { + if (DataItemRowViewModel.TryParseNumeric(history[i].Value, out var d)) + { + values.Add(d); + } + } + + // Non-numeric / empty history → leave Trend null (sparkline stays blank). + if (values.Count > 0) + { + SetTrendOnUi(row, values.ToArray()); + } + } + catch (Exception ex) + { + _logger.LogDebug(ex, "Trend load failed for item {ItemId} on machine {MachineId}.", + row.Id, _machineId); + } + } + } + + private void SetTrendOnUi(DataItemRowViewModel row, IReadOnlyList trend) + { + if (Dispatcher.UIThread.CheckAccess()) + { + row.Trend = trend; + } + else + { + Dispatcher.UIThread.Post(() => + { + if (!_disposed) + { + row.Trend = trend; + } + }); + } + } + [RelayCommand] private void Back() { diff --git a/src/Junction.App/Views/MachineDetailView.axaml b/src/Junction.App/Views/MachineDetailView.axaml index c1b727d..8bf169b 100644 --- a/src/Junction.App/Views/MachineDetailView.axaml +++ b/src/Junction.App/Views/MachineDetailView.axaml @@ -3,6 +3,7 @@ xmlns:d="http://schemas.microsoft.com/expression/blend/2008" xmlns:mc="http://schemas.openxmlformats.org/markup-compatibility/2006" xmlns:vm="clr-namespace:Junction.App.ViewModels" + xmlns:controls="clr-namespace:Junction.App.Controls" mc:Ignorable="d" x:Class="Junction.App.Views.MachineDetailView" x:DataType="vm:MachineDetailViewModel" @@ -68,11 +69,12 @@ BorderBrush="{DynamicResource CardBorderBrush}" Padding="0,0,0,6" Margin="0,0,0,4" IsVisible="{Binding HasItems}"> - + + @@ -82,7 +84,7 @@ - + @@ -90,6 +92,9 @@ + diff --git a/src/Junction.Core/Monitoring/MachineMonitor.cs b/src/Junction.Core/Monitoring/MachineMonitor.cs index b8f765e..4c4f8bd 100644 --- a/src/Junction.Core/Monitoring/MachineMonitor.cs +++ b/src/Junction.Core/Monitoring/MachineMonitor.cs @@ -42,6 +42,15 @@ namespace Junction.Core.Monitoring private readonly ConcurrentDictionary _snapshots = new ConcurrentDictionary(); + // History retention policy (const-fixed for this slice; not config-driven yet). + // TimeSpan can't be a C# const, so these are static readonly. + private static readonly TimeSpan RetentionWindow = TimeSpan.FromDays(7); + private static readonly TimeSpan PruneInterval = TimeSpan.FromHours(1); + + // UTC ticks of the last prune claim. Interlocked-guarded because PersistAsync runs + // concurrently across per-machine poll loops. Init to MinValue so the first persist prunes. + private long _lastPruneTicks = DateTimeOffset.MinValue.UtcTicks; + private readonly object _lifecycleLock = new object(); private readonly List _running = new List(); @@ -462,6 +471,23 @@ namespace Junction.Core.Monitoring "Failed to persist snapshot for machine {MachineId}: {Errors}", snapshot.MachineId, DescribeErrors(result.Errors)); } + + // Append time-series history for live reads only. Synthetic Disconnected + // snapshots carry no items (AppendHistory no-ops) but gating here avoids the + // round-trip and keeps disconnect noise out of the trend. A history failure + // must never undo the latest-snapshot save above. + if (snapshot.ConnectionState == ConnectionState.Connected) + { + Result historyResult = await _repository.AppendHistoryAsync(snapshot, cancellationToken).ConfigureAwait(false); + if (!historyResult.IsSuccess && !historyResult.WasCancelled) + { + _logger.LogWarning( + "Failed to append history for machine {MachineId}: {Errors}", + snapshot.MachineId, DescribeErrors(historyResult.Errors)); + } + + await MaybePruneHistoryAsync(cancellationToken).ConfigureAwait(false); + } } catch (OperationCanceledException) { @@ -473,6 +499,38 @@ namespace Junction.Core.Monitoring } } + /// + /// Prunes history older than at most once per + /// , regardless of poll frequency or machine count. The + /// window is claimed via a single Interlocked CompareExchange so concurrent poll loops + /// don't all prune at once; a lost race simply skips (harmless). + /// + private async Task MaybePruneHistoryAsync(CancellationToken cancellationToken) + { + DateTimeOffset now = DateTimeOffset.UtcNow; + long lastPrune = Interlocked.Read(ref _lastPruneTicks); + + if (now - new DateTimeOffset(lastPrune, TimeSpan.Zero) < PruneInterval) + { + return; + } + + // Claim the interval; only the loop that wins the swap performs the prune. + if (Interlocked.CompareExchange(ref _lastPruneTicks, now.UtcTicks, lastPrune) != lastPrune) + { + return; + } + + DateTimeOffset cutoff = now - RetentionWindow; + Result pruneResult = await _repository.PruneHistoryAsync(cutoff, cancellationToken).ConfigureAwait(false); + if (!pruneResult.IsSuccess && !pruneResult.WasCancelled) + { + _logger.LogWarning( + "Failed to prune history older than {Cutoff}: {Errors}", + cutoff, DescribeErrors(pruneResult.Errors)); + } + } + /// /// Maps ProtocolId (case-insensitive) to factory. On duplicate protocol ids the first /// loaded plugin wins; the collision is logged. diff --git a/src/Junction.Domain/Models/HistoryPoint.cs b/src/Junction.Domain/Models/HistoryPoint.cs new file mode 100644 index 0000000..46e80fe --- /dev/null +++ b/src/Junction.Domain/Models/HistoryPoint.cs @@ -0,0 +1,31 @@ +using System; + +namespace Junction.Domain.Models +{ + /// + /// One time-series sample of one data item: the value a datum had at a point in time. + /// Immutable, protocol-agnostic. Unlike (only-latest), history points + /// accumulate append-only so a trend over time can be reconstructed. + /// + public sealed class HistoryPoint + { + /// Stable identifier of the datum this sample belongs to (see ). + public string ItemId { get; } + + /// + /// Sampled value kept as string to stay type-agnostic across protocols; + /// callers parse to the concrete type they need. + /// + public string Value { get; } + + /// Instant the value was reported/observed. + public DateTimeOffset Timestamp { get; } + + public HistoryPoint(string itemId, string value, DateTimeOffset timestamp) + { + ItemId = itemId ?? ""; + Value = value ?? ""; + Timestamp = timestamp; + } + } +} diff --git a/src/Junction.Domain/Persistence/IMachineRepository.cs b/src/Junction.Domain/Persistence/IMachineRepository.cs index 9f1d524..ceb76ea 100644 --- a/src/Junction.Domain/Persistence/IMachineRepository.cs +++ b/src/Junction.Domain/Persistence/IMachineRepository.cs @@ -42,5 +42,27 @@ namespace Junction.Domain.Persistence /// Returns the latest stored snapshot for the machine, or a failed result when none exists. Task> GetLatestSnapshotAsync(Guid machineId, CancellationToken cancellationToken); + + /// + /// Appends one time-series history row per item in (append-only: + /// never overwrites an earlier sample). A snapshot with no items is a no-op success. + /// This is separate from , which keeps only the latest set. + /// + Task AppendHistoryAsync(MachineSnapshot snapshot, CancellationToken cancellationToken); + + /// + /// Returns time-series samples for a single item of a machine at or after , + /// ascending by timestamp, capped at (keeping the MOST RECENT + /// when more exist). Empty list on none — a success, not a failure. + /// A non-positive yields an empty success. + /// + Task>> GetHistoryAsync(Guid machineId, string itemId, DateTimeOffset since, int maxPoints, CancellationToken cancellationToken); + + /// + /// Deletes all history rows whose timestamp is strictly older than . + /// Idempotent: safe to call when nothing matches. Provides the mechanism only; retention + /// policy (deciding the cutoff) lives above this port. + /// + Task PruneHistoryAsync(DateTimeOffset olderThan, CancellationToken cancellationToken); } } diff --git a/src/Junction.Persistence/SqliteMachineRepository.cs b/src/Junction.Persistence/SqliteMachineRepository.cs index f23d164..d64c279 100644 --- a/src/Junction.Persistence/SqliteMachineRepository.cs +++ b/src/Junction.Persistence/SqliteMachineRepository.cs @@ -189,6 +189,10 @@ namespace Junction.Persistence { var idParam = new { Id = GuidText(id) }; + await connection.ExecuteAsync(new CommandDefinition( + "DELETE FROM snapshot_history WHERE MachineId = @Id;", + idParam, transaction, cancellationToken: cancellationToken)).ConfigureAwait(false); + await connection.ExecuteAsync(new CommandDefinition( "DELETE FROM snapshot_items WHERE MachineId = @Id;", idParam, transaction, cancellationToken: cancellationToken)).ConfigureAwait(false); @@ -354,6 +358,150 @@ namespace Junction.Persistence } } + public async Task AppendHistoryAsync(MachineSnapshot snapshot, CancellationToken cancellationToken) + { + if (snapshot == null) + { + return Result.Fail(OperationError.Of(Source, "Snapshot is null.")); + } + + if (cancellationToken.IsCancellationRequested) + { + return Result.Cancelled(); + } + + // Append-only: nothing to record for an empty snapshot. + if (snapshot.Items.Count == 0) + { + return Result.Ok(); + } + + try + { + using (var connection = OpenConnection()) + using (var transaction = connection.BeginTransaction()) + { + var machineIdText = GuidText(snapshot.MachineId); + + var itemParams = new List(snapshot.Items.Count); + foreach (var item in snapshot.Items) + { + itemParams.Add(new + { + MachineId = machineIdText, + ItemId = item.Id, + item.Name, + item.Value, + item.Category, + Timestamp = IsoText(item.Timestamp) + }); + } + + await connection.ExecuteAsync(new CommandDefinition( + "INSERT INTO snapshot_history (MachineId, ItemId, Name, Value, Category, Timestamp) " + + "VALUES (@MachineId, @ItemId, @Name, @Value, @Category, @Timestamp);", + itemParams, + transaction, cancellationToken: cancellationToken)).ConfigureAwait(false); + + transaction.Commit(); + + return Result.Ok(); + } + } + catch (OperationCanceledException) + { + return Result.Cancelled(); + } + catch (Exception ex) + { + return Result.Fail(OperationError.Of(Source, "AppendHistory failed: " + ex.Message)); + } + } + + public async Task>> GetHistoryAsync( + Guid machineId, string itemId, DateTimeOffset since, int maxPoints, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return Result>.Cancelled(); + } + + // Non-positive cap => nothing requested; empty success (not a failure). + if (maxPoints <= 0) + { + return Result>.Ok(Array.Empty()); + } + + try + { + using (var connection = OpenConnection()) + { + // Take the most-recent N at/after `since` (DESC + LIMIT), then reverse to ascending. + var rows = await connection.QueryAsync(new CommandDefinition( + "SELECT ItemId, Value, Timestamp FROM snapshot_history " + + "WHERE MachineId = @MachineId AND ItemId = @ItemId AND Timestamp >= @Since " + + "ORDER BY Timestamp DESC LIMIT @MaxPoints;", + new + { + MachineId = GuidText(machineId), + ItemId = itemId ?? "", + Since = IsoText(since), + MaxPoints = maxPoints + }, + cancellationToken: cancellationToken)).ConfigureAwait(false); + + var points = new List(); + foreach (var row in rows) + { + points.Add(new HistoryPoint(row.ItemId, row.Value, ParseIso(row.Timestamp))); + } + + // Query is DESC (newest first); reverse to ascending by timestamp. + points.Reverse(); + + return Result>.Ok(points); + } + } + catch (OperationCanceledException) + { + return Result>.Cancelled(); + } + catch (Exception ex) + { + return Result>.Fail( + OperationError.Of(Source, "GetHistory failed: " + ex.Message)); + } + } + + public async Task PruneHistoryAsync(DateTimeOffset olderThan, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return Result.Cancelled(); + } + + try + { + using (var connection = OpenConnection()) + { + await connection.ExecuteAsync(new CommandDefinition( + "DELETE FROM snapshot_history WHERE Timestamp < @OlderThan;", + new { OlderThan = IsoText(olderThan) }, + cancellationToken: cancellationToken)).ConfigureAwait(false); + + return Result.Ok(); + } + } + catch (OperationCanceledException) + { + return Result.Cancelled(); + } + catch (Exception ex) + { + return Result.Fail(OperationError.Of(Source, "PruneHistory failed: " + ex.Message)); + } + } + // -- helpers ------------------------------------------------------------------------- private DbConnection OpenConnection() @@ -471,5 +619,12 @@ namespace Junction.Persistence public string Category { get; set; } = ""; public string Timestamp { get; set; } = ""; } + + private sealed class HistoryRow + { + public string ItemId { get; set; } = ""; + public string Value { get; set; } = ""; + public string Timestamp { get; set; } = ""; + } } } diff --git a/src/Junction.Persistence/SqliteSchema.cs b/src/Junction.Persistence/SqliteSchema.cs index 71d1627..117e877 100644 --- a/src/Junction.Persistence/SqliteSchema.cs +++ b/src/Junction.Persistence/SqliteSchema.cs @@ -6,8 +6,8 @@ namespace Junction.Persistence { /// /// Owns the SQLite schema (DDL). Idempotent: safe to run on every startup. - /// Only-latest policy — one snapshot row and one item-set per machine — but the - /// shape leaves room to add history tables later without breaking these tables. + /// latest_snapshots/snapshot_items keep the only-latest set per machine; snapshot_history + /// is the append-only time series added alongside them without breaking those tables. /// public static class SqliteSchema { @@ -42,6 +42,23 @@ namespace Junction.Persistence PRIMARY KEY (MachineId, ItemId) );"; + // snapshot_history: append-only time series. Many rows per (MachineId, ItemId) — one per + // appended sample — so NO single-row primary key. Never overwritten; pruned by timestamp. + private const string CreateSnapshotHistory = + @"CREATE TABLE IF NOT EXISTS snapshot_history ( + MachineId TEXT NOT NULL, + ItemId TEXT NOT NULL, + Name TEXT NOT NULL, + Value TEXT NOT NULL, + Category TEXT NOT NULL, + Timestamp TEXT NOT NULL + );"; + + // Index for the query path (GetHistory: WHERE MachineId=@ AND ItemId=@ AND Timestamp>=@ ORDER BY Timestamp). + private const string CreateSnapshotHistoryIndex = + @"CREATE INDEX IF NOT EXISTS ix_snapshot_history_machine_item_ts + ON snapshot_history (MachineId, ItemId, Timestamp);"; + /// /// Creates all tables if they do not already exist. Idempotent. /// Returns a failed instead of throwing on store errors. @@ -63,6 +80,8 @@ namespace Junction.Persistence Execute(connection, CreateMachines); Execute(connection, CreateLatestSnapshots); Execute(connection, CreateSnapshotItems); + Execute(connection, CreateSnapshotHistory); + Execute(connection, CreateSnapshotHistoryIndex); // Migration for DBs created by an older schema (before MonitoredItemIdsJson existed): // CREATE TABLE IF NOT EXISTS never alters an existing table, so add the column here if diff --git a/tests/Junction.Tests/Unit/DataItemRowViewModelTests.cs b/tests/Junction.Tests/Unit/DataItemRowViewModelTests.cs new file mode 100644 index 0000000..7d04039 --- /dev/null +++ b/tests/Junction.Tests/Unit/DataItemRowViewModelTests.cs @@ -0,0 +1,39 @@ +using Junction.App.ViewModels; +using Xunit; + +namespace Junction.Tests.Unit +{ + /// + /// Data-level tests for the culture-invariant numeric parse that decides whether a datum can + /// feed the trend sparkline. Pure static helper — no Avalonia platform required. + /// + public class DataItemRowViewModelTests + { + [Theory] + [InlineData("12.5", 12.5)] + [InlineData(" 3 ", 3)] + [InlineData("1000", 1000)] + [InlineData("-4.25", -4.25)] + public void TryParseNumeric_Numeric_ReturnsTrueWithInvariantValue(string input, double expected) + { + var ok = DataItemRowViewModel.TryParseNumeric(input, out var result); + + Assert.True(ok); + Assert.Equal(expected, result); + } + + [Theory] + [InlineData("")] + [InlineData(" ")] + [InlineData("abc")] + [InlineData("ON")] + [InlineData("AVAILABLE")] + [InlineData(null)] + public void TryParseNumeric_NonNumeric_ReturnsFalse(string? input) + { + var ok = DataItemRowViewModel.TryParseNumeric(input, out _); + + Assert.False(ok); + } + } +} diff --git a/tests/Junction.Tests/Unit/MachineDetailViewModelTests.cs b/tests/Junction.Tests/Unit/MachineDetailViewModelTests.cs new file mode 100644 index 0000000..eb47610 --- /dev/null +++ b/tests/Junction.Tests/Unit/MachineDetailViewModelTests.cs @@ -0,0 +1,165 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Junction.App.ViewModels; +using Junction.Core.Monitoring; +using Junction.Domain; +using Junction.Domain.Models; +using Junction.Domain.Persistence; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using Xunit; + +namespace Junction.Tests.Unit +{ + /// + /// Trend-load behaviour of the detail VM: after a load, each row whose history parses as numeric + /// gets a populated (ascending order) with + /// HasTrend true; a row with non-numeric history stays null / HasTrend false. + /// Fixture: Moq'd + + null logger. + /// The fire-and-forget trend load is awaited via the VM's internal TrendLoadTask. + /// + public class MachineDetailViewModelTests + { + private const string NumericItemId = "temp"; + private const string TextItemId = "avail"; + + private static readonly Guid MachineId = Guid.NewGuid(); + + private static MachineSnapshot SnapshotWithBothItems() + { + var now = DateTimeOffset.UtcNow; + var items = new List + { + new DataItem(NumericItemId, "Temperature", "12", "Sample", now), + new DataItem(TextItemId, "Availability", "AVAILABLE", "Event", now), + }; + return new MachineSnapshot(MachineId, now, ConnectionState.Connected, items); + } + + private static IReadOnlyList NumericHistory() + { + var t = DateTimeOffset.UtcNow; + return new List + { + new HistoryPoint(NumericItemId, "10", t.AddSeconds(-30)), + new HistoryPoint(NumericItemId, "12.5", t.AddSeconds(-20)), + new HistoryPoint(NumericItemId, "11", t.AddSeconds(-10)), + }; + } + + private static IReadOnlyList TextHistory() + { + var t = DateTimeOffset.UtcNow; + return new List + { + new HistoryPoint(TextItemId, "AVAILABLE", t.AddSeconds(-20)), + new HistoryPoint(TextItemId, "UNAVAILABLE", t.AddSeconds(-10)), + }; + } + + private static MachineDetailViewModel BuildVm( + Mock repo, + Mock monitor) + { + var vm = new MachineDetailViewModel( + repo.Object, monitor.Object, NullLogger.Instance); + vm.Initialize(MachineId, "VMC Sim (mock)"); + return vm; + } + + private static Mock MonitorWithSnapshot(MachineSnapshot snapshot) + { + var monitor = new Mock(); + var cache = new Dictionary { [MachineId] = snapshot }; + monitor.SetupGet(m => m.LatestSnapshots) + .Returns((IReadOnlyDictionary)cache); + return monitor; + } + + private static DataItemRowViewModel Row(MachineDetailViewModel vm, string id) => + vm.Items.First(r => r.Id == id); + + [Fact] + public async Task LoadAsync_NumericHistory_PopulatesTrendAscending() + { + var snapshot = SnapshotWithBothItems(); + var repo = new Mock(); + repo.Setup(r => r.GetByIdAsync(MachineId, It.IsAny())) + .ReturnsAsync(Result.Ok(new Machine(MachineId, "VMC", "mtconnect", null, TimeSpan.FromSeconds(2)))); + repo.Setup(r => r.GetHistoryAsync(MachineId, NumericItemId, It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(Result>.Ok(NumericHistory())); + repo.Setup(r => r.GetHistoryAsync(MachineId, TextItemId, It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(Result>.Ok(TextHistory())); + + var monitor = MonitorWithSnapshot(snapshot); + var vm = BuildVm(repo, monitor); + + await vm.LoadAsync(); + if (vm.TrendLoadTask != null) + { + await vm.TrendLoadTask; + } + + var numericRow = Row(vm, NumericItemId); + Assert.True(numericRow.HasTrend); + Assert.NotNull(numericRow.Trend); + Assert.Equal(new double[] { 10, 12.5, 11 }, numericRow.Trend!.ToArray()); + } + + [Fact] + public async Task LoadAsync_NonNumericHistory_LeavesTrendNull() + { + var snapshot = SnapshotWithBothItems(); + var repo = new Mock(); + repo.Setup(r => r.GetByIdAsync(MachineId, It.IsAny())) + .ReturnsAsync(Result.Ok(new Machine(MachineId, "VMC", "mtconnect", null, TimeSpan.FromSeconds(2)))); + repo.Setup(r => r.GetHistoryAsync(MachineId, NumericItemId, It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(Result>.Ok(NumericHistory())); + repo.Setup(r => r.GetHistoryAsync(MachineId, TextItemId, It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(Result>.Ok(TextHistory())); + + var monitor = MonitorWithSnapshot(snapshot); + var vm = BuildVm(repo, monitor); + + await vm.LoadAsync(); + if (vm.TrendLoadTask != null) + { + await vm.TrendLoadTask; + } + + var textRow = Row(vm, TextItemId); + Assert.False(textRow.HasTrend); + Assert.Null(textRow.Trend); + } + + [Fact] + public async Task LoadAsync_HistoryFailureForOneItem_DoesNotThrowOrBlockOthers() + { + var snapshot = SnapshotWithBothItems(); + var repo = new Mock(); + repo.Setup(r => r.GetByIdAsync(MachineId, It.IsAny())) + .ReturnsAsync(Result.Ok(new Machine(MachineId, "VMC", "mtconnect", null, TimeSpan.FromSeconds(2)))); + // Numeric item succeeds; text item fails hard. + repo.Setup(r => r.GetHistoryAsync(MachineId, NumericItemId, It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(Result>.Ok(NumericHistory())); + repo.Setup(r => r.GetHistoryAsync(MachineId, TextItemId, It.IsAny(), It.IsAny(), It.IsAny())) + .ThrowsAsync(new InvalidOperationException("boom")); + + var monitor = MonitorWithSnapshot(snapshot); + var vm = BuildVm(repo, monitor); + + await vm.LoadAsync(); + if (vm.TrendLoadTask != null) + { + await vm.TrendLoadTask; + } + + // The failing item leaves the successful item's trend intact. + Assert.True(Row(vm, NumericItemId).HasTrend); + Assert.False(Row(vm, TextItemId).HasTrend); + } + } +} diff --git a/tests/Junction.Tests/Unit/MachineMonitorTests.cs b/tests/Junction.Tests/Unit/MachineMonitorTests.cs index 47f253c..4696a0b 100644 --- a/tests/Junction.Tests/Unit/MachineMonitorTests.cs +++ b/tests/Junction.Tests/Unit/MachineMonitorTests.cs @@ -60,9 +60,19 @@ namespace Junction.Tests.Unit .ReturnsAsync(Result>.Ok(machines)); mock.Setup(r => r.SaveSnapshotAsync(It.IsAny(), It.IsAny())) .ReturnsAsync(Result.Ok()); + mock.Setup(r => r.AppendHistoryAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(Result.Ok()); + mock.Setup(r => r.PruneHistoryAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(Result.Ok()); return mock; } + private static MachineSnapshot SnapshotWithItems(Guid machineId) => + new MachineSnapshot(machineId, DateTimeOffset.UtcNow, ConnectionState.Connected, new[] + { + new DataItem("x1_pos", "X", "12.5", "SAMPLE", DateTimeOffset.UtcNow), + }); + [Fact] public async Task StartAsync_PollsMachine_RaisesEvent_UpdatesCache_Persists() { @@ -95,6 +105,114 @@ namespace Junction.Tests.Unit It.IsAny()), Times.AtLeastOnce); } + [Fact] + public async Task ConnectedSnapshotWithItems_AppendsHistory_AndStillSavesLatest() + { + var machine = MachineWith("test"); + var factory = new FakeFactory("test", m => Result.Ok( + new FakeDriver("test", () => Result.Ok(SnapshotWithItems(machine.Id))))); + + var loader = LoaderReturning(factory); + var repo = RepoReturning(machine); + + var appended = new ManualResetEventSlim(false); + repo.Setup(r => r.AppendHistoryAsync( + It.Is(s => s.MachineId == machine.Id && s.Items.Count > 0), + It.IsAny())) + .ReturnsAsync(Result.Ok()) + .Callback(() => appended.Set()); + + var monitor = NewMonitor(repo.Object, loader.Object); + + await monitor.StartAsync(PluginsDir, CancellationToken.None); + Assert.True(appended.Wait(TimeSpan.FromSeconds(2)), "AppendHistoryAsync was not invoked"); + await monitor.StopAsync(); + + // History appended AND latest-snapshot save still happens (existing contract intact). + repo.Verify(r => r.AppendHistoryAsync( + It.Is(s => s.MachineId == machine.Id), + It.IsAny()), Times.AtLeastOnce); + repo.Verify(r => r.SaveSnapshotAsync( + It.Is(s => s.MachineId == machine.Id), + It.IsAny()), Times.AtLeastOnce); + } + + [Fact] + public async Task HistoryAppendFailure_DoesNotThrow_NorStopLatestSave() + { + var machine = MachineWith("test"); + var factory = new FakeFactory("test", m => Result.Ok( + new FakeDriver("test", () => Result.Ok(SnapshotWithItems(machine.Id))))); + + var loader = LoaderReturning(factory); + var repo = RepoReturning(machine); + + // Append both fails AND throws on alternating calls to exercise both paths. + repo.Setup(r => r.AppendHistoryAsync(It.IsAny(), It.IsAny())) + .ThrowsAsync(new InvalidOperationException("history store boom")); + + var monitor = NewMonitor(repo.Object, loader.Object); + + var eventRaised = new ManualResetEventSlim(false); + monitor.SnapshotUpdated += (_, snap) => + { + if (snap.MachineId == machine.Id) eventRaised.Set(); + }; + + var start = await monitor.StartAsync(PluginsDir, CancellationToken.None); + Assert.True(start.IsSuccess); + + // Monitor keeps running: events keep flowing and latest-save keeps being called. + Assert.True(eventRaised.Wait(TimeSpan.FromSeconds(2)), "monitor stalled after history failure"); + await Task.Delay(120); + await monitor.StopAsync(); + + repo.Verify(r => r.SaveSnapshotAsync( + It.Is(s => s.MachineId == machine.Id), + It.IsAny()), Times.AtLeastOnce); + Assert.True(monitor.LatestSnapshots.ContainsKey(machine.Id)); + } + + [Fact] + public async Task PruneHistory_Throttled_AtMostOncePerInterval_AcrossManySnapshots() + { + // PruneInterval is 1h; many rapid snapshots must trigger at most one prune. + var machine = MachineWith("test"); + var factory = new FakeFactory("test", m => Result.Ok( + new FakeDriver("test", () => Result.Ok(SnapshotWithItems(machine.Id))))); + + var loader = LoaderReturning(factory); + var repo = RepoReturning(machine); + + int appendCount = 0; + int pruneCount = 0; + repo.Setup(r => r.AppendHistoryAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(Result.Ok()) + .Callback(() => Interlocked.Increment(ref appendCount)); + repo.Setup(r => r.PruneHistoryAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(Result.Ok()) + .Callback(() => Interlocked.Increment(ref pruneCount)); + + var monitor = NewMonitor(repo.Object, loader.Object); + + await monitor.StartAsync(PluginsDir, CancellationToken.None); + + // Let many poll cycles run (FastPoll = 25ms => dozens of appends). + var sw = System.Diagnostics.Stopwatch.StartNew(); + while (Volatile.Read(ref appendCount) < 5 && sw.Elapsed < TimeSpan.FromSeconds(2)) + { + await Task.Delay(20); + } + await Task.Delay(150); + await monitor.StopAsync(); + + Assert.True(Volatile.Read(ref appendCount) >= 5, "expected many history appends across poll cycles"); + // Throttle proof: append fired many times, prune fired at most once in the window. + repo.Verify(r => r.PruneHistoryAsync(It.IsAny(), It.IsAny()), Times.AtMostOnce); + // And it did fire on the first persist (MinValue-init guarantees the first call prunes). + Assert.Equal(1, Volatile.Read(ref pruneCount)); + } + [Fact] public async Task StartAsync_UnknownProtocol_SkipsMachine_OthersRun_StillOk() { diff --git a/tests/Junction.Tests/Unit/SqliteMachineRepositoryTests.cs b/tests/Junction.Tests/Unit/SqliteMachineRepositoryTests.cs index 5898f0c..e89bf0b 100644 --- a/tests/Junction.Tests/Unit/SqliteMachineRepositoryTests.cs +++ b/tests/Junction.Tests/Unit/SqliteMachineRepositoryTests.cs @@ -350,6 +350,187 @@ namespace Junction.Tests.Unit Assert.False(result.IsSuccess); } + // -- history (append-only time series) ---------------------------------------------- + + [Fact] + public async Task AppendHistory_ThenGetHistory_ReturnsAppendedPoints_Ascending() + { + var machineId = Guid.NewGuid(); + var t0 = new DateTimeOffset(2026, 7, 22, 8, 0, 0, TimeSpan.Zero); + + // Two snapshots of the same item at different times, appended in reverse order. + await _repo.AppendHistoryAsync( + Snap(machineId, t0.AddSeconds(10), new DataItem("d1", "Speed", "20", "Sample", t0.AddSeconds(10))), + CancellationToken.None); + Result append = await _repo.AppendHistoryAsync( + Snap(machineId, t0, new DataItem("d1", "Speed", "10", "Sample", t0)), + CancellationToken.None); + Assert.True(append.IsSuccess, Describe(append)); + + Result> got = + await _repo.GetHistoryAsync(machineId, "d1", t0.AddSeconds(-1), 100, CancellationToken.None); + Assert.True(got.IsSuccess, Describe(got)); + + // Append is additive (accumulates, not overwrite) and result is ascending by timestamp. + Assert.Equal(2, got.Value.Count); + Assert.Equal("10", got.Value[0].Value); + Assert.Equal(t0, got.Value[0].Timestamp); + Assert.Equal("20", got.Value[1].Value); + Assert.Equal(t0.AddSeconds(10), got.Value[1].Timestamp); + Assert.Equal("d1", got.Value[0].ItemId); + } + + [Fact] + public async Task GetHistory_SinceFilter_ExcludesOlderRows() + { + var machineId = Guid.NewGuid(); + var t0 = new DateTimeOffset(2026, 7, 22, 8, 0, 0, TimeSpan.Zero); + + for (int i = 0; i < 5; i++) + { + await _repo.AppendHistoryAsync( + Snap(machineId, t0.AddMinutes(i), new DataItem("d1", "Speed", i.ToString(), "Sample", t0.AddMinutes(i))), + CancellationToken.None); + } + + // since = t0 + 3min => only minute 3 and 4 remain. + Result> got = + await _repo.GetHistoryAsync(machineId, "d1", t0.AddMinutes(3), 100, CancellationToken.None); + Assert.True(got.IsSuccess, Describe(got)); + + Assert.Equal(2, got.Value.Count); + Assert.Equal("3", got.Value[0].Value); + Assert.Equal("4", got.Value[1].Value); + } + + [Fact] + public async Task GetHistory_MaxPoints_KeepsMostRecentN_Ascending() + { + var machineId = Guid.NewGuid(); + var t0 = new DateTimeOffset(2026, 7, 22, 8, 0, 0, TimeSpan.Zero); + + for (int i = 0; i < 5; i++) + { + await _repo.AppendHistoryAsync( + Snap(machineId, t0.AddMinutes(i), new DataItem("d1", "Speed", i.ToString(), "Sample", t0.AddMinutes(i))), + CancellationToken.None); + } + + // 5 rows exist, ask for at most 2 => most-recent two (3,4), returned ascending. + Result> got = + await _repo.GetHistoryAsync(machineId, "d1", t0.AddMinutes(-1), 2, CancellationToken.None); + Assert.True(got.IsSuccess, Describe(got)); + + Assert.Equal(2, got.Value.Count); + Assert.Equal("3", got.Value[0].Value); + Assert.Equal("4", got.Value[1].Value); + } + + [Fact] + public async Task GetHistory_UnknownItemOrMachine_ReturnsEmptyOk() + { + var machineId = Guid.NewGuid(); + var t0 = new DateTimeOffset(2026, 7, 22, 8, 0, 0, TimeSpan.Zero); + await _repo.AppendHistoryAsync( + Snap(machineId, t0, new DataItem("d1", "Speed", "10", "Sample", t0)), + CancellationToken.None); + + Result> unknownItem = + await _repo.GetHistoryAsync(machineId, "nope", t0.AddSeconds(-1), 100, CancellationToken.None); + Assert.True(unknownItem.IsSuccess, Describe(unknownItem)); + Assert.Empty(unknownItem.Value); + + Result> unknownMachine = + await _repo.GetHistoryAsync(Guid.NewGuid(), "d1", t0.AddSeconds(-1), 100, CancellationToken.None); + Assert.True(unknownMachine.IsSuccess, Describe(unknownMachine)); + Assert.Empty(unknownMachine.Value); + } + + [Fact] + public async Task GetHistory_NonPositiveMaxPoints_ReturnsEmptyOk() + { + var machineId = Guid.NewGuid(); + var t0 = new DateTimeOffset(2026, 7, 22, 8, 0, 0, TimeSpan.Zero); + await _repo.AppendHistoryAsync( + Snap(machineId, t0, new DataItem("d1", "Speed", "10", "Sample", t0)), + CancellationToken.None); + + Result> got = + await _repo.GetHistoryAsync(machineId, "d1", t0.AddSeconds(-1), 0, CancellationToken.None); + Assert.True(got.IsSuccess, Describe(got)); + Assert.Empty(got.Value); + } + + [Fact] + public async Task AppendHistory_EmptySnapshot_IsNoOpSuccess() + { + var machineId = Guid.NewGuid(); + var t0 = new DateTimeOffset(2026, 7, 22, 8, 0, 0, TimeSpan.Zero); + + Result append = await _repo.AppendHistoryAsync( + new MachineSnapshot(machineId, t0, ConnectionState.Connected, null), + CancellationToken.None); + Assert.True(append.IsSuccess, Describe(append)); + + Result> got = + await _repo.GetHistoryAsync(machineId, "d1", t0.AddSeconds(-1), 100, CancellationToken.None); + Assert.True(got.IsSuccess, Describe(got)); + Assert.Empty(got.Value); + } + + [Fact] + public async Task PruneHistory_RemovesOlderThanCutoff_KeepsNewer() + { + var machineId = Guid.NewGuid(); + var t0 = new DateTimeOffset(2026, 7, 22, 8, 0, 0, TimeSpan.Zero); + + for (int i = 0; i < 5; i++) + { + await _repo.AppendHistoryAsync( + Snap(machineId, t0.AddMinutes(i), new DataItem("d1", "Speed", i.ToString(), "Sample", t0.AddMinutes(i))), + CancellationToken.None); + } + + // Cutoff at minute 3: rows < t0+3min (minutes 0,1,2) removed, 3 and 4 kept. + Result prune = await _repo.PruneHistoryAsync(t0.AddMinutes(3), CancellationToken.None); + Assert.True(prune.IsSuccess, Describe(prune)); + + Result> got = + await _repo.GetHistoryAsync(machineId, "d1", DateTimeOffset.MinValue, 100, CancellationToken.None); + Assert.True(got.IsSuccess, Describe(got)); + Assert.Equal(2, got.Value.Count); + Assert.Equal("3", got.Value[0].Value); + Assert.Equal("4", got.Value[1].Value); + + // Idempotent: pruning again with a cutoff below everything removed changes nothing. + Result prune2 = await _repo.PruneHistoryAsync(t0.AddMinutes(3), CancellationToken.None); + Assert.True(prune2.IsSuccess, Describe(prune2)); + } + + [Fact] + public async Task Delete_AlsoClearsMachineHistory() + { + var machineId = Guid.NewGuid(); + var t0 = new DateTimeOffset(2026, 7, 22, 8, 0, 0, TimeSpan.Zero); + + await _repo.UpsertAsync( + new Machine(machineId, "M", "mtconnect", null, TimeSpan.FromSeconds(1)), CancellationToken.None); + await _repo.AppendHistoryAsync( + Snap(machineId, t0, new DataItem("d1", "Speed", "10", "Sample", t0)), + CancellationToken.None); + + Result delete = await _repo.DeleteAsync(machineId, CancellationToken.None); + Assert.True(delete.IsSuccess, Describe(delete)); + + Result> got = + await _repo.GetHistoryAsync(machineId, "d1", DateTimeOffset.MinValue, 100, CancellationToken.None); + Assert.True(got.IsSuccess, Describe(got)); + Assert.Empty(got.Value); + } + + private static MachineSnapshot Snap(Guid machineId, DateTimeOffset capturedAt, params DataItem[] items) => + new MachineSnapshot(machineId, capturedAt, ConnectionState.Connected, items); + private static string Describe(Result result) => result.IsSuccess ? "" : string.Join("; ", result.Errors.Select(e => e.ToString())); diff --git a/tests/Junction.Tests/Unit/SqliteSchemaTests.cs b/tests/Junction.Tests/Unit/SqliteSchemaTests.cs index c518523..6d98c8a 100644 --- a/tests/Junction.Tests/Unit/SqliteSchemaTests.cs +++ b/tests/Junction.Tests/Unit/SqliteSchemaTests.cs @@ -71,6 +71,17 @@ namespace Junction.Tests.Unit Assert.Contains("snapshot_items", tables); } + [Fact] + public void EnsureCreated_CreatesSnapshotHistoryTable() + { + var factory = new SqliteConnectionFactory(_connectionString); + + Result result = SqliteSchema.EnsureCreated(factory); + + Assert.True(result.IsSuccess, DescribeErrors(result)); + Assert.Contains("snapshot_history", QueryTableNames()); + } + [Fact] public void EnsureCreated_IsIdempotent() {