JpaToPgMigration.java

/****************************************************************
 * 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.james;

import static org.apache.james.backends.postgres.utils.PostgresExecutor.DEFAULT_INJECT;
import static org.apache.james.backends.postgres.utils.PostgresExecutor.EAGER_FETCH;
import static org.apache.james.user.postgres.PostgresUserModule.PostgresUserTable.ALGORITHM;
import static org.apache.james.user.postgres.PostgresUserModule.PostgresUserTable.HASHED_PASSWORD;
import static org.apache.james.user.postgres.PostgresUserModule.PostgresUserTable.TABLE_NAME;
import static org.apache.james.user.postgres.PostgresUserModule.PostgresUserTable.USERNAME;
import static org.apache.james.user.postgres.PostgresUserModule.PostgresUserTable.USERNAME_PRIMARY_KEY;

import java.util.Set;

import jakarta.inject.Inject;
import jakarta.inject.Named;

import org.apache.james.backends.postgres.PostgresModule;
import org.apache.james.backends.postgres.utils.PostgresExecutor;
import org.apache.james.core.Username;
import org.apache.james.domainlist.api.DomainListException;
import org.apache.james.domainlist.jpa.JPADomainList;
import org.apache.james.domainlist.lib.DomainListConfiguration;
import org.apache.james.domainlist.postgres.PostgresDomainList;
import org.apache.james.filesystem.api.FileSystem;
import org.apache.james.mailrepository.api.MailRepositoryUrlStore;
import org.apache.james.mailrepository.jpa.JPAMailRepositoryUrlStore;
import org.apache.james.mailrepository.postgres.PostgresMailRepositoryUrlStore;
import org.apache.james.modules.data.JPAEntityManagerModule;
import org.apache.james.modules.data.PostgresCommonModule;
import org.apache.james.modules.data.PostgresDelegationStoreModule;
import org.apache.james.modules.data.PostgresDomainListModule;
import org.apache.james.modules.data.PostgresRecipientRewriteTableModule;
import org.apache.james.modules.data.PostgresUsersRepositoryModule;
import org.apache.james.modules.server.DNSServiceModule;
import org.apache.james.modules.server.DropWizardMetricsModule;
import org.apache.james.rrt.api.RecipientRewriteTableException;
import org.apache.james.rrt.jpa.JPARecipientRewriteTable;
import org.apache.james.rrt.postgres.PostgresRecipientRewriteTable;
import org.apache.james.server.core.configuration.Configuration;
import org.apache.james.server.core.configuration.ConfigurationProvider;
import org.apache.james.server.core.configuration.FileConfigurationProvider;
import org.apache.james.server.core.filesystem.FileSystemImpl;
import org.apache.james.user.lib.model.Algorithm;
import org.apache.james.user.lib.model.DefaultUser;
import org.apache.james.user.postgres.PostgresUsersRepositoryConfiguration;
import org.apache.james.utils.InitializationOperation;
import org.apache.james.utils.InitializationOperations;
import org.apache.james.utils.InitilizationOperationBuilder;
import org.apache.james.utils.PropertiesProvider;
import org.jooq.Field;
import org.jooq.Record;
import org.jooq.Table;
import org.jooq.impl.DSL;
import org.jooq.impl.SQLDataType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.inject.AbstractModule;
import com.google.inject.Guice;
import com.google.inject.Injector;
import com.google.inject.Module;
import com.google.inject.Provides;
import com.google.inject.Scopes;
import com.google.inject.Singleton;
import com.google.inject.multibindings.Multibinder;
import com.google.inject.multibindings.ProvidesIntoSet;
import com.google.inject.util.Modules;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
 * A sample demonstrating how to migrate a limited subset of
 * james code data a JPA PG database to the new Jooq based PG
 * implementation.
 * <p>
 * This demonstrates migration of
 * - domain list
 * - users (limited to JPA users already using PBKDF2 without salt)
 * - recipient rewrite tables
 * - mail repository url
 * use at your own risk  (Test the migration on a restored backup first)
 */
public class JpaToPgMigration {
    private static final Logger LOGGER = LoggerFactory.getLogger(JpaToPgMigration.class);

    private static final Module JPA_MODULES = Modules.combine(
            new JPAEntityManagerModule(),
            new UnboundJPAMigrationModule()
    );
    private static final Module POSTGRESQL_MODULES = Modules.combine(
            new PostgresCommonModule(),
            new PostgresDomainListModule(),
            new PostgresRecipientRewriteTableModule(),
            new PostgresUsersRepositoryModule(),
            new PostgresMailRepositoryUrlStoreModule(),
            new PostgresDelegationStoreModule()
    );


