Merged in feature/jobsync (pull request #63)

Document Sync and Job Sync

* docsyncfunc

* startedQueryWork

* morefixes

* jobsyncsvcstart

* save

* jobsynctext

* save

* save

* docsync

* shorttest

* jobsync

* jobsyncupdateoncollectorupdate

* passlivetest

* collectortest

* lint

api/queryService/controllers.go

@@ -5,6 +5,7 @@ import (
 	"queryorchestration/internal/export"
 	"queryorchestration/internal/job"
 	"queryorchestration/internal/job/collector"
+	collectorupdate "queryorchestration/internal/job/collector/update"
 	"queryorchestration/internal/query"
 	querytest "queryorchestration/internal/query/test"
 
@@ -14,12 +15,13 @@ import (
 const Name = "queryService"
 
 type Services struct {
-	Export    *export.Service
-	Collector *collector.Service
-	Query     *query.Service
-	Client    *client.Service
-	Job       *job.Service
-	QueryTest *querytest.Service
+	Export          *export.Service
+	Collector       *collector.Service
+	CollectorUpdate *collectorupdate.Service
+	Query           *query.Service
+	Client          *client.Service
+	Job             *job.Service
+	QueryTest       *querytest.Service
 }
 
 type Controllers struct {

api/queryService/job_test.go

@@ -9,7 +9,8 @@ import (
 	"queryorchestration/internal/client"
 	"queryorchestration/internal/database"
 	"queryorchestration/internal/database/repository"
-	"queryorchestration/internal/document"
+	cleanversion "queryorchestration/internal/document/clean/version"
+	textversion "queryorchestration/internal/document/text/version"
 	"queryorchestration/internal/job"
 	"queryorchestration/internal/job/collector"
 	"queryorchestration/internal/serviceconfig"
@@ -37,7 +38,8 @@ func TestCreateJob(t *testing.T) {
 	cons := queryservice.NewControllers(validator.New(), &queryservice.Services{
 		Job: job.New(cfg, &job.Services{
 			Collector: collector.New(cfg, &collector.Services{
-				Document: document.New(cfg),
+				CleanVersion: cleanversion.New(cfg),
+				TextVersion:  textversion.New(cfg),
 			}),
 		}),
 	})

api/queryService/jobcollector.go

@@ -3,7 +3,7 @@ package queryservice
 import (
 	"fmt"
 	"net/http"
-	"queryorchestration/internal/job/collector"
+	collectorupdate "queryorchestration/internal/job/collector/update"
 
 	"github.com/google/uuid"
 	"github.com/labstack/echo/v4"
@@ -51,7 +51,7 @@ func (s *Controllers) UpdateJobCollectorByJobId(ctx echo.Context, jobId types.UU
 		fields = &fs
 	}
 
-	err := s.svc.Collector.UpdateByJobId(ctx.Request().Context(), &collector.UpdateParams{
+	err := s.svc.CollectorUpdate.UpdateByJobId(ctx.Request().Context(), &collectorupdate.UpdateParams{
 		JobID:           jobId,
 		ActiveVersion:   req.ActiveVersion,
 		MinCleanVersion: req.MinimumCleanerVersion,

api/queryService/jobcollector_test.go

@@ -2,17 +2,23 @@ package queryservice_test
 
 import (
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	queryservice "queryorchestration/api/queryService"
 	"queryorchestration/internal/database"
 	"queryorchestration/internal/database/repository"
-	"queryorchestration/internal/document"
+	cleanversion "queryorchestration/internal/document/clean/version"
+	textversion "queryorchestration/internal/document/text/version"
 	"queryorchestration/internal/job/collector"
+	collectorupdate "queryorchestration/internal/job/collector/update"
 	"queryorchestration/internal/serviceconfig"
+	"queryorchestration/internal/serviceconfig/queue/jobsync"
+	queuemock "queryorchestration/mocks/queue"
 	"strings"
 	"testing"
 
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
 	"github.com/go-playground/validator/v10"
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5"
@@ -20,6 +26,7 @@ import (
 	"github.com/labstack/echo/v4"
 	"github.com/pashagolub/pgxmock/v3"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 )
 
 func TestUpdateJobCollector(t *testing.T) {
@@ -27,9 +34,14 @@ func TestUpdateJobCollector(t *testing.T) {
 	if err != nil {
 		t.Fatalf("failed to open pgxmock database: %v", err)
 	}
-	cfg := &serviceconfig.BaseConfig{}
+	cfg := &struct {
+		serviceconfig.BaseConfig
+		jobsync.JobSyncConfig
+	}{}
 	cfg.DBPool = pool
 	cfg.DBQueries = repository.New(pool)
+	mockSQS := queuemock.NewMockSQSClient(t)
+	cfg.QueueClient = mockSQS
 
 	current := collector.Collector{
 		ID:            uuid.New(),
@@ -39,8 +51,10 @@ func TestUpdateJobCollector(t *testing.T) {
 	}
 
 	av := int32(2)
+	cv := int32(1)
 	body := queryservice.JobCollectorUpdate{
-		ActiveVersion: &av,
+		ActiveVersion:         &av,
+		MinimumCleanerVersion: &cv,
 		Fields: &[]queryservice.JobCollectorField{
 			{
 				Name:    "a",
@@ -58,8 +72,11 @@ func TestUpdateJobCollector(t *testing.T) {
 	ctx := e.NewContext(req, rec)
 
 	cons := queryservice.NewControllers(validator.New(), &queryservice.Services{
-		Collector: collector.New(cfg, &collector.Services{
-			Document: document.New(cfg),
+		CollectorUpdate: collectorupdate.New(cfg, &collectorupdate.Services{
+			Collector: collector.New(cfg, &collector.Services{
+				CleanVersion: cleanversion.New(cfg),
+				TextVersion:  textversion.New(cfg),
+			}),
 		}),
 	})
 
@@ -76,12 +93,27 @@ func TestUpdateJobCollector(t *testing.T) {
 				AddRow(true),
 		)
 	pool.ExpectBeginTx(pgx.TxOptions{})
+	rv := current.LatestVersion + 1
+	pool.ExpectExec("name: RemoveCollectorCodeVersion :exec").WithArgs(&rv, database.MustToDBUUID(current.ID)).
+		WillReturnResult(pgxmock.NewResult("", 1))
+	pool.ExpectExec("name: AddCollectorCodeVersion :exec").WithArgs(database.MustToDBUUID(current.ID), int32(5), *body.MinimumCleanerVersion, int32(0)).
+		WillReturnResult(pgxmock.NewResult("", 1))
 	pool.ExpectExec("name: AddCollectorQuery :exec").WithArgs(database.MustToDBUUID(current.ID), "a", database.MustToDBUUID((*body.Fields)[0].QueryId), int32(5)).
 		WillReturnResult(pgxmock.NewResult("", 1))
 	pool.ExpectExec("name: UpdateCollector :exec").WithArgs(int32(5), int32(2), database.MustToDBUUID(current.ID)).
 		WillReturnResult(pgxmock.NewResult("", 1))
 	pool.ExpectCommit()
 
+	mockSQS.EXPECT().
+		SendMessage(
+			mock.Anything,
+			mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
+				return *in.QueueUrl == cfg.JobSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", current.JobID.String())
+			}),
+			mock.Anything,
+		).
+		Return(&sqs.SendMessageOutput{}, nil)
+
 	err = cons.UpdateJobCollectorByJobId(ctx, current.JobID)
 	assert.NoError(t, err)
 	assert.Equal(t, http.StatusOK, rec.Code)
@@ -103,7 +135,8 @@ func TestGetJobCollectorByJobId(t *testing.T) {
 	ctx := e.NewContext(req, rec)
 
 	svc := collector.New(cfg, &collector.Services{
-		Document: document.New(cfg),
+		CleanVersion: cleanversion.New(cfg),
+		TextVersion:  textversion.New(cfg),
 	})
 	cons := queryservice.NewControllers(validator.New(), &queryservice.Services{
 		Collector: svc,

api/queryService/query_test.go

@@ -9,6 +9,9 @@ import (
 	"queryorchestration/internal/database"
 	"queryorchestration/internal/database/repository"
 	"queryorchestration/internal/document"
+	cleanversion "queryorchestration/internal/document/clean/version"
+	textversion "queryorchestration/internal/document/text/version"
+	"queryorchestration/internal/job"
 	"queryorchestration/internal/job/collector"
 	"queryorchestration/internal/query"
 	"queryorchestration/internal/query/result"
@@ -208,7 +211,8 @@ func TestTestQuery(t *testing.T) {
 
 	docsvc := document.New(cfg)
 	col := collector.New(cfg, &collector.Services{
-		Document: docsvc,
+		CleanVersion: cleanversion.New(cfg),
+		TextVersion:  textversion.New(cfg),
 	})
 	que := query.New(cfg)
 	cons := queryservice.NewControllers(validator.New(), &queryservice.Services{
@@ -223,9 +227,12 @@ func TestTestQuery(t *testing.T) {
 		}),
 	})
 
+	j := job.Job{
+		ID: uuid.New(),
+	}
 	coll := collector.Collector{
 		ID:    uuid.New(),
-		JobID: uuid.New(),
+		JobID: j.ID,
 	}
 	doc := document.Document{
 		ID:    uuid.New(),