FEAT[DXTA-320]: Create tenant db workflow

cmd/onboarding-worker/main.go

@@ -23,6 +23,12 @@ func main() {
 		log.Fatalln("Failed to load github configuration:", err)
 	}
 
+	createTenantConfig, err := onboarding.LoadCreateTenantConfig()
+
+	if err != nil {
+		log.Fatalln("Failed to load create tenant configuration:", err)
+	}
+
 	temporalClient, err := client.Dial(client.Options{
 		HostPort:  cfg.TemporalHostPort,
 		Namespace: cfg.TemporalOnboardingNamespace,
@@ -54,13 +60,17 @@ func main() {
 	githubInstallationActivities := activity.NewGithubInstallationActivities(*githubAppClient)
 	tenantActivities := activity.NewTenantActivities()
 	githubActivities := activity.NewGithubActivities(*githubConfig)
+	createTenantActivities := activity.NewCreateTenantActivities(*createTenantConfig)
 
 	w.RegisterWorkflow(workflow.CountUsers)
 	w.RegisterWorkflow(workflow.AfterGithubInstallationWorkflow)
+	w.RegisterWorkflow(workflow.CreateTenantDBWorkflow)
+
 	w.RegisterActivity(userActivities)
 	w.RegisterActivity(githubInstallationActivities)
 	w.RegisterActivity(tenantActivities)
 	w.RegisterActivity(githubActivities)
+	w.RegisterActivity(createTenantActivities)
 
 	if err := w.Run(worker.InterruptCh()); err != nil {
 		log.Fatalln("Worker failed to start", err)

internal/internal-api/data/db.go

@@ -17,6 +17,10 @@ func NewDB(DBURL string, ctx context.Context) (DB, error) {
 	driverName := otel.GetDriverName()
 	devToken := os.Getenv("DXTA_DEV_GROUP_TOKEN")
 
+	if devToken == "" {
+		return DB{}, errors.New("no dev group token provided")
+	}
+
 	tenantDB, err := sql.Open(
 		driverName,
 		DBURL+"?authToken="+devToken,

internal/onboarding/activity/organization.go

@@ -20,7 +20,7 @@ func (ta *TenantActivities) GetOrganizationIDByAuthID(ctx context.Context, authI
 	db, err := ta.GetCachedTenantDB(DBURL, ctx)
 
 	if err != nil {
-		return 0, err
+		return 0, errors.New("failed to get cached tenant DB: " + err.Error())
 	}
 
 	var organizationId int64
@@ -38,3 +38,30 @@ func (ta *TenantActivities) GetOrganizationIDByAuthID(ctx context.Context, authI
 
 	return organizationId, nil
 }
+
+func (ta *TenantActivities) CreateOrganization(
+	ctx context.Context,
+	organizationName string,
+	authID string,
+	DBURL string,
+) (bool, error) {
+
+	db, err := ta.GetCachedTenantDB(DBURL, ctx)
+
+	if err != nil {
+		return false, errors.New("failed to get cached tenant DB: " + err.Error())
+	}
+
+	_, err = db.QueryContext(ctx, `
+		INSERT INTO organizations 
+			(name, auth_id) 
+		VALUES 
+			(?, ?);`,
+		organizationName, authID)
+
+	if err != nil {
+		return false, errors.New("failed to create organization: " + err.Error())
+	}
+
+	return true, nil
+}

internal/onboarding/activity/settings.go

@@ -0,0 +1,33 @@
+package activity
+
+import (
+	"context"
+	"errors"
+)
+
+func (ta *TenantActivities) UpsertTenantDBInfo(
+	ctx context.Context,
+	DBName string,
+	DBURL string,
+	DBDomainName string,
+) (bool, error) {
+	db, err := ta.GetCachedTenantDB(DBURL, ctx)
+
+	if err != nil {
+		return false, err
+	}
+
+	_, err = db.QueryContext(ctx, `
+		INSERT INTO settings 
+			(tenant_name, tenant_domain) 
+		VALUES 
+			(?, ?);`,
+		DBName, DBDomainName,
+	)
+
+	if err != nil {
+		return false, errors.New("Failed to upsert tenant db info: " + err.Error())
+	}
+
+	return true, nil
+}

internal/onboarding/activity/tenant_db.go

@@ -0,0 +1,133 @@
+package activity
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+
+	"github.com/dxta-dev/app/internal/onboarding"
+)
+
+type DatabaseData struct {
+	DBID     string `json:"DbId"`
+	Hostname string `json:"Hostname"`
+	Name     string `json:"Name"`
+	DBURL    string
+}
+type CreateTenantDBRes struct {
+	Database DatabaseData `json:"database"`
+}
+
+type CreateTenantActivities struct {
+	config onboarding.CreateTenantConfig
+}
+
+type SeedOpts struct {
+	Type string `json:"type"`
+	Name string `json:"name"`
+}
+type TenantDBRequest struct {
+	Name  string   `json:"name"`
+	Group string   `json:"group"`
+	Seed  SeedOpts `json:"seed"`
+}
+
+func NewCreateTenantActivities(config onboarding.CreateTenantConfig) *CreateTenantActivities {
+	return &CreateTenantActivities{config}
+}
+
+func (cta CreateTenantActivities) CreateTenantDB(
+	ctx context.Context,
+	dbDomainName string,
+) (*CreateTenantDBRes, error) {
+	reqBody := TenantDBRequest{
+		Name:  dbDomainName,
+		Group: cta.config.TursoDBGroupName,
+		Seed: SeedOpts{
+			Type: "database",
+			Name: cta.config.TenantSeedDBURL,
+		},
+	}
+
+	jsonBody, err := json.Marshal(reqBody)
+
+	if err != nil {
+		return nil, errors.New("failed while marshalling request body: " + err.Error())
+	}
+
+	apiUrl := fmt.Sprintf("%s/organizations/%s/databases", cta.config.TursoApiURL, cta.config.TursoOrganizationSlug)
+
+	req, err := http.NewRequest("POST", apiUrl, bytes.NewBuffer(jsonBody))
+
+	if err != nil {
+		return nil, errors.New("failed to create HTTP request: " + err.Error())
+	}
+
+	req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", cta.config.TursoAuthToken))
+	req.Header.Set("Content-Type", "application/json")
+
+	client := http.Client{}
+
+	response, err := client.Do(req)
+
+	if err != nil {
+		return nil, errors.New("failed to send HTTP request: " + err.Error())
+	}
+
+	defer response.Body.Close()
+
+	if response.StatusCode != http.StatusOK {
+		return nil, errors.New("failed to create new tenant db with status code: " + fmt.Sprint(response.StatusCode))
+	}
+
+	responseBodyBytes, err := io.ReadAll(response.Body)
+
+	if err != nil {
+		return nil, errors.New("failed to read response body: " + err.Error())
+	}
+
+	var body CreateTenantDBRes
+
+	err = json.Unmarshal(responseBodyBytes, &body)
+
+	if err != nil {
+		return nil, errors.New("failed to unmarshal response body: " + err.Error())
+	}
+
+	fmt.Printf("Success! Data: %v", body)
+
+	body.Database.DBURL = fmt.Sprintf("libsql://%s", body.Database.Hostname)
+
+	return &body, nil
+}
+
+func (cta CreateTenantActivities) AddTenantDBToMap(
+	ctx context.Context,
+	authId string,
+	DBName string,
+	DBURL string,
+	DBDomainName string,
+) (bool, error) {
+	db, err := onboarding.NewDB(cta.config.OrganizationsTenantMapDBURL, ctx)
+
+	if err != nil {
+		return false, errors.New("failed to get organizations-tenant-map db: " + err.Error())
+	}
+
+	defer db.DB.Close()
+
+	_, err = db.DB.QueryContext(ctx, `
+		INSERT INTO tenants 
+			(organization_id, db_url, name, domain) 
+		VALUES (?, ?, ?, ?);`, authId, DBURL, DBName, DBDomainName)
+
+	if err != nil {
+		return false, errors.New("failed to store tenant db data to organizations-tenant-map db: " + err.Error())
+	}
+
+	return true, nil
+}

internal/onboarding/tenant.go

@@ -4,19 +4,97 @@
 	"context"
 	"database/sql"
 	"errors"
+	"os"
 	"sync"
 
-	internal_api_data "github.com/dxta-dev/app/internal/internal-api/data"
+	"github.com/dxta-dev/app/internal/otel"
 )
 
 var tenantDBConnections = sync.Map{}
 
+type CreateTenantConfig struct {
+	TursoApiURL                 string
+	TursoOrganizationSlug       string
+	TursoDBGroupName            string
+	TursoAuthToken              string
+	TenantSeedDBURL             string
+	OrganizationsTenantMapDBURL string
+}
+
+func LoadCreateTenantConfig() (*CreateTenantConfig, error) {
+	var tursoAuthToken, tenantSeedDBURL, organizationsTenantMapDBURL, tursoApiUrl, tursoOrganizationSlug, tursoDBGroupName string
+
+	if tursoAuthToken = os.Getenv("TURSO_AUTH_TOKEN"); tursoAuthToken == "" {
+		return nil, errors.New("turso auth token not defined")
+	}
+
+	if tenantSeedDBURL = os.Getenv("TENANT_SEED_DB_NAME"); tenantSeedDBURL == "" {
+		return nil, errors.New("seed db url not defined")
+	}
+
+	if organizationsTenantMapDBURL = os.Getenv("ORGANIZATIONS_TENANT_MAP_DB_URL"); organizationsTenantMapDBURL == "" {
+		return nil, errors.New("organizations tenant map db url not defined")
+	}
+
+	if tursoApiUrl = os.Getenv("TURSO_API_URL"); tursoApiUrl == "" {
+		return nil, errors.New("turso api url not defined")
+	}
+
+	if tursoOrganizationSlug = os.Getenv("TURSO_ORGANIZATION_SLUG"); tursoOrganizationSlug == "" {
+		return nil, errors.New("turso organization slug not defined")
+	}
+
+	if tursoDBGroupName = os.Getenv("TURSO_DB_GROUP_NAME"); tursoDBGroupName == "" {
+		return nil, errors.New("turso db group name not defined")
+	}
+
+	return &CreateTenantConfig{
+		TursoApiURL:                 tursoApiUrl,
+		TursoOrganizationSlug:       tursoOrganizationSlug,
+		TursoDBGroupName:            tursoDBGroupName,
+		TursoAuthToken:              tursoAuthToken,
+		TenantSeedDBURL:             tenantSeedDBURL,
+		OrganizationsTenantMapDBURL: organizationsTenantMapDBURL,
+	}, nil
+
+}
+
+type DB struct {
+	DB *sql.DB
+}
+
+func NewDB(DBURL string, ctx context.Context) (DB, error) {
+	driverName := otel.GetDriverName()
+	devToken := os.Getenv("DXTA_DEV_GROUP_TOKEN")
+
+	if devToken == "" {
+		return DB{}, errors.New("no dev group token provided")
+	}
+
+	tenantDB, err := sql.Open(
+		driverName,
+		DBURL+"?authToken="+devToken,
+	)
+
+	if err != nil {
+		return DB{}, errors.New("failed to open db connection " + err.Error())
+	}
+
+	if err := tenantDB.PingContext(ctx); err != nil {
+		return DB{}, errors.New("failed to verify db connection " + err.Error())
+	}
+
+	return DB{
+		DB: tenantDB,
+	}, nil
+}
+
 func GetCachedTenantDB(DBURL string, ctx context.Context) (*sql.DB, error) {
 	if cachedDB, ok := tenantDBConnections.Load(DBURL); ok {
 		return cachedDB.(*sql.DB), nil
 	}
 
-	db, err := internal_api_data.NewDB(DBURL, ctx)
+	db, err := NewDB(DBURL, ctx)
 
 	if err != nil {
 		return nil, errors.New("failed to create tenant db connection: " + err.Error())
@@ -26,3 +104,13 @@
 
 	return db.DB, nil
 }
+
+func GetDB(ctx context.Context, DBURL string) (*sql.DB, error) {
+	db, err := NewDB(DBURL, ctx)
+
+	if err != nil {
+		return nil, errors.New("failed to create db connection: " + err.Error())
+	}
+
+	return db.DB, nil
+}