argoyle
ffcf41b85a
Introduce `AddUserToOrganization`, `RemoveAPIKey`, and `RemoveOrganization` commands to enhance organization management. Implement validation for user addition and API key removal. Update GraphQL schema to support new mutations and add caching for the new events, ensuring that organizations and their relationships are accurately represented in the cache.
511 lines
14 KiB
Go
511 lines
14 KiB
Go
package graph
|
|
|
|
// This file will be automatically regenerated based on the schema, any resolver implementations
|
|
// will be copied through when generating and any unknown code will be moved to the end.
|
|
// Code generated by github.com/99designs/gqlgen
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
|
|
"gitlab.com/unboundsoftware/eventsourced/eventsourced"
|
|
|
|
"gitlab.com/unboundsoftware/schemas/domain"
|
|
"gitlab.com/unboundsoftware/schemas/graph/generated"
|
|
"gitlab.com/unboundsoftware/schemas/graph/model"
|
|
"gitlab.com/unboundsoftware/schemas/middleware"
|
|
"gitlab.com/unboundsoftware/schemas/rand"
|
|
"gitlab.com/unboundsoftware/schemas/sdlmerge"
|
|
)
|
|
|
|
// AddOrganization is the resolver for the addOrganization field.
|
|
func (r *mutationResolver) AddOrganization(ctx context.Context, name string) (*model.Organization, error) {
|
|
sub := middleware.UserFromContext(ctx)
|
|
org := &domain.Organization{}
|
|
h, err := r.handler(ctx, org)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
_, err = h.Handle(ctx, &domain.AddOrganization{
|
|
Name: name,
|
|
Initiator: sub,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return ToGqlOrganization(*org), nil
|
|
}
|
|
|
|
// AddUserToOrganization is the resolver for the addUserToOrganization field.
|
|
func (r *mutationResolver) AddUserToOrganization(ctx context.Context, organizationID string, userID string) (*model.Organization, error) {
|
|
sub := middleware.UserFromContext(ctx)
|
|
org := &domain.Organization{BaseAggregate: eventsourced.BaseAggregateFromString(organizationID)}
|
|
h, err := r.handler(ctx, org)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
_, err = h.Handle(ctx, &domain.AddUserToOrganization{
|
|
UserId: userID,
|
|
Initiator: sub,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return ToGqlOrganization(*org), nil
|
|
}
|
|
|
|
// AddAPIKey is the resolver for the addAPIKey field.
|
|
func (r *mutationResolver) AddAPIKey(ctx context.Context, input *model.InputAPIKey) (*model.APIKey, error) {
|
|
sub := middleware.UserFromContext(ctx)
|
|
org := &domain.Organization{BaseAggregate: eventsourced.BaseAggregateFromString(input.OrganizationID)}
|
|
h, err := r.handler(ctx, org)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
key := fmt.Sprintf("us_ak_%s", rand.String(16))
|
|
_, err = h.Handle(ctx, &domain.AddAPIKey{
|
|
Name: input.Name,
|
|
Key: key,
|
|
Refs: input.Refs,
|
|
Read: input.Read,
|
|
Publish: input.Publish,
|
|
Initiator: sub,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &model.APIKey{
|
|
ID: apiKeyId(input.OrganizationID, input.Name),
|
|
Name: input.Name,
|
|
Key: &key,
|
|
Organization: &model.Organization{
|
|
ID: input.OrganizationID,
|
|
Name: org.Name,
|
|
},
|
|
Refs: input.Refs,
|
|
Read: input.Read,
|
|
Publish: input.Publish,
|
|
}, nil
|
|
}
|
|
|
|
// RemoveAPIKey is the resolver for the removeAPIKey field.
|
|
func (r *mutationResolver) RemoveAPIKey(ctx context.Context, organizationID string, keyName string) (*model.Organization, error) {
|
|
sub := middleware.UserFromContext(ctx)
|
|
org := &domain.Organization{BaseAggregate: eventsourced.BaseAggregateFromString(organizationID)}
|
|
h, err := r.handler(ctx, org)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
_, err = h.Handle(ctx, &domain.RemoveAPIKey{
|
|
KeyName: keyName,
|
|
Initiator: sub,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return ToGqlOrganization(*org), nil
|
|
}
|
|
|
|
// RemoveOrganization is the resolver for the removeOrganization field.
|
|
func (r *mutationResolver) RemoveOrganization(ctx context.Context, organizationID string) (bool, error) {
|
|
sub := middleware.UserFromContext(ctx)
|
|
org := &domain.Organization{BaseAggregate: eventsourced.BaseAggregateFromString(organizationID)}
|
|
h, err := r.handler(ctx, org)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
_, err = h.Handle(ctx, &domain.RemoveOrganization{
|
|
Initiator: sub,
|
|
})
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
return true, nil
|
|
}
|
|
|
|
// UpdateSubGraph is the resolver for the updateSubGraph field.
|
|
func (r *mutationResolver) UpdateSubGraph(ctx context.Context, input model.InputSubGraph) (*model.SubGraph, error) {
|
|
orgId := middleware.OrganizationFromContext(ctx)
|
|
name, err := r.apiKeyCanAccessRef(ctx, input.Ref, true)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
subGraphId := r.Cache.SubGraphId(orgId, input.Ref, input.Service)
|
|
subGraph := &domain.SubGraph{}
|
|
if subGraphId != "" {
|
|
subGraph.BaseAggregate = eventsourced.BaseAggregateFromString(subGraphId)
|
|
}
|
|
handler, err := r.handler(ctx, subGraph)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if strings.TrimSpace(input.Sdl) == strings.TrimSpace(subGraph.Sdl) &&
|
|
r.stringEqual(input.URL, subGraph.Url) &&
|
|
r.stringEqual(input.WsURL, subGraph.WSUrl) {
|
|
return r.toGqlSubGraph(subGraph), nil
|
|
}
|
|
serviceSDLs := []string{input.Sdl}
|
|
services, _ := r.Cache.Services(orgId, input.Ref, "")
|
|
for _, id := range services {
|
|
sg, err := r.fetchSubGraph(ctx, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if sg.Service != input.Service {
|
|
serviceSDLs = append(serviceSDLs, sg.Sdl)
|
|
}
|
|
}
|
|
_, err = sdlmerge.MergeSDLs(serviceSDLs...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
_, err = handler.Handle(ctx, domain.UpdateSubGraph{
|
|
OrganizationId: orgId,
|
|
Ref: input.Ref,
|
|
Service: input.Service,
|
|
Url: input.URL,
|
|
WSUrl: input.WsURL,
|
|
Sdl: input.Sdl,
|
|
Initiator: name,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Publish schema update to subscribers
|
|
go func() {
|
|
services, lastUpdate := r.Cache.Services(orgId, input.Ref, "")
|
|
r.Logger.Info("Publishing schema update after subgraph change",
|
|
"ref", input.Ref,
|
|
"orgId", orgId,
|
|
"lastUpdate", lastUpdate,
|
|
"servicesCount", len(services),
|
|
)
|
|
|
|
subGraphs := make([]*model.SubGraph, len(services))
|
|
for i, id := range services {
|
|
sg, err := r.fetchSubGraph(context.Background(), id)
|
|
if err != nil {
|
|
r.Logger.Error("fetch subgraph for update notification", "error", err)
|
|
continue
|
|
}
|
|
subGraphs[i] = &model.SubGraph{
|
|
ID: sg.ID.String(),
|
|
Service: sg.Service,
|
|
URL: sg.Url,
|
|
WsURL: sg.WSUrl,
|
|
Sdl: sg.Sdl,
|
|
ChangedBy: sg.ChangedBy,
|
|
ChangedAt: sg.ChangedAt,
|
|
}
|
|
}
|
|
|
|
// Generate Cosmo router config
|
|
cosmoConfig, err := GenerateCosmoRouterConfig(subGraphs)
|
|
if err != nil {
|
|
r.Logger.Error("generate cosmo config for update", "error", err)
|
|
cosmoConfig = "" // Send empty if generation fails
|
|
}
|
|
|
|
// Publish to all subscribers of this ref
|
|
update := &model.SchemaUpdate{
|
|
Ref: input.Ref,
|
|
ID: lastUpdate,
|
|
SubGraphs: subGraphs,
|
|
CosmoRouterConfig: &cosmoConfig,
|
|
}
|
|
|
|
r.Logger.Info("Publishing schema update to subscribers",
|
|
"ref", update.Ref,
|
|
"id", update.ID,
|
|
"subGraphsCount", len(update.SubGraphs),
|
|
"cosmoConfigLength", len(cosmoConfig),
|
|
)
|
|
|
|
r.PubSub.Publish(input.Ref, update)
|
|
}()
|
|
|
|
return r.toGqlSubGraph(subGraph), nil
|
|
}
|
|
|
|
// Organizations is the resolver for the organizations field.
|
|
func (r *queryResolver) Organizations(ctx context.Context) ([]*model.Organization, error) {
|
|
sub := middleware.UserFromContext(ctx)
|
|
orgs := r.Cache.OrganizationsByUser(sub)
|
|
return ToGqlOrganizations(orgs), nil
|
|
}
|
|
|
|
// AllOrganizations is the resolver for the allOrganizations field.
|
|
func (r *queryResolver) AllOrganizations(ctx context.Context) ([]*model.Organization, error) {
|
|
// Check if user has admin role
|
|
if !middleware.UserHasRole(ctx, "admin") {
|
|
return nil, fmt.Errorf("unauthorized: admin role required")
|
|
}
|
|
|
|
orgs := r.Cache.AllOrganizations()
|
|
return ToGqlOrganizations(orgs), nil
|
|
}
|
|
|
|
// Supergraph is the resolver for the supergraph field.
|
|
func (r *queryResolver) Supergraph(ctx context.Context, ref string, isAfter *string) (model.Supergraph, error) {
|
|
orgId := middleware.OrganizationFromContext(ctx)
|
|
userId := middleware.UserFromContext(ctx)
|
|
|
|
r.Logger.Info("Supergraph query",
|
|
"ref", ref,
|
|
"orgId", orgId,
|
|
"userId", userId,
|
|
)
|
|
|
|
// If authenticated with API key (organization), check access
|
|
if orgId != "" {
|
|
_, err := r.apiKeyCanAccessRef(ctx, ref, false)
|
|
if err != nil {
|
|
r.Logger.Error("API key cannot access ref", "error", err, "ref", ref)
|
|
return nil, err
|
|
}
|
|
} else if userId != "" {
|
|
// For user authentication, check if user has access to ref through their organizations
|
|
userOrgs := r.Cache.OrganizationsByUser(userId)
|
|
if len(userOrgs) == 0 {
|
|
r.Logger.Error("User has no organizations", "userId", userId)
|
|
return nil, fmt.Errorf("user has no access to any organizations")
|
|
}
|
|
// Use the first organization's ID for querying
|
|
orgId = userOrgs[0].ID.String()
|
|
r.Logger.Info("Using organization from user context", "orgId", orgId)
|
|
} else {
|
|
return nil, fmt.Errorf("no authentication provided")
|
|
}
|
|
|
|
after := ""
|
|
if isAfter != nil {
|
|
after = *isAfter
|
|
}
|
|
services, lastUpdate := r.Cache.Services(orgId, ref, after)
|
|
if after == lastUpdate {
|
|
return &model.Unchanged{
|
|
ID: lastUpdate,
|
|
MinDelaySeconds: 10,
|
|
}, nil
|
|
}
|
|
subGraphs := make([]*model.SubGraph, len(services))
|
|
for i, id := range services {
|
|
sg, err := r.fetchSubGraph(ctx, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
subGraphs[i] = &model.SubGraph{
|
|
ID: sg.ID.String(),
|
|
Service: sg.Service,
|
|
URL: sg.Url,
|
|
WsURL: sg.WSUrl,
|
|
Sdl: sg.Sdl,
|
|
ChangedBy: sg.ChangedBy,
|
|
ChangedAt: sg.ChangedAt,
|
|
}
|
|
}
|
|
|
|
var serviceSDLs []string
|
|
for _, id := range services {
|
|
sg, err := r.fetchSubGraph(ctx, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
serviceSDLs = append(serviceSDLs, sg.Sdl)
|
|
}
|
|
sdl, err := sdlmerge.MergeSDLs(serviceSDLs...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &model.SubGraphs{
|
|
ID: lastUpdate,
|
|
SubGraphs: subGraphs,
|
|
Sdl: sdl,
|
|
MinDelaySeconds: 10,
|
|
}, nil
|
|
}
|
|
|
|
// LatestSchema is the resolver for the latestSchema field.
|
|
func (r *queryResolver) LatestSchema(ctx context.Context, ref string) (*model.SchemaUpdate, error) {
|
|
orgId := middleware.OrganizationFromContext(ctx)
|
|
userId := middleware.UserFromContext(ctx)
|
|
|
|
r.Logger.Info("LatestSchema query",
|
|
"ref", ref,
|
|
"orgId", orgId,
|
|
"userId", userId,
|
|
)
|
|
|
|
// If authenticated with API key (organization), check access
|
|
if orgId != "" {
|
|
_, err := r.apiKeyCanAccessRef(ctx, ref, false)
|
|
if err != nil {
|
|
r.Logger.Error("API key cannot access ref", "error", err, "ref", ref)
|
|
return nil, err
|
|
}
|
|
} else if userId != "" {
|
|
// For user authentication, check if user has access to ref through their organizations
|
|
userOrgs := r.Cache.OrganizationsByUser(userId)
|
|
if len(userOrgs) == 0 {
|
|
r.Logger.Error("User has no organizations", "userId", userId)
|
|
return nil, fmt.Errorf("user has no access to any organizations")
|
|
}
|
|
// Use the first organization's ID for querying
|
|
// In a real-world scenario, you might want to check which org has access to this ref
|
|
orgId = userOrgs[0].ID.String()
|
|
r.Logger.Info("Using organization from user context", "orgId", orgId)
|
|
} else {
|
|
return nil, fmt.Errorf("no authentication provided")
|
|
}
|
|
|
|
// Get current services and schema
|
|
services, lastUpdate := r.Cache.Services(orgId, ref, "")
|
|
r.Logger.Info("Fetching latest schema",
|
|
"ref", ref,
|
|
"orgId", orgId,
|
|
"lastUpdate", lastUpdate,
|
|
"servicesCount", len(services),
|
|
)
|
|
|
|
subGraphs := make([]*model.SubGraph, len(services))
|
|
for i, id := range services {
|
|
sg, err := r.fetchSubGraph(ctx, id)
|
|
if err != nil {
|
|
r.Logger.Error("fetch subgraph", "error", err, "id", id)
|
|
return nil, err
|
|
}
|
|
subGraphs[i] = &model.SubGraph{
|
|
ID: sg.ID.String(),
|
|
Service: sg.Service,
|
|
URL: sg.Url,
|
|
WsURL: sg.WSUrl,
|
|
Sdl: sg.Sdl,
|
|
ChangedBy: sg.ChangedBy,
|
|
ChangedAt: sg.ChangedAt,
|
|
}
|
|
}
|
|
|
|
// Generate Cosmo router config
|
|
cosmoConfig, err := GenerateCosmoRouterConfig(subGraphs)
|
|
if err != nil {
|
|
r.Logger.Error("generate cosmo config", "error", err)
|
|
cosmoConfig = "" // Return empty if generation fails
|
|
}
|
|
|
|
update := &model.SchemaUpdate{
|
|
Ref: ref,
|
|
ID: lastUpdate,
|
|
SubGraphs: subGraphs,
|
|
CosmoRouterConfig: &cosmoConfig,
|
|
}
|
|
|
|
r.Logger.Info("Latest schema fetched",
|
|
"ref", update.Ref,
|
|
"id", update.ID,
|
|
"subGraphsCount", len(update.SubGraphs),
|
|
"cosmoConfigLength", len(cosmoConfig),
|
|
)
|
|
|
|
return update, nil
|
|
}
|
|
|
|
// SchemaUpdates is the resolver for the schemaUpdates field.
|
|
func (r *subscriptionResolver) SchemaUpdates(ctx context.Context, ref string) (<-chan *model.SchemaUpdate, error) {
|
|
orgId := middleware.OrganizationFromContext(ctx)
|
|
|
|
r.Logger.Info("SchemaUpdates subscription started",
|
|
"ref", ref,
|
|
"orgId", orgId,
|
|
)
|
|
|
|
_, err := r.apiKeyCanAccessRef(ctx, ref, false)
|
|
if err != nil {
|
|
r.Logger.Error("API key cannot access ref", "error", err, "ref", ref)
|
|
return nil, err
|
|
}
|
|
|
|
// Subscribe to updates for this ref
|
|
ch := r.PubSub.Subscribe(ref)
|
|
|
|
// Send initial state immediately
|
|
go func() {
|
|
// Use background context for async operation
|
|
bgCtx := context.Background()
|
|
|
|
services, lastUpdate := r.Cache.Services(orgId, ref, "")
|
|
r.Logger.Info("Preparing initial schema update",
|
|
"ref", ref,
|
|
"orgId", orgId,
|
|
"lastUpdate", lastUpdate,
|
|
"servicesCount", len(services),
|
|
)
|
|
|
|
subGraphs := make([]*model.SubGraph, len(services))
|
|
for i, id := range services {
|
|
sg, err := r.fetchSubGraph(bgCtx, id)
|
|
if err != nil {
|
|
r.Logger.Error("fetch subgraph for initial update", "error", err, "id", id)
|
|
continue
|
|
}
|
|
subGraphs[i] = &model.SubGraph{
|
|
ID: sg.ID.String(),
|
|
Service: sg.Service,
|
|
URL: sg.Url,
|
|
WsURL: sg.WSUrl,
|
|
Sdl: sg.Sdl,
|
|
ChangedBy: sg.ChangedBy,
|
|
ChangedAt: sg.ChangedAt,
|
|
}
|
|
}
|
|
|
|
// Generate Cosmo router config
|
|
cosmoConfig, err := GenerateCosmoRouterConfig(subGraphs)
|
|
if err != nil {
|
|
r.Logger.Error("generate cosmo config", "error", err)
|
|
cosmoConfig = "" // Send empty if generation fails
|
|
}
|
|
|
|
// Send initial update
|
|
update := &model.SchemaUpdate{
|
|
Ref: ref,
|
|
ID: lastUpdate,
|
|
SubGraphs: subGraphs,
|
|
CosmoRouterConfig: &cosmoConfig,
|
|
}
|
|
|
|
r.Logger.Info("Sending initial schema update",
|
|
"ref", update.Ref,
|
|
"id", update.ID,
|
|
"subGraphsCount", len(update.SubGraphs),
|
|
"cosmoConfigLength", len(cosmoConfig),
|
|
)
|
|
|
|
ch <- update
|
|
}()
|
|
|
|
// Clean up subscription when context is done
|
|
go func() {
|
|
<-ctx.Done()
|
|
r.PubSub.Unsubscribe(ref, ch)
|
|
}()
|
|
|
|
return ch, nil
|
|
}
|
|
|
|
// Mutation returns generated.MutationResolver implementation.
|
|
func (r *Resolver) Mutation() generated.MutationResolver { return &mutationResolver{r} }
|
|
|
|
// Query returns generated.QueryResolver implementation.
|
|
func (r *Resolver) Query() generated.QueryResolver { return &queryResolver{r} }
|
|
|
|
// Subscription returns generated.SubscriptionResolver implementation.
|
|
func (r *Resolver) Subscription() generated.SubscriptionResolver { return &subscriptionResolver{r} }
|
|
|
|
type (
|
|
mutationResolver struct{ *Resolver }
|
|
queryResolver struct{ *Resolver }
|
|
subscriptionResolver struct{ *Resolver }
|
|
)
|