Merged in bugfix/errhandling (pull request #89)

Error handling

* basic

api/docCleanRunner/runner.go

@@ -3,6 +3,7 @@ package doccleanrunner
 import (
 	"context"
 	"encoding/json"
+	"log/slog"
 
 	documentclean "queryorchestration/internal/document/clean"
 
@@ -34,22 +35,25 @@ type Body struct {
 	ID uuid.UUID `json:"id" validate:"required,uuid"`
 }
 
-func (s Runner) Process(ctx context.Context, req *types.Message) error {
+func (s Runner) Process(ctx context.Context, req *types.Message) bool {
 	var body Body
 	err := json.Unmarshal([]byte(*req.Body), &body)
 	if err != nil {
-		return err
+		slog.Error("parsing body", "messageId", *req.MessageId, "body", *req.Body)
+		return true
 	}
 
 	err = s.validator.Struct(body)
 	if err != nil {
-		return err
+		slog.Error("invalid body", "messageId", *req.MessageId, "error", err)
+		return true
 	}
 
 	err = s.svc.Clean.Clean(ctx, body.ID)
 	if err != nil {
-		return err
+		slog.Error("unable to process", "error", err)
+		return false
 	}
 
-	return nil
+	return true
 }

api/docCleanRunner/runner_test.go

@@ -59,97 +59,108 @@ func TestDocCleanRunner(t *testing.T) {
 	})
 	assert.NotNil(t, runner)
 
-	bod := doccleanrunner.Body{
-		ID: uuid.New(),
-	}
-	bodyBytes, err := json.Marshal(bod)
-	assert.NoError(t, err)
-	body := string(bodyBytes)
-	msg := &types.Message{
-		Body: &body,
-	}
+	t.Run("valid", func(t *testing.T) {
+		bod := doccleanrunner.Body{
+			ID: uuid.New(),
+		}
+		bodyBytes, err := json.Marshal(bod)
+		assert.NoError(t, err)
+		body := string(bodyBytes)
+		msg := &types.Message{
+			Body: &body,
+		}
 
-	doc := document.Document{
-		ID:    bod.ID,
-		JobID: uuid.New(),
-		Hash:  "example_hash",
-	}
-	inloc := document.Location{
-		Bucket: "bucket_name",
-		Key:    "/i/am/here",
-	}
-	dbid := database.MustToDBUUID(doc.ID)
+		doc := document.Document{
+			ID:    bod.ID,
+			JobID: uuid.New(),
+			Hash:  "example_hash",
+		}
+		inloc := document.Location{
+			Bucket: "bucket_name",
+			Key:    "/i/am/here",
+		}
+		dbid := database.MustToDBUUID(doc.ID)
 
-	pool.ExpectQuery("name: HasDocumentCleanEntry :one").WithArgs(dbid).WillReturnRows(
-		pgxmock.NewRows([]string{"isclean"}).
-			AddRow(false),
-	)
-	pool.ExpectQuery("name: GetDocument :one").WithArgs(dbid).
-		WillReturnRows(
-			pgxmock.NewRows([]string{"id", "jobId", "hash"}).
-				AddRow(dbid, database.MustToDBUUID(doc.JobID), doc.Hash),
+		pool.ExpectQuery("name: HasDocumentCleanEntry :one").WithArgs(dbid).WillReturnRows(
+			pgxmock.NewRows([]string{"isclean"}).
+				AddRow(false),
 		)
-	pool.ExpectQuery("name: GetDocumentEntry :one").WithArgs(dbid).WillReturnRows(
-		pgxmock.NewRows([]string{"documentId", "bucket", "key"}).
-			AddRow(dbid, inloc.Bucket, inloc.Key),
-	)
-	mimeType := "application/pdf"
-	dbmimetype := repository.NullCleanmimetype{
-		Valid:         true,
-		Cleanmimetype: repository.Cleanmimetype(mimeType),
-	}
-	pool.ExpectBegin()
-	cleanid := database.MustToDBUUID(uuid.New())
-	pool.ExpectQuery("name: GetMostRecentDocumentCleanEntry :one").WithArgs(dbid).
-		WillReturnRows(
-			pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}),
+		pool.ExpectQuery("name: GetDocument :one").WithArgs(dbid).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"id", "jobId", "hash"}).
+					AddRow(dbid, database.MustToDBUUID(doc.JobID), doc.Hash),
+			)
+		pool.ExpectQuery("name: GetDocumentEntry :one").WithArgs(dbid).WillReturnRows(
+			pgxmock.NewRows([]string{"documentId", "bucket", "key"}).
+				AddRow(dbid, inloc.Bucket, inloc.Key),
 		)
-	pool.ExpectQuery("name: AddDocumentClean :one").WithArgs(dbid, &inloc.Bucket, &inloc.Key, dbmimetype, repository.NullCleanfailtype{}).
-		WillReturnRows(
-			pgxmock.NewRows([]string{"id"}).
-				AddRow(cleanid),
-		)
-	pool.ExpectExec("name: AddDocumentCleanEntry :exec").WithArgs(cleanid, int32(1)).
-		WillReturnResult(pgxmock.NewResult("", 1))
-	pool.ExpectCommit()
+		mimeType := "application/pdf"
+		dbmimetype := repository.NullCleanmimetype{
+			Valid:         true,
+			Cleanmimetype: repository.Cleanmimetype(mimeType),
+		}
+		pool.ExpectBegin()
+		cleanid := database.MustToDBUUID(uuid.New())
+		pool.ExpectQuery("name: GetMostRecentDocumentCleanEntry :one").WithArgs(dbid).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}),
+			)
+		pool.ExpectQuery("name: AddDocumentClean :one").WithArgs(dbid, &inloc.Bucket, &inloc.Key, dbmimetype, repository.NullCleanfailtype{}).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"id"}).
+					AddRow(cleanid),
+			)
+		pool.ExpectExec("name: AddDocumentCleanEntry :exec").WithArgs(cleanid, int32(1)).
+			WillReturnResult(pgxmock.NewResult("", 1))
+		pool.ExpectCommit()
 
-	mockSQS.EXPECT().
-		SendMessage(
-			mock.Anything,
-			mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
-				return *in.QueueUrl == cfg.DocumentTextURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", doc.ID.String())
-			}),
-			mock.Anything,
-		).
-		Return(&sqs.SendMessageOutput{}, nil)
+		mockSQS.EXPECT().
+			SendMessage(
+				mock.Anything,
+				mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
+					return *in.QueueUrl == cfg.DocumentTextURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", doc.ID.String())
+				}),
+				mock.Anything,
+			).
+			Return(&sqs.SendMessageOutput{}, nil)
 
-	mockStore.EXPECT().
-		HeadObject(
-			mock.Anything,
-			mock.MatchedBy(func(in *s3.HeadObjectInput) bool {
-				return *in.Bucket == inloc.Bucket && *in.Key == inloc.Key
-			}),
-			mock.Anything,
-		).
-		Return(&s3.HeadObjectOutput{
-			ContentType: &mimeType,
-		}, nil)
+		mockStore.EXPECT().
+			HeadObject(
+				mock.Anything,
+				mock.MatchedBy(func(in *s3.HeadObjectInput) bool {
+					return *in.Bucket == inloc.Bucket && *in.Key == inloc.Key
+				}),
+				mock.Anything,
+			).
+			Return(&s3.HeadObjectOutput{
+				ContentType: &mimeType,
+			}, nil)
 
-	bodyMsg := io.NopCloser(strings.NewReader(pdfHelloWorld))
-	mockStore.EXPECT().
-		GetObject(
-			mock.Anything,
-			mock.MatchedBy(func(in *s3.GetObjectInput) bool {
-				return *in.Bucket == inloc.Bucket && *in.Key == inloc.Key
-			}),
-			mock.Anything,
-		).
-		Return(&s3.GetObjectOutput{
-			Body: bodyMsg,
-		}, nil)
+		bodyMsg := io.NopCloser(strings.NewReader(pdfHelloWorld))
+		mockStore.EXPECT().
+			GetObject(
+				mock.Anything,
+				mock.MatchedBy(func(in *s3.GetObjectInput) bool {
+					return *in.Bucket == inloc.Bucket && *in.Key == inloc.Key
+				}),
+				mock.Anything,
+			).
+			Return(&s3.GetObjectOutput{
+				Body: bodyMsg,
+			}, nil)
 
