Merged in feature/docclean (pull request #55)

Doc Clean Structure

* outline

* isdocclean

* placeholder for clean log

* versionplustesting

* versionplustesting

* testspassed

internal/database/repository/clean.sql.go

@@ -0,0 +1,64 @@
+// Code generated by sqlc. DO NOT EDIT.
+// versions:
+//   sqlc v1.27.0
+// source: clean.sql
+
+package repository
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v5/pgtype"
+)
+
+const addDocumentCleanEntry = `-- name: AddDocumentCleanEntry :exec
+INSERT INTO documentCleans (documentId, version, bucket, key) VALUES ($1, $2, $3, $4)
+`
+
+type AddDocumentCleanEntryParams struct {
+	Documentid pgtype.UUID `db:"documentid"`
+	Version    int32       `db:"version"`
+	Bucket     string      `db:"bucket"`
+	Key        string      `db:"key"`
+}
+
+// AddDocumentCleanEntry
+//
+//	INSERT INTO documentCleans (documentId, version, bucket, key) VALUES ($1, $2, $3, $4)
+func (q *Queries) AddDocumentCleanEntry(ctx context.Context, arg *AddDocumentCleanEntryParams) error {
+	_, err := q.db.Exec(ctx, addDocumentCleanEntry,
+		arg.Documentid,
+		arg.Version,
+		arg.Bucket,
+		arg.Key,
+	)
+	return err
+}
+
+const isDocumentClean = `-- name: IsDocumentClean :one
+SELECT EXISTS(
+    SELECT 1
+        FROM documentCleans AS dc
+        JOIN documents AS d ON d.id = dc.documentId
+        JOIN collectors as c ON d.jobId = c.jobId
+        JOIN collectorCodeVersions as cv ON c.id = cv.collectorId
+        WHERE dc.documentId = $1 and dc.version >= cv.minCleanVersion
+)
+`
+
+// IsDocumentClean
+//
+//	SELECT EXISTS(
+//	    SELECT 1
+//	        FROM documentCleans AS dc
+//	        JOIN documents AS d ON d.id = dc.documentId
+//	        JOIN collectors as c ON d.jobId = c.jobId
+//	        JOIN collectorCodeVersions as cv ON c.id = cv.collectorId
+//	        WHERE dc.documentId = $1 and dc.version >= cv.minCleanVersion
+//	)
+func (q *Queries) IsDocumentClean(ctx context.Context, documentid pgtype.UUID) (bool, error) {
+	row := q.db.QueryRow(ctx, isDocumentClean, documentid)
+	var exists bool
+	err := row.Scan(&exists)
+	return exists, err
+}

internal/database/repository/clean_test.go

@@ -0,0 +1,77 @@
+package repository_test
+
+import (
+	"context"
+	"os"
+	"path"
+	"queryorchestration/internal/database/repository"
+	"queryorchestration/internal/serviceconfig"
+	"queryorchestration/internal/test"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestClean(t *testing.T) {
+	ctx := context.Background()
+
+	cfg := &serviceconfig.BaseConfig{}
+	test.SetCfgProvider(t, cfg)
+	cfg.SetBasePath(path.Join(os.Getenv("PWD"), "../../.."))
+	_, cleanup := test.CreateDB(t, ctx, &test.CreateDatabaseConfig{
+		Cfg:           cfg,
+		RunMigrations: true,
+	})
+	defer cleanup()
+
+	queries := cfg.GetDBQueries()
+
+	clientId, err := queries.CreateClient(ctx, "example_client")
+	assert.NoError(t, err)
+	jobId, err := queries.CreateJob(ctx, clientId)
+	assert.NoError(t, err)
+
+	hash := "example_hash"
+	id, err := queries.CreateDocument(ctx, &repository.CreateDocumentParams{
+		Jobid: jobId,
+		Hash:  hash,
+	})
+	assert.NoError(t, err)
+	assert.NotEmpty(t, id)
+
+	collId, err := queries.CreateCollector(ctx, jobId)
+	assert.NoError(t, err)
+	err = queries.AddCollectorCodeVersion(ctx, &repository.AddCollectorCodeVersionParams{
+		Collectorid:     collId,
+		Addedversion:    1,
+		Mincleanversion: 2,
+		Mintextversion:  1,
+	})
+	assert.NoError(t, err)
+
+	isclean, err := queries.IsDocumentClean(ctx, id)
+	assert.NoError(t, err)
+	assert.False(t, isclean)
+
+	err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{
+		Documentid: id,
+		Version:    1,
+	})
+	assert.NoError(t, err)
+
+	isclean, err = queries.IsDocumentClean(ctx, id)
+	assert.NoError(t, err)
+	assert.False(t, isclean)
+
+	err = queries.AddCollectorCodeVersion(ctx, &repository.AddCollectorCodeVersionParams{
+		Collectorid:     collId,
+		Addedversion:    1,
+		Mincleanversion: 1,
+		Mintextversion:  1,
+	})
+	assert.NoError(t, err)
+
+	isclean, err = queries.IsDocumentClean(ctx, id)
+	assert.NoError(t, err)
+	assert.True(t, isclean)
+}

internal/database/repository/document.sql.go

@@ -12,20 +12,20 @@ import (
 )
 
 const addDocumentEntry = `-- name: AddDocumentEntry :exec
-INSERT INTO documentEntries (documentId, bucket, location) VALUES ($1, $2, $3)
+INSERT INTO documentEntries (documentId, bucket, key) VALUES ($1, $2, $3)
 `
 
 type AddDocumentEntryParams struct {
 	Documentid pgtype.UUID `db:"documentid"`
 	Bucket     string      `db:"bucket"`
-	Location   string      `db:"location"`
+	Key        string      `db:"key"`
 }
 
 // AddDocumentEntry
 //
