apply transcation on flow and job

Author: gy2006

core/debug.yaml

@@ -3,11 +3,18 @@
 version: '3'
 services:
  db:
- image: mongo:4.2
+ image: bitnami/mongodb:5.0
  restart: always
  container_name: flowci-debug-db
  ports:
  - "27017:27017"
+ environment:
+ MONGODB_REPLICA_SET_MODE: primary
+ MONGODB_ROOT_PASSWORD: flowci
+ MONGODB_USERNAME: flowci
+ MONGODB_PASSWORD: flowci
+ MONGODB_REPLICA_SET_KEY: replicakey123
+ MONGODB_DATABASE: flow_db
 
  zk:
  image: zookeeper:3.6

core/src/main/java/com/flowci/core/api/service/OpenRestServiceImpl.java

@@ -22,8 +22,9 @@
 import com.flowci.core.common.helper.DateHelper;
 import com.flowci.core.config.domain.Config;
 import com.flowci.core.config.service.ConfigService;
-import com.flowci.core.flow.dao.FlowUserDao;
+import com.flowci.core.flow.dao.FlowUsersDao;
 import com.flowci.core.flow.domain.Flow;
+import com.flowci.core.flow.domain.FlowUsers;
 import com.flowci.core.flow.domain.MatrixCounter;
 import com.flowci.core.flow.service.FlowService;
 import com.flowci.core.flow.service.MatrixService;
