CROSSLINK-203 Add creation of crosslink_broker schema

README.md

@@ -61,3 +61,8 @@ Charts use the `LoadBalancer` service type by default, you can change this durin
 ```
 --set service.type=ClusterIP
 ```
+
+For any installation, the application database user must have privileges to create schemas in the database
+(for example, the `CREATE` privilege on the database or the ability to run `CREATE SCHEMA`).
+Database migrations will create and update all required tables and other objects in the `crosslink_broker` schema,
+which is selected via the PostgreSQL `search_path`.

broker/README.md

@@ -75,6 +75,7 @@ Configuration is provided via environment variables:
 | `DB_HOST`                 | Database host                                                                         | `localhost`                               |
 | `DB_DATABASE`             | Database name                                                                         | `crosslink`                               |
 | `DB_PORT`                 | Database port                                                                         | `25432`                                   |
+| `DB_SCHEMA`               | Database schema to use                                                                | `crosslink_broker`                        |
 | `LOG_LEVEL`               | Log level: `ERROR`, `WARN`, `INFO`, `DEBUG`                                           | `INFO`                                    |
 | `ENABLE_JSON_LOG`         | Should JSON log format be enabled                                                     | `false`                                   |
 | `BROKER_MODE`             | Default broker mode if not configured for a peer: `opaque` or `transparent`           | `opaque`                                  |

broker/dbutil/dbutil.go

@@ -1,17 +1,26 @@
 package dbutil
 
 import (
+	"bytes"
 	"context"
+	"database/sql"
 	"fmt"
+	"strings"
+	"text/template"
 
 	"github.com/golang-migrate/migrate/v4"
 	_ "github.com/golang-migrate/migrate/v4/database/postgres"
 	_ "github.com/golang-migrate/migrate/v4/source/file"
+	"github.com/indexdata/go-utils/utils"
 	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/lib/pq"
 )
 
+var DB_SCHEMA = utils.GetEnv("DB_SCHEMA", "crosslink_broker")
+var SchemaParam = "&search_path=" + DB_SCHEMA
+
 func GetConnectionString(typ, user, pass, host, port, db string) string {
-	return fmt.Sprintf("%s://%s:%s@%s:%s/%s?sslmode=disable", typ, user, pass, host, port, db)
+	return fmt.Sprintf("%s://%s:%s@%s:%s/%s?sslmode=disable"+SchemaParam, typ, user, pass, host, port, db)
 }
 
 func InitDbPool(connStr string) (*pgxpool.Pool, error) {
@@ -21,6 +30,10 @@ func InitDbPool(connStr string) (*pgxpool.Pool, error) {
 func RunMigrateScripts(migrateDir, connStr string) (uint, uint, bool, error) {
 	var versionFrom, versionTo uint
 	var dirty bool
+	err := initDBSchema(connStr)
+	if err != nil {
+		return versionFrom, versionTo, dirty, fmt.Errorf("failed to initiate schema: %w", err)
+	}
 	m, err := migrate.New(migrateDir, connStr)
 	if err != nil {
 		return versionFrom, versionTo, dirty, fmt.Errorf("failed to initiate migration: %w", err)
@@ -43,3 +56,48 @@ func RunMigrateScripts(migrateDir, connStr string) (uint, uint, bool, error) {
 	}
 	return versionFrom, versionTo, dirty, nil
 }
+
+func initDBSchema(connStr string) error {
+	connStr = strings.Replace(connStr, SchemaParam, "", 1)
+	db, err := sql.Open("postgres", connStr)
+	if err != nil {
+		return fmt.Errorf("error opening database: : %w", err)
+	}
+	defer db.Close()
+
+	const setupSQL = `
+        DO $$ 
+        BEGIN 
+            IF NOT EXISTS (SELECT FROM pg_catalog.pg_roles WHERE rolname = {{.Literal}}) THEN 
+                CREATE ROLE {{.Identifier}} WITH PASSWORD 'tenant' LOGIN; 
+            END IF; 
+        END 
+        $$; 
+        CREATE SCHEMA IF NOT EXISTS {{.Identifier}} AUTHORIZATION {{.Identifier}};
+        SET search_path TO {{.Identifier}};`
+
+	tmpl, err := template.New("setup").Parse(setupSQL)
+	if err != nil {
+		return err
+	}
+
+	var buf bytes.Buffer
+	data := struct {
+		Literal    string
+		Identifier string
+	}{
+		Literal:    pq.QuoteLiteral(DB_SCHEMA),
+		Identifier: pq.QuoteIdentifier(DB_SCHEMA),
+	}
+
+	if err = tmpl.Execute(&buf, data); err != nil {
+		return err
+	}
+
+	_, err = db.Exec(buf.String())
+	if err != nil {
+		return fmt.Errorf("error executing script: %w", err)
+	}
+
+	return nil
+}

broker/ill_db/illrepo_test.go

@@ -29,6 +29,7 @@ func createDirectoryAdapter(urls ...string) adapter.DirectoryLookupAdapter {
 
 func TestMain(m *testing.M) {
 	ctx, pgc, connStr, err := test.StartPGContainer()
+	connStr = connStr + dbutil.SchemaParam
 	test.Expect(err, "failed to start db container")
 	pgIllRepo := new(PgIllRepo)
 	pgIllRepo.Pool, err = dbutil.InitDbPool(connStr)