public override void OnPut(string key, RavenJObject document, RavenJObject metadata, TransactionInformation transactionInformation)
		{
			if (key.StartsWith("Raven/", StringComparison.OrdinalIgnoreCase) && // we don't deal with system documents
				key.StartsWith("Raven/Hilo/", StringComparison.OrdinalIgnoreCase) == false) // except for hilos
				return;
			using (Database.DisableAllTriggersForCurrentThread())
			{
				var documentMetadata = GetDocumentMetadata(key);
				if (documentMetadata != null)
				{
					RavenJArray history = new RavenJArray(ReplicationData.GetHistory(documentMetadata));
					metadata[Constants.RavenReplicationHistory] = history;

					if (documentMetadata.ContainsKey(Constants.RavenReplicationVersion) && 
						documentMetadata.ContainsKey(Constants.RavenReplicationSource))
					{
						history.Add(new RavenJObject
						{
							{Constants.RavenReplicationVersion, documentMetadata[Constants.RavenReplicationVersion]},
							{Constants.RavenReplicationSource, documentMetadata[Constants.RavenReplicationSource]}
						});
					}

					while (history.Length > Constants.ChangeHistoryLength)
					{
						history.RemoveAt(0);
					}
				}

				metadata[Constants.RavenReplicationVersion] = RavenJToken.FromObject(HiLo.NextId());
				metadata[Constants.RavenReplicationSource] = RavenJToken.FromObject(Database.TransactionalStorage.Id);
			}
		}

		private static void ModifyResult(RavenJObject result)
		{
			var tags = result.Value<string>("Tags");
			if (tags != null)
			{
				result["Tags"] =
					new RavenJArray(tags.Split(new[] {' ', ',', ';'}, StringSplitOptions.RemoveEmptyEntries));
			}
			else
			{
				result["Tags"] = new RavenJArray();
			}
			var deps = result.Value<string>("Dependencies");
			if (deps != null)
			{
				result["Dependencies"] =
					new RavenJArray(deps.Split(new[] {'|'}, StringSplitOptions.RemoveEmptyEntries)
						                .Select(s =>
							                {
								                var strings = s.Split(':');
								                return RavenJObject.FromObject(new {Package = strings[0], Version = strings[1]});
							                }));
			}
			result["PackageId"] = result["Id"];
			result.Remove("Id");
			result.Remove("__metadata");
			result.Remove("DownloadCount");
		}

        protected void RecordActionForDebug(string actionName, string documentKey, RavenJObject documentData, RavenJObject documentMetadata)
        {
            if (debugMode == false)
                return;

            var actions = DebugActions[actionName] as RavenJArray;

            switch (actionName)
            {
                case "LoadDocument":
                case "DeleteDocument":
                    if (actions == null)
                        DebugActions[actionName] = actions = new RavenJArray();

                    actions.Add(documentKey);
                    break;
                case "PutDocument":
                    if (actions == null)
                        DebugActions[actionName] = actions = new RavenJArray();

                    actions.Add(new RavenJObject
                                {
                                    { "Key", documentKey }, 
                                    { "Data", documentData }, 
                                    { "Metadata", documentMetadata }
                                });
                    break;
                default:
                    throw new NotSupportedException("Cannot record action: " + actionName);
            }
        }

        /// <summary>
        /// Adds the cascade delete reference.
        /// </summary>
        /// <param name="session">The session.</param>
        /// <param name="entity">The entity.</param>
        /// <param name="documentKeys">The document keys.</param>
        public static void AddCascadeDeleteReference(this IAdvancedDocumentSessionOperations session, object entity, params string[] documentKeys)
        {
            var metadata = session.GetMetadataFor(entity);

            if(metadata == null)
            {
                throw new InvalidOperationException("The entity must be tracked in the session before calling this method.");
            }

            if(documentKeys.Length == 0)
            {
                throw new ArgumentException("At least one document key must be specified.");
            }

            const string metadataKey = "Raven-Cascade-Delete-Documents";
            RavenJToken token;

            if(!metadata.TryGetValue(metadataKey, out token))
            {
                token = new RavenJArray();
            }

            var list = (RavenJArray)token;
            foreach(var documentKey in documentKeys.Where(key => !list.Contains(key)))
            {
                list.Add(documentKey);
            }

            metadata[metadataKey] = list;
        }

		public override void Respond(IHttpContext context)
		{
			RavenJArray itemsToLoad;
			if(context.Request.HttpMethod == "POST")
				itemsToLoad = context.ReadJsonArray();
			else
				itemsToLoad = new RavenJArray(context.Request.QueryString.GetValues("id"));
			var result = new MultiLoadResult();
			var loadedIds = new HashSet<string>();
			var includes = context.Request.QueryString.GetValues("include") ?? new string[0];
			var transactionInformation = GetRequestTransaction(context);
		    var includedEtags = new List<byte>();
			Database.TransactionalStorage.Batch(actions =>
			{
				foreach (RavenJToken item in itemsToLoad)
				{
					var value = item.Value<string>();
					if(loadedIds.Add(value)==false)
						continue;
					var documentByKey = Database.Get(value, transactionInformation);
					if (documentByKey == null)
						continue;
					result.Results.Add(documentByKey.ToJson());

					if (documentByKey.Etag != null)
					{
						includedEtags.AddRange(documentByKey.Etag.Value.ToByteArray());
					}
					includedEtags.Add((documentByKey.NonAuthoritativeInformation ?? false) ? (byte)0 : (byte)1);
				}

				var addIncludesCommand = new AddIncludesCommand(Database, transactionInformation, (etag, includedDoc) =>
				{
					includedEtags.AddRange(etag.ToByteArray());
					result.Includes.Add(includedDoc);
				}, includes, loadedIds);

				foreach (var item in result.Results.Where(item => item != null))
				{
					addIncludesCommand.Execute(item);
				}
			});

			Guid computedEtag;

			using (var md5 = MD5.Create())
			{
				var computeHash = md5.ComputeHash(includedEtags.ToArray());
				computedEtag = new Guid(computeHash);
			}

			if (context.MatchEtag(computedEtag))
			{
				context.SetStatusToNotModified();
				return;
			}

			context.WriteETag(computedEtag);
			context.WriteJson(result);
		}

		public override void OnPut(string key, Stream data, RavenJObject metadata)
		{
			if (key.StartsWith("Raven/")) // we don't deal with system attachment
				return;
			using (Database.DisableAllTriggersForCurrentThread())
			{
				var attachmentMetadata = GetAttachmentMetadata(key);
				if (attachmentMetadata != null)
				{
					RavenJArray history = new RavenJArray(metadata.Value<RavenJArray>(Constants.RavenReplicationHistory));
					metadata[Constants.RavenReplicationHistory] = history;

					if (attachmentMetadata.ContainsKey(Constants.RavenReplicationVersion) &&
						attachmentMetadata.ContainsKey(Constants.RavenReplicationSource))
					{
						history.Add(new RavenJObject
						{
							{Constants.RavenReplicationVersion, attachmentMetadata[Constants.RavenReplicationVersion]},
							{Constants.RavenReplicationSource, attachmentMetadata[Constants.RavenReplicationSource]}
						});
					}

					if (history.Length > Constants.ChangeHistoryLength)
					{
						history.RemoveAt(0);
					}
				}

				metadata[Constants.RavenReplicationVersion] = RavenJToken.FromObject(HiLo.NextId());
				metadata[Constants.RavenReplicationSource] = RavenJToken.FromObject(Database.TransactionalStorage.Id);
			}
		}

		public ReplicationTopologyDiscoverer(DocumentDatabase database, RavenJArray @from, int ttl, ILog log)
		{
			this.database = database;
			this.ttl = ttl;
			this.log = log;
			[email protected] = @from;
			requestFactory = new HttpRavenRequestFactory();
		}

        private void Initialize()
        {
            if (mappedTypes != null)
                return;

            lock (padlock)
            {
                if (mappedTypes != null)
                    return;

                var collectionData = new CollectionData();

                var jsonDoc = store.DatabaseCommands.Get(collectionNamesDocId);
                if (jsonDoc != null)
                {
                    var collectionNames = jsonDoc.DataAsJson["Collections"] as RavenJArray;
                    foreach (RavenJValue value in collectionNames)
                    {
                        collectionData.Collections.Add(value.Value as string);
                    }
                }

                if (timeoutsEnabled)
                {
                    MapTypeToCollectionName(typeof(TimeoutPersisters.RavenDB.TimeoutData), collectionData);
                }
                if (sagasEnabled)
                {
                    foreach (var sagaType in types.Where(IsSagaEntity))
                    {
                        MapTypeToCollectionName(sagaType, collectionData);
                    }
                }

                if (collectionData.Changed)
                {
                    var newDoc = new RavenJObject();
                    var list = new RavenJArray();
                    foreach (var name in collectionData.Collections)
                    {
                        list.Add(new RavenJValue(name));
                    }
                    newDoc["EndpointName"] = this.endpointName;
                    newDoc["EndpointName"] = this.endpointName;
                    newDoc["Collections"] = list;
                    var metadata = new RavenJObject();
                    store.DatabaseCommands.Put(collectionNamesDocId, null, newDoc, metadata);
                }

                // Completes initialization
                this.mappedTypes = collectionData.Mappings;
            }
        }

        private bool TryHandleArrayValue(int index, Dictionary<string, object> result, KeyValuePair<string, RavenJToken> prop)
        {
            var arrays = new List<RavenJArray>
            {
                (RavenJArray)prop.Value
            };

            for (var i = 0; i < docs.Length; i++)
            {
                if (i == index)
                    continue;

                RavenJToken token;
                if (docs[i].TryGetValue(prop.Key, out token) && token.Type != JTokenType.Array)
                    return false;
                if (token == null)
                    continue;
                if (token.IsSnapshot)
                    token = token.CreateSnapshot();
                arrays.Add((RavenJArray)token);
            }

            var mergedArray = new RavenJArray();
            while (arrays.Count > 0)
            {
                var set = new HashSet<RavenJToken>(RavenJTokenEqualityComparer.Default);
                for (var i = 0; i < arrays.Count; i++)
                {
                    if (arrays[i].Length == 0)
                    {
                        arrays.RemoveAt(i);
                        i -= 1;
                        continue;
                    }
                    set.Add(arrays[i][0]);
                    arrays[i].RemoveAt(0);
                }

                foreach (var ravenJToken in set)
                {
                    mergedArray.Add(ravenJToken);
                }
            }

            if (RavenJTokenEqualityComparer.Default.Equals(mergedArray, prop.Value))
            {
                result.Add(prop.Key, mergedArray);
                return true;
            }

            result.Add(prop.Key, new ArrayWithWarning(mergedArray));
            return true;
        }

		private RavenJArray PerformBulkOperation(string index, IndexQuery indexQuery, bool allowStale, Func<string, TransactionInformation, object> batchOperation)
		{
			var array = new RavenJArray();
			var bulkIndexQuery = new IndexQuery
			{
				Query = indexQuery.Query,
				Start = indexQuery.Start,
				Cutoff = indexQuery.Cutoff,
                WaitForNonStaleResultsAsOfNow = indexQuery.WaitForNonStaleResultsAsOfNow,
				PageSize = int.MaxValue,
				FieldsToFetch = new[] { Constants.DocumentIdFieldName },
				SortedFields = indexQuery.SortedFields,
				HighlighterPreTags = indexQuery.HighlighterPreTags,
				HighlighterPostTags = indexQuery.HighlighterPostTags,
				HighlightedFields = indexQuery.HighlightedFields,
				SortHints = indexQuery.SortHints
			};

			bool stale;
            var queryResults = database.Queries.QueryDocumentIds(index, bulkIndexQuery, tokenSource, out stale);

			if (stale && allowStale == false)
			{
				throw new InvalidOperationException(
						"Bulk operation cancelled because the index is stale and allowStale is false");
			}

		    var token = tokenSource.Token;

			const int batchSize = 1024;
			using (var enumerator = queryResults.GetEnumerator())
			{
				while (true)
				{
					if (timeout != null)
						timeout.Delay();
					var batchCount = 0;
                    token.ThrowIfCancellationRequested();
					database.TransactionalStorage.Batch(actions =>
					{
						while (batchCount < batchSize && enumerator.MoveNext())
						{
							batchCount++;
							var result = batchOperation(enumerator.Current, transactionInformation);
							array.Add(RavenJObject.FromObject(result));
						}
					});
					if (batchCount < batchSize) break;
				}
			}
			return array;
		}

		public override void OnPut(string key, Stream data, RavenJObject metadata)
		{
			using (Database.DisableAllTriggersForCurrentThread())
			{
				metadata.Remove(Constants.RavenReplicationConflict);// you can't put conflicts

				var oldVersion = Database.Attachments.GetStatic(key);
				if (oldVersion == null)
					return;
				if (oldVersion.Metadata[Constants.RavenReplicationConflict] == null)
					return;

				var history = new RavenJArray(ReplicationData.GetHistory(metadata));
				metadata[Constants.RavenReplicationHistory] = history;

				var ravenJTokenEqualityComparer = new RavenJTokenEqualityComparer();
				// this is a conflict document, holding document keys in the 
				// values of the properties
				var conflictData = oldVersion.Data().ToJObject();
				var conflicts = conflictData.Value<RavenJArray>("Conflicts");
				if (conflicts == null)
					return;
				foreach (var prop in conflicts)
				{
					var id = prop.Value<string>();
					Attachment attachment = Database.Attachments.GetStatic(id);
					if(attachment == null)
						continue;
					Database.Attachments.DeleteStatic(id, null);

					// add the conflict history to the mix, so we make sure that we mark that we resolved the conflict
					var conflictHistory = new RavenJArray(ReplicationData.GetHistory(attachment.Metadata));
					conflictHistory.Add(new RavenJObject
					{
						{Constants.RavenReplicationVersion, attachment.Metadata[Constants.RavenReplicationVersion]},
						{Constants.RavenReplicationSource, attachment.Metadata[Constants.RavenReplicationSource]}
					});

					foreach (var item in conflictHistory)
					{
						if (history.Any(x => ravenJTokenEqualityComparer.Equals(x, item)))
							continue;
						history.Add(item);
					}
				}
			}
		}

		public override IEnumerable<Row> Execute(IEnumerable<Row> rows)
		{
			int count = 0;
			foreach (var commentsForPosts in rows.Partition(Constants.BatchSize))
			{
				var cmds = new List<ICommandData>();

				foreach (var commentsForPost in commentsForPosts.GroupBy(row => row["PostId"]))
				{
					var comments = new RavenJArray();
					foreach (var row in commentsForPost)
					{
						comments.Add(new RavenJObject
						{
							{"Score", new RavenJValue(row["Score"])},
							{"CreationDate", new RavenJValue(row["CreationDate"])},
							{"Text", new RavenJValue(row["Text"])},
							{"UserId", new RavenJValue(row["UserId"])}
						});
							
					}
					cmds.Add(new PatchCommandData
					{
						Key = "posts/" + commentsForPost.Key,
						Patches = new[]
						{
							new PatchRequest
							{
								Name = "Comments",
								Type = PatchCommandType.Set,
								Value = comments
							},
						}
					});
				}

				count++;

				WriteCommandsTo("Comments #" + count.ToString("00000") + ".json", cmds);
			}

			yield break;
		}

		public override void OnPut(string key, RavenJObject document, RavenJObject metadata, TransactionInformation transactionInformation)
		{
			using (Database.DisableAllTriggersForCurrentThread())
			{
				metadata.Remove(Constants.RavenReplicationConflict);// you can't put conflicts

				var oldVersion = Database.Get(key, transactionInformation);
				if (oldVersion == null)
					return;
				if (oldVersion.Metadata[Constants.RavenReplicationConflict] == null)
					return;

				RavenJArray history = new RavenJArray(ReplicationData.GetHistory(metadata));
				metadata[Constants.RavenReplicationHistory] = history;

				var ravenJTokenEqualityComparer = new RavenJTokenEqualityComparer();
				// this is a conflict document, holding document keys in the 
				// values of the properties
				var conflicts = oldVersion.DataAsJson.Value<RavenJArray>("Conflicts");
				if(conflicts == null)
					return;
				foreach (var prop in conflicts)
				{
					RavenJObject deletedMetadata;
					Database.Delete(prop.Value<string>(), null, transactionInformation, out deletedMetadata);

					// add the conflict history to the mix, so we make sure that we mark that we resolved the conflict
					var conflictHistory = new RavenJArray(ReplicationData.GetHistory(deletedMetadata));
					conflictHistory.Add(new RavenJObject
					{
						{Constants.RavenReplicationVersion, deletedMetadata[Constants.RavenReplicationVersion]},
						{Constants.RavenReplicationSource, deletedMetadata[Constants.RavenReplicationSource]}
					});

					foreach (var item in conflictHistory)
					{
						if(history.Any(x=>ravenJTokenEqualityComparer.Equals(x, item)))
							continue;
						history.Add(item);
					}
				}
			}
		}

		private static void WriteJsonArray(RavenJArray array, StringWriter sw, int margin, int indent = 0)
		{
			sw.WriteLine('[');
			indent += 1;
			var isFirstItem = true;
			foreach (var token in array.Values())
			{
				if (isFirstItem)
					isFirstItem = false;
				else
					sw.WriteLine(',');
				Indent(sw, indent);
				WriteValue(token, sw, margin, indent);
			}
			sw.WriteLine();
			indent -= 1;
			Indent(sw, indent);
			sw.Write(']');
		}

		private RavenJArray PerformBulkOperation(string index, IndexQuery indexQuery, bool allowStale, Func<string, TransactionInformation, object> batchOperation)
		{
			var array = new RavenJArray();
			var bulkIndexQuery = new IndexQuery
			{
				Query = indexQuery.Query,
				Start = indexQuery.Start,
				Cutoff = indexQuery.Cutoff,
				PageSize = int.MaxValue,
				FieldsToFetch = new[] { Constants.DocumentIdFieldName },
				SortedFields = indexQuery.SortedFields
			};

			bool stale;
			var queryResults = database.QueryDocumentIds(index, bulkIndexQuery, out stale);

			if (stale && allowStale == false)
			{
				throw new InvalidOperationException(
						"Bulk operation cancelled because the index is stale and allowStale is false");
			}

			const int batchSize = 1024;
			using (var enumerator = queryResults.GetEnumerator())
			{
				while (true)
				{
					var batchCount = 0;
					database.TransactionalStorage.Batch(actions =>
					{
						while (batchCount < batchSize && enumerator.MoveNext())
						{
							batchCount++;
							var result = batchOperation(enumerator.Current, transactionInformation);
							array.Add(RavenJObject.FromObject(result, JsonExtensions.CreateDefaultJsonSerializer()));
						}
					});
					if (batchCount < batchSize) break;
				}
			}
			return array;
		}

        private static void WriteJsonArray(RavenJArray array, CountingWriter sw, int width, int numberOfLines)
		{
			sw.WriteLine("[");
			sw.PushIndent();

			var isFirstItem = true;
			foreach (var token in array.Values())
			{
                if (sw.LinesWritten >= numberOfLines)
                    break;

				if (isFirstItem)
					isFirstItem = false;
				else
					sw.WriteLine(",");
                WriteValue(token, sw, width, numberOfLines);
			}
			sw.WriteLine("");
			sw.PopIndent();
			sw.Write("]");
		}

			public void ForEach(RavenJToken result, RavenJToken item, Action<PathPart, RavenJToken, RavenJToken> action)
			{
				if (string.IsNullOrEmpty(FinalName) == false)
				{
					action(this, item, result);
					return;
				}
				RavenJToken newResult = GetTheNewResultOrWireTheDefault(result);
				if (item == null)
				{
					foreach (var pathPart in Items)
						pathPart.Value.ForEach(newResult, null, action);
					return;
				}
				if (item is RavenJArray == false)
				{
					foreach (var pathPart in Items)
						pathPart.Value.ForEach(newResult, item.SelectToken(pathPart.Key), action);
				}
				else
				{
					var jArray = newResult as RavenJArray;
					if (jArray == null)
					{
						jArray = new RavenJArray();
						((RavenJObject)result)[Name] = jArray;
					}
					foreach (var subItem in item.Values<RavenJToken>())
					{
						newResult = new RavenJObject();
						jArray.Add(newResult);
						foreach (var pathPart in Items)
						{
							pathPart.Value.ForEach(newResult, subItem.SelectToken(pathPart.Key), action);
						}
					}
				}
			}

        public async Task<HttpResponseMessage> ReplicationTopologyDiscover()
        {
            var ttlAsString = GetQueryStringValue("ttl");

            int ttl;
            RavenJArray from;

            if (string.IsNullOrEmpty(ttlAsString))
            {
                ttl = 10;
                from = new RavenJArray();
            }
            else
            {
                ttl = int.Parse(ttlAsString);
                from = await ReadJsonArrayAsync();
            }

            var replicationSchemaDiscoverer = new ReplicationTopologyDiscoverer(Database, from, ttl, Log);
            var node = replicationSchemaDiscoverer.Discover();

            return GetMessageWithObject(node);
        }

		public static RavenJToken MinimizeToken(RavenJToken obj, int depth = 0)
		{
			switch (obj.Type)
			{
				case JTokenType.Array:
					var array = new RavenJArray();
					foreach (var item in ((RavenJArray)obj))
					{
						array.Add(MinimizeToken(item, depth + 1));
					}
					return array;
				case JTokenType.Object:
					var ravenJObject = ((RavenJObject)obj);
					if (ravenJObject.ContainsKey(Constants.Metadata) == false)
					{
						// this might be a wrapper object, let check for first level arrays
						if (depth == 0)
						{
							var newRootObj = new RavenJObject();

							foreach (var prop in ravenJObject)
							{
								newRootObj[prop.Key] = prop.Value.Type == JTokenType.Array ?
									MinimizeToken(prop.Value, depth + 1) :
									prop.Value;
							}
							return newRootObj;
						}
						return obj;
					}
					var newObj = new RavenJObject();
					newObj[Constants.Metadata] = ravenJObject[Constants.Metadata];
					return newObj;
				default:
					return obj;
			}
		}

		private RavenJArray GetAttachments(SourceReplicationInformation destinationsReplicationInformationForSource)
		{
			RavenJArray jsonAttachments = null;
			try
			{
				string destinationInstanceId = destinationsReplicationInformationForSource.ServerInstanceId.ToString();

				docDb.TransactionalStorage.Batch(actions =>
				{
					jsonAttachments = new RavenJArray(actions.Attachments.GetAttachmentsAfter(destinationsReplicationInformationForSource.LastAttachmentEtag)
						.Where(x => x.Key.StartsWith("Raven/") == false) // don't replicate system docs
						.Where(x => x.Metadata.Value<string>(ReplicationConstants.RavenReplicationSource) != destinationInstanceId) // prevent replicating back to source
						.Where(x => x.Metadata[ReplicationConstants.RavenReplicationConflict] == null) // don't replicate conflicted documents, that just propgate the conflict
						.Take(100)
						.Select(x => new RavenJObject
						{
							{"@metadata", x.Metadata},
							{"@id", x.Key},
							{"@etag", x.Etag.ToByteArray()},
							{"data", actions.Attachments.GetAttachment(x.Key).Data}
						}));
				});
			}
			catch (Exception e)
			{
				log.WarnException("Could not get documents to replicate after: " + destinationsReplicationInformationForSource.LastAttachmentEtag, e);
			}
			return jsonAttachments;
		}