/*
 * $Header: /home/cvspublic/jakarta-slide/src/stores/org/apache/slide/store/ojb/OJBStore.java,v 1.4 2005/01/23 13:57:27 cvillegas Exp $
 * $Revision: 1.4 $
 * $Date: 2005/01/23 13:57:27 $
 *
 * ====================================================================
 *
 * Copyright 1999-2002 The Apache Software Foundation
 *
 * 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
 *
 *     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.slide.store.ojb;

import java.lang.reflect.Constructor;
import java.sql.SQLException;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.Set;
import java.util.Vector;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;

import org.apache.slide.common.ServiceAccessException;
import org.apache.slide.common.Uri;
import org.apache.slide.content.NodeProperty;
import org.apache.slide.content.NodeRevisionContent;
import org.apache.slide.content.NodeRevisionDescriptor;
import org.apache.slide.content.NodeRevisionDescriptors;
import org.apache.slide.content.NodeRevisionNumber;
import org.apache.slide.content.RevisionAlreadyExistException;
import org.apache.slide.content.RevisionDescriptorNotFoundException;
import org.apache.slide.content.RevisionNotFoundException;
import org.apache.slide.lock.LockTokenNotFoundException;
import org.apache.slide.lock.NodeLock;
import org.apache.slide.security.NodePermission;
import org.apache.slide.structure.LinkNode;
import org.apache.slide.structure.ObjectAlreadyExistsException;
import org.apache.slide.structure.ObjectNode;
import org.apache.slide.structure.ObjectNotFoundException;

import org.apache.ojb.broker.PersistenceBroker;
import org.apache.ojb.broker.PersistenceBrokerException;
import org.apache.ojb.broker.query.*;

import org.apache.slide.store.ConcurrencyConflictError;
import org.apache.slide.store.ojb.peer.*;
import org.apache.slide.store.ojb.property.PropertyHandler;
import org.apache.slide.store.ojb.property.PropertyHandlerFactory;
import org.apache.slide.store.ojb.property.PropertyHandlerException;
import org.apache.slide.util.logger.Logger;

/**
 *
 */
public class OJBStore extends AbstractOJBStore {

    // ==== NodeStore Methods ================================
    
    protected boolean storeObject(PersistenceBroker broker, Uri uri,
            ObjectNode object, boolean create) throws ServiceAccessException {
        String className = object.getClass().getName();
        UriPeer uriPeer;
		Set updatedBindings = object.getUpdatedBindings();
        try {
            Criteria crit = new Criteria();
            crit.addEqualTo("uri.uriString", uri.toString());
            QueryByCriteria query = QueryFactory.newQuery(ObjectPeer.class, crit);
            query.addPrefetchedRelationship("uri");
            ObjectPeer objectPeer = (ObjectPeer)broker.getObjectByQuery(query);
            if ( objectPeer != null ) {
                if (create)
                    return false;
            } else {
                if (!create)
                    return false;
            }

            if (create) {
                // create object in database
                uriPeer = assureUriPeer(broker, uri.toString());
                objectPeer = new ObjectPeer();                
                objectPeer.setUriId(uriPeer.getUriId());
                objectPeer.setClassName(className);
                broker.store(objectPeer);
            } else {
                uriPeer = objectPeer.getUri();
            }

            // update binding...

            clearBinding(broker, uri, updatedBindings);
            Enumeration bindings = object.enumerateBindings();
            while (bindings.hasMoreElements()) {
                ObjectNode.Binding binding = (ObjectNode.Binding) bindings.nextElement();
                //Only insert the binding if it has been updated
                if (updatedBindings.contains(binding.getUuri())) {
                    UriPeer childUriPeer = findUriPeer(broker, binding.getUuri());
                    if ( childUriPeer != null ) {
                        BindingPeer bindingPeer = new BindingPeer();
                        bindingPeer.setUriId(uriPeer.getUriId());
                        bindingPeer.setName(binding.getName());
                        bindingPeer.setChildUuriId(childUriPeer.getUriId());
                        broker.store(bindingPeer);
                    }
                }
            }

            Enumeration parentBindings = object.enumerateParentBindings();
            while (parentBindings.hasMoreElements()) {
                ObjectNode.ParentBinding parentBinding = (ObjectNode.ParentBinding) parentBindings.nextElement();
                UriPeer parentUriPeer = findUriPeer(broker, parentBinding.getUuri());
                if (parentUriPeer != null) {
                    ParentBindingPeer parentBindingPeer = new ParentBindingPeer();
                    parentBindingPeer.setUriId(uriPeer.getUriId());
                    parentBindingPeer.setName(parentBinding.getName());
                    parentBindingPeer.setParentUuriId(parentUriPeer.getUriId());
                    broker.store(parentBindingPeer);
                }
            }

            if (object instanceof LinkNode) {
                // update link target
                crit = new Criteria();
                crit.addEqualTo("uriId", new Long(uriPeer.getUriId()));
                query = QueryFactory.newQuery(LinkPeer.class, crit);
                broker.deleteByQuery(query);
                broker.clearCache();
                UriPeer linkedUriPeer = findUriPeer(broker, ((LinkNode)object).getLinkedUri());
                if ( linkedUriPeer != null ) {
                    LinkPeer linkPeer = new LinkPeer();
                    linkPeer.setUriId(uriPeer.getUriId());
                    linkPeer.setLinkToId(linkedUriPeer.getUriId());
                    broker.store(linkPeer);
                }
            }
        } catch (PersistenceBrokerException e) {
            throw createException(e, uri.toString());
        }
        object.resetUpdatedBindings();
        return true;
    }

    public void storeObject(Uri uri, ObjectNode object)
            throws ServiceAccessException, ObjectNotFoundException {
      
        if (!storeObject(getCurrentBroker(), uri, object, false))
            throw new ObjectNotFoundException(uri.toString());
    }

    public void createObject(Uri uri, ObjectNode object)
            throws ServiceAccessException, ObjectAlreadyExistsException {

        if (!storeObject(getCurrentBroker(), uri, object, true))
            throw new ObjectAlreadyExistsException(uri.toString());
    }

