/*
 * Copyright (c) 2011, 2019 Oracle and/or its affiliates. All rights reserved.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License v. 2.0 which is available at
 * http://www.eclipse.org/legal/epl-2.0,
 * or the Eclipse Distribution License v. 1.0 which is available at
 * http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
 */

// Contributors:
//     James Sutherland - initial API and implementation
package org.eclipse.persistence.platform.database.oracle.dcn;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.eclipse.persistence.annotations.DatabaseChangeNotificationType;
import org.eclipse.persistence.descriptors.CacheIndex;
import org.eclipse.persistence.descriptors.ClassDescriptor;
import org.eclipse.persistence.descriptors.DescriptorEvent;
import org.eclipse.persistence.descriptors.DescriptorEventAdapter;
import org.eclipse.persistence.exceptions.DatabaseException;
import org.eclipse.persistence.internal.databaseaccess.Accessor;
import org.eclipse.persistence.internal.expressions.SQLSelectStatement;
import org.eclipse.persistence.internal.helper.DatabaseField;
import org.eclipse.persistence.internal.helper.DatabaseTable;
import org.eclipse.persistence.internal.identitymaps.CacheId;
import org.eclipse.persistence.internal.identitymaps.CacheKey;
import org.eclipse.persistence.internal.sessions.AbstractRecord;
import org.eclipse.persistence.internal.sessions.AbstractSession;
import org.eclipse.persistence.logging.SessionLog;
import org.eclipse.persistence.mappings.DatabaseMapping;
import org.eclipse.persistence.platform.database.events.DatabaseEventListener;
import org.eclipse.persistence.queries.ValueReadQuery;
import org.eclipse.persistence.sessions.Session;

import oracle.jdbc.OracleConnection;
import oracle.jdbc.OracleStatement;
import oracle.jdbc.dcn.DatabaseChangeEvent;
import oracle.jdbc.dcn.DatabaseChangeListener;
import oracle.jdbc.dcn.DatabaseChangeRegistration;
import oracle.jdbc.dcn.RowChangeDescription;
import oracle.jdbc.dcn.TableChangeDescription;

/**
 * PUBLIC:
 * Listener for Oracle Database Change event Notification (DCN).
 * This allows the EclipseLink cache to be invalidated by database events.
 *
 * @author James Sutherland
 * @since EclipseLink 2.4
 */
public class OracleChangeNotificationListener implements DatabaseEventListener {

    public static String ORA_TRANSACTION_ID = "oracle.dcn.transaction-id";
    public static String ROWID = "ROWID";

    /** The Oracle JDBC registration object. */
    protected DatabaseChangeRegistration register;
    /** Map each table to the descriptor that needs to be invalidated. */
    protected Map<DatabaseTable, ClassDescriptor> descriptorsByTable;

    /** Cache query for transaction id. */
    protected ValueReadQuery transactionIdQuery;

    public OracleChangeNotificationListener() {
        this.transactionIdQuery = new ValueReadQuery("SELECT DBMS_TRANSACTION.LOCAL_TRANSACTION_ID FROM DUAL");
        transactionIdQuery.setName(ORA_TRANSACTION_ID);
    }

