show tim

backend/handler/handle.go

@@ -0,0 +1,114 @@
+package handler
+
+import (
+	"context"
+)
+
+// Handle is a function that handles the in.
+type Handle[Out, In any] func(ctx context.Context, in Out) (out In, err error)
+
+// Middleware is a function that decorates the handle function.
+// It must call the handle function but its up the the middleware to decide when and how.
+type Middleware[In, Out any] func(ctx context.Context, in In, handle Handle[In, Out]) (out Out, err error)
+
+// Chain chains the handle function with the next handler.
+// The next handler is called after the handle function.
+func Chain[In, Out any](handle Handle[In, Out], next Handle[Out, Out]) Handle[In, Out] {
+	return func(ctx context.Context, in In) (out Out, err error) {
+		out, err = handle(ctx, in)
+		if err != nil {
+			return out, err
+		}
+		return next(ctx, out)
+	}
+}
+
+// Chains chains the handle function with the next handlers.
+// The next handlers are called after the handle function.
+// The order of the handlers is preserved.
+func Chains[In, Out any](handle Handle[In, Out], chain ...Handle[Out, Out]) Handle[In, Out] {
+	return func(ctx context.Context, in In) (out Out, err error) {
+		for _, next := range chain {
+			handle = Chain(handle, next)
+		}
+		return handle(ctx, in)
+	}
+}
+
+// Decorate decorates the handle function with the decorate function.
+// The decorate function is called before the handle function.
+func Decorate[In, Out any](handle Handle[In, Out], decorate Middleware[In, Out]) Handle[In, Out] {
+	return func(ctx context.Context, in In) (out Out, err error) {
+		return decorate(ctx, in, handle)
+	}
+}
+
+// Decorates decorates the handle function with the decorate functions.
+// The decorates function is called before the handle function.
+func Decorates[In, Out any](handle Handle[In, Out], decorates ...Middleware[In, Out]) Handle[In, Out] {
+	return func(ctx context.Context, in In) (out Out, err error) {
+		for i := len(decorates) - 1; i >= 0; i-- {
+			handle = Decorate(handle, decorates[i])
+		}
+		return handle(ctx, in)
+	}
+}
+
+// SkipNext skips the next handler if the handle function returns a non-empty output or an error.
+func SkipNext[In, Out any](handle Handle[In, Out], next Handle[In, Out]) Handle[In, Out] {
+	return func(ctx context.Context, in In) (out Out, err error) {
+		var empty Out
+		out, err = handle(ctx, in)
+		// TODO: does this work?
+		if any(out) != any(empty) || err != nil {
+			return out, err
+		}
+		return next(ctx, in)
+	}
+}
+
+// SkipNilHandler skips the handle function if the handler is nil.
+// If handle is nil, an empty output is returned.
+// The function is safe to call with nil handler.
+func SkipNilHandler[O, In, Out any](handler *O, handle Handle[In, Out]) Handle[In, Out] {
+	return func(ctx context.Context, in In) (out Out, err error) {
+		if handler == nil {
+			return out, nil
+		}
+		return handle(ctx, in)
+	}
+}
+
+// SkipReturnPreviousHandler skips the handle function if the handler is nil and returns the input.
+// The function is safe to call with nil handler.
+func SkipReturnPreviousHandler[O, In any](handler *O, handle Handle[In, In]) Handle[In, In] {
+	return func(ctx context.Context, in In) (out In, err error) {
+		if handler == nil {
+			return in, nil
+		}
+		return handle(ctx, in)
+	}
+}
+
+func ResFuncToHandle[In any, Out any](fn func(context.Context, In) Out) Handle[In, Out] {
+	return func(ctx context.Context, in In) (out Out, err error) {
+		return fn(ctx, in), nil
+	}
+}
+
+func ErrFuncToHandle[In any](fn func(context.Context, In) error) Handle[In, In] {
+	return func(ctx context.Context, in In) (out In, err error) {
+		err = fn(ctx, in)
+		if err != nil {
+			return out, err
+		}
+		return in, nil
+	}
+}
+
+func NoReturnToHandle[In any](fn func(context.Context, In)) Handle[In, In] {
+	return func(ctx context.Context, in In) (out In, err error) {
+		fn(ctx, in)
+		return in, nil
+	}
+}

backend/repository/database.go