    public void removeObject(Uri uri, ObjectNode object)
            throws ServiceAccessException, ObjectNotFoundException {

        try {
            PersistenceBroker broker = getCurrentBroker();
            clearBinding(broker, uri);
            
            UriPeer uriPeer = findUriPeer(broker, uri.toString());
            if ( uriPeer != null ) {
                // delete links
                Criteria crit = new Criteria();
                crit.addEqualTo("uriId", new Long(uriPeer.getUriId()));
                Query query = QueryFactory.newQuery(LinkPeer.class, crit);
                broker.deleteByQuery(query);
                // delete version history
                // FIXME: Is this true??? Should the version history be removed if the object is removed???               
                query = QueryFactory.newQuery(VersionHistoryPeer.class, crit);
                broker.deleteByQuery(query);
                // delete version
                query = QueryFactory.newQuery(VersionPeer.class, crit);
                broker.deleteByQuery(query);
                // delete object itself
                query = QueryFactory.newQuery(ObjectPeer.class, crit);
                broker.deleteByQuery(query);
                // finally delete the uri
                broker.delete(uriPeer);
                broker.clearCache();
            } else
                throw new ObjectNotFoundException(uri);
        } catch (PersistenceBrokerException e) {
            throw createException(e, uri.toString());
        }
        object.resetUpdatedBindings();
    }
    
    public ObjectNode retrieveObject(PersistenceBroker broker, Uri uri) 
    		throws ServiceAccessException, ObjectNotFoundException {

        ObjectNode result = null;
        try {
            Vector children = new Vector();
            Vector parents = new Vector();
            Vector links = new Vector();
            UriPeer uriPeer = findUriPeer(broker, uri.toString());
            if ( uriPeer == null ) {
                throw new ObjectNotFoundException(uri);
            }
            Criteria crit = new Criteria();
            crit.addEqualTo("uriId", new Long(uriPeer.getUriId()));
            QueryByCriteria query = QueryFactory.newQuery(ObjectPeer.class, crit);
            query.addPrefetchedRelationship("bindings");
            query.addPrefetchedRelationship("parentBindings");
            query.addPrefetchedRelationship("links");
            ObjectPeer objectPeer = (ObjectPeer)broker.getObjectByQuery(query);
            if ( objectPeer == null )
                throw new ObjectNotFoundException(uri);
            for(Iterator c=objectPeer.getBindings().iterator();c.hasNext();) {
                BindingPeer bindingPeer = (BindingPeer)c.next();
                children.add(new ObjectNode.Binding(bindingPeer.getName(),
                        bindingPeer.getChildUuri().getUriString()));
            }
            for(Iterator p=objectPeer.getParentBindings().iterator();p.hasNext();) {
                ParentBindingPeer parentBindingPeer = (ParentBindingPeer)p.next();
                parents.add(new ObjectNode.ParentBinding(parentBindingPeer.getName(),
                        parentBindingPeer.getParentUuri().getUriString()));
            }
            for(Iterator l=objectPeer.getLinks().iterator();l.hasNext();) {
                UriPeer linkUriPeer = (UriPeer)l.next();
                links.add(linkUriPeer.getUriString());
            }
            
            if (objectPeer.getClassName().equals(LinkNode.class.getName())) {
                crit = new Criteria();
                crit.addEqualTo("uriId", new Long(uriPeer.getUriId()));
                query = QueryFactory.newQuery(LinkPeer.class, crit);
                LinkPeer targetLinkPeer = (LinkPeer)broker.getObjectByQuery(query);
                if ( targetLinkPeer != null )
                    result = new LinkNode(uri.toString(), children, links, 
                            targetLinkPeer.getLinkToUri().getUriString());
                else
                    result = new LinkNode(uri.toString(), children, links);
            } else {
                try {
                    Class objclass = Class.forName(objectPeer.getClassName());
                    Class argClasses[] = { String.class, Vector.class, Vector.class, Vector.class };
                    Object arguments[] = { uri.toString(), children, parents, links };
                    Constructor constructor = objclass.getConstructor(argClasses);
                    result = (ObjectNode) constructor.newInstance(arguments);
                    result.setUri(result.getUuri());
                } catch (Exception e) {
                    throw new ServiceAccessException(this, e);
                }
            }
        } catch (PersistenceBrokerException e) {
            throw createException(e, uri.toString());
        }
        return result;
    }
    
    // Permissions

    public Enumeration enumeratePermissions(PersistenceBroker broker, Uri uri) throws ServiceAccessException {
        Vector permissions = new Vector();
        try {
            UriPeer uriPeer = findUriPeer(broker, uri.toString());
            if ( uriPeer == null )
                return permissions.elements();
            Criteria crit = new Criteria();
            crit.addEqualTo("objectId", new Long(uriPeer.getUriId()));
            QueryByCriteria query = QueryFactory.newQuery(PermissionPeer.class, crit);
            query.addOrderByAscending("succession");
            query.addPrefetchedRelationship("subjectUri");
            query.addPrefetchedRelationship("actionUri");
            Collection permissionPeers = broker.getCollectionByQuery(query);
            for(Iterator p=permissionPeers.iterator(); p.hasNext(); ) {
                PermissionPeer permissionPeer = (PermissionPeer)p.next();
                NodePermission permission = new NodePermission(uriPeer.getUriString(),
                        permissionPeer.getVersionNumber(),
                        permissionPeer.getSubjectUri().getUriString(),
                        permissionPeer.getActionUri().getUriString(),
                        permissionPeer.isInheritable(),
                        permissionPeer.isNegative());
                permissions.add(permission);
            }
        } catch (PersistenceBrokerException e) {
            throw createException(e, uri.toString());
        } catch (Throwable t ) {
            System.out.println("Error enumerating permissions: " + t.getMessage());
            t.printStackTrace();
        }
        return permissions.elements();
    }

