jmon/JMonServer/Program.cs

using JMonAgent;
using JMon.Protos;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddGrpc();

// Track latest full metrics per machine (keyed by HWID:PID)
var latestMetrics = new Dictionary<string, SystemMetrics>();

// Track pending metrics per machine for replay in timestamp order
var pendingMetrics = new Dictionary<string, List<JMon.Protos.DeltaMetrics>>();

// Track which process is active per HWID (for duplicate detection)
var activeProcesses = new Dictionary<string, (int ProcessId, DateTime LastSeen)>();

builder.Services.AddSingleton(latestMetrics);
builder.Services.AddSingleton(pendingMetrics);
builder.Services.AddSingleton(activeProcesses);

var app = builder.Build();
app.MapGrpcService<MetricsServiceImpl>();
// Use environment variables for URL, fallback to HTTP on 0.0.0.0:5001
var urls = Environment.GetEnvironmentVariable("ASPNETCORE_URLS") ?? "http://0.0.0.0:5001";
app.Run(urls);

public class MetricsServiceImpl : MetricsService.MetricsServiceBase
{
    private readonly Dictionary<string, SystemMetrics> _latestMetrics;
    private readonly Dictionary<string, List<JMon.Protos.DeltaMetrics>> _pendingMetrics;
    private readonly Dictionary<string, (int ProcessId, DateTime LastSeen)> _activeProcesses;

    public MetricsServiceImpl(Dictionary<string, SystemMetrics> latestMetrics, Dictionary<string, List<JMon.Protos.DeltaMetrics>> pendingMetrics, Dictionary<string, (int ProcessId, DateTime LastSeen)> activeProcesses)
    {
        _latestMetrics = latestMetrics;
        _pendingMetrics = pendingMetrics;
        _activeProcesses = activeProcesses;
    }

    public override async Task<JMon.Protos.MetricsResponse> ReportMetrics(JMon.Protos.DeltaMetrics request, Grpc.Core.ServerCallContext context)
    {
        var hwid = request.Hwid;
        
        // Check for duplicate processes on same HWID
        var command = MetricsResponse.Types.CommandType.Ack;
        var message = "Metrics received";
        
        if (_activeProcesses.ContainsKey(hwid))
        {
            var (activePid, lastSeen) = _activeProcesses[hwid];
            
            if (activePid != request.ProcessId)
            {
                // Multiple processes detected - SIGTERM the older one, SYNC the newer
                if (request.ProcessId > activePid)
                {
                    // New process is newer - terminate the old one
                    command = MetricsResponse.Types.CommandType.Sigterm;
                    message = $"Duplicate process detected. Terminating PID {activePid}";
                    _activeProcesses[hwid] = (request.ProcessId, DateTime.UtcNow);
                }
                else
                {
                    // Current request is older - ask for full sync
                    command = MetricsResponse.Types.CommandType.Sync;
                    message = $"Duplicate process detected. Newer process (PID {activePid}) is active";
                }
            }
            else
            {
                // Same process, just update last seen
                _activeProcesses[hwid] = (request.ProcessId, DateTime.UtcNow);
            }
        }
        else
        {
            // First process for this HWID
            _activeProcesses[hwid] = (request.ProcessId, DateTime.UtcNow);
        }
        
        // If this is a retried metric, queue it for ordered replay
        if (request.Retried)
        {
            if (!_pendingMetrics.ContainsKey(hwid))
                _pendingMetrics[hwid] = new List<JMon.Protos.DeltaMetrics>();
            _pendingMetrics[hwid].Add(request);
        }
        
        // Process all pending metrics for this machine in timestamp order
        if (_pendingMetrics.ContainsKey(hwid) && _pendingMetrics[hwid].Count > 0)
        {
            var pending = _pendingMetrics[hwid];
            // Sort by timestamp (oldest first)
            var sorted = pending.OrderBy(m => m.TimestampUnixMs).ToList();
            
            foreach (var metric in sorted)
            {
                ApplyMetric(hwid, metric);
            }
            
            // Clear processed pending metrics
            _pendingMetrics[hwid].Clear();
        }
        
        // Apply the current metric
        if (!request.Retried)
        {
            ApplyMetric(hwid, request);
        }
        
        var serverTime = DateTime.UtcNow;
        var agentTime = DateTime.UnixEpoch.AddMilliseconds(request.TimestampUnixMs);
        var timeDeltaMs = Math.Abs((serverTime - agentTime).TotalMilliseconds);
        
        var key = $"{hwid}:{request.ProcessId}";
        if (_latestMetrics.TryGetValue(key, out var latest))
        {
            DisplayMetrics("[gRPC]", latest, timeDeltaMs, request.Retried);
        }
        
        return await Task.FromResult(new JMon.Protos.MetricsResponse
        {
            Success = true,
            Command = command,
            Message = message
        });
    }
    
