SynchronizationResource.cs

using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Waher.Content;
using Waher.IoTGateway;
using Waher.Networking.HTTP;
using Waher.Networking.XMPP;
using Waher.Networking.XMPP.Events;
using Waher.Networking.XMPP.NeuroLedger;
using Waher.Persistence.NeuroLedger;
using Waher.Script;
using Waher.Security;
 
namespace Waher.Service.NeuroLedger.Web
{
    public class SynchronizationResource : HttpSynchronousResource, IHttpPostMethod
    {
        private readonly NeuroLedgerClient client;
        private readonly string userVariable;
        private readonly string[] userPrivileges;
 
        public SynchronizationResource(string ResourceName, NeuroLedgerClient NLClient, string UserVariable, params string[] UserPrivileges)
            : base(ResourceName)
        {
            this.client = NLClient;
            this.userVariable = UserVariable;
            this.userPrivileges = UserPrivileges;
        }
 
        public override bool HandlesSubPaths => false;
 
        public override bool UserSessions => true;
 
        public bool AllowsPOST => true;
 
        public async Task POST(HttpRequest Request, HttpResponse Response)
        {
            if (Request.Session is null ||
                !Request.Session.TryGetVariable(this.userVariable, out Variable v) ||
                !(v.ValueObject is IUser User))
            {
                throw new ForbiddenException("User not authenticated.");
            }
 
            foreach (string Privilege in this.userPrivileges)
            {
                if (!User.HasPrivilege(Privilege))
                    throw new ForbiddenException("User not authorized.");
            }
 
            string TabID = Request.Header["X-TabID"];
            if (string.IsNullOrEmpty(TabID))
                throw new BadRequestException("Expected X-TabID header.");
 
            string Command = Request.Header["X-Command"];
            if (string.IsNullOrEmpty(Command))
                throw new BadRequestException("Expected command.");
 
            if (Gateway.XmppClient.State != XmppState.Connected)
                throw new ConflictException("The Neuro-Ledger is not currently connected to the network. Synchronization cannot start.");
 
            switch (Command)
            {
                case "START":
                    this.Synchronize(TabID);
                    break;
 
                case "STOP":
                    await this.StopSynchronization(TabID);
                    break;
 
                default:
                    throw new BadRequestException("Invalid command.");
            }
        }
 
        private async void Synchronize(string TabID)
        {
            try
            {
                await ClientEvents.PushEvent(new string[] { TabID }, "AddContact", JSON.Encode(new KeyValuePair<string, object>[]
                {
                        new KeyValuePair<string, object>("jid", this.client.Provider.ExternalIdentity),
                        new KeyValuePair<string, object>("name", this.client.Provider.ExternalIdentity)
                }, false), true, this.userVariable, this.userPrivileges);
 
                await ClientEvents.PushEvent(new string[] { TabID }, "ReportAvailability", JSON.Encode(new KeyValuePair<string, object>[]
                {
                        new KeyValuePair<string, object>("jid", this.client.Provider.ExternalIdentity),
                        new KeyValuePair<string, object>("type", "Neuro-Ledger"),
                        new KeyValuePair<string, object>("availability", "Local")
                }, false), true, this.userVariable, this.userPrivileges);
 
                LocalStatus LocalStatus = new LocalStatus()
                {
                    TabID = TabID
                };
 
                await this.client.Provider.RepairRegistry(
                    (Sender, e) =>
                    {
                        LocalStatus.NrBlocks++;
                        return this.Report(LocalStatus, e);
                    },
                    (Sender, e) =>
                    {
                        LocalStatus.NrAdded++;
                        LocalStatus.NrBlocks++;
                        return this.Report(LocalStatus, e);
                    },
                    (Sender, e) =>
                    {
                        LocalStatus.NrUpdated++;
                        LocalStatus.NrBlocks++;
                        return this.Report(LocalStatus, e);
                    },
                    (Sender, e) =>
                    {
                        LocalStatus.NrDeleted++;
                        return this.Report(LocalStatus, e);
                    });
 
                LocalStatus.Last = DateTime.MinValue;
                await this.Report(LocalStatus, true);
 
                LinkedList<Task> Tasks = new LinkedList<Task>();
 
                foreach (RosterItem Item in Gateway.XmppClient.Roster)
                {
                    await ClientEvents.PushEvent(new string[] { TabID }, "AddContact", JSON.Encode(new KeyValuePair<string, object>[]
                    {
                        new KeyValuePair<string, object>("jid", Item.BareJid),
                        new KeyValuePair<string, object>("name", Item.NameOrBareJid)
                    }, false), true, this.userVariable, this.userPrivileges);
 
                    PeerStatus Status = null;
                    PresenceEventArgs Peer = null;
                    bool HasPresence = false;
 
                    foreach (PresenceEventArgs e in Item.Resources)
                    {
                        HasPresence = true;
 
                        Status = await this.client.GetPeerStatus(e);
                        if (Status is null)
                            continue;
 
                        if (Status.IsNeuroLedger)
                        {
                            Peer = e;
                            break;
                        }
                    }
 
                    string Availability;
                    string Type;
 
                    if (!HasPresence)
                    {
                        Type = "Unknown";
                        Availability = "Offline";
                    }
                    else if (!(Peer is null))
                    {
                        Type = "Neuro-Ledger";
                        Availability = Peer.Availability.ToString();
                    }
                    else
                    {
                        Type = "Unknown";
                        Availability = Item.LastPresence?.Availability.ToString() ?? "Offline";
                    }
 
                    await ClientEvents.PushEvent(new string[] { TabID }, "ReportAvailability", JSON.Encode(new KeyValuePair<string, object>[]
                    {
                        new KeyValuePair<string, object>("jid", Item.BareJid),
                        new KeyValuePair<string, object>("type", Type),
                        new KeyValuePair<string, object>("availability", Availability)
                    }, false), true, this.userVariable, this.userPrivileges);
 
                    if (Peer is null || !Peer.IsOnline)
                        continue;
 
                    Tasks.AddLast(Task.Run(() => this.SynchronizePeer(TabID, Status, Item)));
                }
 
                if (Tasks.Count > 0)
                    await Task.WhenAll(Tasks);
            }
            catch (Exception ex)
            {
                await ClientEvents.PushEvent(new string[] { TabID }, "ShowError", ex.Message, false, this.userVariable, this.userPrivileges);
            }
        }
 