    public void grantPermission(Uri uri, NodePermission permission)
        throws ServiceAccessException {

        int succession = 0;

        try {
            // FIXME: This might be useless if insert inserts nothing if insert...select works a i expect
            // FIXME: What happens, if only revision number, inheritable or negative changes?
            // FIXME
            //            revokePermission(connection, uri, permission);

            PersistenceBroker broker = getCurrentBroker();
            Criteria crit = new Criteria();
            crit.addEqualTo("objectUri.uriString", permission.getObjectUri());
            ReportQueryByCriteria query = QueryFactory.newReportQuery(PermissionPeer.class, crit);
            query.setAttributes(new String[] { "max(succession)" });
            Iterator i = broker.getReportQueryIteratorByQuery(query);
            if ( i.hasNext() ) {
                Object[] v = (Object[])i.next();
                if ( v != null && v[0] != null )
                    succession = ((Integer)v[0]).intValue() + 1;
            }
            
            UriPeer objectUriPeer = findUriPeer(broker, permission.getObjectUri());
            UriPeer subjectUriPeer = assureUriPeer(broker, permission.getSubjectUri());
            UriPeer actionUriPeer = assureUriPeer(broker, permission.getActionUri());

            PermissionPeer permissionPeer = new PermissionPeer();
            permissionPeer.setObjectId(objectUriPeer.getUriId());
            permissionPeer.setSubjectId(subjectUriPeer.getUriId());
            permissionPeer.setActionId(actionUriPeer.getUriId());
            permissionPeer.setVersionNumber(getRevisionNumberAsString(permission.getRevisionNumber()));
            permissionPeer.setInheritable(permission.isInheritable());
            permissionPeer.setNegative(permission.isNegative());
            permissionPeer.setSuccession(succession);
            
            broker.store(permissionPeer);
            
        } catch (PersistenceBrokerException e) {
            throw createException(e, uri.toString());
        }
    }

    public void revokePermission(Uri uri, NodePermission permission)
        throws ServiceAccessException {
        
        if (permission == null) return;

        try {
            PersistenceBroker broker = getCurrentBroker();
            
            NodeRevisionNumber revisionNumber = permission.getRevisionNumber();
            Criteria crit = new Criteria();
            crit.addEqualTo("objectUri.uriString", permission.getObjectUri());
            crit.addEqualTo("subjectUri.uriString", permission.getSubjectUri());
            crit.addEqualTo("actionUri.uriString", permission.getActionUri());
            if ( revisionNumber == null )
                crit.addIsNull("versionNumber");
            else
                crit.addEqualTo("versionNumber", revisionNumber.toString());
            Query query = QueryFactory.newQuery(PermissionPeer.class, crit);
            PermissionPeer permissionPeer = (PermissionPeer)broker.getObjectByQuery(query);
            if ( permissionPeer != null )
                broker.delete(permissionPeer);
        } catch (PersistenceBrokerException e) {
            throw createException(e, uri.toString());
        }
    }

    public void revokePermissions(Uri uri) throws ServiceAccessException {

        try {
            PersistenceBroker broker = getCurrentBroker();
            UriPeer uriPeer = findUriPeer(broker, uri.toString());
            if ( uriPeer == null )
                return;
            Criteria crit = new Criteria();
            crit.addEqualTo("objectId", new Long(uriPeer.getUriId()));
            Query query = QueryFactory.newQuery(PermissionPeer.class, crit);
            broker.deleteByQuery(query);
            broker.clearCache();
        } catch (PersistenceBrokerException e) {
            throw createException(e, uri.toString());
        }
    }
    
    // Locks
    
    public Enumeration enumerateLocks(PersistenceBroker broker, Uri uri) throws ServiceAccessException {

        Vector lockVector = new Vector();
        try {
            Criteria crit = new Criteria();
            crit.addEqualTo("objectUri.uriString", uri.toString());
            QueryByCriteria query = QueryFactory.newQuery(LockPeer.class, crit);
            query.addPrefetchedRelationship("lockUri");
            query.addPrefetchedRelationship("subjectUri");
            query.addPrefetchedRelationship("typeUri");
            Iterator l = broker.getIteratorByQuery(query);
            while( l.hasNext() ) {
                LockPeer lockPeer = (LockPeer)l.next();
                NodeLock lock = new NodeLock(lockPeer.getLockUri().getUriString(),
                        uri.toString(),
                        lockPeer.getSubjectUri().getUriString(),
                        lockPeer.getTypeUri().getUriString(),
                        lockPeer.getExpirationDate(),
                        lockPeer.isInheritable(),
                        lockPeer.isExclusive(),
                        lockPeer.getOwner());
                lockVector.add(lock);               
            }
        } catch (PersistenceBrokerException e) {
            throw createException(e, uri.toString());
        }
        return lockVector.elements();
    }
    
    public void killLock(Uri uri, NodeLock lock)
            throws ServiceAccessException, LockTokenNotFoundException {
        removeLock(uri, lock);
    }

    public void putLock(Uri uri, NodeLock lock)
            throws ServiceAccessException {

        try {
            PersistenceBroker broker = getCurrentBroker();
            UriPeer lockUriPeer = assureUriPeer(broker, lock.getLockId());
            UriPeer objectUriPeer = findUriPeer(broker, lock.getObjectUri());
            UriPeer subjectUriPeer = findUriPeer(broker, lock.getSubjectUri());
            UriPeer typeUriPeer = findUriPeer(broker, lock.getTypeUri());
            
            LockPeer lockPeer = new LockPeer();
            lockPeer.setLockId(lockUriPeer.getUriId());
            lockPeer.setObjectId(objectUriPeer.getUriId());
            lockPeer.setSubjectId(subjectUriPeer.getUriId());
            lockPeer.setTypeId(typeUriPeer.getUriId());
            lockPeer.setExpirationDate(lock.getExpirationDate());
            lockPeer.setInheritable(lock.isInheritable());
            lockPeer.setExclusive(lock.isExclusive());
            lockPeer.setOwner(lock.getOwnerInfo());
            
            broker.store(lockPeer);
            
        } catch (PersistenceBrokerException e) {
            throw createException(e, uri.toString());
        }
    }

