/*
 * Copyright 2024-2026 the original author or authors.
 *
 * 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
 *
 *      https://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 com.alibaba.cloud.ai.graph.state;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.function.Supplier;

import static java.util.Collections.unmodifiableList;
import static java.util.Optional.ofNullable;

public class AppenderChannel<T> implements Channel<List<T>> {

	private static final Logger log = LoggerFactory.getLogger(AppenderChannel.class);

	/**
	 * A functional interface that is used to remove elements from a list.
	 *
	 * @param <T> the type of elements in the list
	 */
	@FunctionalInterface
	public interface RemoveIdentifier<T> {

		/**
		 * Compares the specified element with the element at the given index.
		 * @param element the element to be compared
		 * @param atIndex the index of the element to compare with
		 * @return a negative integer, zero, or a positive integer as this object is less
		 * than, equal to, or greater than the specified object.
		 */
		int compareTo(T element, int atIndex);

	}



	private final Reducer<List<T>> reducer;

	private final Supplier<List<T>> defaultProvider;

	/**
	 * Returns an {@link Optional} containing the current reducer if it is non-null.
	 * @return an {@code Optional} describing the current reducer wrapped in a non-empty
	 * optional, or an empty optional if no such reducer exists
	 */
	@Override
	public Optional<Reducer<List<T>>> getReducer() {
		return ofNullable(reducer);
	}

	/**
	 * Returns the default provider or {@code Optional.empty()} if no default provider is
	 * set.
	 * @return an {@code Optional} containing the default provider, or
	 * {@code Optional.empty()}
	 */
	@Override
	public Optional<Supplier<List<T>>> getDefault() {
		return ofNullable(defaultProvider);
	}

	/**
	 * Creates an instance of `AppenderChannel` using the provided supplier to get the
	 * default list.
	 * @param <T> the type of elements in the list
	 * @param defaultProvider a supplier that provides the default list of elements
	 * @return a new instance of `AppenderChannel`
	 */
	public static <T> AppenderChannel<T> of(Supplier<List<T>> defaultProvider) {
		return new AppenderChannel<>(defaultProvider);
	}

	/**
	 * Constructs a new instance of {@code AppenderChannel} with the specified default
	 * provider.
	 * @param defaultProvider a supplier for the default list that will be used when no
	 * other list is available
	 */
	private AppenderChannel(Supplier<List<T>> defaultProvider) {
		this.reducer = new Reducer<>() {
			/**
			 * Combines two lists into one. If the first list is null, the second list is
			 * returned. Otherwise, the second list is added to the end of the first list
			 * and the resulting list is returned.
			 * @param left the first list; may be null
			 * @param right the second list
			 * @return a new list containing all elements from both input lists
			 */
			@Override
			public List<T> apply(List<T> left, List<T> right) {
				if (left == null) {
					return right;
				}
				left.addAll(right);
				return left;
			}
		};
		this.defaultProvider = defaultProvider;
	}

	/**
	 * This method removes elements from a given list based on the specified
	 * {@link RemoveIdentifier}. It creates a copy of the original list, performs the
	 * removal operation, and returns an immutable view of the result.
	 * @param list The list from which elements will be removed.
	 * @param removeIdentifier An instance of {@link RemoveIdentifier} that defines how to
	 * identify elements for removal.
	 * @return An unmodifiable view of the modified list with specified elements removed.
	 */
	private List<T> remove(List<T> list, RemoveIdentifier<T> removeIdentifier) {
		var result = new ArrayList<>(list);
		removeFromList(result, removeIdentifier);
		return unmodifiableList(result);
	}

	/**
	 * Removes an element from the list that matches the specified identifier.
	 *
	 * <p>
	 * This method iterates over the provided list and removes the first element for which
	 * the {@link RemoveIdentifier#compareTo} method returns zero.
	 * </p>
	 * @param result the list to be modified
	 * @param removeIdentifier the identifier used to find the element to remove
	 */
	private void removeFromList(List<T> result, RemoveIdentifier<T> removeIdentifier) {
		for (int i = 0; i < result.size(); i++) {
			if (removeIdentifier.compareTo(result.get(i), i) == 0) {
				result.remove(i);
				break;
			}
		}
	}

	/**
	 * Represents a record for data removal operations with generic types.
	 *
	 * @param <T> the type of elements in the old values list
	 */
	public record RemoveData<T>(List<T> oldValues, List<? extends Object> newValues) {

		// copy constructor. make sure to copy the list to make them modifiable
		public RemoveData {
			oldValues = new ArrayList<>(oldValues);
			newValues = new ArrayList<>(newValues);
		}
	};

	/**
	 * Evaluates the removal of identifiers from the new values list and updates the
	 * RemoveData object accordingly.
	 * @param oldValues a {@code List} of old values
	 * @param newValues a {@code List} of new values containing {@code RemoveIdentifier}s
	 * to be evaluated for removal
	 * @return a {@literal RemoveData<T>} object with updated old and new values after
	 * removing identifiers
	 */
	@SuppressWarnings("unchecked")
	private RemoveData<T> evaluateRemoval(List<T> oldValues, List<?> newValues) {

		final var result = new RemoveData<>(oldValues, newValues);

		newValues.stream().filter(value -> value instanceof RemoveIdentifier<?>).forEach(value -> {
			result.newValues().remove(value);
			var removeIdentifier = (RemoveIdentifier<T>) value;
			removeFromList(result.oldValues(), removeIdentifier);

		});
		return result;

	}

	/**
	 * Updates the value for a given key in the channel.
	 * @param key The key for which the value needs to be updated.
	 * @param oldValue The old value that is being replaced.
	 * @param newValue The new value to be set. If null, the old value will be returned.
	 * @return The updated old value or the new value if the update was successful.
	 * @throws UnsupportedOperationException If the channel does not support updates,
	 * typically due to an immutable list being used.
	 */
	@SuppressWarnings("unchecked")
	public Object update(String key, Object oldValue, Object newValue) {

		if (newValue == null) {
			return oldValue;
		}

		boolean oldValueIsList = oldValue instanceof List<?>;

		try {
			if (oldValueIsList && newValue instanceof RemoveIdentifier<?>) {
				return remove((List<T>) oldValue, (RemoveIdentifier<T>) newValue);
			}
			List<?> list = null;
			if (newValue instanceof List) {
				list = (List<Object>) newValue;
			}
			else if (newValue.getClass().isArray()) {
				list = Arrays.asList((T[]) newValue);
			}
			if (list != null) {
				if (list.isEmpty()) {
					return oldValue;
				}
				if (oldValueIsList) {
					var result = evaluateRemoval((List<T>) oldValue, list);
					return Channel.super.update(key, result.oldValues(), result.newValues());
				}
				return Channel.super.update(key, oldValue, list);
			}
			// this is to allow single value other than List or Array
			try {
				T typedValue = (T) newValue;
				return Channel.super.update(key, oldValue, List.of(typedValue));
			}
			catch (ClassCastException e) {
				log.error("Unsupported content type: {}", newValue.getClass());
				throw e;
			}
		}
		catch (UnsupportedOperationException ex) {
			log.error(
					"Unsupported operation: probably because the appendable channel has been initialized with a immutable List. Check please !");
			throw ex;
		}
	}

}
