Adding tracking of successful updates.

This also adds some logic for how far back to retrieve transactions. If
it has been a while since there was a successful update then we will
retrieve data since the last successful update.

.gitlab-ci.yml

@@ -253,6 +253,7 @@ Dry - Staging:
     kubernetes:
       namespace: monetr-staging
   needs:
+    - "PostgreSQL Tests"
     - "Generate - Staging"
     - "REST API"
   stage: Dry Run
@@ -277,6 +278,7 @@ Dry - Acceptance:
     kubernetes:
       namespace: monetr-acceptance
   needs:
+    - "PostgreSQL Tests"
     - "Generate - Acceptance"
     - "REST API"
   stage: Dry Run

pkg/controller/plaid.go

@@ -17,10 +17,10 @@ import (
 )
 
 func (c *Controller) handlePlaidLinkEndpoints(p router.Party) {
-	p.Get("/token/new", c.newPlaidToken)
 	p.Put("/update/{linkId:uint64}", c.updatePlaidLink)
-	p.Post("/token/callback", c.plaidTokenCallback)
 	p.Post("/update/callback", c.updatePlaidTokenCallback)
+	p.Get("/token/new", c.newPlaidToken)
+	p.Post("/token/callback", c.plaidTokenCallback)
 	p.Get("/setup/wait/{linkId:uint64}", c.waitForPlaid)
 }
 
@@ -281,6 +281,18 @@ func (c *Controller) updatePlaidLink(ctx iris.Context) {
 	})
 }
 
+// Token Callback for Plaid Link
+// @Summary Updated Token Callback
+// @id updated-token-callback
+// @tags Plaid
+// @Description This is used when handling an update flow for a Plaid link. Rather than returning the public token to the normal callback endpoint, this one should be used instead. This one assumes the link already exists and handles it slightly differently than it would for a new link.
+// @Security ApiKeyAuth
+// @Produce json
+// @Accept json
+// @Param Request body swag.UpdatePlaidTokenCallbackRequest true "Update token callback request."
+// @Router /plaid/update/callback [post]
+// @Success 200 {object} swag.LinkResponse
+// @Failure 500 {object} ApiError Something went wrong on our end.
 func (c *Controller) updatePlaidTokenCallback(ctx iris.Context) {
 	var callbackRequest struct {
 		LinkId      uint64 `json:"linkId"`
@@ -339,7 +351,9 @@ func (c *Controller) updatePlaidTokenCallback(ctx iris.Context) {
 // @tags Plaid
 // @description Receives the public token after a user has authenticated their bank account to exchange with plaid.
 // @Security ApiKeyAuth
+// @Accept json
 // @Produce json
+// @Param Request body swag.NewPlaidTokenCallbackRequest true "New token callback request."
 // @Router /plaid/token/callback [post]
 // @Success 200 {object} swag.PlaidTokenCallbackResponse
 // @Failure 500 {object} ApiError Something went wrong on our end.

pkg/internal/migrations/schema/2021052700_LinkUpdatedDate.tx.down.sql

@@ -0,0 +1 @@
+ALTER TABLE "links" DROP COLUMN "last_successful_update";

pkg/internal/migrations/schema/2021052700_LinkUpdatedDate.tx.up.sql

@@ -0,0 +1 @@
+ALTER TABLE "links" ADD COLUMN "last_successful_update" TIMESTAMPTZ NULL;

pkg/jobs/pull_historical_transactions.go

@@ -233,6 +233,7 @@ func (j *jobManagerBase) pullHistoricalTransactions(job *work.Job) error {
 			}
 		}
 
-		return nil
+		link.LastSuccessfulUpdate = myownsanity.TimeP(time.Now().UTC())
+		return repo.UpdateLink(link)
 	})
 }

pkg/jobs/pull_initial_transactions.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"github.com/getsentry/sentry-go"
 	"github.com/gocraft/work"
+	"github.com/monetrapp/rest-api/pkg/internal/myownsanity"
 	"github.com/monetrapp/rest-api/pkg/models"
 	"github.com/monetrapp/rest-api/pkg/repository"
 	"github.com/monetrapp/rest-api/pkg/util"
