make sure that we refresh the access token if it expires during the s…

…ync.

removing references to returned SourceCredential during init. Updated client interface with method to return source credentail pointer whenever required.

clients/factory/factory.go

@@ -18,7 +18,7 @@ import (
 	"net/http"
 )
 
-func GetSourceClient(env pkg.FastenLighthouseEnvType, sourceType pkg.SourceType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (models.SourceClient, *models.SourceCredential, error) {
+func GetSourceClient(env pkg.FastenLighthouseEnvType, sourceType pkg.SourceType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (models.SourceClient, error) {
 	switch sourceType {
 	case pkg.SourceTypeManual:
 		return manual.GetSourceClientManual(env, ctx, globalLogger, sourceCreds, testHttpClient...)
@@ -14498,6 +14498,6 @@ func GetSourceClient(env pkg.FastenLighthouseEnvType, sourceType pkg.SourceType,
 	case pkg.SourceTypeLogica:
 		return sandbox.GetSourceClientLogica(env, ctx, globalLogger, sourceCreds, testHttpClient...)
 	default:
-		return nil, nil, fmt.Errorf("could not find source type")
+		return nil, fmt.Errorf("could not find source type")
 	}
 }

clients/internal/base/base_client.go

@@ -14,6 +14,7 @@ import (
 	"net/url"
 	"os"
 	"strings"
+	"sync"
 	"time"
 )
 
@@ -28,6 +29,12 @@ type SourceClientBase struct {
 
 	UsCoreResources []string
 	FhirVersion     string
+
+	//When test mode is enabled, tokens will not be refreshed, and Http client provided will be used (usually go-vcr for playback)
+	testMode bool
+
+	//Mutex to prevent multiple token refreshes from happening at the same time
+	refreshMutex sync.Mutex
 }
 
 func (c *SourceClientBase) SyncAllBundle(db models.DatabaseRepository, bundleFile *os.File, bundleFhirVersion pkg.FhirVersion) (models.UpsertSummary, error) {
@@ -37,68 +44,12 @@ func (c *SourceClientBase) ExtractPatientId(bundleFile *os.File) (string, pkg.Fh
 	panic("SyncAllBundle functionality is not available on this client")
 }
 
-func NewBaseClient(env pkg.FastenLighthouseEnvType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (*SourceClientBase, *models.SourceCredential, error) {
-	var httpClient *http.Client
-	var updatedSource *models.SourceCredential
-	if len(testHttpClient) == 0 {
-		//check if we need to refresh the access token
-		//https://github.com/golang/oauth2/issues/84#issuecomment-520099526
-		// https://chromium.googlesource.com/external/github.com/golang/oauth2/+/8f816d62a2652f705144857bbbcc26f2c166af9e/oauth2.go#239
-		conf := &oauth2.Config{
-			ClientID:     sourceCreds.GetClientId(),
-			ClientSecret: "",
-			Endpoint: oauth2.Endpoint{
-				AuthURL:  sourceCreds.GetOauthAuthorizationEndpoint(),
-				TokenURL: sourceCreds.GetOauthTokenEndpoint(),
-			},
-			//RedirectURL:  "",
-			//Scopes:       nil,
-		}
-		token := &oauth2.Token{
-			TokenType:    "Bearer",
-			RefreshToken: sourceCreds.GetRefreshToken(),
-			AccessToken:  sourceCreds.GetAccessToken(),
-			Expiry:       time.Unix(sourceCreds.GetExpiresAt(), 0),
-		}
-		if token.Expiry.Before(time.Now()) { // expired so let's update it
-			log.Println("access token expired, refreshing...")
-			src := conf.TokenSource(ctx, token)
-			newToken, err := src.Token() // this actually goes and renews the tokens
-			if err != nil {
-				return nil, nil, err
-			}
-			if newToken.AccessToken != token.AccessToken {
-				token = newToken
-
-				// update the "source" credential with new data (which will need to be sent
-				sourceCreds.RefreshTokens(newToken.AccessToken, newToken.RefreshToken, newToken.Expiry.Unix())
-				updatedSource = &sourceCreds
-				//updatedSource.AccessToken = newToken.AccessToken
-				//updatedSource.ExpiresAt = newToken.Expiry.Unix()
-				//// Don't overwrite `RefreshToken` with an empty value
-				//// if this was a token refreshing request.
-				//if newToken.RefreshToken != "" {
-				//	updatedSource.RefreshToken = newToken.RefreshToken
-				//}
-
-			}
-		}
-
-		// OLD CODE
-		httpClient = oauth2.NewClient(ctx, oauth2.StaticTokenSource(token))
-
-	} else {
-		//Testing mode.
-		httpClient = testHttpClient[0]
-	}
-
-	httpClient.Timeout = 10 * time.Second
+func NewBaseClient(env pkg.FastenLighthouseEnvType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (*SourceClientBase, error) {
 
-	return &SourceClientBase{
+	client := &SourceClientBase{
 		FastenEnv:        env,
 		Context:          ctx,
 		Logger:           globalLogger,
-		OauthClient:      httpClient,
 		SourceCredential: sourceCreds,
 		Headers:          map[string]string{},
 
@@ -130,17 +81,102 @@ func NewBaseClient(env pkg.FastenLighthouseEnvType, ctx context.Context, globalL
 			// "ServiceRequest",
 			// "Specimen",
 		},
-	}, updatedSource, nil
+	}
+
+	if len(testHttpClient) > 0 {
+		//Testing mode.
+		client.testMode = true
+		client.OauthClient = testHttpClient[0]
+		client.OauthClient.Timeout = 10 * time.Second
+	}
+
+	err := client.RefreshAccessToken()
+	if err != nil {
+		return nil, err
+	}
+
+	return client, nil
 }
 
 func (c *SourceClientBase) GetUsCoreResources() []string {
 	return c.UsCoreResources
 }
 
+func (c *SourceClientBase) GetSourceCredential() models.SourceCredential {
+	return c.SourceCredential
+}
+
+func (c *SourceClientBase) RefreshAccessToken() error {
+	if c.testMode {
+		//if test mode is enabled, we cannot refresh the access token
+		return nil
+	}
+
+	c.refreshMutex.Lock()
+	defer c.refreshMutex.Unlock()
+
+	//check if we need to refresh the access token
+	//https://github.com/golang/oauth2/issues/84#issuecomment-520099526
+	// https://chromium.googlesource.com/external/github.com/golang/oauth2/+/8f816d62a2652f705144857bbbcc26f2c166af9e/oauth2.go#239
+	conf := &oauth2.Config{
+		ClientID:     c.SourceCredential.GetClientId(),
+		ClientSecret: "",
+		Endpoint: oauth2.Endpoint{
+			AuthURL:  c.SourceCredential.GetOauthAuthorizationEndpoint(),
+			TokenURL: c.SourceCredential.GetOauthTokenEndpoint(),
+		},
+		//RedirectURL:  "",
+		//Scopes:       nil,
+	}
+
+	token := &oauth2.Token{
+		TokenType:    "Bearer",
+		RefreshToken: c.SourceCredential.GetRefreshToken(),
+		AccessToken:  c.SourceCredential.GetAccessToken(),
+		Expiry:       time.Unix(c.SourceCredential.GetExpiresAt(), 0),
+	}
+
+	if token.Expiry.Before(time.Now().Add(5 * time.Second)) { // expired (or will expire in 5 seconds) so let's update it
+		log.Println("access token expired, refreshing...")
+
+		src := conf.TokenSource(c.Context, token)
+		newToken, err := src.Token() // this actually goes and renews the tokens
+		if err != nil {
+			return err
+		}
+		log.Printf("new token expiry: %s", newToken.Expiry.Format(time.RFC3339))
+		if newToken.AccessToken != token.AccessToken {
+			token = newToken
+
+			// update the "source" credential with new data (which will need to be sent
+			c.SourceCredential.RefreshTokens(newToken.AccessToken, newToken.RefreshToken, newToken.Expiry.Unix())
+			//updatedSource.AccessToken = newToken.AccessToken
+			//updatedSource.ExpiresAt = newToken.Expiry.Unix()
+			//// Don't overwrite `RefreshToken` with an empty value
+			//// if this was a token refreshing request.
+			//if newToken.RefreshToken != "" {
+			//	updatedSource.RefreshToken = newToken.RefreshToken
+			//}
+
+		}
+	}
+
+	c.OauthClient = oauth2.NewClient(c.Context, oauth2.StaticTokenSource(token))
+	c.OauthClient.Timeout = 10 * time.Second
+
+	return nil
+}
+
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 // HttpClient
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 func (c *SourceClientBase) GetRequest(resourceSubpathOrNext string, decodeModelPtr interface{}) error {
+	//check if we need to refresh the access token
+	err := c.RefreshAccessToken()
+	if err != nil {
+		return err
+	}
+
 	resourceUrl, err := url.Parse(resourceSubpathOrNext)
 	if err != nil {
 		return err

clients/internal/base/fhir401_client.go

@@ -18,12 +18,12 @@ type SourceClientFHIR401 struct {
 	*SourceClientBase
 }
 
-func GetSourceClientFHIR401(env pkg.FastenLighthouseEnvType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (*SourceClientFHIR401, *models.SourceCredential, error) {
-	baseClient, updatedSource, err := NewBaseClient(env, ctx, globalLogger, sourceCreds, testHttpClient...)
+func GetSourceClientFHIR401(env pkg.FastenLighthouseEnvType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (*SourceClientFHIR401, error) {
+	baseClient, err := NewBaseClient(env, ctx, globalLogger, sourceCreds, testHttpClient...)
 	baseClient.FhirVersion = "4.0.1"
 	return &SourceClientFHIR401{
 		SourceClientBase: baseClient,
-	}, updatedSource, err
+	}, err
 }
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

clients/internal/base/fhir401_client_test.go

@@ -27,7 +27,7 @@ func TestNewFHIR401Client(t *testing.T) {
 	})
 
 	//test
-	client, _, err := GetSourceClientFHIR401(pkg.FastenLighthouseEnvSandbox, context.Background(), testLogger, sc, &http.Client{})
+	client, err := GetSourceClientFHIR401(pkg.FastenLighthouseEnvSandbox, context.Background(), testLogger, sc, &http.Client{})
 
 	//assert
 	require.NoError(t, err)
@@ -46,7 +46,7 @@ func TestFHIR401Client_ProcessBundle(t *testing.T) {
 	testLogger := logrus.WithFields(logrus.Fields{
 		"type": "test",
 	})
-	client, _, err := GetSourceClientFHIR401(pkg.FastenLighthouseEnvSandbox, context.Background(), testLogger, sc, &http.Client{})
+	client, err := GetSourceClientFHIR401(pkg.FastenLighthouseEnvSandbox, context.Background(), testLogger, sc, &http.Client{})
 	require.NoError(t, err)
 
 	jsonBytes, err := ReadTestFixture("testdata/fixtures/401-R4/bundle/cigna_syntheticuser05-everything.json")

clients/internal/base/fhir430_client.go

@@ -16,12 +16,12 @@ type SourceClientFHIR430 struct {
 	*SourceClientBase
 }
 
-func GetSourceClientFHIR430(env pkg.FastenLighthouseEnvType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (*SourceClientFHIR430, *models.SourceCredential, error) {
-	baseClient, updatedSource, err := NewBaseClient(env, ctx, globalLogger, sourceCreds, testHttpClient...)
+func GetSourceClientFHIR430(env pkg.FastenLighthouseEnvType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (*SourceClientFHIR430, error) {
+	baseClient, err := NewBaseClient(env, ctx, globalLogger, sourceCreds, testHttpClient...)
 	baseClient.FhirVersion = "4.3.0"
 	return &SourceClientFHIR430{
 		baseClient,
-	}, updatedSource, err
+	}, err
 }
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

clients/internal/manual/client.go

@@ -27,6 +27,10 @@ type ManualClient struct {
 	SourceCredential models.SourceCredential
 }
 
+func (m ManualClient) GetSourceCredential() models.SourceCredential {
+	return m.SourceCredential
+}
+
 func (m ManualClient) GetResourceBundle(relativeResourcePath string) (interface{}, error) {
 	//TODO implement me
 	panic("implement me")
@@ -80,7 +84,7 @@ func (m ManualClient) SyncAllBundle(db models.DatabaseRepository, bundleFile *os
 		if err != nil {
 			return summary, fmt.Errorf("an error occurred while parsing 4.3.0 bundle: %w", err)
 		}
-		client, _, err := base.GetSourceClientFHIR430(m.FastenEnv, m.Context, m.Logger, m.SourceCredential, http.DefaultClient)
+		client, err := base.GetSourceClientFHIR430(m.FastenEnv, m.Context, m.Logger, m.SourceCredential, http.DefaultClient)
 		if err != nil {
 			return summary, fmt.Errorf("an error occurred while creating 4.3.0 client: %w", err)
 		}
@@ -102,7 +106,7 @@ func (m ManualClient) SyncAllBundle(db models.DatabaseRepository, bundleFile *os
 		if err != nil {
 			return summary, fmt.Errorf("an error occurred while parsing 4.0.1 bundle: %w", err)
 		}
-		client, _, err := base.GetSourceClientFHIR401(m.FastenEnv, m.Context, m.Logger, m.SourceCredential, http.DefaultClient)
+		client, err := base.GetSourceClientFHIR401(m.FastenEnv, m.Context, m.Logger, m.SourceCredential, http.DefaultClient)
 		if err != nil {
 			return summary, fmt.Errorf("an error occurred while creating 4.0.1 client: %w", err)
 		}
@@ -156,13 +160,13 @@ func (m ManualClient) ExtractPatientId(bundleFile *os.File) (string, pkg.FhirVer
 	}
 }
 
-func GetSourceClientManual(env pkg.FastenLighthouseEnvType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (models.SourceClient, *models.SourceCredential, error) {
+func GetSourceClientManual(env pkg.FastenLighthouseEnvType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (models.SourceClient, error) {
 	return &ManualClient{
 		FastenEnv:        env,
 		Context:          ctx,
 		Logger:           globalLogger,
 		SourceCredential: sourceCreds,
-	}, nil, nil
+	}, nil
 }
 
 //TODO: find a better, more generic way to do this.

clients/internal/platform/allscripts.go

@@ -27,12 +27,12 @@ so we need to swap the code for the access_token on the server
 */
 // https://fhir.fhirpoint.open.allscripts.com/fhirroute/open/CustProProdSand201SMART/metadata
 // https://developer.veradigm.com/Fhir/FHIR_Sandboxes#pehr
-func GetSourceClientAllscripts(env pkg.FastenLighthouseEnvType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (models.SourceClient, *models.SourceCredential, error) {
-	baseClient, updatedSourceCred, err := base.GetSourceClientFHIR401(env, ctx, globalLogger, sourceCreds, testHttpClient...)
+func GetSourceClientAllscripts(env pkg.FastenLighthouseEnvType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (models.SourceClient, error) {
+	baseClient, err := base.GetSourceClientFHIR401(env, ctx, globalLogger, sourceCreds, testHttpClient...)
 	// API requires the following headers for every request
 	baseClient.Headers["Accept"] = "application/json+fhir"
 
-	return SourceClientAllscripts{baseClient}, updatedSourceCred, err
+	return SourceClientAllscripts{baseClient}, err
 }
 
 // Operation-PatientEverything is not supported - https://build.fhir.org/operation-patient-everything.html

clients/internal/platform/careevolution.go

@@ -21,10 +21,10 @@ type SourceClientCareevolution struct {
 // https://fhir.careevolution.com/Master.Adapter1.WebClient/api/fhir-r4/.well-known/smart-configuration
 // https://fhir.careevolution.com/Master.Adapter1.WebClient/api/fhir-r4/metadata
 // https://fhir.careevolution.com/TestPatientAccounts.html
-func GetSourceClientCareevolution(env pkg.FastenLighthouseEnvType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (models.SourceClient, *models.SourceCredential, error) {
-	baseClient, updatedSourceCred, err := base.GetSourceClientFHIR401(env, ctx, globalLogger, sourceCreds, testHttpClient...)
+func GetSourceClientCareevolution(env pkg.FastenLighthouseEnvType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (models.SourceClient, error) {
+	baseClient, err := base.GetSourceClientFHIR401(env, ctx, globalLogger, sourceCreds, testHttpClient...)
 	// API requires the following headers for every request
 	baseClient.Headers["Accept"] = "application/json+fhir"
 
-	return SourceClientCareevolution{baseClient}, updatedSourceCred, err
+	return SourceClientCareevolution{baseClient}, err
 }

clients/internal/platform/cerner.go

@@ -25,12 +25,12 @@ http://fhir.cerner.com/millennium/r4/#authorization
 // https://fhir-ehr.cerner.com/r4/ec2458f2-1e24-41c8-b71b-0e701af7583d/.well-known/smart-configuration
 // https://fhir-myrecord.cerner.com/r4/ec2458f2-1e24-41c8-b71b-0e701af7583d/metadata
 // https://docs.google.com/document/d/10RnVyF1etl_17pyCyK96tyhUWRbrTyEcqpwzW-Z-Ybs/edit
-func GetSourceClientCerner(env pkg.FastenLighthouseEnvType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (models.SourceClient, *models.SourceCredential, error) {
-	baseClient, updatedSourceCred, err := base.GetSourceClientFHIR401(env, ctx, globalLogger, sourceCreds, testHttpClient...)
+func GetSourceClientCerner(env pkg.FastenLighthouseEnvType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (models.SourceClient, error) {
+	baseClient, err := base.GetSourceClientFHIR401(env, ctx, globalLogger, sourceCreds, testHttpClient...)
 	// API requires the following headers for every request
 	baseClient.Headers["Accept"] = "application/json+fhir"
 
-	return SourceClientCerner{baseClient}, updatedSourceCred, err
+	return SourceClientCerner{baseClient}, err
 }
 
 // Operation-PatientEverything is not supported - https://build.fhir.org/operation-patient-everything.html

clients/internal/platform/epic.go

@@ -21,12 +21,12 @@ type SourceClientEpic struct {
 // https://fhir.epic.com/interconnect-fhir-oauth/api/FHIR/R4/.well-known/smart-configuration
 // https://fhir.epic.com/interconnect-fhir-oauth/api/FHIR/R4/metadata
 // https://fhir.epic.com/Documentation?docId=testpatients
-func GetSourceClientEpic(env pkg.FastenLighthouseEnvType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (models.SourceClient, *models.SourceCredential, error) {
-	baseClient, updatedSourceCred, err := base.GetSourceClientFHIR401(env, ctx, globalLogger, sourceCreds, testHttpClient...)
+func GetSourceClientEpic(env pkg.FastenLighthouseEnvType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (models.SourceClient, error) {
+	baseClient, err := base.GetSourceClientFHIR401(env, ctx, globalLogger, sourceCreds, testHttpClient...)
 	// API requires the following headers for every request
 	baseClient.Headers["Accept"] = "application/json+fhir"
 
-	return SourceClientEpic{baseClient}, updatedSourceCred, err
+	return SourceClientEpic{baseClient}, err
 }
 
 // Operation-PatientEverything is not supported - https://build.fhir.org/operation-patient-everything.html

clients/internal/platform/meditech.go

@@ -21,10 +21,10 @@ type SourceClientMeditech struct {
 // https://greenfield-apis.meditech.com/v1/uscore/R4/.well-known/smart-configuration
 // https://greenfield-apis.meditech.com/v1/uscore/R4/metadata
 // https://fhir.meditech.com/explorer/authorization
-func GetSourceClientMeditech(env pkg.FastenLighthouseEnvType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (models.SourceClient, *models.SourceCredential, error) {
-	baseClient, updatedSourceCred, err := base.GetSourceClientFHIR401(env, ctx, globalLogger, sourceCreds, testHttpClient...)
+func GetSourceClientMeditech(env pkg.FastenLighthouseEnvType, ctx context.Context, globalLogger logrus.FieldLogger, sourceCreds models.SourceCredential, testHttpClient ...*http.Client) (models.SourceClient, error) {
+	baseClient, err := base.GetSourceClientFHIR401(env, ctx, globalLogger, sourceCreds, testHttpClient...)
 	// API requires the following headers for every request
 	baseClient.Headers["Accept"] = "application/json+fhir"
 
-	return SourceClientMeditech{baseClient}, updatedSourceCred, err
+	return SourceClientMeditech{baseClient}, err
 }