argoyle
9368d77bc8
Implements the `latestSchema` query to fetch the latest schema updates for an organization. This change enhances the GraphQL API by allowing clients to retrieve the most recent schema version and its associated subgraphs. The resolver performs necessary access checks, logs relevant information, and generates the Cosmo router configuration from fetched subgraph SDLs, returning structured schema update details.
404 lines
11 KiB
Go
404 lines
11 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
|
|
}
|
|
|
|
// 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
|
|
}
|
|
|
|
// 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
|
|
}
|
|
|
|
// 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)
|
|
_, err := r.apiKeyCanAccessRef(ctx, ref, false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
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)
|
|
|
|
r.Logger.Info("LatestSchema query",
|
|
"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
|
|
}
|
|
|
|
// 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 }
|
|
)
|