mirror of
https://github.com/zitadel/zitadel.git
synced 2025-08-11 15:17:33 +00:00
chore(queue): use schema config instead of search_path
and application_name
to configure the database schema (#10075)
Removes manual schema and application name setup via raw SQL and switches to using River’s built-in schema configuration. # Which Problems Are Solved River provides a configuration flag to set the schema of the queue. Zitadel sets the schema through database statements which is not needed anymore. # How the Problems Are Solved Set the schema in the river configuration and removed old code
This commit is contained in:
@@ -1,45 +1,6 @@
|
|||||||
package queue
|
package queue
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"sync"
|
|
||||||
|
|
||||||
"github.com/jackc/pgx/v5"
|
|
||||||
|
|
||||||
"github.com/zitadel/zitadel/internal/database/dialect"
|
|
||||||
)
|
|
||||||
|
|
||||||
const (
|
const (
|
||||||
schema = "queue"
|
schema = "queue"
|
||||||
applicationName = "zitadel_queue"
|
applicationName = "zitadel_queue"
|
||||||
)
|
)
|
||||||
|
|
||||||
var conns = &sync.Map{}
|
|
||||||
|
|
||||||
type queueKey struct{}
|
|
||||||
|
|
||||||
func WithQueue(parent context.Context) context.Context {
|
|
||||||
return context.WithValue(parent, queueKey{}, struct{}{})
|
|
||||||
}
|
|
||||||
|
|
||||||
func init() {
|
|
||||||
dialect.RegisterBeforeAcquire(func(ctx context.Context, c *pgx.Conn) error {
|
|
||||||
if _, ok := ctx.Value(queueKey{}).(struct{}); !ok {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
_, err := c.Exec(ctx, "SET search_path TO "+schema+"; SET application_name TO "+applicationName)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
conns.Store(c, struct{}{})
|
|
||||||
return nil
|
|
||||||
})
|
|
||||||
dialect.RegisterAfterRelease(func(c *pgx.Conn) error {
|
|
||||||
_, ok := conns.LoadAndDelete(c)
|
|
||||||
if !ok {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
_, err := c.Exec(context.Background(), "SET search_path TO DEFAULT; SET application_name TO "+dialect.DefaultAppName)
|
|
||||||
return err
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
@@ -27,11 +27,10 @@ func (m *Migrator) Execute(ctx context.Context) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
migrator, err := rivermigrate.New(m.driver, nil)
|
migrator, err := rivermigrate.New(m.driver, &rivermigrate.Config{Schema: schema})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
ctx = WithQueue(ctx)
|
|
||||||
_, err = migrator.Migrate(ctx, rivermigrate.DirectionUp, nil)
|
_, err = migrator.Migrate(ctx, rivermigrate.DirectionUp, nil)
|
||||||
return err
|
return err
|
||||||
|
|
||||||
|
@@ -41,6 +41,7 @@ func NewQueue(config *Config) (_ *Queue, err error) {
|
|||||||
Queues: make(map[string]river.QueueConfig),
|
Queues: make(map[string]river.QueueConfig),
|
||||||
JobTimeout: -1,
|
JobTimeout: -1,
|
||||||
Middleware: middleware,
|
Middleware: middleware,
|
||||||
|
Schema: schema,
|
||||||
},
|
},
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
@@ -56,7 +57,6 @@ func (q *Queue) Start(ctx context.Context) (err error) {
|
|||||||
if q == nil || !q.shouldStart {
|
if q == nil || !q.shouldStart {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
ctx = WithQueue(ctx)
|
|
||||||
|
|
||||||
q.client, err = river.NewClient(q.driver, q.config)
|
q.client, err = river.NewClient(q.driver, q.config)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -112,7 +112,6 @@ func WithQueueName(name string) InsertOpt {
|
|||||||
|
|
||||||
func (q *Queue) Insert(ctx context.Context, args river.JobArgs, opts ...InsertOpt) error {
|
func (q *Queue) Insert(ctx context.Context, args river.JobArgs, opts ...InsertOpt) error {
|
||||||
options := new(river.InsertOpts)
|
options := new(river.InsertOpts)
|
||||||
ctx = WithQueue(ctx)
|
|
||||||
for _, opt := range opts {
|
for _, opt := range opts {
|
||||||
opt(options)
|
opt(options)
|
||||||
}
|
}
|
||||||
|
Reference in New Issue
Block a user