prs/prs.server/Engines/GPS/GPSEngine.cs

using Comal.Classes;
using InABox.Clients;
using InABox.Core;
using InABox.DigitalMatter;
using InABox.IPC;
using netDxf.Tables;
using PRSServer.Engines;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading.Tasks;
using InABox.Rpc;
using System.Timers;
using PRS.Shared;
using PRSServices;
using com.sun.accessibility.@internal.resources;

namespace PRSServer;

internal class Device
{
    public Guid ID { get; set; }
    public DateTime TimeStamp { get; set; }
    public CoreExpression<GPSBatteryFormulaModel, double>? BatteryFormula { get; set; }

    public Device(Guid iD, DateTime timeStamp, CoreExpression<GPSBatteryFormulaModel, double>? batteryFormula)
    {
        ID = iD;
        TimeStamp = timeStamp;
        BatteryFormula = batteryFormula;
    }

    public double CalculateBatteryLevel(double batteryValue)
    {
        if(BatteryFormula != null)
        {
            return BatteryFormula.Evaluate(new Dictionary<string, object?>
            {
                { nameof(GPSBatteryFormulaModel.BatteryLevel), batteryValue }
            });
        }
        return batteryValue;
    }
}

public class GPSDeviceUpdate : ISerializeBinary
{
    public string AuditTrail { get; set; }

    public GPSTrackerLocation Location { get; set; }

    public void SerializeBinary(CoreBinaryWriter writer)
    {
        writer.Write(AuditTrail ?? "");
        writer.WriteObject(Location);
    }

    public void DeserializeBinary(CoreBinaryReader reader)
    {
        AuditTrail = reader.ReadString();
        Location = reader.ReadObject<GPSTrackerLocation>();
    }
}

public class GPSUpdateQueue
{
    public string QueuePath;

    public GPSUpdateQueue(string queuePath)
    {
        QueuePath = queuePath;
    }

    public void InitQueueFolder()
    {
        try
        {
            Directory.CreateDirectory(QueuePath);
        }
        catch (Exception e)
        {
            throw new Exception($"Could not create directory for device update queue: {QueuePath}", e);
        }
    }

    public int GetNumberOfItems()
    {
        return Directory.EnumerateFiles(QueuePath).Count();
    }

    /// <summary>
    /// Get the first (earliest) items of the directory.
    /// </summary>
    /// <returns>A list of (filename, update) tuples.</returns>
    public IEnumerable<Tuple<string, GPSDeviceUpdate>> GetFirstItems()
    {
        var files = Directory.EnumerateFiles(QueuePath).OrderBy(x => x);

        foreach (var filename in files)
        {
            GPSDeviceUpdate? deviceUpdate = null;
            try
            {
                using var fileStream = new FileStream(filename, FileMode.Open, FileAccess.Read);
                deviceUpdate = Serialization.ReadBinary<GPSDeviceUpdate>(fileStream, BinarySerializationSettings.Latest);
            }
            catch
            {
                // File is probably in use.
            }
            if(deviceUpdate is not null)
            {
                yield return new Tuple<string, GPSDeviceUpdate>(filename, deviceUpdate);
            }
        }
    }

    public void QueueUpdate(GPSDeviceUpdate deviceUpdate)
    {
        var filename = Path.Combine(QueuePath, $"{DateTime.UtcNow.Ticks} - {deviceUpdate.Location.Tracker.ID}");

        using var fileStream = new FileStream(filename, FileMode.OpenOrCreate, FileAccess.Write);
        Serialization.WriteBinary(deviceUpdate, fileStream, BinarySerializationSettings.Latest);
    }

    public void QueueUpdate(string auditTrail, GPSTrackerLocation location) => QueueUpdate(new GPSDeviceUpdate
    {
        AuditTrail = auditTrail,
        Location = location
    });
}

internal class GPSDeviceCache : ConcurrentDictionary<string, Device>
{
    public void Refresh()
    {
        Logger.Send(LogType.Information, "", "Refreshing Tracker Cache");
        var table = new Client<GPSTracker>().Query(
            null,
            new Columns<GPSTracker>(x => x.ID, x => x.DeviceID, x => x.Type.BatteryFormula));

        Logger.Send(LogType.Information, "", string.Format("- Tracker Cache: {0} devices", table.Rows.Count));
        Clear();
        foreach (var row in table.Rows)
        {
            var formula = row.Get<GPSTracker, string?>(x => x.Type.BatteryFormula);
            var expression = string.IsNullOrWhiteSpace(formula) ? null : new CoreExpression<GPSBatteryFormulaModel, double>(formula);
            this[row.Get<GPSTracker, string>(x => x.DeviceID)] =
                new Device(row.Get<GPSTracker, Guid>(x => x.ID), DateTime.MinValue, expression);
        }
    }
}

public class GPSEngine : Engine<GPSServerProperties>
{
    private Listener<SigfoxHandler, SigfoxHandlerProperties> sigfoxListener;
    private OEMListener oemListener;
    private RpcClientPipeTransport transport;

    private GPSDeviceCache DeviceCache = new();

    private Timer RefreshDevicesTimer;

    private Timer UpdateServerTimer;

    private GPSUpdateQueue UpdateQueue;

    public override void Configure(Server server)
    {
        base.Configure(server);

        UpdateQueue = new GPSUpdateQueue(Path.Combine(AppDataFolder, "device_queue"));
    }