    public void renewLock(Uri uri, NodeLock lock)
            throws ServiceAccessException, LockTokenNotFoundException {
        try {
            PersistenceBroker broker = getCurrentBroker();
            Criteria crit = new Criteria();
            crit.addEqualTo("lockUri.uriString", lock.getLockId());
            Query query = QueryFactory.newQuery(LockPeer.class, crit);
            LockPeer lockPeer = (LockPeer)broker.getObjectByQuery(query);
            if ( lockPeer != null ) {
                // just overwrite it
                // removeLock(uri, lock); 
                putLock(uri, lock);
            } else {
                throw new LockTokenNotFoundException(lock);
            }
        } catch (PersistenceBrokerException e) {
            throw createException(e, uri.toString());
        }
    }

    public void removeLock(Uri uri, NodeLock lock)
            throws ServiceAccessException, LockTokenNotFoundException {

        try {
            PersistenceBroker broker = getCurrentBroker();
            Criteria crit = new Criteria();
            crit.addEqualTo("lockUri.uriString", lock.getLockId());
            QueryByCriteria query = QueryFactory.newQuery(LockPeer.class, crit);
            query.addPrefetchedRelationship("lockUri");
            LockPeer lockPeer = (LockPeer)broker.getObjectByQuery(query);
            UriPeer uriPeer = lockPeer.getLockUri();
            broker.delete(lockPeer);
            broker.delete(uriPeer);
        } catch (PersistenceBrokerException e) {
            throw createException(e, uri.toString());
        }
    }
    
    // RevisionDescriptors Store

    public NodeRevisionDescriptors retrieveRevisionDescriptors(PersistenceBroker broker, Uri uri) 
    	throws ServiceAccessException, RevisionDescriptorNotFoundException {
        
        NodeRevisionDescriptors revisionDescriptors = null;
        try {
            NodeRevisionNumber initialRevision = new NodeRevisionNumber();
            // FIXME: Some code might be lost: workingRevisions is not filled with values
            Hashtable workingRevisions = new Hashtable();
            Hashtable latestRevisionNumbers = new Hashtable();
            Hashtable branches = new Hashtable();
            Vector allRevisions = new Vector();
            
            UriPeer uriPeer = findUriPeer(broker, uri.toString());
            if ( uriPeer == null )
                throw new RevisionDescriptorNotFoundException(uri.toString());
            Criteria crit = new Criteria();
            crit.addEqualTo("uriId", new Long(uriPeer.getUriId()));
            QueryByCriteria query = QueryFactory.newQuery(VersionPeer.class, crit);
            VersionPeer versionPeer = (VersionPeer)broker.getObjectByQuery(query);
            if ( versionPeer == null )
                throw new RevisionDescriptorNotFoundException(uri.toString());
            
            // retrieve revision numbers
            query = QueryFactory.newQuery(VersionHistoryPeer.class, crit);
            query.addPrefetchedRelationship("branch");
            /*
               To avoid database dependencies we'll do the ordering here.
               It seems that Slide revision numbers are always of the
               form 1.1 In this case, shouldn't be better to have two
               integer columns in the database? 
             */
            Iterator vhi = broker.getIteratorByQuery(query);
            ArrayList res = new ArrayList();
            while ( vhi.hasNext() ) res.add(vhi.next());
            Collections.sort(res, new RevisionNumberComparator());
            for(Iterator i=res.iterator(); i.hasNext(); ) {
                VersionHistoryPeer vh = (VersionHistoryPeer)i.next();
                NodeRevisionNumber revisionNumber = new NodeRevisionNumber(vh.getRevisionNumber());
                allRevisions.add(revisionNumber);
                latestRevisionNumbers.put(vh.getBranch().getBranchString(), revisionNumber);
            }

            for (Enumeration e = allRevisions.elements(); e.hasMoreElements();) {
                NodeRevisionNumber revisionNumber = (NodeRevisionNumber) e.nextElement();
                // get successors
                crit = new Criteria();
                crit.addEqualTo("uriId", new Long(uriPeer.getUriId()));
                crit.addEqualTo("revisionNumber", revisionNumber.toString());
                query = QueryFactory.newQuery(VersionHistoryPeer.class, crit);
                query.addPrefetchedRelationship("predecessors");
                vhi = broker.getIteratorByQuery(query);
                Vector predecessors = new Vector();
                while( vhi.hasNext() ) {
                    VersionHistoryPeer vh = (VersionHistoryPeer)vhi.next();
                    Iterator vhpi = vh.getPredecessors().iterator();
                    while( vhpi.hasNext() ) {
                        VersionPredecessorPeer vhp = (VersionPredecessorPeer)vhpi.next();
                        predecessors.add(new NodeRevisionNumber(vhp.getPredecessorVersionHistory().getRevisionNumber()));
                    }
                }
                // XXX it is important to call ctor with String, otherwise it does an increment in minor number :(
                branches.put(new NodeRevisionNumber(revisionNumber.toString()), predecessors);
            }
            revisionDescriptors =
                new NodeRevisionDescriptors(
                    uri.toString(),
                    initialRevision,
                    workingRevisions,
                    latestRevisionNumbers,
                    branches,
                    versionPeer.isVersioned());
        } catch (PersistenceBrokerException e) {
            throw createException(e, uri.toString());
        }
        return revisionDescriptors;
        
    }
    
