/**
 * Copyright © 2012-2020 by Data Virtuality GmbH
 */
package org.teiid.metadata;

import com.datavirtuality.dv.core.interfaces.DVCaller;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeUnit;
import org.teiid.adminapi.DVMetadataCache;
import org.teiid.metadata.odata.ODataMetadata;
import org.teiid.metadata.odata.ODataSchemaBuilder;

/**
 * @author Dmitry Pogorelov
 * @date 2020.10.13
 */
public class DVMetadataRecordHolder implements DVMetadataCache {
	private NavigableMap<AbstractMetadataRecord.UUIDKey, AbstractMetadataRecord> uuidHolder = new ConcurrentSkipListMap<>();
	private Timer timer = new Timer("UUID Holder Cleaner", true);
	private static final int THRESHOLD_TO_REMOVE = 10;
	public static final long REFRESH_TIMEOUT = TimeUnit.SECONDS.toMillis(60);
	private volatile boolean testMode;
	private Object lock = new Object();
	private ODataMetadata oDataMetadata;

	// for debug purposes
	private NavigableMap<AbstractMetadataRecord.UUIDKey, AbstractMetadataRecord> uuidHolder2 = new ConcurrentSkipListMap<>();
	private static boolean test;

	public DVMetadataRecordHolder(ODataMetadata oDataMetadata, boolean runCleaner) {
		this.oDataMetadata = oDataMetadata;
		if (runCleaner)
			timer.schedule(new UUIDHolderCleaner(this), 0, REFRESH_TIMEOUT);
	}

	public DVMetadataRecordHolder(String vdbName, boolean runCleaner) {
		this(new ODataMetadata(vdbName), runCleaner);
	}

	public DVMetadataRecordHolder(ODataMetadata oDataMetadata, boolean runCleaner, NavigableMap<AbstractMetadataRecord.UUIDKey, AbstractMetadataRecord> uuidHolder) {
		this(oDataMetadata, runCleaner);
		this.uuidHolder = uuidHolder;
	}

	public boolean isTestMode() {
		return testMode;
	}

	public void setTestMode(boolean testMode) {
		synchronized (lock) {
			this.testMode = testMode;
		}
	}

	public NavigableMap<AbstractMetadataRecord.UUIDKey, AbstractMetadataRecord> getUuidHolder() {
		return uuidHolder;
	}

	public AbstractMetadataRecord getRecord(String uuid) {
		return uuidHolder.get(new AbstractMetadataRecord.UUIDKey(uuid));
	}

	public ODataMetadata getODataMetadata() {
		return oDataMetadata;
	}

	public void cleanCache() {
		synchronized (lock) {
			int count = testMode ? THRESHOLD_TO_REMOVE * 2 : 1;
			Set<AbstractMetadataRecord.UUIDKey> handled = new HashSet<>();
			for (int i = 0; i < count; i++) {
				Map<AbstractMetadataRecord.UUIDKey, AbstractMetadataRecord> toRemove = new HashMap<>();
				for (Map.Entry<AbstractMetadataRecord.UUIDKey, AbstractMetadataRecord> entry : uuidHolder.entrySet()) {
					handleRecord(entry.getValue(), handled, toRemove);
				}
				for (AbstractMetadataRecord record : toRemove.values()) {
					removeRecord(record);
				}
			}
		}
	}

	private void handleRecord(AbstractMetadataRecord record, Set<AbstractMetadataRecord.UUIDKey> handled, Map<AbstractMetadataRecord.UUIDKey, AbstractMetadataRecord> toRemove) {
		AbstractMetadataRecord parent = record.getParent();
		if (parent != null) {
			boolean contains = parent instanceof ColumnSet ? // we don't store ColumnStore in uuidHolder
					true : uuidHolder.containsKey(parent.getUUIDKey());
			handleRecord(parent, handled, toRemove);
			if (!contains) {
				if (!toRemove.containsKey(record.getUUIDKey())) {
					if (record.getThresholdToRemove() == THRESHOLD_TO_REMOVE) {
						toRemove.put(record.getUUIDKey(), record);
					} else if (!handled.contains(record.getUUIDKey())) {
						record.setThresholdToRemove(record.getThresholdToRemove() + 1);
						handled.add(record.getUUIDKey());
					}
				}
			}
		}
	}

