Setting outbox calls on user updates / creations, removing not needed inbox events
This commit is contained in:
parent
953fa2710e
commit
80a3aa5616
|
|
@ -1,30 +0,0 @@
|
|||
package eu.eudat.commons.enums.notification;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonValue;
|
||||
import eu.eudat.commons.enums.EnumUtils;
|
||||
import eu.eudat.data.converters.enums.DatabaseEnum;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
public enum TentativeUserProfile implements DatabaseEnum<Short> {
|
||||
|
||||
Complete((short) 0), Tentative((short) 1);
|
||||
|
||||
private final Short value;
|
||||
|
||||
TentativeUserProfile(Short value) {
|
||||
this.value = value;
|
||||
}
|
||||
|
||||
@JsonValue
|
||||
public Short getValue() {
|
||||
return value;
|
||||
}
|
||||
|
||||
private static final Map<Short, TentativeUserProfile> map = EnumUtils.getEnumValueMap(TentativeUserProfile.class);
|
||||
|
||||
public static TentativeUserProfile of(Short i) {
|
||||
return map.get(i);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -9,7 +9,6 @@ import gr.cite.rabbitmq.consumer.InboxCreator;
|
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
import java.util.ArrayList;
|
||||
|
|
@ -19,29 +18,28 @@ import java.util.List;
|
|||
@EnableConfigurationProperties({OutboxProperties.class, InboxProperties.class})
|
||||
@ConditionalOnProperty(prefix = "queue.rabbitmq", name = "listenerEnabled")
|
||||
public class AppRabbitConfigurer extends RabbitConfigurer {
|
||||
private ApplicationContext applicationContext;
|
||||
private InboxProperties inboxProperties;
|
||||
|
||||
public AppRabbitConfigurer(ApplicationContext applicationContext, InboxProperties inboxProperties) {
|
||||
this.applicationContext = applicationContext;
|
||||
this.inboxProperties = inboxProperties;
|
||||
}
|
||||
private ApplicationContext applicationContext;
|
||||
|
||||
// @Bean
|
||||
public InboxBindings inboxBindingsCreator() {
|
||||
List<String> bindingItems = new ArrayList<>();
|
||||
bindingItems.addAll(this.inboxProperties.getUserRemovalTopic());
|
||||
bindingItems.addAll(this.inboxProperties.getUserTouchedTopic());
|
||||
private InboxProperties inboxProperties;
|
||||
|
||||
return new InboxBindings(bindingItems);
|
||||
}
|
||||
public AppRabbitConfigurer(ApplicationContext applicationContext, InboxProperties inboxProperties) {
|
||||
this.applicationContext = applicationContext;
|
||||
this.inboxProperties = inboxProperties;
|
||||
}
|
||||
|
||||
// @Bean
|
||||
public InboxBindings inboxBindingsCreator() {
|
||||
List<String> bindingItems = new ArrayList<>();
|
||||
|
||||
// @Bean
|
||||
public InboxCreator inboxCreator() {
|
||||
return (params) -> {
|
||||
InboxRepository inboxRepository = this.applicationContext.getBean(InboxRepository.class);
|
||||
return inboxRepository.create(params) != null;
|
||||
};
|
||||
}
|
||||
return new InboxBindings(bindingItems);
|
||||
}
|
||||
|
||||
// @Bean
|
||||
public InboxCreator inboxCreator() {
|
||||
return (params) -> {
|
||||
InboxRepository inboxRepository = this.applicationContext.getBean(InboxRepository.class);
|
||||
return inboxRepository.create(params) != null;
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,42 +1,25 @@
|
|||
package eu.eudat.integrationevent.inbox;
|
||||
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
//import org.springframework.boot.context.properties.ConstructorBinding;
|
||||
import org.springframework.validation.annotation.Validated;
|
||||
|
||||
import jakarta.validation.constraints.NotNull;
|
||||
import java.util.List;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.validation.annotation.Validated;
|
||||
|
||||
@Validated
|
||||
@ConfigurationProperties(prefix = "queue.task.listener.options")
|
||||
//@ConstructorBinding
|
||||
public class InboxProperties {
|
||||
@NotNull
|
||||
private final String exchange;
|
||||
@NotNull
|
||||
private final List<String> userRemovalTopic;
|
||||
@NotNull
|
||||
private final List<String> userTouchedTopic;
|
||||
|
||||
public InboxProperties(
|
||||
String exchange,
|
||||
List<String> userRemovalTopic,
|
||||
List<String> userTouchedTopic) {
|
||||
this.exchange = exchange;
|
||||
this.userRemovalTopic = userRemovalTopic;
|
||||
this.userTouchedTopic = userTouchedTopic;
|
||||
}
|
||||
@NotNull
|
||||
private final String exchange;
|
||||
|
||||
public List<String> getUserRemovalTopic() {
|
||||
return userRemovalTopic;
|
||||
}
|
||||
public InboxProperties(
|
||||
String exchange
|
||||
) {
|
||||
this.exchange = exchange;
|
||||
}
|
||||
|
||||
public List<String> getUserTouchedTopic() {
|
||||
return userTouchedTopic;
|
||||
}
|
||||
public String getExchange() {
|
||||
return exchange;
|
||||
}
|
||||
|
||||
public String getExchange() {
|
||||
return exchange;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,13 +1,10 @@
|
|||
package eu.eudat.integrationevent.inbox;
|
||||
|
||||
|
||||
import eu.eudat.commons.JsonHandlingService;
|
||||
import eu.eudat.commons.enums.IsActive;
|
||||
import eu.eudat.commons.fake.FakeRequestScope;
|
||||
import eu.eudat.data.QueueInboxEntity;
|
||||
import eu.eudat.integrationevent.TrackedEvent;
|
||||
import eu.eudat.integrationevent.inbox.userremoval.UserRemovalIntegrationEventHandler;
|
||||
import eu.eudat.integrationevent.inbox.usertouched.UserTouchedIntegrationEventHandler;
|
||||
import eu.eudat.query.QueueInboxQuery;
|
||||
import gr.cite.queueinbox.entity.QueueInbox;
|
||||
import gr.cite.queueinbox.entity.QueueInboxStatus;
|
||||
|
|
@ -18,11 +15,13 @@ import gr.cite.rabbitmq.consumer.InboxCreatorParams;
|
|||
import gr.cite.tools.data.query.Ordering;
|
||||
import gr.cite.tools.data.query.QueryFactory;
|
||||
import gr.cite.tools.logging.LoggerService;
|
||||
import jakarta.persistence.*;
|
||||
import jakarta.persistence.EntityManager;
|
||||
import jakarta.persistence.EntityManagerFactory;
|
||||
import jakarta.persistence.EntityTransaction;
|
||||
import jakarta.persistence.OptimisticLockException;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
|
|
@ -30,307 +29,318 @@ import java.util.function.Function;
|
|||
|
||||
public class InboxRepositoryImpl implements InboxRepository {
|
||||
|
||||
protected final ApplicationContext applicationContext;
|
||||
private static final LoggerService logger = new LoggerService(LoggerFactory.getLogger(InboxRepositoryImpl.class));
|
||||
private final JsonHandlingService jsonHandlingService;
|
||||
private final InboxProperties inboxProperties;
|
||||
protected final ApplicationContext applicationContext;
|
||||
|
||||
public InboxRepositoryImpl(
|
||||
ApplicationContext applicationContext,
|
||||
InboxProperties inboxProperties
|
||||
) {
|
||||
this.applicationContext = applicationContext;
|
||||
this.jsonHandlingService = this.applicationContext.getBean(JsonHandlingService.class);
|
||||
this.inboxProperties = inboxProperties;
|
||||
}
|
||||
private static final LoggerService logger = new LoggerService(LoggerFactory.getLogger(InboxRepositoryImpl.class));
|
||||
|
||||
private final JsonHandlingService jsonHandlingService;
|
||||
|
||||
@Override
|
||||
public CandidateInfo candidate(Instant lastCandidateCreationTimestamp, MessageOptions options) {
|
||||
EntityTransaction transaction = null;
|
||||
EntityManager entityManager = null;
|
||||
CandidateInfo candidate = null;
|
||||
try (FakeRequestScope fakeRequestScope = new FakeRequestScope()) {
|
||||
try {
|
||||
QueryFactory queryFactory = this.applicationContext.getBean(QueryFactory.class);
|
||||
EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
|
||||
entityManager = entityManagerFactory.createEntityManager();
|
||||
private final InboxProperties inboxProperties;
|
||||
|
||||
transaction = entityManager.getTransaction();
|
||||
transaction.begin();
|
||||
public InboxRepositoryImpl(
|
||||
ApplicationContext applicationContext,
|
||||
InboxProperties inboxProperties
|
||||
) {
|
||||
this.applicationContext = applicationContext;
|
||||
this.jsonHandlingService = this.applicationContext.getBean(JsonHandlingService.class);
|
||||
this.inboxProperties = inboxProperties;
|
||||
}
|
||||
|
||||
QueueInboxEntity item = queryFactory.query(QueueInboxQuery.class)
|
||||
.isActives(IsActive.Active)
|
||||
.status(QueueInboxStatus.PENDING, QueueInboxStatus.ERROR)
|
||||
.retryThreshold(options.getRetryThreashold())
|
||||
.createdAfter(lastCandidateCreationTimestamp)
|
||||
.ordering(new Ordering().addAscending(QueueInboxEntity._createdAt))
|
||||
.first();
|
||||
@Override
|
||||
public CandidateInfo candidate(Instant lastCandidateCreationTimestamp, MessageOptions options) {
|
||||
EntityTransaction transaction = null;
|
||||
EntityManager entityManager = null;
|
||||
CandidateInfo candidate = null;
|
||||
try (FakeRequestScope fakeRequestScope = new FakeRequestScope()) {
|
||||
try {
|
||||
QueryFactory queryFactory = this.applicationContext.getBean(QueryFactory.class);
|
||||
EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
|
||||
entityManager = entityManagerFactory.createEntityManager();
|
||||
|
||||
if (item != null) {
|
||||
QueueInboxStatus prevState = item.getStatus();
|
||||
item.setStatus(QueueInboxStatus.PROCESSING);
|
||||
transaction = entityManager.getTransaction();
|
||||
transaction.begin();
|
||||
|
||||
entityManager.merge(item);
|
||||
entityManager.flush();
|
||||
QueueInboxEntity item = queryFactory.query(QueueInboxQuery.class)
|
||||
.isActives(IsActive.Active)
|
||||
.status(QueueInboxStatus.PENDING, QueueInboxStatus.ERROR)
|
||||
.retryThreshold(options.getRetryThreashold())
|
||||
.createdAfter(lastCandidateCreationTimestamp)
|
||||
.ordering(new Ordering().addAscending(QueueInboxEntity._createdAt))
|
||||
.first();
|
||||
|
||||
candidate = new CandidateInfo();
|
||||
candidate.setId(item.getId());
|
||||
candidate.setCreatedAt(item.getCreatedAt());
|
||||
candidate.setPreviousState(prevState);
|
||||
}
|
||||
if (item != null) {
|
||||
QueueInboxStatus prevState = item.getStatus();
|
||||
item.setStatus(QueueInboxStatus.PROCESSING);
|
||||
|
||||
transaction.commit();
|
||||
} catch (OptimisticLockException ex) {
|
||||
// we get this if/when someone else already modified the notifications. We want to essentially ignore this, and keep working
|
||||
logger.debug("Concurrency exception getting queue inbox. Skipping: {} ", ex.getMessage());
|
||||
if (transaction != null) transaction.rollback();
|
||||
candidate = null;
|
||||
} catch (Exception ex) {
|
||||
logger.error("Problem getting list of queue inbox. Skipping: {}", ex.getMessage(), ex);
|
||||
if (transaction != null) transaction.rollback();
|
||||
candidate = null;
|
||||
} finally {
|
||||
if (entityManager != null) entityManager.close();
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
logger.error("Problem getting list of queue inbox. Skipping: {}", ex.getMessage(), ex);
|
||||
}
|
||||
entityManager.merge(item);
|
||||
entityManager.flush();
|
||||
|
||||
return candidate;
|
||||
}
|
||||
candidate = new CandidateInfo();
|
||||
candidate.setId(item.getId());
|
||||
candidate.setCreatedAt(item.getCreatedAt());
|
||||
candidate.setPreviousState(prevState);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Boolean shouldOmit(CandidateInfo candidate, Function<QueueInbox, Boolean> shouldOmit) {
|
||||
EntityTransaction transaction = null;
|
||||
EntityManager entityManager = null;
|
||||
Boolean success = false;
|
||||
transaction.commit();
|
||||
} catch (OptimisticLockException ex) {
|
||||
// we get this if/when someone else already modified the notifications. We want to essentially ignore this, and keep working
|
||||
logger.debug("Concurrency exception getting queue inbox. Skipping: {} ", ex.getMessage());
|
||||
if (transaction != null)
|
||||
transaction.rollback();
|
||||
candidate = null;
|
||||
} catch (Exception ex) {
|
||||
logger.error("Problem getting list of queue inbox. Skipping: {}", ex.getMessage(), ex);
|
||||
if (transaction != null)
|
||||
transaction.rollback();
|
||||
candidate = null;
|
||||
} finally {
|
||||
if (entityManager != null)
|
||||
entityManager.close();
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
logger.error("Problem getting list of queue inbox. Skipping: {}", ex.getMessage(), ex);
|
||||
}
|
||||
|
||||
try (FakeRequestScope fakeRequestScope = new FakeRequestScope()) {
|
||||
try {
|
||||
EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
|
||||
return candidate;
|
||||
}
|
||||
|
||||
entityManager = entityManagerFactory.createEntityManager();
|
||||
transaction = entityManager.getTransaction();
|
||||
@Override
|
||||
public Boolean shouldOmit(CandidateInfo candidate, Function<QueueInbox, Boolean> shouldOmit) {
|
||||
EntityTransaction transaction = null;
|
||||
EntityManager entityManager = null;
|
||||
Boolean success = false;
|
||||
|
||||
transaction.begin();
|
||||
try (FakeRequestScope fakeRequestScope = new FakeRequestScope()) {
|
||||
try {
|
||||
EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
|
||||
|
||||
QueryFactory queryFactory = this.applicationContext.getBean(QueryFactory.class);
|
||||
QueueInboxEntity item = queryFactory.query(QueueInboxQuery.class).ids(candidate.getId()).first();
|
||||
entityManager = entityManagerFactory.createEntityManager();
|
||||
transaction = entityManager.getTransaction();
|
||||
|
||||
if (item == null) {
|
||||
this.logger.warn("Could not lookup queue inbox {} to process. Continuing...", candidate.getId());
|
||||
} else {
|
||||
if (shouldOmit.apply(item)) {
|
||||
item.setStatus(QueueInboxStatus.OMITTED);
|
||||
transaction.begin();
|
||||
|
||||
entityManager.merge(item);
|
||||
entityManager.flush();
|
||||
success = true;
|
||||
}
|
||||
}
|
||||
QueryFactory queryFactory = this.applicationContext.getBean(QueryFactory.class);
|
||||
QueueInboxEntity item = queryFactory.query(QueueInboxQuery.class).ids(candidate.getId()).first();
|
||||
|
||||
transaction.commit();
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
if (transaction != null) transaction.rollback();
|
||||
success = false;
|
||||
} finally {
|
||||
if (entityManager != null) entityManager.close();
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
}
|
||||
return success;
|
||||
}
|
||||
if (item == null) {
|
||||
this.logger.warn("Could not lookup queue inbox {} to process. Continuing...", candidate.getId());
|
||||
} else {
|
||||
if (shouldOmit.apply(item)) {
|
||||
item.setStatus(QueueInboxStatus.OMITTED);
|
||||
|
||||
@Override
|
||||
public boolean shouldWait(CandidateInfo candidate, Function<QueueInbox, Boolean> itIsTimeFunc) {
|
||||
EntityTransaction transaction = null;
|
||||
EntityManager entityManager = null;
|
||||
Boolean success = false;
|
||||
entityManager.merge(item);
|
||||
entityManager.flush();
|
||||
success = true;
|
||||
}
|
||||
}
|
||||
|
||||
try (FakeRequestScope fakeRequestScope = new FakeRequestScope()) {
|
||||
try {
|
||||
EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
|
||||
transaction.commit();
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
if (transaction != null)
|
||||
transaction.rollback();
|
||||
success = false;
|
||||
} finally {
|
||||
if (entityManager != null)
|
||||
entityManager.close();
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
}
|
||||
return success;
|
||||
}
|
||||
|
||||
entityManager = entityManagerFactory.createEntityManager();
|
||||
transaction = entityManager.getTransaction();
|
||||
@Override
|
||||
public boolean shouldWait(CandidateInfo candidate, Function<QueueInbox, Boolean> itIsTimeFunc) {
|
||||
EntityTransaction transaction = null;
|
||||
EntityManager entityManager = null;
|
||||
Boolean success = false;
|
||||
|
||||
transaction.begin();
|
||||
try (FakeRequestScope fakeRequestScope = new FakeRequestScope()) {
|
||||
try {
|
||||
EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
|
||||
|
||||
QueryFactory queryFactory = this.applicationContext.getBean(QueryFactory.class);
|
||||
QueueInboxEntity item = queryFactory.query(QueueInboxQuery.class).ids(candidate.getId()).first();
|
||||
entityManager = entityManagerFactory.createEntityManager();
|
||||
transaction = entityManager.getTransaction();
|
||||
|
||||
if (item.getRetryCount() != null && item.getRetryCount() >= 1) {
|
||||
Boolean itIsTime = itIsTimeFunc.apply(item);
|
||||
transaction.begin();
|
||||
|
||||
if (!itIsTime) {
|
||||
item.setStatus(candidate.getPreviousState());
|
||||
QueryFactory queryFactory = this.applicationContext.getBean(QueryFactory.class);
|
||||
QueueInboxEntity item = queryFactory.query(QueueInboxQuery.class).ids(candidate.getId()).first();
|
||||
|
||||
entityManager.merge(item);
|
||||
entityManager.flush();
|
||||
success = true;
|
||||
}
|
||||
if (item.getRetryCount() != null && item.getRetryCount() >= 1) {
|
||||
Boolean itIsTime = itIsTimeFunc.apply(item);
|
||||
|
||||
success = !itIsTime;
|
||||
}
|
||||
transaction.commit();
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
if (transaction != null) transaction.rollback();
|
||||
success = false;
|
||||
} finally {
|
||||
if (entityManager != null) entityManager.close();
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
}
|
||||
return success;
|
||||
}
|
||||
if (!itIsTime) {
|
||||
item.setStatus(candidate.getPreviousState());
|
||||
|
||||
@Override
|
||||
public QueueInbox create(InboxCreatorParams inboxCreatorParams) {
|
||||
EntityTransaction transaction = null;
|
||||
EntityManager entityManager = null;
|
||||
Boolean success = false;
|
||||
QueueInboxEntity queueMessage = null;
|
||||
try (FakeRequestScope fakeRequestScope = new FakeRequestScope()) {
|
||||
try {
|
||||
queueMessage = this.createQueueInboxEntity(inboxCreatorParams);
|
||||
EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
|
||||
entityManager.merge(item);
|
||||
entityManager.flush();
|
||||
success = true;
|
||||
}
|
||||
|
||||
entityManager = entityManagerFactory.createEntityManager();
|
||||
transaction = entityManager.getTransaction();
|
||||
success = !itIsTime;
|
||||
}
|
||||
transaction.commit();
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
if (transaction != null)
|
||||
transaction.rollback();
|
||||
success = false;
|
||||
} finally {
|
||||
if (entityManager != null)
|
||||
entityManager.close();
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
}
|
||||
return success;
|
||||
}
|
||||
|
||||
transaction.begin();
|
||||
@Override
|
||||
public QueueInbox create(InboxCreatorParams inboxCreatorParams) {
|
||||
EntityTransaction transaction = null;
|
||||
EntityManager entityManager = null;
|
||||
Boolean success = false;
|
||||
QueueInboxEntity queueMessage = null;
|
||||
try (FakeRequestScope fakeRequestScope = new FakeRequestScope()) {
|
||||
try {
|
||||
queueMessage = this.createQueueInboxEntity(inboxCreatorParams);
|
||||
EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
|
||||
|
||||
entityManager.persist(queueMessage);
|
||||
entityManager.flush();
|
||||
entityManager = entityManagerFactory.createEntityManager();
|
||||
transaction = entityManager.getTransaction();
|
||||
|
||||
transaction.commit();
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
if (transaction != null) transaction.rollback();
|
||||
success = false;
|
||||
} finally {
|
||||
if (entityManager != null) entityManager.close();
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
}
|
||||
return queueMessage;
|
||||
}
|
||||
transaction.begin();
|
||||
|
||||
private QueueInboxEntity createQueueInboxEntity(InboxCreatorParams inboxCreatorParams) {
|
||||
entityManager.persist(queueMessage);
|
||||
entityManager.flush();
|
||||
|
||||
QueueInboxEntity queueMessage = new QueueInboxEntity();
|
||||
queueMessage.setId(UUID.randomUUID());
|
||||
queueMessage.setTenantId(null);
|
||||
queueMessage.setExchange(this.inboxProperties.getExchange());
|
||||
queueMessage.setRoute(inboxCreatorParams.getRoutingKey());
|
||||
queueMessage.setQueue(inboxCreatorParams.getQueueName());
|
||||
queueMessage.setApplicationId(inboxCreatorParams.getAppId());
|
||||
queueMessage.setMessageId(UUID.fromString(inboxCreatorParams.getMessageId()));
|
||||
queueMessage.setMessage(inboxCreatorParams.getMessageBody());
|
||||
queueMessage.setIsActive(IsActive.Active);
|
||||
queueMessage.setStatus(QueueInboxStatus.PENDING);
|
||||
queueMessage.setRetryCount(0);
|
||||
queueMessage.setCreatedAt(Instant.now());
|
||||
transaction.commit();
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
if (transaction != null)
|
||||
transaction.rollback();
|
||||
success = false;
|
||||
} finally {
|
||||
if (entityManager != null)
|
||||
entityManager.close();
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
}
|
||||
return queueMessage;
|
||||
}
|
||||
|
||||
return queueMessage;
|
||||
}
|
||||
private QueueInboxEntity createQueueInboxEntity(InboxCreatorParams inboxCreatorParams) {
|
||||
|
||||
QueueInboxEntity queueMessage = new QueueInboxEntity();
|
||||
queueMessage.setId(UUID.randomUUID());
|
||||
queueMessage.setTenantId(null);
|
||||
queueMessage.setExchange(this.inboxProperties.getExchange());
|
||||
queueMessage.setRoute(inboxCreatorParams.getRoutingKey());
|
||||
queueMessage.setQueue(inboxCreatorParams.getQueueName());
|
||||
queueMessage.setApplicationId(inboxCreatorParams.getAppId());
|
||||
queueMessage.setMessageId(UUID.fromString(inboxCreatorParams.getMessageId()));
|
||||
queueMessage.setMessage(inboxCreatorParams.getMessageBody());
|
||||
queueMessage.setIsActive(IsActive.Active);
|
||||
queueMessage.setStatus(QueueInboxStatus.PENDING);
|
||||
queueMessage.setRetryCount(0);
|
||||
queueMessage.setCreatedAt(Instant.now());
|
||||
|
||||
@Override
|
||||
public Boolean emit(CandidateInfo candidateInfo) {
|
||||
EntityTransaction transaction = null;
|
||||
EntityManager entityManager = null;
|
||||
Boolean success = false;
|
||||
return queueMessage;
|
||||
}
|
||||
|
||||
try (FakeRequestScope fakeRequestScope = new FakeRequestScope()) {
|
||||
try {
|
||||
@Override
|
||||
public Boolean emit(CandidateInfo candidateInfo) {
|
||||
EntityTransaction transaction = null;
|
||||
EntityManager entityManager = null;
|
||||
Boolean success = false;
|
||||
|
||||
EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
|
||||
try (FakeRequestScope fakeRequestScope = new FakeRequestScope()) {
|
||||
try {
|
||||
|
||||
entityManager = entityManagerFactory.createEntityManager();
|
||||
transaction = entityManager.getTransaction();
|
||||
transaction.begin();
|
||||
EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
|
||||
|
||||
QueryFactory queryFactory = this.applicationContext.getBean(QueryFactory.class);
|
||||
QueueInboxEntity queueInboxMessage = queryFactory.query(QueueInboxQuery.class).ids(candidateInfo.getId()).first();
|
||||
entityManager = entityManagerFactory.createEntityManager();
|
||||
transaction = entityManager.getTransaction();
|
||||
transaction.begin();
|
||||
|
||||
if (queueInboxMessage == null) {
|
||||
this.logger.warn("Could not lookup queue inbox {} to process. Continuing...", candidateInfo.getId());
|
||||
} else {
|
||||
QueryFactory queryFactory = this.applicationContext.getBean(QueryFactory.class);
|
||||
QueueInboxEntity queueInboxMessage = queryFactory.query(QueueInboxQuery.class).ids(candidateInfo.getId()).first();
|
||||
|
||||
EventProcessingStatus status = this.processMessage(queueInboxMessage.getRoute(), queueInboxMessage.getMessageId().toString(), queueInboxMessage.getApplicationId(), queueInboxMessage.getMessage());
|
||||
switch (status) {
|
||||
case Success: {
|
||||
queueInboxMessage.setStatus(QueueInboxStatus.SUCCESSFUL);
|
||||
break;
|
||||
}
|
||||
case Postponed: {
|
||||
queueInboxMessage.setStatus(QueueInboxStatus.PARKED);
|
||||
break;
|
||||
}
|
||||
case Error: {
|
||||
queueInboxMessage.setStatus(QueueInboxStatus.ERROR);
|
||||
queueInboxMessage.setRetryCount(queueInboxMessage.getRetryCount() != null ? queueInboxMessage.getRetryCount() + 1 : 0);
|
||||
break;
|
||||
}
|
||||
case Discard:
|
||||
default: {
|
||||
queueInboxMessage.setStatus(QueueInboxStatus.DISCARD);
|
||||
break;
|
||||
}
|
||||
}
|
||||
success = status == EventProcessingStatus.Success;
|
||||
if (queueInboxMessage == null) {
|
||||
this.logger.warn("Could not lookup queue inbox {} to process. Continuing...", candidateInfo.getId());
|
||||
} else {
|
||||
|
||||
entityManager.merge(queueInboxMessage);
|
||||
entityManager.flush();
|
||||
}
|
||||
EventProcessingStatus status = this.processMessage(queueInboxMessage.getRoute(), queueInboxMessage.getMessageId().toString(), queueInboxMessage.getApplicationId(), queueInboxMessage.getMessage());
|
||||
switch (status) {
|
||||
case Success: {
|
||||
queueInboxMessage.setStatus(QueueInboxStatus.SUCCESSFUL);
|
||||
break;
|
||||
}
|
||||
case Postponed: {
|
||||
queueInboxMessage.setStatus(QueueInboxStatus.PARKED);
|
||||
break;
|
||||
}
|
||||
case Error: {
|
||||
queueInboxMessage.setStatus(QueueInboxStatus.ERROR);
|
||||
queueInboxMessage.setRetryCount(queueInboxMessage.getRetryCount() != null ? queueInboxMessage.getRetryCount() + 1 : 0);
|
||||
break;
|
||||
}
|
||||
case Discard:
|
||||
default: {
|
||||
queueInboxMessage.setStatus(QueueInboxStatus.DISCARD);
|
||||
break;
|
||||
}
|
||||
}
|
||||
success = status == EventProcessingStatus.Success;
|
||||
|
||||
transaction.commit();
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
if (transaction != null) transaction.rollback();
|
||||
success = false;
|
||||
} finally {
|
||||
if (entityManager != null) entityManager.close();
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
}
|
||||
return success;
|
||||
}
|
||||
entityManager.merge(queueInboxMessage);
|
||||
entityManager.flush();
|
||||
}
|
||||
|
||||
private EventProcessingStatus processMessage(String routingKey, String messageId, String appId, String message) {
|
||||
IntegrationEventHandler handler;
|
||||
if (this.RoutingKeyMatched(routingKey, this.inboxProperties.getUserRemovalTopic())) handler = this.applicationContext.getBean(UserRemovalIntegrationEventHandler.class);
|
||||
else if (this.RoutingKeyMatched(routingKey, this.inboxProperties.getUserTouchedTopic())) handler = this.applicationContext.getBean(UserTouchedIntegrationEventHandler.class);
|
||||
else handler = null;
|
||||
transaction.commit();
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
if (transaction != null)
|
||||
transaction.rollback();
|
||||
success = false;
|
||||
} finally {
|
||||
if (entityManager != null)
|
||||
entityManager.close();
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
}
|
||||
return success;
|
||||
}
|
||||
|
||||
if (handler == null) return EventProcessingStatus.Discard;
|
||||
private EventProcessingStatus processMessage(String routingKey, String messageId, String appId, String message) {
|
||||
IntegrationEventHandler handler = null;
|
||||
|
||||
IntegrationEventProperties properties = new IntegrationEventProperties();
|
||||
properties.setAppId(appId);
|
||||
properties.setMessageId(messageId);
|
||||
if (handler == null)
|
||||
return EventProcessingStatus.Discard;
|
||||
|
||||
TrackedEvent event = this.jsonHandlingService.fromJsonSafe(TrackedEvent.class, message);
|
||||
IntegrationEventProperties properties = new IntegrationEventProperties();
|
||||
properties.setAppId(appId);
|
||||
properties.setMessageId(messageId);
|
||||
|
||||
TrackedEvent event = this.jsonHandlingService.fromJsonSafe(TrackedEvent.class, message);
|
||||
// using (LogContext.PushProperty(this._logTrackingConfig.LogTrackingContextName, @event.TrackingContextTag))
|
||||
// {
|
||||
try {
|
||||
return handler.handle(properties, message);
|
||||
} catch (Exception ex) {
|
||||
logger.error("problem handling event from routing key " + routingKey + ". Setting nack and continuing...", ex);
|
||||
return EventProcessingStatus.Error;
|
||||
}
|
||||
try {
|
||||
return handler.handle(properties, message);
|
||||
} catch (Exception ex) {
|
||||
logger.error("problem handling event from routing key " + routingKey + ". Setting nack and continuing...", ex);
|
||||
return EventProcessingStatus.Error;
|
||||
}
|
||||
// }
|
||||
}
|
||||
}
|
||||
|
||||
private Boolean RoutingKeyMatched(String routingKey, List<String> topics) {
|
||||
if (topics == null || topics.size() == 0) return false;
|
||||
return topics.stream().anyMatch(x -> x.equals(routingKey));
|
||||
}
|
||||
private Boolean RoutingKeyMatched(String routingKey, List<String> topics) {
|
||||
if (topics == null || topics.size() == 0)
|
||||
return false;
|
||||
return topics.stream().anyMatch(x -> x.equals(routingKey));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,5 +1,7 @@
|
|||
package eu.eudat.integrationevent.inbox;
|
||||
|
||||
public interface IntegrationEventHandler {
|
||||
EventProcessingStatus handle(IntegrationEventProperties properties, String message);
|
||||
|
||||
EventProcessingStatus handle(IntegrationEventProperties properties, String message);
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,22 +1,25 @@
|
|||
package eu.eudat.integrationevent.inbox;
|
||||
|
||||
public class IntegrationEventProperties {
|
||||
private String messageId;
|
||||
private String appId;
|
||||
|
||||
public String getMessageId() {
|
||||
return messageId;
|
||||
}
|
||||
private String messageId;
|
||||
|
||||
public void setMessageId(String messageId) {
|
||||
this.messageId = messageId;
|
||||
}
|
||||
private String appId;
|
||||
|
||||
public String getAppId() {
|
||||
return appId;
|
||||
}
|
||||
public String getMessageId() {
|
||||
return messageId;
|
||||
}
|
||||
|
||||
public void setMessageId(String messageId) {
|
||||
this.messageId = messageId;
|
||||
}
|
||||
|
||||
public String getAppId() {
|
||||
return appId;
|
||||
}
|
||||
|
||||
public void setAppId(String appId) {
|
||||
this.appId = appId;
|
||||
}
|
||||
|
||||
public void setAppId(String appId) {
|
||||
this.appId = appId;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,26 +0,0 @@
|
|||
package eu.eudat.integrationevent.inbox.userremoval;
|
||||
|
||||
import eu.eudat.integrationevent.inbox.ConsistencyHandler;
|
||||
import eu.eudat.query.UserQuery;
|
||||
import gr.cite.tools.data.query.QueryFactory;
|
||||
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
|
||||
import org.springframework.context.annotation.Scope;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
@Component("inboxuserremovalconsistencyhandler")
|
||||
@Scope(value = ConfigurableBeanFactory.SCOPE_PROTOTYPE)
|
||||
public class UserRemovalConsistencyHandler implements ConsistencyHandler<UserRemovalConsistencyPredicates> {
|
||||
|
||||
private final QueryFactory queryFactory;
|
||||
|
||||
public UserRemovalConsistencyHandler(QueryFactory queryFactory) {
|
||||
this.queryFactory = queryFactory;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Boolean isConsistent(UserRemovalConsistencyPredicates consistencyPredicates) {
|
||||
long count = this.queryFactory.query(UserQuery.class).ids(consistencyPredicates.getUserId()).count();
|
||||
if (count == 0) return false;
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,21 +0,0 @@
|
|||
package eu.eudat.integrationevent.inbox.userremoval;
|
||||
|
||||
import eu.eudat.integrationevent.inbox.ConsistencyPredicates;
|
||||
|
||||
import java.util.UUID;
|
||||
|
||||
public class UserRemovalConsistencyPredicates implements ConsistencyPredicates {
|
||||
public UserRemovalConsistencyPredicates(UUID userId) {
|
||||
this.userId = userId;
|
||||
}
|
||||
|
||||
private UUID userId;
|
||||
|
||||
public UUID getUserId() {
|
||||
return userId;
|
||||
}
|
||||
|
||||
public void setUserId(UUID userId) {
|
||||
this.userId = userId;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,26 +0,0 @@
|
|||
package eu.eudat.integrationevent.inbox.userremoval;
|
||||
|
||||
import eu.eudat.integrationevent.TrackedEvent;
|
||||
|
||||
import java.util.UUID;
|
||||
|
||||
public class UserRemovalIntegrationEvent extends TrackedEvent {
|
||||
private UUID userId;
|
||||
private UUID tenant;
|
||||
|
||||
public UUID getUserId() {
|
||||
return userId;
|
||||
}
|
||||
|
||||
public void setUserId(UUID userId) {
|
||||
this.userId = userId;
|
||||
}
|
||||
|
||||
public UUID getTenant() {
|
||||
return tenant;
|
||||
}
|
||||
|
||||
public void setTenant(UUID tenant) {
|
||||
this.tenant = tenant;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,6 +0,0 @@
|
|||
package eu.eudat.integrationevent.inbox.userremoval;
|
||||
|
||||
import eu.eudat.integrationevent.inbox.IntegrationEventHandler;
|
||||
|
||||
public interface UserRemovalIntegrationEventHandler extends IntegrationEventHandler {
|
||||
}
|
||||
|
|
@ -1,125 +0,0 @@
|
|||
package eu.eudat.integrationevent.inbox.userremoval;
|
||||
|
||||
import eu.eudat.audit.AuditableAction;
|
||||
import eu.eudat.commons.JsonHandlingService;
|
||||
import eu.eudat.commons.fake.FakeRequestScope;
|
||||
import eu.eudat.data.TenantEntity;
|
||||
import eu.eudat.errorcode.ErrorThesaurusProperties;
|
||||
import eu.eudat.integrationevent.inbox.EventProcessingStatus;
|
||||
import eu.eudat.integrationevent.inbox.InboxPrincipal;
|
||||
import eu.eudat.integrationevent.inbox.IntegrationEventProperties;
|
||||
import eu.eudat.query.TenantQuery;
|
||||
import eu.eudat.service.user.UserService;
|
||||
import gr.cite.commons.web.oidc.principal.CurrentPrincipalResolver;
|
||||
|
||||
import gr.cite.tools.auditing.AuditService;
|
||||
import gr.cite.tools.data.query.QueryFactory;
|
||||
import gr.cite.tools.exception.MyValidationException;
|
||||
import gr.cite.tools.fieldset.BaseFieldSet;
|
||||
import gr.cite.tools.logging.LoggerService;
|
||||
import jakarta.persistence.*;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.context.MessageSource;
|
||||
import org.springframework.context.annotation.Scope;
|
||||
import org.springframework.context.i18n.LocaleContextHolder;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
|
||||
import java.util.AbstractMap;
|
||||
import java.util.Map;
|
||||
|
||||
@Component("inboxuserremovalintegrationeventhandler")
|
||||
@Scope(value = ConfigurableBeanFactory.SCOPE_PROTOTYPE)
|
||||
public class UserRemovalIntegrationEventHandlerImpl implements UserRemovalIntegrationEventHandler {
|
||||
private static final LoggerService logger = new LoggerService(LoggerFactory.getLogger(UserRemovalIntegrationEventHandlerImpl.class));
|
||||
private final JsonHandlingService jsonHandlingService;
|
||||
protected final ApplicationContext applicationContext;
|
||||
private final ErrorThesaurusProperties errors;
|
||||
private final MessageSource messageSource;
|
||||
|
||||
public UserRemovalIntegrationEventHandlerImpl(
|
||||
JsonHandlingService jsonHandlingService,
|
||||
ApplicationContext applicationContext,
|
||||
ErrorThesaurusProperties errors,
|
||||
MessageSource messageSource
|
||||
) {
|
||||
this.jsonHandlingService = jsonHandlingService;
|
||||
this.applicationContext = applicationContext;
|
||||
this.errors = errors;
|
||||
this.messageSource = messageSource;
|
||||
}
|
||||
|
||||
@Override
|
||||
public EventProcessingStatus handle(IntegrationEventProperties properties, String message) {
|
||||
UserRemovalIntegrationEvent event = this.jsonHandlingService.fromJsonSafe(UserRemovalIntegrationEvent.class, message);
|
||||
if (event == null) return EventProcessingStatus.Error;
|
||||
if (event.getUserId() == null) throw new MyValidationException(this.errors.getModelValidation().getCode(), "userId", messageSource.getMessage("Validation_Required", new Object[]{"userId"}, LocaleContextHolder.getLocale()));
|
||||
|
||||
EntityManager entityManager = null;
|
||||
EntityTransaction transaction = null;
|
||||
try (FakeRequestScope fakeRequestScope = new FakeRequestScope()) {
|
||||
try {
|
||||
QueryFactory queryFactory = this.applicationContext.getBean(QueryFactory.class);
|
||||
// TenantScope scope = this.applicationContext.getBean(TenantScope.class);
|
||||
// if (scope.isMultitenant() && event.getTenant() != null) {
|
||||
// TenantEntity tenant = queryFactory.query(TenantQuery.class).ids(event.getTenant()).firstAs(new BaseFieldSet().ensure(Tenant._id).ensure(Tenant._code));
|
||||
// if (tenant == null) {
|
||||
// logger.error("missing tenant from event message");
|
||||
// return EventProcessingStatus.Error;
|
||||
// }
|
||||
// scope.setTenant(event.getTenant(), tenant.getCode());
|
||||
// } else if (scope.isMultitenant()) {
|
||||
// logger.error("missing tenant from event message");
|
||||
// return EventProcessingStatus.Error;
|
||||
// }
|
||||
|
||||
CurrentPrincipalResolver currentPrincipalResolver = this.applicationContext.getBean(CurrentPrincipalResolver.class);
|
||||
currentPrincipalResolver.push(InboxPrincipal.build(properties));
|
||||
|
||||
UserRemovalConsistencyHandler userRemovalConsistencyHandler = this.applicationContext.getBean(UserRemovalConsistencyHandler.class);
|
||||
if (!(userRemovalConsistencyHandler.isConsistent(new UserRemovalConsistencyPredicates(event.getUserId())))) return EventProcessingStatus.Postponed;
|
||||
|
||||
EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
|
||||
entityManager = entityManagerFactory.createEntityManager();
|
||||
|
||||
transaction = entityManager.getTransaction();
|
||||
transaction.begin();
|
||||
|
||||
try {
|
||||
UserService userService = this.applicationContext.getBean(UserService.class);
|
||||
userService.deleteAndSave(event.getUserId());
|
||||
|
||||
AuditService auditService = this.applicationContext.getBean(AuditService.class);
|
||||
|
||||
auditService.track(AuditableAction.User_Delete, Map.ofEntries(
|
||||
new AbstractMap.SimpleEntry<String, Object>("id", event.getUserId())
|
||||
));
|
||||
//auditService.trackIdentity(AuditableAction.IdentityTracking_Action);
|
||||
|
||||
transaction.commit();
|
||||
} catch (Exception e) {
|
||||
transaction.rollback();
|
||||
throw e;
|
||||
} finally {
|
||||
currentPrincipalResolver.pop();
|
||||
}
|
||||
|
||||
transaction.commit();
|
||||
} catch (OptimisticLockException ex) {
|
||||
// we get this if/when someone else already modified the notifications. We want to essentially ignore this, and keep working
|
||||
this.logger.debug("Concurrency exception getting queue outbox. Skipping: {} ", ex.getMessage());
|
||||
if (transaction != null) transaction.rollback();
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem getting list of queue outbox. Skipping: {}", ex.getMessage(), ex);
|
||||
if (transaction != null) transaction.rollback();
|
||||
} finally {
|
||||
if (entityManager != null) entityManager.close();
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem getting list of queue outbox. Skipping: {}", ex.getMessage(), ex);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,45 +0,0 @@
|
|||
package eu.eudat.integrationevent.inbox.usertouched;
|
||||
|
||||
|
||||
import eu.eudat.integrationevent.TrackedEvent;
|
||||
|
||||
import java.util.UUID;
|
||||
|
||||
public class UserTouchedIntegrationEvent extends TrackedEvent {
|
||||
private UUID id;
|
||||
private UUID tenant;
|
||||
private String firstName;
|
||||
private String lastName;
|
||||
|
||||
public UUID getId() {
|
||||
return id;
|
||||
}
|
||||
|
||||
public void setId(UUID id) {
|
||||
this.id = id;
|
||||
}
|
||||
|
||||
public UUID getTenant() {
|
||||
return tenant;
|
||||
}
|
||||
|
||||
public void setTenant(UUID tenant) {
|
||||
this.tenant = tenant;
|
||||
}
|
||||
|
||||
public String getFirstName() {
|
||||
return firstName;
|
||||
}
|
||||
|
||||
public void setFirstName(String firstName) {
|
||||
this.firstName = firstName;
|
||||
}
|
||||
|
||||
public String getLastName() {
|
||||
return lastName;
|
||||
}
|
||||
|
||||
public void setLastName(String lastName) {
|
||||
this.lastName = lastName;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,6 +0,0 @@
|
|||
package eu.eudat.integrationevent.inbox.usertouched;
|
||||
|
||||
import eu.eudat.integrationevent.inbox.IntegrationEventHandler;
|
||||
|
||||
public interface UserTouchedIntegrationEventHandler extends IntegrationEventHandler {
|
||||
}
|
||||
|
|
@ -1,114 +0,0 @@
|
|||
package eu.eudat.integrationevent.inbox.usertouched;
|
||||
|
||||
import eu.eudat.audit.AuditableAction;
|
||||
import eu.eudat.commons.JsonHandlingService;
|
||||
import eu.eudat.commons.fake.FakeRequestScope;
|
||||
import eu.eudat.integrationevent.inbox.EventProcessingStatus;
|
||||
import eu.eudat.integrationevent.inbox.InboxPrincipal;
|
||||
import eu.eudat.integrationevent.inbox.IntegrationEventProperties;
|
||||
import eu.eudat.service.user.UserService;
|
||||
import gr.cite.commons.web.oidc.principal.CurrentPrincipalResolver;
|
||||
import gr.cite.tools.auditing.AuditService;
|
||||
import gr.cite.tools.data.query.QueryFactory;
|
||||
import gr.cite.tools.logging.LoggerService;
|
||||
import jakarta.persistence.*;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.context.annotation.Scope;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.AbstractMap;
|
||||
import java.util.Map;
|
||||
|
||||
@Component("inboxusertouchedintegrationeventhandler")
|
||||
@Scope(value = ConfigurableBeanFactory.SCOPE_PROTOTYPE)
|
||||
public class UserTouchedIntegrationEventHandlerImpl implements UserTouchedIntegrationEventHandler {
|
||||
private static final LoggerService logger = new LoggerService(LoggerFactory.getLogger(UserTouchedIntegrationEventHandlerImpl.class));
|
||||
private final JsonHandlingService jsonHandlingService;
|
||||
protected final ApplicationContext applicationContext;
|
||||
|
||||
public UserTouchedIntegrationEventHandlerImpl(
|
||||
JsonHandlingService jsonHandlingService,
|
||||
ApplicationContext applicationContext
|
||||
) {
|
||||
this.jsonHandlingService = jsonHandlingService;
|
||||
this.applicationContext = applicationContext;
|
||||
}
|
||||
|
||||
@Override
|
||||
public EventProcessingStatus handle(IntegrationEventProperties properties, String message) {
|
||||
UserTouchedIntegrationEvent event = this.jsonHandlingService.fromJsonSafe(UserTouchedIntegrationEvent.class, message);
|
||||
if (event == null) return EventProcessingStatus.Error;
|
||||
|
||||
// UserTouchedIntegrationEventPersist model = new UserTouchedIntegrationEventPersist();
|
||||
// model.setId(event.getId());
|
||||
// model.setFirstName(event.getFirstName());
|
||||
// model.setLastName(event.getLastName());
|
||||
|
||||
EntityManager entityManager = null;
|
||||
EntityTransaction transaction = null;
|
||||
try (FakeRequestScope fakeRequestScope = new FakeRequestScope()) {
|
||||
try {
|
||||
QueryFactory queryFactory = this.applicationContext.getBean(QueryFactory.class);
|
||||
// TenantScope scope = this.applicationContext.getBean(TenantScope.class);
|
||||
// if (scope.isMultitenant() && event.getTenant() != null) {
|
||||
// TenantEntity tenant = queryFactory.query(TenantQuery.class).ids(event.getTenant()).firstAs(new BaseFieldSet().ensure(Tenant._id).ensure(Tenant._code));
|
||||
// if (tenant == null) {
|
||||
// logger.error("missing tenant from event message");
|
||||
// return EventProcessingStatus.Error;
|
||||
// }
|
||||
// scope.setTenant(event.getTenant(), tenant.getCode());
|
||||
// } else if (scope.isMultitenant()) {
|
||||
// logger.error("missing tenant from event message");
|
||||
// return EventProcessingStatus.Error;
|
||||
// }
|
||||
//
|
||||
// ValidationService validator = this.applicationContext.getBean(ValidationService.class);
|
||||
// validator.validateForce(model);
|
||||
|
||||
|
||||
CurrentPrincipalResolver currentPrincipalResolver = this.applicationContext.getBean(CurrentPrincipalResolver.class);
|
||||
currentPrincipalResolver.push(InboxPrincipal.build(properties));
|
||||
|
||||
EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
|
||||
entityManager = entityManagerFactory.createEntityManager();
|
||||
|
||||
transaction = entityManager.getTransaction();
|
||||
transaction.begin();
|
||||
|
||||
try {
|
||||
UserService userService = this.applicationContext.getBean(UserService.class);
|
||||
// userService.persist(model, null);
|
||||
|
||||
AuditService auditService = this.applicationContext.getBean(AuditService.class);
|
||||
|
||||
auditService.track(AuditableAction.User_Persist, Map.ofEntries(
|
||||
// new AbstractMap.SimpleEntry<String, Object>("model", model)
|
||||
));
|
||||
|
||||
transaction.commit();
|
||||
} catch (Exception e) {
|
||||
transaction.rollback();
|
||||
throw e;
|
||||
} finally {
|
||||
currentPrincipalResolver.pop();
|
||||
}
|
||||
|
||||
transaction.commit();
|
||||
} catch (OptimisticLockException ex) {
|
||||
// we get this if/when someone else already modified the notifications. We want to essentially ignore this, and keep working
|
||||
this.logger.debug("Concurrency exception getting queue outbox. Skipping: {} ", ex.getMessage());
|
||||
if (transaction != null) transaction.rollback();
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem getting list of queue outbox. Skipping: {}", ex.getMessage(), ex);
|
||||
if (transaction != null) transaction.rollback();
|
||||
} finally {
|
||||
if (entityManager != null) entityManager.close();
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
this.logger.error("Problem getting list of queue outbox. Skipping: {}", ex.getMessage(), ex);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,9 +1,8 @@
|
|||
package eu.eudat.integrationevent.outbox.usertouched;
|
||||
|
||||
import eu.eudat.commons.enums.notification.TentativeUserProfile;
|
||||
import eu.eudat.commons.enums.ContactInfoType;
|
||||
import eu.eudat.commons.enums.notification.UserType;
|
||||
import eu.eudat.integrationevent.TrackedEvent;
|
||||
import eu.eudat.model.UserContactInfo;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
|
|
@ -82,22 +81,12 @@ public class UserTouchedIntegrationEvent extends TrackedEvent {
|
|||
|
||||
public static class UserProfile {
|
||||
|
||||
private TentativeUserProfile isTentative;
|
||||
|
||||
private String timezone;
|
||||
|
||||
private String culture;
|
||||
|
||||
private String language;
|
||||
|
||||
public TentativeUserProfile getIsTentative() {
|
||||
return isTentative;
|
||||
}
|
||||
|
||||
public void setIsTentative(TentativeUserProfile isTentative) {
|
||||
this.isTentative = isTentative;
|
||||
}
|
||||
|
||||
public String getTimezone() {
|
||||
return timezone;
|
||||
}
|
||||
|
|
@ -123,4 +112,27 @@ public class UserTouchedIntegrationEvent extends TrackedEvent {
|
|||
}
|
||||
}
|
||||
|
||||
public static class UserContactInfo {
|
||||
|
||||
private ContactInfoType type;
|
||||
|
||||
private String value;
|
||||
|
||||
public ContactInfoType getType() {
|
||||
return type;
|
||||
}
|
||||
|
||||
public void setType(ContactInfoType type) {
|
||||
this.type = type;
|
||||
}
|
||||
|
||||
public String getValue() {
|
||||
return value;
|
||||
}
|
||||
|
||||
public void setValue(String value) {
|
||||
this.value = value;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,10 +1,10 @@
|
|||
package eu.eudat.interceptors;
|
||||
|
||||
|
||||
import eu.eudat.authorization.ClaimNames;
|
||||
import eu.eudat.commons.JsonHandlingService;
|
||||
import eu.eudat.commons.enums.ContactInfoType;
|
||||
import eu.eudat.commons.enums.IsActive;
|
||||
import eu.eudat.commons.enums.notification.UserType;
|
||||
import eu.eudat.commons.lock.LockByKeyManager;
|
||||
import eu.eudat.commons.scope.user.UserScope;
|
||||
import eu.eudat.commons.types.user.AdditionalInfoEntity;
|
||||
|
|
@ -14,6 +14,9 @@ import eu.eudat.data.UserContactInfoEntity;
|
|||
import eu.eudat.data.UserCredentialEntity;
|
||||
import eu.eudat.data.UserEntity;
|
||||
import eu.eudat.data.UserRoleEntity;
|
||||
import eu.eudat.integrationevent.outbox.OutboxIntegrationEvent;
|
||||
import eu.eudat.integrationevent.outbox.OutboxService;
|
||||
import eu.eudat.integrationevent.outbox.usertouched.UserTouchedIntegrationEvent;
|
||||
import eu.eudat.model.UserContactInfo;
|
||||
import eu.eudat.model.UserCredential;
|
||||
import eu.eudat.model.UserRole;
|
||||
|
|
@ -29,6 +32,7 @@ import gr.cite.tools.logging.LoggerService;
|
|||
import jakarta.persistence.EntityManager;
|
||||
import jakarta.persistence.PersistenceContext;
|
||||
import org.apache.commons.validator.routines.EmailValidator;
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.lang.NonNull;
|
||||
|
|
@ -49,328 +53,415 @@ import java.util.concurrent.TimeUnit;
|
|||
|
||||
@Component
|
||||
public class UserInterceptor implements WebRequestInterceptor {
|
||||
private static final LoggerService logger = new LoggerService(LoggerFactory.getLogger(UserInterceptor.class));
|
||||
private final UserScope userScope;
|
||||
private final ClaimExtractor claimExtractor;
|
||||
private final CurrentPrincipalResolver currentPrincipalResolver;
|
||||
private final PlatformTransactionManager transactionManager;
|
||||
private final UserInterceptorCacheService userInterceptorCacheService;
|
||||
private final JsonHandlingService jsonHandlingService;
|
||||
private final QueryFactory queryFactory;
|
||||
private final LockByKeyManager lockByKeyManager;
|
||||
private final LocaleProperties localeProperties;
|
||||
@PersistenceContext
|
||||
public EntityManager entityManager;
|
||||
|
||||
@Autowired
|
||||
public UserInterceptor(
|
||||
UserScope userScope,
|
||||
ClaimExtractor claimExtractor,
|
||||
CurrentPrincipalResolver currentPrincipalResolver,
|
||||
PlatformTransactionManager transactionManager,
|
||||
UserInterceptorCacheService userInterceptorCacheService,
|
||||
JsonHandlingService jsonHandlingService,
|
||||
QueryFactory queryFactory,
|
||||
LockByKeyManager lockByKeyManager,
|
||||
LocaleProperties localeProperties) {
|
||||
this.userScope = userScope;
|
||||
this.currentPrincipalResolver = currentPrincipalResolver;
|
||||
this.claimExtractor = claimExtractor;
|
||||
this.transactionManager = transactionManager;
|
||||
this.userInterceptorCacheService = userInterceptorCacheService;
|
||||
this.jsonHandlingService = jsonHandlingService;
|
||||
this.queryFactory = queryFactory;
|
||||
this.lockByKeyManager = lockByKeyManager;
|
||||
this.localeProperties = localeProperties;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void preHandle(WebRequest request) throws InterruptedException {
|
||||
UUID userId = null;
|
||||
if (this.currentPrincipalResolver.currentPrincipal().isAuthenticated()) {
|
||||
String subjectId = this.claimExtractor.subjectString(this.currentPrincipalResolver.currentPrincipal());
|
||||
if (subjectId == null || subjectId.isBlank()) throw new MyForbiddenException("Empty subjects not allowed");
|
||||
private static final LoggerService logger = new LoggerService(LoggerFactory.getLogger(UserInterceptor.class));
|
||||
|
||||
UserInterceptorCacheService.UserInterceptorCacheValue cacheValue = this.userInterceptorCacheService.lookup(this.userInterceptorCacheService.buildKey(subjectId));
|
||||
if (cacheValue != null && emailExistsToPrincipal(cacheValue.getProviderEmail()) && userRolesSynced(cacheValue.getRoles()) && providerExistsToPrincipal(cacheValue.getExternalProviderNames())) {
|
||||
userId = cacheValue.getUserId();
|
||||
} else {
|
||||
boolean usedResource = false;
|
||||
try {
|
||||
usedResource = this.lockByKeyManager.tryLock(subjectId, 5000, TimeUnit.MILLISECONDS);
|
||||
String email = this.getEmailFromClaims();
|
||||
private final UserScope userScope;
|
||||
|
||||
DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
|
||||
definition.setName(UUID.randomUUID().toString());
|
||||
definition.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
|
||||
definition.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
|
||||
TransactionStatus status = null;
|
||||
try {
|
||||
status = transactionManager.getTransaction(definition);
|
||||
|
||||
userId = this.findExistingUserFromDb(subjectId);
|
||||
boolean isNewUser = userId == null;
|
||||
if (isNewUser) {
|
||||
UserEntity user = this.addNewUser(subjectId, email);
|
||||
userId = user.getId();
|
||||
}
|
||||
this.entityManager.flush();
|
||||
|
||||
if (!isNewUser){
|
||||
this.syncUserWithClaims(userId, subjectId);
|
||||
}
|
||||
private final ClaimExtractor claimExtractor;
|
||||
|
||||
this.entityManager.flush();
|
||||
transactionManager.commit(status);
|
||||
} catch (Exception ex) {
|
||||
if (status != null) transactionManager.rollback(status);
|
||||
throw ex;
|
||||
}
|
||||
|
||||
cacheValue = new UserInterceptorCacheService.UserInterceptorCacheValue(subjectId, userId);
|
||||
cacheValue.setRoles(this.getRolesFromClaims());
|
||||
if (email != null && !email.isBlank()) cacheValue.setProviderEmail(email);
|
||||
UserCredentialEntity userCredential = this.queryFactory.query(UserCredentialQuery.class).externalIds(subjectId).firstAs(new BaseFieldSet().ensure(UserCredential._data));
|
||||
if (userCredential != null && userCredential.getData() != null){
|
||||
UserCredentialDataEntity userCredentialDataEntity = this.jsonHandlingService.fromJsonSafe(UserCredentialDataEntity.class, userCredential.getData());
|
||||
if (userCredentialDataEntity != null) cacheValue.setExternalProviderNames(userCredentialDataEntity.getExternalProviderNames());
|
||||
}
|
||||
private final CurrentPrincipalResolver currentPrincipalResolver;
|
||||
|
||||
this.userInterceptorCacheService.put(cacheValue);
|
||||
} finally {
|
||||
if (usedResource) this.lockByKeyManager.unlock(subjectId);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
this.userScope.setUserId(userId);
|
||||
}
|
||||
|
||||
private void syncUserWithClaims(UUID userId, String subjectId){
|
||||
List<String> existingUserEmails = this.collectUserEmails(userId);
|
||||
if (!this.containsPrincipalEmail(existingUserEmails)){
|
||||
String email = this.getEmailFromClaims();
|
||||
long contactUsedByOthersCount = this.queryFactory.query(UserContactInfoQuery.class).excludedUserIds(userId).types(ContactInfoType.Email).values(email).count();
|
||||
if (contactUsedByOthersCount > 0) {
|
||||
logger.warn("user contact exists to other user" + email);
|
||||
} else {
|
||||
Long emailContactsCount = this.queryFactory.query(UserContactInfoQuery.class).userIds(userId).types(ContactInfoType.Email).count();
|
||||
UserContactInfoEntity contactInfo = this.buildEmailContact(userId, email);
|
||||
contactInfo.setOrdinal(emailContactsCount.intValue());
|
||||
this.entityManager.persist(contactInfo);
|
||||
}
|
||||
}
|
||||
private final PlatformTransactionManager transactionManager;
|
||||
|
||||
List<String> existingUserRoles = this.collectUserRoles(userId);
|
||||
if (!this.userRolesSynced(existingUserRoles)){
|
||||
this.syncRoles(userId);
|
||||
}
|
||||
private final UserInterceptorCacheService userInterceptorCacheService;
|
||||
|
||||
UserCredentialEntity userCredential = this.queryFactory.query(UserCredentialQuery.class).externalIds(subjectId).first();
|
||||
if (userCredential == null) {
|
||||
throw new MyForbiddenException("UserCredential not found");
|
||||
}
|
||||
else {
|
||||
boolean updatedUserCredential = false;
|
||||
UserCredentialDataEntity userCredentialDataEntity = this.jsonHandlingService.fromJsonSafe(UserCredentialDataEntity.class, userCredential.getData());
|
||||
if (userCredentialDataEntity == null) userCredentialDataEntity = new UserCredentialDataEntity();
|
||||
if (userCredentialDataEntity.getExternalProviderNames() == null) userCredentialDataEntity.setExternalProviderNames(new ArrayList<>());
|
||||
|
||||
String email = this.getEmailFromClaims();
|
||||
String provider = this.getProviderFromClaims();
|
||||
|
||||
if (email != null && !email.equalsIgnoreCase(userCredentialDataEntity.getEmail())) {
|
||||
userCredentialDataEntity.setEmail(email);
|
||||
updatedUserCredential = true;
|
||||
}
|
||||
if (provider != null && !provider.isBlank() && userCredentialDataEntity.getExternalProviderNames().stream().noneMatch(provider::equalsIgnoreCase)) {
|
||||
userCredentialDataEntity.getExternalProviderNames().add(provider);
|
||||
updatedUserCredential = true;
|
||||
}
|
||||
if (updatedUserCredential) {
|
||||
userCredential.setData(this.jsonHandlingService.toJsonSafe(userCredentialDataEntity));
|
||||
this.entityManager.persist(userCredential);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private UUID findExistingUserFromDb(String subjectId){
|
||||
UserCredentialEntity userCredential = this.queryFactory.query(UserCredentialQuery.class).externalIds(subjectId).firstAs(new BaseFieldSet().ensure(UserCredential._user));
|
||||
if (userCredential != null) {
|
||||
return userCredential.getUserId();
|
||||
} else {
|
||||
String email = this.getEmailFromClaims();
|
||||
if (email != null && !email.isBlank()) {
|
||||
UserContactInfoEntity userContactInfo = this.queryFactory.query(UserContactInfoQuery.class).types(ContactInfoType.Email).values(email).firstAs(new BaseFieldSet().ensure(UserContactInfo._user));
|
||||
if (userContactInfo != null) {
|
||||
UserCredentialEntity credential = this.buildCredential(userContactInfo.getUserId(), subjectId);
|
||||
this.entityManager.persist(credential);
|
||||
private final JsonHandlingService jsonHandlingService;
|
||||
|
||||
return credential.getUserId();
|
||||
}
|
||||
} else {
|
||||
throw new MyForbiddenException("Email is required");
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private List<String> getRolesFromClaims(){
|
||||
List<String> claimsRoles = claimExtractor.roles(currentPrincipalResolver.currentPrincipal());
|
||||
if (claimsRoles == null) claimsRoles = new ArrayList<>();
|
||||
claimsRoles = claimsRoles.stream().filter(x-> x != null && !x.isBlank()).distinct().toList();
|
||||
return claimsRoles;
|
||||
}
|
||||
|
||||
private void syncRoles(UUID userId){
|
||||
List<UserRoleEntity> existingUserRoles = this.queryFactory.query(UserRoleQuery.class).userIds(userId).collect();
|
||||
List<UUID> foundRoles = new ArrayList<>();
|
||||
for (String claimRole : this.getRolesFromClaims()) {
|
||||
UserRoleEntity roleEntity = existingUserRoles.stream().filter(x-> x.getRole().equals(claimRole)).findFirst().orElse(null);
|
||||
if (roleEntity == null) {
|
||||
roleEntity = this.buildRole(userId, claimRole);
|
||||
this.entityManager.persist(roleEntity);
|
||||
}
|
||||
foundRoles.add(roleEntity.getId());
|
||||
}
|
||||
for (UserRoleEntity existing: existingUserRoles) {
|
||||
if (!foundRoles.contains(existing.getId())){
|
||||
this.entityManager.remove(existing);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private List<String> collectUserRoles(UUID userId){
|
||||
List<UserRoleEntity> items = this.queryFactory.query(UserRoleQuery.class).userIds(userId).collectAs(new BaseFieldSet().ensure(UserRole._role));
|
||||
return items == null ? new ArrayList<>() : items.stream().map(UserRoleEntity::getRole).toList();
|
||||
}
|
||||
private final QueryFactory queryFactory;
|
||||
|
||||
private List<String> collectUserEmails(UUID userId){
|
||||
List<UserContactInfoEntity> items = this.queryFactory.query(UserContactInfoQuery.class).userIds(userId).types(ContactInfoType.Email).collectAs(new BaseFieldSet().ensure(UserContactInfo._value));
|
||||
return items == null ? new ArrayList<>() : items.stream().map(UserContactInfoEntity::getValue).toList();
|
||||
}
|
||||
|
||||
private boolean containsPrincipalEmail(List<String> existingUserEmails){
|
||||
String email = this.getEmailFromClaims();
|
||||
return email == null || email.isBlank() ||
|
||||
(existingUserEmails != null && existingUserEmails.stream().anyMatch(email::equals));
|
||||
}
|
||||
private boolean emailExistsToPrincipal(String existingUserEmail){
|
||||
String email = this.getEmailFromClaims();
|
||||
return email == null || email.isBlank() || email.equalsIgnoreCase(existingUserEmail);
|
||||
}
|
||||
private final LockByKeyManager lockByKeyManager;
|
||||
|
||||
private boolean providerExistsToPrincipal(List<String> principalCredentialProviders){
|
||||
String provider = this.getProviderFromClaims();
|
||||
return provider == null || provider.isBlank() ||
|
||||
(principalCredentialProviders != null && principalCredentialProviders.stream().anyMatch(provider::equalsIgnoreCase));
|
||||
}
|
||||
private final LocaleProperties localeProperties;
|
||||
|
||||
private final OutboxService outboxService;
|
||||
|
||||
private boolean userRolesSynced(List<String> existingUserRoles){
|
||||
List<String> claimsRoles = this.getRolesFromClaims();
|
||||
if (existingUserRoles == null) existingUserRoles = new ArrayList<>();
|
||||
existingUserRoles = existingUserRoles.stream().filter(x-> x != null && !x.isBlank()).distinct().toList();
|
||||
if (claimsRoles.size() != existingUserRoles.size()) return false;
|
||||
|
||||
for (String claim : claimsRoles ) {
|
||||
if (existingUserRoles.stream().noneMatch(claim::equalsIgnoreCase)) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private String getEmailFromClaims(){
|
||||
String email = this.claimExtractor.email(this.currentPrincipalResolver.currentPrincipal());
|
||||
if (email == null || email.isBlank() || !EmailValidator.getInstance().isValid(email)) return null;
|
||||
return email.trim();
|
||||
}
|
||||
@PersistenceContext
|
||||
public EntityManager entityManager;
|
||||
|
||||
private String getProviderFromClaims(){
|
||||
String provider = this.claimExtractor.asString(this.currentPrincipalResolver.currentPrincipal(), ClaimNames.ExternalProviderName);
|
||||
if (provider == null || provider.isBlank()) return null;
|
||||
return provider.trim();
|
||||
}
|
||||
@Autowired
|
||||
public UserInterceptor(
|
||||
UserScope userScope,
|
||||
ClaimExtractor claimExtractor,
|
||||
CurrentPrincipalResolver currentPrincipalResolver,
|
||||
PlatformTransactionManager transactionManager,
|
||||
UserInterceptorCacheService userInterceptorCacheService,
|
||||
JsonHandlingService jsonHandlingService,
|
||||
QueryFactory queryFactory,
|
||||
LockByKeyManager lockByKeyManager,
|
||||
LocaleProperties localeProperties, OutboxService outboxService) {
|
||||
this.userScope = userScope;
|
||||
this.currentPrincipalResolver = currentPrincipalResolver;
|
||||
this.claimExtractor = claimExtractor;
|
||||
this.transactionManager = transactionManager;
|
||||
this.userInterceptorCacheService = userInterceptorCacheService;
|
||||
this.jsonHandlingService = jsonHandlingService;
|
||||
this.queryFactory = queryFactory;
|
||||
this.lockByKeyManager = lockByKeyManager;
|
||||
this.localeProperties = localeProperties;
|
||||
this.outboxService = outboxService;
|
||||
}
|
||||
|
||||
private UserCredentialEntity buildCredential(UUID userId, String subjectId){
|
||||
UserCredentialEntity data = new UserCredentialEntity();
|
||||
UserCredentialDataEntity userCredentialDataEntity = new UserCredentialDataEntity();
|
||||
|
||||
String email = this.getEmailFromClaims();
|
||||
String provider = this.getProviderFromClaims();
|
||||
if (email != null && !email.isBlank()) userCredentialDataEntity.setEmail(email);
|
||||
if (provider != null && !provider.isBlank()) userCredentialDataEntity.setExternalProviderNames(List.of(provider));
|
||||
data.setData(this.jsonHandlingService.toJsonSafe(userCredentialDataEntity));
|
||||
@Override
|
||||
public void preHandle(@NotNull WebRequest request) throws InterruptedException {
|
||||
UUID userId = null;
|
||||
if (this.currentPrincipalResolver.currentPrincipal().isAuthenticated()) {
|
||||
String subjectId = this.claimExtractor.subjectString(this.currentPrincipalResolver.currentPrincipal());
|
||||
if (subjectId == null || subjectId.isBlank())
|
||||
throw new MyForbiddenException("Empty subjects not allowed");
|
||||
|
||||
data.setId(UUID.randomUUID());
|
||||
data.setUserId(userId);
|
||||
data.setCreatedAt(Instant.now());
|
||||
data.setExternalId(subjectId);
|
||||
return data;
|
||||
}
|
||||
UserInterceptorCacheService.UserInterceptorCacheValue cacheValue = this.userInterceptorCacheService.lookup(this.userInterceptorCacheService.buildKey(subjectId));
|
||||
if (cacheValue != null && emailExistsToPrincipal(cacheValue.getProviderEmail()) && userRolesSynced(cacheValue.getRoles()) && providerExistsToPrincipal(cacheValue.getExternalProviderNames())) {
|
||||
userId = cacheValue.getUserId();
|
||||
} else {
|
||||
boolean usedResource = false;
|
||||
try {
|
||||
usedResource = this.lockByKeyManager.tryLock(subjectId, 5000, TimeUnit.MILLISECONDS);
|
||||
String email = this.getEmailFromClaims();
|
||||
|
||||
private UserRoleEntity buildRole(UUID userId, String role){
|
||||
UserRoleEntity data = new UserRoleEntity();
|
||||
data.setId(UUID.randomUUID());
|
||||
data.setUserId(userId);
|
||||
data.setRole(role);
|
||||
data.setCreatedAt(Instant.now());
|
||||
return data;
|
||||
}
|
||||
DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
|
||||
definition.setName(UUID.randomUUID().toString());
|
||||
definition.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
|
||||
definition.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
|
||||
TransactionStatus status = null;
|
||||
try {
|
||||
status = transactionManager.getTransaction(definition);
|
||||
|
||||
private UserContactInfoEntity buildEmailContact(UUID userId, String email){
|
||||
UserContactInfoEntity data = new UserContactInfoEntity();
|
||||
data.setId(UUID.randomUUID());
|
||||
data.setUserId(userId);
|
||||
data.setValue(email);
|
||||
data.setType(ContactInfoType.Email);
|
||||
data.setOrdinal(0);
|
||||
data.setCreatedAt(Instant.now());
|
||||
return data;
|
||||
}
|
||||
userId = this.findExistingUserFromDb(subjectId);
|
||||
boolean isNewUser = userId == null;
|
||||
if (isNewUser) {
|
||||
UserEntity user = this.addNewUser(subjectId, email);
|
||||
userId = user.getId();
|
||||
}
|
||||
this.entityManager.flush();
|
||||
|
||||
if (!isNewUser) {
|
||||
this.syncUserWithClaims(userId, subjectId);
|
||||
}
|
||||
|
||||
this.entityManager.flush();
|
||||
transactionManager.commit(status);
|
||||
} catch (Exception ex) {
|
||||
if (status != null)
|
||||
transactionManager.rollback(status);
|
||||
throw ex;
|
||||
}
|
||||
|
||||
private UserEntity addNewUser(String subjectId, String email){
|
||||
List<String> roles = this.getRolesFromClaims();
|
||||
String name = this.claimExtractor.name(this.currentPrincipalResolver.currentPrincipal());
|
||||
|
||||
UserEntity user = new UserEntity();
|
||||
user.setId(UUID.randomUUID());
|
||||
user.setName(name);
|
||||
user.setCreatedAt(Instant.now());
|
||||
user.setUpdatedAt(Instant.now());
|
||||
user.setIsActive(IsActive.Active);
|
||||
AdditionalInfoEntity additionalInfoEntity = new AdditionalInfoEntity();
|
||||
additionalInfoEntity.setCulture(this.localeProperties.getCulture());
|
||||
additionalInfoEntity.setLanguage(this.localeProperties.getLanguage());
|
||||
additionalInfoEntity.setTimezone(this.localeProperties.getTimezone());
|
||||
user.setAdditionalInfo(this.jsonHandlingService.toJsonSafe(additionalInfoEntity));
|
||||
this.entityManager.persist(user);
|
||||
cacheValue = new UserInterceptorCacheService.UserInterceptorCacheValue(subjectId, userId);
|
||||
cacheValue.setRoles(this.getRolesFromClaims());
|
||||
if (email != null && !email.isBlank())
|
||||
cacheValue.setProviderEmail(email);
|
||||
UserCredentialEntity userCredential = this.queryFactory.query(UserCredentialQuery.class).externalIds(subjectId).firstAs(new BaseFieldSet().ensure(UserCredential._data));
|
||||
if (userCredential != null && userCredential.getData() != null) {
|
||||
UserCredentialDataEntity userCredentialDataEntity = this.jsonHandlingService.fromJsonSafe(UserCredentialDataEntity.class, userCredential.getData());
|
||||
if (userCredentialDataEntity != null)
|
||||
cacheValue.setExternalProviderNames(userCredentialDataEntity.getExternalProviderNames());
|
||||
}
|
||||
|
||||
UserCredentialEntity credential = this.buildCredential(user.getId(), subjectId);
|
||||
this.entityManager.persist(credential);
|
||||
this.userInterceptorCacheService.put(cacheValue);
|
||||
} finally {
|
||||
if (usedResource)
|
||||
this.lockByKeyManager.unlock(subjectId);
|
||||
}
|
||||
|
||||
if (email != null && !email.isBlank()) {
|
||||
UserContactInfoEntity contactInfo = this.buildEmailContact(user.getId(), email);
|
||||
this.entityManager.persist(contactInfo);
|
||||
}
|
||||
if (roles != null) {
|
||||
for (String role: roles) {
|
||||
UserRoleEntity roleEntity = this.buildRole(user.getId(), role);
|
||||
this.entityManager.persist(roleEntity);
|
||||
}
|
||||
}
|
||||
|
||||
return user;
|
||||
}
|
||||
}
|
||||
}
|
||||
this.userScope.setUserId(userId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void postHandle(@NonNull WebRequest request, ModelMap model) {
|
||||
this.userScope.setUserId(null);
|
||||
}
|
||||
private void syncUserWithClaims(UUID userId, String subjectId) {
|
||||
UserTouchedIntegrationEvent event = new UserTouchedIntegrationEvent();
|
||||
|
||||
@Override
|
||||
public void afterCompletion(@NonNull WebRequest request, Exception ex) {
|
||||
}
|
||||
List<String> existingUserEmails = this.collectUserEmails(userId);
|
||||
if (!this.containsPrincipalEmail(existingUserEmails)) {
|
||||
String email = this.getEmailFromClaims();
|
||||
long contactUsedByOthersCount = this.queryFactory.query(UserContactInfoQuery.class).excludedUserIds(userId).types(ContactInfoType.Email).values(email).count();
|
||||
if (contactUsedByOthersCount > 0) {
|
||||
logger.warn("user contact exists to other user" + email);
|
||||
} else {
|
||||
Long emailContactsCount = this.queryFactory.query(UserContactInfoQuery.class).userIds(userId).types(ContactInfoType.Email).count();
|
||||
UserContactInfoEntity contactInfo = this.buildEmailContact(userId, email);
|
||||
contactInfo.setOrdinal(emailContactsCount.intValue());
|
||||
this.entityManager.persist(contactInfo);
|
||||
|
||||
this.initUserTouchedEvent(event, userId);
|
||||
UserTouchedIntegrationEvent.UserContactInfo eventContactInfo = new UserTouchedIntegrationEvent.UserContactInfo();
|
||||
eventContactInfo.setType(ContactInfoType.Email);
|
||||
eventContactInfo.setValue(email);
|
||||
event.getUserContactInfo().add(eventContactInfo);
|
||||
}
|
||||
}
|
||||
|
||||
List<String> existingUserRoles = this.collectUserRoles(userId);
|
||||
if (!this.userRolesSynced(existingUserRoles)) {
|
||||
this.syncRoles(userId);
|
||||
}
|
||||
|
||||
UserCredentialEntity userCredential = this.queryFactory.query(UserCredentialQuery.class).externalIds(subjectId).first();
|
||||
if (userCredential == null) {
|
||||
throw new MyForbiddenException("UserCredential not found");
|
||||
} else {
|
||||
boolean updatedUserCredential = false;
|
||||
UserCredentialDataEntity userCredentialDataEntity = this.jsonHandlingService.fromJsonSafe(UserCredentialDataEntity.class, userCredential.getData());
|
||||
if (userCredentialDataEntity == null)
|
||||
userCredentialDataEntity = new UserCredentialDataEntity();
|
||||
if (userCredentialDataEntity.getExternalProviderNames() == null)
|
||||
userCredentialDataEntity.setExternalProviderNames(new ArrayList<>());
|
||||
|
||||
String email = this.getEmailFromClaims();
|
||||
String provider = this.getProviderFromClaims();
|
||||
|
||||
if (email != null && !email.equalsIgnoreCase(userCredentialDataEntity.getEmail())) {
|
||||
userCredentialDataEntity.setEmail(email);
|
||||
updatedUserCredential = true;
|
||||
}
|
||||
if (provider != null && !provider.isBlank() && userCredentialDataEntity.getExternalProviderNames().stream().noneMatch(provider::equalsIgnoreCase)) {
|
||||
userCredentialDataEntity.getExternalProviderNames().add(provider);
|
||||
updatedUserCredential = true;
|
||||
}
|
||||
if (updatedUserCredential) {
|
||||
userCredential.setData(this.jsonHandlingService.toJsonSafe(userCredentialDataEntity));
|
||||
this.entityManager.persist(userCredential);
|
||||
}
|
||||
}
|
||||
|
||||
OutboxIntegrationEvent message = new OutboxIntegrationEvent();
|
||||
message.setMessageId(UUID.randomUUID());
|
||||
message.setType(OutboxIntegrationEvent.USER_TOUCH);
|
||||
message.setEvent(event);
|
||||
this.outboxService.publish(message);
|
||||
}
|
||||
|
||||
private UUID findExistingUserFromDb(String subjectId) {
|
||||
UserCredentialEntity userCredential = this.queryFactory.query(UserCredentialQuery.class).externalIds(subjectId).firstAs(new BaseFieldSet().ensure(UserCredential._user));
|
||||
if (userCredential != null) {
|
||||
return userCredential.getUserId();
|
||||
} else {
|
||||
String email = this.getEmailFromClaims();
|
||||
if (email != null && !email.isBlank()) {
|
||||
UserContactInfoEntity userContactInfo = this.queryFactory.query(UserContactInfoQuery.class).types(ContactInfoType.Email).values(email).firstAs(new BaseFieldSet().ensure(UserContactInfo._user));
|
||||
if (userContactInfo != null) {
|
||||
UserCredentialEntity credential = this.buildCredential(userContactInfo.getUserId(), subjectId);
|
||||
this.entityManager.persist(credential);
|
||||
|
||||
return credential.getUserId();
|
||||
}
|
||||
} else {
|
||||
throw new MyForbiddenException("Email is required");
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private List<String> getRolesFromClaims() {
|
||||
List<String> claimsRoles = claimExtractor.roles(currentPrincipalResolver.currentPrincipal());
|
||||
if (claimsRoles == null)
|
||||
claimsRoles = new ArrayList<>();
|
||||
claimsRoles = claimsRoles.stream().filter(x -> x != null && !x.isBlank()).distinct().toList();
|
||||
return claimsRoles;
|
||||
}
|
||||
|
||||
private void syncRoles(UUID userId) {
|
||||
List<UserRoleEntity> existingUserRoles = this.queryFactory.query(UserRoleQuery.class).userIds(userId).collect();
|
||||
List<UUID> foundRoles = new ArrayList<>();
|
||||
for (String claimRole : this.getRolesFromClaims()) {
|
||||
UserRoleEntity roleEntity = existingUserRoles.stream().filter(x -> x.getRole().equals(claimRole)).findFirst().orElse(null);
|
||||
if (roleEntity == null) {
|
||||
roleEntity = this.buildRole(userId, claimRole);
|
||||
this.entityManager.persist(roleEntity);
|
||||
}
|
||||
foundRoles.add(roleEntity.getId());
|
||||
}
|
||||
for (UserRoleEntity existing : existingUserRoles) {
|
||||
if (!foundRoles.contains(existing.getId())) {
|
||||
this.entityManager.remove(existing);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private List<String> collectUserRoles(UUID userId) {
|
||||
List<UserRoleEntity> items = this.queryFactory.query(UserRoleQuery.class).userIds(userId).collectAs(new BaseFieldSet().ensure(UserRole._role));
|
||||
return items == null ? new ArrayList<>() : items.stream().map(UserRoleEntity::getRole).toList();
|
||||
}
|
||||
|
||||
private List<String> collectUserEmails(UUID userId) {
|
||||
List<UserContactInfoEntity> items = this.queryFactory.query(UserContactInfoQuery.class).userIds(userId).types(ContactInfoType.Email).collectAs(new BaseFieldSet().ensure(UserContactInfo._value));
|
||||
return items == null ? new ArrayList<>() : items.stream().map(UserContactInfoEntity::getValue).toList();
|
||||
}
|
||||
|
||||
private boolean containsPrincipalEmail(List<String> existingUserEmails) {
|
||||
String email = this.getEmailFromClaims();
|
||||
return email == null || email.isBlank() ||
|
||||
(existingUserEmails != null && existingUserEmails.stream().anyMatch(email::equals));
|
||||
}
|
||||
|
||||
private boolean emailExistsToPrincipal(String existingUserEmail) {
|
||||
String email = this.getEmailFromClaims();
|
||||
return email == null || email.isBlank() || email.equalsIgnoreCase(existingUserEmail);
|
||||
}
|
||||
|
||||
private boolean providerExistsToPrincipal(List<String> principalCredentialProviders) {
|
||||
String provider = this.getProviderFromClaims();
|
||||
return provider == null || provider.isBlank() ||
|
||||
(principalCredentialProviders != null && principalCredentialProviders.stream().anyMatch(provider::equalsIgnoreCase));
|
||||
}
|
||||
|
||||
private boolean userRolesSynced(List<String> existingUserRoles) {
|
||||
List<String> claimsRoles = this.getRolesFromClaims();
|
||||
if (existingUserRoles == null)
|
||||
existingUserRoles = new ArrayList<>();
|
||||
existingUserRoles = existingUserRoles.stream().filter(x -> x != null && !x.isBlank()).distinct().toList();
|
||||
if (claimsRoles.size() != existingUserRoles.size())
|
||||
return false;
|
||||
|
||||
for (String claim : claimsRoles) {
|
||||
if (existingUserRoles.stream().noneMatch(claim::equalsIgnoreCase))
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private String getEmailFromClaims() {
|
||||
String email = this.claimExtractor.email(this.currentPrincipalResolver.currentPrincipal());
|
||||
if (email == null || email.isBlank() || !EmailValidator.getInstance().isValid(email))
|
||||
return null;
|
||||
return email.trim();
|
||||
}
|
||||
|
||||
private String getProviderFromClaims() {
|
||||
String provider = this.claimExtractor.asString(this.currentPrincipalResolver.currentPrincipal(), ClaimNames.ExternalProviderName);
|
||||
if (provider == null || provider.isBlank())
|
||||
return null;
|
||||
return provider.trim();
|
||||
}
|
||||
|
||||
private UserCredentialEntity buildCredential(UUID userId, String subjectId) {
|
||||
UserCredentialEntity data = new UserCredentialEntity();
|
||||
UserCredentialDataEntity userCredentialDataEntity = new UserCredentialDataEntity();
|
||||
|
||||
String email = this.getEmailFromClaims();
|
||||
String provider = this.getProviderFromClaims();
|
||||
if (email != null && !email.isBlank())
|
||||
userCredentialDataEntity.setEmail(email);
|
||||
if (provider != null && !provider.isBlank())
|
||||
userCredentialDataEntity.setExternalProviderNames(List.of(provider));
|
||||
data.setData(this.jsonHandlingService.toJsonSafe(userCredentialDataEntity));
|
||||
|
||||
data.setId(UUID.randomUUID());
|
||||
data.setUserId(userId);
|
||||
data.setCreatedAt(Instant.now());
|
||||
data.setExternalId(subjectId);
|
||||
return data;
|
||||
}
|
||||
|
||||
private UserRoleEntity buildRole(UUID userId, String role) {
|
||||
UserRoleEntity data = new UserRoleEntity();
|
||||
data.setId(UUID.randomUUID());
|
||||
data.setUserId(userId);
|
||||
data.setRole(role);
|
||||
data.setCreatedAt(Instant.now());
|
||||
return data;
|
||||
}
|
||||
|
||||
private UserContactInfoEntity buildEmailContact(UUID userId, String email) {
|
||||
UserContactInfoEntity data = new UserContactInfoEntity();
|
||||
data.setId(UUID.randomUUID());
|
||||
data.setUserId(userId);
|
||||
data.setValue(email);
|
||||
data.setType(ContactInfoType.Email);
|
||||
data.setOrdinal(0);
|
||||
data.setCreatedAt(Instant.now());
|
||||
return data;
|
||||
}
|
||||
|
||||
private UserEntity addNewUser(String subjectId, String email) {
|
||||
UserTouchedIntegrationEvent event = new UserTouchedIntegrationEvent();
|
||||
|
||||
List<String> roles = this.getRolesFromClaims();
|
||||
String name = this.claimExtractor.name(this.currentPrincipalResolver.currentPrincipal());
|
||||
|
||||
UserEntity user = new UserEntity();
|
||||
user.setId(UUID.randomUUID());
|
||||
user.setName(name);
|
||||
user.setCreatedAt(Instant.now());
|
||||
user.setUpdatedAt(Instant.now());
|
||||
user.setIsActive(IsActive.Active);
|
||||
AdditionalInfoEntity additionalInfoEntity = new AdditionalInfoEntity();
|
||||
additionalInfoEntity.setCulture(this.localeProperties.getCulture());
|
||||
additionalInfoEntity.setLanguage(this.localeProperties.getLanguage());
|
||||
additionalInfoEntity.setTimezone(this.localeProperties.getTimezone());
|
||||
user.setAdditionalInfo(this.jsonHandlingService.toJsonSafe(additionalInfoEntity));
|
||||
this.entityManager.persist(user);
|
||||
|
||||
this.initUserTouchedEvent(event, user.getId());
|
||||
event.setType(UserType.Person);
|
||||
event.setFirstName(user.getName().split(" ")[0]);
|
||||
event.setLastName(user.getName().split(" ")[1]);
|
||||
|
||||
UserTouchedIntegrationEvent.UserProfile eventUserProfile = new UserTouchedIntegrationEvent.UserProfile();
|
||||
eventUserProfile.setCulture(additionalInfoEntity.getCulture());
|
||||
eventUserProfile.setTimezone(additionalInfoEntity.getTimezone());
|
||||
eventUserProfile.setLanguage(additionalInfoEntity.getLanguage());
|
||||
event.setProfile(eventUserProfile);
|
||||
|
||||
UserCredentialEntity credential = this.buildCredential(user.getId(), subjectId);
|
||||
this.entityManager.persist(credential);
|
||||
|
||||
if (email != null && !email.isBlank()) {
|
||||
UserContactInfoEntity contactInfo = this.buildEmailContact(user.getId(), email);
|
||||
this.entityManager.persist(contactInfo);
|
||||
|
||||
UserTouchedIntegrationEvent.UserContactInfo userContactInfo = new UserTouchedIntegrationEvent.UserContactInfo();
|
||||
userContactInfo.setType(ContactInfoType.Email);
|
||||
userContactInfo.setValue(email);
|
||||
event.getUserContactInfo().add(userContactInfo);
|
||||
}
|
||||
if (roles != null) {
|
||||
for (String role : roles) {
|
||||
UserRoleEntity roleEntity = this.buildRole(user.getId(), role);
|
||||
this.entityManager.persist(roleEntity);
|
||||
}
|
||||
}
|
||||
|
||||
OutboxIntegrationEvent message = new OutboxIntegrationEvent();
|
||||
message.setMessageId(UUID.randomUUID());
|
||||
message.setType(OutboxIntegrationEvent.USER_TOUCH);
|
||||
message.setEvent(event);
|
||||
this.outboxService.publish(message);
|
||||
|
||||
return user;
|
||||
}
|
||||
|
||||
private void initUserTouchedEvent(UserTouchedIntegrationEvent event, UUID userId) {
|
||||
event.setId(userId);
|
||||
|
||||
UserEntity userEntity = this.entityManager.find(UserEntity.class, userId);
|
||||
if (userEntity != null) {
|
||||
event.setFirstName(userEntity.getName().split(" ")[0]);
|
||||
event.setLastName(userEntity.getName().split(" ")[1]);
|
||||
event.setType(UserType.Person);
|
||||
}
|
||||
|
||||
List<UserTouchedIntegrationEvent.UserContactInfo> userContactInfoList = new ArrayList<>();
|
||||
if (userEntity != null) {
|
||||
List<UserContactInfoEntity> contactInfoEntities = this.queryFactory.query(UserContactInfoQuery.class).userIds(userId).collect();
|
||||
for (UserContactInfoEntity entity : contactInfoEntities) {
|
||||
UserTouchedIntegrationEvent.UserContactInfo userContactInfo = new UserTouchedIntegrationEvent.UserContactInfo();
|
||||
userContactInfo.setType(entity.getType());
|
||||
userContactInfo.setValue(entity.getValue());
|
||||
userContactInfoList.add(userContactInfo);
|
||||
}
|
||||
}
|
||||
event.setUserContactInfo(userContactInfoList);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void postHandle(@NonNull WebRequest request, ModelMap model) {
|
||||
this.userScope.setUserId(null);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void afterCompletion(@NonNull WebRequest request, Exception ex) {
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -48,8 +48,6 @@ queue:
|
|||
enable: true
|
||||
options:
|
||||
exchange: null
|
||||
user-removal-topic: [ "user.remove" ]
|
||||
user-touched-topic: [ "user.touch" ]
|
||||
rabbitmq:
|
||||
enable: true
|
||||
interval-seconds: 30
|
||||
|
|
|
|||
Loading…
Reference in New Issue