    public void createRevisionDescriptors(Uri uri, NodeRevisionDescriptors revisionDescriptors)
            throws ServiceAccessException {
        
        try {
            PersistenceBroker broker = getCurrentBroker();
            UriPeer uriPeer = findUriPeer(broker, uri.toString());
            Long uriId = new Long(uriPeer.getUriId());
            Criteria crit = new Criteria();
            crit.addEqualTo("uriId", uriId);
            QueryByCriteria query = QueryFactory.newQuery(VersionPeer.class, crit);
            VersionPeer versionPeer = (VersionPeer)broker.getObjectByQuery(query);
            if ( versionPeer == null ) {
                versionPeer = new VersionPeer();
                versionPeer.setUriId(uriPeer.getUriId());
                versionPeer.setVersioned(revisionDescriptors.isVersioned());
                broker.store(versionPeer);
            }
            boolean versionHistoryExists;
            NodeRevisionNumber versionNumber = revisionDescriptors.hasRevisions() ? revisionDescriptors.getLatestRevision() : new NodeRevisionNumber();
            crit = new Criteria();
            crit.addEqualTo("uriId", uriId);
            crit.addEqualTo("revisionNumber", versionNumber.toString());
            query = QueryFactory.newQuery(VersionHistoryPeer.class, crit);
            VersionHistoryPeer versionHistoryPeer = (VersionHistoryPeer)broker.getObjectByQuery(query);
            if (versionHistoryPeer == null && revisionDescriptors.getLatestRevision() != null) {
                // FIXME: Create new revisions on the main branch???
                BranchPeer branchPeer = findBranchPeer(broker, NodeRevisionDescriptors.MAIN_BRANCH);
                versionHistoryPeer = new VersionHistoryPeer();
                versionHistoryPeer.setUriId(uriPeer.getUriId());
                versionHistoryPeer.setBranchId(branchPeer.getBranchId());
                versionHistoryPeer.setRevisionNumber(getRevisionNumberAsString(versionNumber));
                broker.store(versionHistoryPeer);
                // reload the object to get generated primary key
                versionHistoryPeer = (VersionHistoryPeer)broker.getObjectByQuery(query);
            }
            
            // Add revision successors
            Enumeration revisionNumbers = revisionDescriptors.enumerateRevisionNumbers();
            while (revisionNumbers.hasMoreElements()) {
                NodeRevisionNumber nodeRevisionNumber = (NodeRevisionNumber) revisionNumbers.nextElement();
                
                Enumeration successors = revisionDescriptors.getSuccessors(nodeRevisionNumber);
                while (successors.hasMoreElements()) {
                    NodeRevisionNumber successor = (NodeRevisionNumber) successors.nextElement();
                    crit = new Criteria();                 
                    crit.addEqualTo("uriId", uriId);
                    crit.addEqualTo("revisionNumber", successor.toString());
                    query = QueryFactory.newQuery(VersionHistoryPeer.class, crit);
                    VersionHistoryPeer successorPeer = (VersionHistoryPeer)broker.getObjectByQuery(query);
                    if ( successorPeer != null && versionHistoryPeer != null) {
                        VersionPredecessorPeer vhp = new VersionPredecessorPeer();
                        vhp.setVersionId(versionHistoryPeer.getVersionId());
                        vhp.setPredecessorId(successorPeer.getVersionId());
                        broker.store(vhp);
                    }
                }
            }
            getLogger().log(
                revisionDescriptors.getOriginalUri() + revisionDescriptors.getInitialRevision(),
                LOG_CHANNEL,
                Logger.DEBUG);
        } catch (PersistenceBrokerException e) {
            throw createException(e, uri.toString());
        }
    }

    public void storeRevisionDescriptors(Uri uri,
            NodeRevisionDescriptors revisionDescriptors)
            throws ServiceAccessException, RevisionDescriptorNotFoundException {
        removeRevisionDescriptors(uri);
        createRevisionDescriptors(uri, revisionDescriptors);
    }

    public void removeRevisionDescriptors(Uri uri)
            throws ServiceAccessException {
        
        try {
            PersistenceBroker broker = getCurrentBroker();
            Criteria crit = new Criteria();
            crit.addEqualTo("versionHistory.uri.uriString", uri.toString());
            Query query = QueryFactory.newQuery(VersionPredecessorPeer.class, crit);
            Iterator res = broker.getIteratorByQuery(query);
            // deleteByQuery doesn't work with this query
            // do it one by one
            while( res.hasNext() ) {
                VersionPredecessorPeer vpp = (VersionPredecessorPeer)res.next();
                broker.delete(vpp);
            }
        } catch (PersistenceBrokerException e) {
            throw createException(e, uri.toString());
        }
    }
    
    // Content Store
    
    public NodeRevisionContent retrieveRevisionContent(Uri uri, NodeRevisionDescriptor revisionDescriptor)
        throws ServiceAccessException, RevisionNotFoundException {
        // TODO Auto-generated method stub
        return null;
    }

    public void createRevisionContent(Uri uri, NodeRevisionDescriptor revisionDescriptor, NodeRevisionContent revisionContent)
    	throws ServiceAccessException, RevisionAlreadyExistException {
        // TODO Auto-generated method stub

    }

    public void storeRevisionContent(Uri uri, NodeRevisionDescriptor revisionDescriptor, NodeRevisionContent revisionContent)
    	throws ServiceAccessException, RevisionNotFoundException {
        // TODO Auto-generated method stub

    }

    /**
     * @see org.apache.slide.store.ContentStore#removeRevisionContent(org.apache.slide.common.Uri, org.apache.slide.content.NodeRevisionDescriptor)
     */
    public void removeRevisionContent(Uri uri, NodeRevisionDescriptor revisionDescriptor)
            throws ServiceAccessException {
        // TODO Auto-generated method stub

    }
    
    // RevisionDescriptor Store
    
    public NodeRevisionDescriptor retrieveRevisionDescriptor(PersistenceBroker broker, Uri uri, NodeRevisionNumber revisionNumber)
    	throws ServiceAccessException, RevisionDescriptorNotFoundException {
        
        NodeRevisionDescriptor revisionDescriptor = null;
        if (revisionNumber == null)
            throw new RevisionDescriptorNotFoundException(uri.toString());
        try {
            VersionHistoryPeer versionHistoryPeer = findVersionHistoryPeer(broker, uri.toString(), revisionNumber.toString());
            revisionDescriptor = new NodeRevisionDescriptor(revisionNumber, versionHistoryPeer.getBranch().getBranchString());
            if ( versionHistoryPeer == null )
                throw new RevisionDescriptorNotFoundException(uri.toString());
            for(Iterator res=versionHistoryPeer.getLabels().iterator(); res.hasNext();) {
                VersionLabelPeer versionLabelPeer = (VersionLabelPeer)res.next();
                revisionDescriptor.addLabel(versionLabelPeer.getLabel().getLabelString());
            }
            for(Iterator res=versionHistoryPeer.getProperties().iterator(); res.hasNext();) {
                PropertyPeer propertyPeer = (PropertyPeer)res.next();
                PropertyHandler handler = PropertyHandlerFactory.getPropertyHandler(propertyPeer.getNamespace(), propertyPeer.getName());
                NodeProperty property = handler.retrieve(broker, propertyPeer);
                revisionDescriptor.setProperty(property);
            }
        } catch (PersistenceBrokerException e) {
            throw createException(e, uri.toString());
        } catch (PropertyHandlerException e) {
            throw createException(e, uri.toString());
        }
        return revisionDescriptor;
    }