    public static void main(String[] args) {
        SMTPRelayConfiguration configuration = SMTPRelayConfiguration.builder()
                .useWorkingDirectoryEnvProperty()
                .build();
        LOGGER.info("Loading configuration {}", configuration.toString());


        var module = Modules.combine(
                new DNSServiceModule(),
                new DropWizardMetricsModule(),
                new MigrationModule(configuration),
                new CoreEntityValidatorsModule(),
                PostgresUsersRepositoryModule.USER_CONFIGURATION_MODULE,
                JPA_MODULES,
                POSTGRESQL_MODULES
        );

        Injector injector = Guice.createInjector(module);
        injector.getInstance(InitializationOperations.class).initModules();
        injector.getInstance(DomainMigration.class).doMigration();
        injector.getInstance(UsersMigration.class).doMigration();
        injector.getInstance(RRTMigration.class).doMigration();
        injector.getInstance(MailRepositoryUrlMigration.class).doMigration();
    }

    static class DomainMigration {
        private static final Logger LOGGER = LoggerFactory.getLogger(DomainMigration.class);

        private final JPADomainList jpaDomainList;
        private final PostgresDomainList pgDomainList;

        @Inject
        DomainMigration(
                JPADomainList jpaDomainList,
                PostgresDomainList pgDomainList
        ) {
            this.jpaDomainList = jpaDomainList;
            this.pgDomainList = pgDomainList;
        }

        void doMigration() {
            LOGGER.info("Start domains migration");
            try {
                jpaDomainList.getDomains().forEach(domain -> {
                            try {
                                pgDomainList.addDomain(domain);
                            } catch (DomainListException e) {
                                if (!e.getMessage().contains("already exists.")) {
                                    LOGGER.warn("Failed to migrate domain {}", domain, e);
                                }
                            }
                        }
                );

            } catch (DomainListException e) {
                throw new RuntimeException("Unable to migrate domains, aborting processing", e);
            }
            LOGGER.info("Domains migration completed");
        }
    }

    static class UsersMigration {
        private static final Logger LOGGER = LoggerFactory.getLogger(UsersMigration.class);

        private final PGJPAUsersDao usersMigrationDao;

        @Inject
        UsersMigration(
                PGJPAUsersDao usersMigrationDao
        ) {
            this.usersMigrationDao = usersMigrationDao;
        }

        void doMigration() {
            LOGGER.info("Start users migration");
            usersMigrationDao.listJpaUsers()
                    .flatMap(usersMigrationDao::addUser)
                    .count()
                    .block();
        }
    }

    static class RRTMigration {
        private static final Logger LOGGER = LoggerFactory.getLogger(RRTMigration.class);

        private final JPARecipientRewriteTable jpaRRTRepository;
        private final PostgresRecipientRewriteTable pgRRTRepository;

        @Inject
        RRTMigration(
                JPARecipientRewriteTable jpaRRTRepository,
                PostgresRecipientRewriteTable pgRRTRepository
        ) {
            this.jpaRRTRepository = jpaRRTRepository;
            this.pgRRTRepository = pgRRTRepository;
        }

        void doMigration() {
            LOGGER.info("Start RRT migration");
            try {
                jpaRRTRepository.getAllMappings().forEach((mappingSource, mappings) ->
                        mappings.forEach(mapping ->
                                pgRRTRepository.addMapping(mappingSource, mapping)
                        )
                );
            } catch (RecipientRewriteTableException e) {
                throw new RuntimeException(e);
            }
            LOGGER.info("RRT migration complete");
        }
    }

    static class PGJPAUsersDao {
        private static final Table<Record> JPA_USERS_TABLE = DSL.table("james_user");
        private static final Field<String> JPA_USERNAME = DSL.field("user_name", SQLDataType.VARCHAR(100).notNull());
        private static final Field<String> JPA_PASSWORD = DSL.field("password", SQLDataType.VARCHAR(128).notNull());
        private static final Field<String> JPA_ALGORITHM = DSL.field("password_hash_algorithm", SQLDataType.VARCHAR(100).notNull());

        private final PostgresExecutor postgresExecutor;
        private final Algorithm algorithm;
        private final Algorithm.HashingMode fallbackHashingMode;

        @Inject
        public PGJPAUsersDao(
                @Named(DEFAULT_INJECT) PostgresExecutor postgresExecutor,
                PostgresUsersRepositoryConfiguration postgresUsersRepositoryConfiguration
        ) {
            this.postgresExecutor = postgresExecutor;
            this.algorithm = postgresUsersRepositoryConfiguration.getPreferredAlgorithm();
            this.fallbackHashingMode = postgresUsersRepositoryConfiguration.getFallbackHashingMode();
        }

        public Flux<DefaultUser> listJpaUsers() {
            return postgresExecutor.executeRows(
                    dslContext -> Flux.from(
                            dslContext.select(
                                    JPA_USERNAME,
                                    JPA_PASSWORD,
                                    JPA_ALGORITHM
                            ).from(JPA_USERS_TABLE)
                    ), EAGER_FETCH
            ).map(record ->
                    new DefaultUser(
                            Username.of(record.get(JPA_USERNAME)),
                            record.get(JPA_PASSWORD),
                            Algorithm.of(record.get(JPA_ALGORITHM), fallbackHashingMode),
                            algorithm
                    )
            );
        }