    private void StartOEMListener()
    {
        if (Properties.ListenPort == 0)
            throw new Exception("Error: OEM Listen Port not Specified\n");

        Logger.Send(LogType.Information, "", "Starting OEM Listener on port " + Properties.ListenPort);

        oemListener = new OEMListener(Properties.ListenPort, DeviceCache, UpdateQueue);
        oemListener.Start();

        Logger.Send(LogType.Information, "", "OEM Listener started on port " + Properties.ListenPort);
    }

    private void StartSigfoxListener()
    {
        if (Properties.SigfoxListenPort == 0)
        {
            Logger.Send(LogType.Information, "", "No Sigfox listen port specified\n");
            return;
        }

        sigfoxListener = new Listener<SigfoxHandler, SigfoxHandlerProperties>(new SigfoxHandlerProperties(DeviceCache, UpdateQueue));
        sigfoxListener.InitPort((ushort)Properties.SigfoxListenPort);

        Logger.Send(LogType.Information, "", "Starting Sigfox Listener on port " + Properties.SigfoxListenPort);

        sigfoxListener.Start();

        //var transport = new RpcClientPipeTransport(DatabaseServerProperties.GetPipeName(Properties.Server));
        //ClientFactory.SetClientType(typeof(RpcClient<>), Platform.GPSEngine, Version, transport);

        //CheckConnection();

        Logger.Send(LogType.Information, "", "Sigfox Listener started on port " + Properties.SigfoxListenPort);
    }
    
    private void StartUpdateServerTask()
    {
        UpdateServerTimer = new Timer(Properties.UpdateTimer);
        UpdateServerTimer.Elapsed += (o, e) => UpdateServer();
        UpdateServerTimer.Start();
    }

    // List of (filename, update)
    private Queue<Tuple<string, GPSDeviceUpdate>> LocationQueueCache = new();

    private void GetLocationQueue(int nLocations)
    {
        LocationQueueCache.EnsureCapacity(LocationQueueCache.Count + nLocations);
        foreach(var item in UpdateQueue.GetFirstItems().Take(nLocations))
        {
            LocationQueueCache.Enqueue(item);
        }
    }

    private void UpdateServer()
    {
        if (!transport.IsConnected()) return;

        // Cache a set of fifty, so that we're not running baack and forth to the filesystem all the time.
        if(LocationQueueCache.Count == 0)
        {
            GetLocationQueue(50);
        }

        if (LocationQueueCache.Count > 0)
        {
            var (filename, update) = LocationQueueCache.Dequeue();

            Logger.Send(LogType.Information, "",
                string.Format("Updating Server ({0}): {1} - {2}", UpdateQueue.GetNumberOfItems(), update.Location.DeviceID, update.AuditTrail));
            new Client<GPSTrackerLocation>().Save(update.Location, update.AuditTrail, (_, exception) =>
            {
                if (exception is not null)
                {
                    Logger.Send(LogType.Error, "", $"Error saving GPS Tracker Location ({update.AuditTrail}): {CoreUtils.FormatException(exception)}");

                    // We won't delete the file in case of an error; this means though that the queue will be wrong, since the item has been dequeued, so we will clear the queue.
                    LocationQueueCache.Clear();
                }
                else
                {
                    try
                    {
                        File.Delete(filename);
                    }
                    catch
                    {
                        // Probably got deleted.
                    }
                }
            });
        }
    }

    public override void Run()
    {
        if (string.IsNullOrWhiteSpace(Properties.Server))
        {
            Logger.Send(LogType.Error, "", "Server is blank!");
            return;
        }

        Logger.Send(LogType.Information, "", "Registering Classes");

        CoreUtils.RegisterClasses();
        ComalUtils.RegisterClasses();
        PRSSharedUtils.RegisterClasses();

        //ClientFactory.SetClientType(typeof(IPCClient<>), Platform.GPSEngine, Version, DatabaseServerProperties.GetPipeName(Properties.Server,false));
        transport = new RpcClientPipeTransport(DatabaseServerProperties.GetPipeName(Properties.Server, true));
        ClientFactory.SetClientType(typeof(RpcClient<>), Platform.GPSEngine, Version, transport);
        transport.OnClose += Transport_OnClose;
        CheckConnection();
        ClientFactory.SetBypass();

        UpdateQueue.InitQueueFolder();

        // Refresh device cache and set up timer.
        DeviceCache.Refresh();
        RefreshDevicesTimer = new Timer(5 * 60 * 1000);
        RefreshDevicesTimer.Elapsed += (o, e) => DeviceCache.Refresh();
        RefreshDevicesTimer.Start();

        DMFactory.Initialise(Properties.DumpFormat, Properties.DumpFile);

        StartOEMListener();
        StartSigfoxListener();
        StartUpdateServerTask();
    }

    private void Transport_OnClose(IRpcTransport transport, RpcTransportCloseArgs e)
    {
        // Try to reconnect when lost connection.
        if (!_connecting)
        {
            CheckConnection();
        }
    }

    private bool _connecting = false;

    private bool CheckConnection()
    {
        _connecting = true;
        // Wait for server connection
        while (!Client.Ping())
        {
            Logger.Send(LogType.Error, "", "Database server unavailable. Trying again in 30 seconds...");
            Task.Delay(30_000).Wait();
            Logger.Send(LogType.Information, "", "Retrying connection...");
        }
        _connecting = false;

        return true;
    }

    public override void Stop()
    {
        oemListener.Stop();
        sigfoxListener.Stop();
        UpdateServerTimer.Stop();
        RefreshDevicesTimer.Stop();
    }
}