	public AbstractMetadataRecord removeRecord(AbstractMetadataRecord record) {
		if (record != null) {
			if (record instanceof Schema) {
				removeSchemaChildren((Schema) record);
			} else if (record instanceof Table) {
				removeTableChildren((Table) record);
			} else if (record instanceof Procedure) {
				removeProcChildren((Procedure) record);
			} else if (record instanceof FunctionMethod) {
				removeFuncChildren((FunctionMethod) record);
			}
			uuidHolder.remove(record.getUUIDKey());
			if (test)
				uuidHolder2.remove(record.getUUIDKey());
			record.setUuidChanger(null);
			oDataMetadata.removeRecord(record);
		}
		return record;
	}

	public AbstractMetadataRecord putRecord(AbstractMetadataRecord oldValue, AbstractMetadataRecord newValue) {
		removeRecord(oldValue);
		if (newValue != null) {
			if (newValue instanceof ProcedureParameter)
				return oldValue;
			else if (newValue instanceof Column) {
				ColumnSet columnSet = ((Column) newValue).getParent();
				if (columnSet != null && columnSet.getParent() instanceof Procedure) {
					return oldValue;
				}
			} else if (newValue instanceof Procedure) {
				Procedure proc = (Procedure) newValue;
				proc.getParameters().stream().forEach(p -> putValue(p));
				if (proc.getResultSet() != null && proc.getResultSet().getColumns() != null)
					proc.getResultSet().getColumns().stream().forEach(c -> putValue(c));
			} else if (newValue instanceof FunctionMethod) {
				FunctionMethod func = (FunctionMethod) newValue;
				func.getInputParameters().stream().forEach(p -> putValue(p));
				if (func.getReturnParameter() != null)
					putValue(func.getReturnParameter());	
			} else if (newValue instanceof KeyRecord) {
				//pretend that each keyrecord contains its own columns
				// we need it to compensate the pg_catalog.getOid(kc.uid) + kc.position as oid in
				// PgCatalogMetadataStore.add_pg_attribute method
				newValue.generateOid(((KeyRecord) newValue).getColumns().size() + 1);
			}
			putValue(newValue);
			oDataMetadata.putRecord(newValue);
		}
		return oldValue;
	}

	private void putValue(AbstractMetadataRecord newValue) {
		((ConcurrentSkipListMap)uuidHolder).putIfAbsent(newValue.getUUIDKey(), newValue);
		if (test)
			((ConcurrentSkipListMap)uuidHolder2).putIfAbsent(newValue.getUUIDKey(), newValue);
		newValue.setUuidChanger((oldUUID, newUUID) -> {
			AbstractMetadataRecord.UUIDKey oldKey = new AbstractMetadataRecord.UUIDKey(oldUUID);
			AbstractMetadataRecord record = uuidHolder.remove(oldKey);
			if (test)
				uuidHolder2.remove(oldKey);
			AbstractMetadataRecord.UUIDKey newKey = new AbstractMetadataRecord.UUIDKey(newUUID);
			if (record != null)
				((ConcurrentSkipListMap)uuidHolder).putIfAbsent(newKey, record);
			if (test)
				((ConcurrentSkipListMap)uuidHolder2).putIfAbsent(newKey, record);
		});
		putODataRecord(newValue);
	}

	public void putODataRecord(AbstractMetadataRecord record) {
		oDataMetadata.putRecord(record);
	}

	public void removeODataProcedureRecord(String schemaName, String procName) {
		oDataMetadata.removeProcedure(schemaName, procName);
	}

	public ODataSchemaBuilder.ODataSchemaInfo buildODataStructuralMetadata(Schema schema, ODataMetadata.SecurityChecker securityChecker) {
		return oDataMetadata.build(schema, securityChecker);
	}

	public static List initList(Integer initialCapacity) {
		List customList = (List) DVCaller.call(181, initialCapacity);
		return customList != null ? customList : (initialCapacity != null ? new ArrayList<>(initialCapacity) : new ArrayList());
	}

	public static List reInitList(List<? extends AbstractMetadataRecord> oldC, List<? extends AbstractMetadataRecord> newC) {
		List customList = (List) DVCaller.call(182, oldC, newC);
		return customList != null ? customList : newC;
	}

	public static NavigableMap initMap(Comparator<String> comparator) {
		NavigableMap customMap = (NavigableMap) DVCaller.call(183, comparator);
		return customMap != null ? customMap : new ConcurrentSkipListMap<>(comparator);
	}