    public void createRevisionDescriptor(Uri uri, NodeRevisionDescriptor revisionDescriptor)
            throws ServiceAccessException {
        
        try {
            PersistenceBroker broker = getCurrentBroker();
            VersionHistoryPeer vhp = assureVersionInfo(broker, uri, revisionDescriptor);
            createVersionLabels(broker, uri, revisionDescriptor);
            for (Enumeration properties = revisionDescriptor.enumerateProperties(); properties.hasMoreElements();) {
                NodeProperty property = (NodeProperty) properties.nextElement();
                PropertyPeer propertyPeer = new PropertyPeer();
                propertyPeer.setVersionId(vhp.getVersionId());
                PropertyHandler handler = PropertyHandlerFactory.getPropertyHandler(property.getNamespace(), property.getName());
                handler.store(broker, propertyPeer, property);
            }
        } catch (PersistenceBrokerException e) {
            throw createException(e, uri.toString());
        } catch (PropertyHandlerException pe) {
            throw createException(pe, uri.toString());
        }
    }

    public void storeRevisionDescriptor(Uri uri, NodeRevisionDescriptor revisionDescriptor)
            throws ServiceAccessException, RevisionDescriptorNotFoundException {
        /*
        removeRevisionDescriptor(uri, revisionDescriptor.getRevisionNumber());
        createRevisionDescriptor(uri, revisionDescriptor);
        */
        try {
            PersistenceBroker broker = getCurrentBroker();
            VersionHistoryPeer versionHistoryPeer = findVersionHistoryPeer(broker, uri.toString(), revisionDescriptor.getRevisionNumber().toString());
            if ( versionHistoryPeer == null )
                throw new RevisionDescriptorNotFoundException(uri.toString());
            removeVersionLabels(broker, uri, revisionDescriptor.getRevisionNumber());
            createVersionLabels(broker, uri, revisionDescriptor);
            Enumeration rp = revisionDescriptor.enumerateRemovedProperties();
            while ( rp.hasMoreElements() ) {
                NodeProperty property = (NodeProperty)rp.nextElement();
                Criteria crit = new Criteria();
                crit.addEqualTo("versionId", new Long(versionHistoryPeer.getVersionId()));
                crit.addEqualTo("name", property.getName());
                crit.addEqualTo("namespace", property.getNamespace());
                Query query = QueryFactory.newQuery(PropertyPeer.class, crit);
                PropertyPeer propertyPeer = (PropertyPeer)broker.getObjectByQuery(query);
                if ( propertyPeer == null ) {
                    getLogger().log("Removed property not found " + property, LOG_CHANNEL, Logger.WARNING);
                    continue;
                }
                PropertyHandler handler = PropertyHandlerFactory.getPropertyHandler(property.getNamespace(), property.getName());
                handler.remove(broker, propertyPeer);
            }
            revisionDescriptor.resetRemovedProperties();
            Enumeration up = revisionDescriptor.enumerateUpdatedProperties();
            while ( up.hasMoreElements() ) {
                NodeProperty property = (NodeProperty)up.nextElement();
                Criteria crit = new Criteria();
                crit.addEqualTo("versionId", new Long(versionHistoryPeer.getVersionId()));
                crit.addEqualTo("name", property.getName());
                crit.addEqualTo("namespace", property.getNamespace());
                Query query = QueryFactory.newQuery(PropertyPeer.class, crit);
                PropertyPeer propertyPeer = (PropertyPeer)broker.getObjectByQuery(query);
                if ( propertyPeer == null ) {
                    // created
                    propertyPeer = new PropertyPeer();
                    propertyPeer.setVersionId(versionHistoryPeer.getVersionId());
                }
                PropertyHandler handler = PropertyHandlerFactory.getPropertyHandler(property.getNamespace(), property.getName());
                handler.store(broker, propertyPeer, property);
            }
            revisionDescriptor.resetUpdatedProperties();
        } catch (PersistenceBrokerException e) {
            throw createException(e, uri.toString());
        } catch (PropertyHandlerException pe) {
            throw createException(pe, uri.toString());
        }
    }

    public void removeRevisionDescriptor(Uri uri, NodeRevisionNumber revisionNumber)
    	throws ServiceAccessException {

        try {
            PersistenceBroker broker = getCurrentBroker();
            removeVersionLabels(broker, uri, revisionNumber);
            Criteria crit = new Criteria();
            crit.addEqualTo("versionHistory.uri.uriString", uri.toString());
            crit.addEqualTo("versionHistory.revisionNumber", revisionNumber.toString());
            Query query = QueryFactory.newQuery(PropertyPeer.class, crit);
            Iterator pi = broker.getIteratorByQuery(query);
            while(pi.hasNext()) {
                PropertyPeer propertyPeer = (PropertyPeer)pi.next();
                PropertyHandler handler = PropertyHandlerFactory.getPropertyHandler(propertyPeer.getNamespace(), propertyPeer.getName());
                handler.remove(broker, propertyPeer);
            }
        } catch (PersistenceBrokerException e) {
            throw createException(e, uri.toString());
        } catch (PropertyHandlerException e) {
            throw createException(e, uri.toString());
        }
    }

    // helper methods
    