@@ -0,0 +1,19 @@
+package repository
+
+import "github.com/zitadel/zitadel/backend/storage/database"
+
+type executor struct {
+	client database.Executor
+}
+
+func execute(client database.Executor) *executor {
+	return &executor{client: client}
+}
+
+type querier struct {
+	client database.Querier
+}
+
+func query(client database.Querier) *querier {
+	return &querier{client: client}
+}

backend/repository/event.go

@@ -0,0 +1,16 @@
+package repository
+
+import (
+	"github.com/zitadel/zitadel/backend/storage/database"
+	"github.com/zitadel/zitadel/backend/storage/eventstore"
+)
+
+type eventStore struct {
+	es *eventstore.Eventstore
+}
+
+func events(client database.Executor) *eventStore {
+	return &eventStore{
+		es: eventstore.New(client),
+	}
+}

backend/repository/instance_cache.go

@@ -0,0 +1,58 @@
+package repository
+
+import (
+	"context"
+	"log"
+
+	"github.com/zitadel/zitadel/backend/storage/cache"
+)
+
+type InstanceCache struct {
+	cache.Cache[InstanceIndex, string, *Instance]
+}
+
+type InstanceIndex uint8
+
+var InstanceIndices = []InstanceIndex{
+	InstanceByID,
+	InstanceByDomain,
+}
+
+const (
+	InstanceByID InstanceIndex = iota
+	InstanceByDomain
+)
+
+var _ cache.Entry[InstanceIndex, string] = (*Instance)(nil)
+
+// Keys implements [cache.Entry].
+func (i *Instance) Keys(index InstanceIndex) (key []string) {
+	switch index {
+	case InstanceByID:
+		return []string{i.ID}
+	case InstanceByDomain:
+		return []string{i.Name}
+	}
+	return nil
+}
+
+func NewInstanceCache(c cache.Cache[InstanceIndex, string, *Instance]) *InstanceCache {
+	return &InstanceCache{c}
+}
+
+func (i *InstanceCache) ByID(ctx context.Context, id string) *Instance {
+	log.Println("cached.instance.byID")
+	instance, _ := i.Cache.Get(ctx, InstanceByID, id)
+	return instance
+}
+
+func (i *InstanceCache) ByDomain(ctx context.Context, domain string) *Instance {
+	log.Println("cached.instance.byDomain")
+	instance, _ := i.Cache.Get(ctx, InstanceByDomain, domain)
+	return instance
+}
+
+func (i *InstanceCache) Set(ctx context.Context, instance *Instance) {
+	log.Println("cached.instance.set")
+	i.Cache.Set(ctx, instance)
+}

backend/repository/instance_db.go

@@ -0,0 +1,61 @@
+package repository
+
+import (
+	"context"
+	"log"
+)
+
+const InstanceByIDStmt = `SELECT id, name FROM instances WHERE id = $1`
+
+func (q *querier) InstanceByID(ctx context.Context, id string) (*Instance, error) {
+	log.Println("sql.instance.byID")
+	row := q.client.QueryRow(ctx, InstanceByIDStmt, id)
+	var instance Instance
+	if err := row.Scan(&instance.ID, &instance.Name); err != nil {
+		return nil, err
+	}
+	return &instance, nil
+}
+
+const instanceByDomainQuery = `SELECT i.id, i.name FROM instances i JOIN instance_domains id ON i.id = id.instance_id WHERE id.domain = $1`
+
+func (q *querier) InstanceByDomain(ctx context.Context, domain string) (*Instance, error) {
+	log.Println("sql.instance.byDomain")
+	row := q.client.QueryRow(ctx, instanceByDomainQuery, domain)
+	var instance Instance
+	if err := row.Scan(&instance.ID, &instance.Name); err != nil {
+		return nil, err
+	}
+	return &instance, nil
+}
+
+func (q *querier) ListInstances(ctx context.Context, request *ListRequest) (res []*Instance, err error) {
+	log.Println("sql.instance.list")
+	rows, err := q.client.Query(ctx, "SELECT id, name FROM instances")
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var instance Instance
+		if err = rows.Scan(&instance.ID, &instance.Name); err != nil {
+			return nil, err
+		}
+		res = append(res, &instance)
+	}
+	if err = rows.Err(); err != nil {
+		return nil, err
+	}
+	return res, nil
+}
+
+const InstanceCreateStmt = `INSERT INTO instances (id, name) VALUES ($1, $2)`
+
+func (e *executor) CreateInstance(ctx context.Context, instance *Instance) (*Instance, error) {
+	log.Println("sql.instance.create")
+	err := e.client.Exec(ctx, InstanceCreateStmt, instance.ID, instance.Name)
+	if err != nil {
+		return nil, err
+	}
+	return instance, nil
+}