    /**
     * INTERNAL:
     * Register the event listener with the database.
     */
    @Override
    public void register(Session session) {
        final AbstractSession databaseSession = (AbstractSession)session;
        // Determine which tables should be tracked for change events.
        this.descriptorsByTable = new HashMap<DatabaseTable, ClassDescriptor>();
        for (ClassDescriptor descriptor : session.getDescriptors().values()) {
            if (!descriptor.getTables().isEmpty()) {
                if ((descriptor.getCachePolicy().getDatabaseChangeNotificationType() != null)
                            && (descriptor.getCachePolicy().getDatabaseChangeNotificationType() != DatabaseChangeNotificationType.NONE)) {
                    this.descriptorsByTable.put(descriptor.getTables().get(0), descriptor);
                }
            }
        }
        Accessor accessor = databaseSession.getAccessor();
        accessor.incrementCallCount(databaseSession);
        try {
            OracleConnection connection = (OracleConnection)databaseSession.getServerPlatform().unwrapConnection(accessor.getConnection());
            databaseSession.log(SessionLog.FINEST, SessionLog.CONNECTION, "dcn_registering");
            Properties properties = new Properties();
            properties.setProperty(OracleConnection.DCN_NOTIFY_ROWIDS, "true");
            properties.setProperty(OracleConnection.DCN_IGNORE_INSERTOP, "true");
            try {
                // Register with the database change notification, the connection is not relevant, the events occur after the connection is closed,
                // and a different connection can be used to unregister the event listener.
                this.register = connection.registerDatabaseChangeNotification(properties);
                final List<DatabaseField> fields = new ArrayList<DatabaseField>();
                fields.add(new DatabaseField(ROWID));
                this.register.addListener(new DatabaseChangeListener() {
                    @Override
                    public void onDatabaseChangeNotification(DatabaseChangeEvent changeEvent) {
                        databaseSession.log(SessionLog.FINEST, SessionLog.CONNECTION, "dcn_change_event", changeEvent);
                        if (changeEvent.getTableChangeDescription() != null) {
                            for (TableChangeDescription tableChange : changeEvent.getTableChangeDescription()) {
                                ClassDescriptor descriptor = OracleChangeNotificationListener.this.descriptorsByTable.get(new DatabaseTable(tableChange.getTableName()));
                                if (descriptor != null) {
                                    CacheIndex index = descriptor.getCachePolicy().getCacheIndex(fields);
                                    for (RowChangeDescription rowChange : tableChange.getRowChangeDescription()) {
                                        CacheId id = new CacheId(new Object[]{rowChange.getRowid().stringValue()});
                                        CacheKey key = databaseSession.getIdentityMapAccessorInstance().getIdentityMapManager().getCacheKeyByIndex(
                                                index, id, true, descriptor);
                                        if (key != null) {
                                            if ((key.getTransactionId() == null) || !key.getTransactionId().equals(changeEvent.getTransactionId(true))) {
                                                databaseSession.log(SessionLog.FINEST, SessionLog.CONNECTION, "dcn_invalidate", key.getKey(), descriptor.getJavaClass().getName());
                                                key.setInvalidationState(CacheKey.CACHE_KEY_INVALID);
                                            }
                                        }
                                    }
                                }
                            }
                        }
                    }
                });
                // Register each table for database events, this is done by executing a select from the table.
                for (DatabaseTable table : this.descriptorsByTable.keySet()) {
                    OracleStatement statement = (OracleStatement)connection.createStatement();
                    statement.setDatabaseChangeRegistration(this.register);
                    try {
                        statement.executeQuery("SELECT ROWID FROM " + table.getQualifiedName()).close();
                        databaseSession.log(SessionLog.FINEST, SessionLog.CONNECTION, "dcn_register_table", table.getQualifiedName());
                    } catch (Exception failed) {
                        // This will fail if the table does not exist,
                        // just log the error to allow table creation to work.
                        databaseSession.logThrowable(SessionLog.WARNING, SessionLog.SQL, failed);
                    } finally {
                        statement.close();
                    }
                }
            } catch (SQLException exception) {
                throw DatabaseException.sqlException(exception, databaseSession.getAccessor(), databaseSession, false);
            }
        } finally {
            accessor.decrementCallCount();
        }
    }

