feat: redirect bluesky events to int-bluesky

Author: akurilov

Makefile

@@ -9,7 +9,8 @@ proto:
 	go install github.com/golang/protobuf/protoc-gen-go@latest
 	go install google.golang.org/grpc/cmd/[email protected]
 	PATH=${PATH}:~/go/bin protoc --go_out=plugins=grpc:. --go_opt=paths=source_relative \
-		api/grpc/*.proto
+		api/grpc/*.proto \
+		api/grpc/events/*.proto
 
 vet: proto
 	go vet

api/grpc/events/client_mock.go

@@ -0,0 +1,31 @@
+package events
+
+import (
+	"context"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+type clientMock struct {
+}
+
+func NewClientMock() ServiceClient {
+	return clientMock{}
+}
+
+func (cm clientMock) SetStream(ctx context.Context, req *SetStreamRequest, opts ...grpc.CallOption) (resp *SetStreamResponse, err error) {
+	switch req.Topic {
+	case "":
+		err = status.Error(codes.InvalidArgument, "empty topic")
+	case "fail":
+		err = status.Error(codes.Internal, "internal failure")
+	default:
+		resp = &SetStreamResponse{}
+	}
+	return
+}
+
+func (cm clientMock) Publish(ctx context.Context, opts ...grpc.CallOption) (Service_PublishClient, error) {
+	return newPublishStreamMock(), nil
+}

api/grpc/events/client_pool.go

@@ -0,0 +1,47 @@
+package events
+
+import (
+	"context"
+	grpcpool "github.com/processout/grpc-go-pool"
+	"google.golang.org/grpc"
+)
+
+type clientPool struct {
+	connPool *grpcpool.Pool
+}
+
+func NewClientPool(connPool *grpcpool.Pool) ServiceClient {
+	return clientPool{
+		connPool: connPool,
+	}
+}
+
+func (cp clientPool) SetStream(ctx context.Context, req *SetStreamRequest, opts ...grpc.CallOption) (resp *SetStreamResponse, err error) {
+	var conn *grpcpool.ClientConn
+	conn, err = cp.connPool.Get(ctx)
+	if err == nil {
+		defer conn.Close()
+	}
+	var client ServiceClient
+	if err == nil {
+		client = NewServiceClient(conn)
+		resp, err = client.SetStream(ctx, req, opts...)
+	}
+	return
+}
+
+func (cp clientPool) Publish(ctx context.Context, opts ...grpc.CallOption) (stream Service_PublishClient, err error) {
+	var conn *grpcpool.ClientConn
+	conn, err = cp.connPool.Get(ctx)
+	var c *grpc.ClientConn
+	if err == nil {
+		c = conn.ClientConn
+		conn.Close() // return back to the conn pool immediately
+	}
+	var client ServiceClient
+	if err == nil {
+		client = NewServiceClient(c)
+		stream, err = client.Publish(ctx, opts...)
+	}
+	return
+}

api/grpc/events/events_writer.go

@@ -0,0 +1,15 @@
+package events
+
+import (
+	"context"
+	"github.com/cloudevents/sdk-go/binding/format/protobuf/v2/pb"
+	"io"
+)
+
+type Writer interface {
+	io.Closer
+
+	// Write writes the specified messages and returns the accepted count preserving the order.
+	// Returns io.EOF if the destination file/connection/whatever is closed.
+	Write(ctx context.Context, msgs []*pb.CloudEvent) (ackCount uint32, err error)
+}

api/grpc/events/logging.go

@@ -0,0 +1,31 @@
+package events
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+)
+
+type loggingMiddleware struct {
+	svc Service
+	log *slog.Logger
+}
+
+func NewLoggingMiddleware(svc Service, log *slog.Logger) Service {
+	return loggingMiddleware{
+		svc: svc,
+		log: log,
+	}
+}
+
+func (lm loggingMiddleware) SetStream(ctx context.Context, topic string, limit uint32) (err error) {
+	err = lm.svc.SetStream(ctx, topic, limit)
+	lm.log.Debug(fmt.Sprintf("events.SetStream(topic=%s, limit=%d): err=%s", topic, limit, err))
+	return
+}
+
+func (lm loggingMiddleware) NewPublisher(ctx context.Context, topic string) (p Writer, err error) {
+	p, err = lm.svc.NewPublisher(ctx, topic)
+	lm.log.Debug(fmt.Sprintf("events.Publish(topic=%s): err=%s", topic, err))
+	return
+}

api/grpc/events/publish_stream_mock.go

@@ -0,0 +1,83 @@
+package events
+
+import (
+	"context"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+	"io"
+)
+
+type publishStreamMock struct {
+	lastReq *PublishRequest
+	lastErr error
+}
+
+func newPublishStreamMock() Service_PublishClient {
+	return &publishStreamMock{}
+}
+
+func (sm *publishStreamMock) Send(req *PublishRequest) (err error) {
+	switch sm.lastErr {
+	case nil:
+		switch req.Topic {
+		case "recv_fail":
+			sm.lastErr = status.Error(codes.Internal, "send failure")
+		case "send_eof":
+			sm.lastErr = io.EOF
+			err = io.EOF
+		case "recv_eof":
+			sm.lastErr = io.EOF
+		case "missing":
+			sm.lastErr = status.Error(codes.NotFound, "queue missing")
+		default:
+			sm.lastReq = req
+		}
+	default:
+		err = io.EOF
+	}
+	return
+}
+
+func (sm *publishStreamMock) Recv() (resp *PublishResponse, err error) {
+	resp = &PublishResponse{}
+	switch sm.lastErr {
+	case nil:
+		resp.AckCount = uint32(len(sm.lastReq.Evts))
+		if resp.AckCount > 2 {
+			resp.AckCount = 2
+		}
+	default:
+		err = sm.lastErr
+	}
+	return
+}
+
+func (sm *publishStreamMock) Header() (metadata.MD, error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (sm *publishStreamMock) Trailer() metadata.MD {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (sm *publishStreamMock) CloseSend() error {
+	return nil
+}
+
+func (sm *publishStreamMock) Context() context.Context {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (sm *publishStreamMock) SendMsg(m interface{}) error {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (sm *publishStreamMock) RecvMsg(m interface{}) error {
+	//TODO implement me
+	panic("implement me")
+}

api/grpc/events/publisher.go

@@ -0,0 +1,46 @@
+package events
+
+import (
+	"context"
+	"github.com/cloudevents/sdk-go/binding/format/protobuf/v2/pb"
+	"io"
+)
+
+type publisher struct {
+	stream Service_PublishClient
+	queue  string
+}
+
+func newPublisher(stream Service_PublishClient, queue string) Writer {
+	return publisher{
+		stream: stream,
+		queue:  queue,
+	}
+}
+
+func (mw publisher) Close() (err error) {
+	err = mw.stream.CloseSend()
+	if err != nil {
+		err = decodeError(err)
+	}
+	return
+}
+
+func (mw publisher) Write(ctx context.Context, msgs []*pb.CloudEvent) (ackCount uint32, err error) {
+	req := PublishRequest{
+		Topic: mw.queue,
+		Evts:  msgs,
+	}
+	err = mw.stream.Send(&req)
+	var resp *PublishResponse
+	if err == nil || err == io.EOF {
+		resp, err = mw.stream.Recv()
+	}
+	if err != nil {
+		err = decodeError(err)
+	}
+	if resp != nil {
+		ackCount = resp.AckCount
+	}
+	return
+}

api/grpc/events/publisher_mock.go

@@ -0,0 +1,31 @@
+package events
+
+import (
+	"context"
+	"github.com/cloudevents/sdk-go/binding/format/protobuf/v2/pb"
+)
+
+type publisherMock struct{}
+
+func NewPublisherMock() Writer {
+	return publisherMock{}
+}
+
+func (mw publisherMock) Close() error {
+	return nil
+}
+
+func (mw publisherMock) Write(ctx context.Context, msgs []*pb.CloudEvent) (ackCount uint32, err error) {
+	for _, msg := range msgs {
+		switch msg.Id {
+		case "queue_fail":
+			err = ErrInternal
+		default:
+			ackCount++
+		}
+		if err != nil {
+			break
+		}
+	}
+	return
+}

api/grpc/events/publisher_test.go

@@ -0,0 +1,84 @@
+package events
+
+import (
+	"context"
+	"github.com/cloudevents/sdk-go/binding/format/protobuf/v2/pb"
+	"github.com/stretchr/testify/assert"
+	"io"
+	"testing"
+)
+
+func TestMessagesWriter_Close(t *testing.T) {
+	mw := newPublisher(newPublishStreamMock(), "queue0")
+	err := mw.Close()
+	assert.Nil(t, err)
+}
+
+func TestMessagesWriter_Write(t *testing.T) {
+	cases := map[string]struct {
+		queue    string
+		msgs     []*pb.CloudEvent
+		ackCount uint32
+		err      error
+	}{
+		"1 => ack 1": {
+			queue: "queue0",
+			msgs: []*pb.CloudEvent{
+				{
+					Id: "msg0",
+				},
+			},
+			ackCount: 1,
+		},
+		"3 => ack 2": {
+			queue: "queue0",
+			msgs: []*pb.CloudEvent{
+				{
+					Id: "msg0",
+				},
+				{
+					Id: "msg1",
+				},
+				{
+					Id: "msg2",
+				},
+			},
+			ackCount: 2,
+		},
+		"send eof": {
+			queue: "send_eof",
+			msgs: []*pb.CloudEvent{
+				{
+					Id: "msg0",
+				},
+			},
+			err: io.EOF,
+		},
+		"recv fail": {
+			queue: "recv_fail",
+			msgs: []*pb.CloudEvent{
+				{
+					Id: "msg0",
+				},
+			},
+			err: ErrInternal,
+		},
+		"recv eof": {
+			queue: "recv_eof",
+			msgs: []*pb.CloudEvent{
+				{
+					Id: "msg0",
+				},
+			},
+			err: io.EOF,
+		},
+	}
+	for k, c := range cases {
+		t.Run(k, func(t *testing.T) {
+			mw := newPublisher(newPublishStreamMock(), c.queue)
+			ackCount, err := mw.Write(context.TODO(), c.msgs)
+			assert.Equal(t, c.ackCount, ackCount)
+			assert.ErrorIs(t, err, c.err)
+		})
+	}
+}