app

Documentation

Index

• Variables
• func NewMessageOfficeServer(producer UpdatesProducer, jc *nats.JetstreamClient, ev *env.Env, ...) (messages.MessageDispatchServiceServer, error)
• func RegisterInternalMsgServer(server InternalGrpcServer, cts cluster_token.ClusterTokenServer, ...)
• type ExternalGrpcServer
• type InfraGRPCClient
• type InternalGrpcServer
• type ProcessErrorArgs
• type RealVectorGrpcClient
• type ResourceUpdateArgs
• type UpdatesProducer

Variables

var Module = fx.Module("app",
	repos.NewFxMongoRepo[*domain.MessageOfficeToken]("mo_tokens", "mot", domain.MOTokenIndexes),
	repos.NewFxMongoRepo[*entities.PlatformEdgeCluster]("platform_edge_clusters", "pec", entities.PlatformEdgeClusterIndexes),
	repos.NewFxMongoRepo[*entities.ClusterAllocation]("cluster_allocations", "ca", entities.ClusterAllocationIndexes),

	fx.Provide(func(jsc *nats.JetstreamClient, logger logging.Logger) UpdatesProducer {
		return msg_nats.NewJetstreamProducer(jsc)
	}),

	fx.Invoke(func(lf fx.Lifecycle, producer UpdatesProducer) {
		lf.Append(fx.Hook{
			OnStop: func(ctx context.Context) error {
				return producer.Stop(ctx)
			},
		})
	}),

	domain.Module,

	fx.Provide(
		func(conn InfraGRPCClient) infra.InfraClient {
			return infra.NewInfraClient(conn)
		},
	),

	fx.Provide(func(logger *slog.Logger, jc *nats.JetstreamClient, producer UpdatesProducer, ev *env.Env, d domain.Domain, infraCli infra.InfraClient) (messages.MessageDispatchServiceServer, error) {
		return NewMessageOfficeServer(producer, jc, ev, d, logger.With("component", "message-office"), infraCli)
	}),

	fx.Provide(func(conn RealVectorGrpcClient) proto_rpc.VectorClient {
		return proto_rpc.NewVectorClient(conn)
	}),

	fx.Provide(func(vectorGrpcClient proto_rpc.VectorClient, logger *slog.Logger, d domain.Domain, ev *env.Env) proto_rpc.VectorServer {
		return &vectorProxyServer{
			realVectorClient:   vectorGrpcClient,
			logger:             logger,
			domain:             d,
			tokenHashingSecret: ev.TokenHashingSecret,
			pushEventsCounter:  0,
		}
	}),

	fx.Provide(newInternalMsgServer),
	fx.Invoke(RegisterInternalMsgServer),

	fx.Invoke(
		func(server ExternalGrpcServer, messageServer messages.MessageDispatchServiceServer) {
			messages.RegisterMessageDispatchServiceServer(server, messageServer)
		},
	),

	fx.Invoke(
		func(server ExternalGrpcServer, vectorServer proto_rpc.VectorServer) {
			proto_rpc.RegisterVectorServer(server, vectorServer)
		},
	),

	fx.Invoke(
		func(server httpServer.Server, d domain.Domain) {
			schema := generated.NewExecutableSchema(generated.Config{Resolvers: &graph.Resolver{Domain: d}})
			server.SetupGraphqlServer(schema)
		},
	),
)

Functions

func NewMessageOfficeServer

func NewMessageOfficeServer(producer UpdatesProducer, jc *nats.JetstreamClient, ev *env.Env, d domain.Domain, logger *slog.Logger, infraCli infra.InfraClient) (messages.MessageDispatchServiceServer, error)

func RegisterInternalMsgServer

func RegisterInternalMsgServer(server InternalGrpcServer, cts cluster_token.ClusterTokenServer, pes platform_edge.PlatformEdgeServer)

Types

type ExternalGrpcServer

type ExternalGrpcServer grpc.Server

type InfraGRPCClient

type InfraGRPCClient grpc.Client

type InternalGrpcServer

type InternalGrpcServer grpc.Server

type ProcessErrorArgs

type ProcessErrorArgs struct {
	AccountName string
	ClusterName string
	Error       *messages.ErrorData
	// contains filtered or unexported fields
}

type RealVectorGrpcClient

type RealVectorGrpcClient grpc.Client

type ResourceUpdateArgs

type ResourceUpdateArgs struct {
	AccountName string
	ClusterName string

	Message *messages.ResourceUpdate
	// contains filtered or unexported fields
}

type UpdatesProducer

type UpdatesProducer messaging.Producer