/*
 * JBoss, Home of Professional Open Source.
 * See the COPYRIGHT.txt file distributed with this work for information
 * regarding copyright ownership.  Some portions may be licensed
 * to Red Hat, Inc. under one or more contributor license agreements.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301 USA.
 * Modified by
 * @author  Data Virtuality GmbH
 * @date    $Date$
 *
 */
package org.teiid.query.metadata;

import com.datavirtuality.dv.core.interfaces.concurrent.ObservableLock;
import com.google.common.collect.ForwardingNavigableMap;
import org.jboss.logging.Logger;
import org.teiid.logging.LogConstants;
import org.teiid.logging.LogManager;
import org.teiid.metadata.MetadataFactory;
import org.teiid.metadata.MetadataRepository;
import org.teiid.metadata.Schema;
import org.teiid.metadata.Table;
import org.teiid.query.QueryPlugin;
import org.teiid.resource.spi.WrappedConnection;
import org.teiid.translator.ExecutionFactory;
import org.teiid.translator.TranslatorException;

import javax.resource.ResourceException;
import java.util.Iterator;
import java.util.NavigableMap;
import java.util.concurrent.ConcurrentHashMap;

@SuppressWarnings("rawtypes")
public class NativeMetadataRepository extends MetadataRepository {
    private static final Logger logger = Logger.getLogger(NativeMetadataRepository.class);

    @Override
	public void loadMetadata(MetadataFactory factory, ExecutionFactory executionFactory, Object connectionFactory) throws TranslatorException {
		if (executionFactory == null) {
			throw new TranslatorException(QueryPlugin.Util.gs(QueryPlugin.Event.TEIID30591, factory.getName()));
		}

		if (connectionFactory == null && executionFactory.isSourceRequiredForMetadata()) {
			throw new TranslatorException(QueryPlugin.Event.TEIID31097, QueryPlugin.Util.gs(QueryPlugin.Event.TEIID31097));
		}

		if (System.getProperty("dv.embedded") == null) {
			ClassLoader originalCL = Thread.currentThread().getContextClassLoader();
			try {
				Thread.currentThread().setContextClassLoader(executionFactory.getClass().getClassLoader());
				getMetadata(factory, executionFactory, connectionFactory);
			} finally {
				Thread.currentThread().setContextClassLoader(originalCL);
			}
		} else {
			getMetadata(factory, executionFactory, connectionFactory);
		}
	}

    private void getMetadata(MetadataFactory factory, ExecutionFactory executionFactory, Object connectionFactory)
			throws TranslatorException {
        if (factory.isLazyMetadataLoading()) {
            final var name = factory.getSchema().getName();
            factory.setSchema(new SchemaProxy(factory, executionFactory, connectionFactory));
            factory.getSchema().setName(name);
            return;
        }
        getActualMetadata(factory, executionFactory, connectionFactory);
    }

    @SuppressWarnings("unchecked")
    private void getActualMetadata(MetadataFactory factory, ExecutionFactory executionFactory, Object connectionFactory)
            throws TranslatorException {
        Object connection = executionFactory.getConnection(connectionFactory, null);
        try {
            Object unwrapped = null;

            if (connection instanceof WrappedConnection) {
                try {
                    unwrapped = ((WrappedConnection)connection).unwrap();
                } catch (ResourceException e) {
                    if (executionFactory.isSourceRequiredForMetadata()) {
                        throw new TranslatorException(QueryPlugin.Event.TEIID30477, QueryPlugin.Util.gs(QueryPlugin.Event.TEIID30477));
                    }
                    connection = null;
                }
            }
			executionFactory.getMetadata(factory, (unwrapped == null) ? connection:unwrapped);
		} finally {
			executionFactory.closeConnection(connection, connectionFactory);
		}
		validateMetadata(factory);
	}

	private void validateMetadata(MetadataFactory schema) {
		for (Iterator<Table> it = schema.getSchema().getTables().values().iterator(); it.hasNext();) {
			Table t = it.next();
			if ((t.getColumns() == null || t.getColumns().isEmpty()) && t.getTableType() != Table.Type.View) {
				it.remove();
				LogManager.logWarning(LogConstants.CTX_RUNTIME, "Can't load this table because it has no columns: " + t.getFullName());
			}
		}
	}

    private class SchemaProxy extends Schema {
        private final MetadataFactory factory;
        private final ExecutionFactory executionFactory;
        private final Object connectionFactory;
        private final ConcurrentHashMap<Object, ObservableLock> locks = new ConcurrentHashMap<>();


        public SchemaProxy(MetadataFactory factory, ExecutionFactory executionFactory, Object connectionFactory) {
            this.factory = factory;
            this.executionFactory = executionFactory;
            this.connectionFactory = connectionFactory;
            final var oldMap = this.tables;
            this.tables = new ForwardingNavigableMap<>() {
                @Override
                protected NavigableMap<String, Table> delegate() {
                    return oldMap;
                }

                @Override
                public Table get(Object key) {
                    Table table = super.get(key);
                    if (table != null)
                        return table;
                    ObservableLock lock = locks.computeIfAbsent(key, k -> new ObservableLock());
                    lock.lock();
                    try {
                        Table table1 = super.get(key);
                        if (table1 != null)
                            return table1;
                        table = loadTable((String) key);
                        if (table == null) return null;
                        addTable(table);
                        return table;
                    } finally {
                        lock.unlock();
                    }
                }
            };
        }

        @Override
        public boolean isLazy() {
            return true;
        }

        @Override
        public Table removeTable(String tableName) {
            ObservableLock lock = locks.computeIfAbsent(tableName, k -> new ObservableLock());
            lock.lock();
            try {
                return super.removeTable(tableName);
            } finally {
                lock.unlock();
                if (!lock.hasQueuedThreads() && !lock.isLocked()) {
                    locks.remove(tableName, lock);
                }
            }
        }

        private Table loadTable(String tableName) {
            Table table;
            var nf = factory.clone();
            nf.getModelProperties().setProperty("importer.tableNamePattern", tableName);
            nf.getSchema().setName(factory.getSchema().getName());
            try {
                getActualMetadata(nf, executionFactory, connectionFactory);
            } catch (TranslatorException e) {
                logger.error("Error lazily loading metadata for table " + tableName, e);
                return null;
            }
            table = nf.getSchema().getTable(tableName);
            return table;
        }
    }
}
