Checks for authorization validity when filtering users

-  Attempt to fix #69.

Author: yatharthranjan

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/route/FitbitPollingRoute.java

@@ -312,7 +312,7 @@ protected Duration getLookbackTime() {
    */
   protected Instant nextPoll(User user) {
     Instant offset = getOffset(user);
-    if (offset.isAfter(user.getEndDate())) {
+    if (offset.isAfter(user.getEndDate().minus(getEndDateThreshold()))) {
       return nearFuture();
     } else {
       Instant nextPoll = lastPollPerUser.getOrDefault(user.getId(), MIN_INSTANT)
@@ -321,6 +321,10 @@ protected Instant nextPoll(User user) {
     }
   }
 
+  private TemporalAmount getEndDateThreshold() {
+    return Duration.ofHours(1);
+  }
+
   /**
    * Generate one date per day, using UTC time zone. The first date will have the time from the
    * given startDate. Following time stamps will start at 00:00. This will not up to the date of

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/user/LocalUser.java

@@ -45,6 +45,9 @@ public class LocalUser implements User {
   @JsonProperty("oauth2")
   private OAuth2UserCredentials oauth2Credentials = new OAuth2UserCredentials();
 
+  @JsonProperty("authorized")
+  private Boolean isAuthorized;
+
   @JsonIgnore
   private SchemaAndValue observationKey;
 
@@ -95,6 +98,15 @@ public void setFitbitUserId(String id) {
     this.externalUserId = id;
   }
 
+  @Override
+  public boolean isAuthorized() {
+    if(isAuthorized == null) {
+      return !oauth2Credentials.isAccessTokenExpired()
+          || oauth2Credentials.hasRefreshToken();
+    }
+    return isAuthorized;
+  }
+
   @Override
   public String getVersion() {
     return version;
@@ -115,6 +127,7 @@ public LocalUser copy() {
     copy.endDate = endDate;
     copy.sourceId = sourceId;
     copy.oauth2Credentials = oauth2Credentials;
+    copy.isAuthorized = isAuthorized;
     return copy;
   }
 
@@ -133,6 +146,7 @@ public String toString() {
         + ", projectId='" + projectId + '\''
         + ", userId='" + userId + '\''
         + ", sourceId='" + sourceId + '\''
+        + ", isAuthorized='" + isAuthorized() + '\''
         + '}';
   }
 
@@ -152,7 +166,8 @@ public boolean equals(Object o) {
         && Objects.equals(userId, localUser.userId)
         && Objects.equals(sourceId, localUser.sourceId)
         && Objects.equals(startDate, localUser.startDate)
-        && Objects.equals(endDate, localUser.endDate);
+        && Objects.equals(endDate, localUser.endDate)
+        && Objects.equals(isAuthorized(), localUser.isAuthorized());
   }
 
   @Override

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/user/User.java

@@ -45,6 +45,8 @@ static SchemaAndValue computeObservationKey(AvroData avroData, User user) {
 
   String getSourceId();
 
+  boolean isAuthorized();
+
   default String getVersionedId() {
     String version = getVersion();
     if (version == null) {
@@ -60,6 +62,7 @@ default Boolean isComplete() {
     return getEndDate() != null
         && getStartDate() != null
         && getProjectId() != null
-        && getUserId() != null;
+        && getUserId() != null
+        && isAuthorized();
   }
 }

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/user/firebase/FirebaseUser.java

@@ -70,6 +70,12 @@ public String getSourceId() {
     return fitbitAuthDetails.getSourceId();
   }
 
+  @Override
+  public boolean isAuthorized() {
+    return !fitbitAuthDetails.getOauth2Credentials().isAccessTokenExpired()
+        || fitbitAuthDetails.getOauth2Credentials().hasRefreshToken();
+  }
+
   public FirebaseUserDetails getFirebaseUserDetails() {
     return firebaseUserDetails;
   }