/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.apache.openjpa.sdo;


import java.util.Collection;
import java.util.Map;
import java.util.Set;

import org.apache.openjpa.kernel.Broker;
import org.apache.openjpa.kernel.Query;
import org.apache.openjpa.kernel.jpql.JPQLParser;
import org.apache.openjpa.persistence.EntityManagerImpl;
import org.apache.openjpa.persistence.OpenJPAQuery;

import commonj.sdo.DataObject;
import commonj.sdo.Type;

/**
 * A specialized EntityManager that accepts SDO DataObject as 
 * arguments for persistence operation.
 * <br>
 * If argument to the persistent operations is DataObject, then this receiver
 * converts them to strongly-typed Java objects, otherwise acts as a mere
 * pass-through to the base function.
 * <br>
 * For query operation, always converts the returned list of objects as 
 * DataObjects before returning to the caller.
 * <br>
 * Currently supports persist(), merge(), find(), remove() and createQuery().
 * 
 * @author Pinaki Poddar
 * @since 0.1.0
 */
public class SDOEntityManager extends EntityManagerImpl {
	// TODO: Change to a smarter cache
	private BiDiMap<DataObject, Object> cache = new BiDiMap<DataObject, Object>();
	
	private SDOMappingRepository repos;
		
	public SDOEntityManager(SDOEntityManagerFactory emf, Broker broker) {
		super(emf, broker);
		repos = ((SDOMetaDataFactory)emf.getConfiguration().
				getMetaDataRepositoryInstance().
				getMetaDataFactory()).
				getSDOMappingRepository();
	}
	
	/**
	 * Finds if the given DataObject is cached, if <code>type</code> equals
	 * <code>DataObject.class</code>, else functions normally for
	 * for strongly-typed Java objects.
	 * 
	 * @param cls if <code>DataObject.class</code> then the <code>id</code>
	 * parameter has to be prefixed with SDO Type name.
	 * <br>
	 * <code> em.find(DataObject.class, "Customer:1234") will return a Customer
	 * DataObject.
	 * @param oid of the identity property prefixed with SDO Type name.
	 * 
	 * This is bit flaky, because to comply with JPA API method signature.
	 */
	@Override
	public <T> T find(Class<T> cls, Object oid) {
		if (cls==DataObject.class) {
			Type type = extractType(oid.toString());
			String id = extractId(oid.toString());
			DataObject data = scanLocalCache(type, id);
			return (T)data;
		} 
		return super.find(cls, oid);
	}
	
	/**
	 * Persist the given entity. If the entity is a DataObject instance then the
	 * closure of the entity i.e. all related entities are persisted too. 
	 */
	@Override
	public void persist(Object entity) {
		if (entity instanceof DataObject) {
			SDOInstanceMapping ctx = new SDOInstanceMapping(this, repos);
			DataObject data = (DataObject)entity;
			ctx.populatePC(data);
			Collection closure = ctx.getClosure().values();
			super.persistAll(closure);
			return;
		}  
		super.persist(entity);
	}
	
	/**
	 * Merges the given entity to the database.
	 */
    @SuppressWarnings("unchecked")
	@Override
    public <T> T merge(T entity) {
    	if (entity instanceof DataObject) {
    		DataObject data = (DataObject)entity;
			SDOInstanceMapping ctx = new SDOInstanceMapping(this, repos);
			ctx.populatePC(data);
    		Map<DataObject, Object> workSet = ctx.getClosure();
    		super.mergeAll(workSet.values());
    		// TODO: This is not done, but how can we get the new version
    		for (DataObject d:workSet.keySet()) {
    			Object original = workSet.get(d);
				//TODO: Version is always integer
    			int newVersion = ((Integer)getVersion(original)).intValue() + 1;
    			ImplHelper.setVersion(d, newVersion);
    		}
    		return (T)data;
    	}
        return super.merge(entity);
    }
    
    /**
     * Removes the DataObject and all DataObjects reachable from it.
     */
    @Override
    public void remove(Object entity) {
    	if (entity instanceof DataObject) {
			DataObject data = (DataObject)entity;
			SDOInstanceMapping ctx = new SDOInstanceMapping(this, repos);
			ctx.populatePC(data);
			Collection closure = ctx.getClosure().values();
			super.removeAll(super.mergeAll(closure));
			return;
    	} 
    	super.remove(entity);
    }

    /**
     * Creates a Query using JPQL that will return list of DataObject as result. 
     */
	@Override
	public OpenJPAQuery createQuery(String jpql) {
		Query kernelQuery = getBroker().newQuery(JPQLParser.LANG_JPQL, jpql);
        return new SDOQuery(this, kernelQuery); 
	}

	Object find(DataObject data) {
		return cache.get(data);
	}
	
	DataObject createDataObject(Object pc) {
		SDOInstanceMapping ctx = new SDOInstanceMapping(this, repos);
		return ctx.populateDataObject(pc);
	}
	
	/**
	 * Scans local cache for a DataObject of given type and stringified 
	 * identity.
	 */
	DataObject scanLocalCache(Type type, String id) {
		Set dataObjects = cache.keySet();
		for (Object d:dataObjects) {
			DataObject data = (DataObject)d;
			if (data.getType().equals(type)) {
				Object idValue = ImplHelper.getIdValue(data);
				if (id.equals(idValue.toString()))
					return data;
			}
		}
		return null;
	}
	
	protected Type extractType(String s) {
		int colon = s.indexOf(':');
		return (colon == -1) ? null : repos.getType(s.substring(0,colon)); 
	}
	
	protected String extractId(String s) {
		int colon = s.indexOf(':');
		return (colon == -1) ? "" : s.substring(colon+1); 
	}
	
	public Object findCached(DataObject data) {
		return cache.get(data);
	}
	
	public DataObject findCached(Object pc) {
		return (DataObject)cache.getReverse(pc);
	}
	
	public void cache(DataObject data, Object pc) {
		cache.put(data, pc);
	}
}