From f99426e9bc74e738193d62bc291f41470c09e27d Mon Sep 17 00:00:00 2001
From: Vladyslav Len <heavenly.beta@gmail.com>
Date: Wed, 8 May 2024 23:56:44 +0200
Subject: [PATCH] fix(): small update for AWS Secret Manager

---
 config/config.go                                    | 2 +-
 config/config_loader.go                             | 5 +++--
 config/config_loader_test.go                        | 2 +-
 internal/secret/secrets.go                          | 2 +-
 internal/secret/secretstorage/aws_secret_manager.go | 9 ++++++---
 public/stream/stream.go                             | 7 +++++--
 6 files changed, 17 insertions(+), 10 deletions(-)

diff --git a/config/config.go b/config/config.go
index 22dae3c..38237df 100644
--- a/config/config.go
+++ b/config/config.go
@@ -39,7 +39,7 @@ type ETCD struct {
 }
 
 type Secrets struct {
-	StorageType secret.SecretsStorageType `yaml:"storage_type"`
+	StorageType secret.SecretsStorageType `yaml:"storage_type" validate:"required"`
 	Config      interface{}               `yaml:"config"`
 }
 
diff --git a/config/config_loader.go b/config/config_loader.go
index 1f52e83..506288e 100644
--- a/config/config_loader.go
+++ b/config/config_loader.go
@@ -74,11 +74,12 @@ func resolveSecrets(config []byte) []byte {
 	}
 
 	secretStorage := secret.NewSecrets(conf.Secrets.StorageType, conf.Secrets.Config)
-
 	replaced := secretsRegex.ReplaceAllFunc(config, func(content []byte) []byte {
 		var value string
 		if len(content) > 4 {
-			varName := string(content[2 : len(content)-1])
+			// 9 - removes #{secret.
+			// -1 - removes }
+			varName := string(content[9 : len(content)-1])
 			secretValue, err := secretStorage.Retrieve(varName)
 			if err != nil {
 				logger.GetInstance().Fatalf("Failed to resolve secret value. Error: %s", err.Error())
diff --git a/config/config_loader_test.go b/config/config_loader_test.go
index 8c68641..50b1ba4 100644
--- a/config/config_loader_test.go
+++ b/config/config_loader_test.go
@@ -69,7 +69,7 @@ func Test_resolveSecrets(t *testing.T) {
 		log.Fatal(err)
 	}
 
-	if cfg.Service.ETCD.Host != "value_secret.etcd/host" {
+	if cfg.Service.ETCD.Host != "value_etcd/host" {
 		t.Fatal("Invalid ETCD host in config")
 	}
 }
diff --git a/internal/secret/secrets.go b/internal/secret/secrets.go
index 0c7ace9..5c3c007 100644
--- a/internal/secret/secrets.go
+++ b/internal/secret/secrets.go
@@ -7,7 +7,7 @@ import (
 
 type SecretsStorageType string
 
-const AwsSecretStorage SecretsStorageType = "aws_secret_storage"
+const AwsSecretStorage SecretsStorageType = "aws_secret_manager"
 const MockSecretStorage SecretsStorageType = "mock_secret_storage"
 
 type Secrets struct{}
diff --git a/internal/secret/secretstorage/aws_secret_manager.go b/internal/secret/secretstorage/aws_secret_manager.go
index a140fdb..157f5e0 100644
--- a/internal/secret/secretstorage/aws_secret_manager.go
+++ b/internal/secret/secretstorage/aws_secret_manager.go
@@ -1,6 +1,8 @@
 package secretstorage
 
 import (
+	"fmt"
+
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/credentials"
 	"github.com/aws/aws-sdk-go/aws/session"
@@ -8,9 +10,9 @@ import (
 )
 
 type AwsSecretManagerConfig struct {
-	SecretKey   string `yaml:"secret_key"`
-	SecretKeyID string `yaml:"secret_key_id"`
-	Region      string `yaml:"region"`
+	SecretKey   string `yaml:"aws_secret_key"`
+	SecretKeyID string `yaml:"aws_secret_key_id"`
+	Region      string `yaml:"aws_region"`
 }
 
 type AwsSecretManager struct {
@@ -35,6 +37,7 @@ func NewAwsSecretManager(config AwsSecretManagerConfig) *AwsSecretManager {
 }
 
 func (a *AwsSecretManager) Retrieve(key string) (string, error) {
+	fmt.Println("Rertienving secret from AWS Secret Manager", key)
 	v, err := a.secretsmanager.GetSecretValue(&secretsmanager.GetSecretValueInput{
 		SecretId: aws.String(key),
 	})
diff --git a/public/stream/stream.go b/public/stream/stream.go
index 8b4ee84..d4f68b0 100644
--- a/public/stream/stream.go
+++ b/public/stream/stream.go
@@ -2,6 +2,9 @@ package stream
 
 import (
 	"errors"
+	"sync"
+	"time"
+
 	"github.com/usedatabrew/blink/config"
 	"github.com/usedatabrew/blink/internal/offset_storage"
 	"github.com/usedatabrew/blink/internal/schema"
@@ -9,8 +12,6 @@ import (
 	"github.com/usedatabrew/blink/internal/stream_context"
 	"github.com/usedatabrew/message"
 	"github.com/usedatabrew/tango"
-	"sync"
-	"time"
 )
 
 type Stream struct {
@@ -32,6 +33,8 @@ func InitFromConfig(config config.Configuration) (*Stream, error) {
 	if config.Service.OffsetStorageURI != "" {
 		offsetStorage := offset_storage.NewOffsetStorage(config.Service.OffsetStorageURI)
 		streamContext.SetOffsetStorage(offsetStorage)
+	} else {
+		streamContext.Logger.Warn("No offset storage URI provided. Offset will not be stored")
 	}
 
 	var processorList []string