Merged in feature/demo (pull request #116)

Demo prep + Fix client sync

* firstversion

* clientsync

* configlint

* fixtests

internal/client/service.go

@@ -14,7 +14,7 @@ type Client struct {
 	CanSync bool
 }
 
-func (c *Client) normalizeCanSyncUpdate(n **bool) {
+func (c *Client) NormalizeCanSyncUpdate(n **bool) {
 	if n == nil || *n == nil {
 		return
 	}
@@ -24,7 +24,7 @@ func (c *Client) normalizeCanSyncUpdate(n **bool) {
 	}
 }
 
-func (c *Client) normalizeNameUpdate(n **string) error {
+func (c *Client) NormalizeNameUpdate(n **string) error {
 	if n == nil || *n == nil {
 		return nil
 	}

internal/client/serviceprivate_test.go

@@ -39,19 +39,19 @@ func TestNormalizeCanSyncUpdate(t *testing.T) {
 		CanSync: true,
 	}
 
-	c.normalizeCanSyncUpdate(nil)
+	c.NormalizeCanSyncUpdate(nil)
 	var val *bool
-	c.normalizeCanSyncUpdate(&val)
+	c.NormalizeCanSyncUpdate(&val)
 	assert.Nil(t, val)
 
 	v := true
 	val = &v
-	c.normalizeCanSyncUpdate(&val)
+	c.NormalizeCanSyncUpdate(&val)
 	assert.Nil(t, val)
 
 	v = false
 	val = &v
-	c.normalizeCanSyncUpdate(&val)
+	c.NormalizeCanSyncUpdate(&val)
 	assert.NotNil(t, *val)
 }
 
@@ -61,24 +61,24 @@ func TestNormalizeNameUpdate(t *testing.T) {
 		Name: "example_name",
 	}
 
-	err := c.normalizeNameUpdate(nil)
+	err := c.NormalizeNameUpdate(nil)
 	require.NoError(t, err)
 
 	v := "update_name"
 	val := &v
-	err = c.normalizeNameUpdate(&val)
+	err = c.NormalizeNameUpdate(&val)
 	require.NoError(t, err)
 	assert.NotNil(t, val)
 	assert.Equal(t, "update_name", *val)
 
 	v = c.Name
 	val = &v
-	err = c.normalizeNameUpdate(&val)
+	err = c.NormalizeNameUpdate(&val)
 	require.NoError(t, err)
 	assert.Nil(t, val)
 
 	v = "###"
 	val = &v
-	err = c.normalizeNameUpdate(&val)
+	err = c.NormalizeNameUpdate(&val)
 	assert.Error(t, err)
 }

internal/client/update/service.go

@@ -0,0 +1,28 @@
+package clientupdate
+
+import (
+	"queryorchestration/internal/client"
+	"queryorchestration/internal/serviceconfig"
+	"queryorchestration/internal/serviceconfig/queue/clientsync"
+)
+
+type ConfigProvider interface {
+	serviceconfig.ConfigProvider
+	clientsync.ConfigProvider
+}
+
+type Services struct {
+	Client *client.Service
+}
+
+type Service struct {
+	cfg ConfigProvider
+	svc *Services
+}
+
+func New(cfg ConfigProvider, svc *Services) *Service {
+	return &Service{
+		cfg,
+		svc,
+	}
+}

internal/client/update/service_test.go

@@ -0,0 +1,24 @@
+package clientupdate_test
+
+import (
+	"testing"
+
+	"queryorchestration/internal/client"
+	"queryorchestration/internal/database/repository"
+	"queryorchestration/internal/serviceconfig"
+
+	"github.com/pashagolub/pgxmock/v3"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestService(t *testing.T) {
+	pool, err := pgxmock.NewPool()
+	require.NoError(t, err)
+	cfg := &serviceconfig.BaseConfig{}
+	cfg.DBPool = pool
+	cfg.DBQueries = repository.New(pool)
+
+	svc := client.New(cfg)
+	assert.NotNil(t, svc)
+}

internal/client/update/update.go

@@ -1,11 +1,14 @@
-package client
+package clientupdate
 
 import (
 	"context"
 	"errors"
 	"log/slog"
 
+	clientsyncrunner "queryorchestration/api/clientSyncRunner"
+	"queryorchestration/internal/client"
 	"queryorchestration/internal/database/repository"
+	"queryorchestration/internal/serviceconfig/queue"
 	"queryorchestration/internal/validation"
 )
 
@@ -15,7 +18,7 @@ type Update struct {
 }
 
 func (s *Service) Update(ctx context.Context, id string, entity *Update) error {
-	current, err := s.Get(ctx, id)
+	current, err := s.svc.Client.Get(ctx, id)
 	if err != nil {
 		return err
 	}
@@ -30,9 +33,27 @@ func (s *Service) Update(ctx context.Context, id string, entity *Update) error {
 		return err
 	}
 
+	err = s.informUpdate(ctx, id, entity)
+	if err != nil {
+		return err
+	}
+
 	return nil
 }
 
+func (s *Service) informUpdate(ctx context.Context, id string, entity *Update) error {
+	if entity.CanSync == nil || !*entity.CanSync {
+		return nil
+	}
+
+	return s.cfg.SendToQueue(ctx, &queue.SendParams{
+		QueueURL: s.cfg.GetClientSyncURL(),
+		Body: clientsyncrunner.Body{
+			ClientID: id,
+		},
+	})
+}
+
 func (s *Service) submitUpdate(ctx context.Context, id string, entity *Update) error {
 	return s.cfg.ExecuteDBTransaction(ctx, func(ctx context.Context, q *repository.Queries) error {
 		id := id
@@ -63,17 +84,17 @@ func (s *Service) submitUpdate(ctx context.Context, id string, entity *Update) e
 	})
 }
 
-func (s *Service) normalizeUpdateParams(id string, current *Client, entity *Update) error {
+func (s *Service) normalizeUpdateParams(id string, current *client.Client, entity *Update) error {
 	if entity == nil || current == nil || current.ID != id {
 		return errors.New("no updates presented")
 	}
 
-	err := current.normalizeNameUpdate(&entity.Name)
+	err := current.NormalizeNameUpdate(&entity.Name)
 	if err != nil {
 		return err
 	}
 
-	current.normalizeCanSyncUpdate(&entity.CanSync)
+	current.NormalizeCanSyncUpdate(&entity.CanSync)
 
 	if validation.AreAllPointersNilExcept(entity) {
 		return errors.New("no updates presented")

internal/client/update/update_test.go

@@ -1,29 +1,42 @@
-package client
+package clientupdate
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
+	"queryorchestration/internal/client"
 	"queryorchestration/internal/database/repository"
 	"queryorchestration/internal/serviceconfig"
+	"queryorchestration/internal/serviceconfig/queue/clientsync"
+	queuemock "queryorchestration/mocks/queue"
 
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
 	"github.com/pashagolub/pgxmock/v3"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 )
 
+type Config struct {
+	serviceconfig.BaseConfig
+	clientsync.ClientSyncConfig
+}
+
 func TestUpdate(t *testing.T) {
 	ctx := context.Background()
 
 	pool, err := pgxmock.NewPool()
 	require.NoError(t, err)
-	cfg := &serviceconfig.BaseConfig{}
+	cfg := &Config{}
 	cfg.DBPool = pool
 	cfg.DBQueries = repository.New(pool)
 
-	svc := New(cfg)
+	svc := New(cfg, &Services{
+		Client: client.New(cfg),
+	})
 
-	c := Client{
+	c := client.Client{
 		ID:      "huhu",
 		Name:    "example_name",
 		CanSync: false,
@@ -72,7 +85,7 @@ func TestNormalizeUpdateParams(t *testing.T) {
 	err := svc.normalizeUpdateParams("huhu", nil, nil)
 	assert.Error(t, err)
 
-	current := &Client{
+	current := &client.Client{
 		ID:   "hi",
 		Name: "client",
 	}
@@ -109,13 +122,13 @@ func TestSubmitUpdate(t *testing.T) {
 
 	pool, err := pgxmock.NewPool()
 	require.NoError(t, err)
-	cfg := &serviceconfig.BaseConfig{}
+	cfg := &Config{}
 	cfg.DBPool = pool
 	cfg.DBQueries = repository.New(pool)
 
-	svc := New(cfg)
+	svc := New(cfg, &Services{})
 
-	c := Client{
+	c := client.Client{
 		ID:      "hello",
 		Name:    "example_name",
 		CanSync: false,
@@ -151,3 +164,49 @@ func TestSubmitUpdate(t *testing.T) {
 	err = svc.submitUpdate(ctx, c.ID, &update)
 	require.NoError(t, err)
 }
+
+func TestInformUpdate(t *testing.T) {
+	ctx := context.Background()
+
+	cfg := &Config{}
+	mockSQS := queuemock.NewMockSQSClient(t)
+	cfg.QueueClient = mockSQS
+
+	svc := New(cfg, &Services{})
+
+	id := "HIELO"
+	t.Run("no cansync", func(t *testing.T) {
+		update := Update{}
+
+		err := svc.informUpdate(ctx, id, &update)
+		require.NoError(t, err)
+	})
+	t.Run("false cansync", func(t *testing.T) {
+		cansync := false
+		update := Update{
+			CanSync: &cansync,
+		}
+
+		err := svc.informUpdate(ctx, id, &update)
+		require.NoError(t, err)
+	})
+	t.Run("true cansync", func(t *testing.T) {
+		cansync := true
+		update := Update{
+			CanSync: &cansync,
+		}
+
+		mockSQS.EXPECT().
+			SendMessage(
+				mock.Anything,
+				mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
+					return *in.QueueUrl == cfg.ClientSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", id)
+				}),
+				mock.Anything,
+			).
+			Return(&sqs.SendMessageOutput{}, nil)
+
+		err := svc.informUpdate(ctx, id, &update)
+		require.NoError(t, err)
+	})
+}

internal/document/store/process.go

@@ -19,7 +19,8 @@ type Params struct {
 type EventS3 string
 
 const (
-	EventS3ObjectCreatedPut = "ObjectCreated:Put"
+	EventS3ObjectCreatedPut                     = "ObjectCreated:Put"
+	EventS3ObjectCreatedCompleteMultipartUpload = "ObjectCreated:CompleteMultipartUpload"
 )
 
 func (s *Service) Process(ctx context.Context, params Params) error {
@@ -52,5 +53,6 @@ func (s *Service) Process(ctx context.Context, params Params) error {
 }
 
 func (s *Service) isSupportedEvent(name EventS3) bool {
-	return name == EventS3ObjectCreatedPut
+	return name == EventS3ObjectCreatedPut ||
+		name == EventS3ObjectCreatedCompleteMultipartUpload
 }

internal/serviceconfig/textract/config.go

@@ -2,10 +2,12 @@ package textract
 
 import (
 	"context"
+	"log/slog"
 
 	awsc "queryorchestration/internal/serviceconfig/aws"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/credentials"
 	"github.com/aws/aws-sdk-go-v2/service/textract"
 )
 
@@ -20,8 +22,12 @@ type ConfigProvider interface {
 }
 
 type TextractConfig struct {
-	AWSEndpointUrlTextract string `env:"AWS_ENDPOINT_URL_TEXTRACT"`
-	TextractClient         TextractClient
+	AWSEndpointUrlTextract     string `env:"AWS_ENDPOINT_URL_TEXTRACT"`
+	AWSTextractAccessKeyId     string `env:"AWS_ACCESS_KEY_ID_TEXTRACT"`
+	AWSTextractSecretAccessKey string `env:"AWS_SECRET_ACCESS_KEY_TEXTRACT"`
+	AWSTextractSessionToken    string `env:"AWS_SESSION_TOKEN_TEXTRACT"`
+	AWSTextractRegion          string `env:"AWS_REGION_TEXTRACT"`
+	TextractClient             TextractClient
 }
 
 func (c *TextractConfig) SetTextractClientWithCfg(ctx context.Context, cfg aws.Config) {
@@ -45,6 +51,15 @@ func (c *TextractConfig) SetTextractClient(ctx context.Context) error {
 		return err
 	}
 
+	if c.AWSTextractAccessKeyId != "" && c.AWSTextractSecretAccessKey != "" {
+		cfg.Credentials = credentials.NewStaticCredentialsProvider(c.AWSTextractAccessKeyId, c.AWSTextractSecretAccessKey, c.AWSTextractSessionToken)
+
+		if c.AWSTextractRegion != "" {
+			cfg.Region = c.AWSTextractRegion
+		}
+		slog.Debug("textract specific credentials", "key_id", c.AWSTextractAccessKeyId, "region", c.AWSTextractRegion)
+	}
+
 	c.SetTextractClientWithCfg(ctx, cfg)
 
 	return nil

internal/serviceconfig/textract/config_test.go

@@ -21,11 +21,25 @@ func TestGetTextractClient(t *testing.T) {
 
 func TestTextractClient(t *testing.T) {
 	os.Clearenv()
-	ctx := context.Background()
-	c := textract.TextractConfig{}
-	err := c.SetTextractClient(ctx)
-	require.NoError(t, err)
-	assert.NotNil(t, c.TextractClient)
+	t.Run("stndard", func(t *testing.T) {
+		ctx := context.Background()
+		c := textract.TextractConfig{}
+		err := c.SetTextractClient(ctx)
+		require.NoError(t, err)
+		assert.NotNil(t, c.TextractClient)
+	})
+	t.Run("textract specific", func(t *testing.T) {
+		ctx := context.Background()
+		c := textract.TextractConfig{
+			AWSTextractAccessKeyId:     "not empty",
+			AWSTextractSecretAccessKey: "not empty",
+			AWSTextractSessionToken:    "not empty",
+			AWSTextractRegion:          "not empty",
+		}
+		err := c.SetTextractClient(ctx)
+		require.NoError(t, err)
+		assert.NotNil(t, c.TextractClient)
+	})
 }
 
 func TestTextractClientWithProfile(t *testing.T) {

internal/test/queue.go

@@ -21,6 +21,7 @@ import (
 )
 
 func SetQueueClient(t testing.TB, ctx context.Context, cfg queue.ConfigProvider) {
+	t.Helper()
 	err := cfg.SetQueueClient(ctx)
 	require.NoError(t, err)
 }
@@ -34,6 +35,7 @@ func GetQueueArn(cfg awsc.ConfigProvider, name RunnerName) string {
 }
 
 func CreateQueue(t testing.TB, ctx context.Context, cfg serviceconfig.ConfigProvider, name string) string {
+	t.Helper()
 	queueM, err := cfg.GetQueueClient().CreateQueue(ctx, &sqs.CreateQueueInput{
 		QueueName: aws.String(name),
 	})
@@ -47,6 +49,7 @@ func CreateQueue(t testing.TB, ctx context.Context, cfg serviceconfig.ConfigProv
 }
 
 func AssertMessage(t testing.TB, ctx context.Context, cfg serviceconfig.ConfigProvider, params *queue.ReceiveParams) types.Message {
+	t.Helper()
 	timeout := time.After(30 * time.Second)
 	tick := time.NewTicker(2 * time.Second)
 	defer tick.Stop()
@@ -73,6 +76,7 @@ func AssertMessage(t testing.TB, ctx context.Context, cfg serviceconfig.ConfigPr
 }
 
 func AssertMessageBody(t testing.TB, ctx context.Context, cfg serviceconfig.ConfigProvider, url string, body *regexp.Regexp) {
+	t.Helper()
 	message := AssertMessage(t, ctx, cfg, &queue.ReceiveParams{
 		QueueURL: url,
 	})
@@ -81,6 +85,7 @@ func AssertMessageBody(t testing.TB, ctx context.Context, cfg serviceconfig.Conf
 }
 
 func AssertMessageAttr(t testing.TB, ctx context.Context, cfg serviceconfig.ConfigProvider, url string, name string, value *regexp.Regexp) {
+	t.Helper()
 	message := AssertMessage(t, ctx, cfg, &queue.ReceiveParams{
 		QueueURL:   url,
 		Attributes: []string{name},