-//	INSERT INTO documentEntries (documentId, bucket, location) VALUES ($1, $2, $3)
+//	INSERT INTO documentEntries (documentId, bucket, key) VALUES ($1, $2, $3)
 func (q *Queries) AddDocumentEntry(ctx context.Context, arg *AddDocumentEntryParams) error {
-	_, err := q.db.Exec(ctx, addDocumentEntry, arg.Documentid, arg.Bucket, arg.Location)
+	_, err := q.db.Exec(ctx, addDocumentEntry, arg.Documentid, arg.Bucket, arg.Key)
 	return err
 }
 
@@ -62,6 +62,26 @@ func (q *Queries) GetDocument(ctx context.Context, id pgtype.UUID) (*Document, e
 	return &i, err
 }
 
+const getDocumentEntry = `-- name: GetDocumentEntry :one
+SELECT documentId, bucket, key FROM documentEntries WHERE documentId = $1 ORDER BY id DESC LIMIT 1
+`
+
+type GetDocumentEntryRow struct {
+	Documentid pgtype.UUID `db:"documentid"`
+	Bucket     string      `db:"bucket"`
+	Key        string      `db:"key"`
+}
+
+// GetDocumentEntry
+//
+//	SELECT documentId, bucket, key FROM documentEntries WHERE documentId = $1 ORDER BY id DESC LIMIT 1
+func (q *Queries) GetDocumentEntry(ctx context.Context, documentid pgtype.UUID) (*GetDocumentEntryRow, error) {
+	row := q.db.QueryRow(ctx, getDocumentEntry, documentid)
+	var i GetDocumentEntryRow
+	err := row.Scan(&i.Documentid, &i.Bucket, &i.Key)
+	return &i, err
+}
+
 const getDocumentIDByHash = `-- name: GetDocumentIDByHash :one
 SELECT id FROM documents WHERE hash = $1 and jobId = $2
 `

internal/database/repository/document_test.go

