more consistent workflow

services/bondster-bco-import/actor/token.go

@@ -25,13 +25,13 @@ import (
 func NilToken(s *System, id string) system.ReceiverFunction {
 	return func(context system.Context) system.ReceiverFunction {
 		context.Self.Tell(context.Data, context.Receiver, context.Sender)
-		if persistence.LoadToken(s.EncryptedStorage, id) == nil {
+		_, err := persistence.LoadToken(s.EncryptedStorage, id)
+		if err != nil {
 			log.Debug().Msgf("token %s Nil -> NonExist", id)
 			return NonExistToken(s, id)
-		} else {
-			log.Debug().Msgf("token %s Nil -> Exist", id)
-			return ExistToken(s, id)
 		}
+		log.Debug().Msgf("token %s Nil -> Exist", id)
+		return ExistToken(s, id)
 	}
 }
 
@@ -46,9 +46,10 @@ func NonExistToken(s *System, id string) system.ReceiverFunction {
 			return NonExistToken(s, id)
 
 		case CreateToken:
-			if persistence.CreateToken(s.EncryptedStorage, id, msg.Username, msg.Password) == nil {
+			err := persistence.CreateToken(s.EncryptedStorage, id, msg.Username, msg.Password)			
+			if err != nil {
 				s.SendMessage(FatalError, context.Sender, context.Receiver)
-				log.Debug().Msgf("token %s (NonExist CreateToken) Error", id)
+				log.Debug().Err(err).Msgf("token %s (NonExist CreateToken) Error", id)
 				return NonExistToken(s, id)
 			}
 
@@ -107,8 +108,8 @@ func ExistToken(s *System, id string) system.ReceiverFunction {
 					context.Self.Tell(SynchornizationDone{}, context.Receiver, context.Receiver)
 				}()
 
-				token := persistence.LoadToken(s.EncryptedStorage, id)
-				if token == nil {
+				token, err := persistence.LoadToken(s.EncryptedStorage, id)
+				if err != nil {
 					return
 				}
 				workflow := integration.NewWorkflow(

services/bondster-bco-import/go.mod

@@ -5,7 +5,7 @@ go 1.17
 require (
 	github.com/DataDog/datadog-go v4.8.3+incompatible
 	github.com/jancajthaml-openbank/actor-system v1.5.4
-	github.com/jancajthaml-openbank/local-fs v1.2.4
+	github.com/jancajthaml-openbank/local-fs v1.2.6
 	github.com/rs/zerolog v1.26.1
 	github.com/stretchr/testify v1.7.0
 )

services/bondster-bco-import/go.sum

@@ -9,8 +9,8 @@ github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
 github.com/jancajthaml-openbank/actor-system v1.5.4 h1:1Eqn4A62uBPkEPW/0k9VxwSqdCdWKUWGlRQEQG9oPtU=
 github.com/jancajthaml-openbank/actor-system v1.5.4/go.mod h1:nIHD5as3cIrWQKj/YH1qJcEoGDJ9bLspb/rEfRwCv4U=
-github.com/jancajthaml-openbank/local-fs v1.2.4 h1:pXzHV0T+uIHj86uSdkr7ooxjWYrHhQF7Vt0QQ2F5u9M=
-github.com/jancajthaml-openbank/local-fs v1.2.4/go.mod h1:W19XcWPceSnj+ZE8UOZapMlinVIxq8xJsGogTfv80XM=
+github.com/jancajthaml-openbank/local-fs v1.2.6 h1:KF9a1A3VN7Qj1epurckNV2shoVS6MREyN9tcOXgvtjI=
+github.com/jancajthaml-openbank/local-fs v1.2.6/go.mod h1:W19XcWPceSnj+ZE8UOZapMlinVIxq8xJsGogTfv80XM=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/pebbe/zmq4 v1.2.7 h1:6EaX83hdFSRUEhgzSW1E/SPoTS3JeYZgYkBvwdcrA9A=
 github.com/pebbe/zmq4 v1.2.7/go.mod h1:nqnPueOapVhE2wItZ0uOErngczsJdLOGkebMxaO8r48=

services/bondster-bco-import/integration/workflow.go

@@ -158,14 +158,14 @@ func importTransactionsFromStatemets(
 
 	log.Info().Msgf("token %s creating transactions from statements for currency %s", token.ID, currency)
 
-	ids, err := plaintextStorage.ListDirectory(persistence.StatementPath(token.ID, currency), true)
+	ids, err := plaintextStorage.ListDirectory("token/" + token.ID + "/statements/" + currency, true)
 	if err != nil {
 		log.Warn().Err(err).Msgf("Unable to obtain transaction ids from storage for token %s currency %s", token.ID, currency)
 		return
 	}
 
 	for _, id := range ids {
-		exists, err := plaintextStorage.Exists(persistence.StatementDoneMarkPath(token.ID, currency, id))
+		exists, err := plaintextStorage.Exists("token/" + token.ID + "/statements/" + currency + "/" + id + "/done")
 		if err != nil {
 			log.Warn().Msgf("Unable to check if statement %s/%s/%s done exists", token.ID, currency, id)
 			continue
@@ -174,7 +174,7 @@ func importTransactionsFromStatemets(
 			continue
 		}
 
-		data, err := plaintextStorage.ReadFileFully(persistence.StatementDataPath(token.ID, currency, id))
+		data, err := plaintextStorage.ReadFileFully("token/" + token.ID + "/statements/" + currency + "/" + id + "/data")
 		if err != nil {
 			log.Warn().Err(err).Msgf("Unable to load statement %s/%s/%s", token.ID, currency, id)
 			continue
@@ -226,7 +226,7 @@ func importTransactionsFromStatemets(
 
 		metrics.TransactionImported(1)
 
-		err = plaintextStorage.TouchFile(persistence.StatementDoneMarkPath(token.ID, currency, id))
+		err = plaintextStorage.TouchFile("token/" + token.ID + "/statements/" + currency + "/" + id + "/done")
 		if err != nil {
 			log.Warn().Msgf("Unable to mark statement done for %s/%s/%s", token.ID, currency, id)
 			continue
@@ -263,7 +263,7 @@ func downloadStatements(
 			log.Warn().Msgf("Unable to marshal statement details of %s/%s/%s", tokenID, currency, transaction.IDTransfer)
 			continue
 		}
-		err = plaintextStorage.WriteFileExclusive(persistence.StatementDataPath(tokenID, currency, transaction.IDTransfer), data)
+		err = plaintextStorage.WriteFileExclusive("token/" + tokenID + "/statements/" + currency + "/" + transaction.IDTransfer + "/data", data)
 		if err != nil {
 			log.Warn().Err(err).Msgf("Unable to persist statement details of %s/%s/%s", tokenID, currency, transaction.IDTransfer)
 			continue
@@ -284,13 +284,13 @@ func yieldUnsynchronizedStatementIds(
 	go func() {
 		defer close(chnl)
 		buffer := make([]string, 0)
-		ids, err := plaintextStorage.ListDirectory(persistence.StatementPath(tokenID, currency), true)
+		ids, err := plaintextStorage.ListDirectory("token/" + tokenID + "/statements/" + currency, true)
 		if err != nil {
 			log.Warn().Msgf("Unable to obtain transaction ids from storage for token %s currency %s", tokenID, currency)
 			return
 		}
 		for _, id := range ids {
-			exists, err := plaintextStorage.Exists(persistence.StatementDataPath(tokenID, currency, id))
+			exists, err := plaintextStorage.Exists("token/" + tokenID + "/statements/" + currency + "/" + id + "/data")
 			if err != nil {
 				log.Warn().Msgf("Unable to check if statement %s/%s/%s data exists", tokenID, currency, id)
 				continue

services/bondster-bco-import/persistence/path.go → services/bondster-bco-import/persistence/package.go

@@ -14,14 +14,6 @@
 
 package persistence
 
-func StatementPath(token string, currency string) string {
-	return "token/" + token + "/statements/" + currency
-}
+import "github.com/jancajthaml-openbank/bondster-bco-import/support/logging"
 
-func StatementDataPath(token string, currency string, transfer string) string {
-	return "token/" + token + "/statements/" + currency + "/" + transfer + "/data"
-}
-
-func StatementDoneMarkPath(token string, currency string, transfer string) string {
-	return "token/" + token + "/statements/" + currency + "/" + transfer + "/done"
-}
+var log = logging.New("persistence")

services/bondster-bco-import/persistence/token.go

@@ -15,6 +15,8 @@
 package persistence
 
 import (
+	"fmt"
+
 	localfs "github.com/jancajthaml-openbank/local-fs"
 
 	"github.com/jancajthaml-openbank/bondster-bco-import/model"
@@ -44,14 +46,18 @@ func LoadTokens(storage localfs.Storage) ([]model.Token, error) {
 }
 
 // LoadToken rehydrates token entity state from storage
-func LoadToken(storage localfs.Storage, id string) *model.Token {
+func LoadToken(storage localfs.Storage, id string) (*model.Token, error) {
 	result := new(model.Token)
 	result.ID = id
-	return HydrateToken(storage, result)
+	err := HydrateToken(storage, result)
+	if err != nil {
+		return nil, err
+	}
+	return result, nil
 }
 
 // CreateToken persist token entity state to storage
-func CreateToken(storage localfs.Storage, id string, username string, password string) *model.Token {
+func CreateToken(storage localfs.Storage, id string, username string, password string) error {
 	token := model.NewToken(id)
 	token.Username = username
 	token.Password = password
@@ -60,52 +66,52 @@ func CreateToken(storage localfs.Storage, id string, username string, password s
 
 // DeleteToken deletes existing token entity
 func DeleteToken(storage localfs.Storage, id string) bool {
-	path := "token/" + id + "/value"
-	return storage.DeleteFile(path) == nil
+	return storage.Delete("token/" + id + "/value") == nil
 }
 
 // PersistToken persist new token entity to storage
-func PersistToken(storage localfs.Storage, entity *model.Token) *model.Token {
+func PersistToken(storage localfs.Storage, entity *model.Token) error {
 	if entity == nil {
-		return nil
+		return fmt.Errorf("nil reference")
 	}
-	path := "token/" + entity.ID + "/value"
 	data, err := entity.Serialize()
 	if err != nil {
-		return nil
+		return err
 	}
-	if storage.WriteFileExclusive(path, data) != nil {
-		return nil
-	}
-	return entity
+	return storage.WriteFileExclusive("token/" + entity.ID + "/value", data)
 }
 
 // HydrateToken hydrate existing token from storage
-func HydrateToken(storage localfs.Storage, entity *model.Token) *model.Token {
+func HydrateToken(storage localfs.Storage, entity *model.Token) error {
 	if entity == nil {
-		return nil
+		return fmt.Errorf("nil reference")
 	}
-	path := "token/" + entity.ID + "/value"
-	data, err := storage.ReadFileFully(path)
-	if err != nil {
-		return nil
+	ok, err := storage.Exists("token/" + entity.ID + "/value")
+	if !ok || err != nil {
+		err = storage.Delete("token/" + entity.ID)
+		if err != nil {
+			log.Warn().Err(err).Msgf("Unable to clean leftover files of no longer existing token %s", entity.ID)
+		} else {
+			log.Info().Msgf("Cleaned files of no longer existing token %s", entity.ID)
+		}
+		return fmt.Errorf("does not exists")
 	}
-	err = entity.Deserialize(data)
+
+	data, err := storage.ReadFileFully("token/" + entity.ID + "/value")
 	if err != nil {
-		return nil
+		return err
 	}
-	return entity
+	return entity.Deserialize(data)
 }
 
 // UpdateToken updates data of existing token to storage
 func UpdateToken(storage localfs.Storage, entity *model.Token) bool {
 	if entity == nil {
 		return false
 	}
-	path := "token/" + entity.ID + "/value"
 	data, err := entity.Serialize()
 	if err != nil {
 		return false
 	}
-	return storage.WriteFile(path, data) == nil
+	return storage.WriteFile("token/" + entity.ID + "/value", data) == nil
 }

services/bondster-bco-rest/go.mod

@@ -5,7 +5,7 @@ go 1.17
 require (
 	github.com/coreos/go-systemd/v22 v22.3.2
 	github.com/jancajthaml-openbank/actor-system v1.5.4
-	github.com/jancajthaml-openbank/local-fs v1.2.4
+	github.com/jancajthaml-openbank/local-fs v1.2.6
 	github.com/labstack/echo/v4 v4.6.1
 	github.com/rs/xid v1.3.0
 	github.com/rs/zerolog v1.26.1

services/bondster-bco-rest/go.sum

@@ -7,8 +7,8 @@ github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5x
 github.com/golang-jwt/jwt v3.2.2+incompatible/go.mod h1:8pz2t5EyA70fFQQSrl6XZXzqecmYZeUEB8OUGHkxJ+I=
 github.com/jancajthaml-openbank/actor-system v1.5.4 h1:1Eqn4A62uBPkEPW/0k9VxwSqdCdWKUWGlRQEQG9oPtU=
 github.com/jancajthaml-openbank/actor-system v1.5.4/go.mod h1:nIHD5as3cIrWQKj/YH1qJcEoGDJ9bLspb/rEfRwCv4U=
-github.com/jancajthaml-openbank/local-fs v1.2.4 h1:pXzHV0T+uIHj86uSdkr7ooxjWYrHhQF7Vt0QQ2F5u9M=
-github.com/jancajthaml-openbank/local-fs v1.2.4/go.mod h1:W19XcWPceSnj+ZE8UOZapMlinVIxq8xJsGogTfv80XM=
+github.com/jancajthaml-openbank/local-fs v1.2.6 h1:KF9a1A3VN7Qj1epurckNV2shoVS6MREyN9tcOXgvtjI=
+github.com/jancajthaml-openbank/local-fs v1.2.6/go.mod h1:W19XcWPceSnj+ZE8UOZapMlinVIxq8xJsGogTfv80XM=
 github.com/labstack/echo/v4 v4.6.1 h1:OMVsrnNFzYlGSdaiYGHbgWQnr+JM7NG+B9suCPie14M=
 github.com/labstack/echo/v4 v4.6.1/go.mod h1:RnjgMWNDB9g/HucVWhQYNQP9PvbYf6adqftqryo7s9k=
 github.com/labstack/gommon v0.3.0 h1:JEeO0bvc78PKdyHxloTKiF8BD5iGrH8T6MSeGvSgob0=