    protected UriPeer findUriPeer(PersistenceBroker broker, String uri) throws PersistenceBrokerException {
        Criteria crit = new Criteria();
        crit.addEqualTo("uriString", uri);
        Query query = QueryFactory.newQuery(UriPeer.class, crit);
        return (UriPeer)broker.getObjectByQuery(query);
    }
    
    protected UriPeer assureUriPeer(PersistenceBroker broker, String uri) throws PersistenceBrokerException {
        Criteria crit = new Criteria();
        crit.addEqualTo("uriString", uri);
        Query query = QueryFactory.newQuery(UriPeer.class, crit);
        UriPeer uriPeer = (UriPeer)broker.getObjectByQuery(query);
        if ( uriPeer != null )
            return uriPeer;
        // not found, create one
        uriPeer = new UriPeer();
        uriPeer.setUriString(uri);
        broker.store(uriPeer);
        return assureUriPeer(broker, uri);
    }
    
    protected void clearBinding(PersistenceBroker broker, Uri uri)
            throws PersistenceBrokerException {
        // clear this uri from having bindings and being bound
        UriPeer uriPeer = findUriPeer(broker, uri.toString());
        if ( uriPeer == null )
            return;
        Criteria crit = new Criteria();
        crit.addEqualTo("uriId", new Long(uriPeer.getUriId()));
        Query query = QueryFactory.newQuery(BindingPeer.class, crit);
        broker.deleteByQuery(query);
        query = QueryFactory.newQuery(ParentBindingPeer.class, crit);
        broker.deleteByQuery(query);
        // Note: deleteByQuery does not delete objects from cache.
        //       Maybe we should just use the empty cache impl since
        //       Slide is doing its own caching anyway.
        broker.clearCache();
    }
    
    protected void clearBinding(PersistenceBroker broker, Uri uri, Set updatedBindings) 
    		throws PersistenceBrokerException {

        // clear this uri from having bindings and being bound
        int bsize = updatedBindings.size();
        //If there are bindings to update, only remove those from the database
        if (bsize > 0) {
            
            UriPeer uriPeer = findUriPeer(broker, uri.toString());
            if ( uriPeer == null )
                return;
            Criteria crit = new Criteria();
            crit.addEqualTo("uriId", new Long(uriPeer.getUriId()));
            Query query = QueryFactory.newQuery(ParentBindingPeer.class, crit);
            broker.deleteByQuery(query);
            // we can't use relationships in a delete query like
            //   crit.addIn("childUuri.uriString", updatedBindings)
            // get the ids with a separate query then.
            Criteria subCrit = new Criteria();
            subCrit.addIn("uriString", updatedBindings);
            ReportQueryByCriteria subQuery = QueryFactory.newReportQuery(UriPeer.class, subCrit);
            subQuery.setAttributes(new String[] { "uriId" } );
            Iterator bi = broker.getReportQueryIteratorByQuery(subQuery);
            Vector bl = new Vector();
            while( bi.hasNext() )
                bl.add(((Object[])bi.next())[0]);
            crit.addIn("childUuriId", bl);
            query = QueryFactory.newQuery(BindingPeer.class, crit);
            broker.deleteByQuery(query);
            broker.clearCache();
        } else {
            //otherwise remove all related to the uri
            clearBinding(broker, uri);
        }
    }

    protected void removeVersionLabels(PersistenceBroker broker, Uri uri, NodeRevisionNumber revisionNumber) 
    	throws PersistenceBrokerException {
        
        Criteria crit = new Criteria();
        crit.addEqualTo("versionHistory.uri.uriString", uri.toString());
        crit.addEqualTo("versionHistory.revisionNumber", revisionNumber.toString());
        Query query = QueryFactory.newQuery(VersionLabelPeer.class, crit);
        Iterator vli = broker.getIteratorByQuery(query);
        while (vli.hasNext())
            broker.delete(vli.next());
    }
    
    protected void createVersionLabels(PersistenceBroker broker, Uri uri, NodeRevisionDescriptor revisionDescriptor) 
    	throws PersistenceBrokerException {

        for (Enumeration labels = revisionDescriptor.enumerateLabels(); labels.hasMoreElements();) {
            LabelPeer labelPeer = assureLabelPeer(broker, (String)labels.nextElement());
            VersionHistoryPeer vhp = findVersionHistoryPeer(broker, uri.toString(), revisionDescriptor.getRevisionNumber().toString());
            if ( vhp != null ) {
                VersionLabelPeer versionLabel = new VersionLabelPeer();
                versionLabel.setLabelId(labelPeer.getLabelId());
                versionLabel.setVersionId(vhp.getVersionId());
                broker.store(versionLabel);
            }
        }
    }
    
    protected VersionHistoryPeer findVersionHistoryPeer(PersistenceBroker broker, String uri, String revisionNumber)
    	throws PersistenceBrokerException {
        
        Criteria crit = new Criteria();
        crit.addEqualTo("uri.uriString", uri);
        crit.addEqualTo("revisionNumber", revisionNumber);
        Query query = QueryFactory.newQuery(VersionHistoryPeer.class, crit);
        return (VersionHistoryPeer)broker.getObjectByQuery(query);
    }
    
    protected LabelPeer assureLabelPeer(PersistenceBroker broker, String label) throws PersistenceBrokerException {
        
        Criteria crit = new Criteria();
        crit.addEqualTo("labelString", label);
        Query query = QueryFactory.newQuery(LabelPeer.class, crit);
        LabelPeer labelPeer = (LabelPeer)broker.getObjectByQuery(query);
        if ( labelPeer != null )
            return labelPeer;
        
        labelPeer = new LabelPeer();
        labelPeer.setLabelString(label);
        broker.store(labelPeer);
        return assureLabelPeer(broker, label);
    }
 
    protected BranchPeer findBranchPeer(PersistenceBroker broker, String branch)
            throws PersistenceBrokerException {

        Criteria crit = new Criteria();
        crit.addEqualTo("branchString", branch);
        Query query = QueryFactory.newQuery(BranchPeer.class, crit);
        BranchPeer branchPeer = (BranchPeer) broker.getObjectByQuery(query);
        return branchPeer;
    }
    