@@ -39,15 +39,6 @@ func TestDocument(t *testing.T) {
 	assert.NoError(t, err)
 	assert.NotEmpty(t, id)
 
-	bucket := "example_bucket"
-	location := "/i/am/here"
-	err = queries.AddDocumentEntry(ctx, &repository.AddDocumentEntryParams{
-		Documentid: id,
-		Bucket:     bucket,
-		Location:   location,
-	})
-	assert.NoError(t, err)
-
 	doc, err := queries.GetDocument(ctx, id)
 	assert.NoError(t, err)
 	assert.EqualExportedValues(t, &repository.Document{
@@ -62,4 +53,49 @@ func TestDocument(t *testing.T) {
 	})
 	assert.NoError(t, err)
 	assert.EqualExportedValues(t, id, docid)
+
+	err = queries.AddDocumentEntry(ctx, &repository.AddDocumentEntryParams{
+		Documentid: id,
+		Bucket:     "bucket_one",
+		Key:        "/i/am/here",
+	})
+	assert.NoError(t, err)
+
+	entry, err := queries.GetDocumentEntry(ctx, id)
+	assert.NoError(t, err)
+	assert.EqualExportedValues(t, &repository.GetDocumentEntryRow{
+		Documentid: id,
+		Bucket:     "bucket_one",
+		Key:        "/i/am/here",
+	}, entry)
+
+	err = queries.AddDocumentEntry(ctx, &repository.AddDocumentEntryParams{
+		Documentid: id,
+		Bucket:     "bucket_two",
+		Key:        "/you/is/there",
+	})
+	assert.NoError(t, err)
+
+	entry, err = queries.GetDocumentEntry(ctx, id)
+	assert.NoError(t, err)
+	assert.EqualExportedValues(t, &repository.GetDocumentEntryRow{
+		Documentid: id,
+		Bucket:     "bucket_two",
+		Key:        "/you/is/there",
+	}, entry)
+
+	err = queries.AddDocumentEntry(ctx, &repository.AddDocumentEntryParams{
+		Documentid: id,
+		Bucket:     "bucket_three",
+		Key:        "/who/is/where",
+	})
+	assert.NoError(t, err)
+
+	entry, err = queries.GetDocumentEntry(ctx, id)
+	assert.NoError(t, err)
+	assert.EqualExportedValues(t, &repository.GetDocumentEntryRow{
+		Documentid: id,
+		Bucket:     "bucket_three",
+		Key:        "/who/is/where",
+	}, entry)
 }

internal/database/repository/models.go

@@ -113,12 +113,19 @@ type Document struct {
 	Hash  string      `db:"hash"`
 }
 
+type Documentclean struct {
+	ID         pgtype.UUID `db:"id"`
+	Documentid pgtype.UUID `db:"documentid"`
+	Version    int32       `db:"version"`
+	Bucket     string      `db:"bucket"`
+	Key        string      `db:"key"`
+}
+
 type Documententry struct {
-	ID         pgtype.UUID      `db:"id"`
-	Documentid pgtype.UUID      `db:"documentid"`
-	Bucket     string           `db:"bucket"`
-	Location   string           `db:"location"`
-	Createdat  pgtype.Timestamp `db:"createdat"`
+	ID         pgtype.UUID `db:"id"`
+	Documentid pgtype.UUID `db:"documentid"`
+	Bucket     string      `db:"bucket"`
+	Key        string      `db:"key"`
 }
 
 type Fullactivecollector struct {

internal/document/clean/clean.go

@@ -0,0 +1,54 @@
+package documentclean
+
+import (
+	"context"
+	"queryorchestration/internal/database"
+	"queryorchestration/internal/database/repository"
+	"queryorchestration/internal/document"
+
+	"github.com/google/uuid"
+)
+
+type CleanParams struct {
+	ID       uuid.UUID
+	Location document.Location
+}
+
+func (s *Service) executeCleanTasks(params *CleanParams) (*document.Location, error) {
+	// TODO - various cleaning tasks
+	return &params.Location, nil
+}
+
+func (s *Service) clean(ctx context.Context, id uuid.UUID) error {
+	docId := database.MustToDBUUID(id)
+
+	entry, err := s.cfg.GetDBQueries().GetDocumentEntry(ctx, docId)
+	if err != nil {
+		return err
+	}
+
+	outLocation, err := s.executeCleanTasks(&CleanParams{
+		ID: id,
+		Location: document.Location{
+			Bucket: entry.Bucket,
+			Key:    entry.Key,
+		},
+	})
+	if err != nil {
+		return err
+	}
+
+	version := s.svc.Document.GetCleanVersion()
+
+	err = s.cfg.GetDBQueries().AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{
+		Documentid: docId,
+		Version:    version,
+		Bucket:     outLocation.Bucket,
+		Key:        outLocation.Key,
+	})
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

internal/document/clean/clean_test.go

@@ -0,0 +1,62 @@
+package documentclean
+
+import (
+	"context"
+	"queryorchestration/internal/database"
+	"queryorchestration/internal/database/repository"
+	"queryorchestration/internal/document"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/pashagolub/pgxmock/v3"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestClean(t *testing.T) {
+	ctx := context.Background()
+	pool, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to open pgxmock database: %v", err)
+	}
+
+	cfg := &DocCleanConfig{}
+	cfg.DBPool = pool
+	cfg.DBQueries = repository.New(pool)
+
+	svc := Service{
+		cfg: cfg,
+		svc: &Services{
+			Document: document.New(cfg),
+		},
+	}
+
+	id := uuid.New()
+	inloc := document.Location{
+		Bucket: "bucket_name",
+		Key:    "/i/am/here",
+	}
+
+	pool.ExpectQuery("name: GetDocumentEntry :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows(
+		pgxmock.NewRows([]string{"documentId", "bucket", "key"}).
+			AddRow(database.MustToDBUUID(id), inloc.Bucket, inloc.Key),
+	)
+	pool.ExpectExec("name: AddDocumentCleanEntry :exec").WithArgs(database.MustToDBUUID(id), int32(1), inloc.Bucket, inloc.Key).
+		WillReturnResult(pgxmock.NewResult("", 1))
+
+	err = svc.clean(ctx, id)
+	assert.NoError(t, err)
+}
+
+func TestExecuteCleanTasks(t *testing.T) {
+	svc := Service{}
+
+	id := uuid.New()
+	location := document.Location{}
+
+	outloc, err := svc.executeCleanTasks(&CleanParams{
+		ID:       id,
+		Location: location,
+	})
+	assert.NoError(t, err)
+	assert.EqualExportedValues(t, location, *outloc)
+}

internal/document/clean/create.go

@@ -0,0 +1,45 @@
+package documentclean
+
+import (
+	"context"
+	"queryorchestration/internal/database"
+	documenttext "queryorchestration/internal/document/text"
+	"queryorchestration/internal/serviceconfig/queue"
+
+	"github.com/google/uuid"
+)
+
+func (s *Service) Create(ctx context.Context, id uuid.UUID) error {
+	isclean, err := s.cfg.GetDBQueries().IsDocumentClean(ctx, database.MustToDBUUID(id))
+	if err != nil {
+		return err
+	}
+
+	if !isclean {
+		err = s.clean(ctx, id)
+		if err != nil {
+			return err
+		}
+	}
+
+	err = s.informClean(ctx, id)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (s *Service) informClean(ctx context.Context, id uuid.UUID) error {
+	err := s.cfg.SendToQueue(ctx, &queue.SendParams{
+		QueueURL: s.cfg.GetDocumentTextURL(),
+		Body: documenttext.Create{
+			ID: id,
+		},
+	})
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

internal/document/clean/create_test.go

@@ -0,0 +1,105 @@
+package documentclean
+
+import (
+	"context"
+	"fmt"
+	"queryorchestration/internal/database"
+	"queryorchestration/internal/database/repository"
+	"queryorchestration/internal/document"
+	"queryorchestration/internal/serviceconfig"
+	"queryorchestration/internal/serviceconfig/objectstore"
+	"queryorchestration/internal/serviceconfig/queue/documenttext"
+	queuemock "queryorchestration/mocks/queue"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
+	"github.com/google/uuid"
+	"github.com/pashagolub/pgxmock/v3"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+)
+
+type DocCleanConfig struct {
+	serviceconfig.BaseConfig
+	documenttext.DocTextConfig
+	objectstore.ObjectStoreConfig
+}
+
+func TestCreate(t *testing.T) {
+	ctx := context.Background()
+	pool, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to open pgxmock database: %v", err)
+	}
+
+	mockSQS := queuemock.NewMockSQSClient(t)
+
+	cfg := &DocCleanConfig{}
+	cfg.QueueClient = mockSQS
+	cfg.DocumentTextURL = "/i/am/here"
+	cfg.DBPool = pool
+	cfg.DBQueries = repository.New(pool)
+
+	svc := Service{
+		cfg: cfg,
+		svc: &Services{
+			Document: document.New(cfg),
+		},
+	}
+
+	id := uuid.New()
+	inloc := document.Location{
+		Bucket: "bucket_name",
+		Key:    "/i/am/here",
+	}
+
+	pool.ExpectQuery("name: IsDocumentClean :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows(
+		pgxmock.NewRows([]string{"isclean"}).
+			AddRow(false),
+	)
+	pool.ExpectQuery("name: GetDocumentEntry :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows(
+		pgxmock.NewRows([]string{"documentId", "bucket", "key"}).
+			AddRow(database.MustToDBUUID(id), inloc.Bucket, inloc.Key),
+	)
+	pool.ExpectExec("name: AddDocumentCleanEntry :exec").WithArgs(database.MustToDBUUID(id), int32(1), inloc.Bucket, inloc.Key).
+		WillReturnResult(pgxmock.NewResult("", 1))
+
+	mockSQS.EXPECT().
+		SendMessage(
+			mock.Anything,
+			mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
+				return *in.QueueUrl == cfg.DocumentTextURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", id)
+			}),
+			mock.Anything,
+		).
+		Return(&sqs.SendMessageOutput{}, nil)
+
+	err = svc.Create(ctx, id)
+	assert.NoError(t, err)
+}
+
+func TestInformClean(t *testing.T) {
+	ctx := context.Background()
+	mockSQS := queuemock.NewMockSQSClient(t)
+
+	cfg := &DocCleanConfig{}
+	cfg.QueueClient = mockSQS
+	cfg.DocumentTextURL = "/i/am/here"
+	svc := Service{
+		cfg: cfg,
+	}
+	id := uuid.New()
+
+	mockSQS.EXPECT().
+		SendMessage(
+			mock.Anything,
+			mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
+				return *in.QueueUrl == cfg.DocumentTextURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", id)
+			}),
+			mock.Anything,
+		).
+		Return(&sqs.SendMessageOutput{}, nil)
+
+	err := svc.informClean(ctx, id)
+	assert.NoError(t, err)
+}

