switch to postgres once the flag is set

cmd/africastalking/main.go

@@ -77,11 +77,13 @@ func main() {
 	var dbDir string
 	var resourceDir string
 	var size uint
+	var database string
 	var engineDebug bool
 	var host string
 	var port uint
 	flag.StringVar(&dbDir, "dbdir", ".state", "database dir to read from")
 	flag.StringVar(&resourceDir, "resourcedir", path.Join("services", "registration"), "resource dir")
+	flag.StringVar(&database, "db", "gdbm", "database to be used")
 	flag.BoolVar(&engineDebug, "d", false, "use engine debug output")
 	flag.UintVar(&size, "s", 160, "max size of output")
 	flag.StringVar(&host, "h", initializers.GetEnv("HOST", "127.0.0.1"), "http host")
@@ -91,6 +93,7 @@ func main() {
 	logg.Infof("start command", "dbdir", dbDir, "resourcedir", resourceDir, "outputsize", size)
 
 	ctx := context.Background()
+	ctx = context.WithValue(ctx, "Database", database)
 	pfp := path.Join(scriptDir, "pp.csv")
 
 	cfg := engine.Config{

cmd/async/main.go

@@ -48,12 +48,14 @@ func main() {
 	var dbDir string
 	var resourceDir string
 	var size uint
+	var database string
 	var engineDebug bool
 	var host string
 	var port uint
 	flag.StringVar(&sessionId, "session-id", "075xx2123", "session id")
 	flag.StringVar(&dbDir, "dbdir", ".state", "database dir to read from")
 	flag.StringVar(&resourceDir, "resourcedir", path.Join("services", "registration"), "resource dir")
+	flag.StringVar(&database, "db", "gdbm", "database to be used")
 	flag.BoolVar(&engineDebug, "d", false, "use engine debug output")
 	flag.UintVar(&size, "s", 160, "max size of output")
 	flag.StringVar(&host, "h", initializers.GetEnv("HOST", "127.0.0.1"), "http host")
@@ -63,6 +65,7 @@ func main() {
 	logg.Infof("start command", "dbdir", dbDir, "resourcedir", resourceDir, "outputsize", size, "sessionId", sessionId)
 
 	ctx := context.Background()
+	ctx = context.WithValue(ctx, "Database", database)
 	pfp := path.Join(scriptDir, "pp.csv")
 
 	cfg := engine.Config{

cmd/http/main.go

@@ -37,11 +37,13 @@ func main() {
 	var dbDir string
 	var resourceDir string
 	var size uint
+	var database string
 	var engineDebug bool
 	var host string
 	var port uint
 	flag.StringVar(&dbDir, "dbdir", ".state", "database dir to read from")
 	flag.StringVar(&resourceDir, "resourcedir", path.Join("services", "registration"), "resource dir")
+	flag.StringVar(&database, "db", "gdbm", "database to be used")
 	flag.BoolVar(&engineDebug, "d", false, "use engine debug output")
 	flag.UintVar(&size, "s", 160, "max size of output")
 	flag.StringVar(&host, "h", initializers.GetEnv("HOST", "127.0.0.1"), "http host")
@@ -51,6 +53,7 @@ func main() {
 	logg.Infof("start command", "dbdir", dbDir, "resourcedir", resourceDir, "outputsize", size)
 
 	ctx := context.Background()
+	ctx = context.WithValue(ctx, "Database", database)
 	pfp := path.Join(scriptDir, "pp.csv")
 
 	cfg := engine.Config{

cmd/main.go

@@ -31,8 +31,10 @@ func main() {
 	var dbDir string
 	var size uint
 	var sessionId string
+	var database string
 	var engineDebug bool
 	flag.StringVar(&sessionId, "session-id", "075xx2123", "session id")
+	flag.StringVar(&database, "db", "gdbm", "database to be used")
 	flag.StringVar(&dbDir, "dbdir", ".state", "database dir to read from")
 	flag.BoolVar(&engineDebug, "d", false, "use engine debug output")
 	flag.UintVar(&size, "s", 160, "max size of output")
@@ -42,6 +44,7 @@ func main() {
 
 	ctx := context.Background()
 	ctx = context.WithValue(ctx, "SessionId", sessionId)
+	ctx = context.WithValue(ctx, "Database", database)
 	pfp := path.Join(scriptDir, "pp.csv")
 
 	cfg := engine.Config{

internal/storage/postgres.go

@@ -0,0 +1,115 @@
+package storage
+
+import (
+	"context"
+
+	"git.defalsify.org/vise.git/db"
+	postgres "git.defalsify.org/vise.git/db/postgres"
+	"git.defalsify.org/vise.git/lang"
+)
+
+var (
+	pdbC map[string]chan db.Db
+)
+
+type ThreadPostgresDb struct {
+	db      db.Db
+	connStr string
+}
+
+func NewThreadPostgresDb() *ThreadPostgresDb {
+	if pdbC == nil {
+		pdbC = make(map[string]chan db.Db)
+	}
+	return &ThreadPostgresDb{}
+}
+
+func (tpdb *ThreadPostgresDb) Connect(ctx context.Context, connStr string) error {
+	var ok bool
+	_, ok = pdbC[connStr]
+	if ok {
+		logg.WarnCtxf(ctx, "already registered thread postgres, skipping", "connStr", connStr)
+		return nil
+	}
+	postgresdb := postgres.NewPgDb().WithSchema("public")
+	err := postgresdb.Connect(ctx, connStr)
+	if err != nil {
+		return err
+	}
+	pdbC[connStr] = make(chan db.Db, 1)
+	pdbC[connStr] <- postgresdb
+	tpdb.connStr = connStr
+	return nil
+}
+
+func (tpdb *ThreadPostgresDb) reserve() {
+	if tpdb.db == nil {
+		tpdb.db = <-pdbC[tpdb.connStr]
+	}
+}
+
+func (tpdb *ThreadPostgresDb) release() {
+	if tpdb.db == nil {
+		return
+	}
+	pdbC[tpdb.connStr] <- tpdb.db
+	tpdb.db = nil
+}
+
+func (tpdb *ThreadPostgresDb) SetPrefix(pfx uint8) {
+	tpdb.reserve()
+	tpdb.db.SetPrefix(pfx)
+}
+
+func (tpdb *ThreadPostgresDb) SetSession(sessionId string) {
+	tpdb.reserve()
+	tpdb.db.SetSession(sessionId)
+}
+
+func (tpdb *ThreadPostgresDb) SetLanguage(lng *lang.Language) {
+	tpdb.reserve()
+	tpdb.db.SetLanguage(lng)
+}
+
+func (tpdb *ThreadPostgresDb) Safe() bool {
+	tpdb.reserve()
+	v := tpdb.db.Safe()
+	tpdb.release()
+	return v
+}
+
+func (tpdb *ThreadPostgresDb) Prefix() uint8 {
+	tpdb.reserve()
+	v := tpdb.db.Prefix()
+	tpdb.release()
+	return v
+}
+
+func (tpdb *ThreadPostgresDb) SetLock(typ uint8, locked bool) error {
+	tpdb.reserve()
+	err := tpdb.db.SetLock(typ, locked)
+	tpdb.release()
+	return err
+}
+
+func (tpdb *ThreadPostgresDb) Put(ctx context.Context, key []byte, val []byte) error {
+	tpdb.reserve()
+	err := tpdb.db.Put(ctx, key, val)
+	tpdb.release()
+	return err
+}
+
+func (tpdb *ThreadPostgresDb) Get(ctx context.Context, key []byte) ([]byte, error) {
+	tpdb.reserve()
+	v, err := tpdb.db.Get(ctx, key)
+	tpdb.release()
+	return v, err
+}
+
+func (tpdb *ThreadPostgresDb) Close() error {
+	tpdb.reserve()
+	close(pdbC[tpdb.connStr])
+	err := tpdb.db.Close()
+	tpdb.db = nil
+	return err
+}

internal/storage/storageservice.go

@@ -8,9 +8,10 @@ import (
 
 	"git.defalsify.org/vise.git/db"
 	fsdb "git.defalsify.org/vise.git/db/fs"
+	"git.defalsify.org/vise.git/logging"
 	"git.defalsify.org/vise.git/persist"
 	"git.defalsify.org/vise.git/resource"
-	"git.defalsify.org/vise.git/logging"
+	"git.grassecon.net/urdt/ussd/initializers"
 )
 
 var (
@@ -24,7 +25,7 @@ type StorageService interface {
 	EnsureDbDir() error
 }
 
-type MenuStorageService struct{
+type MenuStorageService struct {
 	dbDir         string
 	resourceDir   string
 	resourceStore db.Db
@@ -32,6 +33,19 @@ type MenuStorageService struct{
 	userDataStore db.Db
 }
 
+func buildConnStr() string {
+	host := initializers.GetEnv("DB_HOST", "localhost")
+	user := initializers.GetEnv("DB_USER", "postgres")
+	password := initializers.GetEnv("DB_PASSWORD", "")
+	dbName := initializers.GetEnv("DB_NAME", "")
+	port := initializers.GetEnv("DB_PORT", "5432")
+
+	return fmt.Sprintf(
+		"postgres://%s:%s@%s:%s/%s",
+		user, password, host, port, dbName,
+	)
+}
+
 func NewMenuStorageService(dbDir string, resourceDir string) *MenuStorageService {
 	return &MenuStorageService{
 		dbDir:       dbDir,
@@ -52,12 +66,27 @@ func (ms *MenuStorageService) GetPersister(ctx context.Context) (*persist.Persis
 }
 
 func (ms *MenuStorageService) GetUserdataDb(ctx context.Context) (db.Db, error) {
+	database, ok := ctx.Value("Database").(string)
+	if !ok {
+		fmt.Println("The database is not set")
+	}
+
+	if database == "postgres" {
+		ms.userDataStore = NewThreadPostgresDb()
+		connStr := buildConnStr()
+		err := ms.userDataStore.Connect(ctx, connStr)
+		if err != nil {
+			return nil, err
+		}
+	} else {
 		ms.userDataStore = NewThreadGdbmDb()
 		storeFile := path.Join(ms.dbDir, "userdata.gdbm")
 		err := ms.userDataStore.Connect(ctx, storeFile)
 		if err != nil {
 			return nil, err
 		}
+	}
+
 	return ms.userDataStore, nil
 }