Notification service inbox event handlers
This commit is contained in:
parent
df7a43d2de
commit
2fb387825d
|
|
@ -9,6 +9,16 @@ public class AuditableAction {
|
|||
|
||||
public static final EventId User_Available_Notifiers_Query = new EventId(10004, "User_Available_Notifiers_Query");
|
||||
|
||||
public static final EventId User_Query = new EventId(11000, "User_Query");
|
||||
public static final EventId User_Lookup = new EventId(11001, "User_Lookup");
|
||||
public static final EventId User_Persist = new EventId(11002, "User_Persist");
|
||||
public static final EventId User_Delete = new EventId(11003, "User_Delete");
|
||||
|
||||
public static final EventId Tenant_Query = new EventId(12000, "Tenant_Query");
|
||||
public static final EventId Tenant_Lookup = new EventId(12001, "Tenant_Lookup");
|
||||
public static final EventId Tenant_Persist = new EventId(12002, "Tenant_Persist");
|
||||
public static final EventId Tenant_Delete = new EventId(12003, "Tenant_Delete");
|
||||
|
||||
public static final EventId Notification_Query = new EventId(19000, "Notification_Query");
|
||||
public static final EventId Notification_Lookup = new EventId(19001, "Notification_Lookup");
|
||||
public static final EventId Notification_Persist = new EventId(19002, "Notification_Persist");
|
||||
|
|
|
|||
|
|
@ -3,14 +3,13 @@ package gr.cite.notification.common.enums;
|
|||
import com.fasterxml.jackson.annotation.JsonValue;
|
||||
import gr.cite.notification.data.conventers.DatabaseEnum;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
public enum NotificationContactType implements DatabaseEnum<Short> {
|
||||
EMAIL((short)0),
|
||||
SLACK_BROADCAST((short)1),
|
||||
SMS((short)2),
|
||||
IN_APP((short)3);
|
||||
EMAIL((short) 0),
|
||||
SLACK_BROADCAST((short) 1),
|
||||
SMS((short) 2),
|
||||
IN_APP((short) 3);
|
||||
|
||||
private final Short value;
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,48 @@
|
|||
package gr.cite.notification.integrationevent;
|
||||
|
||||
import gr.cite.notification.integrationevent.inbox.InboxProperties;
|
||||
import gr.cite.notification.integrationevent.outbox.OutboxProperties;
|
||||
import gr.cite.queueinbox.repository.InboxRepository;
|
||||
import gr.cite.rabbitmq.RabbitConfigurer;
|
||||
import gr.cite.rabbitmq.consumer.InboxBindings;
|
||||
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;
|
||||
import java.util.List;
|
||||
|
||||
@Configuration
|
||||
@EnableConfigurationProperties({OutboxProperties.class, InboxProperties.class})
|
||||
@ConditionalOnProperty(prefix = "queue.rabbitmq", name = "listenerEnabled")
|
||||
public class AppRabbitConfigurer extends RabbitConfigurer {
|
||||
|
||||
private final ApplicationContext applicationContext;
|
||||
|
||||
private final InboxProperties inboxProperties;
|
||||
|
||||
public AppRabbitConfigurer(ApplicationContext applicationContext, InboxProperties inboxProperties) {
|
||||
this.applicationContext = applicationContext;
|
||||
this.inboxProperties = inboxProperties;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public InboxBindings inboxBindingsCreator() {
|
||||
List<String> bindingItems = new ArrayList<>();
|
||||
bindingItems.addAll(this.inboxProperties.getUserRemovalTopic());
|
||||
bindingItems.addAll(this.inboxProperties.getUserTouchedTopic());
|
||||
|
||||
return new InboxBindings(bindingItems);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public InboxCreator inboxCreator() {
|
||||
return (params) -> {
|
||||
InboxRepository inboxRepository = this.applicationContext.getBean(InboxRepository.class);
|
||||
return inboxRepository.create(params) != null;
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,24 @@
|
|||
package gr.cite.notification.integrationevent;
|
||||
|
||||
import gr.cite.rabbitmq.IntegrationEventContext;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.UUID;
|
||||
|
||||
@Component
|
||||
public class IntegrationEventContextImpl implements IntegrationEventContext {
|
||||
|
||||
private UUID tenant;
|
||||
|
||||
public IntegrationEventContextImpl() {
|
||||
}
|
||||
|
||||
public UUID getTenant() {
|
||||
return tenant;
|
||||
}
|
||||
|
||||
public void setTenant(UUID tenant) {
|
||||
this.tenant = tenant;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,18 @@
|
|||
package gr.cite.notification.integrationevent;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
|
||||
|
||||
@JsonIgnoreProperties(ignoreUnknown = true)
|
||||
public class TrackedEvent {
|
||||
|
||||
public String trackingContextTag;
|
||||
|
||||
public String getTrackingContextTag() {
|
||||
return trackingContextTag;
|
||||
}
|
||||
|
||||
public void setTrackingContextTag(String trackingContextTag) {
|
||||
this.trackingContextTag = trackingContextTag;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,7 @@
|
|||
package gr.cite.notification.integrationevent.inbox;
|
||||
|
||||
public interface ConsistencyHandler<T extends ConsistencyPredicates> {
|
||||
|
||||
Boolean isConsistent(T consistencyPredicates);
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,5 @@
|
|||
package gr.cite.notification.integrationevent.inbox;
|
||||
|
||||
public interface ConsistencyPredicates {
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,8 @@
|
|||
package gr.cite.notification.integrationevent.inbox;
|
||||
|
||||
public enum EventProcessingStatus {
|
||||
Error,
|
||||
Success,
|
||||
Postponed,
|
||||
Discard
|
||||
}
|
||||
|
|
@ -0,0 +1,60 @@
|
|||
package gr.cite.notification.integrationevent.inbox;
|
||||
|
||||
import gr.cite.commons.web.oidc.principal.MyPrincipal;
|
||||
import org.springframework.security.oauth2.core.ClaimAccessor;
|
||||
import org.springframework.security.oauth2.jwt.JwtClaimNames;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.time.temporal.ChronoUnit;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
public class InboxPrincipal implements MyPrincipal, ClaimAccessor {
|
||||
|
||||
private final Map<String, Object> claims;
|
||||
|
||||
private final boolean isAuthenticated;
|
||||
|
||||
public InboxPrincipal(Boolean isAuthenticated, String name) {
|
||||
this.claims = new HashMap<>();
|
||||
this.put(JwtClaimNames.SUB, name);
|
||||
this.isAuthenticated = isAuthenticated;
|
||||
}
|
||||
|
||||
public static InboxPrincipal build(IntegrationEventProperties properties) {
|
||||
InboxPrincipal inboxPrincipal = new InboxPrincipal(true, "IntegrationEventQueueAppId");
|
||||
inboxPrincipal.put("client_id", properties.getAppId());
|
||||
inboxPrincipal.put("active", "true");
|
||||
inboxPrincipal.put("nbf", Instant.now().minus(30, ChronoUnit.SECONDS).toString());
|
||||
inboxPrincipal.put("exp", Instant.now().plus(10, ChronoUnit.MINUTES).toString());
|
||||
return inboxPrincipal;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Boolean isAuthenticated() {
|
||||
return this.isAuthenticated;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Map<String, Object> getClaims() {
|
||||
return this.claims;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<String> getClaimAsStringList(String claim) {
|
||||
if (claims == null)
|
||||
return null;
|
||||
return this.getClaimAsStringList(claim);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getName() {
|
||||
return this.getClaimAsString(JwtClaimNames.SUB);
|
||||
}
|
||||
|
||||
public void put(String key, Object value) {
|
||||
this.claims.put(key, value);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,67 @@
|
|||
package gr.cite.notification.integrationevent.inbox;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.validation.annotation.Validated;
|
||||
|
||||
import javax.validation.constraints.NotNull;
|
||||
import java.util.List;
|
||||
|
||||
@Validated
|
||||
@ConfigurationProperties(prefix = "queue.task.listener.options")
|
||||
public class InboxProperties {
|
||||
|
||||
@NotNull
|
||||
private final String exchange;
|
||||
|
||||
@NotNull
|
||||
private final List<String> notifyTopic;
|
||||
|
||||
@NotNull
|
||||
private final List<String> tenantRemovedTopic;
|
||||
|
||||
@NotNull
|
||||
private final List<String> tenantTouchedTopic;
|
||||
|
||||
@NotNull
|
||||
private final List<String> userRemovalTopic;
|
||||
|
||||
@NotNull
|
||||
private final List<String> userTouchedTopic;
|
||||
|
||||
public InboxProperties(
|
||||
String exchange, List<String> notifyTopic, List<String> tenantRemovedTopic, List<String> tenantTouchedTopic,
|
||||
List<String> userRemovalTopic,
|
||||
List<String> userTouchedTopic) {
|
||||
this.exchange = exchange;
|
||||
this.notifyTopic = notifyTopic;
|
||||
this.tenantRemovedTopic = tenantRemovedTopic;
|
||||
this.tenantTouchedTopic = tenantTouchedTopic;
|
||||
this.userRemovalTopic = userRemovalTopic;
|
||||
this.userTouchedTopic = userTouchedTopic;
|
||||
}
|
||||
|
||||
public List<String> getNotifyTopic() {
|
||||
return notifyTopic;
|
||||
}
|
||||
|
||||
public List<String> getTenantRemovedTopic() {
|
||||
return tenantRemovedTopic;
|
||||
}
|
||||
|
||||
public List<String> getTenantTouchedTopic() {
|
||||
return tenantTouchedTopic;
|
||||
}
|
||||
|
||||
public List<String> getUserRemovalTopic() {
|
||||
return userRemovalTopic;
|
||||
}
|
||||
|
||||
public List<String> getUserTouchedTopic() {
|
||||
return userTouchedTopic;
|
||||
}
|
||||
|
||||
public String getExchange() {
|
||||
return exchange;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,365 @@
|
|||
package gr.cite.notification.integrationevent.inbox;
|
||||
|
||||
import gr.cite.notification.common.JsonHandlingService;
|
||||
import gr.cite.notification.common.enums.IsActive;
|
||||
import gr.cite.notification.common.scope.fake.FakeRequestScope;
|
||||
import gr.cite.notification.data.QueueInboxEntity;
|
||||
import gr.cite.notification.integrationevent.TrackedEvent;
|
||||
import gr.cite.notification.integrationevent.inbox.notify.NotifyIntegrationEventHandler;
|
||||
import gr.cite.notification.integrationevent.inbox.tenantremoval.TenantRemovalIntegrationEventHandler;
|
||||
import gr.cite.notification.integrationevent.inbox.tenanttouched.TenantTouchedIntegrationEventHandler;
|
||||
import gr.cite.notification.integrationevent.inbox.userremoval.UserRemovalIntegrationEventHandler;
|
||||
import gr.cite.notification.integrationevent.inbox.usertouched.UserTouchedIntegrationEventHandler;
|
||||
import gr.cite.notification.query.QueueInboxQuery;
|
||||
import gr.cite.queueinbox.entity.QueueInbox;
|
||||
import gr.cite.queueinbox.entity.QueueInboxStatus;
|
||||
import gr.cite.queueinbox.repository.CandidateInfo;
|
||||
import gr.cite.queueinbox.repository.InboxRepository;
|
||||
import gr.cite.queueinbox.task.MessageOptions;
|
||||
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.EntityManager;
|
||||
import jakarta.persistence.EntityManagerFactory;
|
||||
import jakarta.persistence.EntityTransaction;
|
||||
import jakarta.persistence.OptimisticLockException;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
import java.util.function.Function;
|
||||
|
||||
@Component
|
||||
public class InboxRepositoryImpl implements InboxRepository {
|
||||
|
||||
private static final LoggerService logger = new LoggerService(LoggerFactory.getLogger(InboxRepositoryImpl.class));
|
||||
|
||||
protected final ApplicationContext applicationContext;
|
||||
|
||||
private final JsonHandlingService jsonHandlingService;
|
||||
|
||||
private final InboxProperties inboxProperties;
|
||||
|
||||
public InboxRepositoryImpl(
|
||||
ApplicationContext applicationContext,
|
||||
InboxProperties inboxProperties
|
||||
) {
|
||||
this.applicationContext = applicationContext;
|
||||
this.jsonHandlingService = this.applicationContext.getBean(JsonHandlingService.class);
|
||||
this.inboxProperties = inboxProperties;
|
||||
}
|
||||
|
||||
@Override
|
||||
public CandidateInfo candidate(Instant lastCandidateCreationTimestamp, MessageOptions options) {
|
||||
EntityTransaction transaction = null;
|
||||
EntityManager entityManager = null;
|
||||
CandidateInfo candidate = null;
|
||||
try (FakeRequestScope ignored = new FakeRequestScope()) {
|
||||
try {
|
||||
QueryFactory queryFactory = this.applicationContext.getBean(QueryFactory.class);
|
||||
EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
|
||||
entityManager = entityManagerFactory.createEntityManager();
|
||||
|
||||
transaction = entityManager.getTransaction();
|
||||
transaction.begin();
|
||||
|
||||
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();
|
||||
|
||||
if (item != null) {
|
||||
QueueInboxStatus prevState = item.getStatus();
|
||||
item.setStatus(QueueInboxStatus.PROCESSING);
|
||||
|
||||
entityManager.merge(item);
|
||||
entityManager.flush();
|
||||
|
||||
candidate = new CandidateInfo();
|
||||
candidate.setId(item.getId());
|
||||
candidate.setCreatedAt(item.getCreatedAt());
|
||||
candidate.setPreviousState(prevState);
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
return candidate;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Boolean shouldOmit(CandidateInfo candidate, Function<QueueInbox, Boolean> shouldOmit) {
|
||||
EntityTransaction transaction = null;
|
||||
EntityManager entityManager = null;
|
||||
boolean success = false;
|
||||
|
||||
try (FakeRequestScope ignored = new FakeRequestScope()) {
|
||||
try {
|
||||
EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
|
||||
|
||||
entityManager = entityManagerFactory.createEntityManager();
|
||||
transaction = entityManager.getTransaction();
|
||||
|
||||
transaction.begin();
|
||||
|
||||
QueryFactory queryFactory = this.applicationContext.getBean(QueryFactory.class);
|
||||
QueueInboxEntity item = queryFactory.query(QueueInboxQuery.class).ids(candidate.getId()).first();
|
||||
|
||||
if (item == null) {
|
||||
logger.warn("Could not lookup queue inbox {} to process. Continuing...", candidate.getId());
|
||||
} else {
|
||||
if (shouldOmit.apply(item)) {
|
||||
item.setStatus(QueueInboxStatus.OMITTED);
|
||||
|
||||
entityManager.merge(item);
|
||||
entityManager.flush();
|
||||
success = true;
|
||||
}
|
||||
}
|
||||
|
||||
transaction.commit();
|
||||
} catch (Exception ex) {
|
||||
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) {
|
||||
logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
}
|
||||
return success;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean shouldWait(CandidateInfo candidate, Function<QueueInbox, Boolean> itIsTimeFunc) {
|
||||
EntityTransaction transaction = null;
|
||||
EntityManager entityManager = null;
|
||||
boolean success = false;
|
||||
|
||||
try (FakeRequestScope ignored = new FakeRequestScope()) {
|
||||
try {
|
||||
EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
|
||||
|
||||
entityManager = entityManagerFactory.createEntityManager();
|
||||
transaction = entityManager.getTransaction();
|
||||
|
||||
transaction.begin();
|
||||
|
||||
QueryFactory queryFactory = this.applicationContext.getBean(QueryFactory.class);
|
||||
QueueInboxEntity item = queryFactory.query(QueueInboxQuery.class).ids(candidate.getId()).first();
|
||||
|
||||
if (item.getRetryCount() != null && item.getRetryCount() >= 1) {
|
||||
Boolean itIsTime = itIsTimeFunc.apply(item);
|
||||
|
||||
if (!itIsTime) {
|
||||
item.setStatus(candidate.getPreviousState());
|
||||
|
||||
entityManager.merge(item);
|
||||
entityManager.flush();
|
||||
success = true;
|
||||
}
|
||||
|
||||
success = !itIsTime;
|
||||
}
|
||||
transaction.commit();
|
||||
} catch (Exception ex) {
|
||||
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) {
|
||||
logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
}
|
||||
return success;
|
||||
}
|
||||
|
||||
@Override
|
||||
public QueueInbox create(InboxCreatorParams inboxCreatorParams) {
|
||||
EntityTransaction transaction = null;
|
||||
EntityManager entityManager = null;
|
||||
boolean success = false;
|
||||
QueueInboxEntity queueMessage = null;
|
||||
try (FakeRequestScope ignored = new FakeRequestScope()) {
|
||||
try {
|
||||
queueMessage = this.createQueueInboxEntity(inboxCreatorParams);
|
||||
EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
|
||||
|
||||
entityManager = entityManagerFactory.createEntityManager();
|
||||
transaction = entityManager.getTransaction();
|
||||
|
||||
transaction.begin();
|
||||
|
||||
entityManager.persist(queueMessage);
|
||||
entityManager.flush();
|
||||
|
||||
transaction.commit();
|
||||
} catch (Exception ex) {
|
||||
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) {
|
||||
logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
}
|
||||
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());
|
||||
|
||||
return queueMessage;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Boolean emit(CandidateInfo candidateInfo) {
|
||||
EntityTransaction transaction = null;
|
||||
EntityManager entityManager = null;
|
||||
boolean success = false;
|
||||
|
||||
try (FakeRequestScope ignored = new FakeRequestScope()) {
|
||||
try {
|
||||
|
||||
EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
|
||||
|
||||
entityManager = entityManagerFactory.createEntityManager();
|
||||
transaction = entityManager.getTransaction();
|
||||
transaction.begin();
|
||||
|
||||
QueryFactory queryFactory = this.applicationContext.getBean(QueryFactory.class);
|
||||
QueueInboxEntity queueInboxMessage = queryFactory.query(QueueInboxQuery.class).ids(candidateInfo.getId()).first();
|
||||
|
||||
if (queueInboxMessage == null) {
|
||||
logger.warn("Could not lookup queue inbox {} to process. Continuing...", candidateInfo.getId());
|
||||
} else {
|
||||
|
||||
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;
|
||||
|
||||
entityManager.merge(queueInboxMessage);
|
||||
entityManager.flush();
|
||||
}
|
||||
|
||||
transaction.commit();
|
||||
} catch (Exception ex) {
|
||||
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) {
|
||||
logger.error("Problem executing purge. rolling back any db changes and marking error. Continuing...", ex);
|
||||
}
|
||||
return success;
|
||||
}
|
||||
|
||||
private EventProcessingStatus processMessage(String routingKey, String messageId, String appId, String message) {
|
||||
IntegrationEventHandler handler;
|
||||
if (this.RoutingKeyMatched(routingKey, this.inboxProperties.getNotifyTopic()))
|
||||
handler = this.applicationContext.getBean(NotifyIntegrationEventHandler.class);
|
||||
else if (this.RoutingKeyMatched(routingKey, this.inboxProperties.getTenantRemovedTopic()))
|
||||
handler = this.applicationContext.getBean(TenantRemovalIntegrationEventHandler.class);
|
||||
else if (this.RoutingKeyMatched(routingKey, this.inboxProperties.getTenantTouchedTopic()))
|
||||
handler = this.applicationContext.getBean(TenantTouchedIntegrationEventHandler.class);
|
||||
else 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;
|
||||
|
||||
if (handler == null)
|
||||
return EventProcessingStatus.Discard;
|
||||
|
||||
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;
|
||||
}
|
||||
// }
|
||||
}
|
||||
|
||||
private Boolean RoutingKeyMatched(String routingKey, List<String> topics) {
|
||||
if (topics == null || topics.isEmpty())
|
||||
return false;
|
||||
return topics.stream().anyMatch(x -> x.equals(routingKey));
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,7 @@
|
|||
package gr.cite.notification.integrationevent.inbox;
|
||||
|
||||
public interface IntegrationEventHandler {
|
||||
|
||||
EventProcessingStatus handle(IntegrationEventProperties properties, String message);
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,25 @@
|
|||
package gr.cite.notification.integrationevent.inbox;
|
||||
|
||||
public class IntegrationEventProperties {
|
||||
|
||||
private String messageId;
|
||||
|
||||
private String 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;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,47 @@
|
|||
package gr.cite.notification.integrationevent.inbox.notify;
|
||||
|
||||
import gr.cite.notification.common.StringUtils;
|
||||
import gr.cite.notification.common.enums.IsActive;
|
||||
import gr.cite.notification.common.enums.NotificationContactType;
|
||||
import gr.cite.notification.integrationevent.inbox.ConsistencyHandler;
|
||||
import gr.cite.notification.model.User;
|
||||
import gr.cite.notification.model.builder.UserBuilder;
|
||||
import gr.cite.notification.query.UserQuery;
|
||||
import gr.cite.tools.data.builder.BuilderFactory;
|
||||
import gr.cite.tools.data.query.QueryFactory;
|
||||
import gr.cite.tools.fieldset.BaseFieldSet;
|
||||
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
|
||||
import org.springframework.context.annotation.Scope;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
@Component
|
||||
@Scope(value = ConfigurableBeanFactory.SCOPE_PROTOTYPE)
|
||||
public class NotifyConsistencyHandler implements ConsistencyHandler<NotifyConsistencyPredicates> {
|
||||
|
||||
private final QueryFactory queryFactory;
|
||||
|
||||
private final BuilderFactory builderFactory;
|
||||
|
||||
public NotifyConsistencyHandler(QueryFactory queryFactory, BuilderFactory builderFactory) {
|
||||
this.queryFactory = queryFactory;
|
||||
this.builderFactory = builderFactory;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Boolean isConsistent(NotifyConsistencyPredicates consistencyPredicates) {
|
||||
if (consistencyPredicates.getUserId() != null) {
|
||||
UserQuery query = this.queryFactory.query(UserQuery.class).ids(consistencyPredicates.getUserId()).isActive(IsActive.Active);
|
||||
|
||||
BaseFieldSet fieldSet = new BaseFieldSet(
|
||||
User._id,
|
||||
User._isActive
|
||||
);
|
||||
User user = this.builderFactory.builder(UserBuilder.class).build(fieldSet, query.firstAs(fieldSet));
|
||||
if (user == null)
|
||||
return false;
|
||||
return consistencyPredicates.getContactTypeHint() == null || consistencyPredicates.getContactTypeHint() == NotificationContactType.IN_APP || !StringUtils.isNullOrEmpty(consistencyPredicates.getContactHint());
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,45 @@
|
|||
package gr.cite.notification.integrationevent.inbox.notify;
|
||||
|
||||
import gr.cite.notification.common.enums.NotificationContactType;
|
||||
import gr.cite.notification.integrationevent.inbox.ConsistencyPredicates;
|
||||
|
||||
import java.util.UUID;
|
||||
|
||||
public class NotifyConsistencyPredicates implements ConsistencyPredicates {
|
||||
|
||||
private UUID userId;
|
||||
|
||||
private NotificationContactType contactTypeHint;
|
||||
|
||||
private String contactHint;
|
||||
|
||||
public NotifyConsistencyPredicates(UUID userId, NotificationContactType contactTypeHint, String contactHint) {
|
||||
this.userId = userId;
|
||||
this.contactTypeHint = contactTypeHint;
|
||||
this.contactHint = contactHint;
|
||||
}
|
||||
|
||||
public UUID getUserId() {
|
||||
return userId;
|
||||
}
|
||||
|
||||
public void setUserId(UUID userId) {
|
||||
this.userId = userId;
|
||||
}
|
||||
|
||||
public NotificationContactType getContactTypeHint() {
|
||||
return contactTypeHint;
|
||||
}
|
||||
|
||||
public void setContactTypeHint(NotificationContactType contactTypeHint) {
|
||||
this.contactTypeHint = contactTypeHint;
|
||||
}
|
||||
|
||||
public String getContactHint() {
|
||||
return contactHint;
|
||||
}
|
||||
|
||||
public void setContactHint(String contactHint) {
|
||||
this.contactHint = contactHint;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,80 @@
|
|||
package gr.cite.notification.integrationevent.inbox.notify;
|
||||
|
||||
import gr.cite.notification.common.enums.NotificationContactType;
|
||||
import gr.cite.notification.integrationevent.TrackedEvent;
|
||||
|
||||
import java.util.UUID;
|
||||
|
||||
public class NotifyIntegrationEvent extends TrackedEvent {
|
||||
|
||||
private UUID userId;
|
||||
|
||||
private UUID tenantId;
|
||||
|
||||
private UUID notificationType;
|
||||
|
||||
private NotificationContactType contactTypeHint;
|
||||
|
||||
private String contactHint;
|
||||
|
||||
private String data;
|
||||
|
||||
private String provenanceRef;
|
||||
|
||||
public UUID getUserId() {
|
||||
return userId;
|
||||
}
|
||||
|
||||
public void setUserId(UUID userId) {
|
||||
this.userId = userId;
|
||||
}
|
||||
|
||||
public UUID getTenantId() {
|
||||
return tenantId;
|
||||
}
|
||||
|
||||
public void setTenantId(UUID tenantId) {
|
||||
this.tenantId = tenantId;
|
||||
}
|
||||
|
||||
public UUID getNotificationType() {
|
||||
return notificationType;
|
||||
}
|
||||
|
||||
public void setNotificationType(UUID notificationType) {
|
||||
this.notificationType = notificationType;
|
||||
}
|
||||
|
||||
public NotificationContactType getContactTypeHint() {
|
||||
return contactTypeHint;
|
||||
}
|
||||
|
||||
public void setContactTypeHint(NotificationContactType contactTypeHint) {
|
||||
this.contactTypeHint = contactTypeHint;
|
||||
}
|
||||
|
||||
public String getContactHint() {
|
||||
return contactHint;
|
||||
}
|
||||
|
||||
public void setContactHint(String contactHint) {
|
||||
this.contactHint = contactHint;
|
||||
}
|
||||
|
||||
public String getData() {
|
||||
return data;
|
||||
}
|
||||
|
||||
public void setData(String data) {
|
||||
this.data = data;
|
||||
}
|
||||
|
||||
public String getProvenanceRef() {
|
||||
return provenanceRef;
|
||||
}
|
||||
|
||||
public void setProvenanceRef(String provenanceRef) {
|
||||
this.provenanceRef = provenanceRef;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,7 @@
|
|||
package gr.cite.notification.integrationevent.inbox.notify;
|
||||
|
||||
import gr.cite.notification.integrationevent.inbox.IntegrationEventHandler;
|
||||
|
||||
public interface NotifyIntegrationEventHandler extends IntegrationEventHandler {
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,146 @@
|
|||
package gr.cite.notification.integrationevent.inbox.notify;
|
||||
|
||||
import gr.cite.commons.web.oidc.principal.CurrentPrincipalResolver;
|
||||
import gr.cite.notification.audit.AuditableAction;
|
||||
import gr.cite.notification.common.JsonHandlingService;
|
||||
import gr.cite.notification.common.enums.NotificationNotifyState;
|
||||
import gr.cite.notification.common.scope.fake.FakeRequestScope;
|
||||
import gr.cite.notification.common.scope.tenant.TenantScope;
|
||||
import gr.cite.notification.data.TenantEntity;
|
||||
import gr.cite.notification.errorcode.ErrorThesaurusProperties;
|
||||
import gr.cite.notification.integrationevent.inbox.EventProcessingStatus;
|
||||
import gr.cite.notification.integrationevent.inbox.InboxPrincipal;
|
||||
import gr.cite.notification.integrationevent.inbox.IntegrationEventProperties;
|
||||
import gr.cite.notification.model.Tenant;
|
||||
import gr.cite.notification.model.persist.NotificationPersist;
|
||||
import gr.cite.notification.query.TenantQuery;
|
||||
import gr.cite.notification.service.notification.NotificationService;
|
||||
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.EntityManager;
|
||||
import jakarta.persistence.EntityManagerFactory;
|
||||
import jakarta.persistence.EntityTransaction;
|
||||
import jakarta.persistence.OptimisticLockException;
|
||||
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
|
||||
@Scope(value = ConfigurableBeanFactory.SCOPE_PROTOTYPE)
|
||||
public class NotifyIntegrationEventHandlerImpl implements NotifyIntegrationEventHandler {
|
||||
|
||||
private static final LoggerService logger = new LoggerService(LoggerFactory.getLogger(NotifyIntegrationEventHandlerImpl.class));
|
||||
|
||||
private final JsonHandlingService jsonHandlingService;
|
||||
|
||||
private final ApplicationContext applicationContext;
|
||||
|
||||
private final ErrorThesaurusProperties errors;
|
||||
|
||||
private final MessageSource messageSource;
|
||||
|
||||
public NotifyIntegrationEventHandlerImpl(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) {
|
||||
NotifyIntegrationEvent event = this.jsonHandlingService.fromJsonSafe(NotifyIntegrationEvent.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()));
|
||||
}
|
||||
|
||||
NotificationPersist model = new NotificationPersist();
|
||||
model.setContactHint(event.getContactHint());
|
||||
model.setContactTypeHint(event.getContactTypeHint());
|
||||
model.setNotifiedWith(event.getContactTypeHint());
|
||||
model.setData(event.getData());
|
||||
model.setNotifyState(NotificationNotifyState.PENDING);
|
||||
model.setType(event.getNotificationType());
|
||||
|
||||
EntityManager entityManager = null;
|
||||
EntityTransaction transaction = null;
|
||||
try (FakeRequestScope ignored = new FakeRequestScope()) {
|
||||
try {
|
||||
QueryFactory queryFactory = this.applicationContext.getBean(QueryFactory.class);
|
||||
TenantScope scope = this.applicationContext.getBean(TenantScope.class);
|
||||
if (scope.isMultitenant() && event.getTenantId() != null) {
|
||||
TenantEntity tenant = queryFactory.query(TenantQuery.class).ids(event.getTenantId()).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.getTenantId(), 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));
|
||||
|
||||
NotifyConsistencyHandler notifyConsistencyHandler = this.applicationContext.getBean(NotifyConsistencyHandler.class);
|
||||
if (!(notifyConsistencyHandler.isConsistent(new NotifyConsistencyPredicates(event.getUserId(), event.getContactTypeHint(), event.getContactHint()))))
|
||||
return EventProcessingStatus.Postponed;
|
||||
|
||||
EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
|
||||
entityManager = entityManagerFactory.createEntityManager();
|
||||
|
||||
transaction = entityManager.getTransaction();
|
||||
transaction.begin();
|
||||
|
||||
try {
|
||||
NotificationService notificationService = this.applicationContext.getBean(NotificationService.class);
|
||||
notificationService.persist(model, new BaseFieldSet());
|
||||
|
||||
AuditService auditService = this.applicationContext.getBean(AuditService.class);
|
||||
|
||||
auditService.track(AuditableAction.Notification_Persist, 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
|
||||
logger.debug("Concurrency exception getting queue outbox. Skipping: {} ", ex.getMessage());
|
||||
if (transaction != null)
|
||||
transaction.rollback();
|
||||
} catch (Exception ex) {
|
||||
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) {
|
||||
logger.error("Problem getting list of queue outbox. Skipping: {}", ex.getMessage(), ex);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,26 @@
|
|||
package gr.cite.notification.integrationevent.inbox.tenantremoval;
|
||||
|
||||
import gr.cite.notification.integrationevent.inbox.ConsistencyHandler;
|
||||
import gr.cite.notification.query.TenantQuery;
|
||||
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
|
||||
@Scope(value = ConfigurableBeanFactory.SCOPE_PROTOTYPE)
|
||||
public class TenantRemovalConsistencyHandler implements ConsistencyHandler<TenantRemovalConsistencyPredicates> {
|
||||
|
||||
private final QueryFactory queryFactory;
|
||||
|
||||
public TenantRemovalConsistencyHandler(QueryFactory queryFactory) {
|
||||
this.queryFactory = queryFactory;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Boolean isConsistent(TenantRemovalConsistencyPredicates consistencyPredicates) {
|
||||
long count = this.queryFactory.query(TenantQuery.class).ids(consistencyPredicates.getTenantId()).count();
|
||||
return count > 0;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,23 @@
|
|||
package gr.cite.notification.integrationevent.inbox.tenantremoval;
|
||||
|
||||
import gr.cite.notification.integrationevent.inbox.ConsistencyPredicates;
|
||||
|
||||
import java.util.UUID;
|
||||
|
||||
public class TenantRemovalConsistencyPredicates implements ConsistencyPredicates {
|
||||
|
||||
private UUID tenantId;
|
||||
|
||||
public TenantRemovalConsistencyPredicates(UUID tenantId) {
|
||||
this.tenantId = tenantId;
|
||||
}
|
||||
|
||||
public UUID getTenantId() {
|
||||
return tenantId;
|
||||
}
|
||||
|
||||
public void setTenantId(UUID tenantId) {
|
||||
this.tenantId = tenantId;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,19 @@
|
|||
package gr.cite.notification.integrationevent.inbox.tenantremoval;
|
||||
|
||||
import gr.cite.notification.integrationevent.TrackedEvent;
|
||||
|
||||
import java.util.UUID;
|
||||
|
||||
public class TenantRemovalIntegrationEvent extends TrackedEvent {
|
||||
|
||||
private UUID id;
|
||||
|
||||
public UUID getId() {
|
||||
return id;
|
||||
}
|
||||
|
||||
public void setId(UUID id) {
|
||||
this.id = id;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,7 @@
|
|||
package gr.cite.notification.integrationevent.inbox.tenantremoval;
|
||||
|
||||
import gr.cite.notification.integrationevent.inbox.IntegrationEventHandler;
|
||||
|
||||
public interface TenantRemovalIntegrationEventHandler extends IntegrationEventHandler {
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,111 @@
|
|||
package gr.cite.notification.integrationevent.inbox.tenantremoval;
|
||||
|
||||
import gr.cite.commons.web.oidc.principal.CurrentPrincipalResolver;
|
||||
import gr.cite.notification.audit.AuditableAction;
|
||||
import gr.cite.notification.common.JsonHandlingService;
|
||||
import gr.cite.notification.common.scope.fake.FakeRequestScope;
|
||||
import gr.cite.notification.errorcode.ErrorThesaurusProperties;
|
||||
import gr.cite.notification.integrationevent.inbox.EventProcessingStatus;
|
||||
import gr.cite.notification.integrationevent.inbox.InboxPrincipal;
|
||||
import gr.cite.notification.integrationevent.inbox.IntegrationEventProperties;
|
||||
import gr.cite.notification.service.tenant.TenantService;
|
||||
import gr.cite.tools.auditing.AuditService;
|
||||
import gr.cite.tools.logging.LoggerService;
|
||||
import jakarta.persistence.EntityManager;
|
||||
import jakarta.persistence.EntityManagerFactory;
|
||||
import jakarta.persistence.EntityTransaction;
|
||||
import jakarta.persistence.OptimisticLockException;
|
||||
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.stereotype.Component;
|
||||
|
||||
import java.util.AbstractMap;
|
||||
import java.util.Map;
|
||||
|
||||
@Component
|
||||
@Scope(value = ConfigurableBeanFactory.SCOPE_PROTOTYPE)
|
||||
public class TenantRemovalIntegrationEventHandlerImpl implements TenantRemovalIntegrationEventHandler {
|
||||
|
||||
private static final LoggerService logger = new LoggerService(LoggerFactory.getLogger(TenantRemovalIntegrationEventHandlerImpl.class));
|
||||
|
||||
private final JsonHandlingService jsonHandlingService;
|
||||
|
||||
private final ApplicationContext applicationContext;
|
||||
|
||||
private final ErrorThesaurusProperties errors;
|
||||
|
||||
private final MessageSource messageSource;
|
||||
|
||||
public TenantRemovalIntegrationEventHandlerImpl(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) {
|
||||
TenantRemovalIntegrationEvent event = this.jsonHandlingService.fromJsonSafe(TenantRemovalIntegrationEvent.class, message);
|
||||
if (event == null)
|
||||
return EventProcessingStatus.Error;
|
||||
|
||||
EntityManager entityManager = null;
|
||||
EntityTransaction transaction = null;
|
||||
try (FakeRequestScope ignored = new FakeRequestScope()) {
|
||||
try {
|
||||
CurrentPrincipalResolver currentPrincipalResolver = this.applicationContext.getBean(CurrentPrincipalResolver.class);
|
||||
currentPrincipalResolver.push(InboxPrincipal.build(properties));
|
||||
|
||||
TenantRemovalConsistencyHandler tenantRemovalConsistencyHandler = this.applicationContext.getBean(TenantRemovalConsistencyHandler.class);
|
||||
if (!(tenantRemovalConsistencyHandler.isConsistent(new TenantRemovalConsistencyPredicates(event.getId()))))
|
||||
return EventProcessingStatus.Postponed;
|
||||
|
||||
EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
|
||||
entityManager = entityManagerFactory.createEntityManager();
|
||||
|
||||
transaction = entityManager.getTransaction();
|
||||
transaction.begin();
|
||||
|
||||
try {
|
||||
TenantService tenantService = this.applicationContext.getBean(TenantService.class);
|
||||
tenantService.deleteAndSave(event.getId());
|
||||
|
||||
AuditService auditService = this.applicationContext.getBean(AuditService.class);
|
||||
|
||||
auditService.track(AuditableAction.Tenant_Delete, Map.ofEntries(
|
||||
new AbstractMap.SimpleEntry<String, Object>("id", event.getId())
|
||||
));
|
||||
//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
|
||||
logger.debug("Concurrency exception getting queue outbox. Skipping: {} ", ex.getMessage());
|
||||
if (transaction != null)
|
||||
transaction.rollback();
|
||||
} catch (Exception ex) {
|
||||
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) {
|
||||
logger.error("Problem getting list of queue outbox. Skipping: {}", ex.getMessage(), ex);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,29 @@
|
|||
package gr.cite.notification.integrationevent.inbox.tenanttouched;
|
||||
|
||||
import gr.cite.notification.integrationevent.TrackedEvent;
|
||||
|
||||
import java.util.UUID;
|
||||
|
||||
public class TenantTouchedIntegrationEvent extends TrackedEvent {
|
||||
|
||||
private UUID id;
|
||||
|
||||
private String code;
|
||||
|
||||
public UUID getId() {
|
||||
return id;
|
||||
}
|
||||
|
||||
public void setId(UUID id) {
|
||||
this.id = id;
|
||||
}
|
||||
|
||||
public String getCode() {
|
||||
return code;
|
||||
}
|
||||
|
||||
public void setCode(String code) {
|
||||
this.code = code;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,7 @@
|
|||
package gr.cite.notification.integrationevent.inbox.tenanttouched;
|
||||
|
||||
import gr.cite.notification.integrationevent.inbox.IntegrationEventHandler;
|
||||
|
||||
public interface TenantTouchedIntegrationEventHandler extends IntegrationEventHandler {
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,103 @@
|
|||
package gr.cite.notification.integrationevent.inbox.tenanttouched;
|
||||
|
||||
import gr.cite.commons.web.oidc.principal.CurrentPrincipalResolver;
|
||||
import gr.cite.notification.audit.AuditableAction;
|
||||
import gr.cite.notification.common.JsonHandlingService;
|
||||
import gr.cite.notification.common.scope.fake.FakeRequestScope;
|
||||
import gr.cite.notification.integrationevent.inbox.EventProcessingStatus;
|
||||
import gr.cite.notification.integrationevent.inbox.InboxPrincipal;
|
||||
import gr.cite.notification.integrationevent.inbox.IntegrationEventProperties;
|
||||
import gr.cite.notification.model.persist.TenantTouchedIntegrationEventPersist;
|
||||
import gr.cite.notification.service.tenant.TenantService;
|
||||
import gr.cite.tools.auditing.AuditService;
|
||||
import gr.cite.tools.logging.LoggerService;
|
||||
import jakarta.persistence.EntityManager;
|
||||
import jakarta.persistence.EntityManagerFactory;
|
||||
import jakarta.persistence.EntityTransaction;
|
||||
import jakarta.persistence.OptimisticLockException;
|
||||
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
|
||||
@Scope(value = ConfigurableBeanFactory.SCOPE_PROTOTYPE)
|
||||
public class TenantTouchedIntegrationEventHandlerImpl implements TenantTouchedIntegrationEventHandler {
|
||||
|
||||
private static final LoggerService logger = new LoggerService(LoggerFactory.getLogger(TenantTouchedIntegrationEventHandlerImpl.class));
|
||||
|
||||
protected final ApplicationContext applicationContext;
|
||||
|
||||
private final JsonHandlingService jsonHandlingService;
|
||||
|
||||
public TenantTouchedIntegrationEventHandlerImpl(ApplicationContext applicationContext, JsonHandlingService jsonHandlingService) {
|
||||
this.applicationContext = applicationContext;
|
||||
this.jsonHandlingService = jsonHandlingService;
|
||||
}
|
||||
|
||||
@Override
|
||||
public EventProcessingStatus handle(IntegrationEventProperties properties, String message) {
|
||||
TenantTouchedIntegrationEvent event = this.jsonHandlingService.fromJsonSafe(TenantTouchedIntegrationEvent.class, message);
|
||||
if (event == null)
|
||||
return EventProcessingStatus.Error;
|
||||
|
||||
TenantTouchedIntegrationEventPersist model = new TenantTouchedIntegrationEventPersist();
|
||||
model.setId(event.getId());
|
||||
model.setCode(event.getCode());
|
||||
|
||||
EntityManager entityManager = null;
|
||||
EntityTransaction transaction = null;
|
||||
try (FakeRequestScope ignored = new FakeRequestScope()) {
|
||||
try {
|
||||
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 {
|
||||
TenantService tenantService = this.applicationContext.getBean(TenantService.class);
|
||||
tenantService.persist(model, null);
|
||||
|
||||
AuditService auditService = this.applicationContext.getBean(AuditService.class);
|
||||
|
||||
auditService.track(AuditableAction.Tenant_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
|
||||
logger.debug("Concurrency exception getting queue outbox. Skipping: {} ", ex.getMessage());
|
||||
if (transaction != null)
|
||||
transaction.rollback();
|
||||
} catch (Exception ex) {
|
||||
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) {
|
||||
logger.error("Problem getting list of queue outbox. Skipping: {}", ex.getMessage(), ex);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,25 @@
|
|||
package gr.cite.notification.integrationevent.inbox.userremoval;
|
||||
|
||||
import gr.cite.notification.integrationevent.inbox.ConsistencyHandler;
|
||||
import gr.cite.notification.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
|
||||
@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();
|
||||
return count != 0;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,23 @@
|
|||
package gr.cite.notification.integrationevent.inbox.userremoval;
|
||||
|
||||
import gr.cite.notification.integrationevent.inbox.ConsistencyPredicates;
|
||||
|
||||
import java.util.UUID;
|
||||
|
||||
public class UserRemovalConsistencyPredicates implements ConsistencyPredicates {
|
||||
|
||||
private UUID userId;
|
||||
|
||||
public UserRemovalConsistencyPredicates(UUID userId) {
|
||||
this.userId = userId;
|
||||
}
|
||||
|
||||
public UUID getUserId() {
|
||||
return userId;
|
||||
}
|
||||
|
||||
public void setUserId(UUID userId) {
|
||||
this.userId = userId;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,28 @@
|
|||
package gr.cite.notification.integrationevent.inbox.userremoval;
|
||||
|
||||
import gr.cite.notification.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;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,7 @@
|
|||
package gr.cite.notification.integrationevent.inbox.userremoval;
|
||||
|
||||
import gr.cite.notification.integrationevent.inbox.IntegrationEventHandler;
|
||||
|
||||
public interface UserRemovalIntegrationEventHandler extends IntegrationEventHandler {
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,139 @@
|
|||
package gr.cite.notification.integrationevent.inbox.userremoval;
|
||||
|
||||
import gr.cite.commons.web.oidc.principal.CurrentPrincipalResolver;
|
||||
import gr.cite.notification.audit.AuditableAction;
|
||||
import gr.cite.notification.common.JsonHandlingService;
|
||||
import gr.cite.notification.common.scope.fake.FakeRequestScope;
|
||||
import gr.cite.notification.common.scope.tenant.TenantScope;
|
||||
import gr.cite.notification.data.TenantEntity;
|
||||
import gr.cite.notification.errorcode.ErrorThesaurusProperties;
|
||||
import gr.cite.notification.integrationevent.inbox.EventProcessingStatus;
|
||||
import gr.cite.notification.integrationevent.inbox.InboxPrincipal;
|
||||
import gr.cite.notification.integrationevent.inbox.IntegrationEventProperties;
|
||||
import gr.cite.notification.model.Tenant;
|
||||
import gr.cite.notification.query.TenantQuery;
|
||||
import gr.cite.notification.service.user.UserService;
|
||||
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.EntityManager;
|
||||
import jakarta.persistence.EntityManagerFactory;
|
||||
import jakarta.persistence.EntityTransaction;
|
||||
import jakarta.persistence.OptimisticLockException;
|
||||
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
|
||||
@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;
|
||||
|
||||
private 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 ignored = 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
|
||||
logger.debug("Concurrency exception getting queue outbox. Skipping: {} ", ex.getMessage());
|
||||
if (transaction != null)
|
||||
transaction.rollback();
|
||||
} catch (Exception ex) {
|
||||
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) {
|
||||
logger.error("Problem getting list of queue outbox. Skipping: {}", ex.getMessage(), ex);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,48 @@
|
|||
package gr.cite.notification.integrationevent.inbox.usertouched;
|
||||
|
||||
import gr.cite.notification.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;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,7 @@
|
|||
package gr.cite.notification.integrationevent.inbox.usertouched;
|
||||
|
||||
import gr.cite.notification.integrationevent.inbox.IntegrationEventHandler;
|
||||
|
||||
public interface UserTouchedIntegrationEventHandler extends IntegrationEventHandler {
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,129 @@
|
|||
package gr.cite.notification.integrationevent.inbox.usertouched;
|
||||
|
||||
import gr.cite.commons.web.oidc.principal.CurrentPrincipalResolver;
|
||||
import gr.cite.notification.audit.AuditableAction;
|
||||
import gr.cite.notification.common.JsonHandlingService;
|
||||
import gr.cite.notification.common.scope.fake.FakeRequestScope;
|
||||
import gr.cite.notification.common.scope.tenant.TenantScope;
|
||||
import gr.cite.notification.data.TenantEntity;
|
||||
import gr.cite.notification.integrationevent.inbox.EventProcessingStatus;
|
||||
import gr.cite.notification.integrationevent.inbox.InboxPrincipal;
|
||||
import gr.cite.notification.integrationevent.inbox.IntegrationEventProperties;
|
||||
import gr.cite.notification.model.Tenant;
|
||||
import gr.cite.notification.model.persist.UserTouchedIntegrationEventPersist;
|
||||
import gr.cite.notification.query.TenantQuery;
|
||||
import gr.cite.notification.service.user.UserService;
|
||||
import gr.cite.tools.auditing.AuditService;
|
||||
import gr.cite.tools.data.query.QueryFactory;
|
||||
import gr.cite.tools.fieldset.BaseFieldSet;
|
||||
import gr.cite.tools.logging.LoggerService;
|
||||
import jakarta.persistence.EntityManager;
|
||||
import jakarta.persistence.EntityManagerFactory;
|
||||
import jakarta.persistence.EntityTransaction;
|
||||
import jakarta.persistence.OptimisticLockException;
|
||||
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
|
||||
@Scope(value = ConfigurableBeanFactory.SCOPE_PROTOTYPE)
|
||||
public class UserTouchedIntegrationEventHandlerImpl implements UserTouchedIntegrationEventHandler {
|
||||
|
||||
private static final LoggerService logger = new LoggerService(LoggerFactory.getLogger(UserTouchedIntegrationEventHandlerImpl.class));
|
||||
|
||||
protected final ApplicationContext applicationContext;
|
||||
|
||||
private final JsonHandlingService jsonHandlingService;
|
||||
|
||||
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 ignored = 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
|
||||
logger.debug("Concurrency exception getting queue outbox. Skipping: {} ", ex.getMessage());
|
||||
if (transaction != null)
|
||||
transaction.rollback();
|
||||
} catch (Exception ex) {
|
||||
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) {
|
||||
logger.error("Problem getting list of queue outbox. Skipping: {}", ex.getMessage(), ex);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,95 @@
|
|||
package gr.cite.notification.integrationevent.outbox;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.validation.annotation.Validated;
|
||||
|
||||
import javax.validation.constraints.NotNull;
|
||||
|
||||
@Validated
|
||||
@ConfigurationProperties(prefix = "queue.task.publisher.options")
|
||||
public class OutboxProperties {
|
||||
|
||||
@NotNull
|
||||
private final String exchange;
|
||||
|
||||
@NotNull
|
||||
private final String tenantTouchTopic;
|
||||
|
||||
@NotNull
|
||||
private final String tenantRemovalTopic;
|
||||
|
||||
@NotNull
|
||||
private final String tenantReactivationTopic;
|
||||
|
||||
@NotNull
|
||||
private final String tenantUserInviteTopic;
|
||||
|
||||
@NotNull
|
||||
private final String notifyTopic;
|
||||
|
||||
@NotNull
|
||||
private final String forgetMeCompletedTopic;
|
||||
|
||||
@NotNull
|
||||
private final String whatYouKnowAboutMeCompletedTopic;
|
||||
|
||||
@NotNull
|
||||
private final String generateFileTopic;
|
||||
|
||||
public OutboxProperties(String exchange,
|
||||
String tenantTouchTopic,
|
||||
String tenantRemovalTopic,
|
||||
String tenantReactivationTopic,
|
||||
String tenantUserInviteTopic,
|
||||
String notifyTopic,
|
||||
String forgetMeCompletedTopic,
|
||||
String whatYouKnowAboutMeCompletedTopic,
|
||||
String generateFileTopic
|
||||
) {
|
||||
this.exchange = exchange;
|
||||
this.tenantTouchTopic = tenantTouchTopic;
|
||||
this.tenantRemovalTopic = tenantRemovalTopic;
|
||||
this.tenantReactivationTopic = tenantReactivationTopic;
|
||||
this.tenantUserInviteTopic = tenantUserInviteTopic;
|
||||
this.notifyTopic = notifyTopic;
|
||||
this.forgetMeCompletedTopic = forgetMeCompletedTopic;
|
||||
this.whatYouKnowAboutMeCompletedTopic = whatYouKnowAboutMeCompletedTopic;
|
||||
this.generateFileTopic = generateFileTopic;
|
||||
}
|
||||
|
||||
public String getExchange() {
|
||||
return exchange;
|
||||
}
|
||||
|
||||
public String getTenantTouchTopic() {
|
||||
return tenantTouchTopic;
|
||||
}
|
||||
|
||||
public String getTenantRemovalTopic() {
|
||||
return tenantRemovalTopic;
|
||||
}
|
||||
|
||||
public String getTenantReactivationTopic() {
|
||||
return tenantReactivationTopic;
|
||||
}
|
||||
|
||||
public String getTenantUserInviteTopic() {
|
||||
return tenantUserInviteTopic;
|
||||
}
|
||||
|
||||
public String getNotifyTopic() {
|
||||
return notifyTopic;
|
||||
}
|
||||
|
||||
public String getForgetMeCompletedTopic() {
|
||||
return forgetMeCompletedTopic;
|
||||
}
|
||||
|
||||
public String getWhatYouKnowAboutMeCompletedTopic() {
|
||||
return whatYouKnowAboutMeCompletedTopic;
|
||||
}
|
||||
|
||||
public String getGenerateFileTopic() {
|
||||
return generateFileTopic;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
package gr.cite.notification.model.persist;
|
||||
|
||||
import gr.cite.notification.common.validation.ValidId;
|
||||
|
||||
import javax.validation.constraints.NotEmpty;
|
||||
import javax.validation.constraints.NotNull;
|
||||
import javax.validation.constraints.Size;
|
||||
import java.util.UUID;
|
||||
|
||||
public class TenantTouchedIntegrationEventPersist {
|
||||
|
||||
@ValidId(message = "{validation.invalidid}")
|
||||
@NotNull(message = "{validation.empty}")
|
||||
private UUID id;
|
||||
|
||||
@NotNull(message = "{validation.empty}")
|
||||
@NotEmpty(message = "{validation.empty}")
|
||||
@Size(max = 50, message = "{validation.largerthanmax}")
|
||||
private String code;
|
||||
|
||||
public UUID getId() {
|
||||
return id;
|
||||
}
|
||||
|
||||
public void setId(UUID id) {
|
||||
this.id = id;
|
||||
}
|
||||
|
||||
public String getCode() {
|
||||
return code;
|
||||
}
|
||||
|
||||
public void setCode(String code) {
|
||||
this.code = code;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,23 @@
|
|||
package gr.cite.notification.service.tenant;
|
||||
|
||||
import com.fasterxml.jackson.core.JsonProcessingException;
|
||||
import gr.cite.notification.model.Tenant;
|
||||
import gr.cite.notification.model.User;
|
||||
import gr.cite.notification.model.persist.TenantTouchedIntegrationEventPersist;
|
||||
import gr.cite.notification.model.persist.UserTouchedIntegrationEventPersist;
|
||||
import gr.cite.tools.exception.MyApplicationException;
|
||||
import gr.cite.tools.exception.MyForbiddenException;
|
||||
import gr.cite.tools.exception.MyNotFoundException;
|
||||
import gr.cite.tools.exception.MyValidationException;
|
||||
import gr.cite.tools.fieldset.FieldSet;
|
||||
|
||||
import javax.management.InvalidApplicationException;
|
||||
import java.util.UUID;
|
||||
|
||||
public interface TenantService {
|
||||
|
||||
Tenant persist(TenantTouchedIntegrationEventPersist model, FieldSet fields) throws MyForbiddenException, MyValidationException, MyApplicationException, MyNotFoundException, InvalidApplicationException, JsonProcessingException;
|
||||
|
||||
void deleteAndSave(UUID id) throws MyForbiddenException, InvalidApplicationException;
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,51 @@
|
|||
package gr.cite.notification.service.tenant;
|
||||
|
||||
import com.fasterxml.jackson.core.JsonProcessingException;
|
||||
import gr.cite.commons.web.authz.service.AuthorizationService;
|
||||
import gr.cite.notification.authorization.Permission;
|
||||
import gr.cite.notification.model.Tenant;
|
||||
import gr.cite.notification.model.deleter.TenantDeleter;
|
||||
import gr.cite.notification.model.deleter.UserDeleter;
|
||||
import gr.cite.notification.model.persist.TenantTouchedIntegrationEventPersist;
|
||||
import gr.cite.notification.service.user.UserServiceImpl;
|
||||
import gr.cite.tools.data.deleter.DeleterFactory;
|
||||
import gr.cite.tools.exception.MyApplicationException;
|
||||
import gr.cite.tools.exception.MyForbiddenException;
|
||||
import gr.cite.tools.exception.MyNotFoundException;
|
||||
import gr.cite.tools.exception.MyValidationException;
|
||||
import gr.cite.tools.fieldset.FieldSet;
|
||||
import gr.cite.tools.logging.LoggerService;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import javax.management.InvalidApplicationException;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
|
||||
public class TenantServiceImpl implements TenantService {
|
||||
|
||||
private static final LoggerService logger = new LoggerService(LoggerFactory.getLogger(TenantServiceImpl.class));
|
||||
|
||||
private final AuthorizationService authorizationService;
|
||||
|
||||
private final DeleterFactory deleterFactory;
|
||||
|
||||
public TenantServiceImpl(AuthorizationService authorizationService, DeleterFactory deleterFactory) {
|
||||
this.authorizationService = authorizationService;
|
||||
this.deleterFactory = deleterFactory;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Tenant persist(TenantTouchedIntegrationEventPersist model, FieldSet fields) throws MyForbiddenException, MyValidationException, MyApplicationException, MyNotFoundException, InvalidApplicationException, JsonProcessingException {
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void deleteAndSave(UUID id) throws MyForbiddenException, InvalidApplicationException {
|
||||
logger.debug("deleting Tenant: {}", id);
|
||||
|
||||
this.authorizationService.authorizeForce(Permission.DeleteTenant);
|
||||
|
||||
this.deleterFactory.deleter(TenantDeleter.class).deleteAndSaveByIds(List.of(id));
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,21 @@
|
|||
package gr.cite.notification.service.user;
|
||||
|
||||
import com.fasterxml.jackson.core.JsonProcessingException;
|
||||
import gr.cite.notification.model.User;
|
||||
import gr.cite.notification.model.persist.UserTouchedIntegrationEventPersist;
|
||||
import gr.cite.tools.exception.MyApplicationException;
|
||||
import gr.cite.tools.exception.MyForbiddenException;
|
||||
import gr.cite.tools.exception.MyNotFoundException;
|
||||
import gr.cite.tools.exception.MyValidationException;
|
||||
import gr.cite.tools.fieldset.FieldSet;
|
||||
|
||||
import javax.management.InvalidApplicationException;
|
||||
import java.util.UUID;
|
||||
|
||||
public interface UserService {
|
||||
|
||||
User persist(UserTouchedIntegrationEventPersist model, FieldSet fields) throws MyForbiddenException, MyValidationException, MyApplicationException, MyNotFoundException, InvalidApplicationException, JsonProcessingException;
|
||||
|
||||
void deleteAndSave(UUID id) throws MyForbiddenException, InvalidApplicationException;
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,51 @@
|
|||
package gr.cite.notification.service.user;
|
||||
|
||||
import com.fasterxml.jackson.core.JsonProcessingException;
|
||||
import gr.cite.commons.web.authz.service.AuthorizationService;
|
||||
import gr.cite.notification.authorization.Permission;
|
||||
import gr.cite.notification.model.User;
|
||||
import gr.cite.notification.model.deleter.UserDeleter;
|
||||
import gr.cite.notification.model.persist.UserTouchedIntegrationEventPersist;
|
||||
import gr.cite.tools.data.deleter.DeleterFactory;
|
||||
import gr.cite.tools.exception.MyApplicationException;
|
||||
import gr.cite.tools.exception.MyForbiddenException;
|
||||
import gr.cite.tools.exception.MyNotFoundException;
|
||||
import gr.cite.tools.exception.MyValidationException;
|
||||
import gr.cite.tools.fieldset.FieldSet;
|
||||
import gr.cite.tools.logging.LoggerService;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import javax.management.InvalidApplicationException;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
|
||||
@Service
|
||||
public class UserServiceImpl implements UserService {
|
||||
|
||||
private static final LoggerService logger = new LoggerService(LoggerFactory.getLogger(UserServiceImpl.class));
|
||||
|
||||
private final AuthorizationService authorizationService;
|
||||
|
||||
private final DeleterFactory deleterFactory;
|
||||
|
||||
public UserServiceImpl(AuthorizationService authorizationService, DeleterFactory deleterFactory) {
|
||||
this.authorizationService = authorizationService;
|
||||
this.deleterFactory = deleterFactory;
|
||||
}
|
||||
|
||||
@Override
|
||||
public User persist(UserTouchedIntegrationEventPersist model, FieldSet fields) throws MyForbiddenException, MyValidationException, MyApplicationException, MyNotFoundException, InvalidApplicationException, JsonProcessingException {
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void deleteAndSave(UUID id) throws MyForbiddenException, InvalidApplicationException {
|
||||
logger.debug("deleting User: {}", id);
|
||||
|
||||
this.authorizationService.authorizeForce(Permission.DeleteUser);
|
||||
|
||||
this.deleterFactory.deleter(UserDeleter.class).deleteAndSaveByIds(List.of(id));
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -29,6 +29,24 @@
|
|||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<!-- https://mvnrepository.com/artifact/jakarta.annotation/jakarta.annotation-api -->
|
||||
<dependency>
|
||||
<groupId>jakarta.annotation</groupId>
|
||||
<artifactId>jakarta.annotation-api</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- https://mvnrepository.com/artifact/jakarta.validation/validation-api -->
|
||||
<dependency>
|
||||
<groupId>jakarta.validation</groupId>
|
||||
<artifactId>jakarta.validation-api</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>jakarta.persistence</groupId>
|
||||
<artifactId>jakarta.persistence-api</artifactId>
|
||||
<version>3.1.0</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>gr.cite</groupId>
|
||||
<artifactId>queue-inbox</artifactId>
|
||||
|
|
@ -39,6 +57,12 @@
|
|||
<artifactId>queue-outbox</artifactId>
|
||||
<version>1.0.0</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>gr.cite</groupId>
|
||||
<artifactId>rabbitmq-core</artifactId>
|
||||
<version>1.0.0</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<profiles>
|
||||
|
|
|
|||
Loading…
Reference in New Issue