ChequeHandler, Scheduler and cmAccounts refactoring (#60)

config/config.go

@@ -11,9 +11,9 @@ import (
 	"github.com/ethereum/go-ethereum/crypto"
 )
 
-//
 // ******* Parsed config *******
 //
+//
 
 type Config struct {
 	DeveloperMode bool
@@ -36,7 +36,7 @@ type Config struct {
 	RPCServer     RPCServerConfig
 	PartnerPlugin PartnerPluginConfig
 	Tracing       TracingConfig
-	DB            DBConfig
+	DB            SQLiteDBConfig
 	Matrix        MatrixConfig
 }
 
@@ -60,15 +60,15 @@ type MatrixConfig struct {
 	Store   string
 }
 
-//
+type SQLiteDBConfig struct {
+	Common        UnparsedSQLiteDBConfig
+	Scheduler     UnparsedSQLiteDBConfig
+	ChequeHandler UnparsedSQLiteDBConfig
+}
+
 // ******* Common *******
 //
-
-type DBConfig struct {
-	DBPath         string `mapstructure:"path"`
-	DBName         string `mapstructure:"name"`
-	MigrationsPath string `mapstructure:"migrations_path"`
-}
+//
 
 type RPCServerConfig struct {
 	Enabled        bool   `mapstructure:"enabled"`
@@ -78,9 +78,9 @@ type RPCServerConfig struct {
 	ServerKeyFile  string `mapstructure:"key_file"`
 }
 
-//
 // ******* Unparsed config *******
 //
+//
 
 type UnparsedConfig struct {
 	DeveloperMode bool `mapstructure:"developer_mode"`
@@ -104,8 +104,8 @@ type UnparsedConfig struct {
 	Tracing       UnparsedTracingConfig       `mapstructure:"tracing"`
 	Matrix        UnparsedMatrixConfig        `mapstructure:"matrix"`
 
-	RPCServer RPCServerConfig `mapstructure:"rpc_server"`
-	DB        DBConfig        `mapstructure:"db"`
+	RPCServer RPCServerConfig        `mapstructure:"rpc_server"`
+	DB        UnparsedSQLiteDBConfig `mapstructure:"db"`
 }
 
 type UnparsedTracingConfig struct {
@@ -128,9 +128,14 @@ type UnparsedMatrixConfig struct {
 	Store string `mapstructure:"store"`
 }
 
+type UnparsedSQLiteDBConfig struct {
+	DBPath         string `mapstructure:"path"`
+	MigrationsPath string `mapstructure:"migrations_path"`
+}
+
 func (cfg *Config) unparse() *UnparsedConfig {
 	return &UnparsedConfig{
-		DB:        cfg.DB,
+		DB:        cfg.DB.Common,
 		RPCServer: cfg.RPCServer,
 		Tracing: UnparsedTracingConfig{
 			Enabled:  cfg.Tracing.Enabled,

config/config_reader.go

@@ -21,7 +21,7 @@ const envPrefix = "CMB"
 var (
 	_ Reader = (*reader)(nil)
 
-	errInvalidConfig = errors.New("invalid config")
+	errInvalidRawConfig = errors.New("invalid raw config")
 )
 
 type Reader interface {
@@ -82,7 +82,7 @@ func (cr *reader) ReadConfig() (*Config, error) {
 
 	parsedCfg, err := cr.parseConfig(cfg)
 	if err != nil {
-		return nil, fmt.Errorf("%w: %w", errInvalidConfig, err)
+		return nil, fmt.Errorf("%w: %w", errInvalidRawConfig, err)
 	}
 
 	return parsedCfg, nil
@@ -120,7 +120,17 @@ func (cr *reader) parseConfig(cfg *UnparsedConfig) (*Config, error) {
 	}
 
 	return &Config{
-		DB:        cfg.DB,
+		DB: SQLiteDBConfig{
+			Common: cfg.DB,
+			Scheduler: UnparsedSQLiteDBConfig{
+				DBPath:         cfg.DB.DBPath + "/scheduler",
+				MigrationsPath: cfg.DB.MigrationsPath + "/scheduler",
+			},
+			ChequeHandler: UnparsedSQLiteDBConfig{
+				DBPath:         cfg.DB.DBPath + "/cheque_handler",
+				MigrationsPath: cfg.DB.MigrationsPath + "/cheque_handler",
+			},
+		},
 		RPCServer: cfg.RPCServer,
 		Tracing: TracingConfig{
 			Enabled:  cfg.Tracing.Enabled,

config/config_reader_test.go

@@ -35,7 +35,7 @@ func TestReadConfig(t *testing.T) {
 				cr.viper.Set(flagKeyConfig, nonExistingConfigPath)
 			},
 			flags:       Flags(),
-			expectedErr: errInvalidConfig, // empty bot key
+			expectedErr: errInvalidRawConfig, // empty bot key
 		},
 		"from file": {
 			prepare: func(_ *testing.T, cr *reader) {

config/flags.go

@@ -29,8 +29,7 @@ func Flags() *pflag.FlagSet {
 	flags.Int64("response_timeout", 3000, "The messenger timeout (in milliseconds).")
 
 	// DB config flags
-	flags.String("db.name", "camino_messenger_bot", "Database name.")
-	flags.String("db.path", "cmb.db", "Path to database.")
+	flags.String("db.path", "cmb-db", "Path to database dir.")
 	flags.String("db.migrations_path", "file://./migrations", "Path to migration scripts.")
 
 	// Tracing config flags

config/test_config.yaml

@@ -6,8 +6,7 @@ cheque_expiration_time: 18144000
 cm_account_address: 0xe55E387F5474a012D1b048155E25ea78C7DBfBBC
 db:
     migrations_path: file://./migrations
-    name: camino_messenger_bot
-    path: supplier-bot.db
+    path: supplier-bot-db
 developer_mode: true
 matrix:
     host: messenger.chain4travel.com

examples/config/camino-messenger-bot-distributor-camino.yaml

@@ -51,11 +51,8 @@ db:
     # Path to migrations dir with sql up/down scripts. Schema is mandatory.
     migrations_path: file://./migrations
 
-    # Database name.
-    name: camino_messenger_bot
-
-    # Path to database file.
-    path: distributor-bot.db
+    # Path to database dir.
+    path: distributor-bot-db
 
 
 

examples/config/camino-messenger-bot-distributor-columbus.yaml

@@ -51,11 +51,8 @@ db:
     # Path to migrations dir with sql up/down scripts. Schema is mandatory.
     migrations_path: file://./migrations
 
-    # Database name.
-    name: camino_messenger_bot
-
-    # Path to database file.
-    path: distributor-bot.db
+    # Path to database dir.
+    path: distributor-bot-db
 
 
 

examples/config/camino-messenger-bot-supplier-camino.yaml

@@ -51,11 +51,8 @@ db:
     # Path to migrations dir with sql up/down scripts. Schema is mandatory.
     migrations_path: file://./migrations
 
-    # Database name.
-    name: camino_messenger_bot
-
-    # Path to database file.
-    path: supplier-bot.db
+    # Path to database dir.
+    path: supplier-bot-db
 
 
 

examples/config/camino-messenger-bot-supplier-columbus.yaml

@@ -51,13 +51,8 @@ db:
     # Path to migrations dir with sql up/down scripts. Schema is mandatory.
     migrations_path: file://./migrations
 
-    # Database name.
-    name: camino_messenger_bot
-
-    # Path to database file.
-    path: supplier-bot.db
-
-
+    # Path to database dir.
+    path: supplier-bot-db
 
 ### Matrix
 matrix:

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/hashicorp/golang-lru/v2 v2.0.7
 	github.com/jmoiron/sqlx v1.4.0
+	github.com/jonboulle/clockwork v0.4.0
 	github.com/klauspost/compress v1.17.10
 	github.com/mattn/go-sqlite3 v1.14.23
 	github.com/spf13/cobra v1.8.1

go.sum

@@ -160,6 +160,8 @@ github.com/jackpal/go-nat-pmp v1.0.2 h1:KzKSgb7qkJvOUTqYl9/Hg/me3pWgBmERKrTGD7Bd
 github.com/jackpal/go-nat-pmp v1.0.2/go.mod h1:QPH045xvCAeXUZOxsnwmrtiCoxIr9eob+4orBN1SBKc=
 github.com/jmoiron/sqlx v1.4.0 h1:1PLqN7S1UYp5t4SrVVnt4nUVNemrDAtxlulVe+Qgm3o=
 github.com/jmoiron/sqlx v1.4.0/go.mod h1:ZrZ7UsYB/weZdl2Bxg6jCRO9c3YHl8r3ahlKmRT4JLY=
+github.com/jonboulle/clockwork v0.4.0 h1:p4Cf1aMWXnXAUh8lVfewRBx1zaTSYKrKMF2g3ST4RZ4=
+github.com/jonboulle/clockwork v0.4.0/go.mod h1:xgRqUGwRcjKCO1vbZUEtSLrqKoPSsUpK7fnezOII0kc=
 github.com/klauspost/compress v1.17.10 h1:oXAz+Vh0PMUvJczoi+flxpnBEPxoER1IaAnU/NMPtT0=
 github.com/klauspost/compress v1.17.10/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=

internal/app/app.go

@@ -3,27 +3,35 @@ package app
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/chain4travel/camino-messenger-bot/config"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/ethclient"
+	"github.com/jonboulle/clockwork"
 	"maunium.net/go/mautrix/id"
 
 	"github.com/chain4travel/camino-messenger-bot/internal/compression"
 	"github.com/chain4travel/camino-messenger-bot/internal/matrix"
 	"github.com/chain4travel/camino-messenger-bot/internal/messaging"
 	"github.com/chain4travel/camino-messenger-bot/internal/rpc/client"
 	"github.com/chain4travel/camino-messenger-bot/internal/rpc/server"
-	"github.com/chain4travel/camino-messenger-bot/internal/scheduler"
-	"github.com/chain4travel/camino-messenger-bot/internal/storage"
 	"github.com/chain4travel/camino-messenger-bot/internal/tracing"
+	"github.com/chain4travel/camino-messenger-bot/pkg/chequehandler"
+	chequeHandlerStorage "github.com/chain4travel/camino-messenger-bot/pkg/chequehandler/storage/sqlite"
+	cmaccounts "github.com/chain4travel/camino-messenger-bot/pkg/cm_accounts"
+	"github.com/chain4travel/camino-messenger-bot/pkg/database/sqlite"
+	"github.com/chain4travel/camino-messenger-bot/pkg/scheduler"
+	scheduler_storage "github.com/chain4travel/camino-messenger-bot/pkg/scheduler/storage/sqlite"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 )
 
 const (
-	cashInJobName = "cash_in"
-	appName       = "camino-messenger-bot"
+	cashInJobName        = "cash_in"
+	appName              = "camino-messenger-bot"
+	cmAccountsCacheSize  = 100
+	cashInTxIssueTimeout = 10 * time.Second
 )
 
 func NewApp(ctx context.Context, cfg *config.Config, logger *zap.SugaredLogger) (*App, error) {
@@ -55,13 +63,6 @@ func NewApp(ctx context.Context, cfg *config.Config, logger *zap.SugaredLogger)
 		return nil, err
 	}
 
-	// database
-	storage, err := storage.New(ctx, logger, cfg.DB)
-	if err != nil {
-		logger.Errorf("Failed to create storage: %v", err)
-		return nil, err
-	}
-
 	// partner-plugin rpc client
 	rpcClient, err := client.NewClient(cfg.PartnerPlugin, logger)
 	if err != nil {
@@ -95,27 +96,37 @@ func NewApp(ctx context.Context, cfg *config.Config, logger *zap.SugaredLogger)
 		return nil, err
 	}
 
-	identificationHandler, err := messaging.NewIdentificationHandler(
+	cmAccounts, err := cmaccounts.NewService(
+		logger,
+		cmAccountsCacheSize,
 		evmClient,
+	)
+	if err != nil {
+		logger.Errorf("Failed to create cm accounts service: %v", err)
+		return nil, err
+	}
+
+	chequeHandlerStorage, err := chequeHandlerStorage.New(
+		ctx,
 		logger,
-		cfg.CMAccountAddress,
-		cfg.Matrix.HostURL,
+		sqlite.DBConfig(cfg.DB.ChequeHandler),
 	)
 	if err != nil {
-		logger.Errorf("Failed to create identification handler: %v", err)
+		logger.Errorf("Failed to create cheque handler storage: %v", err)
 		return nil, err
 	}
 
-	chequeHandler, err := messaging.NewChequeHandler(
+	chequeHandler, err := chequehandler.NewChequeHandler(
 		logger,
 		evmClient,
 		cfg.BotKey,
 		cfg.CMAccountAddress,
 		chainID,
-		storage,
-		serviceRegistry,
+		chequeHandlerStorage,
+		cmAccounts,
 		cfg.MinChequeDurationUntilExpiration,
 		cfg.ChequeExpirationTime,
+		cashInTxIssueTimeout,
 	)
 	if err != nil {
 		logger.Errorf("Failed to create cheque handler: %v", err)
@@ -137,13 +148,12 @@ func NewApp(ctx context.Context, cfg *config.Config, logger *zap.SugaredLogger)
 		cfg.NetworkFeeRecipientCMAccountAddress,
 		serviceRegistry,
 		responseHandler,
-		identificationHandler,
 		chequeHandler,
 		messaging.NewCompressor(compression.MaxChunkSize),
+		cmAccounts,
 	)
 
 	// rpc server for incoming requests
-	// TODO@ disable if we don't have port provided, e.g. its supplier bot?
 	rpcServer, err := server.NewServer(
 		cfg.RPCServer,
 		logger,
@@ -157,7 +167,14 @@ func NewApp(ctx context.Context, cfg *config.Config, logger *zap.SugaredLogger)
 	}
 
 	// scheduler for periodic tasks (e.g. cheques cash-in)
-	scheduler := scheduler.New(ctx, logger, storage)
+
+	storage, err := scheduler_storage.New(ctx, logger, sqlite.DBConfig(cfg.DB.Scheduler))
+	if err != nil {
+		logger.Errorf("Failed to create storage: %v", err)
+		return nil, err
+	}
+
+	scheduler := scheduler.New(logger, storage, clockwork.NewRealClock())
 	scheduler.RegisterJobHandler(cashInJobName, func() {
 		_ = chequeHandler.CashIn(context.Background())
 	})
@@ -181,7 +198,7 @@ type App struct {
 	logger           *zap.SugaredLogger
 	tracer           tracing.Tracer
 	scheduler        scheduler.Scheduler
-	chequeHandler    messaging.ChequeHandler
+	chequeHandler    chequehandler.ChequeHandler
 	rpcClient        *client.RPCClient
 	rpcServer        server.Server
 	messageProcessor messaging.Processor