internal/document/clean/service.go

@@ -1,26 +1,30 @@
 package documentclean
 
 import (
-	"errors"
-
-	"github.com/google/uuid"
+	"queryorchestration/internal/document"
+	"queryorchestration/internal/serviceconfig"
+	"queryorchestration/internal/serviceconfig/objectstore"
+	"queryorchestration/internal/serviceconfig/queue/documenttext"
 )
 
-type Create struct {
-	ID uuid.UUID `json:"id" validate:"required,uuid"`
+type ConfigProvider interface {
+	serviceconfig.ConfigProvider
+	objectstore.ConfigProvider
+	documenttext.ConfigProvider
+}
+
+type Services struct {
+	Document *document.Service
 }
 
 type Service struct {
+	cfg ConfigProvider
+	svc *Services
 }
 
-func New() *Service {
-	return &Service{}
-}
-
-func (s *Service) IsValidVersion(v int32) error {
-	if v <= 0 {
-		return errors.New("document clean code version must be > 0")
+func New(cfg ConfigProvider, svc *Services) *Service {
+	return &Service{
+		cfg: cfg,
+		svc: svc,
 	}
-
-	return nil
 }

internal/document/clean/service_test.go

@@ -2,19 +2,22 @@ package documentclean_test
 
 import (
 	documentclean "queryorchestration/internal/document/clean"
+	"queryorchestration/internal/serviceconfig"
+	"queryorchestration/internal/serviceconfig/objectstore"
+	"queryorchestration/internal/serviceconfig/queue/documenttext"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 )
 
+type DocCleanConfig struct {
+	serviceconfig.BaseConfig
+	documenttext.DocTextConfig
+	objectstore.ObjectStoreConfig
+}
+
 func TestService(t *testing.T) {
-	svc := documentclean.New()
+	cfg := &DocCleanConfig{}
+	svc := documentclean.New(cfg, nil)
 	assert.NotNil(t, svc)
 }
-
-func TestIsValidVersion(t *testing.T) {
-	svc := documentclean.New()
-
-	assert.Nil(t, svc.IsValidVersion(2))
-	assert.Error(t, svc.IsValidVersion(-1))
-}

internal/document/init/create.go

@@ -5,15 +5,14 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
+	doccleanrunner "queryorchestration/api/docCleanRunner"
 	"queryorchestration/internal/database"
 	"queryorchestration/internal/database/repository"
 	"queryorchestration/internal/document"
-	documentclean "queryorchestration/internal/document/clean"
 	"queryorchestration/internal/job"
 	"queryorchestration/internal/serviceconfig/queue"
 	"regexp"
 
-	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5/pgtype"
 )
@@ -55,8 +54,7 @@ type createDocumentParams struct {
 	ID       *pgtype.UUID
 	JobID    pgtype.UUID
 	Hash     string
-	Bucket   string
-	Location string
+	Location document.Location
 }
 
 func (s *Service) getCreateParams(ctx context.Context, doc *Create) (*createDocumentParams, error) {
@@ -75,7 +73,6 @@ func (s *Service) getCreateParams(ctx context.Context, doc *Create) (*createDocu
 		ID:       docID,
 		JobID:    database.MustToDBUUID(doc.JobID),
 		Hash:     doc.Hash,
-		Bucket:   doc.Bucket,
 		Location: doc.Location,
 	}, nil
 }
@@ -101,8 +98,8 @@ func (s *Service) submitCreate(ctx context.Context, params *createDocumentParams
 
 		err := s.cfg.GetDBQueries().AddDocumentEntry(ctx, &repository.AddDocumentEntryParams{
 			Documentid: dbid,
-			Bucket:     params.Bucket,
-			Location:   params.Location,
+			Bucket:     params.Location.Bucket,
+			Key:        params.Location.Key,
 		})
 		if err != nil {
 			return err
@@ -126,10 +123,9 @@ func (s *Service) informCreate(ctx context.Context, id uuid.UUID, j *job.Job) er
 
 	err := s.cfg.SendToQueue(ctx, &queue.SendParams{
 		QueueURL: s.cfg.GetDocumentCleanURL(),
-		Body: documentclean.Create{
+		Body: doccleanrunner.Create{
 			ID: id,
 		},
-		Attributes: map[string]types.MessageAttributeValue{},
 	})
 	if err != nil {
 		return err

internal/document/init/create_test.go

@@ -57,18 +57,20 @@ func TestCreate(t *testing.T) {
 		JobID: j.ID,
 		Hash:  "example_hash",
 	}
-	bucket := "eample_bucket"
-	location := "/i/am/here"
+	location := document.Location{
+		Bucket: "example_bucket",
+		Key:    "/i/am/here",
+	}
 
 	pool.ExpectQuery("name: GetJob :one").WithArgs(database.MustToDBUUID(j.ID)).WillReturnRows(
 		pgxmock.NewRows([]string{"id", "clientId", "canSync"}).
 			AddRow(database.MustToDBUUID(j.ID), database.MustToDBUUID(j.ClientID), j.CanSync),
 	)
-	pool.ExpectQuery("-- name: GetClient :one").WithArgs(database.MustToDBUUID(j.ClientID)).WillReturnRows(
+	pool.ExpectQuery("name: GetClient :one").WithArgs(database.MustToDBUUID(j.ClientID)).WillReturnRows(
 		pgxmock.NewRows([]string{"id", "name", "canSync"}).
 			AddRow(database.MustToDBUUID(j.ClientID), "client_name", true),
 	)
-	pool.ExpectQuery("-- name: GetDocumentIDByHash :one").WithArgs(doc.Hash, database.MustToDBUUID(doc.JobID)).WillReturnRows(
+	pool.ExpectQuery("name: GetDocumentIDByHash :one").WithArgs(doc.Hash, database.MustToDBUUID(doc.JobID)).WillReturnRows(
 		pgxmock.NewRows([]string{"id"}),
 	)
 	pool.ExpectBegin()
@@ -77,14 +79,13 @@ func TestCreate(t *testing.T) {
 			pgxmock.NewRows([]string{"id"}).
 				AddRow(database.MustToDBUUID(doc.ID)),
 		)
-	pool.ExpectExec("name: AddDocumentEntry :exec").WithArgs(database.MustToDBUUID(doc.ID), bucket, location).
+	pool.ExpectExec("name: AddDocumentEntry :exec").WithArgs(database.MustToDBUUID(doc.ID), location.Bucket, location.Key).
 		WillReturnResult(pgxmock.NewResult("", 1))
 	pool.ExpectCommit()
 
 	id, err := svc.Create(ctx, &Create{
 		JobID:    doc.JobID,
 		Location: location,
-		Bucket:   bucket,
 		Hash:     doc.Hash,
 	})
 	assert.NoError(t, err)
@@ -145,24 +146,24 @@ func TestGetCreateParams(t *testing.T) {
 		JobID: uuid.New(),
 		Hash:  "example_hash",
 	}
-	bucket := "eample_bucket"
-	location := "/i/am/here"
+	location := document.Location{
+		Bucket: "example_bucket",
+		Key:    "/i/am/here",
+	}
 
-	pool.ExpectQuery("-- name: GetDocumentIDByHash :one").WithArgs(doc.Hash, database.MustToDBUUID(doc.JobID)).WillReturnRows(
+	pool.ExpectQuery("name: GetDocumentIDByHash :one").WithArgs(doc.Hash, database.MustToDBUUID(doc.JobID)).WillReturnRows(
 		pgxmock.NewRows([]string{"id"}),
 	)
 
 	params, err := svc.getCreateParams(ctx, &Create{
 		JobID:    doc.JobID,
 		Location: location,
-		Bucket:   bucket,
 		Hash:     doc.Hash,
 	})
 	assert.NoError(t, err)
 	assert.Equal(t, &createDocumentParams{
 		JobID:    database.MustToDBUUID(doc.JobID),
 		Hash:     doc.Hash,
-		Bucket:   bucket,
 		Location: location,
 	}, params)
 }
@@ -191,10 +192,12 @@ func TestGetCreateParamsExisting(t *testing.T) {
 		JobID: uuid.New(),
 		Hash:  "example_hash",
 	}
-	bucket := "eample_bucket"
-	location := "/i/am/here"
+	location := document.Location{
+		Bucket: "example_bucket",
+		Key:    "/i/am/here",
+	}
 
-	pool.ExpectQuery("-- name: GetDocumentIDByHash :one").WithArgs(doc.Hash, database.MustToDBUUID(doc.JobID)).WillReturnRows(
+	pool.ExpectQuery("name: GetDocumentIDByHash :one").WithArgs(doc.Hash, database.MustToDBUUID(doc.JobID)).WillReturnRows(
 		pgxmock.NewRows([]string{"id"}).
 			AddRow(database.MustToDBUUID(doc.ID)),
 	)
@@ -202,7 +205,6 @@ func TestGetCreateParamsExisting(t *testing.T) {
 	params, err := svc.getCreateParams(ctx, &Create{
 		JobID:    doc.JobID,
 		Location: location,
-		Bucket:   bucket,
 		Hash:     doc.Hash,
 	})
 	assert.NoError(t, err)
@@ -211,7 +213,6 @@ func TestGetCreateParamsExisting(t *testing.T) {
 		ID:       &dbid,
 		JobID:    database.MustToDBUUID(doc.JobID),
 		Hash:     doc.Hash,
-		Bucket:   bucket,
 		Location: location,
 	}, params)
 }
@@ -239,16 +240,17 @@ func TestGetCreateParamsCurrentDocErr(t *testing.T) {
 		JobID: uuid.New(),
 		Hash:  "example_hash",
 	}
-	bucket := "eample_bucket"
-	location := "/i/am/here"
+	location := document.Location{
+		Bucket: "example_bucket",
+		Key:    "/i/am/here",
+	}
 
-	pool.ExpectQuery("-- name: GetDocumentIDByHash :one").WithArgs(doc.Hash, database.MustToDBUUID(doc.JobID)).
+	pool.ExpectQuery("name: GetDocumentIDByHash :one").WithArgs(doc.Hash, database.MustToDBUUID(doc.JobID)).
 		WillReturnError(errors.New("db err"))
 
 	_, err = svc.getCreateParams(ctx, &Create{
 		JobID:    doc.JobID,
 		Location: location,
-		Bucket:   bucket,
 	})
 	assert.Error(t, err)
 }
@@ -275,8 +277,10 @@ func TestSubmitCreate(t *testing.T) {
 		JobID: uuid.New(),
 		Hash:  "example_hash",
 	}
-	bucket := "eample_bucket"
-	location := "/i/am/here"
+	location := document.Location{
+		Bucket: "example_bucket",
+		Key:    "/i/am/here",
+	}
 
 	pool.ExpectBegin()
 	pool.ExpectQuery("name: CreateDocument :one").WithArgs(database.MustToDBUUID(doc.JobID), doc.Hash).
@@ -284,14 +288,13 @@ func TestSubmitCreate(t *testing.T) {
 			pgxmock.NewRows([]string{"id"}).
 				AddRow(database.MustToDBUUID(doc.ID)),
 		)
-	pool.ExpectExec("name: AddDocumentEntry :exec").WithArgs(database.MustToDBUUID(doc.ID), bucket, location).
+	pool.ExpectExec("name: AddDocumentEntry :exec").WithArgs(database.MustToDBUUID(doc.ID), location.Bucket, location.Key).
 		WillReturnResult(pgxmock.NewResult("", 1))
 	pool.ExpectCommit()
 
 	id, err := svc.submitCreate(ctx, &createDocumentParams{
 		JobID:    database.MustToDBUUID(doc.JobID),
 		Location: location,
-		Bucket:   bucket,
 		Hash:     doc.Hash,
 	})
 	assert.NoError(t, err)
@@ -320,11 +323,13 @@ func TestSubmitCreateExists(t *testing.T) {
 		JobID: uuid.New(),
 		Hash:  "example_hash",
 	}
-	bucket := "eample_bucket"
-	location := "/i/am/here"
+	location := document.Location{
+		Bucket: "example_bucket",
+		Key:    "/i/am/here",
+	}
 
 	pool.ExpectBegin()
-	pool.ExpectExec("name: AddDocumentEntry :exec").WithArgs(database.MustToDBUUID(doc.ID), bucket, location).
+	pool.ExpectExec("name: AddDocumentEntry :exec").WithArgs(database.MustToDBUUID(doc.ID), location.Bucket, location.Key).
 		WillReturnResult(pgxmock.NewResult("", 1))
 	pool.ExpectCommit()
 
@@ -333,7 +338,6 @@ func TestSubmitCreateExists(t *testing.T) {
 		ID:       &docid,
 		JobID:    database.MustToDBUUID(doc.JobID),
 		Location: location,
-		Bucket:   bucket,
 		Hash:     doc.Hash,
 	})
 	assert.NoError(t, err)

internal/document/service.go

@@ -6,7 +6,10 @@ import (
 	"github.com/google/uuid"
 )
 
-type Location = string
+type Location struct {
+	Bucket string
+	Key    string
+}
 
 type Document struct {
 	ID    uuid.UUID

internal/document/text/service.go

@@ -1,8 +1,6 @@
 package documenttext
 
 import (
-	"errors"
-
 	"github.com/google/uuid"
 )
 
@@ -13,12 +11,9 @@ func New() *Service {
 	return &Service{}
 }
 
-func (s *Service) IsValidVersion(v int32) error {
-	if v <= 0 {
-		return errors.New("document clean code version must be > 0")
-	}
-
-	return nil
+// TODO - move to api/
+type Create struct {
+	ID uuid.UUID `json:"id" validate:"required,uuid"`
 }
 
 type IsExtractedParams struct {

internal/document/text/service_test.go

@@ -13,13 +13,6 @@ func TestService(t *testing.T) {
 	assert.NotNil(t, svc)
 }
 
-func TestIsValidVersion(t *testing.T) {
-	svc := documenttext.New()
-
-	assert.Nil(t, svc.IsValidVersion(2))
-	assert.Error(t, svc.IsValidVersion(-1))
-}
-
 func TestIsExtracted(t *testing.T) {
 	svc := documenttext.New()
 

internal/document/version.go

@@ -0,0 +1,35 @@
+package document
+
+import (
+	"fmt"
+)
+
+func (s *Service) GetCleanVersion() int32 {
+	// TODO - actual version
+	return 1
+}
+
+func (s *Service) GetTextVersion() int32 {
+	// TODO - actual version
+	return 1
+}
+
+func (s *Service) IsValidCleanVersion(v int32) error {
+	currentVersion := s.GetCleanVersion()
+
+	if v < 1 || v > currentVersion {
+		return fmt.Errorf("document clean code version must be in the range 0 < version <= %d", currentVersion)
+	}
+
+	return nil
+}
+
+func (s *Service) IsValidTextVersion(v int32) error {
+	currentVersion := s.GetTextVersion()
+
+	if v < 1 || v > currentVersion {
+		return fmt.Errorf("document text code version must be in the range 0 < version <= %d", currentVersion)
+	}
+
+	return nil
+}

internal/document/version_test.go

@@ -0,0 +1,43 @@
+package document_test
+
+import (
+	"queryorchestration/internal/document"
+	"queryorchestration/internal/serviceconfig"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIsValidCleanVersion(t *testing.T) {
+	cfg := &serviceconfig.BaseConfig{}
+	svc := document.New(cfg)
+
+	assert.Nil(t, svc.IsValidCleanVersion(1))
+	assert.Error(t, svc.IsValidCleanVersion(2))
+	assert.Error(t, svc.IsValidCleanVersion(0))
+	assert.Error(t, svc.IsValidCleanVersion(-1))
+}
+
+func TestIsValidTextVersion(t *testing.T) {
+	cfg := &serviceconfig.BaseConfig{}
+	svc := document.New(cfg)
+
+	assert.Nil(t, svc.IsValidTextVersion(1))
+	assert.Error(t, svc.IsValidTextVersion(2))
+	assert.Error(t, svc.IsValidTextVersion(0))
+	assert.Error(t, svc.IsValidTextVersion(-1))
+}
+
+func TestGetCleanVersion(t *testing.T) {
+	cfg := &serviceconfig.BaseConfig{}
+	svc := document.New(cfg)
+
+	assert.Equal(t, int32(1), svc.GetCleanVersion())
+}
+
+func TestGetTextVersion(t *testing.T) {
+	cfg := &serviceconfig.BaseConfig{}
+	svc := document.New(cfg)
+
+	assert.Equal(t, int32(1), svc.GetTextVersion())
+}

internal/job/collector/create.go

@@ -41,7 +41,7 @@ type dbCreateParams struct {
 func (s *Service) getCreateParams(ctx context.Context, params *CreateParams) (*dbCreateParams, error) {
 	minClean := params.MinCleanVersion
 	if minClean != nil {
-		err := s.svc.Clean.IsValidVersion(*minClean)
+		err := s.svc.Document.IsValidCleanVersion(*minClean)
 		if err != nil {
 			return nil, err
 		}
@@ -49,7 +49,7 @@ func (s *Service) getCreateParams(ctx context.Context, params *CreateParams) (*d
 
 	minText := params.MinTextVersion
 	if minText != nil {
-		err := s.svc.Text.IsValidVersion(*minText)
+		err := s.svc.Document.IsValidTextVersion(*minText)
 		if err != nil {
 			return nil, err
 		}

internal/job/collector/create_test.go

@@ -28,8 +28,8 @@ func TestCreate(t *testing.T) {
 	svc := collector.New(cfg, &collector.Services{})
 
 	id := uuid.New()
-	minCleanV := int32(2)
-	minTextV := int32(4)
+	minCleanV := int32(1)
+	minTextV := int32(1)
 	create := collector.CreateParams{
 		JobID:           uuid.New(),
 		MinCleanVersion: &minCleanV,

internal/job/collector/createprivate_test.go

@@ -30,8 +30,8 @@ func TestGetCreateParams(t *testing.T) {
 		svc: &Services{},
 	}
 
-	minCleanV := int32(2)
-	minTextV := int32(4)
+	minCleanV := int32(1)
+	minTextV := int32(1)
 	params := CreateParams{
 		JobID:           uuid.New(),
 		MinCleanVersion: &minCleanV,

internal/job/collector/service.go

@@ -1,8 +1,7 @@
 package collector
 
 import (
-	documentclean "queryorchestration/internal/document/clean"
-	documenttext "queryorchestration/internal/document/text"
+	"queryorchestration/internal/document"
 	"queryorchestration/internal/serviceconfig"
 
 	"github.com/google/uuid"
@@ -19,8 +18,7 @@ type Collector struct {
 }
 
 type Services struct {
-	Clean *documentclean.Service
-	Text  *documenttext.Service
+	Document *document.Service
 }
 
 type Service struct {

internal/job/collector/update.go

@@ -97,7 +97,7 @@ func (s *Service) normalizeCodeVersions(current *Collector, params *UpdateParams
 	if params.MinCleanVersion == nil {
 		params.MinCleanVersion = &current.MinCleanVersion
 	} else {
-		err := s.svc.Clean.IsValidVersion(*params.MinCleanVersion)
+		err := s.svc.Document.IsValidCleanVersion(*params.MinCleanVersion)
 		if err != nil {
 			return err
 		}
@@ -106,7 +106,7 @@ func (s *Service) normalizeCodeVersions(current *Collector, params *UpdateParams
 	if params.MinTextVersion == nil {
 		params.MinTextVersion = &current.MinTextVersion
 	} else {
-		err := s.svc.Text.IsValidVersion(*params.MinTextVersion)
+		err := s.svc.Document.IsValidTextVersion(*params.MinTextVersion)
 		if err != nil {
 			return err
 		}

internal/job/collector/update_test.go

@@ -4,8 +4,7 @@ import (
 	"context"
 	"queryorchestration/internal/database"
 	"queryorchestration/internal/database/repository"
-	documentclean "queryorchestration/internal/document/clean"
-	documenttext "queryorchestration/internal/document/text"
+	"queryorchestration/internal/document"
 	"queryorchestration/internal/job/collector"
 	"queryorchestration/internal/serviceconfig"
 	"testing"
@@ -28,8 +27,7 @@ func TestUpdate(t *testing.T) {
 	cfg.DBQueries = repository.New(pool)
 
 	svc := collector.New(cfg, &collector.Services{
-		Clean: documentclean.New(),
-		Text:  documenttext.New(),
+		Document: document.New(cfg),
 	})
 
 	current := collector.Collector{

internal/job/collector/updateprivate_test.go

@@ -4,8 +4,7 @@ import (
 	"context"
 	"queryorchestration/internal/database"
 	"queryorchestration/internal/database/repository"
-	documentclean "queryorchestration/internal/document/clean"
-	documenttext "queryorchestration/internal/document/text"
+	"queryorchestration/internal/document"
 	"queryorchestration/internal/serviceconfig"
 	"testing"
 
@@ -32,8 +31,8 @@ func TestGetUpdateParams(t *testing.T) {
 		svc: &Services{},
 	}
 
-	minCleanV := int32(2)
-	minTextV := int32(4)
+	minCleanV := int32(1)
+	minTextV := int32(1)
 	aV := int32(3)
 	current := Collector{
 		ActiveVersion: 1,
@@ -225,10 +224,10 @@ func TestNormalizeActiveVersion(t *testing.T) {
 }
 
 func TestNormalizeCodeVersions(t *testing.T) {
+	cfg := &serviceconfig.BaseConfig{}
 	svc := Service{
 		svc: &Services{
-			Clean: documentclean.New(),
-			Text:  documenttext.New(),
+			Document: document.New(cfg),
 		},
 	}
 
@@ -247,7 +246,7 @@ func TestNormalizeCodeVersions(t *testing.T) {
 	assert.Nil(t, update.MinCleanVersion)
 	assert.Nil(t, update.MinTextVersion)
 
-	cv := int32(2)
+	cv := int32(1)
 	update.MinCleanVersion = &cv
 	update.MinTextVersion = nil
 	err = svc.normalizeCodeVersions(&current, &update)
@@ -256,15 +255,15 @@ func TestNormalizeCodeVersions(t *testing.T) {
 	assert.Equal(t, int32(0), *update.MinTextVersion)
 
 	update.MinCleanVersion = nil
-	tv := int32(2)
+	tv := int32(1)
 	update.MinTextVersion = &tv
 	err = svc.normalizeCodeVersions(&current, &update)
 	assert.NoError(t, err)
 	assert.Equal(t, int32(0), *update.MinCleanVersion)
 	assert.Equal(t, tv, *update.MinTextVersion)
 
-	current.MinCleanVersion = 2
-	current.MinTextVersion = 2
+	current.MinCleanVersion = 1
+	current.MinTextVersion = 1
 	err = svc.normalizeCodeVersions(&current, &update)
 	assert.NoError(t, err)
 	assert.Nil(t, update.MinCleanVersion)
@@ -290,23 +289,6 @@ func TestNormalizeCodeVersions(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Nil(t, update.MinCleanVersion)
 	assert.Nil(t, update.MinTextVersion)
-
-	cv = current.MinCleanVersion + 1
-	update.MinCleanVersion = &cv
-	update.MinTextVersion = &current.MinTextVersion
-	err = svc.normalizeCodeVersions(&current, &update)
-	assert.NoError(t, err)
-	assert.Equal(t, cv, *update.MinCleanVersion)
-	assert.Equal(t, current.MinTextVersion, *update.MinTextVersion)
-
-	cv = current.MinCleanVersion + 1
-	update.MinCleanVersion = &cv
-	tv = current.MinTextVersion + 1
-	update.MinTextVersion = &tv
-	err = svc.normalizeCodeVersions(&current, &update)
-	assert.NoError(t, err)
-	assert.Equal(t, cv, *update.MinCleanVersion)
-	assert.Equal(t, tv, *update.MinTextVersion)
 }
 
 func TestGetRemoveFields(t *testing.T) {

internal/query/test_test.go

@@ -5,7 +5,6 @@ import (
 	"queryorchestration/internal/database"
 	"queryorchestration/internal/database/repository"
 	"queryorchestration/internal/document"
-	documentclean "queryorchestration/internal/document/clean"
 	documenttext "queryorchestration/internal/document/text"
 	"queryorchestration/internal/job/collector"
 	"queryorchestration/internal/query"
@@ -29,13 +28,11 @@ func TestTest(t *testing.T) {
 	cfg := &serviceconfig.BaseConfig{}
 	cfg.DBPool = pool
 	cfg.DBQueries = repository.New(pool)
-	text := documenttext.New()
-	clean := documentclean.New()
-	col := collector.New(cfg, &collector.Services{
-		Text:  text,
-		Clean: clean,
-	})
 	docsvc := document.New(cfg)
+	col := collector.New(cfg, &collector.Services{
+		Document: docsvc,
+	})
+	text := documenttext.New()
 	svc := query.New(cfg, &query.Services{
 		Text:      text,
 		Document:  docsvc,

internal/server/server_test.go

@@ -9,6 +9,7 @@ import (
 	"queryorchestration/internal/test"
 	"testing"
 
+	"github.com/go-playground/validator/v10"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -36,3 +37,21 @@ func TestNew(t *testing.T) {
 		}
 	}()
 }
+
+func TestSetValidator(t *testing.T) {
+	cfg := server.BaseConfig{}
+
+	assert.Nil(t, cfg.Validator)
+
+	cfg.SetValidator()
+	assert.NotNil(t, cfg.Validator)
+}
+
+func TestGetValidator(t *testing.T) {
+	cfg := server.BaseConfig{}
+
+	assert.Nil(t, cfg.GetValidator())
+	cfg.Validator = validator.New()
+	assert.NotNil(t, cfg.GetValidator())
+	assert.EqualExportedValues(t, validator.New(), cfg.GetValidator())
+}

internal/serviceconfig/queue/documenttext/config.go

@@ -0,0 +1,13 @@
+package documenttext
+
+type DocTextConfig struct {
+	DocumentTextURL string `env:"DOCUMENT_TEXT_URL,required,notEmpty"`
+}
+
+func (c *DocTextConfig) GetDocumentTextURL() string {
+	return c.DocumentTextURL
+}
+
+type ConfigProvider interface {
+	GetDocumentTextURL() string
+}

internal/serviceconfig/queue/documenttext/config_test.go

@@ -0,0 +1,20 @@
+package documenttext_test
+
+import (
+	"queryorchestration/internal/serviceconfig/queue/documenttext"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGetDocumentTextURL(t *testing.T) {
+	cfg := documenttext.DocTextConfig{}
+
+	name := cfg.GetDocumentTextURL()
+	assert.Equal(t, "", name)
+
+	cfg.DocumentTextURL = "name"
+	name = cfg.GetDocumentTextURL()
+	assert.Equal(t, "name", name)
+	assert.Equal(t, cfg.DocumentTextURL, name)
+}

internal/test/runner.go

@@ -14,8 +14,9 @@ import (
 type Runner = string
 
 const (
-	QueryRunner   = queryrunner.Name
-	DocInitRunner = docinitrunner.Name
+	DocInitRunner  = docinitrunner.Name
+	DocCleanRunner = docinitrunner.Name
+	QueryRunner    = queryrunner.Name
 )
 
 type RunnerConfig struct {