	public static AbstractMetadataRecord putIntoMap(AbstractMetadataRecord oldValue, AbstractMetadataRecord newValue) {
		return (AbstractMetadataRecord) DVCaller.call(184, oldValue, newValue);
	}

	private void removeTableChildren(Table table) {
		List<Column> cols = table.getColumns();
		if (cols != null) {
			for (Column column : cols) {
				removeRecord(column);
			}
		}
		for (KeyRecord record : table.getAllKeys()) {
			removeRecord(record);
		}
	}

	private void removeProcChildren(Procedure procedure) {
		List<ProcedureParameter> params = procedure.getParameters();
		if (params != null) {
			for (ProcedureParameter param : params) {
				removeRecord(param);
			}
		}
		ColumnSet<Procedure> columnSet = procedure.getResultSet();
		if (columnSet != null && columnSet.getColumns() != null) {
			for (Column column : columnSet.getColumns()) {
				removeRecord(column);
			}
		}
	}

	private void removeFuncChildren(FunctionMethod function) {
		List<FunctionParameter> params = function.getInputParameters();
		if (params != null) {
			for (FunctionParameter param : params) {
				removeRecord(param);
			}
		}
		if (function.getReturnParameter() != null) {
			removeRecord(function.getReturnParameter());
		}
	}

	private void removeSchemaChildren(Schema schema) {
		for (Table table : schema.getTables().values()) {
			removeRecord(table);
		}
		for (Procedure procedure : schema.getProcedures().values()) {
			removeRecord(procedure);
		}
		for (FunctionMethod function : schema.getFunctions().values()) {
			removeRecord(function);
		}
	}

	@Override
	public DVMetadataCache clone(String vdbName, boolean runCleaner) {
		DVMetadataRecordHolder recordHolder = new DVMetadataRecordHolder(this.oDataMetadata.clone(vdbName), runCleaner,
				new ConcurrentSkipListMap());
		recordHolder.getUuidHolder().putAll(this.uuidHolder);
		return recordHolder;
	}

	public void refreshCache(MetadataStore metadataStore) {
		synchronized (lock) {
			uuidHolder.clear();
			if (test)
				uuidHolder2.clear();
			if (metadataStore != null) {
				for (Datatype datatype : metadataStore.getDatatypes().values()) {
					putRecord(datatype.getUUIDKey(), datatype);
				}
				for (Schema schema : metadataStore.getSchemaList()) {
					for (Table table : schema.getTables().values()) {
						if (table.getColumns() != null) {
							for (Column column : table.getColumns()) {
								putRecord(column.getUUIDKey(), column);
							}
						}
						for (KeyRecord keyRecord : table.getAllKeys()) {
							putRecord(keyRecord.getUUIDKey(), keyRecord);
						}
						putRecord(table.getUUIDKey(), table);
					}
					for (Procedure proc : schema.getProcedures().values()) {
						for (ProcedureParameter parameter : proc.getParameters()) {
							putRecord(parameter.getUUIDKey(), parameter);
						}
						if (proc.getResultSet() != null) {
							for (Column column : proc.getResultSet().getColumns()) {
								putRecord(column.getUUIDKey(), column);
							}
						}
						putRecord(proc.getUUIDKey(), proc);
					}
					for (FunctionMethod function : schema.getFunctions().values()) {
						for (FunctionParameter parameter : function.getInputParameters()) {
							putRecord(parameter.getUUIDKey(), parameter);
						}
						if (function.getReturnParameter() != null) {
							putRecord(function.getReturnParameter().getUUIDKey(), function.getReturnParameter());
						}
						putRecord(function.getUUIDKey(), function);
					}
					putRecord(schema.getUUIDKey(), schema);
				}
			}
		}
	}

	private void putRecord(AbstractMetadataRecord.UUIDKey uuidKey, AbstractMetadataRecord record) {
		uuidHolder.put(uuidKey, record);
	}

	public static class UUIDHolderCleaner extends TimerTask {
		private DVMetadataRecordHolder recordHolder;

		public UUIDHolderCleaner(DVMetadataRecordHolder recordHolder) {
			this.recordHolder = recordHolder;
		}

		@Override
		public void run() {
			if (!recordHolder.isTestMode())
				recordHolder.cleanCache();
		}
	}
}