        public Mono<Void> addUser(DefaultUser user) {
            return postgresExecutor.executeVoid(
                    dslContext -> Mono.from(
                            dslContext.insertInto(TABLE_NAME, USERNAME, HASHED_PASSWORD, ALGORITHM)
                                    .values(
                                            user.getUserName().asString(),
                                            user.getHashedPassword(),
                                            user.getHashAlgorithm().asString()
                                    )
                                    .onConflictOnConstraint(USERNAME_PRIMARY_KEY)
                                    .doNothing())
            );
        }
    }

    static class MailRepositoryUrlMigration {
        private static final Logger LOGGER = LoggerFactory.getLogger(MailRepositoryUrlMigration.class);

        private final JPAMailRepositoryUrlStore jpaMailRepositoryUrlStore;
        private final PostgresMailRepositoryUrlStore pgMailRepositoryUrlStore;

        @Inject
        MailRepositoryUrlMigration(
                JPAMailRepositoryUrlStore jpaMailRepositoryUrlStore,
                PostgresMailRepositoryUrlStore pgMailRepositoryUrlStore
        ) {
            this.jpaMailRepositoryUrlStore = jpaMailRepositoryUrlStore;
            this.pgMailRepositoryUrlStore = pgMailRepositoryUrlStore;
        }

        void doMigration() {
            LOGGER.info("Start mail repository urls migration");
            jpaMailRepositoryUrlStore.listDistinct().forEach(pgMailRepositoryUrlStore::add);
            LOGGER.info("Mail repository urls migration complete");
        }
    }

    static class MigrationModule extends AbstractModule {
        private final Configuration configuration;
        private final FileSystemImpl fileSystem;

        @Inject
        public MigrationModule(Configuration configuration) {
            this.configuration = configuration;
            this.fileSystem = new FileSystemImpl(configuration.directories());
        }

        @Override
        protected void configure() {
            bind(FileSystem.class).toInstance(fileSystem);
            bind(Configuration.class).toInstance(configuration);

            bind(ConfigurationProvider.class).toInstance(new FileConfigurationProvider(fileSystem, configuration));
            bind(DomainMigration.class).in(Scopes.SINGLETON);
            bind(UsersMigration.class).in(Scopes.SINGLETON);
            bind(RRTMigration.class).in(Scopes.SINGLETON);
            bind(MailRepositoryUrlMigration.class).in(Scopes.SINGLETON);
        }

        @Provides
        @jakarta.inject.Singleton
        public Configuration.ConfigurationPath configurationPath() {
            return configuration.configurationPath();
        }

        @Provides
        @jakarta.inject.Singleton
        public PropertiesProvider providePropertiesProvider(FileSystem fileSystem, Configuration.ConfigurationPath configurationPrefix) {
            return new PropertiesProvider(fileSystem, configurationPrefix);
        }

        @Provides
        @Singleton
        DomainListConfiguration domainListConfiguration() {
            return DomainListConfiguration.DEFAULT;
        }
    }

    static class UnboundJPAMigrationModule extends AbstractModule {
        @Override
        protected void configure() {
            bind(JPADomainList.class).in(Scopes.SINGLETON);
            bind(JPARecipientRewriteTable.class).in(Scopes.SINGLETON);
            bind(JPAMailRepositoryUrlStore.class).in(Scopes.SINGLETON);
        }

        @ProvidesIntoSet
        InitializationOperation configureDomainList(DomainListConfiguration configuration, JPADomainList jpaDomainList) {
            return InitilizationOperationBuilder
                    .forClass(JPADomainList.class)
                    .init(() -> jpaDomainList.configure(configuration));
        }
    }

    static class CoreEntityValidatorsModule extends AbstractModule {
        @Override
        protected void configure() {
            Multibinder.newSetBinder(binder(), UserEntityValidator.class).addBinding().to(DefaultUserEntityValidator.class);
            Multibinder.newSetBinder(binder(), UserEntityValidator.class).addBinding().to(RecipientRewriteTableUserEntityValidator.class);
        }

        @Provides
        @Singleton
        UserEntityValidator userEntityValidator(Set<UserEntityValidator> validatorSet) {
            return new AggregateUserEntityValidator(validatorSet);
        }
    }

    static class PostgresMailRepositoryUrlStoreModule extends AbstractModule {
        @Override
        protected void configure() {
            bind(PostgresMailRepositoryUrlStore.class).in(Scopes.SINGLETON);
            bind(MailRepositoryUrlStore.class).to(PostgresMailRepositoryUrlStore.class);
            Multibinder.newSetBinder(binder(), PostgresModule.class)
                    .addBinding().toInstance(org.apache.james.mailrepository.postgres.PostgresMailRepositoryModule.MODULE);
        }
    }

}