    /**
     * Initialize the descriptor to receive database change events.
     * This is called when the descriptor is initialized.
     */
    @Override
    public void initialize(final ClassDescriptor descriptor, AbstractSession session) {
        if (descriptor.getOptimisticLockingPolicy() == null) {
            boolean requiresLocking = descriptor.hasMultipleTables();
            for (DatabaseMapping mapping : descriptor.getMappings()) {
                if (mapping.isCollectionMapping()) {
                    requiresLocking = true;
                }
            }
            if (requiresLocking) {
                session.log(SessionLog.WARNING, SessionLog.METADATA, "locking_required_for_database_change_notification", descriptor.getJavaClass());
            }
        }
        final DatabaseField rowId = descriptor.buildField(new DatabaseField(ROWID));
        final List<DatabaseField> fields = new ArrayList<DatabaseField>();
        fields.add(rowId);
        // May already have the index if has inheritance.
        CacheIndex existingIndex = descriptor.getCachePolicy().getCacheIndex(fields);
        if (existingIndex == null) {
            if (descriptor.isChildDescriptor()) {
                existingIndex = descriptor.getInheritancePolicy().getRootParentDescriptor().getCachePolicy().getCacheIndex(fields);
            }
            if (existingIndex == null) {
                existingIndex = new CacheIndex(fields);
                existingIndex.setIsUpdateable(false);
                existingIndex.setIsInsertable(false);
            }
            descriptor.getCachePolicy().addCacheIndex(existingIndex);
        }

        final CacheIndex index = existingIndex;
        rowId.setInsertable(false);
        rowId.setUpdatable(false);
        rowId.setCreatable(false);
        descriptor.getFields().add(rowId);
        descriptor.getAllFields().add(rowId);

        final ValueReadQuery rowIdQuery = new ValueReadQuery();
        rowIdQuery.setName(ROWID);
        SQLSelectStatement sqlStatement = new SQLSelectStatement();
        sqlStatement.setWhereClause(descriptor.getObjectBuilder().getPrimaryKeyExpression());
        sqlStatement.addField(rowId);
        sqlStatement.addTable(descriptor.getTables().get(0));
        rowIdQuery.setSQLStatement(sqlStatement);
        sqlStatement.normalize(session, null);

        descriptor.getEventManager().addListener(new DescriptorEventAdapter() {
            @Override
            public void postMerge(DescriptorEvent event) {
                if ((event.getChangeSet() != null) && event.getChangeSet().hasChanges()) {
                    Object id = event.getChangeSet().getId();
                    CacheKey cacheKey = event.getChangeSet().getActiveCacheKey();
                    if (cacheKey == null) {
                        cacheKey = event.getSession().getParent().getIdentityMapAccessorInstance().getIdentityMapManager().getCacheKeyForObject(id, descriptor.getJavaClass(), descriptor, false);
                    }
                    cacheKey.setTransactionId(event.getSession().getProperty(ORA_TRANSACTION_ID));
                    if (event.getChangeSet().isNew()) {
                        AbstractRecord row = descriptor.getObjectBuilder().buildRowFromPrimaryKeyValues(id, event.getSession());
                        Object rowid = event.getSession().executeQuery(rowIdQuery, row);
                        CacheId indexValue = new CacheId(new Object[]{rowid});
                        event.getSession().getParent().getIdentityMapAccessorInstance().getIdentityMapManager().putCacheKeyByIndex(index, indexValue, cacheKey, descriptor);
                    }
                }
            }
            @Override
            public void postUpdate(DescriptorEvent event) {
                Object txId = event.getSession().getProperty(ORA_TRANSACTION_ID);
                if (txId == null) {
                    txId = event.getSession().executeQuery(transactionIdQuery);
                    event.getSession().setProperty(ORA_TRANSACTION_ID, txId);
                }
            }
        });
    }

    /**
     * INTERNAL:
     * Remove the event listener from the database.
     */
    @Override
    public void remove(Session session) {
        if (this.register == null) {
            return;
        }
        AbstractSession databaseSession = (AbstractSession)session;
        Accessor accessor = databaseSession.getAccessor();
        accessor.incrementCallCount(databaseSession);
        try {
            OracleConnection connection = (OracleConnection)databaseSession.getServerPlatform().unwrapConnection(accessor.getConnection());
            databaseSession.log(SessionLog.FINEST, SessionLog.CONNECTION, "dcn_unregister");
            try {
                connection.unregisterDatabaseChangeNotification(this.register);
            } catch (SQLException exception) {
                throw DatabaseException.sqlException(exception, databaseSession.getAccessor(), databaseSession, false);
            }
        } finally {
            accessor.decrementCallCount();
        }
    }

    /**
     * INTERNAL:
     * Return the database register.
     */
    public DatabaseChangeRegistration getRegister() {
        return register;
    }

    /**
     * INTERNAL:
     * Set the database register.
     */
    protected void setRegister(DatabaseChangeRegistration register) {
        this.register = register;
    }

    /**
     * INTERNAL:
     * Return the mapping of tables to descriptors.
     */
    public Map<DatabaseTable, ClassDescriptor> getDescriptorsByTable() {
        return descriptorsByTable;
    }

    /**
     * INTERNAL:
     * Set the mapping of tables to descriptors.
     */
    protected void setDescriptorsByTable(Map<DatabaseTable, ClassDescriptor> descriptorsByTable) {
        this.descriptorsByTable = descriptorsByTable;
    }

}