@@ -171,6 +172,7 @@ func (j *jobManagerBase) pullInitialTransactions(job *work.Job) error {
 			return nil // Not good enough of a reason to fail.
 		}
 
-		return nil
+		link.LastSuccessfulUpdate = myownsanity.TimeP(time.Now().UTC())
+		return repo.UpdateLink(link)
 	})
 }

pkg/jobs/pull_latest_transactions.go

@@ -12,6 +12,7 @@ import (
 	"github.com/plaid/plaid-go/plaid"
 	"github.com/sirupsen/logrus"
 	"strconv"
+	"strings"
 	"time"
 )
 
@@ -91,7 +92,7 @@ func (j *jobManagerBase) pullLatestTransactions(job *work.Job) error {
 	span := sentry.StartSpan(ctx, "Job", sentry.TransactionName("Pull Latest Transactions"))
 	defer span.Finish()
 
-	start := time.Now()
+	startTime := time.Now()
 	log := j.getLogForJob(job)
 	log.Infof("pulling account balances")
 
@@ -103,7 +104,7 @@ func (j *jobManagerBase) pullLatestTransactions(job *work.Job) error {
 
 	defer func() {
 		if j.stats != nil {
-			j.stats.JobFinished(PullAccountBalances, accountId, start)
+			j.stats.JobFinished(PullAccountBalances, accountId, startTime)
 		}
 	}()
 
@@ -152,25 +153,44 @@ func (j *jobManagerBase) pullLatestTransactions(job *work.Job) error {
 
 		log.Debugf("retrieving transactions for %d bank account(s)", len(itemBankAccountIds))
 
+		// Request the last 7 days worth of transactions for update.
+		start := time.Now().Add(-7 * 24 * time.Hour)
+		if link.LastSuccessfulUpdate == nil {
+			// But if there has not been a last successful update set yet, then request the last 30 days to handle this
+			// update.
+			start = time.Now().Add(-30 * 24 * time.Hour)
+		} else if start.After(*link.LastSuccessfulUpdate) {
+			// If we haven't seen an update in longer than 7 days, then use the last successful update date instead.
+			start = *link.LastSuccessfulUpdate
+		}
+		end := time.Now()
+
 		transactions, err := j.plaidClient.GetAllTransactions(
 			span.Context(),
 			link.PlaidLink.AccessToken,
-			time.Now().Add(-7*24*time.Hour),
-			time.Now(),
+			start,
+			end,
 			itemBankAccountIds,
 		)
 		if err != nil {
 			log.WithError(err).Error("failed to retrieve transactions from plaid")
 			switch plaidErr := errors.Cause(err).(type) {
 			case plaid.Error:
-				switch plaidErr.ErrorType {
-				case "ITEM_ERROR":
-					link.LinkStatus = models.LinkStatusError
-					link.ErrorCode = &plaidErr.ErrorCode
-					if updateErr := repo.UpdateLink(link); updateErr != nil {
-						log.WithError(updateErr).Error("failed to update link to be an error state")
-					}
+				link.LinkStatus = models.LinkStatusError
+				link.ErrorCode = myownsanity.StringP(strings.Join([]string{
+					plaidErr.ErrorType,
+					plaidErr.ErrorCode,
+				}, "."))
+				if updateErr := repo.UpdateLink(link); updateErr != nil {
+					log.WithError(updateErr).Error("failed to update link to be an error state")
+					return err
 				}
+
+				// Don't return an error here, we set the link to an error state and we don't want to have retry logic
+				// for this right now.
+				return nil
+			default:
+				log.Warnf("unknown error type from Plaid client: %T", plaidErr)
 			}
 
 			return errors.Wrap(err, "failed to retrieve transactions from plaid")
@@ -296,6 +316,7 @@ func (j *jobManagerBase) pullLatestTransactions(job *work.Job) error {
 			}
 		}
 
-		return nil
+		link.LastSuccessfulUpdate = myownsanity.TimeP(time.Now().UTC())
+		return repo.UpdateLink(link)
 	})
 }

pkg/jobs/transactions_removed.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"github.com/getsentry/sentry-go"
 	"github.com/gocraft/work"
+	"github.com/monetrapp/rest-api/pkg/internal/myownsanity"
 	"github.com/monetrapp/rest-api/pkg/repository"
 	"github.com/pkg/errors"
 	"strconv"
@@ -125,6 +126,7 @@ func (j *jobManagerBase) removeTransactions(job *work.Job) error {
 
 		log.Debugf("successfully removed %d transaction(s)", len(transactions))
 
-		return nil
+		link.LastSuccessfulUpdate = myownsanity.TimeP(time.Now().UTC())
+		return repo.UpdateLink(link)
 	})
 }

