From 8f409e60a83827eb5520c66d7c8b7c78b6b34bc2 Mon Sep 17 00:00:00 2001 From: Michael McGuinness Date: Wed, 5 Mar 2025 19:49:03 +0000 Subject: [PATCH] Merged in bugfix/errhandling (pull request #89) Error handling * basic --- api/docCleanRunner/runner.go | 14 +- api/docCleanRunner/runner_test.go | 177 ++++++++------- api/docInitRunner/runner.go | 78 ++++--- api/docInitRunner/runner_test.go | 238 +++++++++++++++----- api/docSyncRunner/runner.go | 14 +- api/docSyncRunner/runner_test.go | 73 +++--- api/docTextRunner/runner.go | 14 +- api/docTextRunner/runner_test.go | 95 ++++---- api/jobSyncRunner/runner.go | 14 +- api/jobSyncRunner/runner_test.go | 65 +++--- api/queryRunner/runner.go | 14 +- api/queryRunner/runner_test.go | 159 +++++++------ api/querySyncRunner/runner.go | 14 +- api/querySyncRunner/runner_test.go | 77 ++++--- api/queryVersionSyncRunner/runner.go | 14 +- api/queryVersionSyncRunner/runner_test.go | 89 ++++---- database/queries/clean.sql | 7 +- database/queries/text.sql | 2 +- internal/database/repository/clean.sql.go | 34 ++- internal/database/repository/clean_test.go | 11 +- internal/database/repository/job_test.go | 18 +- internal/database/repository/result_test.go | 27 +-- internal/database/repository/text.sql.go | 8 +- internal/database/repository/text_test.go | 7 +- internal/document/text/create.go | 10 +- internal/document/text/create_test.go | 88 +++++--- internal/document/text/extract.go | 36 ++- internal/document/text/extract_test.go | 91 ++++++-- internal/query/result/set/set.go | 2 +- internal/query/result/set/set_test.go | 2 +- internal/query/sync/sync.go | 4 +- internal/query/sync/sync_test.go | 83 ++++--- internal/server/runner/listener.go | 2 +- internal/server/runner/poll.go | 9 +- internal/server/runner/poll_test.go | 4 +- mocks/runner/mock_Controller.go | 12 +- 36 files changed, 965 insertions(+), 641 deletions(-) diff --git a/api/docCleanRunner/runner.go b/api/docCleanRunner/runner.go index 9a66a8eb..9836166f 100644 --- a/api/docCleanRunner/runner.go +++ b/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 } diff --git a/api/docCleanRunner/runner_test.go b/api/docCleanRunner/runner_test.go index a6c516ee..0125c6d7 100644 --- a/api/docCleanRunner/runner_test.go +++ b/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 diff --git a/api/docInitRunner/runner.go b/api/docInitRunner/runner.go index 988bc874..dbd85af7 100644 --- a/api/docInitRunner/runner.go +++ b/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 +} diff --git a/api/docInitRunner/runner_test.go b/api/docInitRunner/runner_test.go index 6961163d..751824d0 100644 --- a/api/docInitRunner/runner_test.go +++ b/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")) + }) } diff --git a/api/docSyncRunner/runner.go b/api/docSyncRunner/runner.go index d5e20266..d3d99d35 100644 --- a/api/docSyncRunner/runner.go +++ b/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 } diff --git a/api/docSyncRunner/runner_test.go b/api/docSyncRunner/runner_test.go index 7bc29818..4324b8a3 100644 --- a/api/docSyncRunner/runner_test.go +++ b/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)) + }) } diff --git a/api/docTextRunner/runner.go b/api/docTextRunner/runner.go index e0dcb500..3cc8ec75 100644 --- a/api/docTextRunner/runner.go +++ b/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 } diff --git a/api/docTextRunner/runner_test.go b/api/docTextRunner/runner_test.go index 9c35897a..43ec2860 100644 --- a/api/docTextRunner/runner_test.go +++ b/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)) + }) } diff --git a/api/jobSyncRunner/runner.go b/api/jobSyncRunner/runner.go index bd92cc50..222c55b5 100644 --- a/api/jobSyncRunner/runner.go +++ b/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 } diff --git a/api/jobSyncRunner/runner_test.go b/api/jobSyncRunner/runner_test.go index 9e2bf294..cfcf4a0e 100644 --- a/api/jobSyncRunner/runner_test.go +++ b/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)) + }) } diff --git a/api/queryRunner/runner.go b/api/queryRunner/runner.go index 973599b1..e79e3889 100644 --- a/api/queryRunner/runner.go +++ b/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 } diff --git a/api/queryRunner/runner_test.go b/api/queryRunner/runner_test.go index 0c6a525b..54a12673 100644 --- a/api/queryRunner/runner_test.go +++ b/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)) + }) } diff --git a/api/querySyncRunner/runner.go b/api/querySyncRunner/runner.go index d524937a..d6eab03b 100644 --- a/api/querySyncRunner/runner.go +++ b/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 } diff --git a/api/querySyncRunner/runner_test.go b/api/querySyncRunner/runner_test.go index ecb9cf3a..0c4e0883 100644 --- a/api/querySyncRunner/runner_test.go +++ b/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)) + }) } diff --git a/api/queryVersionSyncRunner/runner.go b/api/queryVersionSyncRunner/runner.go index 5b4c6208..f947589a 100644 --- a/api/queryVersionSyncRunner/runner.go +++ b/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 } diff --git a/api/queryVersionSyncRunner/runner_test.go b/api/queryVersionSyncRunner/runner_test.go index 1472654d..4bc89b24 100644 --- a/api/queryVersionSyncRunner/runner_test.go +++ b/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)) + }) } diff --git a/database/queries/clean.sql b/database/queries/clean.sql index 31155ca6..17fcccc5 100644 --- a/database/queries/clean.sql +++ b/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; diff --git a/database/queries/text.sql b/database/queries/text.sql index 6fe2bda7..f45aff88 100644 --- a/database/queries/text.sql +++ b/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; diff --git a/internal/database/repository/clean.sql.go b/internal/database/repository/clean.sql.go index d7fdc45f..28c0b8cc 100644 --- a/internal/database/repository/clean.sql.go +++ b/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, diff --git a/internal/database/repository/clean_test.go b/internal/database/repository/clean_test.go index dc677ec7..f0aa1ea4 100644 --- a/internal/database/repository/clean_test.go +++ b/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) diff --git a/internal/database/repository/job_test.go b/internal/database/repository/job_test.go index e110a93d..4326ed63 100644 --- a/internal/database/repository/job_test.go +++ b/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{ diff --git a/internal/database/repository/result_test.go b/internal/database/repository/result_test.go index 4b22a782..71ecd02a 100644 --- a/internal/database/repository/result_test.go +++ b/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{ diff --git a/internal/database/repository/text.sql.go b/internal/database/repository/text.sql.go index e057e6df..aa925ab8 100644 --- a/internal/database/repository/text.sql.go +++ b/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, diff --git a/internal/database/repository/text_test.go b/internal/database/repository/text_test.go index 3bd48965..bcba4ca7 100644 --- a/internal/database/repository/text_test.go +++ b/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) diff --git a/internal/document/text/create.go b/internal/document/text/create.go index c6ef37d0..457a2583 100644 --- a/internal/document/text/create.go +++ b/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 } diff --git a/internal/document/text/create_test.go b/internal/document/text/create_test.go index c6434e4e..094a0b79 100644 --- a/internal/document/text/create_test.go +++ b/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) { diff --git a/internal/document/text/extract.go b/internal/document/text/extract.go index 353e653c..90f72ff3 100644 --- a/internal/document/text/extract.go +++ b/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 ¶ms.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 diff --git a/internal/document/text/extract_test.go b/internal/document/text/extract_test.go index accbc413..9a4d1a93 100644 --- a/internal/document/text/extract_test.go +++ b/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) } diff --git a/internal/query/result/set/set.go b/internal/query/result/set/set.go index b465f8b4..e3f2a2ce 100644 --- a/internal/query/result/set/set.go +++ b/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 } diff --git a/internal/query/result/set/set_test.go b/internal/query/result/set/set_test.go index 432b5b35..ba421a29 100644 --- a/internal/query/result/set/set_test.go +++ b/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())), diff --git a/internal/query/sync/sync.go b/internal/query/sync/sync.go index 5e31a329..18a27927 100644 --- a/internal/query/sync/sync.go +++ b/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) diff --git a/internal/query/sync/sync_test.go b/internal/query/sync/sync_test.go index 1127ca2f..8605e20e 100644 --- a/internal/query/sync/sync_test.go +++ b/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)) + }) } diff --git a/internal/server/runner/listener.go b/internal/server/runner/listener.go index b569e752..430a9996 100644 --- a/internal/server/runner/listener.go +++ b/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 { diff --git a/internal/server/runner/poll.go b/internal/server/runner/poll.go index 538229e5..5983139d 100644 --- a/internal/server/runner/poll.go +++ b/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, }) diff --git a/internal/server/runner/poll_test.go b/internal/server/runner/poll_test.go index 5f828dc1..7d87a0be 100644 --- a/internal/server/runner/poll_test.go +++ b/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( diff --git a/mocks/runner/mock_Controller.go b/mocks/runner/mock_Controller.go index 5659ce81..1e9e4d25 100644 --- a/mocks/runner/mock_Controller.go +++ b/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 }