        private class LocalStatus
        {
            public string TabID;
            public int NrBlocks = 0;
            public int NrAdded = 0;
            public int NrUpdated = 0;
            public int NrDeleted = 0;
            public DateTime Last = DateTime.Now;
            public DateTime FirstBlock = DateTime.MaxValue;
            public DateTime LastBlock = DateTime.MinValue;
            public ulong TotalBytes = 0;
        }
 
        private Task Report(LocalStatus LocalStatus, BlockReferenceEventArgs e)
        {
            if (e.Block.Created < LocalStatus.FirstBlock)
                LocalStatus.FirstBlock = e.Block.Created;
 
            if (e.Block.Created > LocalStatus.LastBlock)
                LocalStatus.LastBlock = e.Block.Created;
 
            LocalStatus.TotalBytes += e.Block.Bytes;
 
            return this.Report(LocalStatus, false);
        }
 
        private async Task Report(LocalStatus LocalStatus, bool Done)
        {
            DateTime Now = DateTime.Now;
            if ((Now - LocalStatus.Last).TotalSeconds >= 1)
            {
                LocalStatus.Last = Now;
 
                try
                {
                    await ClientEvents.PushEvent(new string[] { LocalStatus.TabID }, "StatusUpdate", JSON.Encode(new KeyValuePair<string, object>[]
                    {
                        new KeyValuePair<string, object>("jid", this.client.Provider.ExternalIdentity),
                        new KeyValuePair<string, object>("first", LocalStatus.FirstBlock == DateTime.MaxValue ? string.Empty : LocalStatus.FirstBlock.ToShortDateString()),
                        new KeyValuePair<string, object>("last", LocalStatus.LastBlock == DateTime.MinValue ? string.Empty : LocalStatus.LastBlock.ToShortDateString()),
                        new KeyValuePair<string, object>("nrBlocks", LocalStatus.NrBlocks),
                        new KeyValuePair<string, object>("nrNew", LocalStatus.NrAdded),
                        new KeyValuePair<string, object>("nrUpdated", LocalStatus.NrUpdated),
                        new KeyValuePair<string, object>("blockErrors", LocalStatus.NrDeleted),
                        new KeyValuePair<string, object>("totalBytes", LocalStatus.TotalBytes),
                        new KeyValuePair<string, object>("done", Done),
                        new KeyValuePair<string, object>("nrLoaded", string.Empty),
                        new KeyValuePair<string, object>("nrDenied", string.Empty),
                        new KeyValuePair<string, object>("loadedBytes", string.Empty),
                        new KeyValuePair<string, object>("events", string.Empty),
                        new KeyValuePair<string, object>("objectsAdded", string.Empty),
                        new KeyValuePair<string, object>("objectsUpdated", string.Empty),
                        new KeyValuePair<string, object>("objectsDeleted", string.Empty),
                        new KeyValuePair<string, object>("objectErrors", string.Empty)
 
                    }, false), true, this.userVariable, this.userPrivileges);
                }
                catch (Exception)
                {
                    // Ignore
                }
            }
        }
 
