From 83876eae868bd1e4fb6b9a823a6e8173919f290d Mon Sep 17 00:00:00 2001 From: bndw Date: Fri, 13 Feb 2026 18:26:53 -0800 Subject: feat: add Connect (gRPC over HTTP/JSON) support Connect integration: - Buf Connect codegen added to buf.gen.yaml - Connect handler wraps gRPC server - Serves on same port as WebSocket (:8080) - HTTP/2 with h2c for cleartext HTTP/2 Now serving THREE protocols: 1. gRPC (native) on :50051 - binary, high performance 2. Connect on :8080/nostr.v1.NostrRelay/* - HTTP/JSON, browser compatible 3. WebSocket on :8080/ - Nostr standard protocol All three protocols share: - Same storage layer - Same subscription manager - Same validation logic Browser-friendly! Call gRPC methods with fetch() or curl. --- README.md | 25 +- api/nostr/v1/nostrv1connect/nostr.connect.go | 420 +++++++++++++++++++++++++++ buf.gen.yaml | 4 + cmd/relay/main.go | 18 +- go.mod | 7 +- go.sum | 10 + internal/handler/connect/handler.go | 101 +++++++ 7 files changed, 575 insertions(+), 10 deletions(-) create mode 100644 api/nostr/v1/nostrv1connect/nostr.connect.go create mode 100644 internal/handler/connect/handler.go diff --git a/README.md b/README.md index 8fd35c5..35e697c 100644 --- a/README.md +++ b/README.md @@ -28,8 +28,10 @@ make build-all # Build both ``` The relay will start: -- **gRPC** on `:50051` -- **WebSocket** (Nostr) on `:8080` +- **gRPC** (native) on `:50051` +- **HTTP** server on `:8080`: + - **Connect** (gRPC over HTTP/JSON) at `/nostr.v1.NostrRelay/*` + - **WebSocket** (Nostr protocol) at `/` ### Test with Client @@ -64,6 +66,16 @@ nak req -k 1 --limit 10 ws://localhost:8080 echo '{"kind":1,"content":"hello","tags":[]}' | nak event --sec | nak publish ws://localhost:8080 ``` +**With Connect (HTTP/JSON):** +```bash +# Call gRPC methods over HTTP with JSON +curl -X POST http://localhost:8080/nostr.v1.NostrRelay/PublishEvent \ + -H "Content-Type: application/json" \ + -d '{"event": {...}}' + +# Works from browsers, curl, fetch(), etc. +``` + ## gRPC API See [proto/nostr/v1/nostr.proto](proto/nostr/v1/nostr.proto) for the full API. @@ -82,13 +94,16 @@ See [proto/nostr/v1/nostr.proto](proto/nostr/v1/nostr.proto) for the full API. **Phase 1: Complete** ✅ - ✅ SQLite storage with binary-first design - ✅ Event validation (ID, signature) -- ✅ gRPC publish/query API +- ✅ **Triple protocol support:** + - **gRPC** (native binary protocol) + - **Connect** (gRPC over HTTP/JSON - browser compatible!) + - **WebSocket** (NIP-01 - standard Nostr protocol) - ✅ Subscribe/streaming (real-time event delivery) - ✅ Subscription management (filter matching, fan-out) -- ✅ **WebSocket server (NIP-01) - standard Nostr clients work!** **Compatible with:** -- Any gRPC client (custom or generated) +- Any gRPC client (Go, Python, JS, etc.) +- Any HTTP client (curl, fetch, browsers) - Any Nostr client (Damus, Amethyst, Snort, Iris, Gossip, etc.) - nak CLI for testing diff --git a/api/nostr/v1/nostrv1connect/nostr.connect.go b/api/nostr/v1/nostrv1connect/nostr.connect.go new file mode 100644 index 0000000..fd98039 --- /dev/null +++ b/api/nostr/v1/nostrv1connect/nostr.connect.go @@ -0,0 +1,420 @@ +// Code generated by protoc-gen-connect-go. DO NOT EDIT. +// +// Source: nostr/v1/nostr.proto + +package nostrv1connect + +import ( + connect "connectrpc.com/connect" + context "context" + errors "errors" + http "net/http" + v1 "northwest.io/nostr-grpc/api/nostr/v1" + strings "strings" +) + +// This is a compile-time assertion to ensure that this generated file and the connect package are +// compatible. If you get a compiler error that this constant is not defined, this code was +// generated with a version of connect newer than the one compiled into your binary. You can fix the +// problem by either regenerating this code with an older version of connect or updating the connect +// version compiled into your binary. +const _ = connect.IsAtLeastVersion1_13_0 + +const ( + // NostrRelayName is the fully-qualified name of the NostrRelay service. + NostrRelayName = "nostr.v1.NostrRelay" + // RelayAdminName is the fully-qualified name of the RelayAdmin service. + RelayAdminName = "nostr.v1.RelayAdmin" +) + +// These constants are the fully-qualified names of the RPCs defined in this package. They're +// exposed at runtime as Spec.Procedure and as the final two segments of the HTTP route. +// +// Note that these are different from the fully-qualified method names used by +// google.golang.org/protobuf/reflect/protoreflect. To convert from these constants to +// reflection-formatted method names, remove the leading slash and convert the remaining slash to a +// period. +const ( + // NostrRelayPublishEventProcedure is the fully-qualified name of the NostrRelay's PublishEvent RPC. + NostrRelayPublishEventProcedure = "/nostr.v1.NostrRelay/PublishEvent" + // NostrRelaySubscribeProcedure is the fully-qualified name of the NostrRelay's Subscribe RPC. + NostrRelaySubscribeProcedure = "/nostr.v1.NostrRelay/Subscribe" + // NostrRelayUnsubscribeProcedure is the fully-qualified name of the NostrRelay's Unsubscribe RPC. + NostrRelayUnsubscribeProcedure = "/nostr.v1.NostrRelay/Unsubscribe" + // NostrRelayPublishBatchProcedure is the fully-qualified name of the NostrRelay's PublishBatch RPC. + NostrRelayPublishBatchProcedure = "/nostr.v1.NostrRelay/PublishBatch" + // NostrRelayQueryEventsProcedure is the fully-qualified name of the NostrRelay's QueryEvents RPC. + NostrRelayQueryEventsProcedure = "/nostr.v1.NostrRelay/QueryEvents" + // NostrRelayCountEventsProcedure is the fully-qualified name of the NostrRelay's CountEvents RPC. + NostrRelayCountEventsProcedure = "/nostr.v1.NostrRelay/CountEvents" + // RelayAdminGetStatsProcedure is the fully-qualified name of the RelayAdmin's GetStats RPC. + RelayAdminGetStatsProcedure = "/nostr.v1.RelayAdmin/GetStats" + // RelayAdminGetConnectionsProcedure is the fully-qualified name of the RelayAdmin's GetConnections + // RPC. + RelayAdminGetConnectionsProcedure = "/nostr.v1.RelayAdmin/GetConnections" + // RelayAdminBanPublicKeyProcedure is the fully-qualified name of the RelayAdmin's BanPublicKey RPC. + RelayAdminBanPublicKeyProcedure = "/nostr.v1.RelayAdmin/BanPublicKey" + // RelayAdminGetStorageInfoProcedure is the fully-qualified name of the RelayAdmin's GetStorageInfo + // RPC. + RelayAdminGetStorageInfoProcedure = "/nostr.v1.RelayAdmin/GetStorageInfo" +) + +// NostrRelayClient is a client for the nostr.v1.NostrRelay service. +type NostrRelayClient interface { + // Publish a single event + PublishEvent(context.Context, *connect.Request[v1.PublishEventRequest]) (*connect.Response[v1.PublishEventResponse], error) + // Subscribe to events matching filters (streaming) + Subscribe(context.Context, *connect.Request[v1.SubscribeRequest]) (*connect.ServerStreamForClient[v1.Event], error) + // Unsubscribe from an active subscription + Unsubscribe(context.Context, *connect.Request[v1.UnsubscribeRequest]) (*connect.Response[v1.Empty], error) + // gRPC-specific: batch publish + PublishBatch(context.Context, *connect.Request[v1.PublishBatchRequest]) (*connect.Response[v1.PublishBatchResponse], error) + // gRPC-specific: paginated query (non-streaming) + QueryEvents(context.Context, *connect.Request[v1.QueryRequest]) (*connect.Response[v1.QueryResponse], error) + // Event counts (NIP-45) + CountEvents(context.Context, *connect.Request[v1.CountRequest]) (*connect.Response[v1.CountResponse], error) +} + +// NewNostrRelayClient constructs a client for the nostr.v1.NostrRelay service. By default, it uses +// the Connect protocol with the binary Protobuf Codec, asks for gzipped responses, and sends +// uncompressed requests. To use the gRPC or gRPC-Web protocols, supply the connect.WithGRPC() or +// connect.WithGRPCWeb() options. +// +// The URL supplied here should be the base URL for the Connect or gRPC server (for example, +// http://api.acme.com or https://acme.com/grpc). +func NewNostrRelayClient(httpClient connect.HTTPClient, baseURL string, opts ...connect.ClientOption) NostrRelayClient { + baseURL = strings.TrimRight(baseURL, "/") + nostrRelayMethods := v1.File_nostr_v1_nostr_proto.Services().ByName("NostrRelay").Methods() + return &nostrRelayClient{ + publishEvent: connect.NewClient[v1.PublishEventRequest, v1.PublishEventResponse]( + httpClient, + baseURL+NostrRelayPublishEventProcedure, + connect.WithSchema(nostrRelayMethods.ByName("PublishEvent")), + connect.WithClientOptions(opts...), + ), + subscribe: connect.NewClient[v1.SubscribeRequest, v1.Event]( + httpClient, + baseURL+NostrRelaySubscribeProcedure, + connect.WithSchema(nostrRelayMethods.ByName("Subscribe")), + connect.WithClientOptions(opts...), + ), + unsubscribe: connect.NewClient[v1.UnsubscribeRequest, v1.Empty]( + httpClient, + baseURL+NostrRelayUnsubscribeProcedure, + connect.WithSchema(nostrRelayMethods.ByName("Unsubscribe")), + connect.WithClientOptions(opts...), + ), + publishBatch: connect.NewClient[v1.PublishBatchRequest, v1.PublishBatchResponse]( + httpClient, + baseURL+NostrRelayPublishBatchProcedure, + connect.WithSchema(nostrRelayMethods.ByName("PublishBatch")), + connect.WithClientOptions(opts...), + ), + queryEvents: connect.NewClient[v1.QueryRequest, v1.QueryResponse]( + httpClient, + baseURL+NostrRelayQueryEventsProcedure, + connect.WithSchema(nostrRelayMethods.ByName("QueryEvents")), + connect.WithClientOptions(opts...), + ), + countEvents: connect.NewClient[v1.CountRequest, v1.CountResponse]( + httpClient, + baseURL+NostrRelayCountEventsProcedure, + connect.WithSchema(nostrRelayMethods.ByName("CountEvents")), + connect.WithClientOptions(opts...), + ), + } +} + +// nostrRelayClient implements NostrRelayClient. +type nostrRelayClient struct { + publishEvent *connect.Client[v1.PublishEventRequest, v1.PublishEventResponse] + subscribe *connect.Client[v1.SubscribeRequest, v1.Event] + unsubscribe *connect.Client[v1.UnsubscribeRequest, v1.Empty] + publishBatch *connect.Client[v1.PublishBatchRequest, v1.PublishBatchResponse] + queryEvents *connect.Client[v1.QueryRequest, v1.QueryResponse] + countEvents *connect.Client[v1.CountRequest, v1.CountResponse] +} + +// PublishEvent calls nostr.v1.NostrRelay.PublishEvent. +func (c *nostrRelayClient) PublishEvent(ctx context.Context, req *connect.Request[v1.PublishEventRequest]) (*connect.Response[v1.PublishEventResponse], error) { + return c.publishEvent.CallUnary(ctx, req) +} + +// Subscribe calls nostr.v1.NostrRelay.Subscribe. +func (c *nostrRelayClient) Subscribe(ctx context.Context, req *connect.Request[v1.SubscribeRequest]) (*connect.ServerStreamForClient[v1.Event], error) { + return c.subscribe.CallServerStream(ctx, req) +} + +// Unsubscribe calls nostr.v1.NostrRelay.Unsubscribe. +func (c *nostrRelayClient) Unsubscribe(ctx context.Context, req *connect.Request[v1.UnsubscribeRequest]) (*connect.Response[v1.Empty], error) { + return c.unsubscribe.CallUnary(ctx, req) +} + +// PublishBatch calls nostr.v1.NostrRelay.PublishBatch. +func (c *nostrRelayClient) PublishBatch(ctx context.Context, req *connect.Request[v1.PublishBatchRequest]) (*connect.Response[v1.PublishBatchResponse], error) { + return c.publishBatch.CallUnary(ctx, req) +} + +// QueryEvents calls nostr.v1.NostrRelay.QueryEvents. +func (c *nostrRelayClient) QueryEvents(ctx context.Context, req *connect.Request[v1.QueryRequest]) (*connect.Response[v1.QueryResponse], error) { + return c.queryEvents.CallUnary(ctx, req) +} + +// CountEvents calls nostr.v1.NostrRelay.CountEvents. +func (c *nostrRelayClient) CountEvents(ctx context.Context, req *connect.Request[v1.CountRequest]) (*connect.Response[v1.CountResponse], error) { + return c.countEvents.CallUnary(ctx, req) +} + +// NostrRelayHandler is an implementation of the nostr.v1.NostrRelay service. +type NostrRelayHandler interface { + // Publish a single event + PublishEvent(context.Context, *connect.Request[v1.PublishEventRequest]) (*connect.Response[v1.PublishEventResponse], error) + // Subscribe to events matching filters (streaming) + Subscribe(context.Context, *connect.Request[v1.SubscribeRequest], *connect.ServerStream[v1.Event]) error + // Unsubscribe from an active subscription + Unsubscribe(context.Context, *connect.Request[v1.UnsubscribeRequest]) (*connect.Response[v1.Empty], error) + // gRPC-specific: batch publish + PublishBatch(context.Context, *connect.Request[v1.PublishBatchRequest]) (*connect.Response[v1.PublishBatchResponse], error) + // gRPC-specific: paginated query (non-streaming) + QueryEvents(context.Context, *connect.Request[v1.QueryRequest]) (*connect.Response[v1.QueryResponse], error) + // Event counts (NIP-45) + CountEvents(context.Context, *connect.Request[v1.CountRequest]) (*connect.Response[v1.CountResponse], error) +} + +// NewNostrRelayHandler builds an HTTP handler from the service implementation. It returns the path +// on which to mount the handler and the handler itself. +// +// By default, handlers support the Connect, gRPC, and gRPC-Web protocols with the binary Protobuf +// and JSON codecs. They also support gzip compression. +func NewNostrRelayHandler(svc NostrRelayHandler, opts ...connect.HandlerOption) (string, http.Handler) { + nostrRelayMethods := v1.File_nostr_v1_nostr_proto.Services().ByName("NostrRelay").Methods() + nostrRelayPublishEventHandler := connect.NewUnaryHandler( + NostrRelayPublishEventProcedure, + svc.PublishEvent, + connect.WithSchema(nostrRelayMethods.ByName("PublishEvent")), + connect.WithHandlerOptions(opts...), + ) + nostrRelaySubscribeHandler := connect.NewServerStreamHandler( + NostrRelaySubscribeProcedure, + svc.Subscribe, + connect.WithSchema(nostrRelayMethods.ByName("Subscribe")), + connect.WithHandlerOptions(opts...), + ) + nostrRelayUnsubscribeHandler := connect.NewUnaryHandler( + NostrRelayUnsubscribeProcedure, + svc.Unsubscribe, + connect.WithSchema(nostrRelayMethods.ByName("Unsubscribe")), + connect.WithHandlerOptions(opts...), + ) + nostrRelayPublishBatchHandler := connect.NewUnaryHandler( + NostrRelayPublishBatchProcedure, + svc.PublishBatch, + connect.WithSchema(nostrRelayMethods.ByName("PublishBatch")), + connect.WithHandlerOptions(opts...), + ) + nostrRelayQueryEventsHandler := connect.NewUnaryHandler( + NostrRelayQueryEventsProcedure, + svc.QueryEvents, + connect.WithSchema(nostrRelayMethods.ByName("QueryEvents")), + connect.WithHandlerOptions(opts...), + ) + nostrRelayCountEventsHandler := connect.NewUnaryHandler( + NostrRelayCountEventsProcedure, + svc.CountEvents, + connect.WithSchema(nostrRelayMethods.ByName("CountEvents")), + connect.WithHandlerOptions(opts...), + ) + return "/nostr.v1.NostrRelay/", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case NostrRelayPublishEventProcedure: + nostrRelayPublishEventHandler.ServeHTTP(w, r) + case NostrRelaySubscribeProcedure: + nostrRelaySubscribeHandler.ServeHTTP(w, r) + case NostrRelayUnsubscribeProcedure: + nostrRelayUnsubscribeHandler.ServeHTTP(w, r) + case NostrRelayPublishBatchProcedure: + nostrRelayPublishBatchHandler.ServeHTTP(w, r) + case NostrRelayQueryEventsProcedure: + nostrRelayQueryEventsHandler.ServeHTTP(w, r) + case NostrRelayCountEventsProcedure: + nostrRelayCountEventsHandler.ServeHTTP(w, r) + default: + http.NotFound(w, r) + } + }) +} + +// UnimplementedNostrRelayHandler returns CodeUnimplemented from all methods. +type UnimplementedNostrRelayHandler struct{} + +func (UnimplementedNostrRelayHandler) PublishEvent(context.Context, *connect.Request[v1.PublishEventRequest]) (*connect.Response[v1.PublishEventResponse], error) { + return nil, connect.NewError(connect.CodeUnimplemented, errors.New("nostr.v1.NostrRelay.PublishEvent is not implemented")) +} + +func (UnimplementedNostrRelayHandler) Subscribe(context.Context, *connect.Request[v1.SubscribeRequest], *connect.ServerStream[v1.Event]) error { + return connect.NewError(connect.CodeUnimplemented, errors.New("nostr.v1.NostrRelay.Subscribe is not implemented")) +} + +func (UnimplementedNostrRelayHandler) Unsubscribe(context.Context, *connect.Request[v1.UnsubscribeRequest]) (*connect.Response[v1.Empty], error) { + return nil, connect.NewError(connect.CodeUnimplemented, errors.New("nostr.v1.NostrRelay.Unsubscribe is not implemented")) +} + +func (UnimplementedNostrRelayHandler) PublishBatch(context.Context, *connect.Request[v1.PublishBatchRequest]) (*connect.Response[v1.PublishBatchResponse], error) { + return nil, connect.NewError(connect.CodeUnimplemented, errors.New("nostr.v1.NostrRelay.PublishBatch is not implemented")) +} + +func (UnimplementedNostrRelayHandler) QueryEvents(context.Context, *connect.Request[v1.QueryRequest]) (*connect.Response[v1.QueryResponse], error) { + return nil, connect.NewError(connect.CodeUnimplemented, errors.New("nostr.v1.NostrRelay.QueryEvents is not implemented")) +} + +func (UnimplementedNostrRelayHandler) CountEvents(context.Context, *connect.Request[v1.CountRequest]) (*connect.Response[v1.CountResponse], error) { + return nil, connect.NewError(connect.CodeUnimplemented, errors.New("nostr.v1.NostrRelay.CountEvents is not implemented")) +} + +// RelayAdminClient is a client for the nostr.v1.RelayAdmin service. +type RelayAdminClient interface { + GetStats(context.Context, *connect.Request[v1.Empty]) (*connect.Response[v1.RelayStats], error) + GetConnections(context.Context, *connect.Request[v1.Empty]) (*connect.Response[v1.ConnectionList], error) + BanPublicKey(context.Context, *connect.Request[v1.BanRequest]) (*connect.Response[v1.Empty], error) + GetStorageInfo(context.Context, *connect.Request[v1.Empty]) (*connect.Response[v1.StorageStats], error) +} + +// NewRelayAdminClient constructs a client for the nostr.v1.RelayAdmin service. By default, it uses +// the Connect protocol with the binary Protobuf Codec, asks for gzipped responses, and sends +// uncompressed requests. To use the gRPC or gRPC-Web protocols, supply the connect.WithGRPC() or +// connect.WithGRPCWeb() options. +// +// The URL supplied here should be the base URL for the Connect or gRPC server (for example, +// http://api.acme.com or https://acme.com/grpc). +func NewRelayAdminClient(httpClient connect.HTTPClient, baseURL string, opts ...connect.ClientOption) RelayAdminClient { + baseURL = strings.TrimRight(baseURL, "/") + relayAdminMethods := v1.File_nostr_v1_nostr_proto.Services().ByName("RelayAdmin").Methods() + return &relayAdminClient{ + getStats: connect.NewClient[v1.Empty, v1.RelayStats]( + httpClient, + baseURL+RelayAdminGetStatsProcedure, + connect.WithSchema(relayAdminMethods.ByName("GetStats")), + connect.WithClientOptions(opts...), + ), + getConnections: connect.NewClient[v1.Empty, v1.ConnectionList]( + httpClient, + baseURL+RelayAdminGetConnectionsProcedure, + connect.WithSchema(relayAdminMethods.ByName("GetConnections")), + connect.WithClientOptions(opts...), + ), + banPublicKey: connect.NewClient[v1.BanRequest, v1.Empty]( + httpClient, + baseURL+RelayAdminBanPublicKeyProcedure, + connect.WithSchema(relayAdminMethods.ByName("BanPublicKey")), + connect.WithClientOptions(opts...), + ), + getStorageInfo: connect.NewClient[v1.Empty, v1.StorageStats]( + httpClient, + baseURL+RelayAdminGetStorageInfoProcedure, + connect.WithSchema(relayAdminMethods.ByName("GetStorageInfo")), + connect.WithClientOptions(opts...), + ), + } +} + +// relayAdminClient implements RelayAdminClient. +type relayAdminClient struct { + getStats *connect.Client[v1.Empty, v1.RelayStats] + getConnections *connect.Client[v1.Empty, v1.ConnectionList] + banPublicKey *connect.Client[v1.BanRequest, v1.Empty] + getStorageInfo *connect.Client[v1.Empty, v1.StorageStats] +} + +// GetStats calls nostr.v1.RelayAdmin.GetStats. +func (c *relayAdminClient) GetStats(ctx context.Context, req *connect.Request[v1.Empty]) (*connect.Response[v1.RelayStats], error) { + return c.getStats.CallUnary(ctx, req) +} + +// GetConnections calls nostr.v1.RelayAdmin.GetConnections. +func (c *relayAdminClient) GetConnections(ctx context.Context, req *connect.Request[v1.Empty]) (*connect.Response[v1.ConnectionList], error) { + return c.getConnections.CallUnary(ctx, req) +} + +// BanPublicKey calls nostr.v1.RelayAdmin.BanPublicKey. +func (c *relayAdminClient) BanPublicKey(ctx context.Context, req *connect.Request[v1.BanRequest]) (*connect.Response[v1.Empty], error) { + return c.banPublicKey.CallUnary(ctx, req) +} + +// GetStorageInfo calls nostr.v1.RelayAdmin.GetStorageInfo. +func (c *relayAdminClient) GetStorageInfo(ctx context.Context, req *connect.Request[v1.Empty]) (*connect.Response[v1.StorageStats], error) { + return c.getStorageInfo.CallUnary(ctx, req) +} + +// RelayAdminHandler is an implementation of the nostr.v1.RelayAdmin service. +type RelayAdminHandler interface { + GetStats(context.Context, *connect.Request[v1.Empty]) (*connect.Response[v1.RelayStats], error) + GetConnections(context.Context, *connect.Request[v1.Empty]) (*connect.Response[v1.ConnectionList], error) + BanPublicKey(context.Context, *connect.Request[v1.BanRequest]) (*connect.Response[v1.Empty], error) + GetStorageInfo(context.Context, *connect.Request[v1.Empty]) (*connect.Response[v1.StorageStats], error) +} + +// NewRelayAdminHandler builds an HTTP handler from the service implementation. It returns the path +// on which to mount the handler and the handler itself. +// +// By default, handlers support the Connect, gRPC, and gRPC-Web protocols with the binary Protobuf +// and JSON codecs. They also support gzip compression. +func NewRelayAdminHandler(svc RelayAdminHandler, opts ...connect.HandlerOption) (string, http.Handler) { + relayAdminMethods := v1.File_nostr_v1_nostr_proto.Services().ByName("RelayAdmin").Methods() + relayAdminGetStatsHandler := connect.NewUnaryHandler( + RelayAdminGetStatsProcedure, + svc.GetStats, + connect.WithSchema(relayAdminMethods.ByName("GetStats")), + connect.WithHandlerOptions(opts...), + ) + relayAdminGetConnectionsHandler := connect.NewUnaryHandler( + RelayAdminGetConnectionsProcedure, + svc.GetConnections, + connect.WithSchema(relayAdminMethods.ByName("GetConnections")), + connect.WithHandlerOptions(opts...), + ) + relayAdminBanPublicKeyHandler := connect.NewUnaryHandler( + RelayAdminBanPublicKeyProcedure, + svc.BanPublicKey, + connect.WithSchema(relayAdminMethods.ByName("BanPublicKey")), + connect.WithHandlerOptions(opts...), + ) + relayAdminGetStorageInfoHandler := connect.NewUnaryHandler( + RelayAdminGetStorageInfoProcedure, + svc.GetStorageInfo, + connect.WithSchema(relayAdminMethods.ByName("GetStorageInfo")), + connect.WithHandlerOptions(opts...), + ) + return "/nostr.v1.RelayAdmin/", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case RelayAdminGetStatsProcedure: + relayAdminGetStatsHandler.ServeHTTP(w, r) + case RelayAdminGetConnectionsProcedure: + relayAdminGetConnectionsHandler.ServeHTTP(w, r) + case RelayAdminBanPublicKeyProcedure: + relayAdminBanPublicKeyHandler.ServeHTTP(w, r) + case RelayAdminGetStorageInfoProcedure: + relayAdminGetStorageInfoHandler.ServeHTTP(w, r) + default: + http.NotFound(w, r) + } + }) +} + +// UnimplementedRelayAdminHandler returns CodeUnimplemented from all methods. +type UnimplementedRelayAdminHandler struct{} + +func (UnimplementedRelayAdminHandler) GetStats(context.Context, *connect.Request[v1.Empty]) (*connect.Response[v1.RelayStats], error) { + return nil, connect.NewError(connect.CodeUnimplemented, errors.New("nostr.v1.RelayAdmin.GetStats is not implemented")) +} + +func (UnimplementedRelayAdminHandler) GetConnections(context.Context, *connect.Request[v1.Empty]) (*connect.Response[v1.ConnectionList], error) { + return nil, connect.NewError(connect.CodeUnimplemented, errors.New("nostr.v1.RelayAdmin.GetConnections is not implemented")) +} + +func (UnimplementedRelayAdminHandler) BanPublicKey(context.Context, *connect.Request[v1.BanRequest]) (*connect.Response[v1.Empty], error) { + return nil, connect.NewError(connect.CodeUnimplemented, errors.New("nostr.v1.RelayAdmin.BanPublicKey is not implemented")) +} + +func (UnimplementedRelayAdminHandler) GetStorageInfo(context.Context, *connect.Request[v1.Empty]) (*connect.Response[v1.StorageStats], error) { + return nil, connect.NewError(connect.CodeUnimplemented, errors.New("nostr.v1.RelayAdmin.GetStorageInfo is not implemented")) +} diff --git a/buf.gen.yaml b/buf.gen.yaml index e6f485a..01fea00 100644 --- a/buf.gen.yaml +++ b/buf.gen.yaml @@ -13,3 +13,7 @@ plugins: out: api opt: - paths=source_relative + - remote: buf.build/connectrpc/go + out: api + opt: + - paths=source_relative diff --git a/cmd/relay/main.go b/cmd/relay/main.go index 53296b9..9cf6ad6 100644 --- a/cmd/relay/main.go +++ b/cmd/relay/main.go @@ -11,9 +11,14 @@ import ( "context" + "connectrpc.com/connect" + "golang.org/x/net/http2" + "golang.org/x/net/http2/h2c" "google.golang.org/grpc" pb "northwest.io/nostr-grpc/api/nostr/v1" + "northwest.io/nostr-grpc/api/nostr/v1/nostrv1connect" + connecthandler "northwest.io/nostr-grpc/internal/handler/connect" grpchandler "northwest.io/nostr-grpc/internal/handler/grpc" wshandler "northwest.io/nostr-grpc/internal/handler/websocket" "northwest.io/nostr-grpc/internal/storage" @@ -39,7 +44,14 @@ func main() { grpcHandler := grpchandler.NewServer(store) grpcHandler.SetSubscriptionManager(subManager) + connectHandler := connecthandler.NewHandler(grpcHandler) + + mux := http.NewServeMux() + path, handler := nostrv1connect.NewNostrRelayHandler(connectHandler, connect.WithInterceptors()) + mux.Handle(path, handler) + wsHandler := wshandler.NewHandler(store, subManager) + mux.Handle("/", wsHandler) grpcLis, err := net.Listen("tcp", *grpcAddr) if err != nil { @@ -51,11 +63,13 @@ func main() { httpServer := &http.Server{ Addr: *wsAddr, - Handler: wsHandler, + Handler: h2c.NewHandler(mux, &http2.Server{}), } log.Printf("gRPC server listening on %s", *grpcAddr) - log.Printf("WebSocket server listening on %s", *wsAddr) + log.Printf("HTTP server listening on %s", *wsAddr) + log.Printf(" - Connect (gRPC-Web) at %s/nostr.v1.NostrRelay/*", *wsAddr) + log.Printf(" - WebSocket (Nostr) at %s/", *wsAddr) log.Printf("Database: %s", *dbPath) sigChan := make(chan os.Signal, 1) diff --git a/go.mod b/go.mod index 03667f7..a291cec 100644 --- a/go.mod +++ b/go.mod @@ -11,6 +11,7 @@ require ( ) require ( + connectrpc.com/connect v1.19.1 // indirect github.com/btcsuite/btcd/chaincfg/chainhash v1.0.1 // indirect github.com/decred/dcrd/crypto/blake256 v1.0.0 // indirect github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 // indirect @@ -20,9 +21,9 @@ require ( github.com/ncruces/go-strftime v1.0.0 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 // indirect - golang.org/x/net v0.48.0 // indirect - golang.org/x/sys v0.39.0 // indirect - golang.org/x/text v0.32.0 // indirect + golang.org/x/net v0.50.0 // indirect + golang.org/x/sys v0.41.0 // indirect + golang.org/x/text v0.34.0 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 // indirect modernc.org/libc v1.67.6 // indirect modernc.org/mathutil v1.7.1 // indirect diff --git a/go.sum b/go.sum index 6d5fa95..79c1ebe 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +connectrpc.com/connect v1.19.1 h1:R5M57z05+90EfEvCY1b7hBxDVOUl45PrtXtAV2fOC14= +connectrpc.com/connect v1.19.1/go.mod h1:tN20fjdGlewnSFeZxLKb0xwIZ6ozc3OQs2hTXy4du9w= github.com/btcsuite/btcd/btcec/v2 v2.3.2 h1:5n0X6hX0Zk+6omWcihdYvdAlGf2DfasC0GMf7DClJ3U= github.com/btcsuite/btcd/btcec/v2 v2.3.2/go.mod h1:zYzJ8etWJQIv1Ogk7OzpWjowwOdXY1W/17j2MW85J04= github.com/btcsuite/btcd/chaincfg/chainhash v1.0.1 h1:q0rUy8C/TYNBQS1+CGKw68tLOFYSNEs0TFnxxnS9+4U= @@ -50,17 +52,25 @@ golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 h1:mgKeJMpvi0yx/sU5GsxQ7p6s2 golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546/go.mod h1:j/pmGrbnkbPtQfxEe5D0VQhZC6qKbfKifgD0oM7sR70= golang.org/x/mod v0.30.0 h1:fDEXFVZ/fmCKProc/yAXXUijritrDzahmwwefnjoPFk= golang.org/x/mod v0.30.0/go.mod h1:lAsf5O2EvJeSFMiBxXDki7sCgAxEUcZHXoXMKT4GJKc= +golang.org/x/mod v0.32.0 h1:9F4d3PHLljb6x//jOyokMv3eX+YDeepZSEo3mFJy93c= golang.org/x/net v0.48.0 h1:zyQRTTrjc33Lhh0fBgT/H3oZq9WuvRR5gPC70xpDiQU= golang.org/x/net v0.48.0/go.mod h1:+ndRgGjkh8FGtu1w1FGbEC31if4VrNVMuKTgcAAnQRY= +golang.org/x/net v0.50.0 h1:ucWh9eiCGyDR3vtzso0WMQinm2Dnt8cFMuQa9K33J60= +golang.org/x/net v0.50.0/go.mod h1:UgoSli3F/pBgdJBHCTc+tp3gmrU4XswgGRgtnwWTfyM= golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.39.0 h1:CvCKL8MeisomCi6qNZ+wbb0DN9E5AATixKsvNtMoMFk= golang.org/x/sys v0.39.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k= +golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= golang.org/x/text v0.32.0 h1:ZD01bjUt1FQ9WJ0ClOL5vxgxOI/sVCNgX1YtKwcY0mU= golang.org/x/text v0.32.0/go.mod h1:o/rUWzghvpD5TXrTIBuJU77MTaN0ljMWE47kxGJQ7jY= +golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk= +golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA= golang.org/x/tools v0.39.0 h1:ik4ho21kwuQln40uelmciQPp9SipgNDdrafrYA4TmQQ= golang.org/x/tools v0.39.0/go.mod h1:JnefbkDPyD8UU2kI5fuf8ZX4/yUeh9W877ZeBONxUqQ= +golang.org/x/tools v0.41.0 h1:a9b8iMweWG+S0OBnlU36rzLp20z1Rp10w+IY2czHTQc= gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk= gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E= google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 h1:gRkg/vSppuSQoDjxyiGfN4Upv/h/DQmIR10ZU8dh4Ww= diff --git a/internal/handler/connect/handler.go b/internal/handler/connect/handler.go new file mode 100644 index 0000000..f33e4fc --- /dev/null +++ b/internal/handler/connect/handler.go @@ -0,0 +1,101 @@ +package connect + +import ( + "context" + + "connectrpc.com/connect" + "google.golang.org/grpc/metadata" + + pb "northwest.io/nostr-grpc/api/nostr/v1" + "northwest.io/nostr-grpc/api/nostr/v1/nostrv1connect" + grpchandler "northwest.io/nostr-grpc/internal/handler/grpc" +) + +type Handler struct { + grpcServer *grpchandler.Server +} + +func NewHandler(grpcServer *grpchandler.Server) *Handler { + return &Handler{grpcServer: grpcServer} +} + +func (h *Handler) PublishEvent(ctx context.Context, req *connect.Request[pb.PublishEventRequest]) (*connect.Response[pb.PublishEventResponse], error) { + resp, err := h.grpcServer.PublishEvent(ctx, req.Msg) + if err != nil { + return nil, err + } + return connect.NewResponse(resp), nil +} + +func (h *Handler) Subscribe(ctx context.Context, req *connect.Request[pb.SubscribeRequest], stream *connect.ServerStream[pb.Event]) error { + return h.grpcServer.Subscribe(req.Msg, &subscribeStreamAdapter{stream: stream, ctx: ctx}) +} + +func (h *Handler) Unsubscribe(ctx context.Context, req *connect.Request[pb.UnsubscribeRequest]) (*connect.Response[pb.Empty], error) { + resp, err := h.grpcServer.Unsubscribe(ctx, req.Msg) + if err != nil { + return nil, err + } + return connect.NewResponse(resp), nil +} + +func (h *Handler) PublishBatch(ctx context.Context, req *connect.Request[pb.PublishBatchRequest]) (*connect.Response[pb.PublishBatchResponse], error) { + resp, err := h.grpcServer.PublishBatch(ctx, req.Msg) + if err != nil { + return nil, err + } + return connect.NewResponse(resp), nil +} + +func (h *Handler) QueryEvents(ctx context.Context, req *connect.Request[pb.QueryRequest]) (*connect.Response[pb.QueryResponse], error) { + resp, err := h.grpcServer.QueryEvents(ctx, req.Msg) + if err != nil { + return nil, err + } + return connect.NewResponse(resp), nil +} + +func (h *Handler) CountEvents(ctx context.Context, req *connect.Request[pb.CountRequest]) (*connect.Response[pb.CountResponse], error) { + resp, err := h.grpcServer.CountEvents(ctx, req.Msg) + if err != nil { + return nil, err + } + return connect.NewResponse(resp), nil +} + +type subscribeStreamAdapter struct { + stream *connect.ServerStream[pb.Event] + ctx context.Context +} + +func (s *subscribeStreamAdapter) Send(event *pb.Event) error { + return s.stream.Send(event) +} + +func (s *subscribeStreamAdapter) Context() context.Context { + return s.ctx +} + +func (s *subscribeStreamAdapter) SetHeader(md metadata.MD) error { + return nil +} + +func (s *subscribeStreamAdapter) SendHeader(md metadata.MD) error { + return nil +} + +func (s *subscribeStreamAdapter) SetTrailer(md metadata.MD) { +} + +func (s *subscribeStreamAdapter) SendMsg(m any) error { + if event, ok := m.(*pb.Event); ok { + return s.Send(event) + } + return nil +} + +func (s *subscribeStreamAdapter) RecvMsg(m any) error { + return nil +} + +var _ nostrv1connect.NostrRelayHandler = (*Handler)(nil) -- cgit v1.2.3