diff --git a/cmd/migrate.go b/cmd/migrate.go index e7f6eb7fa6..522304bd96 100644 --- a/cmd/migrate.go +++ b/cmd/migrate.go @@ -34,7 +34,7 @@ func RunMigrations(logger *slog.Logger, config db.Config) error { return errors.Wrap(err, "failed to connect to db") } metaschemaRepository := postgres.NewMetaSchemaRepository(logger, dbc) - metaschemaService := metaschema.NewService(metaschemaRepository) + metaschemaService := metaschema.NewService(metaschemaRepository, logger, 0) if err = metaschemaService.MigrateDefault(context.Background()); err != nil { return errors.Wrap(err, "failed to add default schemas to db") } diff --git a/cmd/serve.go b/cmd/serve.go index 0ada531eea..346d9ecad2 100644 --- a/cmd/serve.go +++ b/cmd/serve.go @@ -163,12 +163,18 @@ func StartServer(logger *slog.Logger, cfg *config.Frontier) error { return err } - // load metadata schema in memory from db - if schemas, err := deps.MetaSchemaService.List(context.Background()); err != nil { - logger.Warn("metaschemas initialization failed", "err", err) - } else { - logger.Info("metaschemas loaded", "count", len(schemas)) + // prime the metaschema cache and start its periodic refresh. A failed prime + // means validation would be silently off, so stop startup like a failed + // migration rather than come up with an empty cache. + if err := deps.MetaSchemaService.Init(ctx); err != nil { + return fmt.Errorf("metaschemas initialization: %w", err) } + defer func() { + logger.Debug("cleaning up metaschemas") + if err := deps.MetaSchemaService.Close(); err != nil { + logger.Warn("metaschema service cleanup failed", "err", err) + } + }() // apply schema if err = deps.BootstrapService.MigrateSchema(ctx); err != nil { @@ -489,7 +495,7 @@ func buildAPIDependencies( domainService := domain.NewService(logger, domainRepository, userService, organizationService, membershipService) metaschemaRepository := postgres.NewMetaSchemaRepository(logger, dbc) - metaschemaService := metaschema.NewService(metaschemaRepository) + metaschemaService := metaschema.NewService(metaschemaRepository, logger, cfg.App.Metaschema.RefreshInterval) userPATService := userpat.NewService(logger, userPATRepo, cfg.App.PAT, organizationService, roleService, membershipService, projectService, auditRecordRepository) membershipService.SetUserPATService(userPATService) diff --git a/config/sample.config.yaml b/config/sample.config.yaml index 5d0fec5fb8..f6faf53a2b 100644 --- a/config/sample.config.yaml +++ b/config/sample.config.yaml @@ -204,6 +204,14 @@ app: # this is used to validate the webhook payloads encryption_key: "hash-secret-should-be-32-chars--" + # metaschema cache configuration + metaschema: + # how often each server reloads the metaschema cache from the database, so a + # change made on one server reaches the others. 0 disables the background + # refresh; the cache is still primed once at startup. + # e.g. 30s, 1m, 5m + refresh_interval: 1m + db: driver: postgres url: postgres://frontier:@localhost:5432/frontier?sslmode=disable diff --git a/core/metaschema/config.go b/core/metaschema/config.go new file mode 100644 index 0000000000..0522439ade --- /dev/null +++ b/core/metaschema/config.go @@ -0,0 +1,11 @@ +package metaschema + +import "time" + +// Config holds runtime configuration for the metaschema service. +type Config struct { + // RefreshInterval is how often each server reloads the metaschema cache from + // the database, so a change made on one pod reaches the others. A value of 0 + // disables the background refresh; the cache is still primed once at startup. + RefreshInterval time.Duration `yaml:"refresh_interval" mapstructure:"refresh_interval" default:"1m"` +} diff --git a/core/metaschema/errors.go b/core/metaschema/errors.go index 418d867077..7da1eca4c1 100644 --- a/core/metaschema/errors.go +++ b/core/metaschema/errors.go @@ -8,4 +8,5 @@ var ( ErrConflict = errors.New("metaschema already exist") ErrInvalidDetail = errors.New("invalid metadata detail") ErrInvalidMetaSchema = errors.New("metadata schema validation failed") + ErrInvalidSchema = errors.New("metaschema schema must be a JSON object") ) diff --git a/core/metaschema/service.go b/core/metaschema/service.go index 20e4f0aa8f..f650868e06 100644 --- a/core/metaschema/service.go +++ b/core/metaschema/service.go @@ -2,37 +2,76 @@ package metaschema import ( "context" + "encoding/json" + "fmt" + "log/slog" + "sync" + "time" "github.com/raystack/frontier/pkg/utils" "github.com/pkg/errors" "github.com/raystack/frontier/pkg/metadata" + "github.com/robfig/cron/v3" "github.com/xeipuuv/gojsonschema" ) type Service struct { repository Repository + logger *slog.Logger + refreshInterval time.Duration + + mu sync.RWMutex metaSchemaCache map[string]MetaSchema + + syncJob *cron.Cron + syncJobMu sync.Mutex } -func NewService(repository Repository) *Service { +func NewService(repository Repository, logger *slog.Logger, refreshInterval time.Duration) *Service { return &Service{ repository: repository, + logger: logger, + refreshInterval: refreshInterval, metaSchemaCache: make(map[string]MetaSchema), } } -func (s Service) Create(ctx context.Context, toCreate MetaSchema) (MetaSchema, error) { +// validateSchemaIsObject rejects a schema that is not a JSON object. Only an +// object is a usable JSON schema for validating entity metadata; a number, +// string, boolean, array, or malformed JSON would be stored and then error in +// Validate for every entity of that type. Rejecting it on write keeps that state +// from ever being reached, so a change made through the API always round-trips. +func validateSchemaIsObject(schema string) error { + var root any + if err := json.Unmarshal([]byte(schema), &root); err != nil { + return ErrInvalidSchema + } + if _, ok := root.(map[string]any); !ok { + return ErrInvalidSchema + } + return nil +} + +func (s *Service) Create(ctx context.Context, toCreate MetaSchema) (MetaSchema, error) { + if err := validateSchemaIsObject(toCreate.Schema); err != nil { + return MetaSchema{}, err + } mschema, err := s.repository.Create(ctx, toCreate) if err != nil { return MetaSchema{}, err } + s.mu.Lock() s.metaSchemaCache[mschema.Name] = mschema + s.mu.Unlock() return mschema, nil } -func (s Service) Get(ctx context.Context, idOrName string) (MetaSchema, error) { - if schema, ok := s.metaSchemaCache[idOrName]; ok { +func (s *Service) Get(ctx context.Context, idOrName string) (MetaSchema, error) { + s.mu.RLock() + schema, ok := s.metaSchemaCache[idOrName] + s.mu.RUnlock() + if ok { return schema, nil } @@ -41,24 +80,14 @@ func (s Service) Get(ctx context.Context, idOrName string) (MetaSchema, error) { if err != nil { return MetaSchema{}, err } - return schema, nil } return MetaSchema{}, ErrInvalidID } -func (s Service) List(ctx context.Context) ([]MetaSchema, error) { - if len(s.metaSchemaCache) == 0 { - schemas, err := s.repository.List(ctx) - if err != nil { - return nil, err - } - for _, schema := range schemas { - s.metaSchemaCache[schema.Name] = schema - } - return schemas, nil - } - +func (s *Service) List(ctx context.Context) ([]MetaSchema, error) { + s.mu.RLock() + defer s.mu.RUnlock() schemas := make([]MetaSchema, 0, len(s.metaSchemaCache)) for _, schema := range s.metaSchemaCache { schemas = append(schemas, schema) @@ -66,40 +95,48 @@ func (s Service) List(ctx context.Context) ([]MetaSchema, error) { return schemas, nil } -func (s Service) Update(ctx context.Context, id string, toUpdate MetaSchema) (MetaSchema, error) { +func (s *Service) Update(ctx context.Context, id string, toUpdate MetaSchema) (MetaSchema, error) { if utils.IsValidUUID(id) { + if err := validateSchemaIsObject(toUpdate.Schema); err != nil { + return MetaSchema{}, err + } schema, err := s.repository.Update(ctx, id, toUpdate) if err != nil { return MetaSchema{}, err } + s.mu.Lock() s.metaSchemaCache[schema.Name] = schema + s.mu.Unlock() return schema, nil } return MetaSchema{}, ErrInvalidID } -func (s Service) Delete(ctx context.Context, id string) error { +func (s *Service) Delete(ctx context.Context, id string) error { if utils.IsValidUUID(id) { name, err := s.repository.Delete(ctx, id) if err != nil { return err } - + s.mu.Lock() delete(s.metaSchemaCache, name) + s.mu.Unlock() return nil } return ErrInvalidID } -func (s Service) MigrateDefault(ctx context.Context) error { +func (s *Service) MigrateDefault(ctx context.Context) error { return s.repository.MigrateDefaults(ctx) } -// Validate the metadata against the json-schema. In case metaschema doesn't exists in the cache, it will return nil (no validation) -func (s Service) Validate(mdata metadata.Metadata, name string) error { - var mschema MetaSchema - var ok bool - if mschema, ok = s.metaSchemaCache[name]; !ok { +// Validate checks the metadata against the json-schema. When the named +// metaschema is not in the cache it returns nil (no validation). +func (s *Service) Validate(mdata metadata.Metadata, name string) error { + s.mu.RLock() + mschema, ok := s.metaSchemaCache[name] + s.mu.RUnlock() + if !ok { return nil } @@ -115,3 +152,79 @@ func (s Service) Validate(mdata metadata.Metadata, name string) error { } return nil } + +// reload replaces the cache with the current set of metaschemas from the +// database. It holds the write lock across the read and the swap, so a Create, +// Update, or Delete that runs concurrently applies to the cache after the swap +// and is never lost to a stale snapshot. On a database error it returns without +// touching the cache, and it refuses to swap an empty set over a populated +// cache, so neither a blip nor an unexpected empty read blanks validation. The +// caller decides what to do with the error: fail startup for the initial prime, +// log and keep the cache for a scheduled refresh. +func (s *Service) reload(ctx context.Context) error { + s.mu.Lock() + defer s.mu.Unlock() + schemas, err := s.repository.List(ctx) + if err != nil { + return err + } + if len(schemas) == 0 && len(s.metaSchemaCache) > 0 { + return fmt.Errorf("metaschema list returned no schemas, keeping the current cache of %d", len(s.metaSchemaCache)) + } + fresh := make(map[string]MetaSchema, len(schemas)) + for _, schema := range schemas { + fresh[schema.Name] = schema + } + s.metaSchemaCache = fresh + return nil +} + +// Init primes the cache from the database and, when refreshInterval is greater +// than zero, starts a background job that reloads it on that interval so a +// change made on one pod reaches the others. +func (s *Service) Init(ctx context.Context) error { + // The initial prime must succeed. If it fails, startup stops here rather + // than serving with an empty cache, which would skip validation silently. + if err := s.reload(ctx); err != nil { + return fmt.Errorf("prime metaschema cache: %w", err) + } + s.mu.RLock() + count := len(s.metaSchemaCache) + s.mu.RUnlock() + s.logger.InfoContext(ctx, "metaschemas loaded", "count", count) + + if s.refreshInterval <= 0 { + return nil + } + + s.syncJobMu.Lock() + defer s.syncJobMu.Unlock() + if s.syncJob != nil { + <-s.syncJob.Stop().Done() + } + s.syncJob = cron.New(cron.WithChain( + cron.SkipIfStillRunning(cron.DefaultLogger), + cron.Recover(cron.DefaultLogger), + )) + if _, err := s.syncJob.AddFunc(fmt.Sprintf("@every %s", s.refreshInterval.String()), func() { + // A scheduled refresh keeps the last good cache on error, so a database + // blip does not blank validation between successful reloads. + if err := s.reload(ctx); err != nil { + s.logger.WarnContext(ctx, "metaschema cache refresh failed", "err", err) + } + }); err != nil { + return err + } + s.syncJob.Start() + return nil +} + +// Close stops the background refresh job. It is safe to call when none started. +func (s *Service) Close() error { + s.syncJobMu.Lock() + defer s.syncJobMu.Unlock() + if s.syncJob != nil { + <-s.syncJob.Stop().Done() + } + return nil +} diff --git a/core/metaschema/service_test.go b/core/metaschema/service_test.go new file mode 100644 index 0000000000..9df890e837 --- /dev/null +++ b/core/metaschema/service_test.go @@ -0,0 +1,359 @@ +package metaschema + +import ( + "context" + "errors" + "io" + "log/slog" + "sync" + "testing" + "time" + + "github.com/raystack/frontier/pkg/metadata" +) + +// fakeRepo is an in-memory metaschema.Repository for tests. It is safe for +// concurrent use so tests exercise the service's locking, not the repo's. +type fakeRepo struct { + mu sync.Mutex + byID map[string]MetaSchema + listErr error + listCount int + listHook func() // called by List after it snapshots, with the repo mutex released +} + +func newFakeRepo() *fakeRepo { return &fakeRepo{byID: map[string]MetaSchema{}} } + +func (f *fakeRepo) Create(_ context.Context, m MetaSchema) (MetaSchema, error) { + f.mu.Lock() + defer f.mu.Unlock() + if m.ID == "" { + m.ID = m.Name + } + f.byID[m.ID] = m + return m, nil +} + +func (f *fakeRepo) Get(_ context.Context, id string) (MetaSchema, error) { + f.mu.Lock() + defer f.mu.Unlock() + m, ok := f.byID[id] + if !ok { + return MetaSchema{}, ErrInvalidID + } + return m, nil +} + +func (f *fakeRepo) Update(_ context.Context, id string, m MetaSchema) (MetaSchema, error) { + f.mu.Lock() + defer f.mu.Unlock() + m.ID = id + f.byID[id] = m + return m, nil +} + +func (f *fakeRepo) List(_ context.Context) ([]MetaSchema, error) { + f.mu.Lock() + f.listCount++ + if f.listErr != nil { + err := f.listErr + f.mu.Unlock() + return nil, err + } + out := make([]MetaSchema, 0, len(f.byID)) + for _, m := range f.byID { + out = append(out, m) + } + hook := f.listHook + f.mu.Unlock() + if hook != nil { + hook() + } + return out, nil +} + +func (f *fakeRepo) Delete(_ context.Context, id string) (string, error) { + f.mu.Lock() + defer f.mu.Unlock() + name := f.byID[id].Name + delete(f.byID, id) + return name, nil +} + +func (f *fakeRepo) MigrateDefaults(_ context.Context) error { return nil } + +func (f *fakeRepo) setListHook(fn func()) { + f.mu.Lock() + f.listHook = fn + f.mu.Unlock() +} + +func (f *fakeRepo) listCalls() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.listCount +} + +func (f *fakeRepo) clear() { + f.mu.Lock() + f.byID = map[string]MetaSchema{} + f.mu.Unlock() +} + +// waitForRow blocks until a row with the given name is committed, so a test can +// order a concurrent write ahead of the next step without a fixed sleep. +func (f *fakeRepo) waitForRow(t *testing.T, name string) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + f.mu.Lock() + found := false + for _, m := range f.byID { + if m.Name == name { + found = true + break + } + } + f.mu.Unlock() + if found { + return + } + time.Sleep(time.Millisecond) + } + t.Fatalf("row %q never committed", name) +} + +func discardLogger() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) } + +func waitFor(t *testing.T, timeout time.Duration, cond func() bool) { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(time.Millisecond) + } + t.Fatal("condition not met within timeout") +} + +func TestService_ConcurrentAccess(t *testing.T) { + svc := NewService(newFakeRepo(), discardLogger(), 0) + + const goroutines = 50 + var wg sync.WaitGroup + for i := 0; i < goroutines; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + name := "s" + string(rune('0'+(i%5))) + _, _ = svc.Create(context.Background(), MetaSchema{Name: name, Schema: `{"type":"object"}`}) + _ = svc.Validate(metadata.Metadata{"a": "b"}, name) + _, _ = svc.List(context.Background()) + _, _ = svc.Get(context.Background(), name) + }(i) + } + wg.Wait() +} + +func TestService_reload_picksUpNewSchema(t *testing.T) { + repo := newFakeRepo() + svc := NewService(repo, discardLogger(), 0) + if err := svc.Init(context.Background()); err != nil { + t.Fatalf("Init: %v", err) + } + + // Add a schema straight to the repo, as if another pod created it. + if _, err := repo.Create(context.Background(), MetaSchema{ID: "1", Name: "user", Schema: `{"type":"object"}`}); err != nil { + t.Fatalf("repo.Create: %v", err) + } + + // Not visible yet: Init primed the cache before the row existed. + if _, err := svc.Get(context.Background(), "user"); err == nil { + t.Fatal("expected schema to be absent before reload") + } + + if err := svc.reload(context.Background()); err != nil { + t.Fatalf("reload: %v", err) + } + + got, err := svc.Get(context.Background(), "user") + if err != nil { + t.Fatalf("Get after reload: %v", err) + } + if got.Name != "user" { + t.Fatalf("got %q, want user", got.Name) + } +} + +func TestService_reload_keepsCacheOnListError(t *testing.T) { + repo := newFakeRepo() + _, _ = repo.Create(context.Background(), MetaSchema{ID: "1", Name: "user", Schema: `{"type":"object"}`}) + svc := NewService(repo, discardLogger(), 0) + if err := svc.Init(context.Background()); err != nil { + t.Fatalf("Init: %v", err) + } + + repo.listErr = context.DeadlineExceeded + if err := svc.reload(context.Background()); err == nil { + t.Fatal("reload should return the list error, not swallow it") + } + + // The previous cache survives a failed reload. + if _, err := svc.Get(context.Background(), "user"); err != nil { + t.Fatalf("cache should survive a reload error: %v", err) + } +} + +// TestService_reload_keepsCacheOnEmptyList guards against an error-free empty +// read (a replica blip or a repo bug) blanking a populated cache. +func TestService_reload_keepsCacheOnEmptyList(t *testing.T) { + repo := newFakeRepo() + _, _ = repo.Create(context.Background(), MetaSchema{ID: "org", Name: "organization", Schema: `{"type":"object"}`}) + svc := NewService(repo, discardLogger(), 0) + if err := svc.Init(context.Background()); err != nil { + t.Fatalf("Init: %v", err) + } + + repo.clear() // List now returns an empty slice with no error + + if err := svc.reload(context.Background()); err == nil { + t.Fatal("reload should report an error rather than blank a populated cache") + } + if _, err := svc.Get(context.Background(), "organization"); err != nil { + t.Fatalf("cache should survive an empty list: %v", err) + } +} + +// TestService_reload_doesNotDropConcurrentWrite reproduces the read-modify-write +// clobber: reload snapshots the DB, a Create commits, then reload swaps. With a +// blind swap the created schema is lost; holding the write lock across the read +// and the swap keeps it. +func TestService_reload_doesNotDropConcurrentWrite(t *testing.T) { + repo := newFakeRepo() + // A base schema keeps the cache populated so the empty-list guard is not in play. + _, _ = repo.Create(context.Background(), MetaSchema{ID: "org", Name: "organization", Schema: `{"type":"object"}`}) + svc := NewService(repo, discardLogger(), 0) + if err := svc.Init(context.Background()); err != nil { + t.Fatalf("Init: %v", err) + } + + snapshotted := make(chan struct{}) + release := make(chan struct{}) + var once sync.Once + repo.setListHook(func() { + once.Do(func() { + close(snapshotted) + <-release + }) + }) + + reloadDone := make(chan struct{}) + go func() { + _ = svc.reload(context.Background()) + close(reloadDone) + }() + + <-snapshotted // reload has taken its DB snapshot, which does not include "user" + + createDone := make(chan struct{}) + go func() { + _, _ = svc.Create(context.Background(), MetaSchema{ID: "user", Name: "user", Schema: `{"type":"object"}`}) + close(createDone) + }() + + // Order the Create's row commit ahead of reload's swap: this is the window + // the blind swap would clobber. + repo.waitForRow(t, "user") + close(release) + + <-reloadDone + <-createDone + + if _, err := svc.Get(context.Background(), "user"); err != nil { + t.Fatalf("concurrent Create was clobbered by reload: %v", err) + } +} + +func TestService_Init_refreshDisabled(t *testing.T) { + svc := NewService(newFakeRepo(), discardLogger(), 0) + if err := svc.Init(context.Background()); err != nil { + t.Fatalf("Init: %v", err) + } + if svc.syncJob != nil { + t.Fatal("interval 0 must not start a cron job") + } + if err := svc.Close(); err != nil { + t.Fatalf("Close when no job started: %v", err) + } +} + +// TestService_Init_returnsPrimeError checks the boot-failure signal: a DB that +// is unreachable at startup must fail Init, not start with an empty cache. +func TestService_Init_returnsPrimeError(t *testing.T) { + repo := newFakeRepo() + repo.listErr = context.DeadlineExceeded + svc := NewService(repo, discardLogger(), 0) + if err := svc.Init(context.Background()); err == nil { + t.Fatal("Init should return the initial prime error so startup can fail") + } +} + +// TestService_Init_runsScheduledRefresh exercises the cron lifecycle the PR +// adds: a positive interval starts the job, it refreshes on schedule, and Close +// stops it. +func TestService_Init_runsScheduledRefresh(t *testing.T) { + repo := newFakeRepo() + _, _ = repo.Create(context.Background(), MetaSchema{ID: "org", Name: "organization", Schema: `{"type":"object"}`}) + svc := NewService(repo, discardLogger(), 40*time.Millisecond) + if err := svc.Init(context.Background()); err != nil { + t.Fatalf("Init: %v", err) + } + t.Cleanup(func() { _ = svc.Close() }) + + if svc.syncJob == nil { + t.Fatal("a positive interval must start a cron job") + } + + // Prime is one List; wait for at least one scheduled refresh on top of it. + waitFor(t, 2*time.Second, func() bool { return repo.listCalls() >= 2 }) + + if err := svc.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + after := repo.listCalls() + time.Sleep(120 * time.Millisecond) + if got := repo.listCalls(); got != after { + t.Fatalf("cron kept running after Close: was %d, now %d", after, got) + } +} + +// TestService_Create_rejectsNonObjectSchema checks the write path refuses a +// schema that is not a JSON object, so a non-object metaschema is never a +// reachable state. +func TestService_Create_rejectsNonObjectSchema(t *testing.T) { + svc := NewService(newFakeRepo(), discardLogger(), 0) + for _, bad := range []string{"123", `"x"`, "[]", "true", "{not json", ""} { + if _, err := svc.Create(context.Background(), MetaSchema{Name: "user", Schema: bad}); !errors.Is(err, ErrInvalidSchema) { + t.Fatalf("Create(schema=%q) = %v, want ErrInvalidSchema", bad, err) + } + } +} + +func TestService_Create_acceptsObjectSchema(t *testing.T) { + svc := NewService(newFakeRepo(), discardLogger(), 0) + if _, err := svc.Create(context.Background(), MetaSchema{Name: "user", Schema: `{"type":"object"}`}); err != nil { + t.Fatalf("Create with an object schema: %v", err) + } +} + +// TestService_Update_rejectsNonObjectSchema checks the same guard on the update path. +func TestService_Update_rejectsNonObjectSchema(t *testing.T) { + const id = "11111111-1111-1111-1111-111111111111" + repo := newFakeRepo() + _, _ = repo.Create(context.Background(), MetaSchema{ID: id, Name: "user", Schema: `{"type":"object"}`}) + svc := NewService(repo, discardLogger(), 0) + if _, err := svc.Update(context.Background(), id, MetaSchema{Name: "user", Schema: "123"}); !errors.Is(err, ErrInvalidSchema) { + t.Fatalf("Update with a non-object schema = %v, want ErrInvalidSchema", err) + } +} diff --git a/docs/content/docs/reference/configurations.mdx b/docs/content/docs/reference/configurations.mdx index 0ab171a974..d82753a323 100644 --- a/docs/content/docs/reference/configurations.mdx +++ b/docs/content/docs/reference/configurations.mdx @@ -140,6 +140,13 @@ app: # encryption key used to encrypt the secrets stored in database not to encrypt # the webhook payload encryption_key: "encryption-key-should-be-32-chars--" + # metaschema cache configuration + metaschema: + # how often each server reloads the metaschema cache from the database, so a + # change made on one server reaches the others. 0 disables the background + # refresh; the cache is still primed once at startup. + # e.g. 30s, 1m, 5m + refresh_interval: 1m db: driver: postgres url: postgres://frontier:@localhost:5432/frontier?sslmode=disable @@ -243,6 +250,7 @@ This page contains reference for all the application configurations for Frontier | **app.disable_orgs_listing** | If set to true, disallows non-admin APIs to list all organizations. | | No | | **app.disable_users_listing** | If set to true, disallows non-admin APIs to list all users. | | No | | **app.cors_origin** | Origin value from where CORS is allowed. | | Yes(for Admin UI) | +| **app.metaschema.refresh_interval** | How often each server reloads the metaschema cache from the database so a change made on one server reaches the others. 0 disables the background refresh. | 1m | No (default: 1m) | ### Authentication Configurations diff --git a/internal/api/v1beta1connect/metaschema.go b/internal/api/v1beta1connect/metaschema.go index 64f883ef08..ece9958179 100644 --- a/internal/api/v1beta1connect/metaschema.go +++ b/internal/api/v1beta1connect/metaschema.go @@ -47,6 +47,8 @@ func (h *ConnectHandler) CreateMetaSchema(ctx context.Context, req *connect.Requ }) if err != nil { switch { + case errors.Is(err, metaschema.ErrInvalidSchema): + return nil, connect.NewError(connect.CodeInvalidArgument, metaschema.ErrInvalidSchema) case errors.Is(err, metaschema.ErrNotExist), errors.Is(err, metaschema.ErrInvalidID), errors.Is(err, metaschema.ErrInvalidDetail): @@ -101,6 +103,8 @@ func (h *ConnectHandler) UpdateMetaSchema(ctx context.Context, req *connect.Requ }) if err != nil { switch { + case errors.Is(err, metaschema.ErrInvalidSchema): + return nil, connect.NewError(connect.CodeInvalidArgument, metaschema.ErrInvalidSchema) case errors.Is(err, metaschema.ErrInvalidDetail): return nil, connect.NewError(connect.CodeInvalidArgument, ErrBadRequest) case errors.Is(err, metaschema.ErrInvalidID), diff --git a/internal/api/v1beta1connect/metaschema_test.go b/internal/api/v1beta1connect/metaschema_test.go index f619b4ab05..844aba2558 100644 --- a/internal/api/v1beta1connect/metaschema_test.go +++ b/internal/api/v1beta1connect/metaschema_test.go @@ -168,6 +168,20 @@ func TestConnectHandler_CreateMetaSchema(t *testing.T) { want: nil, wantErr: connect.NewError(connect.CodeInvalidArgument, ErrBadRequest), }, + { + name: "should return invalid argument when schema is not an object", + setup: func(m *mocks.MetaSchemaService) { + m.EXPECT().Create(mock.Anything, mock.Anything).Return(metaschema.MetaSchema{}, metaschema.ErrInvalidSchema) + }, + req: connect.NewRequest(&frontierv1beta1.CreateMetaSchemaRequest{ + Body: &frontierv1beta1.MetaSchemaRequestBody{ + Name: "user", + Schema: "123", + }, + }), + want: nil, + wantErr: connect.NewError(connect.CodeInvalidArgument, metaschema.ErrInvalidSchema), + }, } for _, tt := range tests { @@ -347,6 +361,21 @@ func TestConnectHandler_UpdateMetaSchema(t *testing.T) { want: nil, wantErr: connect.NewError(connect.CodeNotFound, ErrMetaschemaNotFound), }, + { + name: "should return invalid argument when schema is not an object", + setup: func(m *mocks.MetaSchemaService) { + m.EXPECT().Update(mock.Anything, "test_id", mock.Anything).Return(metaschema.MetaSchema{}, metaschema.ErrInvalidSchema) + }, + req: connect.NewRequest(&frontierv1beta1.UpdateMetaSchemaRequest{ + Id: "test_id", + Body: &frontierv1beta1.MetaSchemaRequestBody{ + Name: "user", + Schema: "123", + }, + }), + want: nil, + wantErr: connect.NewError(connect.CodeInvalidArgument, metaschema.ErrInvalidSchema), + }, } for _, tt := range tests { diff --git a/pkg/server/config.go b/pkg/server/config.go index 6f20ba39f0..984309b7c1 100644 --- a/pkg/server/config.go +++ b/pkg/server/config.go @@ -3,6 +3,7 @@ package server import ( "time" + "github.com/raystack/frontier/core/metaschema" "github.com/raystack/frontier/core/userpat" "github.com/raystack/frontier/core/webhook" @@ -83,6 +84,8 @@ type Config struct { Webhook webhook.Config `yaml:"webhook" mapstructure:"webhook"` PAT userpat.Config `yaml:"pat" mapstructure:"pat"` + Metaschema metaschema.Config `yaml:"metaschema" mapstructure:"metaschema"` + // AdditionalTraitsPath is a file path to a YAML file containing additional preference traits // These traits are merged with DefaultTraits at startup AdditionalTraitsPath string `yaml:"additional_traits_path" mapstructure:"additional_traits_path"`