-	err = runner.Process(ctx, msg)
-	assert.NoError(t, err)
+		assert.True(t, runner.Process(ctx, msg))
+	})
+	t.Run("invalid body", func(t *testing.T) {
+		body := "definitely invalid"
+		msgId := "id"
+		msg := &types.Message{
+			Body:      &body,
+			MessageId: &msgId,
+		}
+
+		assert.True(t, runner.Process(ctx, msg))
+	})
 }
 
 const pdfHelloWorld = `%PDF-1.4

api/docInitRunner/runner.go

@@ -3,8 +3,6 @@ package docinitrunner
 import (
 	"context"
 	"encoding/json"
-	"errors"
-	"fmt"
 	"log/slog"
 
 	"queryorchestration/internal/document"
@@ -62,54 +60,64 @@ type S3EventNotification struct {
 	Records []S3EventRecord `json:"Records"`
 }
 
-func (s Runner) Process(ctx context.Context, req *types.Message) error {
+type S3Events string
+
+const (
+	S3EventObjectCreatedPut = "ObjectCreated:Put"
+)
+
+func (s Runner) Process(ctx context.Context, req *types.Message) bool {
 	var body S3EventNotification
 	err := json.Unmarshal([]byte(*req.Body), &body)
 	if err != nil {
-		return err
+		slog.Error("parsing body", "messageId", *req.MessageId, "body", *req.Body)
+		return true
 	}
 
 	err = s.validator.Struct(body)
 	if err != nil {
-		return err
+		slog.Error("invalid body", "messageId", *req.MessageId, "error", err)
+		return true
 	}
 
-	hasErr := false
-
 	for _, record := range body.Records {
-		err := func() error {
-			if record.EventName != "ObjectCreated:Put" {
-				return fmt.Errorf("invalid event name: %s", record.EventName)
-			}
-
-			jobID, err := s.svc.Document.GetJobIDFromKey(record.S3.Object.Key)
-			if err != nil {
-				return err
-			}
-
-			_, err = s.svc.Document.Create(ctx, &documentinit.Create{
-				JobID: jobID,
-				Location: document.Location{
-					Bucket: record.S3.Bucket.Name,
-					Key:    record.S3.Object.Key,
-				},
-				Hash: record.S3.Object.ETag,
-			})
-			if err != nil {
-				return err
-			}
-
-			return nil
-		}()
+		err := s.processRecord(ctx, record)
 		if err != nil {
-			hasErr = true
-			slog.Error("Error processing record", "err", err)
+			slog.Error("unable to process", "error", err)
+			return false
 		}
 	}
 
-	if hasErr {
-		return errors.New("error processing records")
+	return true
+}
+
+func (s Runner) processRecord(ctx context.Context, record S3EventRecord) error {
+	if !s.isSupportedEvent(record.EventName) {
+		slog.Warn("unsupported event", "name", record.EventName)
+		return nil
+	}
+
+	jobID, err := s.svc.Document.GetJobIDFromKey(record.S3.Object.Key)
+	if err != nil {
+		slog.Error("unable to find job id", "key", record.S3.Object.Key, "error", err)
+		return nil
+	}
+
+	_, err = s.svc.Document.Create(ctx, &documentinit.Create{
+		JobID: jobID,
+		Location: document.Location{
+			Bucket: record.S3.Bucket.Name,
+			Key:    record.S3.Object.Key,
+		},
+		Hash: record.S3.Object.ETag,
+	})
+	if err != nil {
+		return err
 	}
 
 	return nil
 }
+
+func (s Runner) isSupportedEvent(name string) bool {
+	return name == S3EventObjectCreatedPut
+}

api/docInitRunner/runner_test.go

@@ -1,4 +1,4 @@
-package docinitrunner_test
+package docinitrunner
 
 import (
 	"context"
@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"testing"
 
-	docinitrunner "queryorchestration/api/docInitRunner"
 	"queryorchestration/internal/database"
 	"queryorchestration/internal/database/repository"
 	"queryorchestration/internal/document"
@@ -43,73 +42,192 @@ func TestDocInitRunner(t *testing.T) {
 	mockSQS := queuemock.NewMockSQSClient(t)
 	cfg.QueueClient = mockSQS
 
-	runner := docinitrunner.New(validator.New(), &docinitrunner.Services{
+	runner := New(validator.New(), &Services{
 		Document: documentinit.New(cfg),
 	})
 	assert.NotNil(t, runner)
 
-	j := job.Job{
-		ID:       uuid.New(),
-		ClientID: uuid.New(),
-	}
-	bucketName := "bucketName"
-	location := fmt.Sprintf("%s/%s/aaa", j.ClientID.String(), j.ID.String())
-	docinfo := document.Document{
-		ID:    uuid.New(),
-		JobID: j.ID,
-		Hash:  "example_hash",
-	}
-	doc := docinitrunner.S3EventNotification{
-		Records: []docinitrunner.S3EventRecord{
-			{
-				EventName: "ObjectCreated:Put",
-				S3: docinitrunner.S3EventRecordDetails{
-					Bucket: docinitrunner.S3Bucket{
-						Name: bucketName,
-					},
-					Object: docinitrunner.S3Object{
-						Key:  location,
-						ETag: docinfo.Hash,
+	t.Run("valid", func(t *testing.T) {
+		j := job.Job{
+			ID:       uuid.New(),
+			ClientID: uuid.New(),
+		}
+		bucketName := "bucketName"
+		location := fmt.Sprintf("%s/%s/aaa", j.ClientID.String(), j.ID.String())
+		docinfo := document.Document{
+			ID:    uuid.New(),
+			JobID: j.ID,
+			Hash:  "example_hash",
+		}
+		doc := S3EventNotification{
+			Records: []S3EventRecord{
+				{
+					EventName: "ObjectCreated:Put",
+					S3: S3EventRecordDetails{
+						Bucket: S3Bucket{
+							Name: bucketName,
+						},
+						Object: S3Object{
+							Key:  location,
+							ETag: docinfo.Hash,
+						},
 					},
 				},
 			},
-		},
-	}
-	bodyBytes, err := json.Marshal(doc)
-	assert.NoError(t, err)
-	body := string(bodyBytes)
-	msg := &types.Message{
-		Body: &body,
-	}
+		}
+		bodyBytes, err := json.Marshal(doc)
+		assert.NoError(t, err)
+		body := string(bodyBytes)
+		msg := &types.Message{
+			Body: &body,
+		}
 
-	pool.ExpectQuery("name: GetDocumentIDByHash :one").WithArgs(docinfo.Hash, database.MustToDBUUID(j.ID)).WillReturnRows(
-		pgxmock.NewRows([]string{"id"}),
-	)
-	pool.ExpectBegin()
-	pool.ExpectQuery("name: CreateDocument :one").WithArgs(database.MustToDBUUID(j.ID), docinfo.Hash).
-		WillReturnRows(
-			pgxmock.NewRows([]string{"id"}).
-				AddRow(database.MustToDBUUID(docinfo.ID)),
-		)
-	pool.ExpectExec("name: AddDocumentEntry :exec").WithArgs(database.MustToDBUUID(docinfo.ID), bucketName, location).
-		WillReturnResult(pgxmock.NewResult("", 1))
-	pool.ExpectCommit()
-	pool.ExpectQuery("name: GetDocument :one").WithArgs(database.MustToDBUUID(docinfo.ID)).
-		WillReturnRows(
-			pgxmock.NewRows([]string{"id", "jobId", "hash"}).
-				AddRow(database.MustToDBUUID(docinfo.ID), database.MustToDBUUID(docinfo.JobID), "example"),
+		pool.ExpectQuery("name: GetDocumentIDByHash :one").WithArgs(docinfo.Hash, database.MustToDBUUID(j.ID)).WillReturnRows(
+			pgxmock.NewRows([]string{"id"}),
 		)
+		pool.ExpectBegin()
+		pool.ExpectQuery("name: CreateDocument :one").WithArgs(database.MustToDBUUID(j.ID), docinfo.Hash).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"id"}).
+					AddRow(database.MustToDBUUID(docinfo.ID)),
+			)
+		pool.ExpectExec("name: AddDocumentEntry :exec").WithArgs(database.MustToDBUUID(docinfo.ID), bucketName, location).
+			WillReturnResult(pgxmock.NewResult("", 1))
+		pool.ExpectCommit()
+		pool.ExpectQuery("name: GetDocument :one").WithArgs(database.MustToDBUUID(docinfo.ID)).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"id", "jobId", "hash"}).
+					AddRow(database.MustToDBUUID(docinfo.ID), database.MustToDBUUID(docinfo.JobID), "example"),
+			)
 
-	mockSQS.EXPECT().
-		SendMessage(
-			mock.Anything,
-			mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
-				return *in.QueueUrl == cfg.DocumentSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", docinfo.ID.String())
-			}),
-			mock.Anything,
-		).
-		Return(&sqs.SendMessageOutput{}, nil)
+		mockSQS.EXPECT().
+			SendMessage(
+				mock.Anything,
+				mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
+					return *in.QueueUrl == cfg.DocumentSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", docinfo.ID.String())
+				}),
+				mock.Anything,
+			).
+			Return(&sqs.SendMessageOutput{}, nil)
 
-	err = runner.Process(ctx, msg)
-	assert.NoError(t, err)
+		assert.True(t, runner.Process(ctx, msg))
+	})
+	t.Run("invalid body", func(t *testing.T) {
+		body := "definitely invalid"
+		msgId := "id"
+		msg := &types.Message{
+			Body:      &body,
+			MessageId: &msgId,
+		}
+
+		assert.True(t, runner.Process(ctx, msg))
+	})
+}
+
+func TestProcessRecord(t *testing.T) {
+	ctx := context.Background()
+
+	pool, err := pgxmock.NewPool()
+	require.NoError(t, err)
+
+	cfg := &DocInitConfig{}
+	cfg.DBPool = pool
+	cfg.DBQueries = repository.New(pool)
+	mockSQS := queuemock.NewMockSQSClient(t)
+	cfg.QueueClient = mockSQS
+
+	runner := New(validator.New(), &Services{
+		Document: documentinit.New(cfg),
+	})
+	assert.NotNil(t, runner)
+
+	t.Run("valid", func(t *testing.T) {
+		j := job.Job{
+			ID:       uuid.New(),
+			ClientID: uuid.New(),
+		}
+		bucketName := "bucketName"
+		location := fmt.Sprintf("%s/%s/aaa", j.ClientID.String(), j.ID.String())
+		docinfo := document.Document{
+			ID:    uuid.New(),
+			JobID: j.ID,
+			Hash:  "example_hash",
+		}
+		record := S3EventRecord{
+			EventName: S3EventObjectCreatedPut,
+			S3: S3EventRecordDetails{
+				Bucket: S3Bucket{
+					Name: bucketName,
+				},
+				Object: S3Object{
+					Key:  location,
+					ETag: docinfo.Hash,
+				},
+			},
+		}
+
+		pool.ExpectQuery("name: GetDocumentIDByHash :one").WithArgs(docinfo.Hash, database.MustToDBUUID(j.ID)).WillReturnRows(
+			pgxmock.NewRows([]string{"id"}),
+		)
+		pool.ExpectBegin()
+		pool.ExpectQuery("name: CreateDocument :one").WithArgs(database.MustToDBUUID(j.ID), docinfo.Hash).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"id"}).
+					AddRow(database.MustToDBUUID(docinfo.ID)),
+			)
+		pool.ExpectExec("name: AddDocumentEntry :exec").WithArgs(database.MustToDBUUID(docinfo.ID), bucketName, location).
+			WillReturnResult(pgxmock.NewResult("", 1))
+		pool.ExpectCommit()
+		pool.ExpectQuery("name: GetDocument :one").WithArgs(database.MustToDBUUID(docinfo.ID)).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"id", "jobId", "hash"}).
+					AddRow(database.MustToDBUUID(docinfo.ID), database.MustToDBUUID(docinfo.JobID), "example"),
+			)
+
+		mockSQS.EXPECT().
+			SendMessage(
+				mock.Anything,
+				mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
+					return *in.QueueUrl == cfg.DocumentSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", docinfo.ID.String())
+				}),
+				mock.Anything,
+			).
+			Return(&sqs.SendMessageOutput{}, nil)
+
+		err = runner.processRecord(ctx, record)
+		assert.NoError(t, err)
+	})
+	t.Run("invalid_event", func(t *testing.T) {
+		record := S3EventRecord{
+			EventName: "invalid_event",
+		}
+
+		err = runner.processRecord(ctx, record)
+		assert.NoError(t, err)
+	})
+	t.Run("invalid id", func(t *testing.T) {
+		location := fmt.Sprintf("%s/cc/aaa", uuid.New())
+		record := S3EventRecord{
+			EventName: S3EventObjectCreatedPut,
+			S3: S3EventRecordDetails{
+				Object: S3Object{
+					Key: location,
+				},
+			},
+		}
+
+		err = runner.processRecord(ctx, record)
+		assert.NoError(t, err)
+	})
+}
+
+func TestIsSupportedEvent(t *testing.T) {
+	runner := Runner{}
+
+	t.Run("put", func(t *testing.T) {
+		assert.True(t, runner.isSupportedEvent(S3EventObjectCreatedPut))
+	})
+	t.Run("invalid", func(t *testing.T) {
+		assert.False(t, runner.isSupportedEvent("invalid"))
+	})
 }

api/docSyncRunner/runner.go

@@ -3,6 +3,7 @@ package docsyncrunner
 import (
 	"context"
 	"encoding/json"
+	"log/slog"
 
 	documentsync "queryorchestration/internal/document/sync"
 
@@ -34,22 +35,25 @@ type Body struct {
 	ID uuid.UUID `json:"id" validate:"required,uuid"`
 }
 
-func (s Runner) Process(ctx context.Context, req *types.Message) error {
+func (s Runner) Process(ctx context.Context, req *types.Message) bool {
 	var body Body
 	err := json.Unmarshal([]byte(*req.Body), &body)
 	if err != nil {
-		return err
+		slog.Error("parsing body", "messageId", *req.MessageId, "body", *req.Body)
+		return true
 	}
 
 	err = s.validator.Struct(body)
 	if err != nil {
-		return err
+		slog.Error("invalid body", "messageId", *req.MessageId, "error", err)
+		return true
 	}
 
 	err = s.svc.Document.Sync(ctx, body.ID)
 	if err != nil {
-		return err
+		slog.Error("unable to process", "error", err)
+		return false
 	}
 
-	return nil
+	return true
 }

api/docSyncRunner/runner_test.go

@@ -48,38 +48,49 @@ func TestDocInitRunner(t *testing.T) {
 	})
 	assert.NotNil(t, runner)
 
-	j := job.Job{
-		ID: uuid.New(),
-	}
-	docinfo := document.Document{
-		ID:    uuid.New(),
-		JobID: j.ID,
-	}
-	doc := docsyncrunner.Body{
-		ID: docinfo.ID,
-	}
-	bodyBytes, err := json.Marshal(doc)
-	assert.NoError(t, err)
-	body := string(bodyBytes)
-	msg := &types.Message{
-		Body: &body,
-	}
+	t.Run("valid", func(t *testing.T) {
+		j := job.Job{
+			ID: uuid.New(),
+		}
+		docinfo := document.Document{
+			ID:    uuid.New(),
+			JobID: j.ID,
+		}
+		doc := docsyncrunner.Body{
+			ID: docinfo.ID,
+		}
+		bodyBytes, err := json.Marshal(doc)
+		assert.NoError(t, err)
+		body := string(bodyBytes)
+		msg := &types.Message{
+			Body: &body,
+		}
 
-	pool.ExpectQuery("name: GetDocument :one").WithArgs(database.MustToDBUUID(docinfo.ID)).
-		WillReturnRows(
-			pgxmock.NewRows([]string{"id", "jobId", "hash"}).
-				AddRow(database.MustToDBUUID(docinfo.ID), database.MustToDBUUID(docinfo.JobID), "example"),
-		)
-	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(
-			pgxmock.NewRows([]string{"id", "name", "canSync"}).
-				AddRow(database.MustToDBUUID(j.ClientID), "client_name", true),
+		pool.ExpectQuery("name: GetDocument :one").WithArgs(database.MustToDBUUID(docinfo.ID)).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"id", "jobId", "hash"}).
+					AddRow(database.MustToDBUUID(docinfo.ID), database.MustToDBUUID(docinfo.JobID), "example"),
+			)
+		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(
+				pgxmock.NewRows([]string{"id", "name", "canSync"}).
+					AddRow(database.MustToDBUUID(j.ClientID), "client_name", true),
+			)
 
-	err = runner.Process(ctx, msg)
-	assert.NoError(t, err)
+		assert.True(t, runner.Process(ctx, msg))
+	})
+	t.Run("invalid body", func(t *testing.T) {
+		body := "definitely invalid"
+		msgId := "id"
+		msg := &types.Message{
+			Body:      &body,
+			MessageId: &msgId,
+		}
+
+		assert.True(t, runner.Process(ctx, msg))
+	})
 }

api/docTextRunner/runner.go

@@ -3,6 +3,7 @@ package doctextrunner
 import (
 	"context"
 	"encoding/json"
+	"log/slog"
 
 	documenttext "queryorchestration/internal/document/text"
 
@@ -34,22 +35,25 @@ type Body struct {
 	ID uuid.UUID `json:"id" validate:"required,uuid"`
 }
 
-func (s Runner) Process(ctx context.Context, req *types.Message) error {
+func (s Runner) Process(ctx context.Context, req *types.Message) bool {
 	var body Body
 	err := json.Unmarshal([]byte(*req.Body), &body)
 	if err != nil {
-		return err
+		slog.Error("parsing body", "messageId", *req.MessageId, "body", *req.Body)
+		return true
 	}
 
 	err = s.validator.Struct(body)
 	if err != nil {
-		return err
+		slog.Error("invalid body", "messageId", *req.MessageId, "error", err)
+		return true
 	}
 
 	err = s.svc.Text.Extract(ctx, body.ID)
 	if err != nil {
-		return err
+		slog.Error("unable to process", "error", err)
+		return false
 	}
 
-	return nil
+	return true
 }

api/docTextRunner/runner_test.go

@@ -51,48 +51,63 @@ func TestDocCleanRunner(t *testing.T) {
 	})
 	assert.NotNil(t, runner)
 
-	doc := doctextrunner.Body{
-		ID: uuid.New(),
-	}
-	bodyBytes, err := json.Marshal(doc)
-	assert.NoError(t, err)
-	body := string(bodyBytes)
-	msg := &types.Message{
-		Body: &body,
-	}
+	t.Run("valid", func(t *testing.T) {
+		doc := doctextrunner.Body{
+			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",
-	}
+		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),
-	)
-	cleanId := database.MustToDBUUID(uuid.New())
-	mimeType := "application/pdf"
-	dbmimetype := repository.NullCleanmimetype{
-		Valid:         true,
-		Cleanmimetype: repository.Cleanmimetype(mimeType),
-	}
-	pool.ExpectQuery("name: GetDocumentCleanEntry :one").WithArgs(database.MustToDBUUID(doc.ID)).WillReturnRows(
-		pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}).
-			AddRow(cleanId, database.MustToDBUUID(doc.ID), &inloc.Bucket, &inloc.Key, int32(1), dbmimetype, repository.NullCleanfailtype{}),
-	)
-	pool.ExpectExec("name: AddDocumentTextEntry :exec").WithArgs(int32(1), inloc.Bucket, inloc.Key, cleanId).
-		WillReturnResult(pgxmock.NewResult("", 1))
+		pool.ExpectQuery("name: IsDocumentTextExtracted :one").WithArgs(database.MustToDBUUID(doc.ID)).WillReturnRows(
+			pgxmock.NewRows([]string{"isextracted"}).
+				AddRow(false),
+		)
+		cleanId := database.MustToDBUUID(uuid.New())
+		mimeType := "application/pdf"
+		dbmimetype := repository.NullCleanmimetype{
+			Valid:         true,
+			Cleanmimetype: repository.Cleanmimetype(mimeType),
+		}
+		pool.ExpectQuery("name: GetCleanEntryByDocId :one").WithArgs(database.MustToDBUUID(doc.ID)).WillReturnRows(
+			pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}).
+				AddRow(cleanId, database.MustToDBUUID(doc.ID), &inloc.Bucket, &inloc.Key, int32(1), dbmimetype, repository.NullCleanfailtype{}),
+		)
+		pool.ExpectQuery("name: GetCleanEntry :one").WithArgs(cleanId).WillReturnRows(
+			pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}).
+				AddRow(cleanId, database.MustToDBUUID(doc.ID), &inloc.Bucket, &inloc.Key, int32(1), dbmimetype, repository.NullCleanfailtype{}),
+		)
+		pool.ExpectExec("name: AddDocumentTextEntry :exec").WithArgs(int32(1), inloc.Bucket, inloc.Key, cleanId).
+			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)
+		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)
+		assert.True(t, runner.Process(ctx, msg))
+	})
+	t.Run("invalid body", func(t *testing.T) {
+		body := "definitely invalid"
+		msgId := "id"
+		msg := &types.Message{
+			Body:      &body,
+			MessageId: &msgId,
+		}
+
+		assert.True(t, runner.Process(ctx, msg))
+	})
 }

api/jobSyncRunner/runner.go

@@ -3,6 +3,7 @@ package jobsyncrunner
 import (
 	"context"
 	"encoding/json"
+	"log/slog"
 
 	jobsync "queryorchestration/internal/job/sync"
 
@@ -34,22 +35,25 @@ type Body struct {
 	ID uuid.UUID `json:"id" validate:"required,uuid"`
 }
 
-func (s *Runner) Process(ctx context.Context, req *types.Message) error {
+func (s *Runner) Process(ctx context.Context, req *types.Message) bool {
 	var body Body
 	err := json.Unmarshal([]byte(*req.Body), &body)
 	if err != nil {
-		return err
+		slog.Error("parsing body", "messageId", *req.MessageId, "body", *req.Body)
+		return true
 	}
 
 	err = s.validator.Struct(body)
 	if err != nil {
-		return err
+		slog.Error("invalid body", "messageId", *req.MessageId, "error", err)
+		return true
 	}
 
 	err = s.svc.JobSync.Sync(ctx, body.ID)
 	if err != nil {
-		return err
+		slog.Error("unable to process", "error", err)
+		return false
 	}
 
-	return nil
+	return true
 }

api/jobSyncRunner/runner_test.go

@@ -49,34 +49,45 @@ func TestQueryRunner(t *testing.T) {
 	})
 	assert.NotNil(t, runner)
 
-	bod := jobsyncrunner.Body{
-		ID: uuid.New(),
-	}
-	bodyBytes, err := json.Marshal(bod)
-	assert.NoError(t, err)
-	body := string(bodyBytes)
-	msg := &types.Message{
-		Body: &body,
-	}
+	t.Run("valid", func(t *testing.T) {
+		bod := jobsyncrunner.Body{
+			ID: uuid.New(),
+		}
+		bodyBytes, err := json.Marshal(bod)
+		assert.NoError(t, err)
+		body := string(bodyBytes)
+		msg := &types.Message{
+			Body: &body,
+		}
 
-	docId := uuid.New()
+		docId := uuid.New()
 
-	total := int64(1)
-	pool.ExpectQuery("name: ListJobDocumentIDsBatch :many").WithArgs(database.MustToDBUUID(bod.ID), int32(100), int32(0)).
-		WillReturnRows(
-			pgxmock.NewRows([]string{"id", "totalCount"}).
-				AddRow(database.MustToDBUUID(docId), &total),
-		)
-	mockSQS.EXPECT().
-		SendMessage(
-			mock.Anything,
-			mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
-				return *in.QueueUrl == cfg.DocumentSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", docId.String())
-			}),
-			mock.Anything,
-		).
-		Return(&sqs.SendMessageOutput{}, nil)
+		total := int64(1)
+		pool.ExpectQuery("name: ListJobDocumentIDsBatch :many").WithArgs(database.MustToDBUUID(bod.ID), int32(100), int32(0)).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"id", "totalCount"}).
+					AddRow(database.MustToDBUUID(docId), &total),
+			)
+		mockSQS.EXPECT().
+			SendMessage(
+				mock.Anything,
+				mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
+					return *in.QueueUrl == cfg.DocumentSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", docId.String())
+				}),
+				mock.Anything,
+			).
+			Return(&sqs.SendMessageOutput{}, nil)
 
-	err = runner.Process(ctx, msg)
-	assert.NoError(t, err)
+		assert.True(t, runner.Process(ctx, msg))
+	})
+	t.Run("invalid body", func(t *testing.T) {
+		body := "definitely invalid"
+		msgId := "id"
+		msg := &types.Message{
+			Body:      &body,
+			MessageId: &msgId,
+		}
+
+		assert.True(t, runner.Process(ctx, msg))
+	})
 }

api/queryRunner/runner.go

@@ -3,6 +3,7 @@ package queryrunner
 import (
 	"context"
 	"encoding/json"
+	"log/slog"
 
 	resultset "queryorchestration/internal/query/result/set"
 
@@ -35,16 +36,18 @@ type Body struct {
 	QueryID    uuid.UUID `json:"query_id" validate:"required,uuid"`
 }
 
-func (s *Runner) Process(ctx context.Context, req *types.Message) error {
+func (s *Runner) Process(ctx context.Context, req *types.Message) bool {
 	var body Body
 	err := json.Unmarshal([]byte(*req.Body), &body)
 	if err != nil {
-		return err
+		slog.Error("parsing body", "messageId", *req.MessageId, "body", *req.Body)
+		return true
 	}
 
 	err = s.validator.Struct(body)
 	if err != nil {
-		return err
+		slog.Error("invalid body", "messageId", *req.MessageId, "error", err)
+		return true
 	}
 
 	err = s.svc.ResultSet.Set(ctx, &resultset.Set{
@@ -52,8 +55,9 @@ func (s *Runner) Process(ctx context.Context, req *types.Message) error {
 		QueryID:    body.QueryID,
 	})
 	if err != nil {
-		return err
+		slog.Error("unable to process", "error", err)
+		return false
 	}
 
-	return nil
+	return true
 }

api/queryRunner/runner_test.go

@@ -58,85 +58,96 @@ func TestQueryRunner(t *testing.T) {
 	})
 	assert.NotNil(t, runner)
 
-	doc := resultset.Set{
-		DocumentID: uuid.New(),
-		QueryID:    uuid.New(),
-	}
-	bodyBytes, err := json.Marshal(doc)
-	assert.NoError(t, err)
-	body := string(bodyBytes)
-	msg := &types.Message{
-		Body: &body,
-	}
+	t.Run("valid", func(t *testing.T) {
+		doc := resultset.Set{
+			DocumentID: uuid.New(),
+			QueryID:    uuid.New(),
+		}
+		bodyBytes, err := json.Marshal(doc)
+		assert.NoError(t, err)
+		body := string(bodyBytes)
+		msg := &types.Message{
+			Body: &body,
+		}
 
-	qcfg := "{\"path\":\"examplekey\"}"
-	query := &resultprocessor.Query{
-		ID:               doc.QueryID,
-		Version:          2,
-		RequiredQueryIDs: &[]uuid.UUID{uuid.New()},
-		Config:           &qcfg,
-	}
-	params := &resultset.Set{
-		DocumentID: doc.DocumentID,
-	}
+		qcfg := "{\"path\":\"examplekey\"}"
+		query := &resultprocessor.Query{
+			ID:               doc.QueryID,
+			Version:          2,
+			RequiredQueryIDs: &[]uuid.UUID{uuid.New()},
+			Config:           &qcfg,
+		}
+		params := &resultset.Set{
+			DocumentID: doc.DocumentID,
+		}
 
-	cleanEntryId := uuid.New()
-	textEntryId := uuid.New()
-	pool.ExpectQuery("name: GetDocumentTextEntry :one").WithArgs(database.MustToDBUUID(params.DocumentID)).
-		WillReturnRows(
-			pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "cleanEntryId"}).
-				AddRow(database.MustToDBUUID(textEntryId), database.MustToDBUUID(params.DocumentID), "buket", "/i/am/here", int32(1), database.MustToDBUUID(cleanEntryId)),
+		cleanEntryId := uuid.New()
+		textEntryId := uuid.New()
+		pool.ExpectQuery("name: GetTextEntryByDocId :one").WithArgs(database.MustToDBUUID(params.DocumentID)).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "cleanEntryId"}).
+					AddRow(database.MustToDBUUID(textEntryId), database.MustToDBUUID(params.DocumentID), "buket", "/i/am/here", int32(1), database.MustToDBUUID(cleanEntryId)),
+			)
+		pool.ExpectQuery("name: GetQuery :one").WithArgs(database.MustToDBUUID(query.ID)).WillReturnRows(
+			pgxmock.NewRows([]string{"id", "type", "activeVersion", "latestVersion", "config", "requiredIds"}).
+				AddRow(database.MustToDBUUID(query.ID), repository.QuerytypeJsonExtractor, query.Version, query.Version, []byte(*query.Config), database.MustToDBUUIDArray(*query.RequiredQueryIDs)),
 		)
-	pool.ExpectQuery("name: GetQuery :one").WithArgs(database.MustToDBUUID(query.ID)).WillReturnRows(
-		pgxmock.NewRows([]string{"id", "type", "activeVersion", "latestVersion", "config", "requiredIds"}).
-			AddRow(database.MustToDBUUID(query.ID), repository.QuerytypeJsonExtractor, query.Version, query.Version, []byte(*query.Config), database.MustToDBUUIDArray(*query.RequiredQueryIDs)),
-	)
-	pool.ExpectQuery("name: GetQueryWithVersion :one").WithArgs(database.MustToDBUUID(query.ID), &query.Version).WillReturnRows(
-		pgxmock.NewRows([]string{"id", "type", "activeVersion", "latestVersion", "config", "requiredIds"}).
-			AddRow(database.MustToDBUUID(query.ID), repository.QuerytypeJsonExtractor, query.Version, query.Version, []byte(*query.Config), database.MustToDBUUIDArray(*query.RequiredQueryIDs)),
-	)
-	requiredResultId := uuid.New()
-	strVal := "{\"examplekey\":\"example_value\"}"
-	pool.ExpectQuery("name: ListQueryRequirementValues :many").WithArgs(database.MustToDBUUID(query.ID), &query.Version, database.MustToDBUUID(params.DocumentID)).
-		WillReturnRows(
-			pgxmock.NewRows([]string{"id", "queryId", "type", "value"}).
-				AddRow(database.MustToDBUUID(requiredResultId), database.MustToDBUUID((*query.RequiredQueryIDs)[0]), repository.QuerytypeContextFull, &strVal),
+		pool.ExpectQuery("name: GetQueryWithVersion :one").WithArgs(database.MustToDBUUID(query.ID), &query.Version).WillReturnRows(
+			pgxmock.NewRows([]string{"id", "type", "activeVersion", "latestVersion", "config", "requiredIds"}).
+				AddRow(database.MustToDBUUID(query.ID), repository.QuerytypeJsonExtractor, query.Version, query.Version, []byte(*query.Config), database.MustToDBUUIDArray(*query.RequiredQueryIDs)),
 		)
-	pool.ExpectQuery("name: GetActiveQueryConfig :one").WithArgs(database.MustToDBUUID(query.ID)).WillReturnRows(
-		pgxmock.NewRows([]string{"config"}).
-			AddRow([]byte(qcfg)),
-	)
-	pool.ExpectBegin()
-	resultId := uuid.New()
-	pool.ExpectQuery("name: AddResult :one").WithArgs(database.MustToDBUUID(query.ID), pgxmock.AnyArg(), database.MustToDBUUID(textEntryId), query.Version).
-		WillReturnRows(
-			pgxmock.NewRows([]string{"id"}).
-				AddRow(database.MustToDBUUID(resultId)),
-		)
-	pool.ExpectQuery("name: ListQueryRequirementValues :many").WithArgs(database.MustToDBUUID(query.ID), &query.Version, database.MustToDBUUID(params.DocumentID)).
-		WillReturnRows(
-			pgxmock.NewRows([]string{"id", "queryId", "type", "value"}).
-				AddRow(database.MustToDBUUID(requiredResultId), database.MustToDBUUID((*query.RequiredQueryIDs)[0]), repository.QuerytypeContextFull, &strVal),
-		)
-	pool.ExpectExec("name: AddResultDependency :exec").WithArgs(database.MustToDBUUID(resultId), database.MustToDBUUID(requiredResultId)).
-		WillReturnResult(pgxmock.NewResult("", 1))
-	pool.ExpectCommit()
-	pool.ExpectQuery("name: ListQueryDirectDependentsByDocumentID :many").WithArgs(database.MustToDBUUID(query.ID), database.MustToDBUUID(doc.DocumentID)).
-		WillReturnRows(
-			pgxmock.NewRows([]string{"queryId"}).
-				AddRow(database.MustToDBUUID((*query.RequiredQueryIDs)[0])),
+		requiredResultId := uuid.New()
+		strVal := "{\"examplekey\":\"example_value\"}"
+		pool.ExpectQuery("name: ListQueryRequirementValues :many").WithArgs(database.MustToDBUUID(query.ID), &query.Version, database.MustToDBUUID(params.DocumentID)).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"id", "queryId", "type", "value"}).
+					AddRow(database.MustToDBUUID(requiredResultId), database.MustToDBUUID((*query.RequiredQueryIDs)[0]), repository.QuerytypeContextFull, &strVal),
+			)
+		pool.ExpectQuery("name: GetActiveQueryConfig :one").WithArgs(database.MustToDBUUID(query.ID)).WillReturnRows(
+			pgxmock.NewRows([]string{"config"}).
+				AddRow([]byte(qcfg)),
 		)
+		pool.ExpectBegin()
+		resultId := uuid.New()
+		pool.ExpectQuery("name: AddResult :one").WithArgs(database.MustToDBUUID(query.ID), pgxmock.AnyArg(), database.MustToDBUUID(textEntryId), query.Version).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"id"}).
+					AddRow(database.MustToDBUUID(resultId)),
+			)
+		pool.ExpectQuery("name: ListQueryRequirementValues :many").WithArgs(database.MustToDBUUID(query.ID), &query.Version, database.MustToDBUUID(params.DocumentID)).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"id", "queryId", "type", "value"}).
+					AddRow(database.MustToDBUUID(requiredResultId), database.MustToDBUUID((*query.RequiredQueryIDs)[0]), repository.QuerytypeContextFull, &strVal),
+			)
+		pool.ExpectExec("name: AddResultDependency :exec").WithArgs(database.MustToDBUUID(resultId), database.MustToDBUUID(requiredResultId)).
+			WillReturnResult(pgxmock.NewResult("", 1))
+		pool.ExpectCommit()
+		pool.ExpectQuery("name: ListQueryDirectDependentsByDocumentID :many").WithArgs(database.MustToDBUUID(query.ID), database.MustToDBUUID(doc.DocumentID)).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"queryId"}).
+					AddRow(database.MustToDBUUID((*query.RequiredQueryIDs)[0])),
+			)
 
-	mockSQS.EXPECT().
-		SendMessage(
-			mock.Anything,
-			mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
-				return *in.QueueUrl == cfg.QueryURL && *in.MessageBody == fmt.Sprintf("{\"document_id\":\"%s\",\"query_id\":\"%s\"}", params.DocumentID.String(), (*query.RequiredQueryIDs)[0].String())
-			}),
-			mock.Anything,
-		).
-		Return(&sqs.SendMessageOutput{}, nil)
+		mockSQS.EXPECT().
+			SendMessage(
+				mock.Anything,
+				mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
+					return *in.QueueUrl == cfg.QueryURL && *in.MessageBody == fmt.Sprintf("{\"document_id\":\"%s\",\"query_id\":\"%s\"}", params.DocumentID.String(), (*query.RequiredQueryIDs)[0].String())
+				}),
+				mock.Anything,
+			).
+			Return(&sqs.SendMessageOutput{}, nil)
 
-	err = runner.Process(ctx, msg)
-	assert.NoError(t, err)
+		assert.True(t, runner.Process(ctx, msg))
+	})
+	t.Run("invalid body", func(t *testing.T) {
+		body := "definitely invalid"
+		msgId := "id"
+		msg := &types.Message{
+			Body:      &body,
+			MessageId: &msgId,
+		}
+
+		assert.True(t, runner.Process(ctx, msg))
+	})
 }

api/querySyncRunner/runner.go

@@ -3,6 +3,7 @@ package querysyncrunner
 import (
 	"context"
 	"encoding/json"
+	"log/slog"
 
 	querysync "queryorchestration/internal/query/sync"
 
@@ -34,22 +35,25 @@ type Body struct {
 	ID uuid.UUID `json:"id" validate:"required,uuid"`
 }
 
-func (s *Runner) Process(ctx context.Context, req *types.Message) error {
+func (s *Runner) Process(ctx context.Context, req *types.Message) bool {
 	var body Body
 	err := json.Unmarshal([]byte(*req.Body), &body)
 	if err != nil {
-		return err
+		slog.Error("parsing body", "messageId", *req.MessageId, "body", *req.Body)
+		return true
 	}
 
 	err = s.validator.Struct(body)
 	if err != nil {
-		return err
+		slog.Error("invalid body", "messageId", *req.MessageId, "error", err)
+		return true
 	}
 
 	err = s.svc.QuerySync.Sync(ctx, body.ID)
 	if err != nil {
-		return err
+		slog.Error("unable to process", "error", err)
+		return false
 	}
 
-	return nil
+	return true
 }

api/querySyncRunner/runner_test.go

@@ -52,41 +52,52 @@ func TestQueryRunner(t *testing.T) {
 	})
 	assert.NotNil(t, runner)
 
-	doc := querysyncrunner.Body{
-		ID: uuid.New(),
-	}
-	bodyBytes, err := json.Marshal(doc)
-	assert.NoError(t, err)
-	body := string(bodyBytes)
-	msg := &types.Message{
-		Body: &body,
-	}
+	t.Run("valid", func(t *testing.T) {
+		doc := querysyncrunner.Body{
+			ID: uuid.New(),
+		}
+		bodyBytes, err := json.Marshal(doc)
+		assert.NoError(t, err)
+		body := string(bodyBytes)
+		msg := &types.Message{
+			Body: &body,
+		}
 
-	qs := []uuid.UUID{
-		uuid.New(),
-	}
+		qs := []uuid.UUID{
+			uuid.New(),
+		}
 
-	pool.ExpectQuery("name: IsDocumentTextExtracted :one").WithArgs(database.MustToDBUUID(doc.ID)).
-		WillReturnRows(
-			pgxmock.NewRows([]string{"isextracted"}).
-				AddRow(true),
-		)
-	pool.ExpectQuery("name: ListUnsyncedNoDepsQueriesByDocId :many").WithArgs(database.MustToDBUUID(doc.ID)).
-		WillReturnRows(
-			pgxmock.NewRows([]string{"id"}).
-				AddRow(database.MustToDBUUID(qs[0])),
-		)
+		pool.ExpectQuery("name: IsDocumentTextExtracted :one").WithArgs(database.MustToDBUUID(doc.ID)).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"isextracted"}).
+					AddRow(true),
+			)
+		pool.ExpectQuery("name: ListUnsyncedNoDepsQueriesByDocId :many").WithArgs(database.MustToDBUUID(doc.ID)).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"id"}).
+					AddRow(database.MustToDBUUID(qs[0])),
+			)
 
-	mockSQS.EXPECT().
-		SendMessage(
-			mock.Anything,
-			mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
-				return *in.QueueUrl == cfg.QueryURL && *in.MessageBody == fmt.Sprintf("{\"document_id\":\"%s\",\"query_id\":\"%s\"}", doc.ID.String(), qs[0].String())
-			}),
-			mock.Anything,
-		).
-		Return(&sqs.SendMessageOutput{}, nil)
+		mockSQS.EXPECT().
+			SendMessage(
+				mock.Anything,
+				mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
+					return *in.QueueUrl == cfg.QueryURL && *in.MessageBody == fmt.Sprintf("{\"document_id\":\"%s\",\"query_id\":\"%s\"}", doc.ID.String(), qs[0].String())
+				}),
+				mock.Anything,
+			).
+			Return(&sqs.SendMessageOutput{}, nil)
 
-	err = runner.Process(ctx, msg)
-	assert.NoError(t, err)
+		assert.True(t, runner.Process(ctx, msg))
+	})
+	t.Run("invalid body", func(t *testing.T) {
+		body := "definitely invalid"
+		msgId := "id"
+		msg := &types.Message{
+			Body:      &body,
+			MessageId: &msgId,
+		}
+
+		assert.True(t, runner.Process(ctx, msg))
+	})
 }

api/queryVersionSyncRunner/runner.go

@@ -3,6 +3,7 @@ package queryversionsyncrunner
 import (
 	"context"
 	"encoding/json"
+	"log/slog"
 
 	queryversionsync "queryorchestration/internal/query/versionsync"
 
@@ -34,22 +35,25 @@ type Body struct {
 	ID uuid.UUID `json:"id" validate:"required,uuid"`
 }
 
-func (s *Runner) Process(ctx context.Context, req *types.Message) error {
+func (s *Runner) Process(ctx context.Context, req *types.Message) bool {
 	var body Body
 	err := json.Unmarshal([]byte(*req.Body), &body)
 	if err != nil {
-		return err
+		slog.Error("parsing body", "messageId", *req.MessageId, "body", *req.Body)
+		return true
 	}
 
 	err = s.validator.Struct(body)
 	if err != nil {
-		return err
+		slog.Error("invalid body", "messageId", *req.MessageId, "error", err)
+		return true
 	}
 
 	err = s.svc.Sync.Sync(ctx, body.ID)
 	if err != nil {
-		return err
+		slog.Error("unable to process", "error", err)
+		return false
 	}
 
-	return nil
+	return true
 }

api/queryVersionSyncRunner/runner_test.go

@@ -49,47 +49,58 @@ func TestQueryRunner(t *testing.T) {
 	})
 	assert.NotNil(t, runner)
 
-	doc := queryversionsyncrunner.Body{
-		ID: uuid.New(),
-	}
-	bodyBytes, err := json.Marshal(doc)
-	assert.NoError(t, err)
-	body := string(bodyBytes)
-	msg := &types.Message{
-		Body: &body,
-	}
+	t.Run("valid", func(t *testing.T) {
+		doc := queryversionsyncrunner.Body{
+			ID: uuid.New(),
+		}
+		bodyBytes, err := json.Marshal(doc)
+		assert.NoError(t, err)
+		body := string(bodyBytes)
+		msg := &types.Message{
+			Body: &body,
+		}
 
-	jobIds := []uuid.UUID{
-		uuid.New(),
-		uuid.New(),
-	}
+		jobIds := []uuid.UUID{
+			uuid.New(),
+			uuid.New(),
+		}
 
-	pool.ExpectQuery("name: ListQueryJobIDs :many").WithArgs(database.MustToDBUUID(doc.ID)).
-		WillReturnRows(
-			pgxmock.NewRows([]string{"jobId"}).
-				AddRow(database.MustToDBUUID(jobIds[0])).
-				AddRow(database.MustToDBUUID(jobIds[1])),
-		)
+		pool.ExpectQuery("name: ListQueryJobIDs :many").WithArgs(database.MustToDBUUID(doc.ID)).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"jobId"}).
+					AddRow(database.MustToDBUUID(jobIds[0])).
+					AddRow(database.MustToDBUUID(jobIds[1])),
+			)
 
-	mockSQS.EXPECT().
-		SendMessage(
-			mock.Anything,
-			mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
-				return *in.QueueUrl == cfg.JobSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", jobIds[0].String())
-			}),
-			mock.Anything,
-		).
-		Return(&sqs.SendMessageOutput{}, nil)
-	mockSQS.EXPECT().
-		SendMessage(
-			mock.Anything,
-			mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
-				return *in.QueueUrl == cfg.JobSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", jobIds[1].String())
-			}),
-			mock.Anything,
-		).
-		Return(&sqs.SendMessageOutput{}, nil)
+		mockSQS.EXPECT().
+			SendMessage(
+				mock.Anything,
+				mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
+					return *in.QueueUrl == cfg.JobSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", jobIds[0].String())
+				}),
+				mock.Anything,
+			).
+			Return(&sqs.SendMessageOutput{}, nil)
+		mockSQS.EXPECT().
+			SendMessage(
+				mock.Anything,
+				mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
+					return *in.QueueUrl == cfg.JobSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", jobIds[1].String())
+				}),
+				mock.Anything,
+			).
+			Return(&sqs.SendMessageOutput{}, nil)
 
-	err = runner.Process(ctx, msg)
-	assert.NoError(t, err)
+		assert.True(t, runner.Process(ctx, msg))
+	})
+	t.Run("invalid body", func(t *testing.T) {
+		body := "definitely invalid"
+		msgId := "id"
+		msg := &types.Message{
+			Body:      &body,
+			MessageId: &msgId,
+		}
+
+		assert.True(t, runner.Process(ctx, msg))
+	})
 }

database/queries/clean.sql

@@ -3,7 +3,12 @@ SELECT EXISTS(
     SELECT 1 FROM currentCleanEntries WHERE documentId = @documentId
 );
 
--- name: GetDocumentCleanEntry :one
+-- name: GetCleanEntry :one
+SELECT id, documentId, bucket, key, version, mimetype, fail
+    FROM currentCleanEntries
+    WHERE id = @cleanId;
+
+-- name: GetCleanEntryByDocId :one
 SELECT id, documentId, bucket, key, version, mimetype, fail
     FROM currentCleanEntries
     WHERE documentId = @documentId;

database/queries/text.sql

@@ -6,7 +6,7 @@ SELECT EXISTS(
 -- name: AddDocumentTextEntry :exec
 INSERT INTO documentTextExtractions (version, bucket, key, cleanEntryId) VALUES ($1, $2, $3, $4);
 
--- name: GetDocumentTextEntry :one
+-- name: GetTextEntryByDocId :one
 SELECT id, documentId, bucket, key, version, cleanEntryId
     FROM currentTextEntries
     WHERE documentId = @documentId;

internal/database/repository/clean.sql.go

@@ -56,19 +56,45 @@ func (q *Queries) AddDocumentCleanEntry(ctx context.Context, arg *AddDocumentCle
 	return err
 }
 
-const getDocumentCleanEntry = `-- name: GetDocumentCleanEntry :one
+const getCleanEntry = `-- name: GetCleanEntry :one
+SELECT id, documentId, bucket, key, version, mimetype, fail
+    FROM currentCleanEntries
+    WHERE id = $1
+`
+
+// GetCleanEntry
+//
+//	SELECT id, documentId, bucket, key, version, mimetype, fail
+//	    FROM currentCleanEntries
+//	    WHERE id = $1
+func (q *Queries) GetCleanEntry(ctx context.Context, cleanid pgtype.UUID) (*Currentcleanentry, error) {
+	row := q.db.QueryRow(ctx, getCleanEntry, cleanid)
+	var i Currentcleanentry
+	err := row.Scan(
+		&i.ID,
+		&i.Documentid,
+		&i.Bucket,
+		&i.Key,
+		&i.Version,
+		&i.Mimetype,
+		&i.Fail,
+	)
+	return &i, err
+}
+
+const getCleanEntryByDocId = `-- name: GetCleanEntryByDocId :one
 SELECT id, documentId, bucket, key, version, mimetype, fail
     FROM currentCleanEntries
     WHERE documentId = $1
 `
 