@@ -54,7 +55,7 @@
 public class OpenRestServiceImpl implements OpenRestService {
 
  @Autowired
- private FlowUserDao flowUserDao;
+ private FlowUsersDao flowUsersDao;
 
  @Autowired
  private JobDao jobDao;
@@ -146,7 +147,12 @@ public void addToJobContext(String flowName, long buildNumber, Map<String, Strin
  @Override
  public List<User> users(String flowName) {
  Flow flow = flowService.get(flowName);
- List<String> emails = flowUserDao.findAllUsers(flow.getId());
+ Optional<FlowUsers> optional = flowUsersDao.findById(flow.getId());
+ if (optional.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ var emails = optional.get().getUsers();
  List<User> users = new ArrayList<>(emails.size());
  for (String email : emails) {
  User user = new User();

core/src/main/java/com/flowci/core/common/config/MongoConfig.java

@@ -34,6 +34,7 @@
 import com.flowci.core.trigger.domain.WebhookTrigger;
 import com.mongodb.ConnectionString;
 import com.mongodb.MongoClientSettings;
+import com.mongodb.MongoCredential;
 import com.mongodb.client.MongoClient;
 import com.mongodb.client.MongoClients;
 import lombok.NonNull;
@@ -43,6 +44,8 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.convert.converter.Converter;
+import org.springframework.data.mongodb.MongoDatabaseFactory;
+import org.springframework.data.mongodb.MongoTransactionManager;
 import org.springframework.data.mongodb.config.AbstractMongoClientConfiguration;
 import org.springframework.data.mongodb.config.EnableMongoAuditing;
 import org.springframework.data.mongodb.core.convert.MongoCustomConversions;
@@ -72,16 +75,16 @@ public MongoConfig(AppProperties appProperties, MongoProperties mongoProperties,
  this.objectMapper = objectMapper;
  }
 
+ @Bean
+ public MongoTransactionManager transactionManager(MongoDatabaseFactory dbFactory) {
+ return new MongoTransactionManager(dbFactory);
+ }
+
  @NonNull
  @Override
  public MongoClient mongoClient() {
  log.info("Mongo URI: {}", mongoProperties.getUri());
- ConnectionString connectionString = new ConnectionString(mongoProperties.getUri());
- MongoClientSettings mongoClientSettings = MongoClientSettings.builder()
- .applyConnectionString(connectionString)
- .build();
-
- return MongoClients.create(mongoClientSettings);
+ return MongoClients.create(mongoProperties.getUri());
  }
 
  @NonNull

core/src/main/java/com/flowci/core/flow/dao/FlowUserDao.java renamed to core/src/main/java/com/flowci/core/flow/dao/CustomizedFlowUserDao.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2019 flow.ci
+ * Copyright (c) 2023 flow.ci
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,25 +20,13 @@
 import java.util.Collection;
 import java.util.List;
 
-public interface FlowUserDao {
-
- void create(String flowId);
-
- /**
- * Remove flow user list
- */
- void delete(String flowId);
+public interface CustomizedFlowUserDao {
 
  /**
  * Find all flows by users email
  */
  List<String> findAllFlowsByUserEmail(String email);
 
- /**
- * Find all users by flow id
- */
- List<String> findAllUsers(String flowId);
-
  /**
  * Batch insert users
  */

core/src/main/java/com/flowci/core/flow/dao/CustomizedFlowUserDaoImpl.java

@@ -0,0 +1,54 @@
+package com.flowci.core.flow.dao;
+
+import com.flowci.core.flow.domain.FlowUsers;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.query.Criteria;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.data.mongodb.core.query.Update;
+
+import java.util.*;
+
+public class CustomizedFlowUserDaoImpl implements CustomizedFlowUserDao {
+
+ @Autowired
+ private MongoTemplate mongoTemplate;
+
+ @Override
+ public List<String> findAllFlowsByUserEmail(String email) {
+ Query q = Query.query(Criteria.where("users").in(email));
+ q.fields().exclude("users");
+
+ List<FlowUsers> lists = mongoTemplate.find(q, FlowUsers.class);
+ List<String> ids = new LinkedList<>();
+
+ for (FlowUsers item : lists) {
+ ids.add(item.getId());
+ }
+
+ return ids;
+ }
+
+ @Override
+ public void insert(String flowId, Collection<String> emails) {
+ Query q = Query.query(Criteria.where("_id").is(flowId));
+ Update u = new Update().addToSet("users").each(emails);
+ mongoTemplate.upsert(q, u, FlowUsers.class);
+ }
+
+ @Override
+ public void remove(String flowId, Collection<String> emails) {
+ Query q = Query.query(Criteria.where("_id").is(flowId));
+ Update u = new Update();
+ u.pullAll("users", emails.toArray());
+ mongoTemplate.updateFirst(q, u, FlowUsers.class);
+ }
+
+ @Override
+ public boolean exist(String flowId, String email) {
+ Query q = new Query();
+ q.addCriteria(Criteria.where("_id").is(flowId));
+ q.addCriteria(Criteria.where("users").is(email));
+ return mongoTemplate.exists(q, FlowUsers.class);
+ }
+}

core/src/main/java/com/flowci/core/flow/dao/FlowUserDaoImpl.java

@@ -0,0 +1,9 @@
+package com.flowci.core.flow.dao;
+
+import com.flowci.core.flow.domain.FlowUsers;
+import org.springframework.data.mongodb.repository.MongoRepository;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public interface FlowUsersDao extends MongoRepository<FlowUsers, String>, CustomizedFlowUserDao {
+}

core/src/main/java/com/flowci/core/flow/dao/FlowUsersDao.java

@@ -17,29 +17,31 @@
 
 package com.flowci.core.flow.domain;
 
+import com.flowci.core.common.domain.Mongoable;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.Setter;
 import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.index.Indexed;
 import org.springframework.data.mongodb.core.mapping.Document;
 
 import java.util.LinkedList;
 import java.util.List;
 
 @Getter
 @Setter
-@EqualsAndHashCode(of = {"flowId"})
 @Document(collection = "flow_users")
 @NoArgsConstructor
-public class FlowUsers {
-
- @Id
- private String flowId;
+public class FlowUsers extends Mongoable {
 
  private List<String> users = new LinkedList<>(); // user email list
 
  public FlowUsers(String flowId) {
- this.flowId = flowId;
+ this.id = flowId;
+ }
+
+ public void add(String email) {
+ users.add(email);
  }
 }

core/src/main/java/com/flowci/core/flow/domain/FlowUsers.java

@@ -3,7 +3,7 @@
 import com.flowci.core.common.manager.SessionManager;
 import com.flowci.core.flow.dao.FlowDao;
 import com.flowci.core.flow.dao.FlowGroupDao;
-import com.flowci.core.flow.dao.FlowUserDao;
+import com.flowci.core.flow.dao.FlowUsersDao;
 import com.flowci.core.flow.domain.Flow;
 import com.flowci.core.flow.domain.FlowGroup;
 import com.flowci.core.flow.domain.FlowItem;
@@ -12,9 +12,9 @@
 import com.flowci.exception.NotFoundException;
 import com.google.common.collect.Sets;
 import lombok.AllArgsConstructor;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.dao.DuplicateKeyException;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 
 import java.util.List;
 import java.util.Optional;
@@ -27,7 +27,7 @@ public class FlowGroupServiceImpl implements FlowGroupService {
 
  private final FlowGroupDao flowGroupDao;
 
- private final FlowUserDao flowUserDao;
+ private final FlowUsersDao flowUsersDao;
 
  private final SessionManager sessionManager;
 
@@ -50,6 +50,7 @@ public FlowGroup getById(String id) {
  }
 
  @Override
+ @Transactional
  public FlowGroup create(String name) {
  var email = sessionManager.getUserEmail();
 
@@ -58,7 +59,7 @@ public FlowGroup create(String name) {
 
  try {
  flowGroupDao.save(group);
- flowUserDao.insert(group.getId(), Sets.newHashSet(email));
+ flowUsersDao.insert(group.getId(), Sets.newHashSet(email));
  return group;
  } catch (DuplicateKeyException e) {
  throw new DuplicateException("Group {0} already exists", name);