Repository: xataio/pgstream
Pull request #62 (2024-08-21 09:18:32)

Create webhook subscriptions table within the pgstream schema


Patch:
diff --git a/pkg/wal/processor/webhook/subscription/store/postgres/pg_subscription_store.go b/pkg/wal/processor/webhook/subscription/store/postgres/pg_subscription_store.go
index 363caf4..b6338da 100644
--- a/pkg/wal/processor/webhook/subscription/store/postgres/pg_subscription_store.go
+++ b/pkg/wal/processor/webhook/subscription/store/postgres/pg_subscription_store.go
@@ -18,7 +18,10 @@ type Store struct {
 
 type Option func(*Store)
 
-const subscriptionsTable = "webhook_subscriptions"
+const (
+	subscriptionsTableName = "webhook_subscriptions"
+	pgstreamSchema         = "pgstream"
+)
 
 func NewSubscriptionStore(ctx context.Context, url string, opts ...Option) (*Store, error) {
 	pgpool, err := pglib.NewConnPool(ctx, url)
@@ -52,13 +55,13 @@ func WithLogger(l loglib.Logger) Option {
 func (s *Store) CreateSubscription(ctx context.Context, subscription *subscription.Subscription) error {
 	query := fmt.Sprintf(`
 	INSERT INTO %s(url, schema_name, table_name, event_types) VALUES($1, $2, $3, $4)
-	ON CONFLICT (url,schema_name,table_name) DO UPDATE SET event_types = EXCLUDED.event_types;`, subscriptionsTable)
+	ON CONFLICT (url,schema_name,table_name) DO UPDATE SET event_types = EXCLUDED.event_types;`, subscriptionsTable())
 	_, err := s.conn.Exec(ctx, query, subscription.URL, subscription.Schema, subscription.Table, subscription.EventTypes)
 	return err
 }
 
 func (s *Store) DeleteSubscription(ctx context.Context, subscription *subscription.Subscription) error {
-	query := fmt.Sprintf(`DELETE FROM %s WHERE url=$1 AND schema_name=$2 AND table=$3;`, subscriptionsTable)
+	query := fmt.Sprintf(`DELETE FROM %s WHERE url=$1 AND schema_name=$2 AND table=$3;`, subscriptionsTable())
 	_, err := s.conn.Exec(ctx, query, subscription.URL, subscription.Schema, subscription.Table)
 	return err
 }
@@ -94,13 +97,13 @@ func (s *Store) createTable(ctx context.Context) error {
 	schema_name TEXT,
 	table_name TEXT,
 	event_types TEXT[],
-	PRIMARY KEY(url,schema_name,table_name))`, subscriptionsTable)
+	PRIMARY KEY(url,schema_name,table_name))`, subscriptionsTable())
 	_, err := s.conn.Exec(ctx, query)
 	return err
 }
 
 func (s *Store) buildGetQuery(action, schema, table string) (string, []any) {
-	query := fmt.Sprintf(`SELECT url, schema_name, table_name, event_types FROM %s`, subscriptionsTable)
+	query := fmt.Sprintf(`SELECT url, schema_name, table_name, event_types FROM %s`, subscriptionsTable())
 
 	separator := func(params []any) string {
 		if len(params) == 0 {
@@ -124,3 +127,7 @@ func (s *Store) buildGetQuery(action, schema, table string) (string, []any) {
 
 	return fmt.Sprintf("%s LIMIT 1000", query), params
 }
+
+func subscriptionsTable() string {
+	return fmt.Sprintf("%s.%s", pgstreamSchema, subscriptionsTableName)
+}


Test patch:
diff --git a/pkg/wal/processor/webhook/subscription/store/postgres/pg_subscription_store_test.go b/pkg/wal/processor/webhook/subscription/store/postgres/pg_subscription_store_test.go
index 3dc355f..63bf1d9 100644
--- a/pkg/wal/processor/webhook/subscription/store/postgres/pg_subscription_store_test.go
+++ b/pkg/wal/processor/webhook/subscription/store/postgres/pg_subscription_store_test.go
@@ -23,25 +23,25 @@ func TestStore_buildGetQuery(t *testing.T) {
 	}{
 		{
 			name:       "no filters",
-			wantQuery:  fmt.Sprintf(`SELECT url, schema_name, table_name, event_types FROM %s LIMIT 1000`, subscriptionsTable),
+			wantQuery:  fmt.Sprintf(`SELECT url, schema_name, table_name, event_types FROM %s LIMIT 1000`, subscriptionsTable()),
 			wantParams: nil,
 		},
 		{
 			name:       "with action filter",
 			action:     "I",
-			wantQuery:  fmt.Sprintf(`SELECT url, schema_name, table_name, event_types FROM %s WHERE ($1=ANY(event_types) OR event_types IS NULL) LIMIT 1000`, subscriptionsTable),
+			wantQuery:  fmt.Sprintf(`SELECT url, schema_name, table_name, event_types FROM %s WHERE ($1=ANY(event_types) OR event_types IS NULL) LIMIT 1000`, subscriptionsTable()),
 			wantParams: []any{"I"},
 		},
 		{
 			name:       "with schema filter",
 			schema:     "test_schema",
-			wantQuery:  fmt.Sprintf(`SELECT url, schema_name, table_name, event_types FROM %s WHERE (schema_name=$1 OR schema_name='') LIMIT 1000`, subscriptionsTable),
+			wantQuery:  fmt.Sprintf(`SELECT url, schema_name, table_name, event_types FROM %s WHERE (schema_name=$1 OR schema_name='') LIMIT 1000`, subscriptionsTable()),
 			wantParams: []any{"test_schema"},
 		},
 		{
 			name:       "with table filter",
 			table:      "test_table",
-			wantQuery:  fmt.Sprintf(`SELECT url, schema_name, table_name, event_types FROM %s WHERE (table_name=$1 OR table_name='') LIMIT 1000`, subscriptionsTable),
+			wantQuery:  fmt.Sprintf(`SELECT url, schema_name, table_name, event_types FROM %s WHERE (table_name=$1 OR table_name='') LIMIT 1000`, subscriptionsTable()),
 			wantParams: []any{"test_table"},
 		},
 		{
@@ -49,7 +49,7 @@ func TestStore_buildGetQuery(t *testing.T) {
 			action: "I",
 			schema: "test_schema",
 			table:  "test_table",
-			wantQuery: fmt.Sprintf(`SELECT url, schema_name, table_name, event_types FROM %s `, subscriptionsTable) +
+			wantQuery: fmt.Sprintf(`SELECT url, schema_name, table_name, event_types FROM %s `, subscriptionsTable()) +
 				"WHERE (schema_name=$1 OR schema_name='') " +
 				"AND (table_name=$2 OR table_name='') " +
 				"AND ($3=ANY(event_types) OR event_types IS NULL) LIMIT 1000",