mirror of
https://github.com/andwati/zitadel.git
synced 2026-11-02 20:32:19 +00:00
feat(backend): state persisted objects (#9870)
This PR initiates the rework of Zitadel's backend to state-persisted objects. This change is a step towards a more scalable and maintainable architecture. ## Changes * **New `/backend/v3` package**: A new package structure has been introduced to house the reworked backend logic. This includes: * `domain`: Contains the core business logic, commands, and repository interfaces. * `storage`: Implements the repository interfaces for database interactions with new transactional tables. * `telemetry`: Provides logging and tracing capabilities. * **Transactional Tables**: New database tables have been defined for `instances`, `instance_domains`, `organizations`, and `org_domains`. * **Projections**: New projections have been created to populate the new relational tables from the existing event store, ensuring data consistency during the migration. * **Repositories**: New repositories provide an abstraction layer for accessing and manipulating the data in the new tables. * **Setup**: A new setup step for `TransactionalTables` has been added to manage the database migrations for the new tables. This PR lays the foundation for future work to fully transition to state-persisted objects for these components, which will improve performance and simplify data access patterns. This PR initiates the rework of ZITADEL's backend to state-persisted objects. This is a foundational step towards a new architecture that will improve performance and maintainability. The following objects are migrated from event-sourced aggregates to state-persisted objects: * Instances * incl. Domains * Orgs * incl. Domains The structure of the new backend implementation follows the software architecture defined in this [wiki page](https://github.com/zitadel/zitadel/wiki/Software-Architecturel). This PR includes: * The initial implementation of the new transactional repositories for the objects listed above. * Projections to populate the new relational tables from the existing event store. * Adjustments to the build and test process to accommodate the new backend structure. This is a work in progress and further changes will be made to complete the migration. --------- Co-authored-by: Copilot <[email protected]> Co-authored-by: Iraq Jaber <[email protected]> Co-authored-by: Iraq <[email protected]> Co-authored-by: Tim Möhlmann <[email protected]>
This commit is contained in:
Silvan
co-authored by
Copilot
Iraq Jaber
Iraq
Tim Möhlmann
Copilot
Iraq Jaber
Iraq
Tim Möhlmann
parent
8cc79c1376
commit
61cab8878e
@@ -0,0 +1,89 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"slices"
|
||||
"strings"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
"github.com/mitchellh/mapstructure"
|
||||
|
||||
"github.com/zitadel/zitadel/backend/v3/storage/database"
|
||||
)
|
||||
|
||||
var (
|
||||
_ database.Connector = (*Config)(nil)
|
||||
Name = "postgres"
|
||||
isMigrated bool
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
*pgxpool.Config
|
||||
*pgxpool.Pool
|
||||
|
||||
// Host string
|
||||
// Port int32
|
||||
// Database string
|
||||
// MaxOpenConns uint32
|
||||
// MaxIdleConns uint32
|
||||
// MaxConnLifetime time.Duration
|
||||
// MaxConnIdleTime time.Duration
|
||||
// User User
|
||||
// // Additional options to be appended as options=<Options>
|
||||
// // The value will be taken as is. Multiple options are space separated.
|
||||
// Options string
|
||||
|
||||
// configuredFields []string
|
||||
}
|
||||
|
||||
// Connect implements [database.Connector].
|
||||
func (c *Config) Connect(ctx context.Context) (database.Pool, error) {
|
||||
pool, err := c.getPool(ctx)
|
||||
if err != nil {
|
||||
return nil, wrapError(err)
|
||||
}
|
||||
if err = pool.Ping(ctx); err != nil {
|
||||
return nil, wrapError(err)
|
||||
}
|
||||
return &pgxPool{Pool: pool}, nil
|
||||
}
|
||||
|
||||
func (c *Config) getPool(ctx context.Context) (*pgxpool.Pool, error) {
|
||||
if c.Pool != nil {
|
||||
return c.Pool, nil
|
||||
}
|
||||
return pgxpool.NewWithConfig(ctx, c.Config)
|
||||
}
|
||||
|
||||
func NameMatcher(name string) bool {
|
||||
return slices.Contains([]string{"postgres", "pg"}, strings.ToLower(name))
|
||||
}
|
||||
|
||||
func DecodeConfig(input any) (database.Connector, error) {
|
||||
switch c := input.(type) {
|
||||
case string:
|
||||
config, err := pgxpool.ParseConfig(c)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &Config{Config: config}, nil
|
||||
case map[string]any:
|
||||
connector := new(Config)
|
||||
decoder, err := mapstructure.NewDecoder(&mapstructure.DecoderConfig{
|
||||
DecodeHook: mapstructure.StringToTimeDurationHookFunc(),
|
||||
WeaklyTypedInput: true,
|
||||
Result: connector,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err = decoder.Decode(c); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &Config{
|
||||
Config: &pgxpool.Config{},
|
||||
}, nil
|
||||
}
|
||||
return nil, errors.New("invalid configuration")
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"github.com/zitadel/zitadel/backend/v3/storage/database"
|
||||
"github.com/zitadel/zitadel/backend/v3/storage/database/dialect/postgres/migration"
|
||||
)
|
||||
|
||||
type pgxConn struct {
|
||||
*pgxpool.Conn
|
||||
}
|
||||
|
||||
var _ database.Client = (*pgxConn)(nil)
|
||||
|
||||
// Release implements [database.Client].
|
||||
func (c *pgxConn) Release(_ context.Context) error {
|
||||
c.Conn.Release()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Begin implements [database.Client].
|
||||
func (c *pgxConn) Begin(ctx context.Context, opts *database.TransactionOptions) (database.Transaction, error) {
|
||||
tx, err := c.BeginTx(ctx, transactionOptionsToPgx(opts))
|
||||
if err != nil {
|
||||
return nil, wrapError(err)
|
||||
}
|
||||
return &pgxTx{tx}, nil
|
||||
}
|
||||
|
||||
// Query implements sql.Client.
|
||||
// Subtle: this method shadows the method (*Conn).Query of pgxConn.Conn.
|
||||
func (c *pgxConn) Query(ctx context.Context, sql string, args ...any) (database.Rows, error) {
|
||||
rows, err := c.Conn.Query(ctx, sql, args...)
|
||||
if err != nil {
|
||||
return nil, wrapError(err)
|
||||
}
|
||||
return &Rows{rows}, nil
|
||||
}
|
||||
|
||||
// QueryRow implements sql.Client.
|
||||
// Subtle: this method shadows the method (*Conn).QueryRow of pgxConn.Conn.
|
||||
func (c *pgxConn) QueryRow(ctx context.Context, sql string, args ...any) database.Row {
|
||||
return &Row{c.Conn.QueryRow(ctx, sql, args...)}
|
||||
}
|
||||
|
||||
// Exec implements [database.Pool].
|
||||
// Subtle: this method shadows the method (Pool).Exec of pgxPool.Pool.
|
||||
func (c *pgxConn) Exec(ctx context.Context, sql string, args ...any) (int64, error) {
|
||||
res, err := c.Conn.Exec(ctx, sql, args...)
|
||||
if err != nil {
|
||||
return 0, wrapError(err)
|
||||
}
|
||||
return res.RowsAffected(), nil
|
||||
}
|
||||
|
||||
// Migrate implements [database.Migrator].
|
||||
func (c *pgxConn) Migrate(ctx context.Context) error {
|
||||
if isMigrated {
|
||||
return nil
|
||||
}
|
||||
err := migration.Migrate(ctx, c.Conn.Conn())
|
||||
isMigrated = err == nil
|
||||
return wrapError(err)
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
// pgxpool v5 implementation of the interfaces defined in the database package.
|
||||
package postgres
|
||||
@@ -0,0 +1,50 @@
|
||||
// embedded is used for testing purposes
|
||||
package embedded
|
||||
|
||||
import (
|
||||
"net"
|
||||
"os"
|
||||
|
||||
embeddedpostgres "github.com/fergusstrange/embedded-postgres"
|
||||
"github.com/zitadel/logging"
|
||||
|
||||
"github.com/zitadel/zitadel/backend/v3/storage/database"
|
||||
"github.com/zitadel/zitadel/backend/v3/storage/database/dialect/postgres"
|
||||
)
|
||||
|
||||
// StartEmbedded starts an embedded postgres v16 instance and returns a database connector and a stop function
|
||||
// the database is started on a random port and data are stored in a temporary directory
|
||||
// its used for testing purposes only
|
||||
func StartEmbedded() (connector database.Connector, stop func(), err error) {
|
||||
path, err := os.MkdirTemp("", "zitadel-embedded-postgres-*")
|
||||
logging.OnError(err).Fatal("unable to create temp dir")
|
||||
|
||||
port, close := getPort()
|
||||
|
||||
config := embeddedpostgres.DefaultConfig().Version(embeddedpostgres.V16).Port(uint32(port)).RuntimePath(path)
|
||||
embedded := embeddedpostgres.NewDatabase(config)
|
||||
|
||||
close()
|
||||
err = embedded.Start()
|
||||
logging.OnError(err).Fatal("unable to start db")
|
||||
|
||||
connector, err = postgres.DecodeConfig(config.GetConnectionURL())
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
return connector, func() {
|
||||
logging.OnError(embedded.Stop()).Error("unable to stop db")
|
||||
}, nil
|
||||
}
|
||||
|
||||
// getPort returns a free port and locks it until close is called
|
||||
func getPort() (port uint16, close func()) {
|
||||
l, err := net.Listen("tcp", ":0")
|
||||
logging.OnError(err).Fatal("unable to get port")
|
||||
port = uint16(l.Addr().(*net.TCPAddr).Port)
|
||||
logging.WithFields("port", port).Info("Port is available")
|
||||
return port, func() {
|
||||
logging.OnError(l.Close()).Error("unable to close port listener")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"errors"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
|
||||
"github.com/zitadel/zitadel/backend/v3/storage/database"
|
||||
)
|
||||
|
||||
func wrapError(err error) error {
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return database.NewNoRowFoundError(err)
|
||||
}
|
||||
var pgxErr *pgconn.PgError
|
||||
if !errors.As(err, &pgxErr) {
|
||||
return database.NewUnknownError(err)
|
||||
}
|
||||
switch pgxErr.Code {
|
||||
// 23514: check_violation - A value violates a CHECK constraint.
|
||||
case "23514":
|
||||
return database.NewCheckError(pgxErr.TableName, pgxErr.ConstraintName, pgxErr)
|
||||
// 23505: unique_violation - A value violates a UNIQUE constraint.
|
||||
case "23505":
|
||||
return database.NewUniqueError(pgxErr.TableName, pgxErr.ConstraintName, pgxErr)
|
||||
// 23503: foreign_key_violation - A value violates a foreign key constraint.
|
||||
case "23503":
|
||||
return database.NewForeignKeyError(pgxErr.TableName, pgxErr.ConstraintName, pgxErr)
|
||||
// 23502: not_null_violation - A value violates a NOT NULL constraint.
|
||||
case "23502":
|
||||
return database.NewNotNullError(pgxErr.TableName, pgxErr.ConstraintName, pgxErr)
|
||||
}
|
||||
return database.NewUnknownError(err)
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
package migration
|
||||
|
||||
import (
|
||||
_ "embed"
|
||||
)
|
||||
|
||||
var (
|
||||
//go:embed 001_instance_table/up.sql
|
||||
up001InstanceTable string
|
||||
//go:embed 001_instance_table/down.sql
|
||||
down001InstanceTable string
|
||||
)
|
||||
|
||||
func init() {
|
||||
registerSQLMigration(1, up001InstanceTable, down001InstanceTable)
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
DROP TABLE zitadel.instances;
|
||||
@@ -0,0 +1,25 @@
|
||||
CREATE TABLE IF NOT EXISTS zitadel.instances(
|
||||
id TEXT NOT NULL CHECK (id <> '') PRIMARY KEY,
|
||||
name TEXT NOT NULL CHECK (name <> ''),
|
||||
default_org_id TEXT, -- NOT NULL,
|
||||
iam_project_id TEXT, -- NOT NULL,
|
||||
console_client_id TEXT, -- NOT NULL,
|
||||
console_app_id TEXT, -- NOT NULL,
|
||||
default_language TEXT, -- NOT NULL,
|
||||
created_at TIMESTAMPTZ DEFAULT NOW() NOT NULL,
|
||||
updated_at TIMESTAMPTZ DEFAULT NOW() NOT NULL
|
||||
);
|
||||
|
||||
CREATE OR REPLACE FUNCTION zitadel.set_updated_at()
|
||||
RETURNS TRIGGER AS $$
|
||||
BEGIN
|
||||
NEW.updated_at := NOW();
|
||||
RETURN NEW;
|
||||
END;
|
||||
$$ LANGUAGE plpgsql;
|
||||
|
||||
CREATE TRIGGER trigger_set_updated_at
|
||||
BEFORE UPDATE ON zitadel.instances
|
||||
FOR EACH ROW
|
||||
WHEN (OLD.updated_at IS NOT DISTINCT FROM NEW.updated_at)
|
||||
EXECUTE FUNCTION zitadel.set_updated_at();
|
||||
@@ -0,0 +1,16 @@
|
||||
package migration
|
||||
|
||||
import (
|
||||
_ "embed"
|
||||
)
|
||||
|
||||
var (
|
||||
//go:embed 002_organization_table/up.sql
|
||||
up002OrganizationTable string
|
||||
//go:embed 002_organization_table/down.sql
|
||||
down002OrganizationTable string
|
||||
)
|
||||
|
||||
func init() {
|
||||
registerSQLMigration(2, up002OrganizationTable, down002OrganizationTable)
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
DROP TABLE zitadel.organizations;
|
||||
DROP Type zitadel.organization_state;
|
||||
@@ -0,0 +1,24 @@
|
||||
CREATE TYPE zitadel.organization_state AS ENUM (
|
||||
'active',
|
||||
'inactive'
|
||||
);
|
||||
|
||||
CREATE TABLE zitadel.organizations(
|
||||
id TEXT NOT NULL CHECK (id <> ''),
|
||||
name TEXT NOT NULL CHECK (name <> ''),
|
||||
instance_id TEXT NOT NULL REFERENCES zitadel.instances (id) ON DELETE CASCADE,
|
||||
state zitadel.organization_state NOT NULL,
|
||||
created_at TIMESTAMPTZ DEFAULT NOW() NOT NULL,
|
||||
updated_at TIMESTAMPTZ DEFAULT NOW() NOT NULL,
|
||||
|
||||
PRIMARY KEY (instance_id, id)
|
||||
);
|
||||
|
||||
CREATE UNIQUE INDEX org_unique_instance_id_name_idx
|
||||
ON zitadel.organizations (instance_id, name);
|
||||
|
||||
CREATE TRIGGER trigger_set_updated_at
|
||||
BEFORE UPDATE ON zitadel.organizations
|
||||
FOR EACH ROW
|
||||
WHEN (OLD.updated_at IS NOT DISTINCT FROM NEW.updated_at)
|
||||
EXECUTE FUNCTION zitadel.set_updated_at();
|
||||
@@ -0,0 +1,16 @@
|
||||
package migration
|
||||
|
||||
import (
|
||||
_ "embed"
|
||||
)
|
||||
|
||||
var (
|
||||
//go:embed 003_domains_table/up.sql
|
||||
up003DomainsTable string
|
||||
//go:embed 003_domains_table/down.sql
|
||||
down003DomainsTable string
|
||||
)
|
||||
|
||||
func init() {
|
||||
registerSQLMigration(3, up003DomainsTable, down003DomainsTable)
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
DROP TABLE IF EXISTS zitadel.instance_domains;
|
||||
DROP TABLE IF EXISTS zitadel.org_domains;
|
||||
DROP TYPE IF EXISTS zitadel.domain_type;
|
||||
DROP TYPE IF EXISTS zitadel.domain_validation_type;
|
||||
DROP FUNCTION IF EXISTS zitadel.check_verified_org_domain();
|
||||
DROP FUNCTION IF EXISTS zitadel.ensure_single_primary_instance_domain();
|
||||
@@ -0,0 +1,137 @@
|
||||
CREATE TYPE zitadel.domain_validation_type AS ENUM (
|
||||
'dns'
|
||||
, 'http'
|
||||
);
|
||||
|
||||
CREATE TYPE zitadel.domain_type AS ENUM (
|
||||
'custom'
|
||||
, 'trusted'
|
||||
);
|
||||
|
||||
CREATE TABLE zitadel.instance_domains(
|
||||
instance_id TEXT NOT NULL
|
||||
, domain TEXT NOT NULL CHECK (LENGTH(domain) BETWEEN 1 AND 255)
|
||||
, is_primary BOOLEAN
|
||||
, is_generated BOOLEAN
|
||||
, type zitadel.domain_type NOT NULL
|
||||
|
||||
, created_at TIMESTAMPTZ DEFAULT NOW() NOT NULL
|
||||
, updated_at TIMESTAMPTZ DEFAULT NOW() NOT NULL
|
||||
|
||||
, PRIMARY KEY (domain)
|
||||
|
||||
, FOREIGN KEY (instance_id) REFERENCES zitadel.instances(id) ON DELETE CASCADE
|
||||
|
||||
, CONSTRAINT primary_cannot_be_trusted CHECK (is_primary IS NULL OR type != 'trusted')
|
||||
, CONSTRAINT generated_cannot_be_trusted CHECK (is_generated IS NULL OR type != 'trusted')
|
||||
, CONSTRAINT custom_values_set CHECK ((is_primary IS NOT NULL AND is_generated IS NOT NULL) OR type != 'custom')
|
||||
);
|
||||
|
||||
CREATE INDEX idx_instance_domain_instance ON zitadel.instance_domains(instance_id);
|
||||
|
||||
CREATE TABLE zitadel.org_domains(
|
||||
instance_id TEXT NOT NULL
|
||||
, org_id TEXT NOT NULL
|
||||
, domain TEXT NOT NULL CHECK (LENGTH(domain) BETWEEN 1 AND 255)
|
||||
, is_verified BOOLEAN NOT NULL DEFAULT FALSE
|
||||
, is_primary BOOLEAN NOT NULL DEFAULT FALSE
|
||||
, validation_type zitadel.domain_validation_type
|
||||
|
||||
, created_at TIMESTAMPTZ DEFAULT NOW() NOT NULL
|
||||
, updated_at TIMESTAMPTZ DEFAULT NOW() NOT NULL
|
||||
|
||||
, PRIMARY KEY (instance_id, org_id, domain)
|
||||
|
||||
, FOREIGN KEY (instance_id, org_id) REFERENCES zitadel.organizations(instance_id, id) ON DELETE CASCADE
|
||||
|
||||
, UNIQUE (instance_id, org_id, domain)
|
||||
);
|
||||
|
||||
CREATE INDEX idx_org_domain ON zitadel.org_domains(instance_id, domain);
|
||||
|
||||
-- Trigger to update the updated_at timestamp on instance_domains
|
||||
CREATE TRIGGER trg_set_updated_at_instance_domains
|
||||
BEFORE UPDATE ON zitadel.instance_domains
|
||||
FOR EACH ROW
|
||||
WHEN (OLD.updated_at IS NOT DISTINCT FROM NEW.updated_at)
|
||||
EXECUTE FUNCTION zitadel.set_updated_at();
|
||||
|
||||
-- Trigger to update the updated_at timestamp on org_domains
|
||||
CREATE TRIGGER trg_set_updated_at_org_domains
|
||||
BEFORE UPDATE ON zitadel.org_domains
|
||||
FOR EACH ROW
|
||||
WHEN (OLD.updated_at IS NOT DISTINCT FROM NEW.updated_at)
|
||||
EXECUTE FUNCTION zitadel.set_updated_at();
|
||||
|
||||
-- Function to check for already verified org domains
|
||||
CREATE OR REPLACE FUNCTION zitadel.check_verified_org_domain()
|
||||
RETURNS TRIGGER AS $$
|
||||
BEGIN
|
||||
-- Check if there's already a verified domain within this instance (excluding the current record being updated)
|
||||
IF EXISTS (
|
||||
SELECT 1
|
||||
FROM zitadel.org_domains
|
||||
WHERE instance_id = NEW.instance_id
|
||||
AND domain = NEW.domain
|
||||
AND is_verified = TRUE
|
||||
AND (TG_OP = 'INSERT' OR (org_id != NEW.org_id))
|
||||
) THEN
|
||||
RAISE EXCEPTION 'org domain is already taken';
|
||||
END IF;
|
||||
|
||||
RETURN NEW;
|
||||
END;
|
||||
$$ LANGUAGE plpgsql;
|
||||
|
||||
-- Trigger to enforce verified domain constraint on org_domains
|
||||
CREATE TRIGGER trg_check_verified_org_domain
|
||||
BEFORE INSERT OR UPDATE ON zitadel.org_domains
|
||||
FOR EACH ROW
|
||||
WHEN (NEW.is_verified IS TRUE)
|
||||
EXECUTE FUNCTION zitadel.check_verified_org_domain();
|
||||
|
||||
-- Function to ensure only one primary domain per instance in instance_domains
|
||||
CREATE OR REPLACE FUNCTION zitadel.ensure_single_primary_instance_domain()
|
||||
RETURNS TRIGGER AS $$
|
||||
BEGIN
|
||||
-- If setting this domain as primary, update all other domains in the same instance to non-primary
|
||||
UPDATE zitadel.instance_domains
|
||||
SET is_primary = FALSE, updated_at = NOW()
|
||||
WHERE instance_id = NEW.instance_id
|
||||
AND domain != NEW.domain
|
||||
AND is_primary = TRUE
|
||||
AND type = 'custom';
|
||||
|
||||
RETURN NEW;
|
||||
END;
|
||||
$$ LANGUAGE plpgsql;
|
||||
|
||||
-- Trigger to enforce single primary domain constraint on instance_domains
|
||||
CREATE TRIGGER trg_ensure_single_primary_instance_domain
|
||||
BEFORE INSERT OR UPDATE ON zitadel.instance_domains
|
||||
FOR EACH ROW
|
||||
WHEN (NEW.is_primary IS TRUE)
|
||||
EXECUTE FUNCTION zitadel.ensure_single_primary_instance_domain();
|
||||
|
||||
-- Function to ensure only one primary domain per organization in org_domains
|
||||
CREATE OR REPLACE FUNCTION zitadel.ensure_single_primary_org_domain()
|
||||
RETURNS TRIGGER AS $$
|
||||
BEGIN
|
||||
-- If setting this domain as primary, update all other domains in the same organization to non-primary
|
||||
UPDATE zitadel.org_domains
|
||||
SET is_primary = FALSE, updated_at = NOW()
|
||||
WHERE instance_id = NEW.instance_id
|
||||
AND org_id = NEW.org_id
|
||||
AND domain != NEW.domain
|
||||
AND is_primary = TRUE;
|
||||
|
||||
RETURN NEW;
|
||||
END;
|
||||
$$ LANGUAGE plpgsql;
|
||||
|
||||
-- Trigger to enforce single primary domain constraint on org_domains
|
||||
CREATE TRIGGER trg_ensure_single_primary_org_domain
|
||||
BEFORE INSERT OR UPDATE ON zitadel.org_domains
|
||||
FOR EACH ROW
|
||||
WHEN (NEW.is_primary IS TRUE)
|
||||
EXECUTE FUNCTION zitadel.ensure_single_primary_org_domain();
|
||||
@@ -0,0 +1,13 @@
|
||||
// This package contains the migration logic for the PostgreSQL dialect.
|
||||
// It uses the [github.com/jackc/tern/v2/migrate] package to handle the migration process.
|
||||
//
|
||||
// **Developer Note**:
|
||||
//
|
||||
// Each migration MUST be registered in an init function.
|
||||
// Create a go file for each migration with the sequence of the migration as prefix and some descriptive name.
|
||||
// The file name MUST be in the format <sequence>_<name>.go.
|
||||
// Each migration SHOULD provide an up and down migration.
|
||||
// Prefer to write SQL statements instead of funcs if it is reasonable.
|
||||
// To keep the folder clean create a folder to store the sql files with the following format: <sequence>_<name>/{up/down}.sql.
|
||||
// And use the go embed directive to embed the sql files.
|
||||
package migration
|
||||
@@ -0,0 +1,33 @@
|
||||
package migration
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/tern/v2/migrate"
|
||||
)
|
||||
|
||||
var migrations []*migrate.Migration
|
||||
|
||||
func Migrate(ctx context.Context, conn *pgx.Conn) error {
|
||||
// we need to ensure that the schema exists before we can run the migration
|
||||
// because creating the migrations table already required the schema
|
||||
_, err := conn.Exec(ctx, "CREATE SCHEMA IF NOT EXISTS zitadel")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
migrator, err := migrate.NewMigrator(ctx, conn, "zitadel.migrations")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
migrator.Migrations = migrations
|
||||
return migrator.Migrate(ctx)
|
||||
}
|
||||
|
||||
func registerSQLMigration(sequence int32, up, down string) {
|
||||
migrations = append(migrations, &migrate.Migration{
|
||||
Sequence: sequence,
|
||||
UpSQL: up,
|
||||
DownSQL: down,
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
package migration_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/muhlemmer/gu"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/zitadel/zitadel/backend/v3/storage/database"
|
||||
"github.com/zitadel/zitadel/backend/v3/storage/database/dialect/postgres/embedded"
|
||||
)
|
||||
|
||||
func TestMigrate(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
stmt string
|
||||
args []any
|
||||
res []any
|
||||
}{
|
||||
{
|
||||
name: "schema",
|
||||
stmt: "SELECT EXISTS(SELECT 1 FROM information_schema.schemata where schema_name = 'zitadel') ;",
|
||||
res: []any{true},
|
||||
},
|
||||
{
|
||||
name: "001",
|
||||
stmt: "SELECT EXISTS(SELECT 1 FROM pg_catalog.pg_tables WHERE schemaname = 'zitadel' and tablename=$1)",
|
||||
args: []any{"instances"},
|
||||
res: []any{true},
|
||||
},
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
connector, stop, err := embedded.StartEmbedded()
|
||||
require.NoError(t, err, "failed to start embedded postgres")
|
||||
defer stop()
|
||||
|
||||
client, err := connector.Connect(ctx)
|
||||
require.NoError(t, err, "failed to connect to embedded postgres")
|
||||
|
||||
err = client.(database.Migrator).Migrate(ctx)
|
||||
require.NoError(t, err, "failed to execute migration steps")
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
got := make([]any, len(tt.res))
|
||||
for i := range got {
|
||||
got[i] = new(any)
|
||||
tt.res[i] = gu.Ptr(tt.res[i])
|
||||
}
|
||||
|
||||
require.NoError(t, client.QueryRow(ctx, tt.stmt, tt.args...).Scan(got...), "failed to execute check query")
|
||||
|
||||
assert.Equal(t, tt.res, got, "query result does not match")
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,100 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"github.com/zitadel/zitadel/backend/v3/storage/database"
|
||||
"github.com/zitadel/zitadel/backend/v3/storage/database/dialect/postgres/migration"
|
||||
)
|
||||
|
||||
type pgxPool struct {
|
||||
*pgxpool.Pool
|
||||
}
|
||||
|
||||
var _ database.Pool = (*pgxPool)(nil)
|
||||
|
||||
func PGxPool(pool *pgxpool.Pool) *pgxPool {
|
||||
return &pgxPool{
|
||||
Pool: pool,
|
||||
}
|
||||
}
|
||||
|
||||
// Acquire implements [database.Pool].
|
||||
func (c *pgxPool) Acquire(ctx context.Context) (database.Client, error) {
|
||||
conn, err := c.Pool.Acquire(ctx)
|
||||
if err != nil {
|
||||
return nil, wrapError(err)
|
||||
}
|
||||
return &pgxConn{Conn: conn}, nil
|
||||
}
|
||||
|
||||
// Query implements [database.Pool].
|
||||
// Subtle: this method shadows the method (Pool).Query of pgxPool.Pool.
|
||||
func (c *pgxPool) Query(ctx context.Context, sql string, args ...any) (database.Rows, error) {
|
||||
rows, err := c.Pool.Query(ctx, sql, args...)
|
||||
if err != nil {
|
||||
return nil, wrapError(err)
|
||||
}
|
||||
return &Rows{rows}, nil
|
||||
}
|
||||
|
||||
// QueryRow implements [database.Pool].
|
||||
// Subtle: this method shadows the method (Pool).QueryRow of pgxPool.Pool.
|
||||
func (c *pgxPool) QueryRow(ctx context.Context, sql string, args ...any) database.Row {
|
||||
return &Row{c.Pool.QueryRow(ctx, sql, args...)}
|
||||
}
|
||||
|
||||
// Exec implements [database.Pool].
|
||||
// Subtle: this method shadows the method (Pool).Exec of pgxPool.Pool.
|
||||
func (c *pgxPool) Exec(ctx context.Context, sql string, args ...any) (int64, error) {
|
||||
res, err := c.Pool.Exec(ctx, sql, args...)
|
||||
if err != nil {
|
||||
return 0, wrapError(err)
|
||||
}
|
||||
return res.RowsAffected(), nil
|
||||
}
|
||||
|
||||
// Begin implements [database.Pool].
|
||||
func (c *pgxPool) Begin(ctx context.Context, opts *database.TransactionOptions) (database.Transaction, error) {
|
||||
tx, err := c.BeginTx(ctx, transactionOptionsToPgx(opts))
|
||||
if err != nil {
|
||||
return nil, wrapError(err)
|
||||
}
|
||||
return &pgxTx{tx}, nil
|
||||
}
|
||||
|
||||
// Close implements [database.Pool].
|
||||
func (c *pgxPool) Close(_ context.Context) error {
|
||||
c.Pool.Close()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Migrate implements [database.Migrator].
|
||||
func (c *pgxPool) Migrate(ctx context.Context) error {
|
||||
if isMigrated {
|
||||
return nil
|
||||
}
|
||||
|
||||
client, err := c.Pool.Acquire(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = migration.Migrate(ctx, client.Conn())
|
||||
isMigrated = err == nil
|
||||
return wrapError(err)
|
||||
}
|
||||
|
||||
// Migrate implements [database.PoolTest].
|
||||
func (c *pgxPool) MigrateTest(ctx context.Context) error {
|
||||
client, err := c.Pool.Acquire(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = migration.Migrate(ctx, client.Conn())
|
||||
isMigrated = err == nil
|
||||
return err
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"github.com/georgysavva/scany/v2/pgxscan"
|
||||
"github.com/jackc/pgx/v5"
|
||||
|
||||
"github.com/zitadel/zitadel/backend/v3/storage/database"
|
||||
)
|
||||
|
||||
var (
|
||||
_ database.Rows = (*Rows)(nil)
|
||||
_ database.CollectableRows = (*Rows)(nil)
|
||||
_ database.Row = (*Row)(nil)
|
||||
)
|
||||
|
||||
type Row struct{ pgx.Row }
|
||||
|
||||
// Scan implements [database.Row].
|
||||
// Subtle: this method shadows the method ([pgx.Row]).Scan of Row.Row.
|
||||
func (r *Row) Scan(dest ...any) error {
|
||||
return wrapError(r.Row.Scan(dest...))
|
||||
}
|
||||
|
||||
type Rows struct{ pgx.Rows }
|
||||
|
||||
// Err implements [database.Rows].
|
||||
// Subtle: this method shadows the method ([pgx.Rows]).Err of Rows.Rows.
|
||||
func (r *Rows) Err() error {
|
||||
return wrapError(r.Rows.Err())
|
||||
}
|
||||
|
||||
func (r *Rows) Scan(dest ...any) error {
|
||||
return wrapError(r.Rows.Scan(dest...))
|
||||
}
|
||||
|
||||
// Collect implements [database.CollectableRows].
|
||||
// See [this page](https://github.com/georgysavva/scany/blob/master/dbscan/doc.go#L8) for additional details.
|
||||
func (r *Rows) Collect(dest any) (err error) {
|
||||
defer func() {
|
||||
closeErr := r.Close()
|
||||
if err == nil {
|
||||
err = closeErr
|
||||
}
|
||||
}()
|
||||
return wrapError(pgxscan.ScanAll(dest, r.Rows))
|
||||
}
|
||||
|
||||
// CollectFirst implements [database.CollectableRows].
|
||||
// See [this page](https://github.com/georgysavva/scany/blob/master/dbscan/doc.go#L8) for additional details.
|
||||
func (r *Rows) CollectFirst(dest any) (err error) {
|
||||
defer func() {
|
||||
closeErr := r.Close()
|
||||
if err == nil {
|
||||
err = closeErr
|
||||
}
|
||||
}()
|
||||
return wrapError(pgxscan.ScanRow(dest, r.Rows))
|
||||
}
|
||||
|
||||
// CollectExactlyOneRow implements [database.CollectableRows].
|
||||
// See [this page](https://github.com/georgysavva/scany/blob/master/dbscan/doc.go#L8) for additional details.
|
||||
func (r *Rows) CollectExactlyOneRow(dest any) (err error) {
|
||||
defer func() {
|
||||
closeErr := r.Close()
|
||||
if err == nil {
|
||||
err = closeErr
|
||||
}
|
||||
}()
|
||||
return wrapError(pgxscan.ScanOne(dest, r.Rows))
|
||||
}
|
||||
|
||||
// Close implements [database.Rows].
|
||||
// Subtle: this method shadows the method (Rows).Close of Rows.Rows.
|
||||
func (r *Rows) Close() error {
|
||||
r.Rows.Close()
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,107 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
|
||||
"github.com/zitadel/zitadel/backend/v3/storage/database"
|
||||
)
|
||||
|
||||
type pgxTx struct{ pgx.Tx }
|
||||
|
||||
var _ database.Transaction = (*pgxTx)(nil)
|
||||
|
||||
// Commit implements [database.Transaction].
|
||||
func (tx *pgxTx) Commit(ctx context.Context) error {
|
||||
err := tx.Tx.Commit(ctx)
|
||||
return wrapError(err)
|
||||
}
|
||||
|
||||
// Rollback implements [database.Transaction].
|
||||
func (tx *pgxTx) Rollback(ctx context.Context) error {
|
||||
err := tx.Tx.Rollback(ctx)
|
||||
return wrapError(err)
|
||||
}
|
||||
|
||||
// End implements [database.Transaction].
|
||||
func (tx *pgxTx) End(ctx context.Context, err error) error {
|
||||
if err != nil {
|
||||
rollbackErr := tx.Rollback(ctx)
|
||||
if rollbackErr != nil {
|
||||
err = errors.Join(err, rollbackErr)
|
||||
}
|
||||
return err
|
||||
}
|
||||
return tx.Commit(ctx)
|
||||
}
|
||||
|
||||
// Query implements [database.Transaction].
|
||||
// Subtle: this method shadows the method (Tx).Query of pgxTx.Tx.
|
||||
func (tx *pgxTx) Query(ctx context.Context, sql string, args ...any) (database.Rows, error) {
|
||||
rows, err := tx.Tx.Query(ctx, sql, args...)
|
||||
if err != nil {
|
||||
return nil, wrapError(err)
|
||||
}
|
||||
return &Rows{rows}, nil
|
||||
}
|
||||
|
||||
// QueryRow implements [database.Transaction].
|
||||
// Subtle: this method shadows the method (Tx).QueryRow of pgxTx.Tx.
|
||||
func (tx *pgxTx) QueryRow(ctx context.Context, sql string, args ...any) database.Row {
|
||||
return &Row{tx.Tx.QueryRow(ctx, sql, args...)}
|
||||
}
|
||||
|
||||
// Exec implements [database.Transaction].
|
||||
// Subtle: this method shadows the method (Pool).Exec of pgxPool.Pool.
|
||||
func (tx *pgxTx) Exec(ctx context.Context, sql string, args ...any) (int64, error) {
|
||||
res, err := tx.Tx.Exec(ctx, sql, args...)
|
||||
if err != nil {
|
||||
return 0, wrapError(err)
|
||||
}
|
||||
return res.RowsAffected(), nil
|
||||
}
|
||||
|
||||
// Begin implements [database.Transaction].
|
||||
// As postgres does not support nested transactions we use savepoints to emulate them.
|
||||
func (tx *pgxTx) Begin(ctx context.Context) (database.Transaction, error) {
|
||||
savepoint, err := tx.Tx.Begin(ctx)
|
||||
if err != nil {
|
||||
return nil, wrapError(err)
|
||||
}
|
||||
return &pgxTx{savepoint}, nil
|
||||
}
|
||||
|
||||
func transactionOptionsToPgx(opts *database.TransactionOptions) pgx.TxOptions {
|
||||
if opts == nil {
|
||||
return pgx.TxOptions{}
|
||||
}
|
||||
|
||||
return pgx.TxOptions{
|
||||
IsoLevel: isolationToPgx(opts.IsolationLevel),
|
||||
AccessMode: accessModeToPgx(opts.AccessMode),
|
||||
}
|
||||
}
|
||||
|
||||
func isolationToPgx(isolation database.IsolationLevel) pgx.TxIsoLevel {
|
||||
switch isolation {
|
||||
case database.IsolationLevelSerializable:
|
||||
return pgx.Serializable
|
||||
case database.IsolationLevelReadCommitted:
|
||||
return pgx.ReadCommitted
|
||||
default:
|
||||
return pgx.Serializable
|
||||
}
|
||||
}
|
||||
|
||||
func accessModeToPgx(accessMode database.AccessMode) pgx.TxAccessMode {
|
||||
switch accessMode {
|
||||
case database.AccessModeReadWrite:
|
||||
return pgx.ReadWrite
|
||||
case database.AccessModeReadOnly:
|
||||
return pgx.ReadOnly
|
||||
default:
|
||||
return pgx.ReadWrite
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user