Feature/persist webhook events

src/main/java/com/impactupgrade/nucleus/controller/StripeController.java

@@ -32,6 +32,7 @@
 import com.stripe.model.Subscription;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.json.JSONObject;
 
 import javax.servlet.http.HttpServletRequest;
 import javax.ws.rs.POST;
@@ -99,10 +100,17 @@ public Response webhook(String json, @Context HttpServletRequest request) throws
       // takes a while, so spin it off as a new thread
       Runnable thread = () -> {
         try {
+          log.info("Saving event with id '{}'...", event.getId());
+          env.webhookRequestService().persist(event.getId(), "stripe", new JSONObject(event));
           processEvent(event.getType(), stripeObject, env);
+
+          // TODO: decide if we want to delete right away or keep it for future analysis/analytics
+          //env.webhookRequestService().delete(event.getId());
         } catch (Exception e) {
           log.error("failed to process the Stripe event", e);
           // TODO: email notification?
+          log.info("Updating event with id '{}' with error details...", event.getId());
+          env.webhookRequestService().updateWithErrorMessage(event.getId(), e.getMessage());
         }
       };
       new Thread(thread).start();

src/main/java/com/impactupgrade/nucleus/dao/Dao.java

@@ -0,0 +1,15 @@
+package com.impactupgrade.nucleus.dao;
+
+import java.util.Optional;
+
+public interface Dao<I, E> {
+
+    E create(E e);
+
+    Optional<E> get(I id);
+
+    E update(E e);
+
+    Optional<E> delete(I id);
+
+}

src/main/java/com/impactupgrade/nucleus/dao/WebhookRequestDao.java

@@ -0,0 +1,65 @@
+package com.impactupgrade.nucleus.dao;
+
+import com.impactupgrade.nucleus.model.WebhookRequest;
+import org.hibernate.Session;
+import org.hibernate.SessionFactory;
+import org.hibernate.Transaction;
+import org.hibernate.boot.registry.StandardServiceRegistryBuilder;
+import org.hibernate.cfg.Configuration;
+import org.springframework.util.StringUtils;
+
+import java.util.Optional;
+
+public class WebhookRequestDao implements Dao<String, WebhookRequest> {
+
+    private final SessionFactory sessionFactory;
+
+    public WebhookRequestDao() {
+        final Configuration configuration = new Configuration();
+        configuration.addAnnotatedClass(WebhookRequest.class);
+        this.sessionFactory = configuration.buildSessionFactory(new StandardServiceRegistryBuilder().build());
+    }
+
+    @Override
+    public WebhookRequest create(WebhookRequest webhookRequest) {
+        final Session session = sessionFactory.openSession();
+        Transaction transaction = session.beginTransaction();
+        session.save(webhookRequest);
+        transaction.commit();
+        session.close();
+        return webhookRequest;
+    }
+
+    @Override
+    public Optional<WebhookRequest> get(String id) {
+        if (StringUtils.isEmpty(id)) {
+            return Optional.empty();
+        }
+
+        final Session session = sessionFactory.openSession();
+        WebhookRequest webhookRequest = session.get(WebhookRequest.class, id);
+        session.close();
+        return Optional.ofNullable(webhookRequest);
+    }
+
+    @Override
+    public WebhookRequest update(WebhookRequest webhookRequest) {
+        final Session session = sessionFactory.openSession();
+        Transaction transaction = session.beginTransaction();
+        session.update(webhookRequest);
+        transaction.commit();
+        session.close();
+        return webhookRequest;
+    }
+
+    @Override
+    public Optional<WebhookRequest> delete(String id) {
+        final Session session = sessionFactory.openSession();
+        Transaction transaction = session.beginTransaction();
+        WebhookRequest webhookRequest = session.get(WebhookRequest.class, id);
+        session.delete(webhookRequest);
+        transaction.commit();
+        session.close();
+        return Optional.ofNullable(webhookRequest);
+    }
+}

src/main/java/com/impactupgrade/nucleus/environment/Environment.java

@@ -12,6 +12,7 @@
 import com.impactupgrade.nucleus.client.TwilioClient;
 import com.impactupgrade.nucleus.service.logic.ContactService;
 import com.impactupgrade.nucleus.service.logic.DonationService;
+import com.impactupgrade.nucleus.service.logic.WebhookRequestService;
 import com.impactupgrade.nucleus.service.logic.MessagingService;
 import com.impactupgrade.nucleus.service.segment.CrmService;
 import com.impactupgrade.nucleus.service.segment.EmailPlatformService;
