From a65f74ad95b43e41e9948fc8c009bdcb570caac8 Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Thu, 16 Jun 2022 15:53:19 -0400
Subject: [PATCH] updates

---
 config/config.go                    |  4 ++++
 config/toml.go                      |  3 +++
 node/node.go                        |  5 ++++-
 state/indexer/sink/pubsub/pubsub.go | 23 ++++++++++++++++++++++-
 4 files changed, 33 insertions(+), 2 deletions(-)

diff --git a/config/config.go b/config/config.go
index 7dc58c030..bfa8ff650 100644
--- a/config/config.go
+++ b/config/config.go
@@ -1069,6 +1069,10 @@ type TxIndexConfig struct {
 	// must ensure the GOOGLE_APPLICATION_CREDENTIALS environment variable is set
 	// to the location of their creds file.
 	PubsubProjectID string `mapstructure:"pubsub-project-id"`
+
+	// PubsubTopic defines the Google Cloud Pubsub topic. If the topic does not
+	// exist, it will be created.
+	PubsubTopic string `mapstructure:"pubsub-topic"`
 }
 
 // DefaultTxIndexConfig returns a default configuration for the transaction indexer.
diff --git a/config/toml.go b/config/toml.go
index e4aca674e..86fa83ea6 100644
--- a/config/toml.go
+++ b/config/toml.go
@@ -487,6 +487,9 @@ psql-conn = "{{ .TxIndex.PsqlConn }}"
 # their creds file.
 pubsub-project-id = "{{ .TxIndex.PubsubProjectID }}"
 
+# The Google Cloud Pubsub topic. If the topic does not exist, it will be created.
+pubsub-topic = "{{ .TxIndex.PubsubTopic }}"
+
 #######################################################
 ###       Instrumentation Configuration Options     ###
 #######################################################
diff --git a/node/node.go b/node/node.go
index 086a7a133..5806fc8c8 100644
--- a/node/node.go
+++ b/node/node.go
@@ -302,8 +302,11 @@ func createAndStartIndexerService(
 		if config.TxIndex.PubsubProjectID == "" {
 			return nil, nil, nil, errors.New("no 'pubsub-project-id' is set for the 'pubsub' indexer")
 		}
+		if config.TxIndex.PubsubTopic == "" {
+			return nil, nil, nil, errors.New("no 'pubsub-topic' is set for the 'pubsub' indexer")
+		}
 
-		sink, err := pubsub.NewEventSink(config.TxIndex.PubsubProjectID, chainID)
+		sink, err := pubsub.NewEventSink(config.TxIndex.PubsubProjectID, config.TxIndex.PubsubTopic, chainID)
 		if err != nil {
 			return nil, nil, nil, fmt.Errorf("creating pubsub indexer: %w", err)
 		}
diff --git a/state/indexer/sink/pubsub/pubsub.go b/state/indexer/sink/pubsub/pubsub.go
index 386acb38c..1e8ba6f01 100644
--- a/state/indexer/sink/pubsub/pubsub.go
+++ b/state/indexer/sink/pubsub/pubsub.go
@@ -22,10 +22,11 @@ const (
 
 type EventSink struct {
 	client  *pubsub.Client
+	topic   *pubsub.Topic
 	chainID string
 }
 
-func NewEventSink(projectID, chainID string) (*EventSink, error) {
+func NewEventSink(projectID, topic, chainID string) (*EventSink, error) {
 	if s := os.Getenv(credsEnvVar); len(s) == 0 {
 		return nil, fmt.Errorf("missing '%s' environment variable", credsEnvVar)
 	}
@@ -38,8 +39,28 @@ func NewEventSink(projectID, chainID string) (*EventSink, error) {
 		return nil, fmt.Errorf("failed to create a Google Cloud Pubsub client: %w", err)
 	}
 
+	// attempt to get the topic. If that fails, we attempt to create it
+	ctx, cancel = context.WithTimeout(context.Background(), 15*time.Second)
+	defer cancel()
+
+	t := c.Topic(topic)
+	ok, err := t.Exists(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to check for topic '%s': %w", topic, err)
+	}
+	if !ok {
+		ctx, cancel = context.WithTimeout(context.Background(), 15*time.Second)
+		defer cancel()
+
+		t, err = c.CreateTopic(ctx, topic)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create topic '%s': %w", topic, err)
+		}
+	}
+
 	return &EventSink{
 		client:  c,
+		topic:   t,
 		chainID: chainID,
 	}, nil
 }