public long GetNextIdentityValueWithoutOverwritingOnExistingDocuments(string key,
 IStorageActionsAccessor actions,
 TransactionInformation transactionInformation)
     {
         int tries;
         return GetNextIdentityValueWithoutOverwritingOnExistingDocuments(key, actions, transactionInformation, out tries);
     }

        protected override bool IsIndexStale(IndexStats indexesStat, IStorageActionsAccessor actions, bool isIdle, Reference<bool> onlyFoundIdleWork)
        {
            var isStale = actions.Staleness.IsMapStale(indexesStat.Id);
            var indexingPriority = indexesStat.Priority;
            if (isStale == false)
                return false;

            if (indexingPriority == IndexingPriority.None)
                return true;

            if ((indexingPriority & IndexingPriority.Normal) == IndexingPriority.Normal)
            {
                onlyFoundIdleWork.Value = false;
                return true;
            }

            if ((indexingPriority & (IndexingPriority.Disabled | IndexingPriority.Error)) != IndexingPriority.None)
                return false;

            if (isIdle == false)
                return false; // everything else is only valid on idle runs

            if ((indexingPriority & IndexingPriority.Idle) == IndexingPriority.Idle)
                return true;

            if ((indexingPriority & IndexingPriority.Abandoned) == IndexingPriority.Abandoned)
            {
                var timeSinceLastIndexing = (SystemTime.UtcNow - indexesStat.LastIndexingTime);

                return (timeSinceLastIndexing > context.Configuration.TimeToWaitBeforeRunningAbandonedIndexes);
            }

            throw new InvalidOperationException("Unknown indexing priority for index " + indexesStat.Id + ": " + indexesStat.Priority);
        }

	    protected override DatabaseTask GetApplicableTask(IStorageActionsAccessor actions)
	    {
		    var removeFromIndexTasks = (DatabaseTask)actions.Tasks.GetMergedTask<RemoveFromIndexTask>();
			var touchReferenceDocumentIfChangedTask = removeFromIndexTasks ?? actions.Tasks.GetMergedTask<TouchReferenceDocumentIfChangedTask>();

		    return touchReferenceDocumentIfChangedTask;
	    }

		public void Delete(string fileName, IStorageActionsAccessor actionsAccessor = null)
		{
            RavenJObject metadata = null;

			Action<IStorageActionsAccessor> delete = accessor =>
			{
				accessor.DeleteConfig(RavenFileNameHelper.ConflictConfigNameForFile(fileName));
				metadata = accessor.GetFile(fileName, 0, 0).Metadata;
				metadata.Remove(SynchronizationConstants.RavenSynchronizationConflict);
				metadata.Remove(SynchronizationConstants.RavenSynchronizationConflictResolution);
				accessor.UpdateFileMetadata(fileName, metadata);
			};

			if (actionsAccessor != null)
			{
				delete(actionsAccessor);
			}
			else
			{
				storage.Batch(delete);
			}

			if (metadata != null)
			{
				index.Index(fileName, metadata);
			}
		}

        public override void IndexDocuments(
			AbstractViewGenerator viewGenerator, 
			IEnumerable<dynamic> documents, 
			WorkContext context, 
			IStorageActionsAccessor actions, 
			DateTime minimumTimestamp)
        {
            actions.Indexing.SetCurrentIndexStatsTo(name);
            var count = 0;
            Func<object, object> documentIdFetcher = null;
            var reduceKeys = new HashSet<string>(StringComparer.InvariantCultureIgnoreCase);
            var documentsWrapped = documents.Select(doc =>
            {
                var documentId = doc.__document_id;
                foreach (var reduceKey in actions.MappedResults.DeleteMappedResultsForDocumentId((string)documentId, name))
                {
                    reduceKeys.Add(reduceKey);
                }
                return doc;
            });
            foreach (var doc in RobustEnumeration(documentsWrapped, viewGenerator.MapDefinition, actions, context))
            {
                count++;

                documentIdFetcher = CreateDocumentIdFetcherIfNeeded(documentIdFetcher, doc);

                var docIdValue = documentIdFetcher(doc);
                if (docIdValue == null)
                    throw new InvalidOperationException("Could not find document id for this document");

                var reduceValue = viewGenerator.GroupByExtraction(doc);
                if (reduceValue == null)
                {
                    logIndexing.DebugFormat("Field {0} is used as the reduce key and cannot be null, skipping document {1}", viewGenerator.GroupByExtraction, docIdValue);
                    continue;
                }
                var reduceKey = ReduceKeyToString(reduceValue);
                var docId = docIdValue.ToString();

                reduceKeys.Add(reduceKey);

                var data = GetMapedData(doc);

                logIndexing.DebugFormat("Mapped result for '{0}': '{1}'", name, data);

                var hash = ComputeHash(name, reduceKey);

                actions.MappedResults.PutMappedResult(name, docId, reduceKey, data, hash);

                actions.Indexing.IncrementSuccessIndexing();
            }

            actions.Tasks.AddTask(new ReduceTask
            {
                Index = name,
                ReduceKeys = reduceKeys.ToArray()
            }, minimumTimestamp);

            logIndexing.DebugFormat("Mapped {0} documents for {1}", count, name);
        }

		private TimeSpan SynchronizationTimeout(IStorageActionsAccessor accessor)
		{
			var timeoutConfigExists = accessor.TryGetConfigurationValue(
				SynchronizationConstants.RavenSynchronizationLockTimeout, out configuredTimeout);

			return timeoutConfigExists ? configuredTimeout : defaultTimeout;
		}

        public IndexingStorageActions(TableStorage tableStorage, IUuidGenerator generator, Reference<SnapshotReader> snapshot, Reference<WriteBatch> writeBatch, IStorageActionsAccessor storageActionsAccessor, IBufferPool bufferPool)
			: base(snapshot, bufferPool)
		{
			this.tableStorage = tableStorage;
			this.generator = generator;
			this.writeBatch = writeBatch;
			this.currentStorageActionsAccessor = storageActionsAccessor;
		}

        private void ReplicateDocument(IStorageActionsAccessor actions, string id, RavenJObject metadata, RavenJObject document, string src)
        {
            var existingDoc = actions.Documents.DocumentByKey(id, null);
            if (existingDoc == null)
            {
                log.DebugFormat("New document {0} replicated successfully from {1}", id, src);
				actions.Documents.AddDocument(id, Guid.Empty, document, metadata);
                return;
            }
            
            var existingDocumentIsInConflict = existingDoc.Metadata[ReplicationConstants.RavenReplicationConflict] != null;
            if (existingDocumentIsInConflict == false &&                    // if the current document is not in conflict, we can continue without having to keep conflict semantics
                (IsDirectChildOfCurrentDocument(existingDoc, metadata))) // this update is direct child of the existing doc, so we are fine with overwriting this
            {
                log.DebugFormat("Existing document {0} replicated successfully from {1}", id, src);
				actions.Documents.AddDocument(id, null, document, metadata);
                return;
            }


        	var newDocumentConflictId = id + "/conflicts/" +
        	                            metadata.Value<string>(ReplicationConstants.RavenReplicationSource) + "/" +
        	                            metadata.Value<string>("@etag");
            metadata.Add(ReplicationConstants.RavenReplicationConflict, RavenJToken.FromObject(true));
			actions.Documents.AddDocument(newDocumentConflictId, null, document, metadata);

            if (existingDocumentIsInConflict) // the existing document is in conflict
            {
                log.DebugFormat("Conflicted document {0} has a new version from {1}, adding to conflicted documents", id, src);
                
                // just update the current doc with the new conflict document
                existingDoc.DataAsJson.Value<RavenJArray>("Conflicts").Add(RavenJToken.FromObject(newDocumentConflictId));
				actions.Documents.AddDocument(id, existingDoc.Etag, existingDoc.DataAsJson, existingDoc.Metadata);
                return;
            }
            log.DebugFormat("Existing document {0} is in conflict with replicated version from {1}, marking document as conflicted", id, src);
                
            // we have a new conflict
            // move the existing doc to a conflict and create a conflict document
        	var existingDocumentConflictId = id + "/conflicts/" + Database.TransactionalStorage.Id + "/" + existingDoc.Etag;
            
            existingDoc.Metadata.Add(ReplicationConstants.RavenReplicationConflict, RavenJToken.FromObject(true));
			actions.Documents.AddDocument(existingDocumentConflictId, null, existingDoc.DataAsJson, existingDoc.Metadata);
        	actions.Documents.AddDocument(id, null,
        	                              new RavenJObject
        	                              {
        	                              	{
        	                              	"Conflicts", new RavenJArray(existingDocumentConflictId, newDocumentConflictId)
        	                              	}
        	                              },
        	                              new RavenJObject
        	                              {
        	                              	{ReplicationConstants.RavenReplicationConflict, true},
        	                              	{"@Http-Status-Code", 409},
        	                              	{"@Http-Status-Description", "Conflict"}
        	                              });
        }

        public MappedResultsStorageActions(TableStorage tableStorage, IUuidGenerator generator, OrderedPartCollection<AbstractDocumentCodec> documentCodecs, Reference<SnapshotReader> snapshot, Reference<WriteBatch> writeBatch, IBufferPool bufferPool, IStorageActionsAccessor storageActionsAccessor)
			: base(snapshot, bufferPool)
		{
			this.tableStorage = tableStorage;
			this.generator = generator;
			this.documentCodecs = documentCodecs;
			this.writeBatch = writeBatch;
	        this.storageActionsAccessor = storageActionsAccessor;
		}

        public MappedResultsStorageActions(TableStorage tableStorage, IUuidGenerator generator, OrderedPartCollection<AbstractDocumentCodec> documentCodecs, Reference<SnapshotReader> snapshot, 
			Reference<WriteBatch> writeBatch, IBufferPool bufferPool, IStorageActionsAccessor storageActionsAccessor, ConcurrentDictionary<int, RemainingReductionPerLevel> ScheduledReductionsPerViewAndLevel)
			: base(snapshot, bufferPool)
		{
			this.tableStorage = tableStorage;
			this.generator = generator;
			this.documentCodecs = documentCodecs;
			this.writeBatch = writeBatch;
	        this.storageActionsAccessor = storageActionsAccessor;
	        this.scheduledReductionsPerViewAndLevel = ScheduledReductionsPerViewAndLevel;
		}

		private TimeSpan SynchronizationTimeout(IStorageActionsAccessor accessor)
		{
            string timeoutConfigKey = string.Empty;
            accessor.TryGetConfigurationValue<string>(SynchronizationConstants.RavenSynchronizationLockTimeout, out timeoutConfigKey);

            TimeSpan timeoutConfiguration;
            if (TimeSpan.TryParse(timeoutConfigKey, out timeoutConfiguration))
                return timeoutConfiguration;

            return defaultTimeout;
		}

		public void LockByCreatingSyncConfiguration(string fileName, ServerInfo sourceServer, IStorageActionsAccessor accessor)
		{
			var syncLock = new SynchronizationLock
			{
				SourceServer = sourceServer,
				FileLockedAt = DateTime.UtcNow
			};

			accessor.SetConfig(RavenFileNameHelper.SyncLockNameForFile(fileName), syncLock.AsConfig());

			log.Debug("File '{0}' was locked", fileName);
		}

		public override void IndexDocuments(
			AbstractViewGenerator viewGenerator, 
			IEnumerable<object> documents,
			WorkContext context,
			IStorageActionsAccessor actions)
		{
			actions.Indexing.SetCurrentIndexStatsTo(name);
			var count = 0;
			Write(indexWriter =>
			{
				bool madeChanges = false;
				PropertyDescriptorCollection properties = null;
				var processedKeys = new HashSet<string>();
				var documentsWrapped = documents.Select((dynamic doc) =>
				{
					var documentId = doc.__document_id.ToString();
					if (processedKeys.Add(documentId) == false)
						return doc;
					madeChanges = true;
					context.IndexUpdateTriggers.Apply(trigger => trigger.OnIndexEntryDeleted(name, documentId));
					indexWriter.DeleteDocuments(new Term("__document_id", documentId));
					return doc;
				});
				foreach (var doc in RobustEnumeration(documentsWrapped, viewGenerator.MapDefinition, actions, context))
				{
					count++;

				    string newDocId;
				    IEnumerable<AbstractField> fields;
                    if (doc is DynamicJsonObject)
                        fields = ExtractIndexDataFromDocument((DynamicJsonObject) doc, out newDocId);
                    else
                        fields = ExtractIndexDataFromDocument(properties, doc, out newDocId);
				   
                    if (newDocId != null)
                    {
                        var luceneDoc = new Document();
                        luceneDoc.Add(new Field("__document_id", newDocId, Field.Store.YES, Field.Index.NOT_ANALYZED));

                    	madeChanges = true;
                        CopyFieldsToDocument(luceneDoc, fields);
                        context.IndexUpdateTriggers.Apply(trigger => trigger.OnIndexEntryCreated(name, newDocId, luceneDoc));
                        log.DebugFormat("Index '{0}' resulted in: {1}", name, luceneDoc);
                        indexWriter.AddDocument(luceneDoc);
                    }

					actions.Indexing.IncrementSuccessIndexing();
				}

				return madeChanges;
			});
			log.DebugFormat("Indexed {0} documents for {1}", count, name);
		}

        public void LockByCreatingSyncConfiguration(string fileName, FileSystemInfo sourceFileSystem, IStorageActionsAccessor accessor)
        {
            var syncLock = new SynchronizationLock
            {
                SourceFileSystem = sourceFileSystem,
                FileLockedAt = DateTime.UtcNow
            };

            accessor.SetConfig(RavenFileNameHelper.SyncLockNameForFile(fileName), JsonExtensions.ToJObject(syncLock));

            log.Debug("File '{0}' was locked", fileName);
        }

        private void ReplicateAttachment(IStorageActionsAccessor actions, string id, JObject metadata, byte[] data, Guid lastEtag ,string src)
        {
            var existingAttachment = actions.Attachments.GetAttachment(id);
            if (existingAttachment == null)
            {
                log.DebugFormat("New attachment {0} replicated successfully from {1}", id, src);
				actions.Attachments.AddAttachment(id, Guid.Empty, data, metadata);
                return;
            }
            
            var existingDocumentIsInConflict = existingAttachment.Metadata[ReplicationConstants.RavenReplicationConflict] != null;
            if (existingDocumentIsInConflict == false &&                    // if the current document is not in conflict, we can continue without having to keep conflict semantics
                (IsDirectChildOfCurrentAttachment(existingAttachment, metadata))) // this update is direct child of the existing doc, so we are fine with overwriting this
            {
                log.DebugFormat("Existing document {0} replicated successfully from {1}", id, src);
                actions.Attachments.AddAttachment(id, null, data, metadata);
                return;
            }


            var newDocumentConflictId = id + "/conflicts/" +
				metadata.Value<string>(ReplicationConstants.RavenReplicationSource) + 
				"/" + lastEtag;
            metadata.Add(ReplicationConstants.RavenReplicationConflict, JToken.FromObject(true));
            actions.Attachments.AddAttachment(newDocumentConflictId, null, data, metadata);

            if (existingDocumentIsInConflict) // the existing document is in conflict
            {
                log.DebugFormat("Conflicted document {0} has a new version from {1}, adding to conflicted documents", id, src);
                
                // just update the current doc with the new conflict document
                existingAttachment.Metadata.Value<JArray>("Conflicts").Add(JToken.FromObject(newDocumentConflictId));
                actions.Attachments.AddAttachment(id, existingAttachment.Etag, existingAttachment.Data, existingAttachment.Metadata);
                return;
            }
            log.DebugFormat("Existing document {0} is in conflict with replicated version from {1}, marking document as conflicted", id, src);
                
            // we have a new conflict
            // move the existing doc to a conflict and create a conflict document
			var existingDocumentConflictId = id + "/conflicts/" + Database.TransactionalStorage.Id + "/" + existingAttachment.Etag;
            
            existingAttachment.Metadata.Add(ReplicationConstants.RavenReplicationConflict, JToken.FromObject(true));
            actions.Attachments.AddAttachment(existingDocumentConflictId, null, existingAttachment.Data, existingAttachment.Metadata);
            actions.Attachments.AddAttachment(id, null,
                                new JObject(
                                    new JProperty("Conflicts", new JArray(existingDocumentConflictId, newDocumentConflictId))).ToBytes(),
                                new JObject(
                                    new JProperty(ReplicationConstants.RavenReplicationConflict, true), 
                                    new JProperty("@Http-Status-Code", 409),
                                    new JProperty("@Http-Status-Description", "Conflict")
                                    ));
        }

        public bool TimeoutExceeded(string fileName, IStorageActionsAccessor accessor)
        {
            SynchronizationLock syncLock;

            try
            {
                syncLock = accessor.GetConfig(RavenFileNameHelper.SyncLockNameForFile(fileName)).JsonDeserialization<SynchronizationLock>();				
            }
            catch (FileNotFoundException)
            {
                return true;
            }

            return (DateTime.UtcNow - syncLock.FileLockedAt).TotalMilliseconds > SynchronizationConfigAccessor.GetOrDefault(accessor).SynchronizationLockTimeoutMiliseconds;
        }

        public static SynchronizationConfig GetOrDefault(IStorageActionsAccessor accessor)
        {
            try
            {
                if (accessor.ConfigExists(SynchronizationConstants.RavenSynchronizationConfig) == false)
                    return new SynchronizationConfig(); // return a default one

                return accessor.GetConfig(SynchronizationConstants.RavenSynchronizationConfig).JsonDeserialization<SynchronizationConfig>();
            }
            catch (Exception e)
            {
                Log.Warn("Could not deserialize a synchronization configuration", e);
                return new SynchronizationConfig(); // return a default one
            }
        }

		public bool TimeoutExceeded(string fileName, IStorageActionsAccessor accessor)
		{
			SynchronizationLock syncLock;

			try
			{
				syncLock =
					accessor.GetConfig(RavenFileNameHelper.SyncLockNameForFile(fileName)).AsObject<SynchronizationLock>();
			}
			catch (FileNotFoundException)
			{
				return true;
			}

			return DateTime.UtcNow - syncLock.FileLockedAt > SynchronizationTimeout(accessor);
		}

		/// <summary>
		/// We need to NOT remove documents that has been removed then added.
		/// We DO remove documents that would be filtered out because of an Entity Name changed, though.
		/// </summary>
		private bool FilterDocuments(WorkContext context, IStorageActionsAccessor accessor, string key)
		{
			var documentMetadataByKey = accessor.Documents.DocumentMetadataByKey(key, null);
			if (documentMetadataByKey == null)
				return true;
			var generator = context.IndexDefinitionStorage.GetViewGenerator(Index);
			if (generator == null)
				return false;

			if (generator.ForEntityNames.Count == 0)
				return false;// there is a new document and this index applies to it

			var entityName = documentMetadataByKey.Metadata.Value<string>(Constants.RavenEntityName);
			if (entityName == null)
				return true; // this document doesn't belong to this index any longer, need to remove it

			return generator.ForEntityNames.Contains(entityName) == false;
		}

		private static bool TryGetDeserializedConfig(IStorageActionsAccessor accessor, string configurationName, out FileVersioningConfiguration fileVersioningConfiguration)
		{
			if (accessor.ConfigExists(configurationName) == false)
			{
				fileVersioningConfiguration = null;
				return false;
			}

			var configuration = accessor.GetConfig(configurationName);
			if (configuration == null)
			{
				fileVersioningConfiguration = null;
				return false;
			}

			fileVersioningConfiguration = configuration.JsonDeserialization<FileVersioningConfiguration>();
			return true;
		}