    private void ApplyMetric(string hwid, JMon.Protos.DeltaMetrics delta)
    {
        var agentTime = DateTime.UnixEpoch.AddMilliseconds(delta.TimestampUnixMs);
        
        // Use HWID + PID as key
        var key = $"{hwid}:{delta.ProcessId}";
        
        // Reconstruct full metrics by merging delta with previous state
        SystemMetrics full;
        if (_latestMetrics.TryGetValue(key, out var previous))
        {
            full = new SystemMetrics(
                delta.MachineName,
                delta.HasOsDescription ? delta.OsDescription : previous.OSDescription,
                delta.HasCpuUsagePercentage ? delta.CpuUsagePercentage : previous.CpuUsagePercentage,
                delta.HasMemoryUsedBytes ? delta.MemoryUsedBytes : previous.MemoryUsedBytes,
                delta.HasMemoryTotalBytes ? delta.MemoryTotalBytes : previous.MemoryTotalBytes,
                delta.HasSwapUsedBytes ? delta.SwapUsedBytes : previous.SwapUsedBytes,
                delta.HasSwapTotalBytes ? delta.SwapTotalBytes : previous.SwapTotalBytes,
                delta.HasDiskUsedBytes ? delta.DiskUsedBytes : previous.DiskUsedBytes,
                delta.HasDiskTotalBytes ? delta.DiskTotalBytes : previous.DiskTotalBytes,
                delta.HasDiskMountPoint ? delta.DiskMountPoint : previous.DiskMountPoint,
                agentTime,
                delta.HasIpAddress ? delta.IpAddress : previous.IpAddress,
                hwid,
                delta.ProcessId
            );
        }
        else
        {
            // First time - initialize with sent values or defaults
            full = new SystemMetrics(
                delta.MachineName,
                delta.HasOsDescription ? delta.OsDescription : "Unknown",
                delta.HasCpuUsagePercentage ? delta.CpuUsagePercentage : 0,
                delta.HasMemoryUsedBytes ? delta.MemoryUsedBytes : 0,
                delta.HasMemoryTotalBytes ? delta.MemoryTotalBytes : 0,
                delta.HasSwapUsedBytes ? delta.SwapUsedBytes : 0,
                delta.HasSwapTotalBytes ? delta.SwapTotalBytes : 0,
                delta.HasDiskUsedBytes ? delta.DiskUsedBytes : 0,
                delta.HasDiskTotalBytes ? delta.DiskTotalBytes : 0,
                delta.HasDiskMountPoint ? delta.DiskMountPoint : "/",
                agentTime,
                delta.HasIpAddress ? delta.IpAddress : "",
                hwid,
                delta.ProcessId
            );
        }
        
        _latestMetrics[key] = full;
    }

    private void DisplayMetrics(string source, SystemMetrics full, double timeDeltaMs, bool retried)
    {
        var tag = retried ? "[RETRIED]" : "";
        Console.WriteLine($"{source} {tag} {full.MachineName} ({full.HwId}) [PID: {full.ProcessId}]:");
        Console.WriteLine($"  Time Skew: {timeDeltaMs:F1}ms");
        
        if (timeDeltaMs > 60000)
        {
            Console.ForegroundColor = ConsoleColor.Yellow;
            Console.WriteLine($"  ⚠️  ALERT: Clock skew detected! {timeDeltaMs:F0}ms difference");
            Console.ResetColor();
        }
        
        if (!string.IsNullOrEmpty(full.IpAddress))
            Console.WriteLine($"  IP: {full.IpAddress}");
        Console.WriteLine($"  CPU: {full.CpuUsagePercentage:F2}%");
        var memPercent = (full.MemoryTotalBytes > 0) ? (full.MemoryUsedBytes * 100.0 / full.MemoryTotalBytes).ToString("F1") : "0.0";
        Console.WriteLine($"  Memory: {full.MemoryUsedBytes / (1024*1024)} MB / {full.MemoryTotalBytes / (1024*1024)} MB ({memPercent}%)");
        var swapPercent = (full.SwapTotalBytes > 0) ? (full.SwapUsedBytes * 100.0 / full.SwapTotalBytes).ToString("F1") : "0.0";
        Console.WriteLine($"  Swap: {full.SwapUsedBytes / (1024*1024)} MB / {full.SwapTotalBytes / (1024*1024)} MB ({swapPercent}%)");
        var diskPercent = (full.DiskTotalBytes > 0) ? (full.DiskUsedBytes * 100.0 / full.DiskTotalBytes).ToString("F1") : "0.0";
        Console.WriteLine($"  Disk ({full.DiskMountPoint}): {full.DiskUsedBytes / (1024*1024)} MB / {full.DiskTotalBytes / (1024*1024)} MB ({diskPercent}%)");
        Console.WriteLine($"  OS: {full.OSDescription}\n");
    }
}