Database drivers

Author: bartekn

README.md

@@ -17,11 +17,13 @@ The `config.toml` file must be present in a working directory. Config file shoul
 
 * `port` - server listening port
 * `api_key` - when set, all requests to gateway server must contain `api_key` parameter with a correct value, otherwise the server will respond with `503 Forbidden`
-* `network_passphrase` - passphrase of the network that will be used with this gateway server, default: `Test SDF Network ; September 2015`
+* `network_passphrase` - passphrase of the network that will be used with this gateway server:
+   * test network: `Test SDF Network ; September 2015`
+   * public network: `Public Global Stellar Network ; September 2015`
 * `horizon` - URL to [horizon](https://github.com/stellar/horizon) server instance
 * `assets` - array of approved assets codes that this server can authorize and send 
 * `database`
-  * `type` - database type (sqlite3, mysql, postgres)
+  * `type` - database type (mysql, postgres)
   * `url` - url to database connection
 * `accounts`
   * `authorizing_seed` - secret seed of the account to send `allow_trust` operations
@@ -33,6 +35,13 @@ The `config.toml` file must be present in a working directory. Config file shoul
 
 Check [`config-example.toml`](./config-example.toml).
 
+The minimal set of config values contains:
+* `port`
+* `network_passphrase`
+* `horizon`
+
+It will start a server with a single endpoint: `/payment`.
+
 ## Getting started
 
 After creating `config.toml` file, you need to run DB migrations:

src/github.com/stellar/gateway/app.go

@@ -5,10 +5,13 @@ import (
 	"flag"
 	"fmt"
 	log "github.com/Sirupsen/logrus"
+	"os"
 	"time"
 
 	"github.com/stellar/gateway/config"
 	"github.com/stellar/gateway/db"
+	"github.com/stellar/gateway/db/drivers/mysql"
+	"github.com/stellar/gateway/db/drivers/postgres"
 	"github.com/stellar/gateway/handlers"
 	"github.com/stellar/gateway/horizon"
 	"github.com/stellar/gateway/listener"
@@ -19,31 +22,62 @@ import (
 
 type App struct {
 	config               config.Config
+	driver               db.Driver
 	entityManager        db.EntityManagerInterface
 	horizon              horizon.HorizonInterface
 	transactionSubmitter *submitter.TransactionSubmitter
 	repository           db.RepositoryInterface
 }
 
 // NewApp constructs an new App instance from the provided config.
-func NewApp(config config.Config) (app *App, err error) {
-	entityManager, err := db.NewEntityManager(config.Database.Type, config.Database.Url)
-	if err != nil {
-		return
+func NewApp(config config.Config, migrateFlag bool) (app *App, err error) {
+	var driver db.Driver
+	switch config.Database.Type {
+	case "mysql":
+		driver = &mysql.MysqlDriver{}
+	case "postgres":
+		driver = &postgres.PostgresDriver{}
+	case "":
+		// Allow to start gateway server with a single endpoint: /payment
+		break
+	default:
+		return nil, fmt.Errorf("%s database has no driver.", config.Database.Type)
 	}
-	repository, err := db.NewRepository(config.Database.Type, config.Database.Url)
-	if err != nil {
-		return
+
+	var entityManager db.EntityManagerInterface
+	var repository db.RepositoryInterface
+
+	if driver != nil {
+		err = driver.Init(config.Database.Url)
+		if err != nil {
+			return
+		}
+
+		entityManager = db.NewEntityManager(driver)
+		repository = db.NewRepository(driver)
 	}
 
-	h := horizon.New(*config.Horizon)
+	if migrateFlag {
+		if driver == nil {
+			log.Fatal("No database driver.")
+			return
+		}
+
+		var migrationsApplied int
+		migrationsApplied, err = driver.MigrateUp()
+		if err != nil {
+			return
+		}
 
-	if config.NetworkPassphrase == "" {
-		config.NetworkPassphrase = "Test SDF Network ; September 2015"
+		log.Info("Applied migrations: ", migrationsApplied)
+		os.Exit(0)
+		return
 	}
 
+	h := horizon.New(*config.Horizon)
+
 	log.Print("Creating and initializing TransactionSubmitter")
-	ts := submitter.NewTransactionSubmitter(&h, &entityManager, config.NetworkPassphrase)
+	ts := submitter.NewTransactionSubmitter(&h, entityManager, config.NetworkPassphrase)
 	if err != nil {
 		return
 	}
@@ -79,7 +113,7 @@ func NewApp(config config.Config) (app *App, err error) {
 		log.Warning("No hooks.receive param. Skipping...")
 	} else {
 		var paymentListener listener.PaymentListener
-		paymentListener, err = listener.NewPaymentListener(&config, &entityManager, &h, &repository, time.Now)
+		paymentListener, err = listener.NewPaymentListener(&config, entityManager, &h, repository, time.Now)
 		if err != nil {
 			return
 		}
@@ -98,9 +132,10 @@ func NewApp(config config.Config) (app *App, err error) {
 
 	app = &App{
 		config:               config,
-		entityManager:        &entityManager,
+		driver:               driver,
+		entityManager:        entityManager,
 		horizon:              &h,
-		repository:           &repository,
+		repository:           repository,
 		transactionSubmitter: &ts,
 	}
 	return

src/github.com/stellar/gateway/cmd/gateway/main.go

@@ -8,7 +8,6 @@ import (
 	"github.com/spf13/viper"
 	"github.com/stellar/gateway"
 	"github.com/stellar/gateway/config"
-	"github.com/stellar/gateway/db/migrations"
 )
 
 var app *gateway.App
@@ -35,18 +34,6 @@ func init() {
 	rootCmd.Flags().BoolVarP(&migrateFlag, "migrate-db", "", false, "migrate DB to the newest schema version")
 }
 
-func migrate(config config.Config) {
-	migrationManager, err := migrations.NewMigrationManager(
-		config.Database.Type,
-		config.Database.Url,
-	)
-	if err != nil {
-		log.Fatal("Error migrating DB")
-		return
-	}
-	migrationManager.MigrateUp()
-}
-
 func run(cmd *cobra.Command, args []string) {
 	log.Print("Reading config.toml file")
 	err := viper.ReadInConfig()
@@ -63,12 +50,7 @@ func run(cmd *cobra.Command, args []string) {
 		return
 	}
 
-	if migrateFlag {
-		migrate(config)
-		return
-	}
-
-	app, err = gateway.NewApp(config)
+	app, err = gateway.NewApp(config, migrateFlag)
 
 	if err != nil {
 		log.Fatal(err.Error())

src/github.com/stellar/gateway/db/driver.go

@@ -0,0 +1,13 @@
+package db
+
+type Driver interface {
+	Init(url string) (err error)
+	MigrateUp() (migrationsApplied int, err error)
+
+	InsertReceivedPayment(object *ReceivedPayment) (id int64, err error)
+	UpdateReceivedPayment(object *ReceivedPayment) (err error)
+	GetLastReceivedPayment() (*ReceivedPayment, error)
+
+	InsertSentTransaction(object *SentTransaction) (id int64, err error)
+	UpdateSentTransaction(object *SentTransaction) (err error)
+}

src/github.com/stellar/gateway/db/migrations/bindata.go renamed to src/github.com/stellar/gateway/db/drivers/mysql/bindata.go

@@ -0,0 +1,110 @@
+package mysql
+
+import (
+	_ "github.com/go-sql-driver/mysql"
+	"github.com/jmoiron/sqlx"
+	"github.com/rubenv/sql-migrate"
+	"github.com/stellar/gateway/db"
+)
+
+type MysqlDriver struct {
+	database *sqlx.DB
+}
+
+func (d *MysqlDriver) Init(url string) (err error) {
+	d.database, err = sqlx.Connect("mysql", url)
+	return
+}
+
+// go-bindata -ignore .+\.go$ -pkg mysql -o bindata.go ./migrations
+func (d *MysqlDriver) MigrateUp() (migrationsApplied int, err error) {
+	source := d.getAssetMigrationSource()
+	migrationsApplied, err = migrate.Exec(d.database.DB, "mysql", source, migrate.Up)
+	return
+}
+
+func (d *MysqlDriver) InsertReceivedPayment(object *db.ReceivedPayment) (id int64, err error) {
+	query := `
+	INSERT INTO ReceivedPayment
+		(operation_id, processed_at, paging_token, status)
+	VALUES
+		(:operation_id, :processed_at, :paging_token, :status)`
+	id, err = d.insert(query, object)
+	return
+}
+
+func (d *MysqlDriver) UpdateReceivedPayment(object *db.ReceivedPayment) (err error) {
+	query := `
+	UPDATE ReceivedPayment SET
+		operation_id = :operation_id,
+		processed_at = :processed_at,
+		paging_token = :paging_token,
+		status = :status
+	WHERE
+		id = :id`
+	err = d.update(query, object)
+	return
+}
+
+func (d *MysqlDriver) GetLastReceivedPayment() (*db.ReceivedPayment, error) {
+	var receivedPayment db.ReceivedPayment
+	err := d.database.Get(&receivedPayment, "SELECT * FROM ReceivedPayment ORDER BY id DESC LIMIT 1")
+	if err != nil {
+		if err.Error() == "sql: no rows in result set" {
+			return nil, nil
+		} else {
+			return nil, err
+		}
+	}
+	return &receivedPayment, nil
+}
+
+func (d *MysqlDriver) InsertSentTransaction(object *db.SentTransaction) (id int64, err error) {
+	query := `
+	INSERT INTO SentTransaction
+		(status, source, submitted_at, succeeded_at, ledger, envelope_xdr, result_xdr)
+	VALUES
+		(:status, :source, :submitted_at, :succeeded_at, :ledger, :envelope_xdr, :result_xdr)`
+	id, err = d.insert(query, object)
+	return
+}
+
+func (d *MysqlDriver) UpdateSentTransaction(object *db.SentTransaction) (err error) {
+	query := `
+	UPDATE SentTransaction SET
+		status = :status,
+		source = :source,
+		submitted_at = :submitted_at,
+		succeeded_at = :succeeded_at,
+		ledger = :ledger,
+		envelope_xdr = :envelope_xdr,
+		result_xdr = :result_xdr
+	WHERE
+		id = :id`
+	err = d.update(query, object)
+	return
+}
+
+func (d *MysqlDriver) insert(query string, object interface{}) (id int64, err error) {
+	result, err := d.database.NamedExec(query, object)
+	if err != nil {
+		return
+	}
+
+	id, err = result.LastInsertId()
+	return
+}
+
+func (d *MysqlDriver) update(query string, object interface{}) (err error) {
+	_, err = d.database.NamedExec(query, object)
+	return
+}
+
+func (d *MysqlDriver) getAssetMigrationSource() (source *migrate.AssetMigrationSource) {
+	source = &migrate.AssetMigrationSource{
+		Asset:    Asset,
+		AssetDir: AssetDir,
+		Dir:      "migrations",
+	}
+	return
+}