NeuroLedgerClient.cs

using System;
using System.Collections.Generic;
using System.IO;
using System.Reflection;
using System.Text;
using System.Threading.Tasks;
using System.Xml;
using Waher.Content;
using Waher.Content.Multipart;
using Waher.Content.Xml;
using Waher.Content.Xml.Text;
using Waher.Events;
using Waher.Networking.XMPP.Events;
using Waher.Networking.XMPP.NeuroLedger.Events;
using Waher.Networking.XMPP.NeuroLedger.PepEvents;
using Waher.Networking.XMPP.NeuroLedger.Web;
using Waher.Networking.XMPP.P2P;
using Waher.Networking.XMPP.PEP;
using Waher.Networking.XMPP.PEP.Events;
using Waher.Networking.XMPP.ServiceDiscovery;
using Waher.Networking.XMPP.StanzaErrors;
using Waher.Persistence;
using Waher.Persistence.Attributes;
using Waher.Persistence.Exceptions;
using Waher.Persistence.Filters;
using Waher.Persistence.NeuroLedger;
using Waher.Persistence.NeuroLedger.Registry;
using Waher.Persistence.Serialization;
using Waher.Runtime.Cache;
using Waher.Runtime.Inventory;
using Waher.Runtime.Timing;
 
namespace Waher.Networking.XMPP.NeuroLedger
{
    public class NeuroLedgerClient : XmppExtension
    {
        public const string NeuroLedgerNamespace = "http://waher.se/NL";
 
        public const int DefaultTimeoutMs = 2 * 60 * 1000;
 
        private static readonly BlockAdded blockAddedRef = new BlockAdded();
        private Cache<string, PeerStatus> peerStatus;
        private readonly Dictionary<string, NodeStatus> nodeStatus = new Dictionary<string, NodeStatus>();
        private readonly Dictionary<string, bool> loading = new Dictionary<string, bool>();
        private readonly BlockResource blockResource;
        private readonly HTTP.HttpResource blockListResource;
        private readonly HTTP.HttpResource multiGetResource;
        private readonly HTTP.HttpServer webServer;
        private readonly EndpointSecurity e2eEncryption;
        private readonly bool internalScheduler;
        private readonly ILedgerExternalEvents externalEvents;
        private NeuroLedgerProvider provider;
        private PepClient pepClient = null;
        private Scheduler scheduler;
 
        public NeuroLedgerClient(XmppClient Client, EndpointSecurity E2eEncryption, HTTP.HttpServer WebServer, NeuroLedgerProvider Provider)
            : base(Client)
        {
            this.provider = Provider;
            this.webServer = WebServer;
            this.e2eEncryption = E2eEncryption;
            this.externalEvents = Provider.ExternalEvents;
 
            this.peerStatus = new Cache<string, PeerStatus>(int.MaxValue, TimeSpan.MaxValue, TimeSpan.FromDays(1), true);
 
            if (Types.TryGetModuleParameter("Scheduler", out Scheduler Scheduler))
            {
                this.scheduler = Scheduler;
                this.internalScheduler = false;
            }
            else
            {
                this.scheduler = new Scheduler();
                this.internalScheduler = true;
            }
 
            this.webServer.Register(this.blockResource = new BlockResource("/NL/B", this.provider, this.client, this, this.webServer, "User"));
            this.blockListResource = this.webServer.Register(new BlockListResource("/NL/L", this.client, this.webServer, "User"));
            this.multiGetResource = this.webServer.Register(new MultiGetResource("/NL/MG", this.client, this.webServer, "User"));
 
            Client.RegisterIqGetHandler("getBlockReferences", NeuroLedgerNamespace, this.GetBlockReferences, true);
 
            foreach (IXmppExtension Extension in Client.Extensions)
            {
                if (Extension is PepClient PepClient)
                {
                    this.pepClient = PepClient;
 
                    this.pepClient.RegisterHandler(typeof(BlockAdded), this.BlockAddedReceived);
                    this.pepClient.RegisterHandler(typeof(BlockDeleted), this.BlockDeletedReceived);
                    break;
                }
            }
 
            Provider.BlockAdded += this.BlockAdded;
            Provider.BlockDeleted += this.BlockDeleted;
 
            this.client.OnStateChanged += this.Client_OnStateChanged;
            this.client.OnPresence += this.Client_OnPresence;
 
            if (this.client.State == XmppState.Connected)
            {
                Task _ = this.ResendPresences();
            }
        }
 
        public NeuroLedgerProvider Provider => this.provider;
 
        private async Task ResendPresences()
        {
            foreach (RosterItem Item in this.client.Roster)
            {
                foreach (PresenceEventArgs e in Item.Resources)
                    await this.Client_OnPresence(this.client, e);
            }
        }
 
        private async Task Client_OnPresence(object Sender, PresenceEventArgs e)
        {
            PeerStatus PeerStatus = await this.GetPeerStatus(e);
            if (PeerStatus is null)
                return;
 
            Task _ = Task.Run(() => this.SynchronizePeer(PeerStatus, false, null));
        }
 
        public async Task<string> SynchronizePeer(PeerStatus PeerStatus, bool StartFromBeginning, CallbackAsync<SynchronizationStatus> Callback)
        {
            try
            {
                if (!PeerStatus.IsNeuroLedger)
                    return "Peer not a Neuro-Ledger node.";
 
                NodeStatus NodeStatus;
 
                lock (this.nodeStatus)
                {
                    if (!this.nodeStatus.TryGetValue(PeerStatus.BareJid, out NodeStatus))
                        NodeStatus = null;
                }
 
                if (NodeStatus is null)
                {
                    NodeStatus = await Database.FindFirstDeleteRest<NodeStatus>(new FilterFieldEqualTo("BareJid", PeerStatus.BareJid));
                    if (NodeStatus is null)
                    {
                        NodeStatus = new NodeStatus()
                        {
                            BareJid = PeerStatus.BareJid
                        };
 
                        await Database.Insert(NodeStatus);
                    }
                }
 
                NodeStatus ToDelete = null;
 
                lock (this.nodeStatus)
                {
                    if (this.nodeStatus.TryGetValue(PeerStatus.BareJid, out NodeStatus NodeStatus2))
                    {
                        if (NodeStatus2 != NodeStatus)
                        {
                            ToDelete = NodeStatus;
                            NodeStatus = NodeStatus2;
                        }
                    }
                    else
                        this.nodeStatus[PeerStatus.BareJid] = NodeStatus;
                }
 
                if (!(ToDelete is null))
                    await Database.Delete(ToDelete);
 
                if (StartFromBeginning && !string.IsNullOrEmpty(NodeStatus.LastBlockId))
                {
                    NodeStatus.LastBlockId = string.Empty;
                    await Database.Update(NodeStatus);
                }
 
                return await this.SynchBlocksAsync(PeerStatus, NodeStatus, Callback, Callback is null, false);
            }
            catch (Exception ex)
            {
                return ex.Message;
            }
        }
 