pkg/models/link.go

@@ -32,6 +32,7 @@ type Link struct {
 	UpdatedAt             time.Time    `json:"updatedAt" pg:"updated_at,notnull"`
 	UpdatedByUserId       *uint64      `json:"updatedByUserId" pg:"updated_by_user_id,on_delete:SET NULL"`
 	UpdatedByUser         *User        `json:"updatedByUser,omitempty" pg:"rel:has-one,fk:updated_by_user_id"`
+	LastSuccessfulUpdate  *time.Time   `json:"lastSuccessfulUpdate" pg:"last_successful_update"`
 
 	BankAccounts []BankAccount `json:"-" pg:"rel:has-many"`
 }

pkg/repository/transaction.go

@@ -39,6 +39,10 @@ func (r *repositoryBase) GetPendingTransactionsForBankAccount(bankAccountId uint
 }
 
 func (r *repositoryBase) GetTransactionsByPlaidId(linkId uint64, plaidTransactionIds []string) (map[string]models.Transaction, error) {
+	if len(plaidTransactionIds) == 0 {
+		return map[string]models.Transaction{}, nil
+	}
+
 	var items []models.Transaction
 	err := r.txn.Model(&items).
 		Join(`INNER JOIN "bank_accounts" AS "bank_account"`).

pkg/swag/links.go

@@ -55,6 +55,9 @@ type LinkResponse struct {
 	UpdatedByUserId *uint64 `json:"updatedByUserId" example:"89547" extensions:"x-nullable"`
 	// The last time this link was updated. Currently this field is not really maintained, eventually this timestamp
 	// will indicate the last time a sync occurred between monetr and Plaid. Manual links don't change this field at
-	// all.
+	// all. **OLD**
 	UpdatedAt time.Time `json:"updatedAt" example:"2021-05-21T05:24:12.958309Z"`
+	// The last time transactions were successfully retrieved for this link. This date does not indicate the most recent
+	// transaction retrieved, simply the most recent attempt to retrieve transactions that was successful.
+	LastSuccessfulUpdate *time.Time `json:"lastSuccessfulUpdate" example:"2021-05-21T05:24:12.958309Z" extensions:"x-nullable"`
 }

pkg/swag/plaid.go

@@ -7,11 +7,23 @@ type PlaidNewLinkTokenResponse struct {
 }
 
 type PlaidTokenCallbackResponse struct {
-	Success bool    `json:"success"`
+	Success bool `json:"success"`
 	// LinkId will always be included in a successful response. It can be used when webhooks are enabled to wait for the
 	// initial transactions to be retrieved.
-	LinkId  uint64  `json:"linkId"`
+	LinkId uint64 `json:"linkId"`
 	// If webhooks are not enabled then a job Id is returned with the response. This job Id can also be used to check
 	// for initial transactions being retrieved.
-	JobId   *string `json:"jobId" extensions:"x-nullable"`
+	JobId *string `json:"jobId" extensions:"x-nullable"`
+}
+
+type UpdatePlaidTokenCallbackRequest struct {
+	LinkId      uint64 `json:"linkId"`
+	PublicToken string `json:"publicToken"`
+}
+
+type NewPlaidTokenCallbackRequest struct {
+	PublicToken     string   `json:"publicToken"`
+	InstitutionId   string   `json:"institutionId" example:"ins_117212"`
+	InstitutionName string   `json:"institutionName" example:"Navy Federal Credit Union"`
+	AccountIds      []string `json:"accountIds" example:"KEdQjMo39lFwXKqKLlqEt6R3AgBWW1C6l8vDn,r3DVlexNymfJkgZgonZeSQ4n5Koqqjtyrwvkp"`
 }