    protected BranchPeer assureBranchPeer(PersistenceBroker broker, String branch) throws PersistenceBrokerException {
        
        Criteria crit = new Criteria();
        crit.addEqualTo("branchString", branch);
        Query query = QueryFactory.newQuery(BranchPeer.class, crit);
        BranchPeer branchPeer = (BranchPeer)broker.getObjectByQuery(query);
        if ( branchPeer != null )
            return branchPeer;
        
        branchPeer = new BranchPeer();
        branchPeer.setBranchString(branch);
        broker.store(branchPeer);

        return assureBranchPeer(broker, branch);
    }
    
    protected VersionHistoryPeer assureVersionInfo(PersistenceBroker broker, Uri uri,
            NodeRevisionDescriptor revisionDescriptor) throws PersistenceBrokerException {
        
        UriPeer uriPeer = assureUriPeer(broker, uri.toString());
        
        Criteria crit = new Criteria();
        crit.addEqualTo("uriId", new Long(uriPeer.getUriId()));
        QueryByCriteria query = QueryFactory.newQuery(VersionPeer.class, crit);
        VersionPeer versionPeer = (VersionPeer)broker.getObjectByQuery(query);

        // FIXME: Is it true, that the default for isVersion is false ??
        if (versionPeer == null) {
            versionPeer = new VersionPeer();
            versionPeer.setUriId(uriPeer.getUriId());
            versionPeer.setVersioned(false);
            broker.store(versionPeer);
        }
        
        crit = new Criteria();
        crit.addEqualTo("uriId", new Long(uriPeer.getUriId()));
        crit.addEqualTo("revisionNumber", revisionDescriptor.getRevisionNumber().toString());
        query = QueryFactory.newQuery(VersionHistoryPeer.class, crit);
        VersionHistoryPeer versionHistoryPeer = (VersionHistoryPeer)broker.getObjectByQuery(query);
        
        if ( versionHistoryPeer == null ) {
            BranchPeer branchPeer = assureBranchPeer(broker, revisionDescriptor.getBranchName());
            versionHistoryPeer = new VersionHistoryPeer();
            versionHistoryPeer.setUriId(uriPeer.getUriId());
            versionHistoryPeer.setBranchId(branchPeer.getBranchId());
            versionHistoryPeer.setRevisionNumber(getRevisionNumberAsString(revisionDescriptor.getRevisionNumber()));
            broker.store(versionHistoryPeer);
            // reload the object to get the autogenerated primary key
            versionHistoryPeer = (VersionHistoryPeer)broker.getObjectByQuery(query);
        }     
        return versionHistoryPeer;
    }
    
    // This is taken from StandardRDBMSStore, may not be correct 
    protected ServiceAccessException createException(Exception pe, String uri) {
        /*
         * For Postgresql error states (match SQL92 ones for class 23
         * and 40) see
         * http://developer.postgresql.org/docs/postgres/errcodes-appendix.html
         */
        Throwable e = pe;
        if ( pe instanceof PersistenceBrokerException)
            e = ((PersistenceBrokerException)pe).getSourceException();
        if ( pe instanceof PropertyHandlerException ) {
            Throwable he = ((PropertyHandlerException)pe).getCause();
            if ( he instanceof PersistenceBrokerException )
                e = ((PersistenceBrokerException)he).getSourceException();
        }
        if ( e != null && e instanceof SQLException ) {
            SQLException se = (SQLException) e;
            String sqlstate = se.getSQLState();

            if (sqlstate != null) {
                if (sqlstate.startsWith("23")) {
                    getLogger().log(se.getErrorCode() + ": Low isolation conflict for "
                                    + uri, LOG_CHANNEL, Logger.WARNING);
                    throw new ConcurrencyConflictError(se, uri);

                } else if (sqlstate.startsWith("40")) {
                    getLogger().log(se.getErrorCode() + ": Deadlock resolved on " + uri,
                                    LOG_CHANNEL, Logger.WARNING);
                    throw new ConcurrencyConflictError(se, uri);
                }
            }
            getLogger().log("SQL error (" + se.getErrorCode() + ") state"
                            + se.getSQLState() + " on " + uri + ": "
                            + se.getMessage(), LOG_CHANNEL, Logger.ERROR);
        }
        return new ServiceAccessException(this, pe);
    }
    
    // null means permission is valid for all revisions
    protected String getRevisionNumberAsString(NodeRevisionNumber revisionNumber) {
        return revisionNumber != null ? revisionNumber.toString() : null;
    }
    
    /**
     * Compares objects of type VersionHistoryPeer based on the
     * revisionNumber field.
     *
     */
    public class RevisionNumberComparator implements Comparator {
        
        public int compare(Object o1, Object o2) {
            if ( !(o1 instanceof VersionHistoryPeer) || !(o2 instanceof VersionHistoryPeer ))
                throw new ClassCastException();
            String r1 = ((VersionHistoryPeer)o1).getRevisionNumber();
            String r2 = ((VersionHistoryPeer)o2).getRevisionNumber();
            while(true) {
                int i=r1.indexOf('.');
                String s1 = i == -1 ? r1 : r1.substring(0, i);
                int j=r2.indexOf('.');
                String s2 = j == -1 ? r2 : r2.substring(0, j);
                int n1 = 0;
                try {
                    n1 = Integer.parseInt(s1);
                } catch (NumberFormatException nfe ) {
                    getLogger().log("Error parsing revision number: " + s1, LOG_CHANNEL, Logger.WARNING);
                }
                int n2 = 0;
                try {
                    n2 = Integer.parseInt(s2);
                } catch (NumberFormatException nfe ) {
                    getLogger().log("Error parsing revision number: " + s2, LOG_CHANNEL, Logger.WARNING);
                }
                if ( n1 != n2 || ( i == -1 && j == -1) )
                    return (n1 - n2);
                // this makes 1.1.1 > 1.1
                if ( i == -1 )
                    return -1;
                if ( j == -1 )
                    return 1;
                r1 = r1.substring(i+1);
                r2 = r2.substring(j+1);
            }
        }
    }
}