        public async Task<PeerStatus> GetPeerStatus(PresenceEventArgs e)
        {
            if (!e.IsOnline || e.FromBareJID == e.Client.BareJID || e.FromBareJID == e.From || this.peerStatus is null)
                return null;
 
            if (!e.HasEntityCapabilities)
                return null;
 
            if (!this.peerStatus.TryGetValue(e.From, out PeerStatus PeerStatus))
            {
                PeerStatus = new PeerStatus()
                {
                    FullJid = e.From,
                    BareJid = e.FromBareJID
                };
 
                this.peerStatus[e.From] = PeerStatus;
            }
 
            if (PeerStatus.RequestingCapabilitiesVersion == e.EntityCapabilityVersion ||
                PeerStatus.RequestingCapabilitiesNode == e.EntityCapabilityNode ||
                PeerStatus.RequestingCapabilitiesFunction == e.EntityCapabilityHashFunction)
            {
                return null;
            }
 
            if (PeerStatus.CapabilitiesVersion != e.EntityCapabilityVersion ||
                PeerStatus.CapabilitiesNode != e.EntityCapabilityNode ||
                PeerStatus.CapabilitiesFunction != e.EntityCapabilityHashFunction)
            {
                PeerStatus.RequestingCapabilitiesFunction = e.EntityCapabilityHashFunction;
                PeerStatus.RequestingCapabilitiesNode = e.EntityCapabilityNode;
                PeerStatus.RequestingCapabilitiesVersion = e.EntityCapabilityVersion;
 
                try
                {
                    string Node = e.EntityCapabilityNode + "#" + e.EntityCapabilityVersion;
                    ServiceDiscoveryEventArgs e2 = await e.Client.ServiceDiscoveryAsync(null, e.From, Node);
 
                    string[] Features = new string[e2.Features.Count];
                    e2.Features.Keys.CopyTo(Features, 0);
 
                    PeerStatus.CapabilitiesFunction = e.EntityCapabilityHashFunction;
                    PeerStatus.CapabilitiesNode = e.EntityCapabilityNode;
                    PeerStatus.CapabilitiesVersion = e.EntityCapabilityVersion;
                    PeerStatus.Features = Features;
                    PeerStatus.IsNeuroLedger = e2.Features.ContainsKey(NeuroLedgerNamespace);
                }
                finally
                {
                    PeerStatus.RequestingCapabilitiesFunction = string.Empty;
                    PeerStatus.RequestingCapabilitiesNode = string.Empty;
                    PeerStatus.RequestingCapabilitiesVersion = string.Empty;
                }
            }
 
            return PeerStatus;
        }
 
        public void StopSynchronization(PeerStatus PeerStatus)
        {
            lock (PeerStatus)
            {
                if (PeerStatus.IsSynchronizing)
                {
                    PeerStatus.IsSynchronizing = false;
                    PeerStatus.SynchCounter++;
                }
            }
        }
 