        private async Task StopSynchronization(string TabID)
        {
            try
            {
                foreach (RosterItem Item in Gateway.XmppClient.Roster)
                {
                    PeerStatus Status = null;
                    PresenceEventArgs Peer = null;
 
                    foreach (PresenceEventArgs e in Item.Resources)
                    {
                        Status = await this.client.GetPeerStatus(e);
                        if (Status is null)
                            continue;
 
                        if (Status.IsNeuroLedger)
                        {
                            Peer = e;
                            break;
                        }
                    }
 
                    if (Peer is null)
                        continue;
 
                    this.client.StopSynchronization(Status);
                }
            }
            catch (Exception ex)
            {
                await ClientEvents.PushEvent(new string[] { TabID }, "ShowError", ex.Message, false, this.userVariable, this.userPrivileges);
            }
        }
 
        private async Task SynchronizePeer(string TabID, PeerStatus Status, RosterItem Item)
        {
            try
            {
                string Msg = await this.client.SynchronizePeer(Status, true, async (Progress) =>
                {
                    await ClientEvents.PushEvent(new string[] { TabID }, "StatusUpdate", JSON.Encode(new KeyValuePair<string, object>[]
                    {
                        new KeyValuePair<string, object>("jid", Item.BareJid),
                        new KeyValuePair<string, object>("first", Progress.First == DateTime.MaxValue ? string.Empty : Progress.First.ToShortDateString()),
                        new KeyValuePair<string, object>("last", Progress.Last == DateTime.MinValue ? string.Empty : Progress.Last.ToShortDateString()),
                        new KeyValuePair<string, object>("nrBlocks", Progress.NrBlocks),
                        new KeyValuePair<string, object>("nrNew", Progress.NrNew),
                        new KeyValuePair<string, object>("nrUpdated", Progress.NrUpdated),
                        new KeyValuePair<string, object>("nrLoaded", Progress.NrLoaded),
                        new KeyValuePair<string, object>("nrDenied", Progress.NrDenied),
                        new KeyValuePair<string, object>("blockErrors", Progress.BlockErrors),
                        new KeyValuePair<string, object>("totalBytes", Progress.TotalBytes),
                        new KeyValuePair<string, object>("loadedBytes", Progress.LoadedBytes),
                        new KeyValuePair<string, object>("events", Progress.Events),
                        new KeyValuePair<string, object>("objectsAdded", Progress.ObjectsAdded),
                        new KeyValuePair<string, object>("objectsUpdated", Progress.ObjectsUpdated),
                        new KeyValuePair<string, object>("objectsDeleted", Progress.ObjectsDeleted),
                        new KeyValuePair<string, object>("objectErrors", Progress.ObjectErrors),
                        new KeyValuePair<string, object>("done", Progress.Done)
                    }, false), true, this.userVariable, this.userPrivileges);
 
                    KeyValuePair<string, string>[] Errors = Progress.PopBlockErrorMessages();
                    if (!(Errors is null))
                    {
                        foreach (KeyValuePair<string, string> P in Errors)
                        {
                            await ClientEvents.PushEvent(new string[] { TabID }, "BlockError", JSON.Encode(new KeyValuePair<string, object>[]
                            {
                                new KeyValuePair<string, object>("time", DateTime.Now.ToLongTimeString()),
                                new KeyValuePair<string, object>("peer", P.Key),
                                new KeyValuePair<string, object>("message", P.Value)
                            }, false), true, this.userVariable, this.userPrivileges);
                        }
                    }
                });
 
                if (!string.IsNullOrEmpty(Msg))
                {
                    await ClientEvents.PushEvent(new string[] { TabID }, "UnableToSynchronize", JSON.Encode(new KeyValuePair<string, object>[]
                    {
                        new KeyValuePair<string, object>("jid", Item.BareJid),
                        new KeyValuePair<string, object>("message", Msg)
                    }, false), true, this.userVariable, this.userPrivileges);
                }
            }
            catch (Exception ex)
            {
                await ClientEvents.PushEvent(new string[] { TabID }, "ShowError", ex.Message, false, this.userVariable, this.userPrivileges);
            }
        }
 
    }
}