diff --git a/api/docInitRunner/runner_test.go b/api/docInitRunner/runner_test.go index eaa7b4eb..234808da 100644 --- a/api/docInitRunner/runner_test.go +++ b/api/docInitRunner/runner_test.go @@ -7,23 +7,29 @@ import ( "queryorchestration/internal/client" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" + "queryorchestration/internal/document" documentinit "queryorchestration/internal/document/init" "queryorchestration/internal/job" "queryorchestration/internal/job/collector" "queryorchestration/internal/serviceconfig" + "queryorchestration/internal/serviceconfig/objectstore" "queryorchestration/internal/serviceconfig/queue/documentclean" + objectstoremock "queryorchestration/mocks/objectstore" "testing" + "github.com/aws/aws-sdk-go-v2/service/s3" "github.com/aws/aws-sdk-go-v2/service/sqs/types" "github.com/go-playground/validator/v10" "github.com/google/uuid" "github.com/pashagolub/pgxmock/v3" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" ) type DocInitConfig struct { serviceconfig.BaseConfig documentclean.DocCleanConfig + objectstore.ObjectStoreConfig } func TestDocInitRunner(t *testing.T) { @@ -37,6 +43,8 @@ func TestDocInitRunner(t *testing.T) { cfg := &DocInitConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) + mockStore := objectstoremock.NewMockS3Client(t) + cfg.StoreClient = mockStore runner := docinitrunner.New(validator.New(), &docinitrunner.Services{ Document: documentinit.New(cfg, &documentinit.Services{ @@ -54,6 +62,7 @@ func TestDocInitRunner(t *testing.T) { } doc := documentinit.Create{ JobID: j.ID, + Bucket: "bucket_name", Location: "/I/am/here", } bodyBytes, err := json.Marshal(doc) @@ -63,6 +72,11 @@ func TestDocInitRunner(t *testing.T) { Body: &body, } + docinfo := document.Document{ + ID: uuid.New(), + Hash: "example_hash", + } + 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), @@ -71,11 +85,30 @@ func TestDocInitRunner(t *testing.T) { pgxmock.NewRows([]string{"id", "name", "canSync"}). AddRow(database.MustToDBUUID(j.ClientID), "client_name", true), ) - pool.ExpectQuery("name: CreateDocument :one").WithArgs(database.MustToDBUUID(doc.JobID), pgxmock.AnyArg(), doc.Location). + pool.ExpectQuery("-- name: GetDocumentIDByHash :one").WithArgs(docinfo.Hash, database.MustToDBUUID(doc.JobID)).WillReturnRows( + pgxmock.NewRows([]string{"id"}), + ) + pool.ExpectBegin() + pool.ExpectQuery("name: CreateDocument :one").WithArgs(database.MustToDBUUID(doc.JobID), docinfo.Hash). WillReturnRows( pgxmock.NewRows([]string{"id"}). - AddRow(database.MustToDBUUID(uuid.New())), + AddRow(database.MustToDBUUID(docinfo.ID)), ) + pool.ExpectExec("name: AddDocumentEntry :exec").WithArgs(database.MustToDBUUID(docinfo.ID), doc.Bucket, doc.Location). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectCommit() + + mockStore.EXPECT(). + HeadObject( + mock.Anything, + mock.MatchedBy(func(in *s3.HeadObjectInput) bool { + return *in.Bucket == doc.Bucket && *in.Key == doc.Location + }), + mock.Anything, + ). + Return(&s3.HeadObjectOutput{ + ETag: &docinfo.Hash, + }, nil) err = runner.Process(ctx, msg) assert.NoError(t, err) diff --git a/api/queryService/query_test.go b/api/queryService/query_test.go index e403c117..7af4580d 100644 --- a/api/queryService/query_test.go +++ b/api/queryService/query_test.go @@ -226,10 +226,9 @@ func TestTestQuery(t *testing.T) { JobID: uuid.New(), } doc := document.Document{ - ID: uuid.New(), - JobID: coll.JobID, - Hash: "example_hash", - Location: "example_location", + ID: uuid.New(), + JobID: coll.JobID, + Hash: "example_hash", } params := &query.Test{ QueryID: uuid.New(), @@ -252,8 +251,8 @@ func TestTestQuery(t *testing.T) { reqID := database.MustToDBUUID(uuid.New()) pool.ExpectQuery("name: GetDocument :one").WithArgs(database.MustToDBUUID(doc.ID)). WillReturnRows( - pgxmock.NewRows([]string{"id", "jobId", "hash", "location"}). - AddRow(database.MustToDBUUID(doc.ID), database.MustToDBUUID(doc.JobID), doc.Hash, doc.Location), + pgxmock.NewRows([]string{"id", "jobId", "hash"}). + AddRow(database.MustToDBUUID(doc.ID), database.MustToDBUUID(doc.JobID), doc.Hash), ) pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(database.MustToDBUUID(doc.JobID)). WillReturnRows( diff --git a/cmd/docInitRunner/main.go b/cmd/docInitRunner/main.go index 739e56fc..cd40d29d 100644 --- a/cmd/docInitRunner/main.go +++ b/cmd/docInitRunner/main.go @@ -62,7 +62,7 @@ func main() { os.Exit(1) } - err = cfg.SetStoreClientAndPingByName(ctx, cfg.BucketName) + err = cfg.SetStoreClient(ctx) if err != nil { slog.Error(err.Error()) os.Exit(1) diff --git a/database/migrations/00000000000006_documents.up.sql b/database/migrations/00000000000006_documents.up.sql index 3f921e28..b64719fa 100644 --- a/database/migrations/00000000000006_documents.up.sql +++ b/database/migrations/00000000000006_documents.up.sql @@ -2,6 +2,15 @@ CREATE TABLE documents ( id uuid primary key DEFAULT gen_random_uuid(), jobId uuid not null, hash text not null, + foreign key (jobId) references jobs(id), + unique(jobId, hash) +); + +CREATE TABLE documentEntries ( + id uuid primary key DEFAULT gen_random_uuid(), + documentId uuid not null, + bucket text not null, location text not null, - foreign key (jobId) references jobs(id) + createdAt timestamp default now(), + foreign key (documentId) references documents(id) ); \ No newline at end of file diff --git a/database/queries/document.sql b/database/queries/document.sql index 65bb3987..0338821c 100644 --- a/database/queries/document.sql +++ b/database/queries/document.sql @@ -1,5 +1,11 @@ -- name: GetDocument :one -SELECT id, jobId, hash, location FROM documents WHERE id = $1 LIMIT 1; +SELECT id, jobId, hash FROM documents WHERE id = $1 LIMIT 1; -- name: CreateDocument :one -INSERT INTO documents (jobId, hash, location) VALUES ($1, $2, $3) RETURNING id; \ No newline at end of file +INSERT INTO documents (jobId, hash) VALUES ($1, $2) RETURNING id; + +-- name: AddDocumentEntry :exec +INSERT INTO documentEntries (documentId, bucket, location) VALUES ($1, $2, $3); + +-- name: GetDocumentIDByHash :one +SELECT id FROM documents WHERE hash = $1 and jobId = $2; \ No newline at end of file diff --git a/internal/database/repository/document.sql.go b/internal/database/repository/document.sql.go index 38ee47a1..0adceae5 100644 --- a/internal/database/repository/document.sql.go +++ b/internal/database/repository/document.sql.go @@ -11,41 +11,72 @@ import ( "github.com/jackc/pgx/v5/pgtype" ) +const addDocumentEntry = `-- name: AddDocumentEntry :exec +INSERT INTO documentEntries (documentId, bucket, location) VALUES ($1, $2, $3) +` + +type AddDocumentEntryParams struct { + Documentid pgtype.UUID `db:"documentid"` + Bucket string `db:"bucket"` + Location string `db:"location"` +} + +// AddDocumentEntry +// +// INSERT INTO documentEntries (documentId, bucket, location) VALUES ($1, $2, $3) +func (q *Queries) AddDocumentEntry(ctx context.Context, arg *AddDocumentEntryParams) error { + _, err := q.db.Exec(ctx, addDocumentEntry, arg.Documentid, arg.Bucket, arg.Location) + return err +} + const createDocument = `-- name: CreateDocument :one -INSERT INTO documents (jobId, hash, location) VALUES ($1, $2, $3) RETURNING id +INSERT INTO documents (jobId, hash) VALUES ($1, $2) RETURNING id ` type CreateDocumentParams struct { - Jobid pgtype.UUID `db:"jobid"` - Hash string `db:"hash"` - Location string `db:"location"` + Jobid pgtype.UUID `db:"jobid"` + Hash string `db:"hash"` } // CreateDocument // -// INSERT INTO documents (jobId, hash, location) VALUES ($1, $2, $3) RETURNING id +// INSERT INTO documents (jobId, hash) VALUES ($1, $2) RETURNING id func (q *Queries) CreateDocument(ctx context.Context, arg *CreateDocumentParams) (pgtype.UUID, error) { - row := q.db.QueryRow(ctx, createDocument, arg.Jobid, arg.Hash, arg.Location) + row := q.db.QueryRow(ctx, createDocument, arg.Jobid, arg.Hash) var id pgtype.UUID err := row.Scan(&id) return id, err } const getDocument = `-- name: GetDocument :one -SELECT id, jobId, hash, location FROM documents WHERE id = $1 LIMIT 1 +SELECT id, jobId, hash FROM documents WHERE id = $1 LIMIT 1 ` // GetDocument // -// SELECT id, jobId, hash, location FROM documents WHERE id = $1 LIMIT 1 +// SELECT id, jobId, hash FROM documents WHERE id = $1 LIMIT 1 func (q *Queries) GetDocument(ctx context.Context, id pgtype.UUID) (*Document, error) { row := q.db.QueryRow(ctx, getDocument, id) var i Document - err := row.Scan( - &i.ID, - &i.Jobid, - &i.Hash, - &i.Location, - ) + err := row.Scan(&i.ID, &i.Jobid, &i.Hash) return &i, err } + +const getDocumentIDByHash = `-- name: GetDocumentIDByHash :one +SELECT id FROM documents WHERE hash = $1 and jobId = $2 +` + +type GetDocumentIDByHashParams struct { + Hash string `db:"hash"` + Jobid pgtype.UUID `db:"jobid"` +} + +// GetDocumentIDByHash +// +// SELECT id FROM documents WHERE hash = $1 and jobId = $2 +func (q *Queries) GetDocumentIDByHash(ctx context.Context, arg *GetDocumentIDByHashParams) (pgtype.UUID, error) { + row := q.db.QueryRow(ctx, getDocumentIDByHash, arg.Hash, arg.Jobid) + var id pgtype.UUID + err := row.Scan(&id) + return id, err +} diff --git a/internal/database/repository/document_test.go b/internal/database/repository/document_test.go index 2cfeb3ad..5f264028 100644 --- a/internal/database/repository/document_test.go +++ b/internal/database/repository/document_test.go @@ -39,6 +39,15 @@ func TestDocument(t *testing.T) { assert.NoError(t, err) assert.NotEmpty(t, id) + bucket := "example_bucket" + location := "/i/am/here" + err = queries.AddDocumentEntry(ctx, &repository.AddDocumentEntryParams{ + Documentid: id, + Bucket: bucket, + Location: location, + }) + assert.NoError(t, err) + doc, err := queries.GetDocument(ctx, id) assert.NoError(t, err) assert.EqualExportedValues(t, &repository.Document{ @@ -46,4 +55,11 @@ func TestDocument(t *testing.T) { Jobid: jobId, Hash: hash, }, doc) + + docid, err := queries.GetDocumentIDByHash(ctx, &repository.GetDocumentIDByHashParams{ + Hash: hash, + Jobid: jobId, + }) + assert.NoError(t, err) + assert.EqualExportedValues(t, id, docid) } diff --git a/internal/database/repository/models.go b/internal/database/repository/models.go index df0d1e58..95f75243 100644 --- a/internal/database/repository/models.go +++ b/internal/database/repository/models.go @@ -108,10 +108,17 @@ type Collectorquerydependencytree struct { } type Document struct { - ID pgtype.UUID `db:"id"` - Jobid pgtype.UUID `db:"jobid"` - Hash string `db:"hash"` - Location string `db:"location"` + ID pgtype.UUID `db:"id"` + Jobid pgtype.UUID `db:"jobid"` + Hash string `db:"hash"` +} + +type Documententry struct { + ID pgtype.UUID `db:"id"` + Documentid pgtype.UUID `db:"documentid"` + Bucket string `db:"bucket"` + Location string `db:"location"` + Createdat pgtype.Timestamp `db:"createdat"` } type Fullactivecollector struct { diff --git a/internal/document/get.go b/internal/document/get.go index 4e7d44fb..ca1141ce 100644 --- a/internal/document/get.go +++ b/internal/document/get.go @@ -14,9 +14,8 @@ func (s *Service) Get(ctx context.Context, id uuid.UUID) (*Document, error) { } return &Document{ - ID: database.MustToUUID(doc.ID), - JobID: database.MustToUUID(doc.Jobid), - Hash: doc.Hash, - Location: doc.Location, + ID: database.MustToUUID(doc.ID), + JobID: database.MustToUUID(doc.Jobid), + Hash: doc.Hash, }, nil } diff --git a/internal/document/get_test.go b/internal/document/get_test.go index 6b6cf0b2..46f69516 100644 --- a/internal/document/get_test.go +++ b/internal/document/get_test.go @@ -27,16 +27,15 @@ func TestGet(t *testing.T) { svc := document.New(cfg) doc := document.Document{ - ID: uuid.New(), - JobID: uuid.New(), - Hash: "example_hash", - Location: "example_location", + ID: uuid.New(), + JobID: uuid.New(), + Hash: "example_hash", } pool.ExpectQuery("name: GetDocument :one").WithArgs(database.MustToDBUUID(doc.ID)). WillReturnRows( - pgxmock.NewRows([]string{"id", "jobId", "hash", "location"}). - AddRow(database.MustToDBUUID(doc.ID), database.MustToDBUUID(doc.JobID), doc.Hash, doc.Location), + pgxmock.NewRows([]string{"id", "jobId", "hash"}). + AddRow(database.MustToDBUUID(doc.ID), database.MustToDBUUID(doc.JobID), doc.Hash), ) adoc, err := svc.Get(ctx, doc.ID) diff --git a/internal/document/init/create.go b/internal/document/init/create.go index fd43799b..b7044a6d 100644 --- a/internal/document/init/create.go +++ b/internal/document/init/create.go @@ -2,6 +2,8 @@ package documentinit import ( "context" + "database/sql" + "errors" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" "queryorchestration/internal/document" @@ -9,13 +11,16 @@ import ( "queryorchestration/internal/job" "queryorchestration/internal/serviceconfig/queue" + "github.com/aws/aws-sdk-go-v2/service/s3" "github.com/aws/aws-sdk-go-v2/service/sqs/types" "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" ) type Create struct { JobID uuid.UUID `json:"jobId" validate:"required,uuid"` Location document.Location `json:"location" validate:"required"` + Bucket string `json:"bucket" validate:"required"` } func (s *Service) Create(ctx context.Context, doc *Create) (uuid.UUID, error) { @@ -42,38 +47,82 @@ func (s *Service) Create(ctx context.Context, doc *Create) (uuid.UUID, error) { return id, nil } -func (s *Service) getCreateParams(ctx context.Context, doc *Create) (*repository.CreateDocumentParams, error) { - // TODO - get document +type createDocumentParams struct { + ID *pgtype.UUID + JobID pgtype.UUID + Hash string + Bucket string + Location string +} - hash, err := s.getHash() +func (s *Service) getCreateParams(ctx context.Context, doc *Create) (*createDocumentParams, error) { + res, err := s.cfg.GetStoreClient().HeadObject(ctx, &s3.HeadObjectInput{ + Bucket: &doc.Bucket, + Key: &doc.Location, + }) if err != nil { return nil, err } - return &repository.CreateDocumentParams{ - Jobid: database.MustToDBUUID(doc.JobID), - Hash: hash, + idbyhash, err := s.cfg.GetDBQueries().GetDocumentIDByHash(ctx, &repository.GetDocumentIDByHashParams{ + Jobid: database.MustToDBUUID(doc.JobID), + Hash: *res.ETag, + }) + var docID *pgtype.UUID + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return nil, err + } else if err == nil { + docID = &idbyhash + } + + return &createDocumentParams{ + ID: docID, + JobID: database.MustToDBUUID(doc.JobID), + Hash: *res.ETag, + Bucket: doc.Bucket, Location: doc.Location, }, nil } -func (s *Service) submitCreate(ctx context.Context, params *repository.CreateDocumentParams) (uuid.UUID, error) { - // TODO create - or if hash exists log attempt - dbid, err := s.cfg.GetDBQueries().CreateDocument(ctx, params) +func (s *Service) submitCreate(ctx context.Context, params *createDocumentParams) (uuid.UUID, error) { + var id uuid.UUID + + err := s.cfg.ExecuteDBTransaction(ctx, func(ctx context.Context, q *repository.Queries) error { + var dbid pgtype.UUID + if params.ID == nil { + createid, err := s.cfg.GetDBQueries().CreateDocument(ctx, &repository.CreateDocumentParams{ + Jobid: params.JobID, + Hash: params.Hash, + }) + if err != nil { + return err + } + + dbid = createid + } else { + dbid = *params.ID + } + + err := s.cfg.GetDBQueries().AddDocumentEntry(ctx, &repository.AddDocumentEntryParams{ + Documentid: dbid, + Bucket: params.Bucket, + Location: params.Location, + }) + if err != nil { + return err + } + + id = database.MustToUUID(dbid) + + return nil + }) if err != nil { return uuid.Nil, err } - id := database.MustToUUID(dbid) - return id, nil } -func (s *Service) getHash() (string, error) { - // TODO - return "example_hash", nil -} - func (s *Service) informCreate(ctx context.Context, id uuid.UUID, j *job.Job) error { if !j.CanSync { return nil diff --git a/internal/document/init/create_test.go b/internal/document/init/create_test.go index 3ccda88d..8453f21b 100644 --- a/internal/document/init/create_test.go +++ b/internal/document/init/create_test.go @@ -2,6 +2,7 @@ package documentinit import ( "context" + "errors" "fmt" "queryorchestration/internal/client" "queryorchestration/internal/database" @@ -9,10 +10,13 @@ import ( "queryorchestration/internal/document" "queryorchestration/internal/job" "queryorchestration/internal/serviceconfig" + "queryorchestration/internal/serviceconfig/objectstore" "queryorchestration/internal/serviceconfig/queue/documentclean" + objectstoremock "queryorchestration/mocks/objectstore" queuemock "queryorchestration/mocks/queue" "testing" + "github.com/aws/aws-sdk-go-v2/service/s3" "github.com/aws/aws-sdk-go-v2/service/sqs" "github.com/google/uuid" "github.com/pashagolub/pgxmock/v3" @@ -23,6 +27,7 @@ import ( type DocInitConfig struct { serviceconfig.BaseConfig documentclean.DocCleanConfig + objectstore.ObjectStoreConfig } func TestCreate(t *testing.T) { @@ -35,6 +40,8 @@ func TestCreate(t *testing.T) { cfg := &DocInitConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) + mockStore := objectstoremock.NewMockS3Client(t) + cfg.StoreClient = mockStore svc := New(cfg, &Services{ Job: job.New(cfg, &job.Services{ @@ -47,10 +54,12 @@ func TestCreate(t *testing.T) { ClientID: uuid.New(), } doc := document.Document{ - ID: uuid.New(), - JobID: j.ID, - Location: "example_location", + ID: uuid.New(), + JobID: j.ID, + Hash: "example_hash", } + bucket := "eample_bucket" + location := "/i/am/here" pool.ExpectQuery("name: GetJob :one").WithArgs(database.MustToDBUUID(j.ID)).WillReturnRows( pgxmock.NewRows([]string{"id", "clientId", "canSync"}). @@ -60,15 +69,35 @@ func TestCreate(t *testing.T) { pgxmock.NewRows([]string{"id", "name", "canSync"}). AddRow(database.MustToDBUUID(j.ClientID), "client_name", true), ) - pool.ExpectQuery("name: CreateDocument :one").WithArgs(database.MustToDBUUID(doc.JobID), pgxmock.AnyArg(), doc.Location). + pool.ExpectQuery("-- name: GetDocumentIDByHash :one").WithArgs(doc.Hash, database.MustToDBUUID(doc.JobID)).WillReturnRows( + pgxmock.NewRows([]string{"id"}), + ) + pool.ExpectBegin() + pool.ExpectQuery("name: CreateDocument :one").WithArgs(database.MustToDBUUID(doc.JobID), doc.Hash). WillReturnRows( pgxmock.NewRows([]string{"id"}). AddRow(database.MustToDBUUID(doc.ID)), ) + pool.ExpectExec("name: AddDocumentEntry :exec").WithArgs(database.MustToDBUUID(doc.ID), bucket, location). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectCommit() + + mockStore.EXPECT(). + HeadObject( + mock.Anything, + mock.MatchedBy(func(in *s3.HeadObjectInput) bool { + return *in.Bucket == bucket && *in.Key == location + }), + mock.Anything, + ). + Return(&s3.HeadObjectOutput{ + ETag: &doc.Hash, + }, nil) id, err := svc.Create(ctx, &Create{ JobID: doc.JobID, - Location: doc.Location, + Location: location, + Bucket: bucket, }) assert.NoError(t, err) assert.Equal(t, doc.ID, id) @@ -104,3 +133,255 @@ func TestInformCreate(t *testing.T) { err = svc.informCreate(ctx, id, j) assert.NoError(t, err) } + +func TestGetCreateParams(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + cfg := &DocInitConfig{} + cfg.DBPool = pool + cfg.DBQueries = repository.New(pool) + mockStore := objectstoremock.NewMockS3Client(t) + cfg.StoreClient = mockStore + + svc := New(cfg, &Services{ + Job: job.New(cfg, &job.Services{ + Client: client.New(cfg), + }), + }) + + doc := document.Document{ + JobID: uuid.New(), + Hash: "example_hash", + } + bucket := "eample_bucket" + location := "/i/am/here" + + pool.ExpectQuery("-- name: GetDocumentIDByHash :one").WithArgs(doc.Hash, database.MustToDBUUID(doc.JobID)).WillReturnRows( + pgxmock.NewRows([]string{"id"}), + ) + + mockStore.EXPECT(). + HeadObject( + mock.Anything, + mock.MatchedBy(func(in *s3.HeadObjectInput) bool { + return *in.Bucket == bucket && *in.Key == location + }), + mock.Anything, + ). + Return(&s3.HeadObjectOutput{ + ETag: &doc.Hash, + }, nil) + + params, err := svc.getCreateParams(ctx, &Create{ + JobID: doc.JobID, + Location: location, + Bucket: bucket, + }) + assert.NoError(t, err) + assert.Equal(t, &createDocumentParams{ + JobID: database.MustToDBUUID(doc.JobID), + Hash: doc.Hash, + Bucket: bucket, + Location: location, + }, params) +} + +func TestGetCreateParamsExisting(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + cfg := &DocInitConfig{} + cfg.DBPool = pool + cfg.DBQueries = repository.New(pool) + mockStore := objectstoremock.NewMockS3Client(t) + cfg.StoreClient = mockStore + + svc := New(cfg, &Services{ + Job: job.New(cfg, &job.Services{ + Client: client.New(cfg), + }), + }) + + doc := document.Document{ + ID: uuid.New(), + JobID: uuid.New(), + Hash: "example_hash", + } + bucket := "eample_bucket" + location := "/i/am/here" + + pool.ExpectQuery("-- name: GetDocumentIDByHash :one").WithArgs(doc.Hash, database.MustToDBUUID(doc.JobID)).WillReturnRows( + pgxmock.NewRows([]string{"id"}). + AddRow(database.MustToDBUUID(doc.ID)), + ) + + mockStore.EXPECT(). + HeadObject( + mock.Anything, + mock.MatchedBy(func(in *s3.HeadObjectInput) bool { + return *in.Bucket == bucket && *in.Key == location + }), + mock.Anything, + ). + Return(&s3.HeadObjectOutput{ + ETag: &doc.Hash, + }, nil) + + params, err := svc.getCreateParams(ctx, &Create{ + JobID: doc.JobID, + Location: location, + Bucket: bucket, + }) + assert.NoError(t, err) + dbid := database.MustToDBUUID(doc.ID) + assert.Equal(t, &createDocumentParams{ + ID: &dbid, + JobID: database.MustToDBUUID(doc.JobID), + Hash: doc.Hash, + Bucket: bucket, + Location: location, + }, params) +} + +func TestGetCreateParamsCurrentDocErr(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + cfg := &DocInitConfig{} + cfg.DBPool = pool + cfg.DBQueries = repository.New(pool) + mockStore := objectstoremock.NewMockS3Client(t) + cfg.StoreClient = mockStore + + svc := New(cfg, &Services{ + Job: job.New(cfg, &job.Services{ + Client: client.New(cfg), + }), + }) + + doc := document.Document{ + JobID: uuid.New(), + Hash: "example_hash", + } + bucket := "eample_bucket" + location := "/i/am/here" + + pool.ExpectQuery("-- name: GetDocumentIDByHash :one").WithArgs(doc.Hash, database.MustToDBUUID(doc.JobID)). + WillReturnError(errors.New("db err")) + + mockStore.EXPECT(). + HeadObject( + mock.Anything, + mock.MatchedBy(func(in *s3.HeadObjectInput) bool { + return *in.Bucket == bucket && *in.Key == location + }), + mock.Anything, + ). + Return(&s3.HeadObjectOutput{ + ETag: &doc.Hash, + }, nil) + + _, err = svc.getCreateParams(ctx, &Create{ + JobID: doc.JobID, + Location: location, + Bucket: bucket, + }) + assert.Error(t, err) +} + +func TestSubmitCreate(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + cfg := &DocInitConfig{} + cfg.DBPool = pool + cfg.DBQueries = repository.New(pool) + + svc := New(cfg, &Services{ + Job: job.New(cfg, &job.Services{ + Client: client.New(cfg), + }), + }) + + doc := document.Document{ + ID: uuid.New(), + JobID: uuid.New(), + Hash: "example_hash", + } + bucket := "eample_bucket" + location := "/i/am/here" + + pool.ExpectBegin() + pool.ExpectQuery("name: CreateDocument :one").WithArgs(database.MustToDBUUID(doc.JobID), doc.Hash). + WillReturnRows( + pgxmock.NewRows([]string{"id"}). + AddRow(database.MustToDBUUID(doc.ID)), + ) + pool.ExpectExec("name: AddDocumentEntry :exec").WithArgs(database.MustToDBUUID(doc.ID), bucket, location). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectCommit() + + id, err := svc.submitCreate(ctx, &createDocumentParams{ + JobID: database.MustToDBUUID(doc.JobID), + Location: location, + Bucket: bucket, + Hash: doc.Hash, + }) + assert.NoError(t, err) + assert.Equal(t, doc.ID, id) +} + +func TestSubmitCreateExists(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + cfg := &DocInitConfig{} + cfg.DBPool = pool + cfg.DBQueries = repository.New(pool) + + svc := New(cfg, &Services{ + Job: job.New(cfg, &job.Services{ + Client: client.New(cfg), + }), + }) + + doc := document.Document{ + ID: uuid.New(), + JobID: uuid.New(), + Hash: "example_hash", + } + bucket := "eample_bucket" + location := "/i/am/here" + + pool.ExpectBegin() + pool.ExpectExec("name: AddDocumentEntry :exec").WithArgs(database.MustToDBUUID(doc.ID), bucket, location). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectCommit() + + docid := database.MustToDBUUID(doc.ID) + id, err := svc.submitCreate(ctx, &createDocumentParams{ + ID: &docid, + JobID: database.MustToDBUUID(doc.JobID), + Location: location, + Bucket: bucket, + Hash: doc.Hash, + }) + assert.NoError(t, err) + assert.Equal(t, doc.ID, id) +} diff --git a/internal/document/init/service.go b/internal/document/init/service.go index cb62ad90..66090395 100644 --- a/internal/document/init/service.go +++ b/internal/document/init/service.go @@ -3,6 +3,7 @@ package documentinit import ( "queryorchestration/internal/job" "queryorchestration/internal/serviceconfig" + "queryorchestration/internal/serviceconfig/objectstore" "queryorchestration/internal/serviceconfig/queue/documentclean" ) @@ -13,6 +14,7 @@ type Services struct { type ConfigProvider interface { serviceconfig.ConfigProvider documentclean.ConfigProvider + objectstore.ConfigProvider } type Service struct { diff --git a/internal/document/service.go b/internal/document/service.go index 19d1b07e..1c20b776 100644 --- a/internal/document/service.go +++ b/internal/document/service.go @@ -9,10 +9,9 @@ import ( type Location = string type Document struct { - ID uuid.UUID - JobID uuid.UUID - Hash string - Location Location + ID uuid.UUID + JobID uuid.UUID + Hash string } type Service struct { diff --git a/internal/job/collector/create.go b/internal/job/collector/create.go index fc12d542..984607e3 100644 --- a/internal/job/collector/create.go +++ b/internal/job/collector/create.go @@ -4,7 +4,7 @@ import ( "context" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" - "queryorchestration/internal/server/validation" + "queryorchestration/internal/validation" "github.com/google/uuid" "github.com/jackc/pgx/v5/pgtype" diff --git a/internal/job/collector/update.go b/internal/job/collector/update.go index cf8e6468..3e236234 100644 --- a/internal/job/collector/update.go +++ b/internal/job/collector/update.go @@ -5,7 +5,7 @@ import ( "errors" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" - "queryorchestration/internal/server/validation" + "queryorchestration/internal/validation" "github.com/google/uuid" "github.com/jackc/pgx/v5/pgtype" diff --git a/internal/query/normalize.go b/internal/query/normalize.go index 2728fd17..690db057 100644 --- a/internal/query/normalize.go +++ b/internal/query/normalize.go @@ -7,7 +7,7 @@ import ( "fmt" "queryorchestration/internal/database" resultprocessor "queryorchestration/internal/query/result/processor" - "queryorchestration/internal/server/validation" + "queryorchestration/internal/validation" "strings" "github.com/google/uuid" diff --git a/internal/query/test_test.go b/internal/query/test_test.go index a55b63f4..4f253e9c 100644 --- a/internal/query/test_test.go +++ b/internal/query/test_test.go @@ -48,10 +48,9 @@ func TestTest(t *testing.T) { JobID: uuid.New(), } doc := document.Document{ - ID: uuid.New(), - JobID: coll.JobID, - Hash: "example_hash", - Location: "example_location", + ID: uuid.New(), + JobID: coll.JobID, + Hash: "example_hash", } params := &query.Test{ QueryID: uuid.New(), @@ -62,8 +61,8 @@ func TestTest(t *testing.T) { reqID := database.MustToDBUUID(uuid.New()) pool.ExpectQuery("name: GetDocument :one").WithArgs(database.MustToDBUUID(doc.ID)). WillReturnRows( - pgxmock.NewRows([]string{"id", "jobId", "hash", "location"}). - AddRow(database.MustToDBUUID(doc.ID), database.MustToDBUUID(doc.JobID), doc.Hash, doc.Location), + pgxmock.NewRows([]string{"id", "jobId", "hash"}). + AddRow(database.MustToDBUUID(doc.ID), database.MustToDBUUID(doc.JobID), doc.Hash), ) pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(database.MustToDBUUID(doc.JobID)). WillReturnRows( diff --git a/internal/server/runner/listener_test.go b/internal/server/runner/listener_test.go index 837ab157..7c250683 100644 --- a/internal/server/runner/listener_test.go +++ b/internal/server/runner/listener_test.go @@ -44,8 +44,8 @@ func TestNew(t *testing.T) { cfg.QueueURL = test.CreateQueue(t, ctx, cfg, "queueName") srvPtr, err := New(ctx, cfg) - assert.NotNil(t, srvPtr) assert.NoError(t, err) + assert.NotNil(t, srvPtr) } func TestListen(t *testing.T) { diff --git a/internal/serviceconfig/database/pool.go b/internal/serviceconfig/database/pool.go index 1a9af075..2aec2038 100644 --- a/internal/serviceconfig/database/pool.go +++ b/internal/serviceconfig/database/pool.go @@ -88,7 +88,7 @@ func (b *DBConfig) DBPing(ctx context.Context) error { slog.Info("Successful database ping") return nil } - slog.Info("Attempted database ping") + slog.Info("Attempted database ping", "host", b.DBHost, "port", b.DBPort) case <-ctx.Done(): return ctx.Err() } diff --git a/internal/serviceconfig/objectstore/config.go b/internal/serviceconfig/objectstore/config.go index 6c468400..591d6b3c 100644 --- a/internal/serviceconfig/objectstore/config.go +++ b/internal/serviceconfig/objectstore/config.go @@ -12,7 +12,6 @@ import ( ) type ObjectStoreConfig struct { - BucketName string `env:"BUCKET_NAME,required,notEmpty"` AWSEndpointUrlS3 string `env:"AWS_ENDPOINT_URL_S3"` AWSS3UsePathStyle bool `env:"AWS_S3_USE_PATH_STYLE" envDefault:"false"` StoreClient S3Client @@ -74,7 +73,7 @@ func (c *ObjectStoreConfig) PingStoreByName(ctx context.Context, name string) er slog.Info("Successful bucket ping", "name", name) return nil } - slog.Info("Attempted bucket ping", "name", name) + slog.Info("Attempted bucket ping", "name", name, "address", c.GetS3Endpoint()) case <-ctx.Done(): return ctx.Err() } diff --git a/internal/serviceconfig/queue/config.go b/internal/serviceconfig/queue/config.go index 075f1fe6..fd0fa291 100644 --- a/internal/serviceconfig/queue/config.go +++ b/internal/serviceconfig/queue/config.go @@ -58,7 +58,7 @@ func (c *QueueConfig) PingQueueByURL(ctx context.Context, url string) error { slog.Info("Successful queue ping", "url", url, "approx_msgs", r.Attributes[string(types.QueueAttributeNameApproximateNumberOfMessages)]) return nil } - slog.Info("Attempted queue ping", "url", url) + slog.Info("Attempted queue ping", "url", url, "address", c.GetSQSEndpoint()) case <-ctx.Done(): return ctx.Err() } diff --git a/internal/serviceconfig/queue/receive.go b/internal/serviceconfig/queue/receive.go index ef07998b..0f3e09aa 100644 --- a/internal/serviceconfig/queue/receive.go +++ b/internal/serviceconfig/queue/receive.go @@ -15,8 +15,8 @@ func (c *QueueConfig) ReceiveFromQueue(ctx context.Context, params *ReceiveParam return c.QueueClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{ QueueUrl: ¶ms.QueueURL, MaxNumberOfMessages: 1, - WaitTimeSeconds: 20, - VisibilityTimeout: 30, + WaitTimeSeconds: 30, + VisibilityTimeout: 5, MessageAttributeNames: params.Attributes, }) } diff --git a/internal/test/queue.go b/internal/test/queue.go index 8428a8c3..5fb5f2e1 100644 --- a/internal/test/queue.go +++ b/internal/test/queue.go @@ -4,6 +4,7 @@ import ( "context" "queryorchestration/internal/serviceconfig" "queryorchestration/internal/serviceconfig/queue" + "regexp" "testing" "github.com/aws/aws-sdk-go-v2/aws" @@ -38,20 +39,20 @@ func AssertMessage(t *testing.T, ctx context.Context, cfg serviceconfig.ConfigPr return result.Messages[0] } -func AssertMessageBody(t *testing.T, ctx context.Context, cfg serviceconfig.ConfigProvider, url string, body string) { +func AssertMessageBody(t *testing.T, ctx context.Context, cfg serviceconfig.ConfigProvider, url string, body *regexp.Regexp) { message := AssertMessage(t, ctx, cfg, &queue.ReceiveParams{ QueueURL: url, }) - assert.Equal(t, body, *message.Body) + assert.Regexp(t, body, *message.Body) } -func AssertMessageAttr(t *testing.T, ctx context.Context, cfg serviceconfig.ConfigProvider, url string, name string, value string) { +func AssertMessageAttr(t *testing.T, ctx context.Context, cfg serviceconfig.ConfigProvider, url string, name string, value *regexp.Regexp) { message := AssertMessage(t, ctx, cfg, &queue.ReceiveParams{ QueueURL: url, Attributes: []string{name}, }) assert.NotNil(t, message.MessageAttributes[name]) - assert.Equal(t, value, *(message.MessageAttributes[name]).StringValue) + assert.Regexp(t, value, *(message.MessageAttributes[name]).StringValue) } diff --git a/internal/test/queue_test.go b/internal/test/queue_test.go index 67254b5f..083efb4f 100644 --- a/internal/test/queue_test.go +++ b/internal/test/queue_test.go @@ -4,6 +4,7 @@ import ( "context" "queryorchestration/internal/serviceconfig" "queryorchestration/internal/serviceconfig/queue" + "regexp" "testing" "github.com/aws/aws-sdk-go-v2/aws" @@ -84,7 +85,7 @@ func TestAssertMessageBodyWait(t *testing.T) { }) assert.NoError(t, err) - AssertMessageBody(t, ctx, cfg, url, "\"body\"") + AssertMessageBody(t, ctx, cfg, url, regexp.MustCompile("\"body\"")) } func TestAssertMessageAttrWait(t *testing.T) { @@ -119,5 +120,5 @@ func TestAssertMessageAttrWait(t *testing.T) { }) assert.NoError(t, err) - AssertMessageAttr(t, ctx, cfg, url, name, value) + AssertMessageAttr(t, ctx, cfg, url, name, regexp.MustCompile(value)) } diff --git a/internal/server/validation/validation.go b/internal/validation/validation.go similarity index 100% rename from internal/server/validation/validation.go rename to internal/validation/validation.go diff --git a/internal/server/validation/validation_test.go b/internal/validation/validation_test.go similarity index 97% rename from internal/server/validation/validation_test.go rename to internal/validation/validation_test.go index a3302232..f9f14ef2 100644 --- a/internal/server/validation/validation_test.go +++ b/internal/validation/validation_test.go @@ -1,7 +1,7 @@ package validation_test import ( - "queryorchestration/internal/server/validation" + "queryorchestration/internal/validation" "testing" "github.com/google/uuid" diff --git a/test/docInitRunner/docinitrunner_test.go b/test/docInitRunner/docinitrunner_test.go index dc639ac1..11fb6237 100644 --- a/test/docInitRunner/docinitrunner_test.go +++ b/test/docInitRunner/docinitrunner_test.go @@ -11,8 +11,11 @@ import ( documentcleanc "queryorchestration/internal/serviceconfig/queue/documentclean" "queryorchestration/internal/test" queryservice "queryorchestration/pkg/queryService" + "regexp" + "strings" "testing" + "github.com/aws/aws-sdk-go-v2/service/s3" "github.com/stretchr/testify/assert" ) @@ -55,7 +58,6 @@ func TestDocInitRunner(t *testing.T) { Name: test.DocInitRunner, Env: map[string]string{ "DOCUMENT_CLEAN_URL": doccleanurl, - "BUCKET_NAME": bucketName, }, }, }, @@ -89,21 +91,24 @@ func TestDocInitRunner(t *testing.T) { }) assert.NoError(t, err) - // TODO - upload example doc location := "/i/am/here" - - document := documentinit.Create{ - JobID: jobRes.JSON201.Id, - Location: location, - } - - err = cfg.SendToQueue(ctx, &queue.SendParams{ - QueueURL: net.Runners[test.DocInitRunner].URI, - Body: document, + body := strings.NewReader("hello world") + _, err = cfg.StoreClient.PutObject(ctx, &s3.PutObjectInput{ + Bucket: &bucketName, + Key: &location, + Body: body, }) assert.NoError(t, err) - _ = test.AssertMessage(t, ctx, cfg, &queue.ReceiveParams{ - QueueURL: doccleanurl, + err = cfg.SendToQueue(ctx, &queue.SendParams{ + QueueURL: net.Runners[test.DocInitRunner].URI, + Body: documentinit.Create{ + JobID: jobRes.JSON201.Id, + Location: location, + }, }) + assert.NoError(t, err) + + resRegex := regexp.MustCompile(`{"id": "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"`) + test.AssertMessageBody(t, ctx, cfg, doccleanurl, resRegex) }