backend/repository/instance_event.go

@@ -0,0 +1,15 @@
+package repository
+
+import (
+	"context"
+	"log"
+)
+
+func (s *eventStore) CreateInstance(ctx context.Context, instance *Instance) (*Instance, error) {
+	log.Println("event.instance.create")
+	err := s.es.Push(ctx, instance)
+	if err != nil {
+		return nil, err
+	}
+	return instance, nil
+}

backend/repository/instance_test.go

@@ -0,0 +1,258 @@
+package repository_test
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"os"
+	"reflect"
+	"testing"
+
+	"github.com/zitadel/zitadel/backend/repository"
+	"github.com/zitadel/zitadel/backend/storage/cache"
+	"github.com/zitadel/zitadel/backend/storage/cache/connector/gomap"
+	"github.com/zitadel/zitadel/backend/storage/database"
+	"github.com/zitadel/zitadel/backend/storage/database/mock"
+	"github.com/zitadel/zitadel/backend/telemetry/logging"
+	"github.com/zitadel/zitadel/backend/telemetry/tracing"
+)
+
+func Test_instance_Create(t *testing.T) {
+	type args struct {
+		ctx      context.Context
+		tx       database.Transaction
+		instance *repository.Instance
+	}
+	tests := []struct {
+		name    string
+		opts    []repository.Option[repository.InstanceOptions]
+		args    args
+		want    *repository.Instance
+		wantErr bool
+	}{
+		{
+			name: "simple",
+			opts: []repository.Option[repository.InstanceOptions]{
+				repository.WithTracer[repository.InstanceOptions](tracing.NewTracer("test")),
+				repository.WithLogger[repository.InstanceOptions](logging.New(slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug})))),
+				repository.WithInstanceCache(
+					repository.NewInstanceCache(gomap.NewCache[repository.InstanceIndex, string, *repository.Instance](context.Background(), repository.InstanceIndices, cache.Config{})),
+				),
+			},
+			args: args{
+				ctx: context.Background(),
+				tx:  mock.NewTransaction(t, mock.ExpectExec(repository.InstanceCreateStmt, "ID", "Name")),
+				instance: &repository.Instance{
+					ID:   "ID",
+					Name: "Name",
+				},
+			},
+			want: &repository.Instance{
+				ID:   "ID",
+				Name: "Name",
+			},
+			wantErr: false,
+		},
+		{
+			name: "without cache",
+			opts: []repository.Option[repository.InstanceOptions]{
+				repository.WithTracer[repository.InstanceOptions](tracing.NewTracer("test")),
+				repository.WithLogger[repository.InstanceOptions](logging.New(slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug})))),
+			},
+			args: args{
+				ctx: context.Background(),
+				tx:  mock.NewTransaction(t, mock.ExpectExec(repository.InstanceCreateStmt, "ID", "Name")),
+				instance: &repository.Instance{
+					ID:   "ID",
+					Name: "Name",
+				},
+			},
+			want: &repository.Instance{
+				ID:   "ID",
+				Name: "Name",
+			},
+			wantErr: false,
+		},
+		{
+			name: "without cache, tracer",
+			opts: []repository.Option[repository.InstanceOptions]{
+				repository.WithLogger[repository.InstanceOptions](logging.New(slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug})))),
+			},
+			args: args{
+				ctx: context.Background(),
+				tx:  mock.NewTransaction(t, mock.ExpectExec(repository.InstanceCreateStmt, "ID", "Name")),
+				instance: &repository.Instance{
+					ID:   "ID",
+					Name: "Name",
+				},
+			},
+			want: &repository.Instance{
+				ID:   "ID",
+				Name: "Name",
+			},
+			wantErr: false,
+		},
+		{
+			name: "without cache, tracer, logger",
+			args: args{
+				ctx: context.Background(),
+				tx:  mock.NewTransaction(t, mock.ExpectExec(repository.InstanceCreateStmt, "ID", "Name")),
+				instance: &repository.Instance{
+					ID:   "ID",
+					Name: "Name",
+				},
+			},
+			want: &repository.Instance{
+				ID:   "ID",
+				Name: "Name",
+			},
+			wantErr: false,
+		},
+		{
+			name: "without cache, tracer, logger, eventStore",
+			args: args{
+				ctx: context.Background(),
+				tx:  mock.NewTransaction(t, mock.ExpectExec(repository.InstanceCreateStmt, "ID", "Name")),
+				instance: &repository.Instance{
+					ID:   "ID",
+					Name: "Name",
+				},
+			},
+			want: &repository.Instance{
+				ID:   "ID",
+				Name: "Name",
+			},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			fmt.Printf("------------------------ %s ------------------------\n", tt.name)
+			i := repository.NewInstance(tt.opts...)
+			got, err := i.Create(tt.args.ctx, tt.args.tx, tt.args.instance)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("instance.Create() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("instance.Create() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_instance_ByID(t *testing.T) {
+	type args struct {
+		ctx context.Context
+		tx  database.Transaction
+		id  string
+	}
+	tests := []struct {
+		name    string
+		opts    []repository.Option[repository.InstanceOptions]
+		args    args
+		want    *repository.Instance
+		wantErr bool
+	}{
+		{
+			name: "simple, not cached",
+			opts: []repository.Option[repository.InstanceOptions]{
+				repository.WithTracer[repository.InstanceOptions](tracing.NewTracer("test")),
+				repository.WithLogger[repository.InstanceOptions](logging.New(slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug})))),
+				repository.WithInstanceCache(
+					repository.NewInstanceCache(gomap.NewCache[repository.InstanceIndex, string, *repository.Instance](context.Background(), repository.InstanceIndices, cache.Config{})),
+				),
+			},
+			args: args{
+				ctx: context.Background(),
+				tx: mock.NewTransaction(t,
+					mock.ExpectQueryRow(mock.NewRow(t, "id", "Name"), repository.InstanceByIDStmt, "id"),
+				),
+				id: "id",
+			},
+			want: &repository.Instance{
+				ID:   "id",
+				Name: "Name",
+			},
+			wantErr: false,
+		},
+		{
+			name: "simple, cached",
+			opts: []repository.Option[repository.InstanceOptions]{
+				repository.WithTracer[repository.InstanceOptions](tracing.NewTracer("test")),
+				repository.WithLogger[repository.InstanceOptions](logging.New(slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug})))),
+				repository.WithInstanceCache(
+					func() *repository.InstanceCache {
+						c := repository.NewInstanceCache(gomap.NewCache[repository.InstanceIndex, string, *repository.Instance](context.Background(), repository.InstanceIndices, cache.Config{}))
+						c.Set(context.Background(), &repository.Instance{
+							ID:   "id",
+							Name: "Name",
+						})
+						return c
+					}(),
+				),
+			},
+			args: args{
+				ctx: context.Background(),
+				tx: mock.NewTransaction(t,
+					mock.ExpectQueryRow(mock.NewRow(t, "id", "Name"), repository.InstanceByIDStmt, "id"),
+				),
+				id: "id",
+			},
+			want: &repository.Instance{
+				ID:   "id",
+				Name: "Name",
+			},
+			wantErr: false,
+		},
+		// {
+		// 	name: "without cache, tracer",
+		// 	opts: []repository.Option[repository.InstanceOptions]{
+		// 		repository.WithLogger[repository.InstanceOptions](logging.New(slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug})))),
+		// 	},
+		// 	args: args{
+		// 		ctx: context.Background(),
+		// 		tx:  mock.NewTransaction(),
+		// 		id: &repository.Instance{
+		// 			ID:   "ID",
+		// 			Name: "Name",
+		// 		},
+		// 	},
+		// 	want: &repository.Instance{
+		// 		ID:   "ID",
+		// 		Name: "Name",
+		// 	},
+		// 	wantErr: false,
+		// },
+		// {
+		// 	name: "without cache, tracer, logger",
+		// 	args: args{
+		// 		ctx: context.Background(),
+		// 		tx:  mock.NewTransaction(),
+		// 		id: &repository.Instance{
+		// 			ID:   "ID",
+		// 			Name: "Name",
+		// 		},
+		// 	},
+		// 	want: &repository.Instance{
+		// 		ID:   "ID",
+		// 		Name: "Name",
+		// 	},
+		// 	wantErr: false,
+		// },
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			fmt.Printf("------------------------ %s ------------------------\n", tt.name)
+			i := repository.NewInstance(tt.opts...)
+			got, err := i.ByID(tt.args.ctx, tt.args.tx, tt.args.id)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("instance.ByID() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("instance.ByID() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

backend/repository/option.go

@@ -0,0 +1,35 @@
+package repository
+
+import (
+	"github.com/zitadel/zitadel/backend/telemetry/logging"
+	"github.com/zitadel/zitadel/backend/telemetry/tracing"
+)
+
+// options are the default options for orchestrators.
+type options[T any] struct {
+	custom T
+	defaultOptions
+}
+
+type defaultOptions struct {
+	tracer *tracing.Tracer
+	logger *logging.Logger
+}
+
+type Option[T any] func(*options[T])
+
+func WithTracer[T any](tracer *tracing.Tracer) Option[T] {
+	return func(o *options[T]) {
+		o.tracer = tracer
+	}
+}
+
+func WithLogger[T any](logger *logging.Logger) Option[T] {
+	return func(o *options[T]) {
+		o.logger = logger
+	}
+}
+
+func (o Option[T]) apply(opts *options[T]) {
+	o(opts)
+}

backend/repository/user_cache.go

@@ -0,0 +1,52 @@
+package repository
+
+import (
+	"context"
+	"log"
+
+	"github.com/zitadel/zitadel/backend/storage/cache"
+)
+
+type UserCache struct {
+	cache.Cache[UserIndex, string, *User]
+}
+
+type UserIndex uint8
+
+var UserIndices = []UserIndex{
+	UserByIDIndex,
+	UserByUsernameIndex,
+}
+
+const (
+	UserByIDIndex UserIndex = iota
+	UserByUsernameIndex
+)
+
+var _ cache.Entry[UserIndex, string] = (*User)(nil)
+
+// Keys implements [cache.Entry].
+func (u *User) Keys(index UserIndex) (key []string) {
+	switch index {
+	case UserByIDIndex:
+		return []string{u.ID}
+	case UserByUsernameIndex:
+		return []string{u.Username}
+	}
+	return nil
+}
+
+func NewUserCache(c cache.Cache[UserIndex, string, *User]) *UserCache {
+	return &UserCache{c}
+}
+
+func (c *UserCache) ByID(ctx context.Context, id string) *User {
+	log.Println("cached.user.byID")
+	user, _ := c.Cache.Get(ctx, UserByIDIndex, id)
+	return user
+}
+
+func (c *UserCache) Set(ctx context.Context, user *User) {
+	log.Println("cached.user.set")
+	c.Cache.Set(ctx, user)
+}

backend/repository/user_db.go

@@ -0,0 +1,27 @@
+package repository
+
+import (
+	"context"
+	"log"
+)
+
+const userByIDQuery = `SELECT id, username FROM users WHERE id = $1`
+
+func (q *querier) UserByID(ctx context.Context, id string) (res *User, err error) {
+	log.Println("sql.user.byID")
+	row := q.client.QueryRow(ctx, userByIDQuery, id)
+	var user User
+	if err := row.Scan(&user.ID, &user.Username); err != nil {
+		return nil, err
+	}
+	return &user, nil
+}
+
+func (e *executor) CreateUser(ctx context.Context, user *User) (res *User, err error) {
+	log.Println("sql.user.create")
+	err = e.client.Exec(ctx, "INSERT INTO users (id, username) VALUES ($1, $2)", user.ID, user.Username)
+	if err != nil {
+		return nil, err
+	}
+	return user, nil
+}

backend/repository/user_event.go

@@ -0,0 +1,15 @@
+package repository
+
+import (
+	"context"
+	"log"
+)
+
+func (s *eventStore) CreateUser(ctx context.Context, user *User) (*User, error) {
+	log.Println("event.user.create")
+	err := s.es.Push(ctx, user)
+	if err != nil {
+		return nil, err
+	}
+	return user, nil
+}