feat(relational): org metadata (#10761)

This pull request introduces a new feature that allows adding, updating,
and querying metadata for organizations. The changes are primarily in
the backend and include new database tables, repositories, and domain
logic to support organization metadata.

## Changes

* New `org_metadata` table: A new table `zitadel.org_metadata` is
introduced to store metadata for organizations. It includes columns for
`instance_id`, `org_id`, `key`, and `value`. The `value` is stored as a
`BYTEA` type to allow for flexible data storage.
* New `OrganizationMetadataRepository`: A new repository
`OrganizationMetadataRepository` is created to handle all database
operations for organization metadata. It provides methods to `Get`,
`List`, `Set`, and `Remove` metadata.
* New `org_metadata_relational_projection`: A new projection
`org_metadata_relational_projection` is added to update the
`zitadel.org_metadata` table based on events. It handles `MetadataSet`,
`MetadataRemoved`, and `MetadataRemovedAll` events.
* Updated `OrganizationRepository`: The `OrganizationRepository` is
updated to support loading organization metadata. A new method
`LoadMetadata` is added to enable joining the `org_metadata` table when
querying for organizations.
* Updated Organization domain: The Organization domain model is updated
to include a new field `Metadata` of type `[]*OrganizationMetadata`.

## Additional Info

* Extensible Design: The new metadata feature is designed to be
extensible, allowing for future enhancements such as indexing on
specific JSON fields within the `value` column.
* closes https://github.com/zitadel/zitadel/issues/10206
* closes https://github.com/zitadel/zitadel/issues/10214

backend/v3/domain/metadata.go

@@ -0,0 +1,42 @@
+package domain
+
+import (
+	"time"
+
+	"github.com/zitadel/zitadel/backend/v3/storage/database"
+)
+
+type Metadata struct {
+	InstanceID string `json:"instanceId,omitempty" db:"instance_id"`
+	Key        string `json:"key,omitempty" db:"key"`
+	// Value is a byte slice that might be json encoded.
+	// the API does not require json encoding so we keep it as a byte slice here.
+	Value []byte `json:"value,omitempty" db:"value"`
+
+	CreatedAt time.Time `json:"createdAt,omitzero" db:"created_at"`
+	UpdatedAt time.Time `json:"updatedAt,omitzero" db:"updated_at"`
+}
+
+type MetadataColumns interface {
+	// PrimaryKeyColumns returns the columns for the primary key.
+	PrimaryKeyColumns() []database.Column
+	// InstanceIDColumn returns the column for the instance id field.
+	InstanceIDColumn() database.Column
+	// KeyColumn returns the column for the key field.
+	KeyColumn() database.Column
+	// ValueColumn returns the column for the value field.
+	ValueColumn() database.Column
+	// CreatedAtColumn returns the column for the created at field.
+	CreatedAtColumn() database.Column
+	// UpdatedAtColumn returns the column for the updated at field.
+	UpdatedAtColumn() database.Column
+}
+
+type MetadataConditions interface {
+	// InstanceIDCondition returns a filter on the instance id field.
+	InstanceIDCondition(instanceID string) database.Condition
+	// KeyCondition returns a filter on the key field.
+	KeyCondition(op database.TextOperation, key string) database.Condition
+	// ValueCondition returns a filter on the value field.
+	ValueCondition(op database.BytesOperation, value []byte) database.Condition
+}

backend/v3/domain/organization.go

@@ -23,7 +23,8 @@ type Organization struct {
 	CreatedAt  time.Time `json:"createdAt,omitzero" db:"created_at"`
 	UpdatedAt  time.Time `json:"updatedAt,omitzero" db:"updated_at"`
 
-	Domains []*OrganizationDomain `json:"domains,omitempty" db:"-"` // domains need to be handled separately
+	Domains  []*OrganizationDomain   `json:"domains,omitempty" db:"-"`  // domains need to be handled separately
+	Metadata []*OrganizationMetadata `json:"metadata,omitempty" db:"-"` // metadata need to be handled separately
 }
 
 // organizationColumns define all the columns of the instance table.
@@ -54,6 +55,8 @@ type organizationConditions interface {
 	StateCondition(state OrgState) database.Condition
 	// ExistsDomain returns a filter on the organizations domains.
 	ExistsDomain(cond database.Condition) database.Condition
+	// ExistsMetadata returns a filter on the organizations metadata.
+	ExistsMetadata(cond database.Condition) database.Condition
 }
 
 // organizationChanges define all the changes for the instance table.
@@ -80,6 +83,9 @@ type OrganizationRepository interface {
 	// LoadDomains loads the domains of the given organizations.
 	// If it is called the [Organization].Domains field will be set on future calls to Get or List.
 	LoadDomains() OrganizationRepository
+	// LoadMetadata loads the metadata of the given organizations.
+	// If it is called the [Organization].Metadata field will be set on future calls to Get or List.
+	LoadMetadata() OrganizationRepository
 }
 
 type CreateOrganization struct {

backend/v3/domain/organization_metadata.go

@@ -0,0 +1,45 @@
+package domain
+
+import (
+	"context"
+
+	"github.com/zitadel/zitadel/backend/v3/storage/database"
+)
+
+type OrganizationMetadata struct {
+	Metadata
+	OrganizationID string `json:"orgId,omitempty" db:"organization_id"`
+}
+
+type organizationMetadataColumns interface {
+	MetadataColumns
+	// OrganizationIDColumn returns the column for the org id field.
+	OrganizationIDColumn() database.Column
+}
+
+type organizationMetadataConditions interface {
+	MetadataConditions
+	// PrimaryKeyCondition returns the condition for the primary key fields.
+	PrimaryKeyCondition(instanceID, orgID, key string) database.Condition
+	// OrganizationIDCondition returns a filter on the org id field.
+	OrganizationIDCondition(orgID string) database.Condition
+}
+
+type OrganizationMetadataRepository interface {
+	organizationMetadataColumns
+	organizationMetadataConditions
+
+	// Get returns one metadata based on the criteria.
+	// If none is found, it returns an error of type [database.ErrNotFound].
+	// If multiple were found, it returns an error of type [database.ErrMultipleRows].
+	Get(ctx context.Context, client database.QueryExecutor, opts ...database.QueryOption) (*OrganizationMetadata, error)
+	// List returns a list of metadata based on the criteria.
+	// If none are found, it returns an empty slice.
+	List(ctx context.Context, client database.QueryExecutor, opts ...database.QueryOption) ([]*OrganizationMetadata, error)
+
+	// Set sets the given metadata for the organization.
+	// If a metadata with the same key already exists, it will be overwritten, otherwise created.
+	Set(ctx context.Context, client database.QueryExecutor, metadata ...*OrganizationMetadata) error
+	// Remove removes a metadata from the organization.
+	Remove(ctx context.Context, client database.QueryExecutor, condition database.Condition) (int64, error)
+}

backend/v3/storage/database/dialect/postgres/error.go

@@ -21,6 +21,11 @@ func wrapError(err error) error {
 		return database.NewMultipleRowsFoundError(err)
 	}
 
+	var pgxErr *pgconn.PgError
+	if errors.As(err, &pgxErr) {
+		return wrapPgError(pgxErr)
+	}
+
 	// scany only exports its errors as strings
 	if strings.HasPrefix(err.Error(), "scany: expected 1 row, got: ") {
 		return database.NewMultipleRowsFoundError(err)
@@ -29,23 +34,24 @@ func wrapError(err error) error {
 		return database.NewScanError(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)
 }
+
+func wrapPgError(err *pgconn.PgError) error {
+	switch err.Code {
+	// 23514: check_violation - A value violates a CHECK constraint.
+	case "23514":
+		return database.NewCheckError(err.TableName, err.ConstraintName, err)
+	// 23505: unique_violation - A value violates a UNIQUE constraint.
+	case "23505":
+		return database.NewUniqueError(err.TableName, err.ConstraintName, err)
+	// 23503: foreign_key_violation - A value violates a foreign key constraint.
+	case "23503":
+		return database.NewForeignKeyError(err.TableName, err.ConstraintName, err)
+	// 23502: not_null_violation - A value violates a NOT NULL constraint.
+	case "23502":
+		return database.NewNotNullError(err.TableName, err.ConstraintName, err)
+	default:
+		return database.NewUnknownError(err)
+	}
+}

backend/v3/storage/database/dialect/postgres/migration/007_org_metadata_table.go

@@ -0,0 +1,16 @@
+package migration
+
+import (
+	_ "embed"
+)
+
+var (
+	//go:embed 007_org_metadata_table/up.sql
+	up007OrgMetadataTable string
+	//go:embed 007_org_metadata_table/down.sql
+	down007OrgMetadataTable string
+)
+
+func init() {
+	registerSQLMigration(7, up007OrgMetadataTable, down007OrgMetadataTable)
+}

backend/v3/storage/database/dialect/postgres/migration/007_org_metadata_table/down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS zitadel.organization_metadata;

backend/v3/storage/database/dialect/postgres/migration/007_org_metadata_table/up.sql

@@ -0,0 +1,28 @@
+CREATE TABLE zitadel.organization_metadata (
+    instance_id TEXT NOT NULL
+    , organization_id TEXT NOT NULL
+    , key TEXT NOT NULL CHECK (key <> '')
+    , value BYTEA NOT NULL
+
+    , created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+    , updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+    
+    , PRIMARY KEY (instance_id, organization_id, key)
+    
+    , CONSTRAINT fk_organization_metadata_org FOREIGN KEY (instance_id, organization_id) REFERENCES zitadel.organizations (instance_id, id) ON DELETE CASCADE
+);
+
+CREATE INDEX idx_organization_metadata_key ON zitadel.organization_metadata (key);
+CREATE INDEX idx_organization_metadata_value ON zitadel.organization_metadata (sha256(value));
+
+-- TODO(adlerhurst): these indexes can currently not be used by Postgres, because of the type conversion
+-- the value can be a json but doesn't have to be.
+-- CREATE INDEX idx_organization_metadata_value_number ON zitadel.organization_metadata ((value::NUMERIC)) WHERE jsonb_typeof(value) = 'number';
+-- CREATE INDEX idx_organization_metadata_value_string ON zitadel.organization_metadata ((value#>>'{}')) WHERE jsonb_typeof(value) = 'string';
+-- CREATE INDEX idx_organization_metadata_value_boolean ON zitadel.organization_metadata ((value::BOOLEAN)) WHERE jsonb_typeof(value) = 'boolean';
+
+CREATE TRIGGER trg_set_updated_at_organization_metadata
+  BEFORE INSERT OR UPDATE ON zitadel.organization_metadata
+  FOR EACH ROW
+  WHEN (NEW.updated_at IS NULL)
+  EXECUTE FUNCTION zitadel.set_updated_at();

backend/v3/storage/database/dialect/sql/error.go

@@ -21,6 +21,11 @@ func wrapError(err error) error {
 		return database.NewNoRowFoundError(err)
 	}
 
+	var pgxErr *pgconn.PgError
+	if errors.As(err, &pgxErr) {
+		return wrapPgError(pgxErr)
+	}
+
 	// scany only exports its errors as strings
 	if strings.HasPrefix(err.Error(), "scany: expected 1 row, got: ") {
 		return database.NewMultipleRowsFoundError(err)
@@ -29,23 +34,24 @@ func wrapError(err error) error {
 		return database.NewScanError(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)
 }
+
+func wrapPgError(err *pgconn.PgError) error {
+	switch err.Code {
+	// 23514: check_violation - A value violates a CHECK constraint.
+	case "23514":
+		return database.NewCheckError(err.TableName, err.ConstraintName, err)
+	// 23505: unique_violation - A value violates a UNIQUE constraint.
+	case "23505":
+		return database.NewUniqueError(err.TableName, err.ConstraintName, err)
+	// 23503: foreign_key_violation - A value violates a foreign key constraint.
+	case "23503":
+		return database.NewForeignKeyError(err.TableName, err.ConstraintName, err)
+	// 23502: not_null_violation - A value violates a NOT NULL constraint.
+	case "23502":
+		return database.NewNotNullError(err.TableName, err.ConstraintName, err)
+	default:
+		return database.NewUnknownError(err)
+	}
+}

backend/v3/storage/database/events_testing/id_provider_org_test.go

@@ -247,7 +247,7 @@ func TestServer_TestIDProviderOrgReduces(t *testing.T) {
 		})
 		require.NoError(t, err)
 
-		retryDuration, tick := integration.WaitForAndTickWithMaxDuration(IAMCTX, time.Second*20)
+		retryDuration, tick := integration.WaitForAndTickWithMaxDuration(IAMCTX, time.Minute)
 		assert.EventuallyWithT(t, func(t *assert.CollectT) {
 			_, err := idpRepo.Get(IAMCTX, pool,
 				idpRepo.IDCondition(addOIDC.IdpId),

backend/v3/storage/database/events_testing/org_metadata_test.go

@@ -0,0 +1,296 @@
+//go:build integration
+
+package events_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/zitadel/zitadel/backend/v3/domain"
+	"github.com/zitadel/zitadel/backend/v3/storage/database"
+	"github.com/zitadel/zitadel/backend/v3/storage/database/repository"
+	"github.com/zitadel/zitadel/internal/integration"
+	v2beta "github.com/zitadel/zitadel/pkg/grpc/org/v2beta"
+)
+
+func TestServer_TestOrgMetadataReduces(t *testing.T) {
+	orgMetadataRepo := repository.OrganizationMetadataRepository()
+
+	// The API call also sets the metadata as primary, so we don't do a separate test for that.
+	t.Run("test organization metadata add reduces", func(t *testing.T) {
+		// Add a metadata to the organization
+		org := createTestScopedOrg(t)
+		beforeAdd := time.Now()
+		_, err := OrgClient.SetOrganizationMetadata(CTX, &v2beta.SetOrganizationMetadataRequest{
+			OrganizationId: org.GetId(),
+			Metadata: []*v2beta.Metadata{
+				{
+					Key:   "test-1-bool",
+					Value: []byte("false"),
+				},
+				{
+					Key:   "test-2-number",
+					Value: []byte("123"),
+				},
+				{
+					Key:   "test-3-object",
+					Value: []byte(`{"text":"value", "number":123, "bool": false}`),
+				},
+				{
+					Key:   "test-4-text",
+					Value: []byte(`"test-value"`),
+				},
+				{
+					Key:   "test-5-bytes",
+					Value: []byte(`test-value`),
+				},
+			},
+		})
+		require.NoError(t, err)
+		afterAdd := time.Now()
+
+		// Test that metadata add reduces
+		retryDuration, tick := integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
+		var gottenMetadata []*domain.OrganizationMetadata
+
+		assert.EventuallyWithT(t, func(t *assert.CollectT) {
+			gottenMetadata, err = orgMetadataRepo.List(CTX, pool,
+				database.WithCondition(
+					database.And(
+						orgMetadataRepo.InstanceIDCondition(Instance.Instance.Id),
+						orgMetadataRepo.OrganizationIDCondition(org.Id),
+						orgMetadataRepo.KeyCondition(database.TextOperationStartsWith, "test-"),
+					),
+				),
+				database.WithOrderByAscending(orgMetadataRepo.KeyColumn()),
+			)
+			require.NoError(t, err)
+			require.Len(t, gottenMetadata, 5)
+		}, retryDuration, tick)
+
+		assert.Equal(t, Instance.Instance.Id, gottenMetadata[0].InstanceID)
+		assert.Equal(t, org.Id, gottenMetadata[0].OrganizationID)
+		assert.Equal(t, "test-1-bool", gottenMetadata[0].Key)
+		assert.Equal(t, []byte(`false`), gottenMetadata[0].Value)
+		assert.WithinRange(t, gottenMetadata[0].CreatedAt, beforeAdd, afterAdd)
+		assert.WithinRange(t, gottenMetadata[0].UpdatedAt, beforeAdd, afterAdd)
+
+		assert.Equal(t, Instance.Instance.Id, gottenMetadata[1].InstanceID)
+		assert.Equal(t, org.Id, gottenMetadata[1].OrganizationID)
+		assert.Equal(t, "test-2-number", gottenMetadata[1].Key)
+		assert.Equal(t, []byte(`123`), gottenMetadata[1].Value)
+		assert.WithinRange(t, gottenMetadata[1].CreatedAt, beforeAdd, afterAdd)
+		assert.WithinRange(t, gottenMetadata[1].UpdatedAt, beforeAdd, afterAdd)
+
+		assert.Equal(t, Instance.Instance.Id, gottenMetadata[2].InstanceID)
+		assert.Equal(t, org.Id, gottenMetadata[2].OrganizationID)
+		assert.Equal(t, "test-3-object", gottenMetadata[2].Key)
+		assert.JSONEq(t, `{"text":"value", "number":123, "bool": false}`, string(gottenMetadata[2].Value))
+		assert.WithinRange(t, gottenMetadata[2].CreatedAt, beforeAdd, afterAdd)
+		assert.WithinRange(t, gottenMetadata[2].UpdatedAt, beforeAdd, afterAdd)
+
+		assert.Equal(t, Instance.Instance.Id, gottenMetadata[3].InstanceID)
+		assert.Equal(t, org.Id, gottenMetadata[3].OrganizationID)
+		assert.Equal(t, "test-4-text", gottenMetadata[3].Key)
+		assert.Equal(t, []byte(`"test-value"`), gottenMetadata[3].Value)
+		assert.WithinRange(t, gottenMetadata[3].CreatedAt, beforeAdd, afterAdd)
+		assert.WithinRange(t, gottenMetadata[3].UpdatedAt, beforeAdd, afterAdd)
+
+		assert.Equal(t, Instance.Instance.Id, gottenMetadata[4].InstanceID)
+		assert.Equal(t, org.Id, gottenMetadata[4].OrganizationID)
+		assert.Equal(t, "test-5-bytes", gottenMetadata[4].Key)
+		assert.Equal(t, []byte(`test-value`), gottenMetadata[4].Value)
+		assert.WithinRange(t, gottenMetadata[4].CreatedAt, beforeAdd, afterAdd)
+		assert.WithinRange(t, gottenMetadata[4].UpdatedAt, beforeAdd, afterAdd)
+	})
+
+	// The API call also sets the metadata as primary, so we don't do a separate test for that.
+	t.Run("ensure update works", func(t *testing.T) {
+		// Add a metadata to the organization
+		org := createTestScopedOrg(t)
+		beforeAdd := time.Now()
+		_, err := OrgClient.SetOrganizationMetadata(CTX, &v2beta.SetOrganizationMetadataRequest{
+			OrganizationId: org.GetId(),
+			Metadata: []*v2beta.Metadata{
+				{
+					Key:   "test-1-bool",
+					Value: []byte("false"),
+				},
+			},
+		})
+		require.NoError(t, err)
+		afterAdd := time.Now()
+
+		// Test that metadata add reduces
+		retryDuration, tick := integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
+		var gottenMetadata []*domain.OrganizationMetadata
+		assert.EventuallyWithT(t, func(t *assert.CollectT) {
+			gottenMetadata, err = orgMetadataRepo.List(CTX, pool,
+				database.WithCondition(
+					database.And(
+						orgMetadataRepo.InstanceIDCondition(Instance.Instance.Id),
+						orgMetadataRepo.OrganizationIDCondition(org.Id),
+						orgMetadataRepo.KeyCondition(database.TextOperationStartsWith, "test-"),
+					),
+				),
+				database.WithOrderByAscending(orgMetadataRepo.KeyColumn()),
+			)
+			require.NoError(t, err)
+			require.Len(t, gottenMetadata, 1)
+		}, retryDuration, tick)
+
+		assert.Equal(t, Instance.Instance.Id, gottenMetadata[0].InstanceID)
+		assert.Equal(t, org.Id, gottenMetadata[0].OrganizationID)
+		assert.Equal(t, "test-1-bool", gottenMetadata[0].Key)
+		assert.Equal(t, []byte(`false`), gottenMetadata[0].Value)
+		assert.WithinRange(t, gottenMetadata[0].CreatedAt, beforeAdd, afterAdd)
+		assert.WithinRange(t, gottenMetadata[0].UpdatedAt, beforeAdd, afterAdd)
+
+		_, err = OrgClient.SetOrganizationMetadata(CTX, &v2beta.SetOrganizationMetadataRequest{
+			OrganizationId: org.GetId(),
+			Metadata: []*v2beta.Metadata{
+				{
+					Key:   "test-1-bool",
+					Value: []byte("true"),
+				},
+			},
+		})
+		require.NoError(t, err)
+		afterUpdate := time.Now()
+
+		// Test that metadata add reduces
+		retryDuration, tick = integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
+		assert.EventuallyWithT(t, func(t *assert.CollectT) {
+			gottenMetadata, err = orgMetadataRepo.List(CTX, pool,
+				database.WithCondition(
+					database.And(
+						orgMetadataRepo.InstanceIDCondition(Instance.Instance.Id),
+						orgMetadataRepo.OrganizationIDCondition(org.Id),
+						orgMetadataRepo.KeyCondition(database.TextOperationStartsWith, "test-"),
+					),
+				),
+				database.WithOrderByAscending(orgMetadataRepo.KeyColumn()),
+			)
+			require.NoError(t, err)
+			require.Len(t, gottenMetadata, 1)
+			assert.Greater(t, gottenMetadata[0].UpdatedAt, gottenMetadata[0].CreatedAt)
+		}, retryDuration, tick)
+
+		assert.Equal(t, Instance.Instance.Id, gottenMetadata[0].InstanceID)
+		assert.Equal(t, org.Id, gottenMetadata[0].OrganizationID)
+		assert.Equal(t, "test-1-bool", gottenMetadata[0].Key)
+		assert.Equal(t, []byte(`true`), gottenMetadata[0].Value)
+		assert.WithinRange(t, gottenMetadata[0].CreatedAt, beforeAdd, afterAdd)
+		assert.WithinRange(t, gottenMetadata[0].UpdatedAt, afterAdd, afterUpdate)
+	})
+
+	t.Run("test org metadata remove reduces", func(t *testing.T) {
+		// Add a metadata to the organization
+		org := createTestScopedOrg(t)
+
+		_, err := OrgClient.SetOrganizationMetadata(CTX, &v2beta.SetOrganizationMetadataRequest{
+			OrganizationId: org.GetId(),
+			Metadata: []*v2beta.Metadata{
+				{
+					Key:   "test-bool",
+					Value: []byte("false"),
+				},
+				{
+					Key:   "test-number",
+					Value: []byte("123"),
+				},
+				{
+					Key:   "test-text",
+					Value: []byte(`"test-value"`),
+				},
+			},
+		})
+		require.NoError(t, err)
+
+		// Remove the metadata
+		_, err = OrgClient.DeleteOrganizationMetadata(CTX, &v2beta.DeleteOrganizationMetadataRequest{
+			OrganizationId: org.GetId(),
+			Keys:           []string{"test-bool", "test-text"},
+		})
+		require.NoError(t, err)
+
+		// Test that metadata remove reduces
+		retryDuration, tick := integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
+		assert.EventuallyWithT(t, func(t *assert.CollectT) {
+			metadata, err := orgMetadataRepo.Get(CTX, pool,
+				database.WithCondition(
+					database.And(
+						orgMetadataRepo.InstanceIDCondition(Instance.Instance.Id),
+						orgMetadataRepo.OrganizationIDCondition(org.Id),
+					),
+				),
+			)
+			// event instance.metadata.removed
+			require.NoError(t, err)
+			assert.Equal(t, "test-number", metadata.Key)
+		}, retryDuration, tick)
+	})
+
+	t.Run("test org metadata removed on org remove", func(t *testing.T) {
+		// Add a metadata to the organization
+		org := createOrg(t)
+
+		_, err := OrgClient.SetOrganizationMetadata(CTX, &v2beta.SetOrganizationMetadataRequest{
+			OrganizationId: org.GetId(),
+			Metadata: []*v2beta.Metadata{
+				{
+					Key:   "test-bool",
+					Value: []byte("false"),
+				},
+				{
+					Key:   "test-number",
+					Value: []byte("123"),
+				},
+				{
+					Key:   "test-text",
+					Value: []byte(`"test-value"`),
+				},
+			},
+		})
+		require.NoError(t, err)
+
+		// await metadata creation
+		retryDuration, tick := integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
+		assert.EventuallyWithT(t, func(t *assert.CollectT) {
+			metadata, err := orgMetadataRepo.List(CTX, pool,
+				database.WithCondition(
+					database.And(
+						orgMetadataRepo.InstanceIDCondition(Instance.Instance.Id),
+						orgMetadataRepo.OrganizationIDCondition(org.Id),
+					),
+				),
+			)
+			require.NoError(t, err)
+			assert.Len(t, metadata, 3)
+		}, retryDuration, tick)
+
+		_, err = OrgClient.DeleteOrganization(CTX, &v2beta.DeleteOrganizationRequest{
+			Id: org.Id,
+		})
+		require.NoError(t, err)
+
+		// Test that metadata remove reduces
+		retryDuration, tick = integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
+		assert.EventuallyWithT(t, func(t *assert.CollectT) {
+			metadata, err := orgMetadataRepo.List(CTX, pool,
+				database.WithCondition(
+					database.And(
+						orgMetadataRepo.InstanceIDCondition(Instance.Instance.Id),
+						orgMetadataRepo.OrganizationIDCondition(org.Id),
+					),
+				),
+			)
+			// event instance.metadata.removed
+			require.NoError(t, err)
+			assert.Len(t, metadata, 0)
+		}, retryDuration, tick)
+	})
+}

backend/v3/storage/database/events_testing/org_test.go

@@ -0,0 +1,55 @@
+//go:build integration
+
+package events_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/brianvoe/gofakeit/v6"
+	"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/repository"
+	"github.com/zitadel/zitadel/internal/integration"
+	org "github.com/zitadel/zitadel/pkg/grpc/org/v2beta"
+	v2beta "github.com/zitadel/zitadel/pkg/grpc/org/v2beta"
+)
+
+func createOrg(t *testing.T) *org.CreateOrganizationResponse {
+	t.Helper()
+	org, err := OrgClient.CreateOrganization(CTX, &org.CreateOrganizationRequest{
+		Name: gofakeit.Name(),
+	})
+	require.NoError(t, err)
+
+	orgRepo := repository.OrganizationRepository()
+	retryDuration, tick := integration.WaitForAndTickWithMaxDuration(t.Context(), time.Minute)
+	assert.EventuallyWithT(t, func(tc *assert.CollectT) {
+		_, err := orgRepo.Get(t.Context(), pool,
+			database.WithCondition(database.And(
+				orgRepo.InstanceIDCondition(Instance.Instance.Id),
+				orgRepo.IDCondition(org.GetId())),
+			),
+		)
+		assert.NoError(tc, err)
+	}, retryDuration, tick)
+
+	return org
+}
+
+func createTestScopedOrg(t *testing.T) *org.CreateOrganizationResponse {
+	org := createOrg(t)
+
+	t.Cleanup(func() {
+		_, err := OrgClient.DeleteOrganization(CTX, &v2beta.DeleteOrganizationRequest{
+			Id: org.GetId(),
+		})
+		if err != nil {
+			t.Logf("Failed to delete organization on cleanup: %v", err)
+		}
+	})
+
+	return org
+}

backend/v3/storage/database/repository/dates.go

@@ -0,0 +1,8 @@
+package repository
+
+import "time"
+
+type changedDates struct {
+	CreatedAt time.Time `db:"created_at"`
+	UpdatedAt time.Time `db:"updated_at"`
+}

backend/v3/storage/database/repository/org.go

@@ -14,8 +14,10 @@ import (
 var _ domain.OrganizationRepository = (*org)(nil)
 
 type org struct {
-	shouldLoadDomains bool
-	domainRepo        orgDomain
+	shouldLoadDomains  bool
+	domainRepo         orgDomain
+	shouldLoadMetadata bool
+	metadataRepo       orgMetadata
 }
 
 func (o org) unqualifiedTableName() string {
@@ -28,12 +30,14 @@ func OrganizationRepository() domain.OrganizationRepository {
 
 const queryOrganizationStmt = `SELECT organizations.id, organizations.name, organizations.instance_id, organizations.state, organizations.created_at, organizations.updated_at` +
 	` , jsonb_agg(json_build_object('instanceId', org_domains.instance_id, 'orgId', org_domains.org_id, 'domain', org_domains.domain, 'isVerified', org_domains.is_verified, 'isPrimary', org_domains.is_primary, 'validationType', org_domains.validation_type, 'createdAt', org_domains.created_at, 'updatedAt', org_domains.updated_at)) FILTER (WHERE org_domains.org_id IS NOT NULL) AS domains` +
+	` , jsonb_agg(json_build_object('instanceId', organization_metadata.instance_id, 'orgId', organization_metadata.organization_id, 'key', organization_metadata.key, 'value', encode(organization_metadata.value, 'base64'), 'createdAt', organization_metadata.created_at, 'updatedAt', organization_metadata.updated_at)) FILTER (WHERE organization_metadata.organization_id IS NOT NULL) AS metadata` +
 	` FROM zitadel.organizations`
 
 // Get implements [domain.OrganizationRepository].
 func (o org) Get(ctx context.Context, client database.QueryExecutor, opts ...database.QueryOption) (*domain.Organization, error) {
 	opts = append(opts,
 		o.joinDomains(),
+		o.joinMetadata(),
 		database.WithGroupBy(o.InstanceIDColumn(), o.IDColumn()),
 	)
 
@@ -57,6 +61,7 @@ func (o org) Get(ctx context.Context, client database.QueryExecutor, opts ...dat
 func (o org) List(ctx context.Context, client database.QueryExecutor, opts ...database.QueryOption) ([]*domain.Organization, error) {
 	opts = append(opts,
 		o.joinDomains(),
+		o.joinMetadata(),
 		database.WithGroupBy(o.InstanceIDColumn(), o.IDColumn()),
 	)
 
@@ -187,6 +192,31 @@ func (o org) ExistsDomain(cond database.Condition) database.Condition {
 	)
 }
 
+// ExistsMetadata creates a correlated [database.Exists] condition on organization_metadata.
+// Use this when you want to filter organizations by a metadata condition but still return all metadata
+// of the organization in the aggregated result.
+// Example usage:
+//
+//	metadataRepo := orgRepo.Metadata(true) // ensure metadata are loaded/aggregated
+//	org, _ := orgRepo.Get(ctx,
+//	    database.WithCondition(
+//	        database.And(
+//	            orgRepo.InstanceIDCondition(instanceID),
+//	            orgRepo.MetadataExists(metadataRepo.KeyCondition(database.TextOperationEqual, "urn:zitadel:org:custom:my-key")),
+//	        ),
+//	    ),
+//	)
+func (o org) ExistsMetadata(cond database.Condition) database.Condition {
+	return database.Exists(
+		o.metadataRepo.qualifiedTableName(),
+		database.And(
+			database.NewColumnCondition(o.InstanceIDColumn(), o.metadataRepo.InstanceIDColumn()),
+			database.NewColumnCondition(o.IDColumn(), o.metadataRepo.OrganizationIDColumn()),
+			cond,
+		),
+	)
+}
+
 // -------------------------------------------------------------
 // columns
 // -------------------------------------------------------------
@@ -227,7 +257,8 @@ func (o org) UpdatedAtColumn() database.Column {
 
 type rawOrg struct {
 	*domain.Organization
-	Domains JSONArray[domain.OrganizationDomain] `json:"domains,omitempty" db:"domains"`
+	Domains  JSONArray[domain.OrganizationDomain]   `json:"domains,omitempty" db:"domains"`
+	Metadata JSONArray[domain.OrganizationMetadata] `json:"metadata,omitempty" db:"metadata"`
 }
 
 func scanOrganization(ctx context.Context, querier database.Querier, builder *database.StatementBuilder) (*domain.Organization, error) {
@@ -241,6 +272,7 @@ func scanOrganization(ctx context.Context, querier database.Querier, builder *da
 		return nil, err
 	}
 	org.Organization.Domains = org.Domains
+	org.Organization.Metadata = org.Metadata
 
 	return org.Organization, nil
 }
@@ -260,6 +292,7 @@ func scanOrganizations(ctx context.Context, querier database.Querier, builder *d
 	for i, org := range orgs {
 		result[i] = org.Organization
 		result[i].Domains = org.Domains
+		result[i].Metadata = org.Metadata
 	}
 
 	return result, nil
@@ -271,7 +304,8 @@ func scanOrganizations(ctx context.Context, querier database.Querier, builder *d
 
 func (o org) LoadDomains() domain.OrganizationRepository {
 	return &org{
-		shouldLoadDomains: true,
+		shouldLoadDomains:  true,
+		shouldLoadMetadata: o.shouldLoadMetadata,
 	}
 }
 
@@ -293,3 +327,29 @@ func (o org) joinDomains() database.QueryOption {
 		database.And(columns...),
 	)
 }
+
+func (o org) LoadMetadata() domain.OrganizationRepository {
+	return &org{
+		shouldLoadDomains:  o.shouldLoadDomains,
+		shouldLoadMetadata: true,
+	}
+}
+
+func (o org) joinMetadata() database.QueryOption {
+	columns := make([]database.Condition, 0, 3)
+	columns = append(columns,
+		database.NewColumnCondition(o.InstanceIDColumn(), o.metadataRepo.InstanceIDColumn()),
+		database.NewColumnCondition(o.IDColumn(), o.metadataRepo.OrganizationIDColumn()),
+	)
+
+	// If metadata should not be joined, we make sure to return null for the metadata columns
+	// the query optimizer of the dialect should optimize this away if no metadata are requested
+	if !o.shouldLoadMetadata {
+		columns = append(columns, database.IsNull(o.metadataRepo.OrganizationIDColumn()))
+	}
+
+	return database.WithLeftJoin(
+		o.metadataRepo.qualifiedTableName(),
+		database.And(columns...),
+	)
+}

backend/v3/storage/database/repository/org_metadata.go

@@ -0,0 +1,228 @@
+package repository
+
+import (
+	"context"
+	"errors"
+
+	"github.com/zitadel/zitadel/backend/v3/domain"
+	"github.com/zitadel/zitadel/backend/v3/storage/database"
+)
+
+var _ domain.OrganizationMetadataRepository = (*orgMetadata)(nil)
+
+type orgMetadata struct{}
+
+func OrganizationMetadataRepository() domain.OrganizationMetadataRepository {
+	return new(orgMetadata)
+}
+
+func (o orgMetadata) qualifiedTableName() string {
+	return "zitadel.organization_metadata"
+}
+
+func (o orgMetadata) unqualifiedTableName() string {
+	return "organization_metadata"
+}
+
+// -------------------------------------------------------------
+// repository
+// -------------------------------------------------------------
+
+const queryOrganizationMetadataStmt = `SELECT instance_id, organization_id, key, value, created_at, updated_at ` +
+	`FROM zitadel.organization_metadata`
+
+// Get implements [domain.OrganizationMetadataRepository].
+func (o orgMetadata) Get(ctx context.Context, client database.QueryExecutor, opts ...database.QueryOption) (*domain.OrganizationMetadata, error) {
+	options := new(database.QueryOpts)
+	for _, opt := range opts {
+		opt(options)
+	}
+	if !options.Condition.IsRestrictingColumn(o.InstanceIDColumn()) {
+		return nil, database.NewMissingConditionError(o.InstanceIDColumn())
+	}
+
+	var builder database.StatementBuilder
+	builder.WriteString(queryOrganizationMetadataStmt)
+	options.Write(&builder)
+
+	return scanOrganizationMetadata(ctx, client, &builder)
+}
+
+// List implements [domain.OrganizationMetadataRepository].
+func (o orgMetadata) List(ctx context.Context, client database.QueryExecutor, opts ...database.QueryOption) ([]*domain.OrganizationMetadata, error) {
+	options := new(database.QueryOpts)
+	for _, opt := range opts {
+		opt(options)
+	}
+	if !options.Condition.IsRestrictingColumn(o.InstanceIDColumn()) {
+		return nil, database.NewMissingConditionError(o.InstanceIDColumn())
+	}
+
+	var builder database.StatementBuilder
+	builder.WriteString(queryOrganizationMetadataStmt)
+	options.Write(&builder)
+
+	return scanOrganizationMetadataList(ctx, client, &builder)
+}
+
+// Set implements [domain.OrganizationMetadataRepository].
+func (o orgMetadata) Set(ctx context.Context, client database.QueryExecutor, metadata ...*domain.OrganizationMetadata) error {
+	if len(metadata) == 0 {
+		return database.ErrNoChanges
+	}
+
+	var builder database.StatementBuilder
+	builder.WriteString(`INSERT INTO zitadel.organization_metadata (instance_id, organization_id, key, value, created_at, updated_at) VALUES `)
+	for i, m := range metadata {
+		var createdAt, updatedAt any = database.DefaultInstruction, database.NullInstruction
+		if !m.CreatedAt.IsZero() {
+			createdAt = m.CreatedAt
+		}
+		if !m.UpdatedAt.IsZero() {
+			updatedAt = m.UpdatedAt
+		}
+		if i > 0 {
+			builder.WriteString(", ")
+		}
+		builder.WriteRune('(')
+		builder.WriteArgs(m.InstanceID, m.OrganizationID, m.Key, m.Value, createdAt, updatedAt)
+		builder.WriteRune(')')
+	}
+	builder.WriteString(` ON CONFLICT (instance_id, organization_id, key) DO UPDATE SET value = EXCLUDED.value, updated_at = EXCLUDED.updated_at RETURNING created_at, updated_at`)
+
+	res, err := client.Query(ctx, builder.String(), builder.Args()...)
+	if err != nil {
+		return err
+	}
+	if res.Err() != nil {
+		return res.Err()
+	}
+	dates := make([]changedDates, 0, len(metadata))
+	err = res.(database.CollectableRows).Collect(&dates)
+	if err != nil {
+		return err
+	}
+	if len(dates) != len(metadata) {
+		return errors.New("repository: returned changed dates count does not match list count")
+	}
+	for i, d := range dates {
+		metadata[i].CreatedAt = d.CreatedAt
+		metadata[i].UpdatedAt = d.UpdatedAt
+	}
+	return nil
+}
+
+// Remove implements [domain.OrganizationMetadataRepository].
+func (o orgMetadata) Remove(ctx context.Context, client database.QueryExecutor, condition database.Condition) (int64, error) {
+	if err := checkPKCondition(o, condition); err != nil {
+		return 0, err
+	}
+
+	var builder database.StatementBuilder
+	builder.WriteString(`DELETE FROM zitadel.organization_metadata `)
+	writeCondition(&builder, condition)
+
+	return client.Exec(ctx, builder.String(), builder.Args()...)
+}
+
+// -------------------------------------------------------------
+// conditions
+// -------------------------------------------------------------
+
+// PrimaryKeyCondition implements [domain.OrganizationMetadataRepository].
+func (o orgMetadata) PrimaryKeyCondition(instanceID string, orgID string, key string) database.Condition {
+	return database.And(
+		o.InstanceIDCondition(instanceID),
+		o.OrganizationIDCondition(orgID),
+		o.KeyCondition(database.TextOperationEqual, key),
+	)
+}
+
+// InstanceIDCondition implements [domain.OrganizationMetadataRepository].
+func (o orgMetadata) InstanceIDCondition(instanceID string) database.Condition {
+	return database.NewTextCondition(o.InstanceIDColumn(), database.TextOperationEqual, instanceID)
+}
+
+// KeyCondition implements [domain.OrganizationMetadataRepository].
+func (o orgMetadata) KeyCondition(op database.TextOperation, key string) database.Condition {
+	return database.NewTextCondition(o.KeyColumn(), op, key)
+}
+
+// OrganizationIDCondition implements [domain.OrganizationMetadataRepository].
+func (o orgMetadata) OrganizationIDCondition(orgID string) database.Condition {
+	return database.NewTextCondition(o.OrganizationIDColumn(), database.TextOperationEqual, orgID)
+}
+
+// ValueCondition implements [domain.OrganizationMetadataRepository].
+func (o orgMetadata) ValueCondition(op database.BytesOperation, value []byte) database.Condition {
+	return database.NewBytesCondition[[]byte](database.SHA256Column(o.ValueColumn()), op, database.SHA256Value(value))
+}
+
+// -------------------------------------------------------------
+// columns
+// -------------------------------------------------------------
+
+// PrimaryKeyColumns implements [domain.OrganizationMetadataRepository].
+func (o orgMetadata) PrimaryKeyColumns() []database.Column {
+	return []database.Column{o.InstanceIDColumn(), o.OrganizationIDColumn(), o.KeyColumn()}
+}
+
+// CreatedAtColumn implements [domain.OrganizationMetadataRepository].
+func (o orgMetadata) CreatedAtColumn() database.Column {
+	return database.NewColumn(o.unqualifiedTableName(), "created_at")
+}
+
+// InstanceIDColumn implements [domain.OrganizationMetadataRepository].
+func (o orgMetadata) InstanceIDColumn() database.Column {
+	return database.NewColumn(o.unqualifiedTableName(), "instance_id")
+}
+
+// KeyColumn implements [domain.OrganizationMetadataRepository].
+func (o orgMetadata) KeyColumn() database.Column {
+	return database.NewColumn(o.unqualifiedTableName(), "key")
+}
+
+// OrganizationIDColumn implements [domain.OrganizationMetadataRepository].
+func (o orgMetadata) OrganizationIDColumn() database.Column {
+	return database.NewColumn(o.unqualifiedTableName(), "organization_id")
+}
+
+// UpdatedAtColumn implements [domain.OrganizationMetadataRepository].
+func (o orgMetadata) UpdatedAtColumn() database.Column {
+	return database.NewColumn(o.unqualifiedTableName(), "updated_at")
+}
+
+// ValueColumn implements [domain.OrganizationMetadataRepository].
+func (o orgMetadata) ValueColumn() database.Column {
+	return database.NewColumn(o.unqualifiedTableName(), "value")
+}
+
+// -------------------------------------------------------------
+// scanners
+// -------------------------------------------------------------
+
+func scanOrganizationMetadata(ctx context.Context, client database.Querier, builder *database.StatementBuilder) (*domain.OrganizationMetadata, error) {
+	rows, err := client.Query(ctx, builder.String(), builder.Args()...)
+	if err != nil {
+		return nil, err
+	}
+
+	metadata := new(domain.OrganizationMetadata)
+	if err := rows.(database.CollectableRows).CollectExactlyOneRow(metadata); err != nil {
+		return nil, err
+	}
+	return metadata, nil
+}
+
+func scanOrganizationMetadataList(ctx context.Context, client database.Querier, builder *database.StatementBuilder) ([]*domain.OrganizationMetadata, error) {
+	rows, err := client.Query(ctx, builder.String(), builder.Args()...)
+	if err != nil {
+		return nil, err
+	}
+
+	var metadataList []*domain.OrganizationMetadata
+	if err := rows.(database.CollectableRows).Collect(&metadataList); err != nil {
+		return nil, err
+	}
+	return metadataList, nil
+}

backend/v3/storage/database/repository/org_metadata_test.go

@@ -0,0 +1,934 @@
+package repository_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/brianvoe/gofakeit/v6"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/zitadel/zitadel/backend/v3/domain"
+	"github.com/zitadel/zitadel/backend/v3/storage/database"
+	"github.com/zitadel/zitadel/backend/v3/storage/database/repository"
+)
+
+func TestGetOrganizationMetadata(t *testing.T) {
+	tx, err := pool.Begin(t.Context(), nil)
+	require.NoError(t, err)
+	defer func() {
+		err = tx.Rollback(t.Context())
+		if err != nil {
+			t.Log("error during rollback:", err)
+		}
+	}()
+
+	instanceRepo := repository.InstanceRepository()
+	orgRepo := repository.OrganizationRepository()
+	metadataRepo := repository.OrganizationMetadataRepository()
+
+	// create instance
+	instanceID := gofakeit.UUID()
+	instance := domain.Instance{
+		ID:              instanceID,
+		Name:            gofakeit.Name(),
+		DefaultOrgID:    "defaultOrgId",
+		IAMProjectID:    "iamProject",
+		ConsoleClientID: "consoleClient",
+		ConsoleAppID:    "consoleApp",
+		DefaultLanguage: "defaultLanguage",
+	}
+	err = instanceRepo.Create(t.Context(), tx, &instance)
+	require.NoError(t, err)
+
+	// create organization
+	orgA := domain.Organization{
+		ID:         "1",
+		Name:       gofakeit.Name(),
+		InstanceID: instanceID,
+		State:      domain.OrgStateActive,
+	}
+	err = orgRepo.Create(t.Context(), tx, &orgA)
+	require.NoError(t, err)
+
+	err = metadataRepo.Set(
+		t.Context(),
+		tx,
+		&domain.OrganizationMetadata{
+			OrganizationID: orgA.ID,
+			Metadata: domain.Metadata{
+				InstanceID: instanceID,
+				Key:        "urn:zitadel:key",
+				Value:      []byte("1234"),
+			},
+		},
+		&domain.OrganizationMetadata{
+			OrganizationID: orgA.ID,
+			Metadata: domain.Metadata{
+				InstanceID: instanceID,
+				Key:        "urn:zitadel:key2",
+				Value:      []byte("asdf"),
+			},
+		},
+	)
+	require.NoError(t, err)
+
+	orgB := domain.Organization{
+		ID:         "2",
+		Name:       gofakeit.Name(),
+		InstanceID: instanceID,
+		State:      domain.OrgStateActive,
+	}
+	err = orgRepo.Create(t.Context(), tx, &orgB)
+	require.NoError(t, err)
+
+	err = metadataRepo.Set(
+		t.Context(), tx,
+		&domain.OrganizationMetadata{
+			OrganizationID: orgB.ID,
+			Metadata: domain.Metadata{
+				InstanceID: instanceID,
+				Key:        "urn:zitadel:key",
+				Value:      []byte("5678"),
+			},
+		},
+		&domain.OrganizationMetadata{
+			OrganizationID: orgB.ID,
+			Metadata: domain.Metadata{
+				InstanceID: instanceID,
+				Key:        "urn:zitadel:key2",
+				Value:      []byte(`"asdf"`),
+			},
+		},
+	)
+	require.NoError(t, err)
+
+	t.Run("metadata without instance condition", func(t *testing.T) {
+		metadata, err := metadataRepo.Get(
+			t.Context(), tx,
+			database.WithCondition(metadataRepo.OrganizationIDCondition(orgA.ID)),
+		)
+		assert.ErrorIs(t, err, new(database.MissingConditionError))
+		assert.Nil(t, metadata)
+	})
+
+	t.Run("no metadata found", func(t *testing.T) {
+		metadata, err := metadataRepo.Get(
+			t.Context(), tx,
+			database.WithCondition(metadataRepo.InstanceIDCondition("non-existing")),
+		)
+		assert.ErrorIs(t, err, new(database.NoRowFoundError))
+		assert.Empty(t, metadata)
+	})
+
+	t.Run("multiple metadata found", func(t *testing.T) {
+		metadata, err := metadataRepo.Get(
+			t.Context(), tx,
+			database.WithCondition(metadataRepo.InstanceIDCondition(instanceID)),
+		)
+		require.ErrorIs(t, err, new(database.MultipleRowsFoundError))
+		assert.Empty(t, metadata)
+	})
+
+	t.Run("metadata by key", func(t *testing.T) {
+		metadata, err := metadataRepo.Get(
+			t.Context(), tx,
+			database.WithCondition(
+				database.And(
+					metadataRepo.InstanceIDCondition(instanceID),
+					metadataRepo.OrganizationIDCondition(orgA.ID),
+					metadataRepo.KeyCondition(database.TextOperationEqual, "urn:zitadel:key2"),
+				),
+			),
+			database.WithOrderByAscending(metadataRepo.OrganizationIDColumn()),
+		)
+		require.NoError(t, err)
+		assert.Equal(t, "urn:zitadel:key2", metadata.Key)
+		assert.Equal(t, []byte(`asdf`), metadata.Value)
+	})
+
+	t.Run("metadata by value", func(t *testing.T) {
+		metadata, err := metadataRepo.Get(
+			t.Context(), tx,
+			database.WithCondition(
+				database.And(
+					metadataRepo.InstanceIDCondition(instanceID),
+					metadataRepo.OrganizationIDCondition(orgA.ID),
+					metadataRepo.ValueCondition(database.BytesOperationEqual, []byte("asdf")),
+				),
+			),
+			database.WithOrderByAscending(metadataRepo.OrganizationIDColumn()),
+		)
+		require.NoError(t, err)
+		assert.Equal(t, "urn:zitadel:key2", metadata.Key)
+	})
+}
+
+func TestListOrganizationMetadata(t *testing.T) {
+	tx, err := pool.Begin(t.Context(), nil)
+	require.NoError(t, err)
+	defer func() {
+		err = tx.Rollback(t.Context())
+		if err != nil {
+			t.Log("error during rollback:", err)
+		}
+	}()
+
+	instanceRepo := repository.InstanceRepository()
+	orgRepo := repository.OrganizationRepository()
+	metadataRepo := repository.OrganizationMetadataRepository()
+
+	// create instance
+	instanceID := gofakeit.UUID()
+	instance := domain.Instance{
+		ID:              instanceID,
+		Name:            gofakeit.Name(),
+		DefaultOrgID:    "defaultOrgId",
+		IAMProjectID:    "iamProject",
+		ConsoleClientID: "consoleClient",
+		ConsoleAppID:    "consoleApp",
+		DefaultLanguage: "defaultLanguage",
+	}
+	err = instanceRepo.Create(t.Context(), tx, &instance)
+	require.NoError(t, err)
+
+	// create organization
+	orgA := domain.Organization{
+		ID:         "1",
+		Name:       gofakeit.Name(),
+		InstanceID: instanceID,
+		State:      domain.OrgStateActive,
+	}
+	err = orgRepo.Create(t.Context(), tx, &orgA)
+	require.NoError(t, err)
+
+	err = metadataRepo.Set(
+		t.Context(), tx,
+		&domain.OrganizationMetadata{
+			OrganizationID: orgA.ID,
+			Metadata: domain.Metadata{
+				InstanceID: instanceID,
+				Key:        "urn:zitadel:key",
+				Value:      []byte("1234"),
+			},
+		},
+		&domain.OrganizationMetadata{
+			OrganizationID: orgA.ID,
+			Metadata: domain.Metadata{
+				InstanceID: instanceID,
+				Key:        "urn:zitadel:key2",
+				Value:      []byte("asdf"),
+			},
+		},
+	)
+	require.NoError(t, err)
+
+	metadataOrgA, err := metadataRepo.List(
+		t.Context(), tx,
+		database.WithCondition(
+			database.And(
+				metadataRepo.OrganizationIDCondition(orgA.ID),
+				metadataRepo.InstanceIDCondition(instanceID),
+			),
+		),
+	)
+	require.NoError(t, err)
+
+	orgB := domain.Organization{
+		ID:         "2",
+		Name:       gofakeit.Name(),
+		InstanceID: instanceID,
+		State:      domain.OrgStateActive,
+	}
+	err = orgRepo.Create(t.Context(), tx, &orgB)
+	require.NoError(t, err)
+
+	err = metadataRepo.Set(
+		t.Context(), tx,
+		&domain.OrganizationMetadata{
+			OrganizationID: orgB.ID,
+			Metadata: domain.Metadata{
+				InstanceID: instanceID,
+				Key:        "urn:zitadel:key",
+				Value:      []byte("5678"),
+			},
+		},
+		&domain.OrganizationMetadata{
+			OrganizationID: orgB.ID,
+			Metadata: domain.Metadata{
+				InstanceID: instanceID,
+				Key:        "urn:zitadel:key2",
+				Value:      []byte(`"asdf"`),
+			},
+		},
+	)
+	require.NoError(t, err)
+
+	metadataOrgB, err := metadataRepo.List(
+		t.Context(), tx,
+		database.WithCondition(
+			database.And(
+				metadataRepo.OrganizationIDCondition(orgB.ID),
+				metadataRepo.InstanceIDCondition(instanceID),
+			),
+		),
+	)
+	require.NoError(t, err)
+
+	t.Run("metadata without instance condition", func(t *testing.T) {
+		metadata, err := metadataRepo.List(
+			t.Context(), tx,
+			database.WithCondition(metadataRepo.OrganizationIDCondition(orgA.ID)),
+		)
+		assert.ErrorIs(t, err, new(database.MissingConditionError))
+		assert.Nil(t, metadata)
+	})
+
+	t.Run("no metadata found", func(t *testing.T) {
+		metadata, err := metadataRepo.List(
+			t.Context(), tx,
+			database.WithCondition(metadataRepo.InstanceIDCondition("non-existing")),
+		)
+		assert.NoError(t, err)
+		assert.Len(t, metadata, 0)
+	})
+
+	t.Run("all metadata of instance", func(t *testing.T) {
+		metadata, err := metadataRepo.List(
+			t.Context(), tx,
+			database.WithCondition(metadataRepo.InstanceIDCondition(instanceID)),
+		)
+		require.NoError(t, err)
+		assert.ElementsMatch(t, metadata, append(metadataOrgA, metadataOrgB...))
+	})
+
+	t.Run("metadata by org id", func(t *testing.T) {
+		metadata, err := metadataRepo.List(
+			t.Context(), tx,
+			database.WithCondition(
+				database.And(
+					metadataRepo.InstanceIDCondition(instanceID),
+					metadataRepo.OrganizationIDCondition(orgA.ID),
+				),
+			),
+		)
+		require.NoError(t, err)
+		assert.ElementsMatch(t, metadata, metadataOrgA)
+	})
+
+	t.Run("metadata by key", func(t *testing.T) {
+		metadata, err := metadataRepo.List(
+			t.Context(), tx,
+			database.WithCondition(
+				database.And(
+					metadataRepo.InstanceIDCondition(instanceID),
+					metadataRepo.KeyCondition(database.TextOperationEqual, "urn:zitadel:key2"),
+				),
+			),
+			database.WithOrderByAscending(metadataRepo.OrganizationIDColumn()),
+		)
+		require.NoError(t, err)
+		require.Len(t, metadata, 2)
+		assert.Equal(t, []byte("asdf"), metadata[0].Value)
+		assert.Equal(t, []byte(`"asdf"`), metadata[1].Value)
+	})
+
+	t.Run("metadata by value", func(t *testing.T) {
+		metadata, err := metadataRepo.List(
+			t.Context(), tx,
+			database.WithCondition(
+				database.And(
+					metadataRepo.InstanceIDCondition(instanceID),
+					metadataRepo.ValueCondition(database.BytesOperationEqual, []byte("asdf")),
+				),
+			),
+			database.WithOrderByAscending(metadataRepo.OrganizationIDColumn()),
+		)
+		require.NoError(t, err)
+		require.Len(t, metadata, 1)
+		assert.Equal(t, "urn:zitadel:key2", metadata[0].Key)
+		assert.Equal(t, "1", metadata[0].OrganizationID)
+	})
+}
+
+func TestSetOrganizationMetadata_UpdatedAt(t *testing.T) {
+	instanceRepo := repository.InstanceRepository()
+	orgRepo := repository.OrganizationRepository()
+	metadataRepo := repository.OrganizationMetadataRepository()
+
+	// create instance
+	instanceID := gofakeit.UUID()
+	instance := domain.Instance{
+		ID:              instanceID,
+		Name:            gofakeit.Name(),
+		DefaultOrgID:    "defaultOrgId",
+		IAMProjectID:    "iamProject",
+		ConsoleClientID: "consoleClient",
+		ConsoleAppID:    "consoleApp",
+		DefaultLanguage: "defaultLanguage",
+	}
+	err := instanceRepo.Create(t.Context(), pool, &instance)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		_, err := instanceRepo.Delete(context.Background(), pool, instanceID)
+		require.NoError(t, err)
+	})
+
+	// create organization
+	orgID := gofakeit.UUID()
+	organization := domain.Organization{
+		ID:         orgID,
+		Name:       gofakeit.Name(),
+		InstanceID: instanceID,
+		State:      domain.OrgStateActive,
+	}
+	err = orgRepo.Create(t.Context(), pool, &organization)
+	require.NoError(t, err)
+
+	metadata := &domain.OrganizationMetadata{
+		OrganizationID: orgID,
+		Metadata: domain.Metadata{
+			InstanceID: instanceID,
+			Key:        "urn:zitadel:key",
+			Value:      []byte("some-value"),
+		},
+	}
+
+	beforeCreate := time.Now()
+	err = metadataRepo.Set(t.Context(), pool, metadata)
+	require.NoError(t, err)
+	afterCreate := time.Now()
+
+	assert.WithinRange(t, metadata.CreatedAt, beforeCreate, afterCreate)
+	assert.WithinRange(t, metadata.UpdatedAt, beforeCreate, afterCreate)
+
+	metadata.Value = []byte("other-value")
+	metadata.CreatedAt = time.Time{}
+	metadata.UpdatedAt = time.Time{}
+	beforeUpdate := time.Now()
+	err = metadataRepo.Set(t.Context(), pool, metadata)
+	require.NoError(t, err)
+	afterUpdate := time.Now()
+
+	assert.WithinRange(t, metadata.CreatedAt, beforeCreate, afterCreate)
+	assert.WithinRange(t, metadata.UpdatedAt, beforeUpdate, afterUpdate)
+}
+
+func TestSetOrganizationMetadata(t *testing.T) {
+	beforeSet := time.Now()
+
+	tx, err := pool.Begin(t.Context(), nil)
+	require.NoError(t, err)
+	defer func() {
+		err = tx.Rollback(t.Context())
+		if err != nil {
+			t.Log("error during rollback:", err)
+		}
+	}()
+
+	instanceRepo := repository.InstanceRepository()
+	orgRepo := repository.OrganizationRepository()
+	metadataRepo := repository.OrganizationMetadataRepository()
+
+	// create instance
+	instanceID := gofakeit.UUID()
+	instance := domain.Instance{
+		ID:              instanceID,
+		Name:            gofakeit.Name(),
+		DefaultOrgID:    "defaultOrgId",
+		IAMProjectID:    "iamProject",
+		ConsoleClientID: "consoleClient",
+		ConsoleAppID:    "consoleApp",
+		DefaultLanguage: "defaultLanguage",
+	}
+	err = instanceRepo.Create(t.Context(), tx, &instance)
+	require.NoError(t, err)
+
+	// create organization
+	orgID := gofakeit.UUID()
+	organization := domain.Organization{
+		ID:         orgID,
+		Name:       gofakeit.Name(),
+		InstanceID: instanceID,
+		State:      domain.OrgStateActive,
+	}
+	err = orgRepo.Create(t.Context(), tx, &organization)
+	require.NoError(t, err)
+
+	t.Run("check fields all fields scanned", func(t *testing.T) {
+		tx, err := tx.Begin(t.Context())
+		require.NoError(t, err)
+		defer func() {
+			require.NoError(t, tx.Rollback(t.Context()))
+		}()
+
+		metadata := &domain.OrganizationMetadata{
+			OrganizationID: orgID,
+			Metadata: domain.Metadata{
+				InstanceID: instanceID,
+				Key:        "urn:zitadel:key",
+				Value:      []byte("some-value"),
+			},
+		}
+
+		err = metadataRepo.Set(t.Context(), tx, metadata)
+		require.NoError(t, err)
+		afterSet := time.Now()
+
+		assert.Equal(t, orgID, metadata.OrganizationID)
+		assert.Equal(t, instanceID, metadata.InstanceID)
+		assert.Equal(t, "urn:zitadel:key", metadata.Key)
+		assert.Equal(t, []byte("some-value"), metadata.Value)
+		assert.WithinRange(t, metadata.CreatedAt, beforeSet, afterSet)
+		assert.WithinRange(t, metadata.UpdatedAt, beforeSet, afterSet)
+		assert.Equal(t, metadata.CreatedAt, metadata.UpdatedAt)
+	})
+
+	t.Run("set one organization metadata", func(t *testing.T) {
+		tx, err := tx.Begin(t.Context())
+		require.NoError(t, err)
+		defer func() {
+			require.NoError(t, tx.Rollback(t.Context()))
+		}()
+
+		err = metadataRepo.Set(t.Context(), tx, &domain.OrganizationMetadata{
+			OrganizationID: orgID,
+			Metadata: domain.Metadata{
+				InstanceID: instanceID,
+				Key:        "urn:zitadel:key",
+				Value:      []byte("some-value"),
+			},
+		})
+		require.NoError(t, err)
+		afterSet := time.Now()
+
+		savedMetadata, err := metadataRepo.Get(
+			t.Context(), tx, database.WithCondition(
+				database.And(
+					metadataRepo.InstanceIDCondition(instanceID),
+					metadataRepo.OrganizationIDCondition(orgID),
+					metadataRepo.KeyCondition(database.TextOperationEqual, "urn:zitadel:key"),
+				),
+			),
+		)
+		require.NoError(t, err)
+		assert.Equal(t, orgID, savedMetadata.OrganizationID)
+		assert.Equal(t, instanceID, savedMetadata.InstanceID)
+		assert.Equal(t, "urn:zitadel:key", savedMetadata.Key)
+		assert.Equal(t, []byte("some-value"), savedMetadata.Value)
+		assert.WithinRange(t, savedMetadata.CreatedAt, beforeSet, afterSet)
+		assert.WithinRange(t, savedMetadata.UpdatedAt, beforeSet, afterSet)
+		assert.Equal(t, savedMetadata.CreatedAt, savedMetadata.UpdatedAt)
+	})
+
+	t.Run("set multiple organization metadata", func(t *testing.T) {
+		tx, err := tx.Begin(t.Context())
+		require.NoError(t, err)
+		defer func() {
+			require.NoError(t, tx.Rollback(t.Context()))
+		}()
+
+		err = metadataRepo.Set(t.Context(), tx,
+			&domain.OrganizationMetadata{
+				OrganizationID: orgID,
+				Metadata: domain.Metadata{
+					InstanceID: instanceID,
+					Key:        "urn:zitadel:key",
+					Value:      []byte("some-value"),
+				},
+			},
+			&domain.OrganizationMetadata{
+				OrganizationID: orgID,
+				Metadata: domain.Metadata{
+					InstanceID: instanceID,
+					Key:        "urn:zitadel:key2",
+					Value:      []byte("1234"),
+				},
+			},
+		)
+		require.NoError(t, err)
+		afterSet := time.Now()
+
+		savedMetadata, err := metadataRepo.List(
+			t.Context(), tx,
+			database.WithCondition(
+				database.And(
+					metadataRepo.InstanceIDCondition(instanceID),
+					metadataRepo.OrganizationIDCondition(orgID),
+					metadataRepo.KeyCondition(database.TextOperationStartsWith, "urn:zitadel:key"),
+				),
+			),
+			database.WithOrderByAscending(metadataRepo.KeyColumn()),
+		)
+		require.NoError(t, err)
+		require.Len(t, savedMetadata, 2)
+
+		for _, saved := range savedMetadata {
+			assert.Equal(t, orgID, saved.OrganizationID)
+			assert.Equal(t, instanceID, saved.InstanceID)
+			assert.WithinRange(t, saved.CreatedAt, beforeSet, afterSet)
+			assert.WithinRange(t, saved.UpdatedAt, beforeSet, afterSet)
+			assert.Equal(t, saved.CreatedAt, saved.UpdatedAt)
+		}
+
+		assert.Equal(t, "urn:zitadel:key", savedMetadata[0].Key)
+		assert.Equal(t, []byte("some-value"), savedMetadata[0].Value)
+
+		assert.Equal(t, "urn:zitadel:key2", savedMetadata[1].Key)
+		assert.Equal(t, []byte("1234"), savedMetadata[1].Value)
+	})
+
+	t.Run("set no organization metadata", func(t *testing.T) {
+		tx, err := tx.Begin(t.Context())
+		require.NoError(t, err)
+		defer func() {
+			require.NoError(t, tx.Rollback(t.Context()))
+		}()
+
+		err = metadataRepo.Set(t.Context(), tx)
+		require.ErrorIs(t, err, database.ErrNoChanges)
+	})
+
+	t.Run("overwrite organization metadata", func(t *testing.T) {
+		tx, err := tx.Begin(t.Context())
+		require.NoError(t, err)
+		defer func() {
+			require.NoError(t, tx.Rollback(t.Context()))
+		}()
+
+		err = metadataRepo.Set(t.Context(), tx, &domain.OrganizationMetadata{
+			OrganizationID: orgID,
+			Metadata: domain.Metadata{
+				InstanceID: instanceID,
+				Key:        "urn:zitadel:key",
+				Value:      []byte("some-value"),
+			},
+		})
+		require.NoError(t, err)
+		afterSet := time.Now()
+
+		savedMetadata, err := metadataRepo.Get(
+			t.Context(), tx, database.WithCondition(
+				database.And(
+					metadataRepo.InstanceIDCondition(instanceID),
+					metadataRepo.OrganizationIDCondition(orgID),
+					metadataRepo.KeyCondition(database.TextOperationEqual, "urn:zitadel:key"),
+				),
+			),
+		)
+		require.NoError(t, err)
+		assert.Equal(t, orgID, savedMetadata.OrganizationID)
+		assert.Equal(t, instanceID, savedMetadata.InstanceID)
+		assert.Equal(t, "urn:zitadel:key", savedMetadata.Key)
+		assert.Equal(t, []byte("some-value"), savedMetadata.Value)
+		assert.WithinRange(t, savedMetadata.CreatedAt, beforeSet, afterSet)
+		assert.WithinRange(t, savedMetadata.UpdatedAt, beforeSet, afterSet)
+		assert.Equal(t, savedMetadata.CreatedAt, savedMetadata.UpdatedAt)
+
+		err = metadataRepo.Set(t.Context(), tx, &domain.OrganizationMetadata{
+			OrganizationID: orgID,
+			Metadata: domain.Metadata{
+				InstanceID: instanceID,
+				Key:        "urn:zitadel:key",
+				Value:      []byte("1234"),
+			},
+		})
+		require.NoError(t, err)
+		afterSet = time.Now()
+
+		savedMetadata, err = metadataRepo.Get(
+			t.Context(), tx, database.WithCondition(
+				database.And(
+					metadataRepo.InstanceIDCondition(instanceID),
+					metadataRepo.OrganizationIDCondition(orgID),
+					metadataRepo.KeyCondition(database.TextOperationEqual, "urn:zitadel:key"),
+				),
+			),
+		)
+		require.NoError(t, err)
+
+		assert.Equal(t, orgID, savedMetadata.OrganizationID)
+		assert.Equal(t, instanceID, savedMetadata.InstanceID)
+		assert.Equal(t, "urn:zitadel:key", savedMetadata.Key)
+		assert.Equal(t, []byte("1234"), savedMetadata.Value)
+		assert.WithinRange(t, savedMetadata.CreatedAt, beforeSet, afterSet)
+		assert.WithinRange(t, savedMetadata.UpdatedAt, beforeSet, afterSet)
+		// we cannot check if the updated at did change because we are in the same transaction, so we skip this check
+	})
+
+	t.Run("from events", func(t *testing.T) {
+		t.Run("create", func(t *testing.T) {
+			tx, err := tx.Begin(t.Context())
+			require.NoError(t, err)
+			defer func() {
+				require.NoError(t, tx.Rollback(t.Context()))
+			}()
+
+			err = metadataRepo.Set(t.Context(), tx, &domain.OrganizationMetadata{
+				OrganizationID: orgID,
+				Metadata: domain.Metadata{
+					InstanceID: instanceID,
+					Key:        "urn:zitadel:key",
+					Value:      []byte("some-value"),
+					CreatedAt:  beforeSet.Add(-time.Hour),
+					UpdatedAt:  beforeSet.Add(-time.Hour),
+				},
+			})
+			require.NoError(t, err)
+
+			savedMetadata, err := metadataRepo.Get(
+				t.Context(), tx, database.WithCondition(
+					database.And(
+						metadataRepo.InstanceIDCondition(instanceID),
+						metadataRepo.OrganizationIDCondition(orgID),
+						metadataRepo.KeyCondition(database.TextOperationEqual, "urn:zitadel:key"),
+					),
+				),
+			)
+			require.NoError(t, err)
+			assert.Equal(t, orgID, savedMetadata.OrganizationID)
+			assert.Equal(t, instanceID, savedMetadata.InstanceID)
+			assert.Equal(t, "urn:zitadel:key", savedMetadata.Key)
+			assert.Equal(t, []byte("some-value"), savedMetadata.Value)
+			assert.Less(t, savedMetadata.CreatedAt, beforeSet)
+			assert.Less(t, savedMetadata.UpdatedAt, beforeSet)
+			assert.Equal(t, savedMetadata.CreatedAt, savedMetadata.UpdatedAt)
+		})
+
+		t.Run("update", func(t *testing.T) {
+			tx, err := tx.Begin(t.Context())
+			require.NoError(t, err)
+			defer func() {
+				require.NoError(t, tx.Rollback(t.Context()))
+			}()
+
+			// first event
+			firstEventCreatedAt := beforeSet.Add(-2 * time.Hour).Round(time.Microsecond) // the timestamps are rounded because postgres does not store nanoseconds
+			err = metadataRepo.Set(t.Context(), tx, &domain.OrganizationMetadata{
+				OrganizationID: orgID,
+				Metadata: domain.Metadata{
+					InstanceID: instanceID,
+					Key:        "urn:zitadel:key",
+					Value:      []byte("some-value"),
+					CreatedAt:  firstEventCreatedAt,
+					UpdatedAt:  firstEventCreatedAt,
+				},
+			})
+			require.NoError(t, err)
+
+			// second event
+			secondEventCreatedAt := beforeSet.Add(-time.Hour).Round(time.Microsecond) // the timestamps are rounded because postgres does not store nanoseconds
+			savedMetadata := &domain.OrganizationMetadata{
+				OrganizationID: orgID,
+				Metadata: domain.Metadata{
+					InstanceID: instanceID,
+					Key:        "urn:zitadel:key",
+					Value:      []byte("some-other-value"),
+					CreatedAt:  secondEventCreatedAt,
+					UpdatedAt:  secondEventCreatedAt,
+				},
+			}
+			err = metadataRepo.Set(t.Context(), tx, savedMetadata)
+			require.NoError(t, err)
+
+			require.NoError(t, err)
+			assert.Equal(t, orgID, savedMetadata.OrganizationID)
+			assert.Equal(t, instanceID, savedMetadata.InstanceID)
+			assert.Equal(t, "urn:zitadel:key", savedMetadata.Key)
+			assert.Equal(t, []byte("some-other-value"), savedMetadata.Value)
+			assert.True(t, savedMetadata.CreatedAt.Equal(firstEventCreatedAt), "created at should have been %v, but was %v", firstEventCreatedAt, savedMetadata.CreatedAt)
+			assert.True(t, savedMetadata.UpdatedAt.Equal(secondEventCreatedAt), "updated at should have been %v, but was %v", secondEventCreatedAt, savedMetadata.UpdatedAt)
+		})
+	})
+
+	t.Run("invalid input", func(t *testing.T) {
+		for _, testCase := range []struct {
+			name        string
+			metadata    *domain.OrganizationMetadata
+			expectedErr error
+		}{
+			{
+				name: "missing organization_id",
+				metadata: &domain.OrganizationMetadata{
+					Metadata: domain.Metadata{
+						InstanceID: instanceID,
+						Key:        "urn:zitadel:key",
+						Value:      []byte("1234"),
+					},
+				},
+				expectedErr: new(database.ForeignKeyError),
+			},
+			{
+				name: "missing instance_id",
+				metadata: &domain.OrganizationMetadata{
+					OrganizationID: orgID,
+					Metadata: domain.Metadata{
+						Key:   "urn:zitadel:key",
+						Value: []byte("1234"),
+					},
+				},
+				expectedErr: new(database.ForeignKeyError),
+			},
+			{
+				name: "missing key",
+				metadata: &domain.OrganizationMetadata{
+					OrganizationID: orgID,
+					Metadata: domain.Metadata{
+						InstanceID: instanceID,
+						Value:      []byte("1234"),
+					},
+				},
+				expectedErr: new(database.CheckError),
+			},
+			{
+				name: "missing value",
+				metadata: &domain.OrganizationMetadata{
+					OrganizationID: orgID,
+					Metadata: domain.Metadata{
+						InstanceID: instanceID,
+						Key:        "urn:zitadel:key",
+					},
+				},
+				expectedErr: new(database.NotNullError),
+			},
+		} {
+			t.Run(testCase.name, func(t *testing.T) {
+				tx, err := tx.Begin(t.Context())
+				require.NoError(t, err)
+				defer func() {
+					require.NoError(t, tx.Rollback(t.Context()))
+				}()
+
+				err = metadataRepo.Set(t.Context(), tx, testCase.metadata)
+				assert.ErrorIs(t, err, testCase.expectedErr)
+			})
+		}
+	})
+}
+
+func TestRemoveOrganizationMetadata(t *testing.T) {
+	tx, err := pool.Begin(t.Context(), nil)
+	require.NoError(t, err)
+	defer func() {
+		err = tx.Rollback(t.Context())
+		if err != nil {
+			t.Log("error during rollback:", err)
+		}
+	}()
+
+	instanceRepo := repository.InstanceRepository()
+	orgRepo := repository.OrganizationRepository()
+	metadataRepo := repository.OrganizationMetadataRepository()
+
+	// create instance
+	instanceID := gofakeit.UUID()
+	instance := domain.Instance{
+		ID:              instanceID,
+		Name:            gofakeit.Name(),
+		DefaultOrgID:    "defaultOrgId",
+		IAMProjectID:    "iamProject",
+		ConsoleClientID: "consoleClient",
+		ConsoleAppID:    "consoleApp",
+		DefaultLanguage: "defaultLanguage",
+	}
+	err = instanceRepo.Create(t.Context(), tx, &instance)
+	require.NoError(t, err)
+
+	// create organization
+	orgA := domain.Organization{
+		ID:         "1",
+		Name:       gofakeit.Name(),
+		InstanceID: instanceID,
+		State:      domain.OrgStateActive,
+	}
+	err = orgRepo.Create(t.Context(), tx, &orgA)
+	require.NoError(t, err)
+
+	err = metadataRepo.Set(
+		t.Context(), tx,
+		&domain.OrganizationMetadata{
+			OrganizationID: orgA.ID,
+			Metadata: domain.Metadata{
+				InstanceID: instanceID,
+				Key:        "urn:zitadel:key",
+				Value:      []byte("1234"),
+			},
+		},
+		&domain.OrganizationMetadata{
+			OrganizationID: orgA.ID,
+			Metadata: domain.Metadata{
+				InstanceID: instanceID,
+				Key:        "urn:zitadel:key2",
+				Value:      []byte("asdf"),
+			},
+		},
+	)
+	require.NoError(t, err)
+
+	orgB := domain.Organization{
+		ID:         "2",
+		Name:       gofakeit.Name(),
+		InstanceID: instanceID,
+		State:      domain.OrgStateActive,
+	}
+	err = orgRepo.Create(t.Context(), tx, &orgB)
+	require.NoError(t, err)
+
+	err = metadataRepo.Set(
+		t.Context(), tx,
+		&domain.OrganizationMetadata{
+			OrganizationID: orgB.ID,
+			Metadata: domain.Metadata{
+				InstanceID: instanceID,
+				Key:        "urn:zitadel:key",
+				Value:      []byte("5678"),
+			},
+		},
+		&domain.OrganizationMetadata{
+			OrganizationID: orgB.ID,
+			Metadata: domain.Metadata{
+				InstanceID: instanceID,
+				Key:        "urn:zitadel:key2",
+				Value:      []byte(`"asdf"`),
+			},
+		},
+	)
+	require.NoError(t, err)
+
+	t.Run("without instance condition", func(t *testing.T) {
+		affected, err := metadataRepo.Remove(
+			t.Context(), tx,
+			metadataRepo.OrganizationIDCondition(orgA.ID),
+		)
+		assert.ErrorIs(t, err, new(database.MissingConditionError))
+		assert.Equal(t, int64(0), affected)
+	})
+
+	t.Run("without org condition", func(t *testing.T) {
+		affected, err := metadataRepo.Remove(
+			t.Context(), tx,
+			metadataRepo.InstanceIDCondition("non-existing"),
+		)
+		assert.ErrorIs(t, err, new(database.MissingConditionError))
+		assert.Equal(t, int64(0), affected)
+	})
+
+	t.Run("successful", func(t *testing.T) {
+		tx, err := tx.Begin(t.Context())
+		require.NoError(t, err)
+		defer func() {
+			require.NoError(t, tx.Rollback(t.Context()))
+		}()
+		affected, err := metadataRepo.Remove(
+			t.Context(), tx,
+			database.And(
+				metadataRepo.InstanceIDCondition(instanceID),
+				metadataRepo.OrganizationIDCondition(orgA.ID),
+				metadataRepo.KeyCondition(database.TextOperationStartsWith, "urn:zitadel:key"),
+			),
+		)
+		require.NoError(t, err)
+		assert.Equal(t, int64(2), affected)
+	})
+}

backend/v3/storage/database/repository/org_test.go

@@ -1132,4 +1132,75 @@ func TestGetOrganizationWithSubResources(t *testing.T) {
 			assert.Len(t, returnedOrg.Domains, 0)
 		})
 	})
+
+	t.Run("metadata", func(t *testing.T) {
+		metadataRepo := repository.OrganizationMetadataRepository()
+
+		metadata := []*domain.OrganizationMetadata{
+			{
+				OrganizationID: org.ID,
+				Metadata: domain.Metadata{
+					InstanceID: org.InstanceID,
+					Key:        "key1",
+					Value:      []byte("value1"),
+				},
+			},
+			{
+				OrganizationID: org.ID,
+				Metadata: domain.Metadata{
+					InstanceID: org.InstanceID,
+					Key:        "key2",
+					Value:      []byte("value2"),
+				},
+			},
+		}
+		err = metadataRepo.Set(t.Context(), tx, metadata...)
+		require.NoError(t, err)
+
+		t.Run("org by metadata key", func(t *testing.T) {
+			orgRepo := orgRepo.LoadMetadata()
+
+			returnedOrg, err := orgRepo.Get(t.Context(), tx,
+				database.WithCondition(
+					database.And(
+						orgRepo.InstanceIDCondition(instanceId),
+						orgRepo.ExistsMetadata(metadataRepo.KeyCondition(database.TextOperationEqual, "key1")),
+					),
+				),
+			)
+			require.NoError(t, err)
+			assert.Equal(t, org.ID, returnedOrg.ID)
+			assert.Len(t, returnedOrg.Metadata, 2)
+		})
+
+		t.Run("org by metadata value", func(t *testing.T) {
+			orgRepo := orgRepo.LoadMetadata()
+
+			returnedOrg, err := orgRepo.Get(t.Context(), tx,
+				database.WithCondition(
+					database.And(
+						orgRepo.InstanceIDCondition(instanceId),
+						orgRepo.ExistsMetadata(metadataRepo.ValueCondition(database.BytesOperationEqual, []byte("value1"))),
+					),
+				),
+			)
+			require.NoError(t, err)
+			assert.Equal(t, org.ID, returnedOrg.ID)
+			assert.Len(t, returnedOrg.Metadata, 2)
+		})
+
+		t.Run("ensure org by metadata key works without LoadMetadata", func(t *testing.T) {
+			returnedOrg, err := orgRepo.Get(t.Context(), tx,
+				database.WithCondition(
+					database.And(
+						orgRepo.InstanceIDCondition(instanceId),
+						orgRepo.ExistsMetadata(metadataRepo.KeyCondition(database.TextOperationEqual, "key1")),
+					),
+				),
+			)
+			require.NoError(t, err)
+			assert.Equal(t, org.ID, returnedOrg.ID)
+			assert.Len(t, returnedOrg.Metadata, 0)
+		})
+	})
 }

internal/query/projection/org_metadata_relational.go

@@ -0,0 +1,114 @@
+package projection
+
+import (
+	"context"
+	"database/sql"
+
+	"github.com/zitadel/zitadel/backend/v3/domain"
+	"github.com/zitadel/zitadel/backend/v3/storage/database"
+	v3_sql "github.com/zitadel/zitadel/backend/v3/storage/database/dialect/sql"
+	"github.com/zitadel/zitadel/backend/v3/storage/database/repository"
+	"github.com/zitadel/zitadel/internal/eventstore"
+	"github.com/zitadel/zitadel/internal/eventstore/handler/v2"
+	"github.com/zitadel/zitadel/internal/repository/org"
+	"github.com/zitadel/zitadel/internal/zerrors"
+)
+
+type orgMetadataRelationalProjection struct{}
+
+func newOrgMetadataRelationalProjection(ctx context.Context, config handler.Config) *handler.Handler {
+	return handler.NewHandler(ctx, &config, new(orgMetadataRelationalProjection))
+}
+
+func (*orgMetadataRelationalProjection) Name() string {
+	return "zitadel.org_metadata"
+}
+
+func (p *orgMetadataRelationalProjection) Reducers() []handler.AggregateReducer {
+	return []handler.AggregateReducer{
+		{
+			Aggregate: org.AggregateType,
+			EventReducers: []handler.EventReducer{
+				{
+					Event:  org.MetadataSetType,
+					Reduce: p.reduceSet,
+				},
+				{
+					Event:  org.MetadataRemovedType,
+					Reduce: p.reduceRemoved,
+				},
+				// This event cannot be tested because it was never used in the past
+				{
+					Event:  org.MetadataRemovedAllType,
+					Reduce: p.reduceRemovedAll,
+				},
+			},
+		},
+	}
+}
+
+func (p *orgMetadataRelationalProjection) reduceSet(event eventstore.Event) (*handler.Statement, error) {
+	e, ok := event.(*org.MetadataSetEvent)
+	if !ok {
+		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-xOO4e", "reduce.wrong.event.type %s", org.MetadataSetType)
+	}
+	return handler.NewStatement(e, func(ctx context.Context, ex handler.Executer, projectionName string) error {
+		tx, ok := ex.(*sql.Tx)
+		if !ok {
+			return zerrors.ThrowInvalidArgumentf(nil, "HANDL-xg4IJ", "reduce.wrong.db.pool %T", ex)
+		}
+		return repository.OrganizationMetadataRepository().Set(ctx, v3_sql.SQLTx(tx), &domain.OrganizationMetadata{
+			Metadata: domain.Metadata{
+				InstanceID: e.Aggregate().InstanceID,
+				Key:        e.Key,
+				Value:      e.Value,
+				CreatedAt:  e.CreationDate(),
+				UpdatedAt:  e.CreationDate(),
+			},
+			OrganizationID: e.Aggregate().ResourceOwner,
+		})
+	}), nil
+}
+
+func (p *orgMetadataRelationalProjection) reduceRemoved(event eventstore.Event) (*handler.Statement, error) {
+	e, ok := event.(*org.MetadataRemovedEvent)
+	if !ok {
+		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-XE6TF", "reduce.wrong.event.type %s", org.MetadataRemovedType)
+	}
+	return handler.NewStatement(e, func(ctx context.Context, ex handler.Executer, projectionName string) error {
+		tx, ok := ex.(*sql.Tx)
+		if !ok {
+			return zerrors.ThrowInvalidArgumentf(nil, "HANDL-QKMlz", "reduce.wrong.db.pool %T", ex)
+		}
+		domainRepo := repository.OrganizationMetadataRepository()
+		_, err := domainRepo.Remove(ctx, v3_sql.SQLTx(tx),
+			database.And(
+				domainRepo.InstanceIDCondition(e.Aggregate().InstanceID),
+				domainRepo.OrganizationIDCondition(e.Aggregate().ResourceOwner),
+				domainRepo.KeyCondition(database.TextOperationEqual, e.Key),
+			),
+		)
+		return err
+	}), nil
+}
+
+func (p *orgMetadataRelationalProjection) reduceRemovedAll(event eventstore.Event) (*handler.Statement, error) {
+	e, ok := event.(*org.MetadataRemovedAllEvent)
+	if !ok {
+		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-EmyAe", "reduce.wrong.event.type %s", org.MetadataRemovedAllType)
+	}
+	return handler.NewStatement(e, func(ctx context.Context, ex handler.Executer, projectionName string) error {
+		tx, ok := ex.(*sql.Tx)
+		if !ok {
+			return zerrors.ThrowInvalidArgumentf(nil, "HANDL-hjEHg", "reduce.wrong.db.pool %T", ex)
+		}
+		domainRepo := repository.OrganizationMetadataRepository()
+		_, err := domainRepo.Remove(ctx, v3_sql.SQLTx(tx),
+			database.And(
+				domainRepo.InstanceIDCondition(e.Aggregate().InstanceID),
+				domainRepo.OrganizationIDCondition(e.Aggregate().ResourceOwner),
+			),
+		)
+		return err
+	}), nil
+}

internal/query/projection/projection.go

@@ -89,13 +89,14 @@ var (
 	HostedLoginTranslationProjection    *handler.Handler
 	OrganizationSettingsProjection      *handler.Handler
 
-	InstanceRelationalProjection           *handler.Handler
-	OrganizationRelationalProjection       *handler.Handler
-	InstanceDomainRelationalProjection     *handler.Handler
-	OrganizationDomainRelationalProjection *handler.Handler
-	IDPTemplateRelationalProjection        *handler.Handler
-	ProjectRelationalProjection            *handler.Handler
-	ProjectRoleRelationalProjection        *handler.Handler
+	InstanceRelationalProjection             *handler.Handler
+	OrganizationRelationalProjection         *handler.Handler
+	InstanceDomainRelationalProjection       *handler.Handler
+	OrganizationDomainRelationalProjection   *handler.Handler
+	IDPTemplateRelationalProjection          *handler.Handler
+	ProjectRelationalProjection              *handler.Handler
+	ProjectRoleRelationalProjection          *handler.Handler
+	OrganizationMetadataRelationalProjection *handler.Handler
 
 	ProjectGrantFields      *handler.FieldHandler
 	OrgDomainVerifiedFields *handler.FieldHandler
@@ -218,6 +219,7 @@ func Create(ctx context.Context, sqlClient *database.DB, es handler.EventStore,
 	IDPTemplateRelationalProjection = newIDPTemplateRelationalProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["idp_templates_relational"]))
 	ProjectRelationalProjection = newProjectRelationalProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["projects_relational"]))
 	ProjectRoleRelationalProjection = newProjectRoleRelationalProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["project_roles_relational"]))
+	OrganizationMetadataRelationalProjection = newOrgMetadataRelationalProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["organization_metadata_relational"]))
 
 	newProjectionsList()
 	newFieldsList()
@@ -409,5 +411,6 @@ func newProjectionsList() {
 		IDPTemplateRelationalProjection,
 		ProjectRelationalProjection,
 		ProjectRoleRelationalProjection,
+		OrganizationMetadataRelationalProjection,
 	}
 }