@@ -101,6 +102,7 @@ public void setOtherContext(MultivaluedMap<String, String> otherContext) {
   public DonationService donationService() { return new DonationService(this); }
   public ContactService contactService() { return new ContactService(this); }
   public MessagingService messagingService() { return new MessagingService(this); }
+  public WebhookRequestService webhookRequestService() { return new WebhookRequestService(this); }
 
   // segment services
 

src/main/java/com/impactupgrade/nucleus/model/WebhookRequest.java

@@ -0,0 +1,27 @@
+package com.impactupgrade.nucleus.model;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.Id;
+import javax.persistence.Table;
+import java.util.Date;
+
+@Entity
+@Table(name = "webhook_request", schema = "public")
+public class WebhookRequest {
+    @Id
+    public String id;
+    @Column(name = "payload_type", nullable = false)
+    public String payloadType;
+    @Column(name = "payload", nullable = false)
+    public String payload;
+    @Column(name = "error_message")
+    public String errorMessage;
+    @Column(name = "attempt_count")
+    public int attemptCount;
+    @Column(name = "first_attempt_time", nullable = false)
+    public Date firstAttemptTime;
+    @Column(name = "last_attempt_time")
+    public Date lastAttemptTime;
+
+}

src/main/java/com/impactupgrade/nucleus/service/logic/WebhookRequestService.java

@@ -0,0 +1,89 @@
+package com.impactupgrade.nucleus.service.logic;
+
+import com.google.common.base.Strings;
+import com.impactupgrade.nucleus.dao.WebhookRequestDao;
+import com.impactupgrade.nucleus.environment.Environment;
+import com.impactupgrade.nucleus.model.WebhookRequest;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.json.JSONObject;
+
+import java.util.Date;
+import java.util.Objects;
+import java.util.Optional;
+
+public class WebhookRequestService {
+
+    private static final Logger log = LogManager.getLogger(WebhookRequestService.class);
+
+    private Environment env;
+    private WebhookRequestDao webhookRequestDao;
+
+    public WebhookRequestService(Environment env) {
+        this.env = env;
+        this.webhookRequestDao = new WebhookRequestDao();
+    }
+
+    public void persist(String id, String payloadType, JSONObject jsonObject) {
+        if (Strings.isNullOrEmpty(id) || Strings.isNullOrEmpty(payloadType) || Objects.isNull(jsonObject)) {
+            return;
+        }
+        try {
+            upsert(newWebhookRequest(id, payloadType, jsonObject));
+        } catch (Exception e) {
+            log.error("Failed to persist webhook request!", e);
+        }
+    }
+
+    public void updateWithErrorMessage(String id, String errorMessage) {
+        if (Strings.isNullOrEmpty(id)) {
+            return;
+        }
+        try {
+            Optional<WebhookRequest> existingRecord = webhookRequestDao.get(id);
+            if (existingRecord.isPresent()) {
+                WebhookRequest webhookRequest = existingRecord.get();
+                webhookRequest.errorMessage = errorMessage;
+                webhookRequestDao.update(webhookRequest);
+            }
+        } catch (Exception e) {
+            log.error("Failed to update webhook request!", e);
+        }
+    }
+
+    public void delete(String id) {
+        if (Strings.isNullOrEmpty(id)) {
+            return;
+        }
+        try {
+            webhookRequestDao.delete(id);
+        } catch (Exception e) {
+            log.error("Failed to delete failed request!", e);
+        }
+    }
+
+    // Utils
+    private WebhookRequest newWebhookRequest(String id, String payloadType, JSONObject payload) {
+        WebhookRequest webhookRequest = new WebhookRequest();
+        webhookRequest.id = id;
+        webhookRequest.payloadType = payloadType;
+        webhookRequest.payload = payload.toString();
+        webhookRequest.attemptCount = 1;
+        webhookRequest.firstAttemptTime = new Date();
+        webhookRequest.lastAttemptTime = new Date();
+        return webhookRequest;
+    }
+
+    private void upsert(WebhookRequest webhookRequest) {
+        Optional<WebhookRequest> existingRecord = webhookRequestDao.get(webhookRequest.id);
+        if (existingRecord.isEmpty()) {
+            webhookRequestDao.create(webhookRequest);
+        } else {
+            int attempts = existingRecord.get().attemptCount + 1;
+            webhookRequest.attemptCount = attempts;
+            webhookRequest.lastAttemptTime =new Date();
+            webhookRequestDao.update(webhookRequest);
+        }
+    }
+
+}