Add subscriptions store cache

cmd/config.go

@@ -18,7 +18,7 @@ import (
 	"github.com/xataio/pgstream/pkg/wal/processor/search/opensearch"
 	"github.com/xataio/pgstream/pkg/wal/processor/translator"
 	"github.com/xataio/pgstream/pkg/wal/processor/webhook/notifier"
-	"github.com/xataio/pgstream/pkg/wal/processor/webhook/server"
+	"github.com/xataio/pgstream/pkg/wal/processor/webhook/subscription/server"
 	pgreplication "github.com/xataio/pgstream/pkg/wal/replication/postgres"
 )
 
@@ -181,7 +181,11 @@ func parseWebhookProcessorConfig() *stream.WebhookProcessorConfig {
 	}
 
 	return &stream.WebhookProcessorConfig{
-		SubscriptionStoreURL: subscriptionStore,
+		SubscriptionStore: stream.WebhookSubscriptionStoreConfig{
+			URL:                  subscriptionStore,
+			CacheEnabled:         viper.GetBool("PGSTREAM_WEBHOOK_SUBSCRIPTION_STORE_CACHE_ENABLED"),
+			CacheRefreshInterval: viper.GetDuration("PGSTREAM_WEBHOOK_SUBSCRIPTION_STORE_CACHE_REFRESH_INTERVAL"),
+		},
 		Notifier: notifier.Config{
 			MaxQueueBytes:  viper.GetInt64("PGSTREAM_WEBHOOK_NOTIFIER_MAX_QUEUE_BYTES"),
 			URLWorkerCount: viper.GetUint("PGSTREAM_WEBHOOK_NOTIFIER_WORKER_COUNT"),

pg2webhook.env

@@ -3,3 +3,5 @@ PGSTREAM_POSTGRES_LISTENER_URL="postgres://postgres:postgres@localhost?sslmode=d
 
 # Processor config
 PGSTREAM_WEBHOOK_SUBSCRIPTION_STORE_URL="postgres://postgres:postgres@localhost?sslmode=disable"
+PGSTREAM_WEBHOOK_SUBSCRIPTION_STORE_CACHE_ENABLED=true
+PGSTREAM_WEBHOOK_SUBSCRIPTION_STORE_CACHE_REFRESH_INTERVAL="60s"

pkg/stream/config.go

@@ -4,6 +4,7 @@ package stream
 
 import (
 	"errors"
+	"time"
 
 	kafkacheckpoint "github.com/xataio/pgstream/pkg/wal/checkpointer/kafka"
 	kafkalistener "github.com/xataio/pgstream/pkg/wal/listener/kafka"
@@ -12,7 +13,7 @@ import (
 	"github.com/xataio/pgstream/pkg/wal/processor/search/opensearch"
 	"github.com/xataio/pgstream/pkg/wal/processor/translator"
 	"github.com/xataio/pgstream/pkg/wal/processor/webhook/notifier"
-	"github.com/xataio/pgstream/pkg/wal/processor/webhook/server"
+	"github.com/xataio/pgstream/pkg/wal/processor/webhook/subscription/server"
 	pgreplication "github.com/xataio/pgstream/pkg/wal/replication/postgres"
 )
 
@@ -53,9 +54,15 @@ type SearchProcessorConfig struct {
 }
 
 type WebhookProcessorConfig struct {
-	SubscriptionStoreURL string
-	Notifier             notifier.Config
-	SubscriptionServer   server.Config
+	Notifier           notifier.Config
+	SubscriptionServer server.Config
+	SubscriptionStore  WebhookSubscriptionStoreConfig
+}
+
+type WebhookSubscriptionStoreConfig struct {
+	URL                  string
+	CacheEnabled         bool
+	CacheRefreshInterval time.Duration
 }
 
 func (c *Config) IsValid() error {

pkg/stream/stream_start.go

@@ -19,10 +19,12 @@ import (
 	"github.com/xataio/pgstream/pkg/wal/processor/search"
 	"github.com/xataio/pgstream/pkg/wal/processor/search/opensearch"
 	"github.com/xataio/pgstream/pkg/wal/processor/translator"
-	"github.com/xataio/pgstream/pkg/wal/processor/webhook"
 	webhooknotifier "github.com/xataio/pgstream/pkg/wal/processor/webhook/notifier"
-	pgwebhook "github.com/xataio/pgstream/pkg/wal/processor/webhook/postgres"
-	webhookserver "github.com/xataio/pgstream/pkg/wal/processor/webhook/server"
+	subscriptionserver "github.com/xataio/pgstream/pkg/wal/processor/webhook/subscription/server"
+	webhookstore "github.com/xataio/pgstream/pkg/wal/processor/webhook/subscription/store"
+	subscriptionstorecache "github.com/xataio/pgstream/pkg/wal/processor/webhook/subscription/store/cache"
+	pgwebhook "github.com/xataio/pgstream/pkg/wal/processor/webhook/subscription/store/postgres"
+
 	"github.com/xataio/pgstream/pkg/wal/replication"
 	replicationinstrumentation "github.com/xataio/pgstream/pkg/wal/replication/instrumentation"
 	pgreplication "github.com/xataio/pgstream/pkg/wal/replication/postgres"
@@ -133,15 +135,28 @@ func Start(ctx context.Context, logger loglib.Logger, config *Config, meter metr
 		})
 
 	case config.Processor.Webhook != nil:
-		var subscriptionStore webhook.SubscriptionStore
+		var subscriptionStore webhookstore.Store
 		var err error
 		subscriptionStore, err = pgwebhook.NewSubscriptionStore(ctx,
-			config.Processor.Webhook.SubscriptionStoreURL,
+			config.Processor.Webhook.SubscriptionStore.URL,
 			pgwebhook.WithLogger(logger),
 		)
 		if err != nil {
 			return err
 		}
+
+		if config.Processor.Webhook.SubscriptionStore.CacheEnabled {
+			logger.Info("setting up subscription store cache...")
+			subscriptionStore, err = subscriptionstorecache.New(ctx, subscriptionStore,
+				&subscriptionstorecache.Config{
+					SyncInterval: config.Processor.Webhook.SubscriptionStore.CacheRefreshInterval,
+				},
+				subscriptionstorecache.WithLogger(logger))
+			if err != nil {
+				return err
+			}
+		}
+
 		notifier := webhooknotifier.New(
 			&config.Processor.Webhook.Notifier,
 			subscriptionStore,
@@ -150,10 +165,10 @@ func Start(ctx context.Context, logger loglib.Logger, config *Config, meter metr
 		defer notifier.Close()
 		processor = notifier
 
-		subscriptionServer := webhookserver.New(
+		subscriptionServer := subscriptionserver.New(
 			&config.Processor.Webhook.SubscriptionServer,
 			subscriptionStore,
-			webhookserver.WithLogger(logger))
+			subscriptionserver.WithLogger(logger))
 
 		eg.Go(func() error {
 			logger.Info("running subscription server...")

pkg/wal/processor/webhook/notifier/helper_test.go

@@ -6,15 +6,16 @@ import (
 	"errors"
 
 	"github.com/xataio/pgstream/pkg/wal"
-	"github.com/xataio/pgstream/pkg/wal/processor/webhook"
+	"github.com/xataio/pgstream/pkg/wal/processor/webhook/subscription"
 )
 
-var testCommitPos = wal.CommitPosition("test-pos")
-
-var errTest = errors.New("oh noes")
+var (
+	testCommitPos = wal.CommitPosition("test-pos")
+	errTest       = errors.New("oh noes")
+)
 
-func newTestSubscription(url, schema, table string, eventTypes []string) *webhook.Subscription {
-	return &webhook.Subscription{
+func newTestSubscription(url, schema, table string, eventTypes []string) *subscription.Subscription {
+	return &subscription.Subscription{
 		URL:        url,
 		Schema:     schema,
 		Table:      table,

pkg/wal/processor/webhook/notifier/webhook_notifier.go

@@ -18,7 +18,7 @@ import (
 	"github.com/xataio/pgstream/pkg/wal"
 	"github.com/xataio/pgstream/pkg/wal/checkpointer"
 	"github.com/xataio/pgstream/pkg/wal/processor"
-	"github.com/xataio/pgstream/pkg/wal/processor/webhook"
+	"github.com/xataio/pgstream/pkg/wal/processor/webhook/subscription"
 )
 
 // Notifier represents the process that notifies any subscribed webhooks when
@@ -38,7 +38,7 @@ type Notifier struct {
 }
 
 type subscriptionRetriever interface {
-	GetSubscriptions(ctx context.Context, action, schema, table string) ([]*webhook.Subscription, error)
+	GetSubscriptions(ctx context.Context, action, schema, table string) ([]*subscription.Subscription, error)
 }
 
 type Option func(*Notifier)
@@ -92,7 +92,7 @@ func (n *Notifier) ProcessWALEvent(ctx context.Context, walEvent *wal.Event) (er
 		}
 	}()
 
-	subscriptions := []*webhook.Subscription{}
+	subscriptions := []*subscription.Subscription{}
 	if walEvent.Data != nil {
 		data := walEvent.Data
 		subscriptions, err = n.subscriptionStore.GetSubscriptions(ctx, data.Action, data.Schema, data.Table)

pkg/wal/processor/webhook/notifier/webhook_notifier_test.go

@@ -20,7 +20,8 @@ import (
 	"github.com/xataio/pgstream/pkg/wal/checkpointer"
 	"github.com/xataio/pgstream/pkg/wal/processor"
 	"github.com/xataio/pgstream/pkg/wal/processor/webhook"
-	"github.com/xataio/pgstream/pkg/wal/processor/webhook/mocks"
+	"github.com/xataio/pgstream/pkg/wal/processor/webhook/subscription"
+	"github.com/xataio/pgstream/pkg/wal/processor/webhook/subscription/store/mocks"
 )
 
 func TestNotifier_ProcessWALEvent(t *testing.T) {
@@ -35,7 +36,7 @@ func TestNotifier_ProcessWALEvent(t *testing.T) {
 		CommitPosition: testCommitPos,
 	}
 
-	testSubscription := func(url string) *webhook.Subscription {
+	testSubscription := func(url string) *subscription.Subscription {
 		return newTestSubscription(url, "", "", nil)
 	}
 
@@ -54,9 +55,9 @@ func TestNotifier_ProcessWALEvent(t *testing.T) {
 	}{
 		{
 			name: "ok - no subscriptions for event",
-			store: &mocks.SubscriptionStore{
-				GetSubscriptionsFn: func(ctx context.Context, action, schema, table string) ([]*webhook.Subscription, error) {
-					return []*webhook.Subscription{}, nil
+			store: &mocks.Store{
+				GetSubscriptionsFn: func(ctx context.Context, action, schema, table string) ([]*subscription.Subscription, error) {
+					return []*subscription.Subscription{}, nil
 				},
 			},
 			weightedSemaphore: &syncmocks.WeightedSemaphore{
@@ -72,9 +73,9 @@ func TestNotifier_ProcessWALEvent(t *testing.T) {
 		},
 		{
 			name: "ok - subscriptions for event",
-			store: &mocks.SubscriptionStore{
-				GetSubscriptionsFn: func(ctx context.Context, action, schema, table string) ([]*webhook.Subscription, error) {
-					return []*webhook.Subscription{
+			store: &mocks.Store{
+				GetSubscriptionsFn: func(ctx context.Context, action, schema, table string) ([]*subscription.Subscription, error) {
+					return []*subscription.Subscription{
 						testSubscription("url-1"), testSubscription("url-2"),
 					}, nil
 				},
@@ -94,8 +95,8 @@ func TestNotifier_ProcessWALEvent(t *testing.T) {
 		},
 		{
 			name: "error - getting subscriptions",
-			store: &mocks.SubscriptionStore{
-				GetSubscriptionsFn: func(ctx context.Context, action, schema, table string) ([]*webhook.Subscription, error) {
+			store: &mocks.Store{
+				GetSubscriptionsFn: func(ctx context.Context, action, schema, table string) ([]*subscription.Subscription, error) {
 					return nil, errTest
 				},
 			},
@@ -106,9 +107,9 @@ func TestNotifier_ProcessWALEvent(t *testing.T) {
 		},
 		{
 			name: "error - serialising payload",
-			store: &mocks.SubscriptionStore{
-				GetSubscriptionsFn: func(ctx context.Context, action, schema, table string) ([]*webhook.Subscription, error) {
-					return []*webhook.Subscription{
+			store: &mocks.Store{
+				GetSubscriptionsFn: func(ctx context.Context, action, schema, table string) ([]*subscription.Subscription, error) {
+					return []*subscription.Subscription{
 						testSubscription("url-1"), testSubscription("url-2"),
 					}, nil
 				},
@@ -121,9 +122,9 @@ func TestNotifier_ProcessWALEvent(t *testing.T) {
 		},
 		{
 			name: "error - acquiring semaphore",
-			store: &mocks.SubscriptionStore{
-				GetSubscriptionsFn: func(ctx context.Context, action, schema, table string) ([]*webhook.Subscription, error) {
-					return []*webhook.Subscription{
+			store: &mocks.Store{
+				GetSubscriptionsFn: func(ctx context.Context, action, schema, table string) ([]*subscription.Subscription, error) {
+					return []*subscription.Subscription{
 						testSubscription("url-1"), testSubscription("url-2"),
 					}, nil
 				},
@@ -139,8 +140,8 @@ func TestNotifier_ProcessWALEvent(t *testing.T) {
 		},
 		{
 			name: "error - panic recovery",
-			store: &mocks.SubscriptionStore{
-				GetSubscriptionsFn: func(ctx context.Context, action, schema, table string) ([]*webhook.Subscription, error) {
+			store: &mocks.Store{
+				GetSubscriptionsFn: func(ctx context.Context, action, schema, table string) ([]*subscription.Subscription, error) {
 					panic(errTest)
 				},
 			},
@@ -299,7 +300,7 @@ func TestNotifier_Notify(t *testing.T) {
 			doneChan := make(chan struct{}, 1)
 			defer close(doneChan)
 
-			n := New(testCfg, &mocks.SubscriptionStore{})
+			n := New(testCfg, &mocks.Store{})
 			n.client = tc.client
 			n.queueBytesSema = tc.semaphore
 			n.checkpointer = tc.checkpointer(doneChan)

pkg/wal/processor/webhook/notifier/webhook_notify_msg.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/xataio/pgstream/pkg/wal"
 	"github.com/xataio/pgstream/pkg/wal/processor/webhook"
+	"github.com/xataio/pgstream/pkg/wal/processor/webhook/subscription"
 )
 
 type notifyMsg struct {
@@ -17,7 +18,7 @@ type notifyMsg struct {
 
 type serialiser func(any) ([]byte, error)
 
-func newNotifyMsg(event *wal.Event, subscriptions []*webhook.Subscription, serialiser serialiser) (*notifyMsg, error) {
+func newNotifyMsg(event *wal.Event, subscriptions []*subscription.Subscription, serialiser serialiser) (*notifyMsg, error) {
 	var payload []byte
 	urls := make([]string, 0, len(subscriptions))
 	if len(subscriptions) > 0 {