        private async Task<string> SynchBlocksAsync(PeerStatus PeerStatus, NodeStatus NodeStatus, CallbackAsync<SynchronizationStatus> Callback,
            bool IgnoreIfRunning, bool Repopulate)
        {
            int SynchCounter;
 
            lock (PeerStatus)
            {
                if (PeerStatus.IsSynchronizing)
                {
                    if (IgnoreIfRunning)
                        return string.Empty;
                }
 
                PeerStatus.IsSynchronizing = true;
                SynchCounter = ++PeerStatus.SynchCounter;
            }
 
            SynchronizationStatus Status = new SynchronizationStatus();
            DateTime LastReported = DateTime.Now;
            int MaxCount = 1000;
            bool More;
 
            try
            {
                List<BlockReference> Blocks = new List<BlockReference>();
                Dictionary<string, bool> Collections = new Dictionary<string, bool>();
                string LastId = NodeStatus.LastBlockId;
                List<KeyValuePair<BlockReference, string>> Bulk = new List<KeyValuePair<BlockReference, string>>();
                ulong BulkSize = 0;
                int Count;
 
                do
                {
                    LastReported = await this.UpdateStatus(LastReported, Status, Callback);
 
                    XmlElement Result = await this.GetBlockList(NodeStatus.BareJid, LastId, MaxCount);
 
                    if (Result is null || Result.LocalName != "blockReferences" || Result.NamespaceURI != NeuroLedgerNamespace)
                        break;
 
                    Count = 0;
                    foreach (XmlNode N in Result.ChildNodes)
                    {
                        if (PeerStatus.SynchCounter != SynchCounter || this.ClientDisposed)
                            return null;
 
                        if (N is XmlElement E2 && E2.LocalName == "ref" && E2.NamespaceURI == NeuroLedgerNamespace)
                        {
                            LastReported = await this.UpdateStatus(LastReported, Status, Callback);
 
                            string Id = XML.Attribute(E2, "id");
                            BlockAdded Event = (BlockAdded)blockAddedRef.Parse(E2);
 
                            BlockReference Ref = await this.GetReference(Event, Status, !Repopulate);
                            if (!(Ref is null))
                            {
                                Collections[Ref.Collection] = true;
 
                                Bulk.Add(new KeyValuePair<BlockReference, string>(Ref, Event.Url));
                                BulkSize += Ref.Bytes;
 
                                if (Bulk.Count >= 100 || BulkSize >= 16 * 1024 * 1024)
                                {
                                    await this.BulkLoad(PeerStatus, Bulk.ToArray(), LastReported, Status, Callback, false, Blocks);
                                    Bulk.Clear();
                                    BulkSize = 0;
                                }
                            }
 
                            Count++;
                            LastId = Id;
                        }
                    }
 
                    More = Count >= MaxCount;
                }
                while (More && !this.ClientDisposed);
 
                if (Bulk.Count > 0)
                    await this.BulkLoad(PeerStatus, Bulk.ToArray(), LastReported, Status, Callback, false, Blocks);
 
                if (Blocks.Count > 0)
                {
                    Blocks.Reverse();
 
                    List<BlockReference> Subset = new List<BlockReference>();
                    List<ObjectState> ObjectsInBlock = new List<ObjectState>();
                    bool Cleared;
 
                    foreach (string Collection in Collections.Keys)
                    {
                        foreach (BlockReference Ref in Blocks)
                        {
                            if (Ref.Collection != Collection)
                                continue;
 
                            if (Ref.Status != BlockStatus.Valid || Ref.AccessDenied)
                                continue;
 
                            Subset.Add(Ref);
                        }
 
                        if (Subset.Count == 0)
                            continue;
 
                        IPersistentDictionary PersistentDictionary = await Database.GetDictionary(Collection);
                        await PersistentDictionary.ClearAsync();
 
                        CachedStringDictionary ObjectEvents = new CachedStringDictionary(100000, PersistentDictionary);
                        try
                        {
                            await Database.StartBulk();
                            try
                            {
                                Cleared = false;
                                Count = 0;
 
                                foreach (BlockReference Ref in Subset)
                                {
                                    using (BlockEnumerator TempBlockEnumerator = new BlockEnumerator(Ref, this.provider))
                                    {
                                        using (ObjectEnumerator<GenericObject> e = await ObjectEnumerator<GenericObject>.Create(TempBlockEnumerator, this.provider))
                                        {
                                            while (await e.MoveNextAsync())
                                                ObjectsInBlock.Add(new ObjectState(e.CurrentEntry.Type, e.Current));
                                        }
 
                                        ObjectsInBlock.Reverse();
 
                                        foreach (ObjectState ObjectState in ObjectsInBlock)
                                        {
                                            Status.Events++;
                                            LastReported = await this.UpdateStatus(LastReported, Status, Callback);
 
                                            if (ObjectState.Type == EntryType.Clear)
                                            {
                                                Cleared = true;
                                                break;
                                            }
 
                                            try
                                            {
                                                string Key = ObjectState.Object.ObjectId.ToString();
 
                                                if (await ObjectEvents.ContainsKeyAsync(Key))
                                                    continue;   // Only latest is of importance.
 
                                                await ObjectEvents.AddAsync(Key, ObjectState);
                                            }
                                            catch (Exception ex)
                                            {
                                                Log.Error("Unable to enumerate objects in block properly when repairing collection:\r\n\r\n" +
                                                    ex.Message, TempBlockEnumerator.Current.FileName, string.Empty, string.Empty,
                                                    EventLevel.Major, string.Empty, string.Empty, Log.CleanStackTrace(ex.StackTrace),
                                                    new KeyValuePair<string, object>("Collection", Collection));
 
                                                break;
                                            }
 
                                            if (++Count >= 100)
                                            {
                                                await Database.EndBulk();
                                                await Database.StartBulk();
                                                Count = 0;
                                            }
                                        }
 
                                        ObjectsInBlock.Clear();
                                    }
 
                                    if (Cleared)
                                        break;
                                }
 
                                if (Cleared)
                                {
                                    await Database.Provider.Clear(Collection);
                                    this.externalEvents?.RaiseCollectionCleared(Collection);
                                }
                            }
                            finally
                            {
                                await Database.EndBulk();
                            }
 
                            Tuple<uint, uint, uint, uint> Counts = await this.provider.Process(ObjectEvents, Collection);
 
                            Status.ObjectsAdded += Counts.Item1;
                            Status.ObjectsUpdated += Counts.Item2;
                            Status.ObjectsDeleted += Counts.Item3;
                            Status.ObjectErrors += Counts.Item4;
 
                            LastReported = await this.UpdateStatus(LastReported, Status, Callback);
                        }
                        finally
                        {
                            await ObjectEvents.ClearAsync();
                            ObjectEvents.DeleteAndDispose();
 
                            Subset.Clear();
                        }
                    }
 
                    NodeStatus.LastBlockId = LastId;
                    await Database.Update(NodeStatus);
                }
 
                foreach (BlockReference Ref in await Database.Find<BlockReference>(new FilterAnd(
                    new FilterFieldEqualTo("FileName", string.Empty),
                    new FilterFieldEqualTo("Creator", NodeStatus.BareJid),
                    new FilterFieldEqualTo("AccessDenied", false))))
                {
                    if (Ref.Sources is null)
                        continue;
 
                    foreach (string Source in Ref.Sources)
                    {
                        if (PeerStatus.SynchCounter != SynchCounter || this.ClientDisposed)
                            return null;
 
                        try
                        {
                            Uri Uri = new Uri(Source);
                            if (Uri.Authority == NodeStatus.BareJid)
                            {
                                LastReported = await this.UpdateStatus(LastReported, Status, Callback);
                                await this.RetrieveBlock(Ref, Source, Status, true);
                                break;
                            }
                        }
                        catch (Exception ex)
                        {
                            Log.Exception(ex);
                            Status.IncBlockError(NodeStatus.BareJid, ex.Message + " (1)");
                        }
                    }
                }
 
                foreach (BlockReference Ref in await Database.Find<BlockReference>(new FilterAnd(
                    new FilterFieldEqualTo("Creator", NodeStatus.BareJid),
                    new FilterFieldEqualTo("AccessDenied", false),
                    new FilterFieldEqualTo("Unpacked", false))))
                {
                    if (Ref.Sources is null)
                        continue;
 
                    foreach (string Source in Ref.Sources)
                    {
                        if (PeerStatus.SynchCounter != SynchCounter || this.ClientDisposed)
                            return null;
 
                        try
                        {
                            Uri Uri = new Uri(Source);
                            if (Uri.Authority == NodeStatus.BareJid)
                            {
                                LastReported = await this.UpdateStatus(LastReported, Status, Callback);
                                await this.RetrieveBlock(Ref, Source, Status, true);
                                break;
                            }
                        }
                        catch (Exception ex)
                        {
                            Log.Exception(ex);
                            Status.IncBlockError(NodeStatus.BareJid, ex.Message + " (2)");
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                Log.Exception(ex);
                Status.IncBlockError(NodeStatus.BareJid, ex.Message + " (3)");
            }
            finally
            {
                if (SynchCounter == PeerStatus.SynchCounter)
                {
                    lock (PeerStatus)
                    {
                        PeerStatus.IsSynchronizing = false;
                    }
 
                    Status.Done = true;
                    await this.UpdateStatus(DateTime.MinValue, Status, Callback);
                }
            }
 
            return string.Empty;
        }
 
        private async Task BulkLoad(PeerStatus PeerStatus, KeyValuePair<BlockReference, string>[] Resources,
            DateTime LastReported, SynchronizationStatus Status, CallbackAsync<SynchronizationStatus> Callback,
            bool Unpack, List<BlockReference> Blocks)
        {
            string Msg;
            int Processed = 0;
 
            try
            {
                StringBuilder Csv = new StringBuilder();
 
                foreach (KeyValuePair<BlockReference, string> Item in Resources)
                {
                    this.client.Information("Retrieving block (" + Item.Key.Collection + "): " + Item.Value);
 
                    Csv.Append(Item.Value);
                    Csv.AppendLine(",application/octet-stream");
                }
 
                Uri Uri = new Uri("httpx://" + PeerStatus.BareJid + "/NL/MG");
                ContentBinaryResponse Response = await InternetContent.PostAsync(Uri,
                    Encoding.UTF8.GetBytes(Csv.ToString()), "text/csv; charset=utf-8",
                    new KeyValuePair<string, string>("Accept", "multipart/mixed"));
 
                Response.AssertOk();
                ContentResponse ContentResponse = await InternetContent.DecodeAsync(Response.ContentType, Response.Encoded, Uri);
                ContentResponse.AssertOk();
 
                if (ContentResponse.Decoded is MixedContent Result)
                {
                    int i = 0;
                    int c = Resources.Length;
 
                    foreach (EmbeddedContent Content in Result.Content)
                    {
                        if (i >= c)
                            break;
 
                        LastReported = await this.UpdateStatus(LastReported, Status, Callback);
                        Processed++;
 
                        KeyValuePair<BlockReference, string> P = Resources[i++];
                        BlockReference Ref = P.Key;
                        string Url = P.Value;
 
                        if (!int.TryParse(Content.Description, out int StatusCode))
                        {
                            Status?.IncBlockError(Url, "Invalid content description: " + Content.Description);
                            continue;
                        }
 
                        if (StatusCode == 200)
                        {
                            byte[] Block = Content.TransferDecoded;
                            this.client.Information("Block retrieved (" + Ref.Collection + "): " + Url);
 
                            if (this.provider is null)
                            {
                                Log.Warning("Bulk loading of blocks aborted. Neuro-Ledger closed.");
                                return;
                            }
 
                            using (MemoryStream ms = new MemoryStream(Block))
                            {
                                await this.provider.AddBlockFile(ms, P.Key);
                            }
 
                            this.client.Information("Block added (" + Ref.Collection + "): " + Url);
 
                            if (Ref.AccessDenied)
                            {
                                Ref.AccessDenied = false;
                                await Database.Update(Ref);
                            }
 
                            if (!(Status is null))
                            {
                                Status.NrLoaded++;
                                Status.LoadedBytes += Ref.Bytes;
                            }
 
                            if (Unpack)
                                await this.UnpackObjects(Ref, Status);
 
                            Blocks.Add(Ref);
                        }
                        else if (StatusCode == HTTP.ForbiddenException.Code ||
                            StatusCode == HTTP.FailedDependencyException.Code)
                        {
                            this.client.Information("Access to block denied (" + Ref.Collection + "): " + Url + ". Cause: " + Content.Decoded?.ToString());
 
                            if (!Ref.AccessDenied)
                            {
                                Ref.AccessDenied = true;
                                await Database.Update(Ref);
                            }
 
                            if (!(Status is null))
                                Status.NrDenied++;
                        }
                        else if (StatusCode == HTTP.NotFoundException.Code)
                        {
                            this.client.Warning(Msg = "Block not found (" + Ref.Collection + "): " + Url + ". Cause: " + Content.Decoded?.ToString());
 
                            if (Ref.Creator == this.provider.ExternalIdentity &&
                                Content.Decoded is string ErrorMsg &&
                                ErrorMsg == BlockResource.ErrorMsg_BlockFileHasBeenRemoved)
                            {
                                await Database.Delete(Ref);
 
                                if (!(Status is null))
                                    Status.NrBlocks--;
                            }
                            else
                                Status?.IncBlockError(Url, Msg);
                        }
                        else if (!this.ClientDisposed && !this.client.Disposed)
                        {
                            this.client.Warning(Msg = "Unable to retrieve block. Trying again later. Error reported: " + StatusCode.ToString() + ", " + Content.Decoded?.ToString());
                            this.scheduler?.Add(DateTime.Now.AddHours(1), this.RetryProcessing, new object[] { Ref, Url });
 
                            Status?.IncBlockError(Url, Msg);
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                this.client.Error(Msg = "Unable to retrieve blocks. Error reported: " + ex.Message);
                Status?.IncBlockError(PeerStatus.BareJid, Msg, (uint)(Resources.Length - Processed));
            }
        }
 
        private async Task<XmlElement> GetBlockList(string Jid, string LastBlockId, int MaxCount)
        {
            StringBuilder Url = new StringBuilder();
 
            Url.Append("httpx://");
            Url.Append(Jid);
            Url.Append("/NL/L?Max=");
            Url.Append(MaxCount.ToString());
 
            if (!string.IsNullOrEmpty(LastBlockId))
            {
                Url.Append("&Last=");
                Url.Append(LastBlockId);
            }
            
            using ContentStreamResponse Content = await InternetContent.GetTempStreamAsync(new Uri(Url.ToString()), DefaultTimeoutMs,
                new KeyValuePair<string, string>("Accept", XmlCodec.DefaultContentType));
            Content.AssertOk();
 
            XmlDocument Doc = new XmlDocument();
 
            Content.Encoded.Position = 0;
            Doc.Load(Content.Encoded);
 
            return Doc.DocumentElement;
        }
 
        private async Task<DateTime> UpdateStatus(DateTime LastReported, SynchronizationStatus Status, CallbackAsync<SynchronizationStatus> Callback)
        {
            DateTime TP = DateTime.Now;
            if ((TP - LastReported).TotalSeconds > 1)
            {
                LastReported = TP;
                if (!(Callback is null))
                {
                    try
                    {
                        await Callback(Status);
                    }
                    catch (Exception ex)
                    {
                        Status?.IncBlockError(string.Empty, ex.Message);
                    }
                }
            }
 
            return LastReported;
        }
 
        private async Task GetBlockReferences(object Sender, IqEventArgs e)
        {
            int Max = XML.Attribute(e.Query, "max", 50);
 
            if (Max <= 0)
            {
                await e.IqError(new BadRequestException("Maximum number of records must be positive.", e.IQ));
                return;
            }
 
            RosterItem Item = (Sender as XmppClient)?.GetRosterItem(e.FromBareJid);
            if (Item is null || (Item.State != SubscriptionState.Both && Item.State != SubscriptionState.From))
            {
                await e.IqError(new ForbiddenException("You must have an active presence subscription to the node to request information.", e.IQ));
                return;
            }
 
            if (Max > 50)
                Max = 50;
 
            IEnumerable<BlockReference> Blocks;
            FilterCustom<BlockReference> CollectionFilter = new FilterCustom<BlockReference>((Ref) => Item.IsInGroup(Ref.Collection));
            string Last = XML.Attribute(e.Query, "last");
            int i;
 
            if (string.IsNullOrEmpty(Last))
                Blocks = await Database.Find<BlockReference>(0, Max + 1, CollectionFilter, "ObjectId");
            else
            {
                Blocks = await Database.Find<BlockReference>(0, Max + 1,
                    new FilterAnd(new FilterFieldGreaterThan("ObjectId", Last), CollectionFilter), "ObjectId");
            }
 
            i = 0;
            foreach (BlockReference Ref in Blocks)
                i++;
 
            StringBuilder Xml = new StringBuilder();
 
            Xml.Append("<blockReferences xmlns=\"");
            Xml.Append(NeuroLedgerNamespace);
            Xml.Append('"');
 
            if (i > Max)
                Xml.Append(" more=\"true\"");
 
            if (i > 0)
            {
                Xml.Append('>');
 
                foreach (BlockReference Ref in Blocks)
                {
                    if (--Max < 0)
                        break;
 
                    Xml.Append("<ref id='");
                    Xml.Append(Ref.ObjectId);
 
                    Xml.Append("' d='");
                    Xml.Append(Convert.ToBase64String(Ref.Digest));
                    Xml.Append("' s='");
                    Xml.Append(Convert.ToBase64String(Ref.Signature));
 
                    if (!(Ref.Link is null))
                    {
                        Xml.Append("' l='");
                        Xml.Append(Convert.ToBase64String(Ref.Link));
                    }
 
                    Xml.Append("' cn='");
                    Xml.Append(XML.Encode(Ref.Collection));
                    Xml.Append("' cr='");
                    Xml.Append(XML.Encode(Ref.Creator));
                    Xml.Append("' ct='");
                    Xml.Append(XML.Encode(Ref.Created));
 
                    if (Ref.Updated != DateTime.MinValue)
                    {
                        Xml.Append("' u='");
                        Xml.Append(XML.Encode(Ref.Updated));
                    }
 
                    if (Ref.Expires != DateTime.MaxValue)
                    {
                        Xml.Append("' x='");
                        Xml.Append(XML.Encode(Ref.Expires));
                    }
 
                    if (Ref.Status != BlockStatus.Valid)
                    {
                        Xml.Append("' t='");
                        Xml.Append(Ref.Status.ToString());
                    }
 
                    Xml.Append("' r='");
                    Xml.Append("httpx://");
                    Xml.Append(this.client.BareJID);
                    Xml.Append("/NL/B/");
                    Xml.Append(Base64Url.Encode(Ref.Digest));
                    Xml.Append("' b='");
                    Xml.Append(Ref.Bytes.ToString());
                    Xml.Append("'/>");
                }
 
                Xml.Append("</blockReferences>");
            }
            else
                Xml.Append("/>");
 
            await e.IqResult(Xml.ToString());
        }
 
        private Task Client_OnStateChanged(object Sender, XmppState NewState)
        {
            if (NewState == XmppState.Connected)
                Task.Run(() => this.CheckOutgoingEvents());
 
            return Task.CompletedTask;
        }
 
        private async Task CheckOutgoingEvents()
        {
            try
            {
                IEnumerable<OutgoingEvent> Events;
 
                try
                {
                    Events = await Database.Find<OutgoingEvent>();
                }
                catch (Exception ex)
                {
                    Log.Exception(ex);
                    await Database.Clear("OutgoingPEP");
                    return;
                }
 
                foreach (OutgoingEvent Event in Events)
                {
                    if (this.pepClient is null)
                        break;
 
                    await this.pepClient.PublishAsync(Event.Event);
                    await Database.Delete(Event);
                }
            }
            catch (Exception ex)
            {
                Log.Exception(ex);
            }
        }
 
        public override string[] Extensions => new string[] { };
 
        public override void Dispose()
        {
            if (!(this.provider is null))
            {
                this.webServer.Unregister(this.blockResource);
                this.webServer.Unregister(this.blockListResource);
                this.webServer.Unregister(this.multiGetResource);
 
                this.provider.BlockAdded -= this.BlockAdded;
                this.provider.BlockDeleted -= this.BlockDeleted;
 
                this.client.OnStateChanged -= this.Client_OnStateChanged;
                this.client.OnPresence -= this.Client_OnPresence;
                this.client.UnregisterIqGetHandler("getBlockReferences", NeuroLedgerNamespace, this.GetBlockReferences, true);
 
                this.pepClient.UnregisterHandler(typeof(BlockAdded), this.BlockAddedReceived);
                this.pepClient.UnregisterHandler(typeof(BlockDeleted), this.BlockDeletedReceived);
 
                this.provider.BlockAdded -= this.BlockAdded;
                this.provider.BlockDeleted -= this.BlockDeleted;
 
                if (this.internalScheduler)
                    this.scheduler?.Dispose();
 
                this.scheduler = null;
 
                this.peerStatus?.Dispose();
                this.peerStatus = null;
 
                this.provider = null;
                this.pepClient = null;
            }
    
            base.Dispose();
        }
 
        private async Task Queue(NeuroLedgerPepEvent Event)
        {
            try
            {
                await Database.Insert(new OutgoingEvent()
                {
                    Event = Event
                });
            }
            catch (Exception ex)
            {
                Log.Exception(ex);
            }
        }
 
        private Task BlockAdded(object Sender, BlockReferenceEventArgs e)
        {
            BlockAdded Event = new BlockAdded();
            Task T;
 
            this.Fill(Event, e);
 
            if (this.pepClient.Client.State == XmppState.Connected)
            {
                this.pepClient.Publish(Event, (sender, e2) =>
                {
                    if (!e2.Ok)
                        T = this.Queue(Event);
 
                    return Task.CompletedTask;
 
                }, null);
            }
            else
                T = this.Queue(Event);
 
            return Task.CompletedTask;
        }
 
        private void Fill(BlockEvent Event, BlockReferenceEventArgs e)
        {
            BlockReference Block = e.Block;
 
            Event.Collection = Block.Collection;
            Event.Created = Block.Created;
            Event.Creator = Block.Creator;
            Event.Digest = Block.Digest;
            Event.Expires = Block.Expires;
            Event.Link = Block.Link;
            Event.Signature = Block.Signature;
            Event.Status = Block.Status;
            Event.Updated = Block.Updated;
            Event.Bytes = Block.Bytes;
            Event.Url = "httpx://" + this.client.BareJID + "/NL/B/" + Base64Url.Encode(Block.Digest);
        }
 
        private Task BlockDeleted(object Sender, BlockReferenceEventArgs e)
        {
            BlockDeleted Event = new BlockDeleted();
            Task T;
 
            this.Fill(Event, e);
 
            if (this.pepClient.Client.State == XmppState.Connected)
            {
                this.pepClient.Publish(Event, (sender, e2) =>
                {
                    if (!e2.Ok)
                        T = this.Queue(Event);
 
                    return Task.CompletedTask;
 
                }, null);
            }
            else
                T = this.Queue(Event);
 
            return Task.CompletedTask;
        }
 
        private async Task BlockAddedReceived(object Sender, PersonalEventNotificationEventArgs e)
        {
            if (e.PersonalEvent is BlockAdded Event)
            {
                lock (this.processingQueue)
                {
                    if (this.processing)
                    {
                        this.processingQueue.AddLast(Event);
                        return;
                    }
 
                    this.processing = true;
                }
 
                do
                {
                    try
                    {
                        await this.Process(Event, null);
                    }
                    catch (ForbiddenException)
                    {
                        // Node not authorized to access blocks from the corresponding collection. Ignore error.
                    }
                    catch (Exception ex)
                    {
                        Log.Exception(ex);
                    }
 
                    lock (this.processingQueue)
                    {
                        if (this.processingQueue.First is null)
                        {
                            Event = null;
                            this.processing = false;
                        }
                        else
                        {
                            Event = this.processingQueue.First.Value;
                            this.processingQueue.RemoveFirst();
                        }
                    }
                }
                while (!(Event is null));
            }
        }
 
        private readonly LinkedList<BlockAdded> processingQueue = new LinkedList<BlockAdded>();
        private bool processing = false;
 
        private async Task Process(BlockAdded Event, SynchronizationStatus Status)
        {
            BlockReference Ref = await this.GetReference(Event, Status, true);
            if (!(Ref is null))
                await this.RetrieveBlock(Ref, Event.Url, Status, true);
        }
 
        private async Task<BlockReference> GetReference(BlockAdded Event, SynchronizationStatus Status, bool NullIfNotChanged)
        {
            if (!(Status is null))
            {
                Status.NrBlocks++;
                Status.TotalBytes += Event.Bytes;
 
                if (Event.Created < Status.First)
                    Status.First = Event.Created;
 
                if (Event.Created > Status.Last)
                    Status.Last = Event.Created;
            }
 
            BlockReference Ref = await NeuroLedgerProvider.FindReference(Event.Digest);
 
            if (Ref is null)
            {
                Ref = new BlockReference()
                {
                    Collection = Event.Collection,
                    Created = Event.Created,
                    Creator = Event.Creator,
                    Digest = Event.Digest,
                    Expires = Event.Expires,
                    Link = Event.Link,
                    Signature = Event.Signature,
                    Status = Event.Status,
                    Updated = Event.Updated,
                    Bytes = Event.Bytes,
                    Sources = new string[] { Event.Url },
                    AccessDenied = false,
                    Unpacked = false
                };
 
                await Database.Insert(Ref);
 
                if (!(Status is null))
                    Status.NrNew++;
 
                return Ref;
            }
            else
            {
                bool Updated = false;
                bool Changed = false;
 
                if (Ref.Creator != this.provider?.ExternalIdentity &&
                    (Ref.Collection != Event.Collection ||
                    !NeuroLedgerProvider.AreEqual(Ref.Created, Event.Created) ||
                    !NeuroLedgerProvider.AreEqual(Ref.Updated, Event.Updated) ||
                    !NeuroLedgerProvider.AreEqual(Ref.Expires, Event.Expires) ||
                    Ref.Creator != Event.Creator ||
                    !Compare(Ref.Digest, Event.Digest) ||
                    !Compare(Ref.Link, Event.Link) ||
                    !Compare(Ref.Signature, Event.Signature) ||
                    Ref.Status != Event.Status ||
                    Ref.Bytes != Event.Bytes ||
                    Ref.AccessDenied))
                {
                    Ref.Collection = Event.Collection;
                    Ref.Created = Event.Created;
                    Ref.Creator = Event.Creator;
                    Ref.Expires = Event.Expires;
                    Ref.Digest = Event.Digest;
                    Ref.Link = Event.Link;
                    Ref.Signature = Event.Signature;
                    Ref.Status = Event.Status;
                    Ref.Updated = Event.Updated;
                    Ref.Bytes = Event.Bytes;
                    Ref.AccessDenied = false;
 
                    Changed = true;
                }
 
                if (Ref.Sources is null || Array.IndexOf(Ref.Sources, Event.Url) < 0)
                {
                    int c = Ref.Sources?.Length ?? 0;
 
                    string[] s = new string[c + 1];
                    if (c > 0)
                        Array.Copy(Ref.Sources, 0, s, 0, c);
 
                    s[c] = Event.Url;
                    Ref.Sources = s;
                    Updated = true;
                }
 
                if (Updated || Changed)
                {
                    await Database.Update(Ref);
 
                    if (!(Status is null))
                        Status.NrUpdated++;
 
                    if (Changed || !NullIfNotChanged)
                        return Ref;
                    else
                        return null;
                }
                else if (string.IsNullOrEmpty(Ref.FileName) || !NullIfNotChanged)
                    return Ref;
                else
                    return null;
            }
        }
 
        private static bool Compare(byte[] A1, byte[] A2)
        {
            if ((A1 is null) ^ (A2 is null))
                return false;
 
            if (A1 is null)
                return true;
 
            int i, c = A1.Length;
            if (A2.Length != c)
                return false;
 
            for (i = 0; i < c; i++)
            {
                if (A1[i] != A2[i])
                    return false;
            }
 
            return true;
        }
 
        internal async Task<bool> RetrieveBlock(BlockReference Ref, string Url, SynchronizationStatus Status, bool Unpack)
        {
            string FileName = this.provider?.GetFullFileName(Ref.FileName);
 
            if (string.IsNullOrEmpty(Ref.FileName) || !File.Exists(FileName))
            {
                string Key = Convert.ToBase64String(Ref.Digest);
 
                lock (this.loading)
                {
                    if (this.loading.ContainsKey(Key))
                        return true;
 
                    this.loading[Key] = true;
                }
 
                try
                {
                    this.client.Information("Retrieving block (" + Ref.Collection + "): " + Url);
 
                    using ContentStreamResponse Content = await InternetContent.GetTempStreamAsync(new Uri(Url), DefaultTimeoutMs,
                        new KeyValuePair<string, string>("Accept", "application/octet-stream"));
                    Content.AssertOk();
 
                    this.client.Information("Block retrieved (" + Ref.Collection + "): " + Url);
 
                    await this.provider.AddBlockFile(Content.Encoded, Ref);
 
                    this.client.Information("Block added (" + Ref.Collection + "): " + Url);
 
                    if (Ref.AccessDenied)
                    {
                        Ref.AccessDenied = false;
                        await Database.Update(Ref);
                    }
 
                    if (!(Status is null))
                    {
                        Status.NrLoaded++;
                        Status.LoadedBytes += Ref.Bytes;
                    }
 
                    if (Unpack)
                        await this.UnpackObjects(Ref, Status);
 
                    return true;
                }
                catch (HTTP.ForbiddenException ex)
                {
                    string ErrorMessage = await this.GetErrorMessage(ex);
 
                    this.client.Information("Access to block denied (" + Ref.Collection + "): " + Url + ". Cause: " + ErrorMessage);
 
                    if (!Ref.AccessDenied)
                    {
                        Ref.AccessDenied = true;
                        await Database.Update(Ref);
                    }
 
                    if (!(Status is null))
                        Status.NrDenied++;
 
                    return false;
                }
                catch (HTTP.FailedDependencyException ex)
                {
                    string ErrorMessage = await this.GetErrorMessage(ex);
 
                    this.client.Information("Access to block denied (" + Ref.Collection + "): " + Url + ". Cause: " + ErrorMessage);
 
                    if (!Ref.AccessDenied)
                    {
                        Ref.AccessDenied = true;
                        await Database.Update(Ref);
                    }
 
                    if (!(Status is null))
                        Status.NrDenied++;
 
                    return false;
                }
                catch (HTTP.NotFoundException ex)
                {
                    string ErrorMessage = await this.GetErrorMessage(ex);
 
                    if (ErrorMessage == BlockResource.ErrorMsg_BlockFileHasBeenRemoved)
                    {
                        this.client.Warning("Block not found (" + Ref.Collection + "): " + Url + ". Cause: " + ErrorMessage);
 
                        await Database.Delete(Ref);
 
                        if (!(Status is null))
                            Status.NrBlocks--;
                    }
 
                    return false;
                }
                catch (Exception ex)
                {
                    this.client.Warning("Unable to retrieve block. Trying again later. Error reported: " + ex.Message + " (" + ex.GetType().FullName + ")");
                    this.scheduler?.Add(DateTime.Now.AddHours(1), this.RetryProcessing, new object[] { Ref, Url });
 
                    Status?.IncBlockError(Url, ex.Message);
 
                    return false;
                }
                finally
                {
                    lock (this.loading)
                    {
                        this.loading.Remove(Key);
                    }
                }
            }
            else
                return true;
        }
 
        private async Task<string> GetErrorMessage(HTTP.HttpException ex)
        {
            object ContentObject = await ex.GetContentObjectAsync();
            if (ContentObject is string s)
                return s;
            else if (!(ex.Content is null))
                return Encoding.UTF8.GetString(ex.Content);
            else
                return ex.Message;
        }
 
        private async Task UnpackObjects(BlockReference Ref, SynchronizationStatus Status)
        {
            using (BlockEnumerator TempBlockEnumerator = new BlockEnumerator(Ref, this.provider))
            {
                using ObjectEnumerator<GenericObject> e = await ObjectEnumerator<GenericObject>.Create(TempBlockEnumerator, this.provider);
                Dictionary<string, bool> UnpackType = new Dictionary<string, bool>();
                int NrAdded = 0;
                int NrUpdated = 0;
                int NrDeleted = 0;
                int NrErrors = 0;
                int NrIgnored = 0;
 
                await Database.StartBulk();
                try
                {
                    while (await e.MoveNextAsync())
                    {
                        GenericObject Obj = e.Current;
 
                        if (!UnpackType.TryGetValue(Obj.TypeName, out bool UnpackObject))
                        {
                            Type T = Types.GetType(Obj.TypeName);
                            UnpackObject = (T is null) || !(T.GetCustomAttribute(typeof(ArchivingTimeAttribute)) is null);
                            UnpackType[Obj.TypeName] = UnpackObject;
                        }
 
                        if (!UnpackObject)
                        {
                            NrIgnored++;
                            continue;
                        }
 
                        switch (e.CurrentEntry.Type)
                        {
                            case EntryType.New:
                                try
                                {
                                    await Database.Provider.Insert(Obj);    // Calling Provider method avoids raising event.
                                    NrAdded++;
 
                                    this.externalEvents?.RaiseEntryAdded(Obj);
 
                                    if (!(Status is null))
                                        Status.ObjectsAdded++;
                                }
                                catch (KeyAlreadyExistsException)
                                {
                                    // Object already exists, perhaps in a newer state. Keep that version for now.
                                }
                                catch (Exception ex)
                                {
                                    this.client.Error(ex.Message);
                                    NrErrors++; // Object already exists locally.
 
                                    if (!(Status is null))
                                        Status.ObjectErrors++;
                                }
                                break;
 
                            case EntryType.Update:
                                try
                                {
                                    await Database.Provider.Update(Obj);    // Calling Provider method avoids raising event.
                                    NrUpdated++;
 
                                    this.externalEvents?.RaiseEntryUpdated(Obj);
 
                                    if (!(Status is null))
                                        Status.ObjectsUpdated++;
                                }
                                catch (KeyNotFoundException)
                                {
                                    try
                                    {
                                        await Database.Provider.Insert(Obj);    // Calling Provider method avoids raising event.
                                        NrAdded++;
 
                                        this.externalEvents?.RaiseEntryAdded(Obj);
 
                                        if (!(Status is null))
                                            Status.ObjectsAdded++;
                                    }
                                    catch (Exception ex)
                                    {
                                        this.client.Error(ex.Message);
                                        NrErrors++;
 
                                        if (!(Status is null))
                                            Status.ObjectErrors++;
                                    }
                                }
                                catch (Exception ex)
                                {
                                    this.client.Error(ex.Message);
                                    NrErrors++;
 
                                    if (!(Status is null))
                                        Status.ObjectErrors++;
                                }
                                break;
 
                            case EntryType.Delete:
                                try
                                {
                                    await Database.Provider.Delete(Obj);    // Calling Provider method avoids raising event.
                                    NrDeleted++;
 
                                    this.externalEvents?.RaiseEntryDeleted(Obj);
 
                                    if (!(Status is null))
                                        Status.ObjectsDeleted++;
                                }
                                catch (KeyNotFoundException)
                                {
                                    // Already deleted.
                                }
                                catch (Exception ex)
                                {
                                    this.client.Error(ex.Message);
                                    NrErrors++; // Object does not exist locally.
 
                                    if (!(Status is null))
                                        Status.ObjectErrors++;
                                }
                                break;
 
                            case EntryType.Clear:
                                //await Database.Provider.Clear(Ref.Collection);  // Calling Provider method avoids raising event.
                                //this.client.Information("Collection cleared: " + Ref.Collection);
                                //
                                //NrAdded = 0;
                                //NrUpdated = 0;
                                //NrDeleted = 0;
                                //NrErrors = 0;
 
                                // TODO: Clear only objects from corresponding collection & creator. Also raise corresponding events.
                                continue;
 
                            default:
                                continue;
                        }
                    }
                }
                finally
                {
                    await Database.EndBulk();
                }
 
                if (NrAdded > 0)
                    this.client.Information("Number of objects added to " + Ref.Collection + ": " + NrAdded.ToString());
 
                if (NrUpdated > 0)
                    this.client.Information("Number of objects updated in " + Ref.Collection + ": " + NrUpdated.ToString());
 
                if (NrDeleted > 0)
                    this.client.Information("Number of objects deleted from " + Ref.Collection + ": " + NrDeleted.ToString());
 
                if (NrErrors > 0)
                    this.client.Information("Number of errors related to " + Ref.Collection + ": " + NrErrors.ToString());
 
                if (NrIgnored > 0)
                    this.client.Information("Number of objects ignored in " + Ref.Collection + ": " + NrIgnored.ToString());
            }
 
            if (!Ref.Unpacked)
            {
                Ref.Unpacked = true;
                await Database.Update(Ref);
            }
        }
 
        private async void RetryProcessing(object State)
        {
            try
            {
                if (this.ClientDisposed)
                    return;
 
                object[] P = (object[])State;
                BlockReference Ref = (BlockReference)P[0];
                string Url = (string)P[1];
 
                RosterItem Item = this.client[Ref.Creator];
                if (Item is null)
                    return;
 
                foreach (PresenceEventArgs e in Item.Resources)
                {
                    PeerStatus PeerStatus = await this.GetPeerStatus(e);
                    if (!(PeerStatus is null) && PeerStatus.IsNeuroLedger)
                    {
                        await this.RetrieveBlock(Ref, Url, null, true);
                        return;
                    }
                }
            }
            catch (Exception ex)
            {
                Log.Exception(ex);
            }
        }
 
        private Task BlockDeletedReceived(object Sender, PersonalEventNotificationEventArgs e)
        {
            if (e.PersonalEvent is BlockDeleted)
            {
                // TODO: Delete BlockReference
                // TODO: Delete Objects
            }
 
            return Task.CompletedTask;
        }
 
        public Task ProcessBlockRequest(HTTP.HeaderFields.HttpFieldAccept Accept, HTTP.HttpResponse Response, BlockReference Ref)
        {
            return this.blockResource.ProcessRequest(Accept, Response, Ref);
        }
 
    }
}