updates

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.

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     ###
 #######################################################

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)
 		}

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
 }