-// GetDocumentCleanEntry
+// GetCleanEntryByDocId
 //
 //	SELECT id, documentId, bucket, key, version, mimetype, fail
 //	    FROM currentCleanEntries
 //	    WHERE documentId = $1
-func (q *Queries) GetDocumentCleanEntry(ctx context.Context, documentid pgtype.UUID) (*Currentcleanentry, error) {
-	row := q.db.QueryRow(ctx, getDocumentCleanEntry, documentid)
+func (q *Queries) GetCleanEntryByDocId(ctx context.Context, documentid pgtype.UUID) (*Currentcleanentry, error) {
+	row := q.db.QueryRow(ctx, getCleanEntryByDocId, documentid)
 	var i Currentcleanentry
 	err := row.Scan(
 		&i.ID,

internal/database/repository/clean_test.go

@@ -47,9 +47,6 @@ func TestClean(t *testing.T) {
 	assert.NoError(t, err)
 	assert.False(t, isclean)
 
-	_, err = queries.GetDocumentCleanEntry(ctx, id)
-	assert.Error(t, err)
-
 	bucket := "example_bucket"
 	key := "example_key"
 	mimetype := repository.NullCleanmimetype{
@@ -84,7 +81,7 @@ func TestClean(t *testing.T) {
 	assert.NoError(t, err)
 	assert.True(t, isclean)
 
-	clean, err := queries.GetDocumentCleanEntry(ctx, id)
+	clean, err := queries.GetCleanEntry(ctx, failcleanid)
 	assert.NoError(t, err)
 	assert.Equal(t, id, clean.Documentid)
 	assert.Nil(t, clean.Bucket)
@@ -110,7 +107,7 @@ func TestClean(t *testing.T) {
 	assert.NoError(t, err)
 	assert.True(t, isclean)
 
-	clean, err = queries.GetDocumentCleanEntry(ctx, id)
+	clean, err = queries.GetCleanEntry(ctx, cleanid)
 	assert.NoError(t, err)
 	assert.Equal(t, id, clean.Documentid)
 	assert.False(t, clean.Fail.Valid)
@@ -120,6 +117,10 @@ func TestClean(t *testing.T) {
 	assert.Equal(t, int32(2), clean.Version)
 	assert.Equal(t, cleanid, clean.ID)
 
+	docclean, err := queries.GetCleanEntryByDocId(ctx, id)
+	assert.NoError(t, err)
+	assert.EqualExportedValues(t, clean, docclean)
+
 	recent, err := queries.GetMostRecentDocumentCleanEntry(ctx, id)
 	assert.NoError(t, err)
 	assert.Equal(t, clean.Documentid, recent.Documentid)

internal/database/repository/job_test.go

@@ -293,14 +293,11 @@ func TestJobSync(t *testing.T) {
 	assert.NoError(t, err)
 	assert.False(t, isSynced)
 
-	cleanentry, err := queries.GetDocumentCleanEntry(ctx, documentID)
-	assert.NoError(t, err)
-
 	err = queries.AddDocumentTextEntry(ctx, &repository.AddDocumentTextEntryParams{
 		Version:      1,
 		Bucket:       "hi",
 		Key:          "hello",
-		Cleanentryid: cleanentry.ID,
+		Cleanentryid: cleantwoid,
 	})
 	assert.NoError(t, err)
 
@@ -308,7 +305,7 @@ func TestJobSync(t *testing.T) {
 	assert.NoError(t, err)
 	assert.False(t, isSynced)
 
-	textentry, err := queries.GetDocumentTextEntry(ctx, documentID)
+	textentry, err := queries.GetTextEntryByDocId(ctx, documentID)
 	assert.NoError(t, err)
 
 	depresultid, err := queries.AddResult(ctx, &repository.AddResultParams{
@@ -395,7 +392,7 @@ func TestJobSync(t *testing.T) {
 		Version:      1,
 		Bucket:       "hi",
 		Key:          "hello",
-		Cleanentryid: cleanentry.ID,
+		Cleanentryid: cleantwoid,
 	})
 	assert.NoError(t, err)
 
@@ -403,7 +400,7 @@ func TestJobSync(t *testing.T) {
 	assert.NoError(t, err)
 	assert.False(t, isSynced)
 
-	textentry, err = queries.GetDocumentTextEntry(ctx, documentID)
+	textentry, err = queries.GetTextEntryByDocId(ctx, documentID)
 	assert.NoError(t, err)
 
 	depresultid, err = queries.AddResult(ctx, &repository.AddResultParams{
@@ -460,14 +457,11 @@ func TestJobSync(t *testing.T) {
 	assert.NoError(t, err)
 	assert.False(t, isSynced)
 
-	cleanentry, err = queries.GetDocumentCleanEntry(ctx, documentID)
-	assert.NoError(t, err)
-
 	err = queries.AddDocumentTextEntry(ctx, &repository.AddDocumentTextEntryParams{
 		Version:      1,
 		Bucket:       "hi",
 		Key:          "hello",
-		Cleanentryid: cleanentry.ID,
+		Cleanentryid: cleanthreeid,
 	})
 	assert.NoError(t, err)
 
@@ -475,7 +469,7 @@ func TestJobSync(t *testing.T) {
 	assert.NoError(t, err)
 	assert.False(t, isSynced)
 
-	textentry, err = queries.GetDocumentTextEntry(ctx, documentID)
+	textentry, err = queries.GetTextEntryByDocId(ctx, documentID)
 	assert.NoError(t, err)
 
 	depresultid, err = queries.AddResult(ctx, &repository.AddResultParams{

internal/database/repository/result_test.go

@@ -94,14 +94,11 @@ func TestResults(t *testing.T) {
 	assert.NoError(t, err)
 	assert.False(t, issynced)
 
-	cleanentry, err := queries.GetDocumentCleanEntry(ctx, documentID)
-	assert.NoError(t, err)
-
 	err = queries.AddDocumentTextEntry(ctx, &repository.AddDocumentTextEntryParams{
 		Version:      1,
 		Bucket:       "hi",
 		Key:          "hello",
-		Cleanentryid: cleanentry.ID,
+		Cleanentryid: cleanid,
 	})
 	assert.NoError(t, err)
 
@@ -109,7 +106,7 @@ func TestResults(t *testing.T) {
 	assert.NoError(t, err)
 	assert.True(t, issynced)
 
-	textentry, err := queries.GetDocumentTextEntry(ctx, documentID)
+	textentry, err := queries.GetTextEntryByDocId(ctx, documentID)
 	assert.NoError(t, err)
 
 	err = queries.AddCollectorQuery(ctx, &repository.AddCollectorQueryParams{
@@ -255,16 +252,14 @@ func TestResultValues(t *testing.T) {
 		Version: 1,
 	})
 	assert.NoError(t, err)
-	cleanentry, err := queries.GetDocumentCleanEntry(ctx, documentID)
-	assert.NoError(t, err)
 	err = queries.AddDocumentTextEntry(ctx, &repository.AddDocumentTextEntryParams{
 		Version:      1,
 		Bucket:       "hi",
 		Key:          "hello",
-		Cleanentryid: cleanentry.ID,
+		Cleanentryid: cleanid,
 	})
 	assert.NoError(t, err)
-	textentry, err := queries.GetDocumentTextEntry(ctx, documentID)
+	textentry, err := queries.GetTextEntryByDocId(ctx, documentID)
 	assert.NoError(t, err)
 
 	result := repository.AddResultParams{
@@ -465,18 +460,15 @@ func TestUnsyncedNoDepsQueries(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Len(t, qs, 0)
 
-	cleanentry, err := queries.GetDocumentCleanEntry(ctx, documentID)
-	assert.NoError(t, err)
-
 	err = queries.AddDocumentTextEntry(ctx, &repository.AddDocumentTextEntryParams{
 		Version:      1,
 		Bucket:       "hi",
 		Key:          "hello",
-		Cleanentryid: cleanentry.ID,
+		Cleanentryid: cleanid,
 	})
 	assert.NoError(t, err)
 
-	textentry, err := queries.GetDocumentTextEntry(ctx, documentID)
+	textentry, err := queries.GetTextEntryByDocId(ctx, documentID)
 	assert.NoError(t, err)
 
 	qs, err = queries.ListUnsyncedNoDepsQueriesByDocId(ctx, documentID)
@@ -541,18 +533,15 @@ func TestUnsyncedNoDepsQueries(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Len(t, qs, 0)
 
-	cleantwoentry, err := queries.GetDocumentCleanEntry(ctx, documentTwoID)
-	assert.NoError(t, err)
-
 	err = queries.AddDocumentTextEntry(ctx, &repository.AddDocumentTextEntryParams{
 		Version:      1,
 		Bucket:       "hi",
 		Key:          "hello",
-		Cleanentryid: cleantwoentry.ID,
+		Cleanentryid: cleantwoid,
 	})
 	assert.NoError(t, err)
 
-	texttwoentry, err := queries.GetDocumentTextEntry(ctx, documentTwoID)
+	texttwoentry, err := queries.GetTextEntryByDocId(ctx, documentTwoID)
 	assert.NoError(t, err)
 
 	_, err = queries.AddResult(ctx, &repository.AddResultParams{

internal/database/repository/text.sql.go

@@ -35,19 +35,19 @@ func (q *Queries) AddDocumentTextEntry(ctx context.Context, arg *AddDocumentText
 	return err
 }
 
-const getDocumentTextEntry = `-- name: GetDocumentTextEntry :one
+const getTextEntryByDocId = `-- name: GetTextEntryByDocId :one
 SELECT id, documentId, bucket, key, version, cleanEntryId
     FROM currentTextEntries
     WHERE documentId = $1
 `
 
-// GetDocumentTextEntry
+// GetTextEntryByDocId
 //
 //	SELECT id, documentId, bucket, key, version, cleanEntryId
 //	    FROM currentTextEntries
 //	    WHERE documentId = $1
-func (q *Queries) GetDocumentTextEntry(ctx context.Context, documentid pgtype.UUID) (*Currenttextentry, error) {
-	row := q.db.QueryRow(ctx, getDocumentTextEntry, documentid)
+func (q *Queries) GetTextEntryByDocId(ctx context.Context, documentid pgtype.UUID) (*Currenttextentry, error) {
+	row := q.db.QueryRow(ctx, getTextEntryByDocId, documentid)
 	var i Currenttextentry
 	err := row.Scan(
 		&i.ID,

internal/database/repository/text_test.go

@@ -62,9 +62,6 @@ func TestTextExtraction(t *testing.T) {
 	})
 	assert.NoError(t, err)
 
-	cleanentry, err := queries.GetDocumentCleanEntry(ctx, id)
-	assert.NoError(t, err)
-
 	isextract, err := queries.IsDocumentTextExtracted(ctx, id)
 	assert.NoError(t, err)
 	assert.False(t, isextract)
@@ -73,7 +70,7 @@ func TestTextExtraction(t *testing.T) {
 		Version:      1,
 		Bucket:       bucket,
 		Key:          key,
-		Cleanentryid: cleanentry.ID,
+		Cleanentryid: cleanid,
 	})
 	assert.NoError(t, err)
 
@@ -81,7 +78,7 @@ func TestTextExtraction(t *testing.T) {
 	assert.NoError(t, err)
 	assert.True(t, isextract)
 
-	text, err := queries.GetDocumentTextEntry(ctx, id)
+	text, err := queries.GetTextEntryByDocId(ctx, id)
 	assert.NoError(t, err)
 	assert.Equal(t, id, text.Documentid)
 	assert.Equal(t, bucket, text.Bucket)

internal/document/text/create.go

@@ -20,7 +20,15 @@ func (s *Service) Extract(ctx context.Context, id uuid.UUID) error {
 	}
 
 	if !isextracted {
-		err = s.extract(ctx, id)
+		entry, err := s.cfg.GetDBQueries().GetCleanEntryByDocId(ctx, database.MustToDBUUID(id))
+		if err != nil {
+			return err
+		} else if entry.Fail.Valid {
+			slog.Error("document has no clean document", "id", id)
+			return nil
+		}
+
+		err = s.extract(ctx, database.MustToUUID(entry.ID))
 		if err != nil {
 			return err
 		}

internal/document/text/create_test.go

@@ -47,41 +47,65 @@ func TestCreate(t *testing.T) {
 		},
 	}
 
-	id := uuid.New()
-	inloc := document.Location{
-		Bucket: "bucket_name",
-		Key:    "/i/am/here",
-	}
+	t.Run("valid", func(t *testing.T) {
+		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),
-	)
-	cleanId := database.MustToDBUUID(uuid.New())
-	mimeType := "application/pdf"
-	dbmimetype := repository.NullCleanmimetype{
-		Valid:         true,
-		Cleanmimetype: repository.Cleanmimetype(mimeType),
-	}
-	pool.ExpectQuery("name: GetDocumentCleanEntry :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows(
-		pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}).
-			AddRow(cleanId, database.MustToDBUUID(id), &inloc.Bucket, &inloc.Key, int32(1), dbmimetype, repository.NullCleanfailtype{}),
-	)
-	pool.ExpectExec("name: AddDocumentTextEntry :exec").WithArgs(int32(1), inloc.Bucket, inloc.Key, cleanId).
-		WillReturnResult(pgxmock.NewResult("", 1))
+		pool.ExpectQuery("name: IsDocumentTextExtracted :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows(
+			pgxmock.NewRows([]string{"isextracted"}).
+				AddRow(false),
+		)
+		cleanId := database.MustToDBUUID(uuid.New())
+		mimeType := "application/pdf"
+		dbmimetype := repository.NullCleanmimetype{
+			Valid:         true,
+			Cleanmimetype: repository.Cleanmimetype(mimeType),
+		}
+		pool.ExpectQuery("name: GetCleanEntryByDocId :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows(
+			pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}).
+				AddRow(cleanId, database.MustToDBUUID(id), &inloc.Bucket, &inloc.Key, int32(1), dbmimetype, repository.NullCleanfailtype{}),
+		)
+		pool.ExpectQuery("name: GetCleanEntry :one").WithArgs(cleanId).WillReturnRows(
+			pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}).
+				AddRow(cleanId, database.MustToDBUUID(id), &inloc.Bucket, &inloc.Key, int32(1), dbmimetype, repository.NullCleanfailtype{}),
+		)
+		pool.ExpectExec("name: AddDocumentTextEntry :exec").WithArgs(int32(1), inloc.Bucket, inloc.Key, cleanId).
+			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)
+		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.Extract(ctx, id)
-	assert.NoError(t, err)
+		err = svc.Extract(ctx, id)
+		assert.NoError(t, err)
+	})
+	t.Run("invalid clean", func(t *testing.T) {
+		id := uuid.New()
+		fail := repository.NullCleanfailtype{
+			Valid:         true,
+			Cleanfailtype: repository.CleanfailtypeInvalidMimetype,
+		}
+
+		pool.ExpectQuery("name: IsDocumentTextExtracted :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows(
+			pgxmock.NewRows([]string{"isextracted"}).AddRow(false),
+		)
+		pool.ExpectQuery("name: GetCleanEntryByDocId :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows(
+			pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}).
+				AddRow(database.MustToDBUUID(id), database.MustToDBUUID(uuid.New()), nil, nil, int32(1), nil, fail),
+		)
+
+		err = svc.Extract(ctx, id)
+		assert.NoError(t, err)
+	})
 }
 
 func TestInformClean(t *testing.T) {

internal/document/text/extract.go

@@ -11,32 +11,22 @@ import (
 	"github.com/google/uuid"
 )
 
-type ExtractionParams struct {
-	ID       uuid.UUID
-	Location document.Location
-}
-
-func (s *Service) executeExtraction(params *ExtractionParams) (*document.Location, error) {
-	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)
+func (s *Service) executeExtraction(ctx context.Context, cleanId uuid.UUID) (*document.Location, error) {
+	entry, err := s.cfg.GetDBQueries().GetCleanEntry(ctx, database.MustToDBUUID(cleanId))
 	if err != nil {
-		return err
+		return nil, err
 	} else if entry.Fail.Valid {
-		return fmt.Errorf("no valid cleaning")
+		return nil, fmt.Errorf("no valid cleaning")
 	}
 
-	outLocation, err := s.executeExtraction(&ExtractionParams{
-		ID: id,
-		Location: document.Location{
-			Bucket: *entry.Bucket,
-			Key:    *entry.Key,
-		},
-	})
+	return &document.Location{
+		Bucket: *entry.Bucket,
+		Key:    *entry.Key,
+	}, nil
+}
+
+func (s *Service) extract(ctx context.Context, cleanId uuid.UUID) error {
+	outLocation, err := s.executeExtraction(ctx, cleanId)
 	if err != nil {
 		return err
 	}
@@ -47,7 +37,7 @@ func (s *Service) extract(ctx context.Context, id uuid.UUID) error {
 		Version:      version,
 		Bucket:       outLocation.Bucket,
 		Key:          outLocation.Key,
-		Cleanentryid: entry.ID,
+		Cleanentryid: database.MustToDBUUID(cleanId),
 	})
 	if err != nil {
 		return err

internal/document/text/extract_test.go

@@ -32,39 +32,80 @@ func TestExtract(t *testing.T) {
 		},
 	}
 
-	id := uuid.New()
-	inloc := document.Location{
-		Bucket: "bucket_name",
-		Key:    "/i/am/here",
-	}
+	t.Run("successful clean", func(t *testing.T) {
+		id := uuid.New()
+		inloc := document.Location{
+			Bucket: "bucket_name",
+			Key:    "/i/am/here",
+		}
 
-	cleanId := database.MustToDBUUID(uuid.New())
-	mimeType := "application/pdf"
-	dbmimetype := repository.NullCleanmimetype{
-		Valid:         true,
-		Cleanmimetype: repository.Cleanmimetype(mimeType),
-	}
-	pool.ExpectQuery("name: GetDocumentCleanEntry :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows(
-		pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}).
-			AddRow(cleanId, database.MustToDBUUID(id), &inloc.Bucket, &inloc.Key, int32(1), dbmimetype, repository.NullCleanfailtype{}),
-	)
-	pool.ExpectExec("name: AddDocumentTextEntry :exec").WithArgs(int32(1), inloc.Bucket, inloc.Key, cleanId).
-		WillReturnResult(pgxmock.NewResult("", 1))
+		cleanId := uuid.New()
+		dbCleanId := database.MustToDBUUID(cleanId)
+		mimeType := "application/pdf"
+		dbmimetype := repository.NullCleanmimetype{
+			Valid:         true,
+			Cleanmimetype: repository.Cleanmimetype(mimeType),
+		}
+		pool.ExpectQuery("name: GetCleanEntry :one").WithArgs(dbCleanId).WillReturnRows(
+			pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}).
+				AddRow(dbCleanId, database.MustToDBUUID(id), &inloc.Bucket, &inloc.Key, int32(1), dbmimetype, repository.NullCleanfailtype{}),
+		)
+		pool.ExpectExec("name: AddDocumentTextEntry :exec").WithArgs(int32(1), inloc.Bucket, inloc.Key, dbCleanId).
+			WillReturnResult(pgxmock.NewResult("", 1))
 
-	err = svc.extract(ctx, id)
-	assert.NoError(t, err)
+		err = svc.extract(ctx, cleanId)
+		assert.NoError(t, err)
+	})
+	t.Run("fail clean", func(t *testing.T) {
+		id := uuid.New()
+
+		cleanId := database.MustToDBUUID(uuid.New())
+		fail := repository.NullCleanfailtype{
+			Valid:         true,
+			Cleanfailtype: repository.CleanfailtypeInvalidMimetype,
+		}
+		pool.ExpectQuery("name: GetCleanEntry :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows(
+			pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}).
+				AddRow(cleanId, database.MustToDBUUID(id), nil, nil, int32(1), nil, fail),
+		)
+
+		err = svc.extract(ctx, id)
+		assert.Error(t, err)
+	})
 }
 
 func TestExecuteExtraction(t *testing.T) {
-	svc := Service{}
+	pool, err := pgxmock.NewPool()
+	require.NoError(t, err)
 
+	cfg := &DocTextConfig{}
+	cfg.DBPool = pool
+	cfg.DBQueries = repository.New(pool)
+
+	svc := Service{
+		cfg: cfg,
+		svc: &Services{
+			Version: textversion.New(cfg),
+		},
+	}
+
+	ctx := context.Background()
 	id := uuid.New()
-	location := document.Location{}
+	location := document.Location{
+		Bucket: "buck",
+		Key:    "key",
+	}
+	dbmimetype := repository.NullCleanmimetype{
+		Valid:         true,
+		Cleanmimetype: repository.CleanmimetypeApplicationPdf,
+	}
 
-	outloc, err := svc.executeExtraction(&ExtractionParams{
-		ID:       id,
-		Location: location,
-	})
+	pool.ExpectQuery("name: GetCleanEntry :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows(
+		pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}).
+			AddRow(database.MustToDBUUID(uuid.New()), database.MustToDBUUID(id), &location.Bucket, &location.Key, int32(1), dbmimetype, repository.NullCleanfailtype{}),
+	)
+
+	outloc, err := svc.executeExtraction(ctx, id)
 	assert.NoError(t, err)
 	assert.EqualExportedValues(t, location, *outloc)
 }

internal/query/result/set/set.go

@@ -20,7 +20,7 @@ type Set struct {
 
 func (s *Service) Set(ctx context.Context, params *Set) error {
 	dbid := database.MustToDBUUID(params.DocumentID)
-	textVersion, err := s.cfg.GetDBQueries().GetDocumentTextEntry(ctx, dbid)
+	textVersion, err := s.cfg.GetDBQueries().GetTextEntryByDocId(ctx, dbid)
 	if err != nil {
 		return err
 	}

internal/query/result/set/set_test.go

@@ -66,7 +66,7 @@ func TestSet(t *testing.T) {
 	}
 
 	textEntryId := uuid.New()
-	pool.ExpectQuery("name: GetDocumentTextEntry :one").WithArgs(database.MustToDBUUID(params.DocumentID)).
+	pool.ExpectQuery("name: GetTextEntryByDocId :one").WithArgs(database.MustToDBUUID(params.DocumentID)).
 		WillReturnRows(
 			pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "cleanEntryId"}).
 				AddRow(database.MustToDBUUID(textEntryId), database.MustToDBUUID(params.DocumentID), "buket", "/i/am/here", int32(1), database.MustToDBUUID(uuid.New())),

internal/query/sync/sync.go

@@ -2,7 +2,6 @@ package querysync
 
 import (
 	"context"
-	"fmt"
 	"log/slog"
 
 	"queryorchestration/internal/database"
@@ -17,7 +16,8 @@ func (s *Service) Sync(ctx context.Context, id uuid.UUID) error {
 	if err != nil {
 		return err
 	} else if !isextracted {
-		return fmt.Errorf("document text must be extracted")
+		slog.Error("document has no extracted text", "id", id)
+		return nil
 	}
 
 	unsyncedQueries, err := s.cfg.GetDBQueries().ListUnsyncedNoDepsQueriesByDocId(ctx, dbid)

internal/query/sync/sync_test.go

@@ -42,43 +42,56 @@ func TestSync(t *testing.T) {
 		},
 	}
 
-	id := uuid.New()
-	qs := []uuid.UUID{
-		uuid.New(),
-		uuid.New(),
-	}
+	t.Run("valid", func(t *testing.T) {
+		id := uuid.New()
+		qs := []uuid.UUID{
+			uuid.New(),
+			uuid.New(),
+		}
 
-	pool.ExpectQuery("name: IsDocumentTextExtracted :one").WithArgs(database.MustToDBUUID(id)).
-		WillReturnRows(
-			pgxmock.NewRows([]string{"isextracted"}).
-				AddRow(true),
-		)
-	pool.ExpectQuery("name: ListUnsyncedNoDepsQueriesByDocId :many").WithArgs(database.MustToDBUUID(id)).
-		WillReturnRows(
-			pgxmock.NewRows([]string{"id"}).
-				AddRow(database.MustToDBUUID(qs[0])).
-				AddRow(database.MustToDBUUID(qs[1])),
-		)
+		pool.ExpectQuery("name: IsDocumentTextExtracted :one").WithArgs(database.MustToDBUUID(id)).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"isextracted"}).
+					AddRow(true),
+			)
+		pool.ExpectQuery("name: ListUnsyncedNoDepsQueriesByDocId :many").WithArgs(database.MustToDBUUID(id)).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"id"}).
+					AddRow(database.MustToDBUUID(qs[0])).
+					AddRow(database.MustToDBUUID(qs[1])),
+			)
 
-	mockSQS.EXPECT().
-		SendMessage(
-			mock.Anything,
-			mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
-				return *in.QueueUrl == cfg.QueryURL && *in.MessageBody == fmt.Sprintf("{\"document_id\":\"%s\",\"query_id\":\"%s\"}", id.String(), qs[0].String())
-			}),
-			mock.Anything,
-		).
-		Return(&sqs.SendMessageOutput{}, nil)
+		mockSQS.EXPECT().
+			SendMessage(
+				mock.Anything,
+				mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
+					return *in.QueueUrl == cfg.QueryURL && *in.MessageBody == fmt.Sprintf("{\"document_id\":\"%s\",\"query_id\":\"%s\"}", id.String(), qs[0].String())
+				}),
+				mock.Anything,
+			).
+			Return(&sqs.SendMessageOutput{}, nil)
 
-	mockSQS.EXPECT().
-		SendMessage(
-			mock.Anything,
-			mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
-				return *in.QueueUrl == cfg.QueryURL && *in.MessageBody == fmt.Sprintf("{\"document_id\":\"%s\",\"query_id\":\"%s\"}", id.String(), qs[1].String())
-			}),
-			mock.Anything,
-		).
-		Return(&sqs.SendMessageOutput{}, nil)
+		mockSQS.EXPECT().
+			SendMessage(
+				mock.Anything,
+				mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
+					return *in.QueueUrl == cfg.QueryURL && *in.MessageBody == fmt.Sprintf("{\"document_id\":\"%s\",\"query_id\":\"%s\"}", id.String(), qs[1].String())
+				}),
+				mock.Anything,
+			).
+			Return(&sqs.SendMessageOutput{}, nil)
 
-	assert.Nil(t, svc.Sync(ctx, id))
+		assert.Nil(t, svc.Sync(ctx, id))
+	})
+	t.Run("not extracted", func(t *testing.T) {
+		id := uuid.New()
+
+		pool.ExpectQuery("name: IsDocumentTextExtracted :one").WithArgs(database.MustToDBUUID(id)).
+			WillReturnRows(
+				pgxmock.NewRows([]string{"isextracted"}).
+					AddRow(false),
+			)
+
+		assert.Nil(t, svc.Sync(ctx, id))
+	})
 }

internal/server/runner/listener.go

@@ -12,7 +12,7 @@ import (
 )
 
 type Controller interface {
-	Process(ctx context.Context, message *types.Message) error
+	Process(ctx context.Context, message *types.Message) bool
 }
 
 type ListenerConfig interface {

internal/server/runner/poll.go

@@ -2,6 +2,7 @@ package runner
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
 
@@ -58,12 +59,12 @@ func (c *Server) pollMessage(ctx context.Context) error {
 func (c *Server) processMessage(ctx context.Context, message *types.Message) error {
 	slog.Debug("processing message", "id", *message.MessageId, "body", *message.Body)
 
-	err := c.cfg.GetController().Process(ctx, message)
-	if err != nil {
-		return fmt.Errorf("message process fail: %v", err)
+	isProcessed := c.cfg.GetController().Process(ctx, message)
+	if !isProcessed {
+		return errors.New("message process fail")
 	}
 
-	err = c.cfg.DeleteFromQueue(ctx, &queue.DeleteParams{
+	err := c.cfg.DeleteFromQueue(ctx, &queue.DeleteParams{
 		QueueURL:      c.cfg.GetQueueURL(),
 		ReceiptHandle: message.ReceiptHandle,
 	})

internal/server/runner/poll_test.go

@@ -90,7 +90,7 @@ func TestPollMessage(t *testing.T) {
 				return *in.MessageId == id && *in.Body == body
 			}),
 		).
-		Return(nil)
+		Return(true)
 
 	mockSQS.EXPECT().
 		DeleteMessage(
@@ -137,7 +137,7 @@ func TestProcessMessage(t *testing.T) {
 				return *in.MessageId == id && *in.Body == body
 			}),
 		).
-		Return(nil)
+		Return(true)
 
 	mockSQS.EXPECT().
 		DeleteMessage(

mocks/runner/mock_Controller.go

@@ -24,18 +24,18 @@ func (_m *MockController) EXPECT() *MockController_Expecter {
 }
 
 // Process provides a mock function with given fields: ctx, message
-func (_m *MockController) Process(ctx context.Context, message *types.Message) error {
+func (_m *MockController) Process(ctx context.Context, message *types.Message) bool {
 	ret := _m.Called(ctx, message)
 
 	if len(ret) == 0 {
 		panic("no return value specified for Process")
 	}
 
-	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, *types.Message) error); ok {
+	var r0 bool
+	if rf, ok := ret.Get(0).(func(context.Context, *types.Message) bool); ok {
 		r0 = rf(ctx, message)
 	} else {
-		r0 = ret.Error(0)
+		r0 = ret.Get(0).(bool)
 	}
 
 	return r0
@@ -60,12 +60,12 @@ func (_c *MockController_Process_Call) Run(run func(ctx context.Context, message
 	return _c
 }
 
-func (_c *MockController_Process_Call) Return(_a0 error) *MockController_Process_Call {
+func (_c *MockController_Process_Call) Return(_a0 bool) *MockController_Process_Call {
 	_c.Call.Return(_a0)
 	return _c
 }
 
-func (_c *MockController_Process_Call) RunAndReturn(run func(context.Context, *types.Message) error) *MockController_Process_Call {
+func (_c *MockController_Process_Call) RunAndReturn(run func(context.Context, *types.Message) bool) *MockController_Process_Call {
 	_c.Call.Return(run)
 	return _c
 }