Merged in feature/doctextstructure (pull request #56)

DocTextRunner Structure

* firstround

* shorttests

api/docCleanRunner/runner_test.go

@@ -25,7 +25,7 @@ import (
 	"github.com/stretchr/testify/mock"
 )
 
-type DocTextConfig struct {
+type DocCleanConfig struct {
 	serviceconfig.BaseConfig
 	documenttext.DocTextConfig
 	objectstore.ObjectStoreConfig
@@ -39,7 +39,7 @@ func TestDocCleanRunner(t *testing.T) {
 		t.Fatalf("failed to open pgxmock database: %v", err)
 	}
 
-	cfg := &DocTextConfig{}
+	cfg := &DocCleanConfig{}
 	cfg.DBPool = pool
 	cfg.DBQueries = repository.New(pool)
 	mockStore := objectstoremock.NewMockS3Client(t)

api/docTextRunner/runner.go

@@ -0,0 +1,54 @@
+package doctextrunner
+
+import (
+	"context"
+	"encoding/json"
+	documenttext "queryorchestration/internal/document/text"
+
+	"github.com/go-playground/validator/v10"
+	"github.com/google/uuid"
+
+	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
+)
+
+const Name = "docTextRunner"
+
+type Services struct {
+	Text *documenttext.Service
+}
+
+type Runner struct {
+	validator *validator.Validate
+	svc       *Services
+}
+
+func New(validator *validator.Validate, svc *Services) Runner {
+	return Runner{
+		validator: validator,
+		svc:       svc,
+	}
+}
+
+type Create struct {
+	ID uuid.UUID `json:"id" validate:"required,uuid"`
+}
+
+func (s Runner) Process(ctx context.Context, req *types.Message) error {
+	var body Create
+	err := json.Unmarshal([]byte(*req.Body), &body)
+	if err != nil {
+		return err
+	}
+
+	err = s.validator.Struct(body)
+	if err != nil {
+		return err
+	}
+
+	err = s.svc.Text.Create(ctx, body.ID)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

api/docTextRunner/runner_test.go

@@ -0,0 +1,96 @@
+package doctextrunner_test
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	doctextrunner "queryorchestration/api/docTextRunner"
+	"queryorchestration/internal/database"
+	"queryorchestration/internal/database/repository"
+	"queryorchestration/internal/document"
+	documenttext "queryorchestration/internal/document/text"
+	"queryorchestration/internal/serviceconfig"
+	"queryorchestration/internal/serviceconfig/objectstore"
+	"queryorchestration/internal/serviceconfig/queue/querysync"
+	objectstoremock "queryorchestration/mocks/objectstore"
+	queuemock "queryorchestration/mocks/queue"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
+	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
+	"github.com/go-playground/validator/v10"
+	"github.com/google/uuid"
+	"github.com/pashagolub/pgxmock/v3"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+)
+
+type DocTextConfig struct {
+	serviceconfig.BaseConfig
+	querysync.QuerySyncConfig
+	objectstore.ObjectStoreConfig
+}
+
+func TestDocCleanRunner(t *testing.T) {
+	ctx := context.Background()
+
+	pool, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to open pgxmock database: %v", err)
+	}
+
+	cfg := &DocTextConfig{}
+	cfg.DBPool = pool
+	cfg.DBQueries = repository.New(pool)
+	mockStore := objectstoremock.NewMockS3Client(t)
+	cfg.StoreClient = mockStore
+	mockSQS := queuemock.NewMockSQSClient(t)
+	cfg.QueueClient = mockSQS
+	cfg.QuerySyncURL = "/i/am/here"
+
+	runner := doctextrunner.New(validator.New(), &doctextrunner.Services{
+		Text: documenttext.New(cfg, &documenttext.Services{
+			Document: document.New(cfg),
+		}),
+	})
+	assert.NotNil(t, runner)
+
+	doc := doctextrunner.Create{
+		ID: uuid.New(),
+	}
+	bodyBytes, err := json.Marshal(doc)
+	assert.NoError(t, err)
+	body := string(bodyBytes)
+	msg := &types.Message{
+		Body: &body,
+	}
+
+	inloc := document.Location{
+		Bucket: "bucket_name",
+		Key:    "/i/am/here",
+	}
+
+	pool.ExpectQuery("name: IsDocumentTextExtracted :one").WithArgs(database.MustToDBUUID(doc.ID)).WillReturnRows(
+		pgxmock.NewRows([]string{"isextracted"}).
+			AddRow(false),
+	)
+	pool.ExpectQuery("name: GetDocumentCleanEntry :one").WithArgs(database.MustToDBUUID(doc.ID)).WillReturnRows(
+		pgxmock.NewRows([]string{"documentId", "bucket", "key"}).
+			AddRow(database.MustToDBUUID(doc.ID), inloc.Bucket, inloc.Key),
+	)
+	pool.ExpectExec("name: AddDocumentTextEntry :exec").WithArgs(database.MustToDBUUID(doc.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.QuerySyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", doc.ID.String())
+			}),
+			mock.Anything,
+		).
+		Return(&sqs.SendMessageOutput{}, nil)
+
+	err = runner.Process(ctx, msg)
+	assert.NoError(t, err)
+}

api/queryRunner/runner_test.go

@@ -6,7 +6,7 @@ import (
 	controllers "queryorchestration/api/queryRunner"
 	"queryorchestration/internal/database"
 	"queryorchestration/internal/database/repository"
-	documenttext "queryorchestration/internal/document/text"
+	"queryorchestration/internal/document"
 	"queryorchestration/internal/job/collector"
 	"queryorchestration/internal/query"
 	"queryorchestration/internal/query/result"
@@ -34,8 +34,8 @@ func TestQueryRunner(t *testing.T) {
 
 	svc := query.New(cfg, &query.Services{
 		Result:    result.New(cfg),
-		Text:      documenttext.New(),
 		Collector: collector.New(cfg, &collector.Services{}),
+		Document:  document.New(cfg),
 	})
 
 	runner := controllers.New(validator.New(), &controllers.Services{

api/querySyncRunner/runner.go

@@ -0,0 +1,11 @@
+package querysyncrunner
+
+import (
+	"github.com/google/uuid"
+)
+
+const Name = "querySyncRunner"
+
+type Create struct {
+	ID uuid.UUID `json:"id" validate:"required,uuid"`
+}

cmd/docTextRunner/main.go

@@ -0,0 +1,46 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+	"os"
+	doctextrunner "queryorchestration/api/docTextRunner"
+	"queryorchestration/internal/document"
+	documenttext "queryorchestration/internal/document/text"
+	"queryorchestration/internal/server/runner"
+	"queryorchestration/internal/serviceconfig/objectstore"
+	"queryorchestration/internal/serviceconfig/queue/querysync"
+
+	_ "github.com/lib/pq"
+)
+
+type DocTextConfig struct {
+	runner.BaseConfig
+	objectstore.ObjectStoreConfig
+	querysync.QuerySyncConfig
+}
+
+func main() {
+	ctx := context.Background()
+
+	cfg := &DocTextConfig{}
+
+	cfg.ControllerFunc = func() runner.Controller {
+		doc := document.New(cfg)
+		text := documenttext.New(cfg, &documenttext.Services{
+			Document: doc,
+		})
+
+		return doctextrunner.New(cfg.GetValidator(), &doctextrunner.Services{
+			Text: text,
+		})
+	}
+
+	server, err := runner.New(ctx, cfg)
+	if err != nil {
+		slog.Error(err.Error())
+		os.Exit(1)
+	}
+
+	server.Listen(ctx)
+}

cmd/queryRunner/main.go

@@ -6,7 +6,6 @@ import (
 	"os"
 	controllers "queryorchestration/api/queryRunner"
 	"queryorchestration/internal/document"
-	documenttext "queryorchestration/internal/document/text"
 	"queryorchestration/internal/job/collector"
 	"queryorchestration/internal/query"
 	"queryorchestration/internal/query/result"
@@ -21,7 +20,6 @@ func main() {
 	cfg := &runner.BaseConfig{}
 
 	cfg.ControllerFunc = func() runner.Controller {
-		text := documenttext.New()
 		doc := document.New(cfg)
 		col := collector.New(cfg, &collector.Services{
 			Document: doc,
@@ -29,7 +27,6 @@ func main() {
 		res := result.New(cfg)
 		svc := query.New(cfg, &query.Services{
 			Result:    res,
-			Text:      text,
 			Collector: col,
 			Document:  doc,
 		})

cmd/queryService/main.go

@@ -8,7 +8,6 @@ import (
 	queryservice "queryorchestration/api/queryService"
 	"queryorchestration/internal/client"
 	"queryorchestration/internal/document"
-	documenttext "queryorchestration/internal/document/text"
 	"queryorchestration/internal/export"
 	"queryorchestration/internal/job"
 	"queryorchestration/internal/job/collector"
@@ -27,14 +26,12 @@ func main() {
 
 	cfg.RegisterHandlersFunc = func() (*openapi3.T, error) {
 		exp := export.New()
-		extract := documenttext.New()
 		res := result.New(cfg)
 		doc := document.New(cfg)
 		col := collector.New(cfg, &collector.Services{
 			Document: doc,
 		})
 		que := query.New(cfg, &query.Services{
-			Text:      extract,
 			Result:    res,
 			Collector: col,
 			Document:  doc,

database/migrations/00000000000006_documents.up.sql

@@ -22,3 +22,12 @@ CREATE TABLE documentCleans (
     key text not null,
     foreign key (documentId) references documents(id)
 );
+
+CREATE TABLE documentTextExtractions (
+    id uuid primary key DEFAULT uuid_generate_v7(),
+    documentId uuid not null,
+    version int not null,
+    bucket text not null,
+    key text not null,
+    foreign key (documentId) references documents(id)
+);

database/queries/clean.sql

@@ -9,4 +9,13 @@ SELECT EXISTS(
 );
 
 -- name: AddDocumentCleanEntry :exec
-INSERT INTO documentCleans (documentId, version, bucket, key) VALUES ($1, $2, $3, $4);
+INSERT INTO documentCleans (documentId, version, bucket, key) VALUES ($1, $2, $3, $4);
+
+-- name: GetDocumentCleanEntry :one
+SELECT dc.documentId, dc.bucket, dc.key
+    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
+    ORDER BY d.id DESC LIMIT 1;

database/queries/text.sql

@@ -0,0 +1,21 @@
+-- name: IsDocumentTextExtracted :one
+SELECT EXISTS(
+    SELECT 1
+        FROM documentTextExtractions 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.minTextVersion
+);
+
+-- name: AddDocumentTextEntry :exec
+INSERT INTO documentTextExtractions (documentId, version, bucket, key) VALUES ($1, $2, $3, $4);
+
+-- name: GetDocumentTextEntry :one
+SELECT dc.documentId, dc.bucket, dc.key
+    FROM documentTextExtractions 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.minTextVersion
+    ORDER BY d.id DESC LIMIT 1;

deployments/compose.local.yaml

@@ -46,6 +46,29 @@ services:
       AWS_S3_USE_PATH_STYLE: true
     networks:
       - server-network
+  doc_text_runner:
+    image: queryorchestration:latest
+    command: ["./docTextRunner"]
+    depends_on:
+      - db
+      - localstack
+    environment:
+      QUEUE_URL: ${DOCUMENT_TEXT_URL}
+      QUERY_SYNC_URL: ${QUERY_SYNC_URL}
+      AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID}
+      AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY}
+      AWS_SESSION_TOKEN: ${AWS_SESSION_TOKEN}
+      AWS_REGION: ${AWS_REGION}
+      DB_USER: ${DB_USER}
+      DB_PASS: ${DB_PASS}
+      DB_HOST: db
+      DB_PORT: 5432
+      DB_NAME: ${DB_NAME}
+      DB_NOSSL: ${DB_NOSSL}
+      AWS_ENDPOINT_URL: "http://localstack:4566"
+      AWS_S3_USE_PATH_STYLE: true
+    networks:
+      - server-network
   query_runner:
     image: queryorchestration:latest
     command: ["./queryRunner"]

devbox.json

@@ -43,10 +43,12 @@
     "QNAME_DOCUMENT_INIT":   "document_init",
     "QNAME_DOCUMENT_CLEAN":  "document_clean",
     "QNAME_DOCUMENT_TEXT":  "document_text",
+    "QNAME_QUERY_SYNC":    "query_sync",
     "QNAME_QUERY_RUNNER":    "query_runner",
     "DOCUMENT_INIT_URL": "http://localstack:4566/queue/us-east-1/000000000000/document_init",
     "DOCUMENT_CLEAN_URL": "http://localstack:4566/queue/us-east-1/000000000000/document_clean",
     "DOCUMENT_TEXT_URL": "http://localstack:4566/queue/us-east-1/000000000000/document_text",
+    "QUERY_SYNC_URL": "http://localstack:4566/queue/us-east-1/000000000000/query_sync",
     "QUERY_RUNNER_URL": "http://localstack:4566/queue/us-east-1/000000000000/query_runner",
     "BUCKET_IN":             "documentin"
   },

internal/database/repository/clean.sql.go

@@ -35,6 +35,38 @@ func (q *Queries) AddDocumentCleanEntry(ctx context.Context, arg *AddDocumentCle
 	return err
 }
 
+const getDocumentCleanEntry = `-- name: GetDocumentCleanEntry :one
+SELECT dc.documentId, dc.bucket, dc.key
+    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
+    ORDER BY d.id DESC LIMIT 1
+`
+
+type GetDocumentCleanEntryRow struct {
+	Documentid pgtype.UUID `db:"documentid"`
+	Bucket     string      `db:"bucket"`
+	Key        string      `db:"key"`
+}
+
+// GetDocumentCleanEntry
+//
+//	SELECT dc.documentId, dc.bucket, dc.key
+//	    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
+//	    ORDER BY d.id DESC LIMIT 1
+func (q *Queries) GetDocumentCleanEntry(ctx context.Context, documentid pgtype.UUID) (*GetDocumentCleanEntryRow, error) {
+	row := q.db.QueryRow(ctx, getDocumentCleanEntry, documentid)
+	var i GetDocumentCleanEntryRow
+	err := row.Scan(&i.Documentid, &i.Bucket, &i.Key)
+	return &i, err
+}
+
 const isDocumentClean = `-- name: IsDocumentClean :one
 SELECT EXISTS(
     SELECT 1

internal/database/repository/clean_test.go

@@ -13,6 +13,9 @@ import (
 )
 
 func TestClean(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping long test in short mode")
+	}
 	ctx := context.Background()
 
 	cfg := &serviceconfig.BaseConfig{}
@@ -53,9 +56,13 @@ func TestClean(t *testing.T) {
 	assert.NoError(t, err)
 	assert.False(t, isclean)
 
+	bucket := "example_bucket"
+	key := "example_key"
 	err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{
 		Documentid: id,
 		Version:    1,
+		Bucket:     bucket,
+		Key:        key,
 	})
 	assert.NoError(t, err)
 
@@ -74,4 +81,12 @@ func TestClean(t *testing.T) {
 	isclean, err = queries.IsDocumentClean(ctx, id)
 	assert.NoError(t, err)
 	assert.True(t, isclean)
+
+	clean, err := queries.GetDocumentCleanEntry(ctx, id)
+	assert.NoError(t, err)
+	assert.EqualExportedValues(t, &repository.GetDocumentCleanEntryRow{
+		Documentid: id,
+		Bucket:     bucket,
+		Key:        key,
+	}, clean)
 }

internal/database/repository/client_test.go

@@ -13,6 +13,9 @@ import (
 )
 
 func TestClient(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping long test in short mode")
+	}
 	ctx := context.Background()
 
 	cfg := &serviceconfig.BaseConfig{}

internal/database/repository/collector_test.go

@@ -16,6 +16,9 @@ import (
 )
 
 func TestCollector(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping long test in short mode")
+	}
 	ctx := context.Background()
 
 	cfg := &serviceconfig.BaseConfig{}

internal/database/repository/document_test.go

@@ -13,6 +13,9 @@ import (
 )
 
 func TestDocument(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping long test in short mode")
+	}
 	ctx := context.Background()
 
 	cfg := &serviceconfig.BaseConfig{}

internal/database/repository/job_test.go

@@ -13,6 +13,9 @@ import (
 )
 
 func TestJob(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping long test in short mode")
+	}
 	ctx := context.Background()
 
 	cfg := &serviceconfig.BaseConfig{}

internal/database/repository/models.go

@@ -128,6 +128,14 @@ type Documententry struct {
 	Key        string      `db:"key"`
 }
 
+type Documenttextextraction struct {
+	ID         pgtype.UUID `db:"id"`
+	Documentid pgtype.UUID `db:"documentid"`
+	Version    int32       `db:"version"`
+	Bucket     string      `db:"bucket"`
+	Key        string      `db:"key"`
+}
+
 type Fullactivecollector struct {
 	ID              pgtype.UUID `db:"id"`
 	Jobid           pgtype.UUID `db:"jobid"`

internal/database/repository/query_test.go

@@ -16,6 +16,9 @@ import (
 )
 
 func TestQueries(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping long test in short mode")
+	}
 	ctx := context.Background()
 
 	cfg := &serviceconfig.BaseConfig{}
@@ -164,6 +167,9 @@ func TestQueries(t *testing.T) {
 }
 
 func TestQueryDependencyTree(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping long test in short mode")
+	}
 	ctx := context.Background()
 
 	cfg := &serviceconfig.BaseConfig{}
@@ -235,6 +241,9 @@ func TestQueryDependencyTree(t *testing.T) {
 }
 
 func TestQueriesList(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping long test in short mode")
+	}
 	ctx := context.Background()
 
 	cfg := &serviceconfig.BaseConfig{}

internal/database/repository/result_test.go

@@ -14,6 +14,9 @@ import (
 )
 
 func TestResults(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping long test in short mode")
+	}
 	ctx := context.Background()
 
 	cfg := &serviceconfig.BaseConfig{}
@@ -73,6 +76,9 @@ func TestResults(t *testing.T) {
 }
 
 func TestResultValues(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping long test in short mode")
+	}
 	ctx := context.Background()
 
 	cfg := &serviceconfig.BaseConfig{}
@@ -171,6 +177,9 @@ func TestResultValues(t *testing.T) {
 }
 
 func TestUnsyncedQueries(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping long test in short mode")
+	}
 	ctx := context.Background()
 
 	cfg := &serviceconfig.BaseConfig{}

internal/database/repository/text.sql.go

@@ -0,0 +1,96 @@
+// Code generated by sqlc. DO NOT EDIT.
+// versions:
+//   sqlc v1.27.0
+// source: text.sql
+
+package repository
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v5/pgtype"
+)
+
+const addDocumentTextEntry = `-- name: AddDocumentTextEntry :exec
+INSERT INTO documentTextExtractions (documentId, version, bucket, key) VALUES ($1, $2, $3, $4)
+`
+
+type AddDocumentTextEntryParams struct {
+	Documentid pgtype.UUID `db:"documentid"`
+	Version    int32       `db:"version"`
+	Bucket     string      `db:"bucket"`
+	Key        string      `db:"key"`
+}
+
+// AddDocumentTextEntry
+//
+//	INSERT INTO documentTextExtractions (documentId, version, bucket, key) VALUES ($1, $2, $3, $4)
+func (q *Queries) AddDocumentTextEntry(ctx context.Context, arg *AddDocumentTextEntryParams) error {
+	_, err := q.db.Exec(ctx, addDocumentTextEntry,
+		arg.Documentid,
+		arg.Version,
+		arg.Bucket,
+		arg.Key,
+	)
+	return err
+}
+
+const getDocumentTextEntry = `-- name: GetDocumentTextEntry :one
+SELECT dc.documentId, dc.bucket, dc.key
+    FROM documentTextExtractions 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.minTextVersion
+    ORDER BY d.id DESC LIMIT 1
+`
+
+type GetDocumentTextEntryRow struct {
+	Documentid pgtype.UUID `db:"documentid"`
+	Bucket     string      `db:"bucket"`
+	Key        string      `db:"key"`
+}
+
+// GetDocumentTextEntry
+//
+//	SELECT dc.documentId, dc.bucket, dc.key
+//	    FROM documentTextExtractions 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.minTextVersion
+//	    ORDER BY d.id DESC LIMIT 1
+func (q *Queries) GetDocumentTextEntry(ctx context.Context, documentid pgtype.UUID) (*GetDocumentTextEntryRow, error) {
+	row := q.db.QueryRow(ctx, getDocumentTextEntry, documentid)
+	var i GetDocumentTextEntryRow
+	err := row.Scan(&i.Documentid, &i.Bucket, &i.Key)
+	return &i, err
+}
+
+const isDocumentTextExtracted = `-- name: IsDocumentTextExtracted :one
+SELECT EXISTS(
+    SELECT 1
+        FROM documentTextExtractions 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.minTextVersion
+)
+`
+
+// IsDocumentTextExtracted
+//
+//	SELECT EXISTS(
+//	    SELECT 1
+//	        FROM documentTextExtractions 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.minTextVersion
+//	)
+func (q *Queries) IsDocumentTextExtracted(ctx context.Context, documentid pgtype.UUID) (bool, error) {
+	row := q.db.QueryRow(ctx, isDocumentTextExtracted, documentid)
+	var exists bool
+	err := row.Scan(&exists)
+	return exists, err
+}

internal/database/repository/text_test.go

@@ -0,0 +1,92 @@
+package repository_test
+
+import (
+	"context"
+	"os"
+	"path"
+	"queryorchestration/internal/database/repository"
+	"queryorchestration/internal/serviceconfig"
+	"queryorchestration/internal/test"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestTextExtraction(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping long test in short mode")
+	}
+	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: 1,
+		Mintextversion:  2,
+	})
+	assert.NoError(t, err)
+
+	isextract, err := queries.IsDocumentTextExtracted(ctx, id)
+	assert.NoError(t, err)
+	assert.False(t, isextract)
+
+	bucket := "example_bucket"
+	key := "example_key"
+	err = queries.AddDocumentTextEntry(ctx, &repository.AddDocumentTextEntryParams{
+		Documentid: id,
+		Version:    1,
+		Bucket:     bucket,
+		Key:        key,
+	})
+	assert.NoError(t, err)
+
+	isextract, err = queries.IsDocumentTextExtracted(ctx, id)
+	assert.NoError(t, err)
+	assert.False(t, isextract)
+
+	err = queries.AddCollectorCodeVersion(ctx, &repository.AddCollectorCodeVersionParams{
+		Collectorid:     collId,
+		Addedversion:    1,
+		Mincleanversion: 1,
+		Mintextversion:  1,
+	})
+	assert.NoError(t, err)
+
+	isextract, err = queries.IsDocumentTextExtracted(ctx, id)
+	assert.NoError(t, err)
+	assert.True(t, isextract)
+
+	clean, err := queries.GetDocumentTextEntry(ctx, id)
+	assert.NoError(t, err)
+	assert.EqualExportedValues(t, &repository.GetDocumentTextEntryRow{
+		Documentid: id,
+		Bucket:     bucket,
+		Key:        key,
+	}, clean)
+}

internal/document/clean/create.go

@@ -2,8 +2,8 @@ package documentclean
 
 import (
 	"context"
+	doctextrunner "queryorchestration/api/docTextRunner"
 	"queryorchestration/internal/database"
-	documenttext "queryorchestration/internal/document/text"
 	"queryorchestration/internal/serviceconfig/queue"
 
 	"github.com/google/uuid"
@@ -33,7 +33,7 @@ func (s *Service) Create(ctx context.Context, id uuid.UUID) error {
 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{
+		Body: doctextrunner.Create{
 			ID: id,
 		},
 	})

internal/document/text/create.go

@@ -0,0 +1,45 @@
+package documenttext
+
+import (
+	"context"
+	querysyncrunner "queryorchestration/api/querySyncRunner"
+	"queryorchestration/internal/database"
+	"queryorchestration/internal/serviceconfig/queue"
+
+	"github.com/google/uuid"
+)
+
+func (s *Service) Create(ctx context.Context, id uuid.UUID) error {
+	isextracted, err := s.cfg.GetDBQueries().IsDocumentTextExtracted(ctx, database.MustToDBUUID(id))
+	if err != nil {
+		return err
+	}
+
+	if !isextracted {
+		err = s.extract(ctx, id)
+		if err != nil {
+			return err
+		}
+	}
+
+	err = s.informExtraction(ctx, id)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (s *Service) informExtraction(ctx context.Context, id uuid.UUID) error {
+	err := s.cfg.SendToQueue(ctx, &queue.SendParams{
+		QueueURL: s.cfg.GetQuerySyncURL(),
+		Body: querysyncrunner.Create{
+			ID: id,
+		},
+	})
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

internal/document/text/create_test.go

@@ -0,0 +1,105 @@
+package documenttext
+
+import (
+	"context"
+	"fmt"
+	"queryorchestration/internal/database"
+	"queryorchestration/internal/database/repository"
+	"queryorchestration/internal/document"
+	"queryorchestration/internal/serviceconfig"
+	"queryorchestration/internal/serviceconfig/objectstore"
+	"queryorchestration/internal/serviceconfig/queue/querysync"
+	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 DocTextConfig struct {
+	serviceconfig.BaseConfig
+	objectstore.ObjectStoreConfig
+	querysync.QuerySyncConfig
+}
+
+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 := &DocTextConfig{}
+	cfg.QueueClient = mockSQS
+	cfg.QuerySyncURL = "/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: IsDocumentTextExtracted :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows(
+		pgxmock.NewRows([]string{"isextracted"}).
+			AddRow(false),
+	)
+	pool.ExpectQuery("name: GetDocumentCleanEntry :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows(
+		pgxmock.NewRows([]string{"documentId", "bucket", "key"}).
+			AddRow(database.MustToDBUUID(id), inloc.Bucket, inloc.Key),
+	)
+	pool.ExpectExec("name: AddDocumentTextEntry :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.QuerySyncURL && *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 := &DocTextConfig{}
+	cfg.QueueClient = mockSQS
+	cfg.QuerySyncURL = "/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.QuerySyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", id)
+			}),
+			mock.Anything,
+		).
+		Return(&sqs.SendMessageOutput{}, nil)
+
+	err := svc.informExtraction(ctx, id)
+	assert.NoError(t, err)
+}

internal/document/text/extract.go

@@ -0,0 +1,54 @@
+package documenttext
+
+import (
+	"context"
+	"queryorchestration/internal/database"
+	"queryorchestration/internal/database/repository"
+	"queryorchestration/internal/document"
+
+	"github.com/google/uuid"
+)
+
+type ExtractionParams struct {
+	ID       uuid.UUID
+	Location document.Location
+}
+
+func (s *Service) executeExtraction(params *ExtractionParams) (*document.Location, error) {
+	// TODO - various extraction tasks
+	return &params.Location, nil
+}
+
+func (s *Service) extract(ctx context.Context, id uuid.UUID) error {
+	docId := database.MustToDBUUID(id)
+
+	entry, err := s.cfg.GetDBQueries().GetDocumentCleanEntry(ctx, docId)
+	if err != nil {
+		return err
+	}
+
+	outLocation, err := s.executeExtraction(&ExtractionParams{
+		ID: id,
+		Location: document.Location{
+			Bucket: entry.Bucket,
+			Key:    entry.Key,
+		},
+	})
+	if err != nil {
+		return err
+	}
+
+	version := s.svc.Document.GetTextVersion()
+
+	err = s.cfg.GetDBQueries().AddDocumentTextEntry(ctx, &repository.AddDocumentTextEntryParams{
+		Documentid: docId,
+		Version:    version,
+		Bucket:     outLocation.Bucket,
+		Key:        outLocation.Key,
+	})
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

internal/document/text/extract_test.go

@@ -0,0 +1,62 @@
+package documenttext
+
+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 TestExtract(t *testing.T) {
+	ctx := context.Background()
+	pool, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to open pgxmock database: %v", err)
+	}
+
+	cfg := &DocTextConfig{}
+	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: GetDocumentCleanEntry :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows(
+		pgxmock.NewRows([]string{"documentId", "bucket", "key"}).
+			AddRow(database.MustToDBUUID(id), inloc.Bucket, inloc.Key),
+	)
+	pool.ExpectExec("name: AddDocumentTextEntry :exec").WithArgs(database.MustToDBUUID(id), int32(1), inloc.Bucket, inloc.Key).
+		WillReturnResult(pgxmock.NewResult("", 1))
+
+	err = svc.extract(ctx, id)
+	assert.NoError(t, err)
+}
+
+func TestExecuteExtraction(t *testing.T) {
+	svc := Service{}
+
+	id := uuid.New()
+	location := document.Location{}
+
+	outloc, err := svc.executeExtraction(&ExtractionParams{
+		ID:       id,
+		Location: location,
+	})
+	assert.NoError(t, err)
+	assert.EqualExportedValues(t, location, *outloc)
+}

internal/document/text/service.go

@@ -1,27 +1,30 @@
 package documenttext
 
 import (
-	"github.com/google/uuid"
+	"queryorchestration/internal/document"
+	"queryorchestration/internal/serviceconfig"
+	"queryorchestration/internal/serviceconfig/objectstore"
+	"queryorchestration/internal/serviceconfig/queue/querysync"
 )
 
+type ConfigProvider interface {
+	serviceconfig.ConfigProvider
+	objectstore.ConfigProvider
+	querysync.ConfigProvider
+}
+
+type Services struct {
+	Document *document.Service
+}
+
 type Service struct {
+	cfg ConfigProvider
+	svc *Services
 }
 
-func New() *Service {
-	return &Service{}
-}
-
-// TODO - move to api/
-type Create struct {
-	ID uuid.UUID `json:"id" validate:"required,uuid"`
-}
-
-type IsExtractedParams struct {
-	DocumentID      uuid.UUID
-	MinCleanVersion int32
-	MinTextVersion  int32
-}
-
-func (s *Service) IsExtracted(params *IsExtractedParams) error {
-	return nil
+func New(cfg ConfigProvider, svc *Services) *Service {
+	return &Service{
+		cfg: cfg,
+		svc: svc,
+	}
 }

internal/document/text/service_test.go

@@ -1,24 +1,26 @@
 package documenttext_test
 
 import (
+	"queryorchestration/internal/document"
 	documenttext "queryorchestration/internal/document/text"
+	"queryorchestration/internal/serviceconfig"
+	"queryorchestration/internal/serviceconfig/objectstore"
+	"queryorchestration/internal/serviceconfig/queue/querysync"
 	"testing"
 
-	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 )
 
+type DocTextConfig struct {
+	serviceconfig.BaseConfig
+	objectstore.ObjectStoreConfig
+	querysync.QuerySyncConfig
+}
+
 func TestService(t *testing.T) {
-	svc := documenttext.New()
+	cfg := &DocTextConfig{}
+	svc := documenttext.New(cfg, &documenttext.Services{
+		Document: document.New(cfg),
+	})
 	assert.NotNil(t, svc)
 }
-
-func TestIsExtracted(t *testing.T) {
-	svc := documenttext.New()
-
-	assert.Nil(t, svc.IsExtracted(&documenttext.IsExtractedParams{
-		DocumentID:      uuid.New(),
-		MinCleanVersion: int32(1),
-		MinTextVersion:  int32(1),
-	}))
-}

internal/document/version.go

@@ -2,8 +2,21 @@ package document
 
 import (
 	"fmt"
+
+	"github.com/google/uuid"
 )
 
+type IsTextExtractedParams struct {
+	DocumentID      uuid.UUID
+	MinCleanVersion int32
+	MinTextVersion  int32
+}
+
+func (s *Service) IsTextExtracted(params *IsTextExtractedParams) error {
+	// TODO
+	return nil
+}
+
 func (s *Service) GetCleanVersion() int32 {
 	// TODO - actual version
 	return 1

internal/document/version_test.go

@@ -5,6 +5,7 @@ import (
 	"queryorchestration/internal/serviceconfig"
 	"testing"
 
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -41,3 +42,14 @@ func TestGetTextVersion(t *testing.T) {
 
 	assert.Equal(t, int32(1), svc.GetTextVersion())
 }
+
+func TestIsTextExtracted(t *testing.T) {
+	cfg := &serviceconfig.BaseConfig{}
+	svc := document.New(cfg)
+
+	assert.Nil(t, svc.IsTextExtracted(&document.IsTextExtractedParams{
+		DocumentID:      uuid.New(),
+		MinCleanVersion: int32(1),
+		MinTextVersion:  int32(1),
+	}))
+}

internal/query/service.go

@@ -2,7 +2,6 @@ package query
 
 import (
 	"queryorchestration/internal/document"
-	documenttext "queryorchestration/internal/document/text"
 	"queryorchestration/internal/job/collector"
 	"queryorchestration/internal/query/result"
 	"queryorchestration/internal/serviceconfig"
@@ -11,7 +10,6 @@ import (
 )
 
 type Services struct {
-	Text      *documenttext.Service
 	Result    *result.Service
 	Collector *collector.Service
 	Document  *document.Service

internal/query/sync.go

@@ -3,7 +3,7 @@ package query
 import (
 	"context"
 	"errors"
-	documenttext "queryorchestration/internal/document/text"
+	"queryorchestration/internal/document"
 	"queryorchestration/internal/query/result"
 	resultprocessor "queryorchestration/internal/query/result/processor"
 	"sync"
@@ -12,7 +12,7 @@ import (
 )
 
 func (s *Service) Sync(ctx context.Context, doc *Document) error {
-	err := s.svc.Text.IsExtracted(&documenttext.IsExtractedParams{
+	err := s.svc.Document.IsTextExtracted(&document.IsTextExtractedParams{
 		DocumentID:      doc.ID,
 		MinCleanVersion: doc.CleanVersion,
 		MinTextVersion:  doc.TextVersion,

internal/query/test_test.go

@@ -5,7 +5,6 @@ import (
 	"queryorchestration/internal/database"
 	"queryorchestration/internal/database/repository"
 	"queryorchestration/internal/document"
-	documenttext "queryorchestration/internal/document/text"
 	"queryorchestration/internal/job/collector"
 	"queryorchestration/internal/query"
 	"queryorchestration/internal/query/result"
@@ -32,9 +31,7 @@ func TestTest(t *testing.T) {
 	col := collector.New(cfg, &collector.Services{
 		Document: docsvc,
 	})
-	text := documenttext.New()
 	svc := query.New(cfg, &query.Services{
-		Text:      text,
 		Document:  docsvc,
 		Collector: col,
 		Result:    result.New(cfg),

internal/serviceconfig/queue/querysync/config.go

@@ -0,0 +1,13 @@
+package querysync
+
+type QuerySyncConfig struct {
+	QuerySyncURL string `env:"QUERY_SYNC_URL,required,notEmpty"`
+}
+
+func (c *QuerySyncConfig) GetQuerySyncURL() string {
+	return c.QuerySyncURL
+}
+
+type ConfigProvider interface {
+	GetQuerySyncURL() string
+}

internal/serviceconfig/queue/querysync/config_test.go

@@ -0,0 +1,20 @@
+package querysync_test
+
+import (
+	"queryorchestration/internal/serviceconfig/queue/querysync"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGetQuerySyncURL(t *testing.T) {
+	cfg := querysync.QuerySyncConfig{}
+
+	name := cfg.GetQuerySyncURL()
+	assert.Equal(t, "", name)
+
+	cfg.QuerySyncURL = "name"
+	name = cfg.GetQuerySyncURL()
+	assert.Equal(t, "name", name)
+	assert.Equal(t, cfg.QuerySyncURL, name)
+}

internal/test/runner.go

@@ -2,7 +2,9 @@ package test
 
 import (
 	"context"
+	doccleanrunner "queryorchestration/api/docCleanRunner"
 	docinitrunner "queryorchestration/api/docInitRunner"
+	doctextrunner "queryorchestration/api/docTextRunner"
 	queryrunner "queryorchestration/api/queryRunner"
 	"testing"
 
@@ -14,9 +16,11 @@ import (
 type Runner = string
 
 const (
-	DocInitRunner  = docinitrunner.Name
-	DocCleanRunner = docinitrunner.Name
-	QueryRunner    = queryrunner.Name
+	DocInitRunner   = docinitrunner.Name
+	DocCleanRunner  = doccleanrunner.Name
+	DocTextRunner   = doctextrunner.Name
+	QuerySyncRunner = queryrunner.Name
+	QueryRunner     = queryrunner.Name
 )
 
 type RunnerConfig struct {

scripts/compose.yml

@@ -36,6 +36,7 @@ tasks:
       - aws sqs create-queue --queue-name $QNAME_DOCUMENT_INIT
       - aws sqs create-queue --queue-name $QNAME_DOCUMENT_CLEAN
       - aws sqs create-queue --queue-name $QNAME_DOCUMENT_TEXT
+      - aws sqs create-queue --queue-name $QNAME_QUERY_SYNC
       - aws sqs create-queue --queue-name $QNAME_QUERY_RUNNER
       - |
         aws s3api put-bucket-notification-configuration \

test/process_test.go

@@ -65,7 +65,8 @@ func TestProcess(t *testing.T) {
 	assert.NoError(t, err)
 
 	doccleanurl := test.CreateQueue(t, ctx, cfg, test.DocCleanRunner)
-	doctexturl := test.CreateQueue(t, ctx, cfg, "doctextrunner")
+	doctexturl := test.CreateQueue(t, ctx, cfg, test.DocTextRunner)
+	querysyncurl := test.CreateQueue(t, ctx, cfg, test.QuerySyncRunner)
 
 	net, cleanup := test.CreateRunnersAndServicesNetwork(t, ctx, &test.EcosystemNetworkConfig{
 		Cfg:     cfg,
@@ -83,6 +84,12 @@ func TestProcess(t *testing.T) {
 					"DOCUMENT_TEXT_URL": doctexturl,
 				},
 			},
+			{
+				Name: test.DocTextRunner,
+				Env: map[string]string{
+					"QUERY_SYNC_URL": querysyncurl,
+				},
+			},
 		},
 		Services: []*test.ServiceNetworkConfig{
 			{
@@ -124,5 +131,5 @@ func TestProcess(t *testing.T) {
 	assert.NoError(t, err)
 
 	resRegex := regexp.MustCompile(`{"id": "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"`)
-	test.AssertMessageBody(t, ctx, cfg, doctexturl, resRegex)
+	test.AssertMessageBody(t, ctx, cfg, querysyncurl, resRegex)
 }