This is the fix for the intermittent issues when applying policies on devices

Please find the GitHub issue from [1]

[1]: https://github.com/wso2/product-iots/issues/1833

components/device-mgt/org.wso2.carbon.device.mgt.common/src/main/java/org/wso2/carbon/device/mgt/common/operation/mgt/OperationManager.java

@@ -22,6 +22,8 @@ import org.wso2.carbon.device.mgt.common.DeviceManagementException;
 import org.wso2.carbon.device.mgt.common.InvalidDeviceException;
 import org.wso2.carbon.device.mgt.common.PaginationRequest;
 import org.wso2.carbon.device.mgt.common.PaginationResult;
+import org.wso2.carbon.device.mgt.common.policy.mgt.Policy;
+
 import org.wso2.carbon.device.mgt.common.push.notification.NotificationStrategy;
 
 import java.util.List;
@@ -44,6 +46,10 @@ public interface OperationManager {
     Activity addOperation(Operation operation, List<DeviceIdentifier> devices) throws OperationManagementException,
             InvalidDeviceException;
 
+
+    void addOperationsForPolicyRevoke(Policy policy, List<DeviceIdentifier> devices) throws OperationManagementException,
+            InvalidDeviceException;
+
     /**
      * Method to retrieve the list of all operations to a device.
      *

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/operation/mgt/OperationManagerImpl.java

@@ -38,6 +38,8 @@ import org.wso2.carbon.device.mgt.common.operation.mgt.ActivityStatus;
 import org.wso2.carbon.device.mgt.common.operation.mgt.Operation;
 import org.wso2.carbon.device.mgt.common.operation.mgt.OperationManagementException;
 import org.wso2.carbon.device.mgt.common.operation.mgt.OperationManager;
+import org.wso2.carbon.device.mgt.common.policy.mgt.Policy;
+import org.wso2.carbon.device.mgt.common.policy.mgt.ProfileFeature;
 import org.wso2.carbon.device.mgt.common.push.notification.NotificationContext;
 import org.wso2.carbon.device.mgt.common.push.notification.NotificationStrategy;
 import org.wso2.carbon.device.mgt.common.push.notification.PushNotificationConfig;
@@ -64,13 +66,7 @@ import org.wso2.carbon.device.mgt.core.task.impl.DeviceTaskManagerImpl;
 import org.wso2.carbon.device.mgt.core.util.DeviceManagerUtil;
 
 import java.sql.SQLException;
-import java.util.ArrayList;
+import java.util.*;
-import java.util.Calendar;
-import java.util.Collections;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
 
 /**
  * This class implements all the functionality exposed as part of the OperationManager. Any transaction initiated
@@ -326,6 +322,160 @@ public class OperationManagerImpl implements OperationManager {
         }
     }
 
+    private Operation getPolicyRevokeOperation() {
+        CommandOperation policyRevokeOperation = new CommandOperation();
+        policyRevokeOperation.setEnabled(true);
+        policyRevokeOperation.setCode(OperationMgtConstants.OperationCodes.POLICY_REVOKE);
+        policyRevokeOperation.setType(Operation.Type.COMMAND);
+        return policyRevokeOperation;
+    }
+    private Operation transformPolicy(Policy policy) {
+        List<ProfileFeature> effectiveFeatures = policy.getProfile().getProfileFeaturesList();
+        List<ProfileOperation> profileOperationList = new ArrayList<ProfileOperation>();
+        PolicyOperation policyOperation = new PolicyOperation();
+        policyOperation.setEnabled(true);
+        policyOperation.setType(org.wso2.carbon.device.mgt.common.operation.mgt.Operation.Type.POLICY);
+        policyOperation.setCode(PolicyOperation.POLICY_OPERATION_CODE);
+        for (ProfileFeature feature : effectiveFeatures) {
+            ProfileOperation profileOperation = new ProfileOperation();
+            profileOperation.setCode(feature.getFeatureCode());
+            profileOperation.setEnabled(true);
+            profileOperation.setStatus(org.wso2.carbon.device.mgt.common.operation.mgt.Operation.Status.PENDING);
+            profileOperation.setType(org.wso2.carbon.device.mgt.common.operation.mgt.Operation.Type.PROFILE);
+            profileOperation.setPayLoad(feature.getContent());
+            profileOperationList.add(profileOperation);
+        }
+        policyOperation.setProfileOperations(profileOperationList);
+        policyOperation.setPayLoad(policyOperation.getProfileOperations());
+        return policyOperation;
+    }
+    @Override
+    public void addOperationsForPolicyRevoke(Policy policy, List<DeviceIdentifier> deviceIds)
+            throws OperationManagementException, InvalidDeviceException {
+        Operation revokeOperation = getPolicyRevokeOperation();
+        Operation operation = transformPolicy(policy);
+        if (log.isDebugEnabled()) {
+            log.debug("operation:[" + operation.toString() + "]");
+            for (DeviceIdentifier deviceIdentifier : deviceIds) {
+                log.debug("device identifier id:[" + deviceIdentifier.getId() + "] type:[" +
+                        deviceIdentifier.getType() + "]");
+            }
+        }
+        try {
+            DeviceIDHolder deviceValidationResult = DeviceManagerUtil.validateDeviceIdentifiers(deviceIds);
+            List<DeviceIdentifier> validDeviceIds = deviceValidationResult.getValidDeviceIDList();
+            if (validDeviceIds.size() > 0) {
+                DeviceIDHolder deviceAuthorizationResult = this.authorizeDevices(operation, validDeviceIds);
+                List<DeviceIdentifier> authorizedDeviceList = deviceAuthorizationResult.getValidDeviceIDList();
+                OperationManagementDAOFactory.beginTransaction();
+                org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation policyOperationDto =
+                        OperationDAOUtil.convertOperation(operation);
+                org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation revokeOperationDto =
+                        OperationDAOUtil.convertOperation(revokeOperation);
+                boolean isScheduledOperation = this.isTaskScheduledOperation(operation);
+                boolean isNotRepeated = false;
+                boolean isScheduled = false;
+                NotificationStrategy notificationStrategy = getNotificationStrategy();
+                // check whether device list is greater than batch size notification strategy has enable to send push
+                // notification using scheduler task
+                if (DeviceConfigurationManager.getInstance().getDeviceManagementConfig().
+                        getPushNotificationConfiguration().getSchedulerBatchSize() <= authorizedDeviceList.size() &&
+                        notificationStrategy != null) {
+                    isScheduled = notificationStrategy.getConfig().isScheduled();
+                }
+                List<org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation> operationList = new LinkedList<org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation>();
+                operationList.add(revokeOperationDto);
+                operationList.add(policyOperationDto);
+                List<Integer> operationIds = this.lookupOperationDAO(operation).addOperations(operationList);
+                List<Device> devices = new ArrayList<>();
+                if (org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation.Control.NO_REPEAT == policyOperationDto.
+                        getControl()) {
+                    isNotRepeated = true;
+                }
+                //Need to happen for both revoke and new policy operation
+                addOperationMappings(authorizedDeviceList, revokeOperationDto, operationIds.get(0), isScheduledOperation,
+                        isNotRepeated, isScheduled, devices);
+                sendPushNotifications(revokeOperation, operationIds.get(0), isScheduled, notificationStrategy, devices);
+                //Need to happen for both revoke and new policy operation
+                addOperationMappings(authorizedDeviceList, policyOperationDto, operationIds.get(1), isScheduledOperation,
+                        isNotRepeated, isScheduled, devices);
+                sendPushNotifications(operation, operationIds.get(1), isScheduled, notificationStrategy, devices);
+                OperationManagementDAOFactory.commitTransaction();
+            } else {
+                throw new InvalidDeviceException("Invalid device Identifiers found.");
+            }
+        } catch (OperationManagementDAOException e) {
+            OperationManagementDAOFactory.rollbackTransaction();
+            throw new OperationManagementException("Error occurred while adding operation", e);
+        } catch (TransactionManagementException e) {
+            throw new OperationManagementException("Error occurred while initiating the transaction", e);
+        } finally {
+            OperationManagementDAOFactory.closeConnection();
+        }
+    }
+    private String addOperationMappings(List<DeviceIdentifier> authorizedDeviceList, org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation operationDto, int operationId, boolean isScheduledOperation, boolean isNotRepeated, boolean isScheduled, List<Device> devices) throws OperationManagementException, OperationManagementDAOException {
+        int enrolmentId;
+        boolean hasExistingTaskOperation;//TODO have to create a sql to load device details from deviceDAO using single query.
+        String operationCode = operationDto.getCode();
+        for (DeviceIdentifier deviceId : authorizedDeviceList) {
+            Device device = getDevice(deviceId);
+            devices.add(device);
+            enrolmentId = device.getEnrolmentInfo().getId();
+            //Do not repeat the task operations
+            if (isScheduledOperation) {
+                hasExistingTaskOperation = operationDAO.updateTaskOperation(enrolmentId, operationCode);
+                if (!hasExistingTaskOperation) {
+                    operationMappingDAO.addOperationMapping(operationId, enrolmentId, isScheduled);
+                }
+            } else if (isNotRepeated) {
+                operationDAO.updateEnrollmentOperationsStatus(enrolmentId, operationCode,
+                        org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation.Status.PENDING,
+                        org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation.Status.REPEATED);
+                operationMappingDAO.addOperationMapping(operationId, enrolmentId, isScheduled);
+            } else {
+                operationMappingDAO.addOperationMapping(operationId, enrolmentId, isScheduled);
+            }
+        }
+        return operationCode;
+    }
+    /*
+     * If notification strategy has not enable to send push notification using scheduler task we will send
+     * notification immediately. This is done in separate loop inorder to prevent overlap with DB insert
+     * operations with the possible db update operations trigger followed by pending operation call.
+     * Otherwise device may call pending operation while DB is locked for write and deadlock can occur.
+     */
+    private void sendPushNotifications(Operation operation, int operationId, boolean isScheduled, NotificationStrategy notificationStrategy, List<Device> devices) {
+        int enrolmentId;
+        if (notificationStrategy != null && !isScheduled) {
+            for (Device device : devices) {
+                DeviceIdentifier deviceId = new DeviceIdentifier(device.getDeviceIdentifier(), device.getType());
+                if (log.isDebugEnabled()) {
+                    log.debug("Sending push notification to " + deviceId + " from add operation method.");
+                }
+                operation.setId(operationId);
+                operation.setActivityId(DeviceManagementConstants.OperationAttributes.ACTIVITY + operationId);
+                try {
+                    notificationStrategy.execute(new NotificationContext(deviceId, operation));
+                } catch (PushNotificationExecutionFailedException e) {
+                    log.error("Error occurred while sending push notifications to " + deviceId.getType() +
+                            " device carrying id '" + deviceId + "'", e);
+                    /*
+                     Reschedule if push notification failed. Doing db transactions in atomic way to prevent
+                     deadlocks.
+                     */
+                    enrolmentId = device.getEnrolmentInfo().getId();
+                    try {
+                        operationMappingDAO.updateOperationMapping(operationId, enrolmentId, org.wso2.carbon
+                                .device.mgt.core.dto.operation.mgt.Operation.PushNotificationStatus.SCHEDULED);
+                    } catch (OperationManagementDAOException ex) {
+                        // Not throwing this exception in order to keep sending remaining notifications if any.
+                        log.error("Error occurred while setting push notification status to SCHEDULED.", ex);
+                    }
+                }
+            }
+        }
+    }
+
     private List<ActivityStatus> getActivityStatus(DeviceIDHolder deviceIdValidationResult, DeviceIDHolder deviceAuthResult,
                                                    String deviceType) {
         List<ActivityStatus> activityStatuses = new ArrayList<>();

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/operation/mgt/dao/OperationDAO.java

@@ -30,6 +30,8 @@ public interface OperationDAO {
 
     int addOperation(Operation operation) throws OperationManagementDAOException;
 
+    List<Integer> addOperations(List<Operation> operations) throws OperationManagementDAOException;
+
     Operation getOperation(int operationId) throws OperationManagementDAOException;
 
     Operation getOperationByDeviceAndId(int enrolmentId, int operationId) throws OperationManagementDAOException;

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/operation/mgt/dao/impl/GenericOperationDAOImpl.java

@@ -86,6 +86,37 @@ public class GenericOperationDAOImpl implements OperationDAO {
         }
     }
 
+    //This implementation has been done this way due to H2 not supporting batch inserts properly.
+    //Even though records are added in batch mode, only the id of the last added record will be returned, which is a problem.
+    public List<Integer> addOperations(List<Operation> operations) throws OperationManagementDAOException {
+        List<Integer> ids = new LinkedList<Integer>();
+        for (Operation operation : operations) {
+            PreparedStatement stmt = null;
+            ResultSet rs = null;
+            try {
+                Connection connection = OperationManagementDAOFactory.getConnection();
+                String sql = "INSERT INTO DM_OPERATION(TYPE, CREATED_TIMESTAMP, RECEIVED_TIMESTAMP, OPERATION_CODE)  " +
+                        "VALUES (?, ?, ?, ?)";
+                stmt = connection.prepareStatement(sql, new String[]{"id"});
+                stmt.setString(1, operation.getType().toString());
+                stmt.setTimestamp(2, new Timestamp(new Date().getTime()));
+                stmt.setTimestamp(3, null);
+                stmt.setString(4, operation.getCode());
+                stmt.executeUpdate();
+                rs = stmt.getGeneratedKeys();
+                int id = -1;
+                if (rs.next()) {
+                    ids.add(rs.getInt(1));
+                }
+            } catch (SQLException e) {
+                throw new OperationManagementDAOException("Error occurred while adding operation metadata", e);
+            } finally {
+                OperationManagementDAOUtil.cleanupResources(stmt, rs);
+            }
+        }
+        return ids;
+    }
+
     public boolean updateOperationStatus(int enrolmentId, int operationId, Operation.Status status)
             throws OperationManagementDAOException {
         PreparedStatement stmt = null;

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/operation/mgt/dao/impl/PolicyOperationDAOImpl.java

@@ -20,6 +20,7 @@ package org.wso2.carbon.device.mgt.core.operation.mgt.dao.impl;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.wso2.carbon.device.mgt.core.dto.operation.mgt.CommandOperation;
 import org.wso2.carbon.device.mgt.core.dto.operation.mgt.Operation;
 import org.wso2.carbon.device.mgt.core.dto.operation.mgt.PolicyOperation;
 import org.wso2.carbon.device.mgt.core.operation.mgt.dao.OperationManagementDAOException;
@@ -46,7 +47,6 @@ public class PolicyOperationDAOImpl extends GenericOperationDAOImpl {
             operation.setCreatedTimeStamp(new Timestamp(new java.util.Date().getTime()).toString());
             operation.setId(operationId);
             operation.setEnabled(true);
-            PolicyOperation policyOperation = (PolicyOperation) operation;
             Connection conn = OperationManagementDAOFactory.getConnection();
             stmt = conn.prepareStatement("INSERT INTO DM_POLICY_OPERATION(OPERATION_ID, OPERATION_DETAILS) " +
                     "VALUES(?, ?)");
@@ -82,6 +82,79 @@ public class PolicyOperationDAOImpl extends GenericOperationDAOImpl {
         return operationId;
     }
 
+
+    private int addCommandOperation(int operationId, Operation operation) throws OperationManagementDAOException {
+        CommandOperation commandOp = (CommandOperation) operation;
+        PreparedStatement stmt = null;
+        try {
+            Connection conn = OperationManagementDAOFactory.getConnection();
+            stmt = conn.prepareStatement("INSERT INTO DM_COMMAND_OPERATION(OPERATION_ID, ENABLED) VALUES(?, ?)");
+            stmt.setInt(1, operationId);
+            stmt.setBoolean(2, commandOp.isEnabled());
+            stmt.executeUpdate();
+        } catch (SQLException e) {
+            throw new OperationManagementDAOException("Error occurred while adding command operation", e);
+        } finally {
+            OperationManagementDAOUtil.cleanupResources(stmt);
+        }
+        return operationId;
+    }
+    @Override
+    public List<Integer> addOperations(List<Operation> operations) throws OperationManagementDAOException {
+        List<Integer> operationIds;
+        int counter = 0;
+        operationIds = super.addOperations(operations);
+        for(Operation operation : operations) {
+            if(operation.getType().equals(Operation.Type.COMMAND)){
+                addCommandOperation(operationIds.get(counter), operation);
+            } else if(operation.getType().equals(Operation.Type.POLICY)){
+                addPolicyOperation(operationIds.get(counter), operation);
+            }
+            counter++;
+        }
+        return operationIds;
+    }
+    private void addPolicyOperation(int operationId, Operation operation) throws OperationManagementDAOException {
+        PreparedStatement stmt = null;
+        ByteArrayOutputStream bao = null;
+        ObjectOutputStream oos = null;
+        try {
+            operation.setCreatedTimeStamp(new Timestamp(new java.util.Date().getTime()).toString());
+            operation.setId(operationId);
+            operation.setEnabled(true);
+            Connection conn = OperationManagementDAOFactory.getConnection();
+            stmt = conn.prepareStatement("INSERT INTO DM_POLICY_OPERATION(OPERATION_ID, OPERATION_DETAILS) " +
+                    "VALUES(?, ?)");
+            bao = new ByteArrayOutputStream();
+            oos = new ObjectOutputStream(bao);
+            oos.writeObject(operation);
+            stmt.setInt(1, operationId);
+            stmt.setBytes(2, bao.toByteArray());
+            stmt.executeUpdate();
+        } catch (SQLException e) {
+            throw new OperationManagementDAOException("Error occurred while adding policy operation", e);
+        } catch (IOException e) {
+            throw new OperationManagementDAOException("Error occurred while serializing policy operation object", e);
+        } finally {
+            if (bao != null) {
+                try {
+                    bao.close();
+                } catch (IOException e) {
+                    log.warn("Error occurred while closing ByteArrayOutputStream", e);
+                }
+            }
+            if (oos != null) {
+                try {
+                    oos.close();
+                } catch (IOException e) {
+                    log.warn("Error occurred while closing ObjectOutputStream", e);
+                }
+            }
+            OperationManagementDAOUtil.cleanupResources(stmt);
+        }
+    }
+
+
     @Override
     public Operation getOperation(int operationId) throws OperationManagementDAOException {
         PreparedStatement stmt = null;

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/service/DeviceManagementProviderService.java

@@ -32,6 +32,7 @@ import org.wso2.carbon.device.mgt.common.license.mgt.License;
 import org.wso2.carbon.device.mgt.common.operation.mgt.Activity;
 import org.wso2.carbon.device.mgt.common.operation.mgt.Operation;
 import org.wso2.carbon.device.mgt.common.operation.mgt.OperationManagementException;
+import org.wso2.carbon.device.mgt.common.policy.mgt.Policy;
 import org.wso2.carbon.device.mgt.common.policy.mgt.PolicyMonitoringManager;
 import org.wso2.carbon.device.mgt.common.pull.notification.PullNotificationExecutionFailedException;
 import org.wso2.carbon.device.mgt.common.push.notification.NotificationStrategy;
@@ -541,6 +542,10 @@ public interface DeviceManagementProviderService {
     Activity addOperation(String type, Operation operation,
                           List<DeviceIdentifier> devices) throws OperationManagementException, InvalidDeviceException;
 
+    void addPolicyOperations(String type, Policy policy,
+            List<DeviceIdentifier> devices) throws OperationManagementException, InvalidDeviceException;
+
+
     List<? extends Operation> getOperations(DeviceIdentifier deviceId) throws OperationManagementException;
 
     PaginationResult getOperations(DeviceIdentifier deviceId,

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/service/DeviceManagementProviderServiceImpl.java

@@ -53,6 +53,7 @@ import org.wso2.carbon.device.mgt.common.operation.mgt.Activity;
 import org.wso2.carbon.device.mgt.common.operation.mgt.Operation;
 import org.wso2.carbon.device.mgt.common.operation.mgt.OperationManagementException;
 import org.wso2.carbon.device.mgt.common.operation.mgt.OperationManager;
+import org.wso2.carbon.device.mgt.common.policy.mgt.Policy;
 import org.wso2.carbon.device.mgt.common.policy.mgt.PolicyMonitoringManager;
 import org.wso2.carbon.device.mgt.common.pull.notification.PullNotificationExecutionFailedException;
 import org.wso2.carbon.device.mgt.common.pull.notification.PullNotificationSubscriber;
@@ -1423,6 +1424,12 @@ public class DeviceManagementProviderServiceImpl implements DeviceManagementProv
         return pluginRepository.getOperationManager(type, this.getTenantId()).addOperation(operation, devices);
     }
 
+    @Override
+    public void addPolicyOperations(String type, Policy policy,
+            List<DeviceIdentifier> devices) throws OperationManagementException, InvalidDeviceException {
+        pluginRepository.getOperationManager(type, this.getTenantId()).addOperationsForPolicyRevoke(policy, devices);
+    }
+
     @Override
     public List<? extends Operation> getOperations(DeviceIdentifier deviceId) throws OperationManagementException {
         return pluginRepository.getOperationManager(deviceId.getType(), this.getTenantId()).getOperations(deviceId);

components/policy-mgt/org.wso2.carbon.policy.mgt.core/src/main/java/org/wso2/carbon/policy/mgt/core/enforcement/PolicyEnforcementDelegator.java

@@ -33,4 +33,7 @@ public interface PolicyEnforcementDelegator {
 
     void addPolicyRevokeOperation(List<DeviceIdentifier> deviceIdentifiers) throws PolicyDelegationException;
 
+    void revokePolicyOperation(List<DeviceIdentifier> deviceIdentifiers, Policy policy) throws PolicyDelegationException;
+
+
 }

components/policy-mgt/org.wso2.carbon.policy.mgt.core/src/main/java/org/wso2/carbon/policy/mgt/core/enforcement/PolicyEnforcementDelegatorImpl.java

@@ -81,8 +81,7 @@ public class PolicyEnforcementDelegatorImpl implements PolicyEnforcementDelegato
                  */
                 if (devicePolicy == null || devicePolicy.getId() != policy.getId() || updatedPolicyIds.contains
                         (policy.getId())) {
-                    this.addPolicyRevokeOperation(deviceIdentifiers);
+                    this.revokePolicyOperation(deviceIdentifiers, policy);
-                    this.addPolicyOperation(deviceIdentifiers, policy);
                 }
             } else {
                 //This means all the applicable policies have been removed from device. Hence calling a policy revoke.
@@ -167,6 +166,27 @@ public class PolicyEnforcementDelegatorImpl implements PolicyEnforcementDelegato
         return policyRevokeOperation;
     }
 
+    @Override
+    public void revokePolicyOperation(List<DeviceIdentifier> deviceIdentifiers, Policy policy) throws
+            PolicyDelegationException {
+        try {
+            String type = null;
+            if (deviceIdentifiers.size() > 0) {
+                type = deviceIdentifiers.get(0).getType();
+            }
+            PolicyManagementDataHolder.getInstance().getDeviceManagementService().addPolicyOperations(type, policy,
+                    deviceIdentifiers);
+        } catch (InvalidDeviceException e) {
+            String msg = "Invalid DeviceIdentifiers found.";
+            log.error(msg, e);
+            throw new PolicyDelegationException(msg, e);
+        } catch (OperationManagementException e) {
+            String msg = "Error occurred while adding the operation to device.";
+            log.error(msg, e);
+            throw new PolicyDelegationException(msg, e);
+        }
+    }
+
     /**
      * Provides the applied policy for give device
      *