/*
 * Copyright (C) 2020 ActiveJ LLC.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.activej.fs.adapter.impl;

import io.activej.async.function.AsyncBiConsumer;
import io.activej.async.function.AsyncConsumer;
import io.activej.bytebuf.ByteBuf;
import io.activej.common.annotation.ExposedInternals;
import io.activej.csp.consumer.ChannelConsumer;
import io.activej.csp.supplier.ChannelSupplier;
import io.activej.fs.FileMetadata;
import io.activej.fs.IFileSystem;
import io.activej.fs.exception.FileSystemBatchException;
import io.activej.fs.exception.FileSystemScalarException;
import io.activej.fs.exception.ForbiddenPathException;
import io.activej.promise.Promise;
import org.jetbrains.annotations.Nullable;

import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
import java.util.function.Predicate;

import static io.activej.common.Checks.checkArgument;
import static io.activej.common.Utils.isBijection;
import static java.lang.Boolean.TRUE;
import static java.util.stream.Collectors.*;

@ExposedInternals
public final class Filter implements IFileSystem {

	public final IFileSystem parent;
	public final Predicate<String> predicate;

	public Filter(IFileSystem parent, Predicate<String> predicate) {
		this.parent = parent;
		this.predicate = predicate;
	}

	@Override
	public Promise<ChannelConsumer<ByteBuf>> upload(String name) {
		if (!predicate.test(name)) {
			return Promise.ofException(new ForbiddenPathException());
		}
		return parent.upload(name);
	}

	@Override
	public Promise<ChannelConsumer<ByteBuf>> upload(String name, long size) {
		if (!predicate.test(name)) {
			return Promise.ofException(new ForbiddenPathException());
		}
		return parent.upload(name);
	}

	@Override
	public Promise<ChannelConsumer<ByteBuf>> append(String name, long offset) {
		if (!predicate.test(name)) {
			return Promise.ofException(new ForbiddenPathException());
		}
		return parent.append(name, offset);
	}

	@Override
	public Promise<ChannelSupplier<ByteBuf>> download(String name, long offset, long limit) {
		if (!predicate.test(name)) {
			return Promise.ofException(new ForbiddenPathException());
		}
		return parent.download(name, offset, limit);
	}

	@Override
	public Promise<Void> copy(String name, String target) {
		return filteringOp(name, target, parent::copy);
	}

	@Override
	public Promise<Void> copyAll(Map<String, String> sourceToTarget) {
		checkArgument(isBijection(sourceToTarget), "Targets must be unique");
		if (sourceToTarget.isEmpty()) return Promise.complete();

		return filteringOp(sourceToTarget, parent::copyAll);
	}

	@Override
	public Promise<Void> move(String name, String target) {
		return filteringOp(name, target, parent::move);
	}

	@Override
	public Promise<Void> moveAll(Map<String, String> sourceToTarget) {
		checkArgument(isBijection(sourceToTarget), "Targets must be unique");
		if (sourceToTarget.isEmpty()) return Promise.complete();

		return filteringOp(sourceToTarget, parent::moveAll);
	}

	@Override
	public Promise<Map<String, FileMetadata>> list(String glob) {
		return parent.list(glob)
			.map(map -> map.entrySet().stream()
				.filter(entry -> predicate.test(entry.getKey()))
				.collect(toMap(Map.Entry::getKey, Map.Entry::getValue)));
	}

	@Override
	public Promise<@Nullable FileMetadata> info(String name) {
		if (!predicate.test(name)) {
			return Promise.of(null);
		}
		return parent.info(name);
	}

	@Override
	public Promise<Map<String, FileMetadata>> infoAll(Set<String> names) {
		Map<Boolean, Set<String>> partitioned = names.stream().collect(partitioningBy(predicate, toSet()));
		Set<String> query = partitioned.get(TRUE);
		return query.isEmpty() ?
			Promise.of(Map.of()) :
			parent.infoAll(query);
	}

	@Override
	public Promise<Void> ping() {
		return parent.ping();
	}

	@Override
	public Promise<Void> delete(String name) {
		if (!predicate.test(name)) {
			return Promise.complete();
		}
		return parent.delete(name);
	}

	@Override
	public Promise<Void> deleteAll(Set<String> toDelete) {
		return parent.deleteAll(toDelete.stream()
			.filter(predicate)
			.collect(toSet()));
	}

	private Promise<Void> filteringOp(String source, String target, AsyncBiConsumer<String, String> original) {
		if (!predicate.test(source)) {
			return Promise.ofException(new ForbiddenPathException("Path '" + source + "' is forbidden"));
		}
		if (!predicate.test(target)) {
			return Promise.ofException(new ForbiddenPathException("Path '" + target + "' is forbidden"));
		}
		return original.accept(source, target);
	}

	private Promise<Void> filteringOp(Map<String, String> sourceToTarget, AsyncConsumer<Map<String, String>> original) {
		Map<String, String> renamed = new LinkedHashMap<>();
		Map<String, FileSystemScalarException> exceptions = new HashMap<>();
		for (Map.Entry<String, String> entry : sourceToTarget.entrySet()) {
			String source = entry.getKey();
			if (!predicate.test(source)) {
				exceptions.put(source, new ForbiddenPathException("Path '" + source + "' is forbidden"));
			}
			String target = entry.getValue();
			if (!predicate.test(target)) {
				exceptions.put(source, new ForbiddenPathException("Path '" + target + "' is forbidden"));
			}
			renamed.put(source, target);
		}
		if (!exceptions.isEmpty()) {
			return Promise.ofException(new FileSystemBatchException(exceptions));
		}
		return original.accept(renamed);
	}
}
