feat: connect to spicedb

12 files changed, 433 insertions, 0 deletions

diff --git a/vendor/github.com/authzed/spicedb/internal/middleware/chain.go b/vendor/github.com/authzed/spicedb/internal/middleware/chain.go
new file mode 100644
index 0000000..de08ffc
--- /dev/null
+++ b/vendor/github.com/authzed/spicedb/internal/middleware/chain.go
@@ -0,0 +1,58 @@
+package middleware
+
+import (
+	"context"
+
+	"google.golang.org/grpc"
+)
+
+// Vendored from grpc-go-middleware
+// These were removed in v2, see: https://github.com/grpc-ecosystem/go-grpc-middleware/pull/385
+
+// ChainUnaryServer creates a single interceptor out of a chain of many interceptors.
+//
+// Execution is done in left-to-right order, including passing of context.
+// For example ChainUnaryServer(one, two, three) will execute one before two before three, and three
+// will see context changes of one and two.
+func ChainUnaryServer(interceptors ...grpc.UnaryServerInterceptor) grpc.UnaryServerInterceptor {
+	n := len(interceptors)
+
+	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+		chainer := func(currentInter grpc.UnaryServerInterceptor, currentHandler grpc.UnaryHandler) grpc.UnaryHandler {
+			return func(currentCtx context.Context, currentReq interface{}) (interface{}, error) {
+				return currentInter(currentCtx, currentReq, info, currentHandler)
+			}
+		}
+
+		chainedHandler := handler
+		for i := n - 1; i >= 0; i-- {
+			chainedHandler = chainer(interceptors[i], chainedHandler)
+		}
+
+		return chainedHandler(ctx, req)
+	}
+}
+
+// ChainStreamServer creates a single interceptor out of a chain of many interceptors.
+//
+// Execution is done in left-to-right order, including passing of context.
+// For example ChainUnaryServer(one, two, three) will execute one before two before three.
+// If you want to pass context between interceptors, use WrapServerStream.
+func ChainStreamServer(interceptors ...grpc.StreamServerInterceptor) grpc.StreamServerInterceptor {
+	n := len(interceptors)
+
+	return func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+		chainer := func(currentInter grpc.StreamServerInterceptor, currentHandler grpc.StreamHandler) grpc.StreamHandler {
+			return func(currentSrv interface{}, currentStream grpc.ServerStream) error {
+				return currentInter(currentSrv, currentStream, info, currentHandler)
+			}
+		}
+
+		chainedHandler := handler
+		for i := n - 1; i >= 0; i-- {
+			chainedHandler = chainer(interceptors[i], chainedHandler)
+		}
+
+		return chainedHandler(srv, ss)
+	}
+}
diff --git a/vendor/github.com/authzed/spicedb/internal/middleware/datastore/datastore.go b/vendor/github.com/authzed/spicedb/internal/middleware/datastore/datastore.go
new file mode 100644
index 0000000..8c321b3
--- /dev/null
+++ b/vendor/github.com/authzed/spicedb/internal/middleware/datastore/datastore.go
@@ -0,0 +1,85 @@
+package datastore
+
+import (
+	"context"
+
+	middleware "github.com/grpc-ecosystem/go-grpc-middleware/v2"
+	"google.golang.org/grpc"
+
+	"github.com/authzed/spicedb/pkg/datastore"
+)
+
+type ctxKeyType struct{}
+
+var datastoreKey ctxKeyType = struct{}{}
+
+type datastoreHandle struct {
+	datastore datastore.Datastore
+}
+
+// ContextWithHandle adds a placeholder to a context that will later be
+// filled by the datastore
+func ContextWithHandle(ctx context.Context) context.Context {
+	return context.WithValue(ctx, datastoreKey, &datastoreHandle{})
+}
+
+// FromContext reads the selected datastore out of a context.Context
+// and returns nil if it does not exist.
+func FromContext(ctx context.Context) datastore.Datastore {
+	if c := ctx.Value(datastoreKey); c != nil {
+		handle := c.(*datastoreHandle)
+		return handle.datastore
+	}
+	return nil
+}
+
+// MustFromContext reads the selected datastore out of a context.Context and panics if it does not exist
+func MustFromContext(ctx context.Context) datastore.Datastore {
+	datastore := FromContext(ctx)
+	if datastore == nil {
+		panic("datastore middleware did not inject datastore")
+	}
+
+	return datastore
+}
+
+// SetInContext adds a datastore to the given context
+func SetInContext(ctx context.Context, datastore datastore.Datastore) error {
+	handle := ctx.Value(datastoreKey)
+	if handle == nil {
+		return nil
+	}
+	handle.(*datastoreHandle).datastore = datastore
+	return nil
+}
+
+// ContextWithDatastore adds the handle and datastore in one step
+func ContextWithDatastore(ctx context.Context, datastore datastore.Datastore) context.Context {
+	return context.WithValue(ctx, datastoreKey, &datastoreHandle{datastore: datastore})
+}
+
+// UnaryServerInterceptor returns a new unary server interceptor that adds the
+// datastore to the context
+func UnaryServerInterceptor(datastore datastore.Datastore) grpc.UnaryServerInterceptor {
+	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+		newCtx := ContextWithHandle(ctx)
+		if err := SetInContext(newCtx, datastore); err != nil {
+			return nil, err
+		}
+
+		return handler(newCtx, req)
+	}
+}
+
+// StreamServerInterceptor returns a new stream server interceptor that adds the
+// datastore to the context
+func StreamServerInterceptor(datastore datastore.Datastore) grpc.StreamServerInterceptor {
+	return func(srv interface{}, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+		wrapped := middleware.WrapServerStream(stream)
+		wrapped.WrappedContext = ContextWithHandle(wrapped.WrappedContext)
+		if err := SetInContext(wrapped.WrappedContext, datastore); err != nil {
+			return err
+		}
+		return handler(srv, wrapped)
+	}
+}
diff --git a/vendor/github.com/authzed/spicedb/internal/middleware/datastore/doc.go b/vendor/github.com/authzed/spicedb/internal/middleware/datastore/doc.go
new file mode 100644
index 0000000..a4d0cf0
--- /dev/null
+++ b/vendor/github.com/authzed/spicedb/internal/middleware/datastore/doc.go
@@ -0,0 +1,2 @@
+// Package datastore defines middleware that injects the datastore into the context.
+package datastore
diff --git a/vendor/github.com/authzed/spicedb/internal/middleware/doc.go b/vendor/github.com/authzed/spicedb/internal/middleware/doc.go
new file mode 100644
index 0000000..b11553e
--- /dev/null
+++ b/vendor/github.com/authzed/spicedb/internal/middleware/doc.go
@@ -0,0 +1,2 @@
+// Package middleware defines various custom middlewares.
+package middleware
diff --git a/vendor/github.com/authzed/spicedb/internal/middleware/handwrittenvalidation/doc.go b/vendor/github.com/authzed/spicedb/internal/middleware/handwrittenvalidation/doc.go
new file mode 100644
index 0000000..2d9aa01
--- /dev/null
+++ b/vendor/github.com/authzed/spicedb/internal/middleware/handwrittenvalidation/doc.go
@@ -0,0 +1,2 @@
+// Package handwrittenvalidation defines middleware that runs custom-made validations on incoming requests.
+package handwrittenvalidation
diff --git a/vendor/github.com/authzed/spicedb/internal/middleware/handwrittenvalidation/handwrittenvalidation.go b/vendor/github.com/authzed/spicedb/internal/middleware/handwrittenvalidation/handwrittenvalidation.go
new file mode 100644
index 0000000..2adc4b3
--- /dev/null
+++ b/vendor/github.com/authzed/spicedb/internal/middleware/handwrittenvalidation/handwrittenvalidation.go
@@ -0,0 +1,54 @@
+package handwrittenvalidation
+
+import (
+	"context"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+type handwrittenValidator interface {
+	HandwrittenValidate() error
+}
+
+// UnaryServerInterceptor returns a new unary server interceptor that runs the handwritten validation
+// on the incoming request, if any.
+func UnaryServerInterceptor(ctx context.Context, req interface{}, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+	validator, ok := req.(handwrittenValidator)
+	if ok {
+		err := validator.HandwrittenValidate()
+		if err != nil {
+			return nil, status.Errorf(codes.InvalidArgument, "%s", err)
+		}
+	}
+
+	return handler(ctx, req)
+}
+
+// StreamServerInterceptor returns a new stream server interceptor that runs the handwritten validation
+// on the incoming request messages, if any.
+func StreamServerInterceptor(srv interface{}, stream grpc.ServerStream, _ *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+	wrapper := &recvWrapper{stream}
+	return handler(srv, wrapper)
+}
+
+type recvWrapper struct {
+	grpc.ServerStream
+}
+
+func (s *recvWrapper) RecvMsg(m interface{}) error {
+	if err := s.ServerStream.RecvMsg(m); err != nil {
+		return err
+	}
+
+	validator, ok := m.(handwrittenValidator)
+	if ok {
+		err := validator.HandwrittenValidate()
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
diff --git a/vendor/github.com/authzed/spicedb/internal/middleware/servicespecific/doc.go b/vendor/github.com/authzed/spicedb/internal/middleware/servicespecific/doc.go
new file mode 100644
index 0000000..7e3ae0f
--- /dev/null
+++ b/vendor/github.com/authzed/spicedb/internal/middleware/servicespecific/doc.go
@@ -0,0 +1,2 @@
+// Package servicespecific defines middleware that injects other middlewares.
+package servicespecific
diff --git a/vendor/github.com/authzed/spicedb/internal/middleware/servicespecific/servicespecific.go b/vendor/github.com/authzed/spicedb/internal/middleware/servicespecific/servicespecific.go
new file mode 100644
index 0000000..10fe753
--- /dev/null
+++ b/vendor/github.com/authzed/spicedb/internal/middleware/servicespecific/servicespecific.go
@@ -0,0 +1,39 @@
+package servicespecific
+
+import (
+	"context"
+
+	"google.golang.org/grpc"
+)
+
+// ExtraUnaryInterceptor is an interface for a service which has its own bundled
+// unary interceptors that must be run.
+type ExtraUnaryInterceptor interface {
+	UnaryInterceptor() grpc.UnaryServerInterceptor
+}
+
+// ExtraStreamInterceptor is an interface for a service which has its own bundled
+// stream interceptors that must be run.
+type ExtraStreamInterceptor interface {
+	StreamInterceptor() grpc.StreamServerInterceptor
+}
+
+// UnaryServerInterceptor returns a new unary server interceptor that runs bundled interceptors.
+func UnaryServerInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+	if hasExtraInterceptor, ok := info.Server.(ExtraUnaryInterceptor); ok {
+		interceptor := hasExtraInterceptor.UnaryInterceptor()
+		return interceptor(ctx, req, info, handler)
+	}
+
+	return handler(ctx, req)
+}
+
+// StreamServerInterceptor returns a new stream server interceptor that runs bundled interceptors.
+func StreamServerInterceptor(srv interface{}, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+	if hasExtraInterceptor, ok := srv.(ExtraStreamInterceptor); ok {
+		interceptor := hasExtraInterceptor.StreamInterceptor()
+		return interceptor(srv, stream, info, handler)
+	}
+
+	return handler(srv, stream)
+}
diff --git a/vendor/github.com/authzed/spicedb/internal/middleware/streamtimeout/doc.go b/vendor/github.com/authzed/spicedb/internal/middleware/streamtimeout/doc.go
new file mode 100644
index 0000000..9eceb4d
--- /dev/null
+++ b/vendor/github.com/authzed/spicedb/internal/middleware/streamtimeout/doc.go
@@ -0,0 +1,2 @@
+// Package streamtimeout defines middleware that cancels the context after a timeout if no new data has been received.
+package streamtimeout
diff --git a/vendor/github.com/authzed/spicedb/internal/middleware/streamtimeout/streamtimeout.go b/vendor/github.com/authzed/spicedb/internal/middleware/streamtimeout/streamtimeout.go
new file mode 100644
index 0000000..8f09fdb
--- /dev/null
+++ b/vendor/github.com/authzed/spicedb/internal/middleware/streamtimeout/streamtimeout.go
@@ -0,0 +1,57 @@
+package streamtimeout
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+
+	"github.com/authzed/spicedb/pkg/spiceerrors"
+)
+
+// MustStreamServerInterceptor returns a new stream server interceptor that cancels the context
+// after a timeout if no new data has been received.
+func MustStreamServerInterceptor(timeout time.Duration) grpc.StreamServerInterceptor {
+	if timeout <= 0 {
+		panic("timeout must be >= 0 for streaming timeout interceptor")
+	}
+
+	return func(srv any, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+		ctx := stream.Context()
+		withCancel, internalCancelFn := context.WithCancelCause(ctx)
+		timer := time.AfterFunc(timeout, func() {
+			internalCancelFn(spiceerrors.WithCodeAndDetailsAsError(fmt.Errorf("operation took longer than allowed %v to complete", timeout), codes.DeadlineExceeded))
+		})
+		wrapper := &sendWrapper{stream, withCancel, timer, timeout}
+		return handler(srv, wrapper)
+	}
+}
+
+type sendWrapper struct {
+	grpc.ServerStream
+
+	ctx     context.Context
+	timer   *time.Timer
+	timeout time.Duration
+}
+
+func (s *sendWrapper) Context() context.Context {
+	return s.ctx
+}
+
+func (s *sendWrapper) SetTrailer(_ metadata.MD) {
+	s.timer.Stop()
+}
+
+func (s *sendWrapper) SendMsg(m any) error {
+	err := s.ServerStream.SendMsg(m)
+	if err != nil {
+		s.timer.Stop()
+	} else {
+		s.timer.Reset(s.timeout)
+	}
+	return err
+}
diff --git a/vendor/github.com/authzed/spicedb/internal/middleware/usagemetrics/doc.go b/vendor/github.com/authzed/spicedb/internal/middleware/usagemetrics/doc.go
new file mode 100644
index 0000000..c05cacc
--- /dev/null
+++ b/vendor/github.com/authzed/spicedb/internal/middleware/usagemetrics/doc.go
@@ -0,0 +1,2 @@
+// Package usagemetrics defines middleware that adds usage data (e.g. dispatch counts) to the response.
+package usagemetrics
diff --git a/vendor/github.com/authzed/spicedb/internal/middleware/usagemetrics/usagemetrics.go b/vendor/github.com/authzed/spicedb/internal/middleware/usagemetrics/usagemetrics.go
new file mode 100644
index 0000000..32f5676
--- /dev/null
+++ b/vendor/github.com/authzed/spicedb/internal/middleware/usagemetrics/usagemetrics.go
@@ -0,0 +1,128 @@
+package usagemetrics
+
+import (
+	"context"
+	"strconv"
+	"time"
+
+	"github.com/authzed/authzed-go/pkg/responsemeta"
+	"github.com/authzed/grpcutil"
+	"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	"google.golang.org/grpc"
+
+	log "github.com/authzed/spicedb/internal/logging"
+	dispatch "github.com/authzed/spicedb/pkg/proto/dispatch/v1"
+)
+
+var (
+	// DispatchedCountLabels are the labels that DispatchedCountHistogram will
+	// have by default.
+	DispatchedCountLabels = []string{"method", "cached"}
+
+	// DispatchedCountHistogram is the metric that SpiceDB uses to keep track
+	// of the number of downstream dispatches that are performed to answer a
+	// single query.
+	DispatchedCountHistogram = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: "spicedb",
+		Subsystem: "services",
+		Name:      "dispatches",
+		Help:      "Histogram of cluster dispatches performed by the instance.",
+		Buckets:   []float64{1, 5, 10, 25, 50, 100, 250},
+	}, DispatchedCountLabels)
+)
+
+type reporter struct{}
+
+func (r *reporter) ServerReporter(ctx context.Context, callMeta interceptors.CallMeta) (interceptors.Reporter, context.Context) {
+	_, methodName := grpcutil.SplitMethodName(callMeta.FullMethod())
+	ctx = ContextWithHandle(ctx)
+	return &serverReporter{ctx: ctx, methodName: methodName}, ctx
+}
+
+type serverReporter struct {
+	interceptors.NoopReporter
+	ctx        context.Context
+	methodName string
+}
+
+func (r *serverReporter) PostCall(_ error, _ time.Duration) {
+	responseMeta := FromContext(r.ctx)
+	if responseMeta == nil {
+		responseMeta = &dispatch.ResponseMeta{}
+	}
+
+	err := annotateAndReportForMetadata(r.ctx, r.methodName, responseMeta)
+	// if context is cancelled, the stream will be closed, and gRPC will return ErrIllegalHeaderWrite
+	// this prevents logging unnecessary error messages
+	if r.ctx.Err() != nil {
+		return
+	}
+	if err != nil {
+		log.Ctx(r.ctx).Warn().Err(err).Msg("usagemetrics: could not report metadata")
+	}
+}
+
+// UnaryServerInterceptor implements a gRPC Middleware for reporting usage metrics
+// in both the trailer of the request, as well as to the registered prometheus
+// metrics.
+func UnaryServerInterceptor() grpc.UnaryServerInterceptor {
+	return interceptors.UnaryServerInterceptor(&reporter{})
+}
+
+// StreamServerInterceptor implements a gRPC Middleware for reporting usage metrics
+// in both the trailer of the request, as well as to the registered prometheus
+// metrics
+func StreamServerInterceptor() grpc.StreamServerInterceptor {
+	return interceptors.StreamServerInterceptor(&reporter{})
+}
+
+func annotateAndReportForMetadata(ctx context.Context, methodName string, metadata *dispatch.ResponseMeta) error {
+	DispatchedCountHistogram.WithLabelValues(methodName, "false").Observe(float64(metadata.DispatchCount))
+	DispatchedCountHistogram.WithLabelValues(methodName, "true").Observe(float64(metadata.CachedDispatchCount))
+
+	return responsemeta.SetResponseTrailerMetadata(ctx, map[responsemeta.ResponseMetadataTrailerKey]string{
+		responsemeta.DispatchedOperationsCount: strconv.Itoa(int(metadata.DispatchCount)),
+		responsemeta.CachedOperationsCount:     strconv.Itoa(int(metadata.CachedDispatchCount)),
+	})
+}
+
+// Create a new type to prevent context collisions
+type responseMetaKey string
+
+var metadataCtxKey responseMetaKey = "dispatched-response-meta"
+
+type metaHandle struct{ metadata *dispatch.ResponseMeta }
+
+// SetInContext should be called in a gRPC handler to correctly set the response metadata
+// for the dispatched request.
+func SetInContext(ctx context.Context, metadata *dispatch.ResponseMeta) {
+	possibleHandle := ctx.Value(metadataCtxKey)
+	if possibleHandle == nil {
+		return
+	}
+
+	handle := possibleHandle.(*metaHandle)
+	handle.metadata = metadata
+}
+
+// FromContext returns any metadata that was stored in the context.
+//
+// This is useful for testing that a handler is properly setting the context.
+func FromContext(ctx context.Context) *dispatch.ResponseMeta {
+	possibleHandle := ctx.Value(metadataCtxKey)
+	if possibleHandle == nil {
+		return nil
+	}
+	return possibleHandle.(*metaHandle).metadata
+}
+
+// ContextWithHandle creates a new context with a location to store metadata
+// returned from a dispatched request.
+//
+// This should only be called in middleware or testing functions.
+func ContextWithHandle(ctx context.Context) context.Context {
+	var handle metaHandle
+	return context.WithValue(ctx, metadataCtxKey, &handle)
+}